Add support to redirect from a Hive catalog to an Iceberg catalog

presto-hive/src/main/java/io/prestosql/plugin/hive/HiveConfig.java

@@ -134,6 +134,9 @@ public class HiveConfig
 
     private boolean projectionPushdownEnabled = true;
 
+    private boolean redirectToIcebergEnabled;
+    private String redirectToIcebergCatalog = "iceberg";
+
     public int getMaxInitialSplits()
     {
         return maxInitialSplits;
@@ -977,4 +980,31 @@ public HiveConfig setProjectionPushdownEnabled(boolean projectionPushdownEnabled
         this.projectionPushdownEnabled = projectionPushdownEnabled;
         return this;
     }
+
+    public boolean isRedirectToIcebergEnabled()
+    {
+        return redirectToIcebergEnabled;
+    }
+
+    @Config("hive.redirect-to-iceberg-enabled")
+    @ConfigDescription("Redirect to a catalog configured with Iceberg Connector")
+    public HiveConfig setRedirectToIcebergEnabled(boolean redirectToIcebergEnabled)
+    {
+        this.redirectToIcebergEnabled = redirectToIcebergEnabled;
+        return this;
+    }
+
+    @NotNull
+    public String getRedirectToIcebergCatalog()
+    {
+        return redirectToIcebergCatalog;
+    }
+
+    @Config("hive.redirect-to-iceberg-catalog")
+    @ConfigDescription("The Iceberg catalog to redirect to")
+    public HiveConfig setRedirectToIcebergCatalog(String redirectToIcebergCatalog)
+    {
+        this.redirectToIcebergCatalog = redirectToIcebergCatalog;
+        return this;
+    }
 }

presto-hive/src/main/java/io/prestosql/plugin/hive/HiveMetadata.java

@@ -120,6 +120,8 @@
 import java.util.function.Function;
 import java.util.function.Predicate;
 import java.util.function.Supplier;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 import java.util.stream.Stream;
@@ -159,11 +161,13 @@
 import static io.prestosql.plugin.hive.HivePartitionManager.extractPartitionValues;
 import static io.prestosql.plugin.hive.HiveSessionProperties.getCompressionCodec;
 import static io.prestosql.plugin.hive.HiveSessionProperties.getHiveStorageFormat;
+import static io.prestosql.plugin.hive.HiveSessionProperties.getRedirectToIcebergCatalog;
 import static io.prestosql.plugin.hive.HiveSessionProperties.isBucketExecutionEnabled;
 import static io.prestosql.plugin.hive.HiveSessionProperties.isCollectColumnStatisticsOnWrite;
 import static io.prestosql.plugin.hive.HiveSessionProperties.isCreateEmptyBucketFiles;
 import static io.prestosql.plugin.hive.HiveSessionProperties.isOptimizedMismatchedBucketCount;
 import static io.prestosql.plugin.hive.HiveSessionProperties.isProjectionPushdownEnabled;
+import static io.prestosql.plugin.hive.HiveSessionProperties.isRedirectToIcebergEnabled;
 import static io.prestosql.plugin.hive.HiveSessionProperties.isRespectTableFormat;
 import static io.prestosql.plugin.hive.HiveSessionProperties.isSortedWritingEnabled;
 import static io.prestosql.plugin.hive.HiveSessionProperties.isStatisticsEnabled;
@@ -279,6 +283,8 @@ public class HiveMetadata
     public static final String AVRO_SCHEMA_URL_KEY = "avro.schema.url";
     public static final String SPARK_TABLE_PROVIDER_KEY = "spark.sql.sources.provider";
     public static final String DELTA_LAKE_PROVIDER = "delta";
+    public static final String TABLE_TYPE_PROP = "table_type";
+    public static final String ICEBERG_TABLE_TYPE_VALUE = "iceberg";
 
     private static final String CSV_SEPARATOR_KEY = OpenCSVSerde.SEPARATORCHAR;
     private static final String CSV_QUOTE_KEY = OpenCSVSerde.QUOTECHAR;
@@ -298,6 +304,12 @@ public class HiveMetadata
     private final HiveStatisticsProvider hiveStatisticsProvider;
     private final AccessControlMetadata accessControlMetadata;
 
+    // Copied from IcebergTableHandle
+    private static final Pattern ICEBERG_TABLE_NAME_PATTERN = Pattern.compile("" +
+            "(?<table>[^$@]+)" +
+            "(?:@(?<ver1>[0-9]+))?" +
+            "(?:\\$(?<type>[^@]+)(?:@(?<ver2>[0-9]+))?)?");
+
     public HiveMetadata(
             CatalogName catalogName,
             SemiTransactionalHiveMetastore metastore,
@@ -2654,6 +2666,37 @@ public void cleanupQuery(ConnectorSession session)
         metastore.cleanupQuery(session);
     }
 
