feature: Implement alias db lookups

.gitignore

@@ -35,6 +35,7 @@ Thumbs.db
 ################
 *~
 *.swp
+.factorypath
 
 # Gradle Files #
 ################
@@ -47,6 +48,8 @@ Thumbs.db
 */target
 /build
 */build
+/bin
+*/bin
 
 # IntelliJ specific files/directories
 out

metacat-common-server/src/main/java/com/netflix/metacat/common/server/usermetadata/AliasService.java

@@ -36,4 +36,14 @@ public interface AliasService {
     default QualifiedName getTableName(@NonNull final QualifiedName tableAlias) {
         return tableAlias;
     }
+
+    /**
+     * Returns true if the given name refers to a table alias rather than to a table itself.
+     *
+     * @param name the table name.
+     * @return true if the name is an alias, false otherwise.
+     */
+    default boolean isAlias(@NonNull final QualifiedName name) {
+        return !name.equals(getTableName(name));
+    }
 }

metacat-connector-polaris/src/functionalTest/resources/schema.sql

@@ -1,3 +1,4 @@
+DROP TABLE IF EXISTS ALIASES;
 DROP TABLE IF EXISTS TBLS;
 DROP TABLE IF EXISTS DBS;
 
@@ -31,3 +32,15 @@ CREATE TABLE TBLS (
                     CONSTRAINT unique_catalog_db_tbl UNIQUE (catalog_name, db_name, tbl_name),
                     CONSTRAINT fk_tbls_db FOREIGN KEY (catalog_name, db_name) REFERENCES DBS(catalog_name, name) ON DELETE CASCADE ON UPDATE CASCADE
 );
+
+CREATE TABLE ALIASES (
+                    id UUID DEFAULT gen_random_uuid() NOT NULL PRIMARY KEY,
+                    db_id UUID NOT NULL REFERENCES DBS(id),
+                    name VARCHAR(255) NOT NULL,
+                    source_tbl_id UUID NOT NULL REFERENCES TBLS(id),
+                    created_by VARCHAR(255),
+                    created_date TIMESTAMP NOT NULL,
+                    last_updated_by VARCHAR(255),
+                    last_updated_date TIMESTAMP NOT NULL,
+                    CONSTRAINT unique_alias_name_db UNIQUE (name, db_id)
+);

metacat-connector-polaris/src/functionalTest/scripts/datastores/aurora/docker-entrypoint-initdb-primary.d/schema.sql

@@ -1,6 +1,7 @@
 -- Enable the uuid-ossp extension for UUID generation
 CREATE EXTENSION IF NOT EXISTS "uuid-ossp";
 
+DROP TABLE IF EXISTS ALIASES;
 DROP TABLE IF EXISTS TBLS;
 DROP TABLE IF EXISTS DBS;
 
@@ -37,3 +38,15 @@ CREATE TABLE TBLS (
                     CONSTRAINT unique_catalog_db_tbl UNIQUE (catalog_name, db_name, tbl_name),
                     CONSTRAINT fk_tbls_db FOREIGN KEY (catalog_name, db_name) REFERENCES DBS(catalog_name, name) ON DELETE RESTRICT ON UPDATE CASCADE
 );
+
+CREATE TABLE ALIASES (
+                    id UUID DEFAULT uuid_generate_v4() NOT NULL PRIMARY KEY,
+                    db_id UUID NOT NULL REFERENCES DBS(id),
+                    name VARCHAR(255) NOT NULL,
+                    source_tbl_id UUID NOT NULL REFERENCES TBLS(id),
+                    created_by VARCHAR(255),
+                    created_date TIMESTAMP NOT NULL,
+                    last_updated_by VARCHAR(255),
+                    last_updated_date TIMESTAMP NOT NULL,
+                    CONSTRAINT unique_alias_name_db UNIQUE (name, db_id)
+);

metacat-connector-polaris/src/functionalTest/scripts/datastores/aurora/docker-entrypoint-initdb-reader.d/schema.sql

@@ -1,6 +1,7 @@
 -- Enable the uuid-ossp extension for UUID generation
 CREATE EXTENSION IF NOT EXISTS "uuid-ossp";
 