+    @Override
+    public Optional<String> redirectCatalog(ConnectorSession session, SchemaTableName tableName)
+    {
+        if (!isRedirectToIcebergEnabled(session)) {
+            return Optional.empty();
+        }
+        if (!filterSchema(tableName.getSchemaName())) {
+            return Optional.empty();
+        }
+        String table = tableName.getTableName();
+        Matcher icebergNameMatch = ICEBERG_TABLE_NAME_PATTERN.matcher(table);
+        if (icebergNameMatch.matches()) {
+            table = icebergNameMatch.group("table");
+        }
+        Optional<Table> hiveTable = metastore.getTable(new HiveIdentity(session), tableName.getSchemaName(), table);
+        if (hiveTable.isEmpty()) {
+            return Optional.empty();
+        }
+
+        if (isIcebergTable(hiveTable.get())) {
+            return Optional.of(getRedirectToIcebergCatalog(session));
+        }
+
+        return Optional.empty();
+    }
+
+    private static boolean isIcebergTable(Table table)
+    {
+        return ICEBERG_TABLE_TYPE_VALUE.equalsIgnoreCase(table.getParameters().get(TABLE_TYPE_PROP));
+    }
+
     public static Optional<SchemaTableName> getSourceTableNameFromSystemTable(SchemaTableName tableName)
     {
         return Stream.of(SystemTableHandler.values())

presto-hive/src/main/java/io/prestosql/plugin/hive/HiveSessionProperties.java

@@ -91,6 +91,8 @@ public final class HiveSessionProperties
     private static final String QUERY_PARTITION_FILTER_REQUIRED = "query_partition_filter_required";
     private static final String PROJECTION_PUSHDOWN_ENABLED = "projection_pushdown_enabled";
     private static final String PARQUET_OPTIMIZED_WRITER_ENABLED = "parquet_optimized_writer_enabled";
+    private static final String REDIRECT_TO_ICEBERG_ENABLED = "redirect_to_iceberg_enabled";
+    private static final String REDIRECT_TO_ICEBERG_CATALOG = "redirect_to_iceberg_catalog";
 
     private final List<PropertyMetadata<?>> sessionProperties;
 
@@ -369,6 +371,16 @@ public HiveSessionProperties(
                         PARQUET_OPTIMIZED_WRITER_ENABLED,
                         "Experimental: Enable optimized writer",
                         parquetWriterConfig.isParquetOptimizedWriterEnabled(),
+                        false),
+                booleanProperty(
+                        REDIRECT_TO_ICEBERG_ENABLED,
+                        "Enable redirecting to a catalog configured with Iceberg Connector",
+                        hiveConfig.isRedirectToIcebergEnabled(),
+                        false),
+                stringProperty(
+                        REDIRECT_TO_ICEBERG_CATALOG,
+                        "The target Iceberg catalog for redirection",
+                        hiveConfig.getRedirectToIcebergCatalog(),
                         false));
     }
 
@@ -635,4 +647,14 @@ public static boolean isParquetOptimizedWriterEnabled(ConnectorSession session)
     {
         return session.getProperty(PARQUET_OPTIMIZED_WRITER_ENABLED, Boolean.class);
     }
+
+    public static boolean isRedirectToIcebergEnabled(ConnectorSession session)
+    {
+        return session.getProperty(REDIRECT_TO_ICEBERG_ENABLED, Boolean.class);
+    }
+
+    public static String getRedirectToIcebergCatalog(ConnectorSession session)
+    {
+        return session.getProperty(REDIRECT_TO_ICEBERG_CATALOG, String.class);
+    }
 }

presto-hive/src/test/java/io/prestosql/plugin/hive/TestHiveConfig.java

@@ -96,7 +96,9 @@ public void testDefaults()
                 .setAllowRegisterPartition(false)
                 .setQueryPartitionFilterRequired(false)
                 .setPartitionUseColumnNames(false)
-                .setProjectionPushdownEnabled(true));
+                .setProjectionPushdownEnabled(true)
+                .setRedirectToIcebergEnabled(false)
+                .setRedirectToIcebergCatalog("iceberg"));
     }
 
     @Test
@@ -166,6 +168,8 @@ public void testExplicitPropertyMappings()
                 .put("hive.query-partition-filter-required", "true")
                 .put("hive.partition-use-column-names", "true")
                 .put("hive.projection-pushdown-enabled", "false")
+                .put("hive.redirect-to-iceberg-enabled", "true")
+                .put("hive.redirect-to-iceberg-catalog", "myiceberg")
                 .build();
 
         HiveConfig expected = new HiveConfig()
@@ -231,7 +235,9 @@ public void testExplicitPropertyMappings()
                 .setAllowRegisterPartition(true)
                 .setQueryPartitionFilterRequired(true)
                 .setPartitionUseColumnNames(true)
-                .setProjectionPushdownEnabled(false);
+                .setProjectionPushdownEnabled(false)
+                .setRedirectToIcebergEnabled(true)
+                .setRedirectToIcebergCatalog("myiceberg");
 
         assertFullMapping(properties, expected);
     }

presto-main/src/main/java/io/prestosql/execution/AddColumnTask.java

@@ -36,6 +36,7 @@
 
 import static com.google.common.util.concurrent.Futures.immediateFuture;
 import static io.prestosql.metadata.MetadataUtil.createQualifiedObjectName;
+import static io.prestosql.metadata.MetadataUtil.redirectToNewCatalogIfNecessary;
 import static io.prestosql.spi.StandardErrorCode.COLUMN_ALREADY_EXISTS;
 import static io.prestosql.spi.StandardErrorCode.COLUMN_TYPE_UNKNOWN;
 import static io.prestosql.spi.StandardErrorCode.NOT_FOUND;