+DROP TABLE IF EXISTS ALIASES;
 DROP TABLE IF EXISTS TBLS;
 DROP TABLE IF EXISTS DBS;
 
@@ -37,3 +38,15 @@ CREATE TABLE TBLS (
                     CONSTRAINT unique_catalog_db_tbl UNIQUE (catalog_name, db_name, tbl_name),
                     CONSTRAINT fk_tbls_db FOREIGN KEY (catalog_name, db_name) REFERENCES DBS(catalog_name, name) ON DELETE RESTRICT ON UPDATE CASCADE
 );
+
+CREATE TABLE ALIASES (
+                    id UUID DEFAULT uuid_generate_v4() NOT NULL PRIMARY KEY,
+                    db_id UUID NOT NULL REFERENCES DBS(id),
+                    name VARCHAR(255) NOT NULL,
+                    source_tbl_id UUID NOT NULL REFERENCES TBLS(id),
+                    created_by VARCHAR(255),
+                    created_date TIMESTAMP NOT NULL,
+                    last_updated_by VARCHAR(255),
+                    last_updated_date TIMESTAMP NOT NULL,
+                    CONSTRAINT unique_alias_name_db UNIQUE (name, db_id)
+);

metacat-connector-polaris/src/main/java/com/netflix/metacat/connector/polaris/PolarisAliasService.java