@@ -64,6 +65,7 @@ public ListenableFuture<?> execute(AddColumn statement, TransactionManager trans
     {
         Session session = stateMachine.getSession();
         QualifiedObjectName tableName = createQualifiedObjectName(session, statement, statement.getName());
+        tableName = redirectToNewCatalogIfNecessary(session, tableName, metadata);
         Optional<TableHandle> tableHandle = metadata.getTableHandle(session, tableName);
         if (tableHandle.isEmpty()) {
             if (!statement.isTableExists()) {
@@ -72,8 +74,9 @@ public ListenableFuture<?> execute(AddColumn statement, TransactionManager trans
             return immediateFuture(null);
         }
 
-        CatalogName catalogName = metadata.getCatalogHandle(session, tableName.getCatalogName())
-                .orElseThrow(() -> new PrestoException(NOT_FOUND, "Catalog does not exist: " + tableName.getCatalogName()));
+        String catalog = tableName.getCatalogName();
+        CatalogName catalogName = metadata.getCatalogHandle(session, catalog)
+                .orElseThrow(() -> new PrestoException(NOT_FOUND, "Catalog does not exist: " + catalog));
 
         accessControl.checkCanAddColumns(session.toSecurityContext(), tableName);
 

presto-main/src/main/java/io/prestosql/execution/CommentTask.java

@@ -31,6 +31,7 @@
 
 import static com.google.common.util.concurrent.Futures.immediateFuture;
 import static io.prestosql.metadata.MetadataUtil.createQualifiedObjectName;
+import static io.prestosql.metadata.MetadataUtil.redirectToNewCatalogIfNecessary;
 import static io.prestosql.spi.StandardErrorCode.COLUMN_NOT_FOUND;
 import static io.prestosql.spi.StandardErrorCode.MISSING_TABLE;
 import static io.prestosql.spi.StandardErrorCode.NOT_SUPPORTED;
@@ -53,6 +54,7 @@ public ListenableFuture<?> execute(Comment statement, TransactionManager transac
 
         if (statement.getType() == Comment.Type.TABLE) {
             QualifiedObjectName tableName = createQualifiedObjectName(session, statement, statement.getName());
+            tableName = redirectToNewCatalogIfNecessary(session, tableName, metadata);
             Optional<TableHandle> tableHandle = metadata.getTableHandle(session, tableName);
             if (tableHandle.isEmpty()) {
                 throw semanticException(TABLE_NOT_FOUND, statement, "Table does not exist: %s", tableName);
@@ -69,6 +71,7 @@ else if (statement.getType() == Comment.Type.COLUMN) {
             }
 
             QualifiedObjectName tableName = createQualifiedObjectName(session, statement, prefix.get());
+            tableName = redirectToNewCatalogIfNecessary(session, tableName, metadata);
             Optional<TableHandle> tableHandle = metadata.getTableHandle(session, tableName);
             if (tableHandle.isEmpty()) {
                 throw semanticException(TABLE_NOT_FOUND, statement, "Table does not exist: " + tableName);

presto-main/src/main/java/io/prestosql/execution/CreateTableTask.java

@@ -51,6 +51,7 @@
 import static com.google.common.collect.ImmutableSet.toImmutableSet;
 import static com.google.common.util.concurrent.Futures.immediateFuture;
 import static io.prestosql.metadata.MetadataUtil.createQualifiedObjectName;
+import static io.prestosql.metadata.MetadataUtil.redirectToNewCatalogIfNecessary;
 import static io.prestosql.spi.StandardErrorCode.ALREADY_EXISTS;
 import static io.prestosql.spi.StandardErrorCode.CATALOG_NOT_FOUND;
 import static io.prestosql.spi.StandardErrorCode.COLUMN_TYPE_UNKNOWN;
@@ -98,6 +99,7 @@ ListenableFuture<?> internalExecute(CreateTable statement, Metadata metadata, Ac
 
         Map<NodeRef<Parameter>, Expression> parameterLookup = parameterExtractor(statement, parameters);
         QualifiedObjectName tableName = createQualifiedObjectName(session, statement, statement.getName());
+        tableName = redirectToNewCatalogIfNecessary(session, tableName, metadata);
         Optional<TableHandle> tableHandle = metadata.getTableHandle(session, tableName);
         if (tableHandle.isPresent()) {
             if (!statement.isNotExists()) {
@@ -106,8 +108,9 @@ ListenableFuture<?> internalExecute(CreateTable statement, Metadata metadata, Ac
             return immediateFuture(null);
         }
 
-        CatalogName catalogName = metadata.getCatalogHandle(session, tableName.getCatalogName())
-                .orElseThrow(() -> new PrestoException(NOT_FOUND, "Catalog does not exist: " + tableName.getCatalogName()));
+        String catalog = tableName.getCatalogName();
+        CatalogName catalogName = metadata.getCatalogHandle(session, catalog)
+                .orElseThrow(() -> new PrestoException(NOT_FOUND, "Catalog does not exist: " + catalog));
 
         LinkedHashMap<String, ColumnMetadata> columns = new LinkedHashMap<>();
         Map<String, Object> inheritedProperties = ImmutableMap.of();
@@ -154,14 +157,16 @@ ListenableFuture<?> internalExecute(CreateTable statement, Metadata metadata, Ac
             else if (element instanceof LikeClause) {
                 LikeClause likeClause = (LikeClause) element;
                 QualifiedObjectName likeTableName = createQualifiedObjectName(session, statement, likeClause.getTableName());
+                likeTableName = redirectToNewCatalogIfNecessary(session, likeTableName, metadata);
                 if (metadata.getCatalogHandle(session, likeTableName.getCatalogName()).isEmpty()) {
                     throw semanticException(CATALOG_NOT_FOUND, statement, "LIKE table catalog '%s' does not exist", likeTableName.getCatalogName());
                 }
                 if (!tableName.getCatalogName().equals(likeTableName.getCatalogName())) {
                     throw semanticException(NOT_SUPPORTED, statement, "LIKE table across catalogs is not supported");
                 }
+                QualifiedObjectName finalLikeTableName = likeTableName;
                 TableHandle likeTable = metadata.getTableHandle(session, likeTableName)
-                        .orElseThrow(() -> semanticException(TABLE_NOT_FOUND, statement, "LIKE table '%s' does not exist", likeTableName));
+                        .orElseThrow(() -> semanticException(TABLE_NOT_FOUND, statement, "LIKE table '%s' does not exist", finalLikeTableName));
 
                 TableMetadata likeTableMetadata = metadata.getTableMetadata(session, likeTable);
 

presto-main/src/main/java/io/prestosql/execution/CreateViewTask.java

@@ -37,6 +37,7 @@
 import static com.google.common.collect.ImmutableList.toImmutableList;
 import static com.google.common.util.concurrent.Futures.immediateFuture;
 import static io.prestosql.metadata.MetadataUtil.createQualifiedObjectName;
+import static io.prestosql.metadata.MetadataUtil.redirectToNewCatalogIfNecessary;
 import static io.prestosql.spi.connector.ConnectorViewDefinition.ViewColumn;
 import static io.prestosql.sql.ParameterUtils.parameterExtractor;
 import static io.prestosql.sql.SqlFormatterUtil.getFormattedSql;
@@ -72,7 +73,7 @@ public ListenableFuture<?> execute(CreateView statement, TransactionManager tran
     {
         Session session = stateMachine.getSession();
         QualifiedObjectName name = createQualifiedObjectName(session, statement, statement.getName());
-
+        name = redirectToNewCatalogIfNecessary(session, name, metadata);
         accessControl.checkCanCreateView(session.toSecurityContext(), name);
 
         String sql = getFormattedSql(statement.getQuery(), sqlParser);

presto-main/src/main/java/io/prestosql/execution/DropColumnTask.java

@@ -29,6 +29,7 @@
 
 import static com.google.common.util.concurrent.Futures.immediateFuture;
 import static io.prestosql.metadata.MetadataUtil.createQualifiedObjectName;
+import static io.prestosql.metadata.MetadataUtil.redirectToNewCatalogIfNecessary;
 import static io.prestosql.spi.StandardErrorCode.COLUMN_NOT_FOUND;
 import static io.prestosql.spi.StandardErrorCode.NOT_SUPPORTED;
 import static io.prestosql.spi.StandardErrorCode.TABLE_NOT_FOUND;
@@ -49,6 +50,7 @@ public ListenableFuture<?> execute(DropColumn statement, TransactionManager tran
     {
         Session session = stateMachine.getSession();
         QualifiedObjectName tableName = createQualifiedObjectName(session, statement, statement.getTable());
+        tableName = redirectToNewCatalogIfNecessary(session, tableName, metadata);
         Optional<TableHandle> tableHandleOptional = metadata.getTableHandle(session, tableName);
 
         if (tableHandleOptional.isEmpty()) {

presto-main/src/main/java/io/prestosql/execution/DropTableTask.java

@@ -28,6 +28,7 @@
 
 import static com.google.common.util.concurrent.Futures.immediateFuture;
 import static io.prestosql.metadata.MetadataUtil.createQualifiedObjectName;
+import static io.prestosql.metadata.MetadataUtil.redirectToNewCatalogIfNecessary;
 import static io.prestosql.spi.StandardErrorCode.TABLE_NOT_FOUND;
 import static io.prestosql.sql.analyzer.SemanticExceptions.semanticException;
 
@@ -45,6 +46,7 @@ public ListenableFuture<?> execute(DropTable statement, TransactionManager trans
     {
         Session session = stateMachine.getSession();
         QualifiedObjectName tableName = createQualifiedObjectName(session, statement, statement.getTableName());
+        tableName = redirectToNewCatalogIfNecessary(session, tableName, metadata);
 
         Optional<TableHandle> tableHandle = metadata.getTableHandle(session, tableName);
         if (tableHandle.isEmpty()) {

presto-main/src/main/java/io/prestosql/execution/DropViewTask.java

@@ -28,6 +28,7 @@
 
 import static com.google.common.util.concurrent.Futures.immediateFuture;
 import static io.prestosql.metadata.MetadataUtil.createQualifiedObjectName;
+import static io.prestosql.metadata.MetadataUtil.redirectToNewCatalogIfNecessary;
 import static io.prestosql.spi.StandardErrorCode.TABLE_NOT_FOUND;
 import static io.prestosql.sql.analyzer.SemanticExceptions.semanticException;
 
@@ -45,6 +46,7 @@ public ListenableFuture<?> execute(DropView statement, TransactionManager transa
     {
         Session session = stateMachine.getSession();
         QualifiedObjectName name = createQualifiedObjectName(session, statement, statement.getName());
+        name = redirectToNewCatalogIfNecessary(session, name, metadata);
 
         Optional<ConnectorViewDefinition> view = metadata.getView(session, name);
         if (view.isEmpty()) {

presto-main/src/main/java/io/prestosql/execution/GrantTask.java

@@ -33,6 +33,7 @@
 import static com.google.common.util.concurrent.Futures.immediateFuture;
 import static io.prestosql.metadata.MetadataUtil.createPrincipal;
 import static io.prestosql.metadata.MetadataUtil.createQualifiedObjectName;
+import static io.prestosql.metadata.MetadataUtil.redirectToNewCatalogIfNecessary;
 import static io.prestosql.spi.StandardErrorCode.INVALID_PRIVILEGE;
 import static io.prestosql.spi.StandardErrorCode.TABLE_NOT_FOUND;
 import static io.prestosql.sql.analyzer.SemanticExceptions.semanticException;
@@ -51,6 +52,7 @@ public ListenableFuture<?> execute(Grant statement, TransactionManager transacti
     {
         Session session = stateMachine.getSession();
         QualifiedObjectName tableName = createQualifiedObjectName(session, statement, statement.getTableName());
+        tableName = redirectToNewCatalogIfNecessary(session, tableName, metadata);
         Optional<TableHandle> tableHandle = metadata.getTableHandle(session, tableName);
         if (tableHandle.isEmpty()) {
             throw semanticException(TABLE_NOT_FOUND, statement, "Table '%s' does not exist", tableName);