@@ -0,0 +1,64 @@
+package com.netflix.metacat.connector.polaris;
+
+import com.netflix.metacat.common.QualifiedName;
+import com.netflix.metacat.common.server.properties.Config;
+import com.netflix.metacat.common.server.usermetadata.AliasService;
+import com.netflix.metacat.connector.polaris.store.repos.PolarisAliasRepository;
+import lombok.NonNull;
+import lombok.extern.slf4j.Slf4j;
+
+import java.util.Optional;
+
+/**
+ * Resolves table aliases against the polaris store.
+ */
+@Slf4j
+public class PolarisAliasService implements AliasService {
+    private final PolarisAliasRepository aliasRepository;
+    private final Config config;
+
+    /**
+     * Constructor.
+     *
+     * @param aliasRepository the alias repository
+     * @param config the runtime config
+     */
+    public PolarisAliasService(final PolarisAliasRepository aliasRepository, final Config config) {
+        this.aliasRepository = aliasRepository;
+        this.config = config;
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    @Override
+    public QualifiedName getTableName(@NonNull final QualifiedName tableAlias) {
+        return resolve(tableAlias).orElse(tableAlias);
+    }
+
+    /**
+     * {@inheritDoc}
+     *
+     * <p>Overridden so that the answer comes from the same cached lookup as
+     * {@link #getTableName(QualifiedName)} rather than costing a second resolution.
+     */
+    @Override
+    public boolean isAlias(@NonNull final QualifiedName name) {
+        return config.isTableAliasEnabled() && resolve(name).isPresent();
+    }
+
+    private Optional<QualifiedName> resolve(final QualifiedName name) {
+        if (name.getType() != QualifiedName.Type.TABLE) {
+            return Optional.empty();
+        }
+        try {
+            return aliasRepository
+                .findSourceTable(name.getCatalogName(), name.getDatabaseName(), name.getTableName())
+                .map(table -> QualifiedName.ofTable(
+                    table.getCatalogName(), table.getDbName(), table.getTblName()));
+        } catch (Exception e) {
+            log.error("Failed to resolve alias for {}", name, e);
+            throw new IllegalStateException("Failed to resolve alias for " + name, e);
+        }
+    }
+}

metacat-connector-polaris/src/main/java/com/netflix/metacat/connector/polaris/configs/PolarisConnectorConfig.java

@@ -1,6 +1,7 @@
 package com.netflix.metacat.connector.polaris.configs;
 
 import com.netflix.metacat.common.server.connectors.ConnectorContext;
+import com.netflix.metacat.common.server.usermetadata.AliasService;
 import com.netflix.metacat.common.server.util.ThreadServiceManager;
 import com.netflix.metacat.connector.hive.commonview.CommonViewHandler;
 import com.netflix.metacat.connector.hive.converters.HiveConnectorInfoConverter;
@@ -9,18 +10,22 @@
 import com.netflix.metacat.connector.hive.iceberg.IcebergTableHandler;
 import com.netflix.metacat.connector.hive.iceberg.IcebergTableOpWrapper;
 import com.netflix.metacat.connector.hive.iceberg.IcebergTableOpsProxy;
+import com.netflix.metacat.connector.polaris.PolarisAliasService;
 import com.netflix.metacat.connector.polaris.PolarisConnectorDatabaseService;
 import com.netflix.metacat.connector.polaris.PolarisConnectorPartitionService;
 import com.netflix.metacat.connector.polaris.PolarisConnectorTableService;
 import com.netflix.metacat.connector.polaris.mappers.PolarisTableMapper;
 import com.netflix.metacat.connector.polaris.store.PolarisStoreService;
+import com.netflix.metacat.connector.polaris.store.repos.PolarisAliasRepository;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
 import org.springframework.context.annotation.Bean;
 
+
 /**
  * Config for polaris connector.
  */
 public class PolarisConnectorConfig {
+
     /**
      * Creates a new instance of a polaris connector partition service.
      *
@@ -87,6 +92,22 @@ public PolarisConnectorTableService polarisTableService(
         );
     }
 
+    /**
+     * Create the alias service backing table alias resolution for this catalog.
+     *
+     * @param aliasRepository  alias repository
+     * @param connectorContext server context
+     * @return AliasService
+     */
+    @Bean
+    @ConditionalOnMissingBean(AliasService.class)
+    public AliasService polarisAliasService(
+        final PolarisAliasRepository aliasRepository,
+        final ConnectorContext connectorContext
+    ) {
+        return new PolarisAliasService(aliasRepository, connectorContext.getConfig());
+    }
+
     /**
      * Create PolarisTableMapper.
      * @param connectorContext server context

metacat-connector-polaris/src/main/java/com/netflix/metacat/connector/polaris/store/entities/PolarisAliasEntity.java

@@ -0,0 +1,55 @@
+package com.netflix.metacat.connector.polaris.store.entities;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.EqualsAndHashCode;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.ToString;
+import org.hibernate.annotations.GenericGenerator;
+import org.springframework.data.jpa.domain.support.AuditingEntityListener;
+
+import jakarta.persistence.Basic;
+import jakarta.persistence.Column;
+import jakarta.persistence.Embedded;
+import jakarta.persistence.Entity;
+import jakarta.persistence.EntityListeners;
+import jakarta.persistence.GeneratedValue;
+import jakarta.persistence.Id;
+import jakarta.persistence.Table;
+
+/**
+ * Entity class for a table alias.
+ */
+@Getter
+@AllArgsConstructor
+@NoArgsConstructor
+@Builder(toBuilder = true)
+@EqualsAndHashCode
+@Entity
+@ToString(callSuper = true)
+@Table(name = "ALIASES")
+@EntityListeners(AuditingEntityListener.class)
+public class PolarisAliasEntity {
+    @Basic
+    @Id
+    @GeneratedValue(generator = "uuid")
+    @GenericGenerator(name = "uuid", strategy = "uuid2")
+    @Column(name = "id", nullable = false, unique = true, updatable = false)
+    private String aliasId;
+
+    @Basic
+    @Column(name = "db_id", nullable = false, updatable = false)
+    private String dbId;
+
+    @Basic
+    @Column(name = "name", nullable = false, updatable = false)
+    private String name;
+
+    @Basic
+    @Column(name = "source_tbl_id", nullable = false)
+    private String sourceTblId;
+
+    @Embedded
+    private AuditEntity audit;
+}

metacat-connector-polaris/src/main/java/com/netflix/metacat/connector/polaris/store/repos/PolarisAliasRepository.java

@@ -0,0 +1,36 @@
+package com.netflix.metacat.connector.polaris.store.repos;
+
+import com.netflix.metacat.connector.polaris.store.entities.PolarisAliasEntity;
+import com.netflix.metacat.connector.polaris.store.entities.PolarisTableEntity;
+import org.springframework.data.jpa.repository.JpaRepository;
+import org.springframework.data.jpa.repository.Query;
+import org.springframework.data.repository.query.Param;
+import org.springframework.stereotype.Repository;
+
+import java.util.Optional;
+
+/**
+ * JPA repository implementation for storing PolarisAliasEntity.
+ */
+@Repository
+public interface PolarisAliasRepository extends JpaRepository<PolarisAliasEntity, String> {
+
+    /**
+     * Resolves an alias to the table it points at.
+     *
+     * @param catalogName catalog the alias is defined in
+     * @param dbName      database the alias is defined in
+     * @param aliasName   the alias
+     * @return the source table, or empty if the name is not an alias
+     */
+    @Query("SELECT t FROM PolarisAliasEntity a "
+        + "JOIN PolarisDatabaseEntity d ON d.dbId = a.dbId "
+        + "JOIN PolarisTableEntity t ON t.tblId = a.sourceTblId "
+        + "WHERE d.catalogName = :catalogName "
+        + "AND d.dbName = :dbName "
+        + "AND a.name = :aliasName")
+    Optional<PolarisTableEntity> findSourceTable(
+        @Param("catalogName") String catalogName,
+        @Param("dbName") String dbName,
+        @Param("aliasName") String aliasName);
+}

metacat-connector-polaris/src/test/java/com/netflix/metacat/connector/polaris/PolarisAliasServiceTest.java

@@ -0,0 +1,81 @@
+package com.netflix.metacat.connector.polaris;
+
+import com.netflix.metacat.common.QualifiedName;
+import com.netflix.metacat.common.server.properties.Config;
+import com.netflix.metacat.connector.polaris.store.entities.PolarisTableEntity;
+import com.netflix.metacat.connector.polaris.store.repos.PolarisAliasRepository;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+import java.util.Optional;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * Tests for alias resolution against the polaris store.
+ */
+public class PolarisAliasServiceTest {
+
+    private PolarisAliasRepository aliasRepository;
+    private Config config;
+
+    private PolarisAliasService service;
+    private QualifiedName alias;
+    private QualifiedName source;
+
+    @BeforeEach
+    public void setup() {
+        aliasRepository = Mockito.mock(PolarisAliasRepository.class);
+        config = Mockito.mock(Config.class);
+        when(config.isTableAliasEnabled()).thenReturn(true);
+        service = new PolarisAliasService(aliasRepository, config);
+        alias = QualifiedName.ofTable("prodhive", "db", "the_alias");
+        source = QualifiedName.ofTable("prodhive", "db", "the_source");
+    }
+
+    private void givenAlias(final String catalog, final String db, final String table) {
+        when(aliasRepository.findSourceTable("prodhive", "db", "the_alias"))
+            .thenReturn(Optional.of(PolarisTableEntity.builder()
+                .catalogName(catalog).dbName(db).tblName(table).build()));
+    }
+
+    @Test
+    public void testResolvesAliasToSourceTable() {
+        givenAlias("prodhive", "db", "the_source");
+        assertThat(service.getTableName(alias)).isEqualTo(source);
+        assertThat(service.isAlias(alias)).isTrue();
+    }
+
+    @Test
+    public void testReturnsInputWhenNotAnAlias() {
+        when(aliasRepository.findSourceTable("prodhive", "db", "the_alias")).thenReturn(Optional.empty());
+        assertThat(service.getTableName(alias)).isEqualTo(alias);
+        assertThat(service.isAlias(alias)).isFalse();
+    }
+
+    @Test
+    public void testResolvesAliasPointingAtAnotherDatabase() {
+        givenAlias("prodhive", "other_db", "the_source");
+        assertThat(service.getTableName(alias))
+            .isEqualTo(QualifiedName.ofTable("prodhive", "other_db", "the_source"));
+    }
+
+    @Test
+    public void testNonTableNamesAreNotLookedUp() {
+        final QualifiedName database = QualifiedName.ofDatabase("prodhive", "db");
+        assertThat(service.getTableName(database)).isEqualTo(database);
+        assertThat(service.isAlias(database)).isFalse();
+        verify(aliasRepository, never()).findSourceTable("prodhive", "db", null);
+    }
+
+    @Test
+    public void testDoesNotResolveAliasWhenAliasNotEnabled() {
+        givenAlias("prodhive", "other_db", "the_source");
+        when(config.isTableAliasEnabled()).thenReturn(false);
+        assertThat(service.isAlias(QualifiedName.ofTable("prodhive", "other_db", "the_source"))).isFalse();
+    }
+}