Add support to redirect table reads from Hive to Iceberg

plugin/trino-hive/src/main/java/io/trino/plugin/hive/DefaultHiveTableRedirectionsProvider.java

@@ -0,0 +1,40 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.trino.plugin.hive;
+
+import io.trino.plugin.hive.metastore.Table;
+import io.trino.spi.connector.CatalogSchemaTableName;
+import io.trino.spi.connector.ConnectorSession;
+
+import java.util.Optional;
+
+import static io.trino.plugin.hive.HiveSessionProperties.getIcebergCatalogName;
+import static io.trino.plugin.hive.util.HiveUtil.isIcebergTable;
+
+public class DefaultHiveTableRedirectionsProvider
+        implements HiveTableRedirectionsProvider
+{
+    @Override
+    public Optional<CatalogSchemaTableName> redirectTable(ConnectorSession session, Table table)
+    {
+        Optional<String> targetCatalogName = getIcebergCatalogName(session);
+        if (targetCatalogName.isEmpty()) {
+            return Optional.empty();
+        }
+        if (isIcebergTable(table)) {
+            return targetCatalogName.map(catalog -> new CatalogSchemaTableName(catalog, table.getSchemaTableName()));
+        }
+        return Optional.empty();
+    }
+}

plugin/trino-hive/src/main/java/io/trino/plugin/hive/HiveConfig.java

@@ -153,6 +153,8 @@ public class HiveConfig
     private boolean optimizeSymlinkListing = true;
 
     private boolean legacyHiveViewTranslation;
+    private Optional<String> icebergCatalogName = Optional.empty();
+
     private DataSize targetMaxFileSize = DataSize.of(1, GIGABYTE);
 
     private boolean sizeBasedSplitWeightsEnabled = true;
@@ -1092,6 +1094,19 @@ public boolean isLegacyHiveViewTranslation()
         return this.legacyHiveViewTranslation;
     }
 
+    public Optional<String> getIcebergCatalogName()
+    {
+        return icebergCatalogName;
+    }
+
+    @Config("hive.iceberg-catalog-name")
+    @ConfigDescription("The catalog to redirect iceberg tables to")
+    public HiveConfig setIcebergCatalogName(String icebergCatalogName)
+    {
+        this.icebergCatalogName = Optional.ofNullable(icebergCatalogName);
+        return this;
+    }
+
     @Config("hive.size-based-split-weights-enabled")
     public HiveConfig setSizeBasedSplitWeightsEnabled(boolean sizeBasedSplitWeightsEnabled)
     {

plugin/trino-hive/src/main/java/io/trino/plugin/hive/HiveModule.java

@@ -91,7 +91,7 @@ public void configure(Binder binder)
         newOptionalBinder(binder, TransactionalMetadataFactory.class)
                 .setDefault().to(HiveMetadataFactory.class).in(Scopes.SINGLETON);
         newOptionalBinder(binder, HiveTableRedirectionsProvider.class)
-                .setDefault().toInstance(HiveTableRedirectionsProvider.NO_REDIRECTIONS);
+                .setDefault().to(DefaultHiveTableRedirectionsProvider.class);
         binder.bind(HiveTransactionManager.class).in(Scopes.SINGLETON);
         binder.bind(ConnectorSplitManager.class).to(HiveSplitManager.class).in(Scopes.SINGLETON);
         newExporter(binder).export(ConnectorSplitManager.class).as(generator -> generator.generatedNameOf(HiveSplitManager.class));

plugin/trino-hive/src/main/java/io/trino/plugin/hive/HiveSessionProperties.java

@@ -31,6 +31,7 @@
 
 import java.util.Collection;
 import java.util.List;
+import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.ThreadLocalRandom;
 
@@ -111,6 +112,7 @@ public final class HiveSessionProperties
     private static final String DYNAMIC_FILTERING_WAIT_TIMEOUT = "dynamic_filtering_wait_timeout";
     private static final String OPTIMIZE_SYMLINK_LISTING = "optimize_symlink_listing";
     private static final String LEGACY_HIVE_VIEW_TRANSLATION = "legacy_hive_view_translation";
+    private static final String ICEBERG_CATALOG_NAME = "iceberg_catalog_name";
     public static final String SIZE_BASED_SPLIT_WEIGHTS_ENABLED = "size_based_split_weights_enabled";
     public static final String MINIMUM_ASSIGNED_SPLIT_WEIGHT = "minimum_assigned_split_weight";
     public static final String NON_TRANSACTIONAL_OPTIMIZE_ENABLED = "non_transactional_optimize_enabled";
@@ -463,6 +465,11 @@ public HiveSessionProperties(
                         "Use legacy Hive view translation mechanism",
                         hiveConfig.isLegacyHiveViewTranslation(),
                         false),
+                stringProperty(
+                        ICEBERG_CATALOG_NAME,
+                        "Catalog to redirect to when an Iceberg table is referenced",
+                        hiveConfig.getIcebergCatalogName().orElse(null),
+                        false),
                 booleanProperty(
                         SIZE_BASED_SPLIT_WEIGHTS_ENABLED,
                         "Enable estimating split weights based on size in bytes",
@@ -790,6 +797,11 @@ public static boolean isLegacyHiveViewTranslation(ConnectorSession session)
         return session.getProperty(LEGACY_HIVE_VIEW_TRANSLATION, Boolean.class);
     }
 
+    public static Optional<String> getIcebergCatalogName(ConnectorSession session)
+    {
+        return Optional.ofNullable(session.getProperty(ICEBERG_CATALOG_NAME, String.class));
+    }
+
     public static boolean isSizeBasedSplitWeightsEnabled(ConnectorSession session)
     {
         return session.getProperty(SIZE_BASED_SPLIT_WEIGHTS_ENABLED, Boolean.class);

plugin/trino-hive/src/main/java/io/trino/plugin/hive/HiveWriterFactory.java

@@ -39,7 +39,6 @@
 import io.trino.spi.TrinoException;
 import io.trino.spi.connector.ConnectorSession;
 import io.trino.spi.connector.SortOrder;
-import io.trino.spi.session.PropertyMetadata;
 import io.trino.spi.type.Type;
 import io.trino.spi.type.TypeManager;
 import org.apache.hadoop.conf.Configuration;
@@ -62,6 +61,7 @@
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Map.Entry;
 import java.util.Optional;
 import java.util.OptionalInt;
 import java.util.Properties;
@@ -72,6 +72,7 @@
 import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.collect.ImmutableList.toImmutableList;
 import static com.google.common.collect.ImmutableMap.toImmutableMap;
+import static com.google.common.collect.Maps.immutableEntry;
 import static io.trino.plugin.hive.HiveErrorCode.HIVE_FILESYSTEM_ERROR;
 import static io.trino.plugin.hive.HiveErrorCode.HIVE_INVALID_METADATA;
 import static io.trino.plugin.hive.HiveErrorCode.HIVE_PARTITION_READ_ONLY;
@@ -256,8 +257,12 @@ public HiveWriterFactory(
 
         requireNonNull(hiveSessionProperties, "hiveSessionProperties is null");
         this.sessionProperties = hiveSessionProperties.getSessionProperties().stream()
-                .collect(toImmutableMap(PropertyMetadata::getName,
-                        entry -> session.getProperty(entry.getName(), entry.getJavaType()).toString()));
+                .map(propertyMetadata -> immutableEntry(
+                        propertyMetadata.getName(),
+                        session.getProperty(propertyMetadata.getName(), propertyMetadata.getJavaType())))
+                // The session properties collected here are used for events only. Filter out nulls to avoid problems with downstream consumers
+                .filter(entry -> entry.getValue() != null)
+                .collect(toImmutableMap(Entry::getKey, entry -> entry.getValue().toString()));
 
         Configuration conf = hdfsEnvironment.getConfiguration(new HdfsContext(session), writePath);
         configureCompression(conf, getCompressionCodec(session));

plugin/trino-hive/src/test/java/io/trino/plugin/hive/TestHiveConfig.java

@@ -105,6 +105,7 @@ public void testDefaults()
                 .setTimestampPrecision(HiveTimestampPrecision.DEFAULT_PRECISION)
                 .setOptimizeSymlinkListing(true)
                 .setLegacyHiveViewTranslation(false)
+                .setIcebergCatalogName(null)
                 .setSizeBasedSplitWeightsEnabled(true)
                 .setMinimumAssignedSplitWeight(0.05));
     }
@@ -182,6 +183,7 @@ public void testExplicitPropertyMappings()
                 .put("hive.timestamp-precision", "NANOSECONDS")
                 .put("hive.optimize-symlink-listing", "false")
                 .put("hive.legacy-hive-view-translation", "true")
+                .put("hive.iceberg-catalog-name", "iceberg")
                 .put("hive.size-based-split-weights-enabled", "false")
                 .put("hive.minimum-assigned-split-weight", "1.0")
                 .build();
@@ -256,6 +258,7 @@ public void testExplicitPropertyMappings()
                 .setTimestampPrecision(HiveTimestampPrecision.NANOSECONDS)
                 .setOptimizeSymlinkListing(false)
                 .setLegacyHiveViewTranslation(true)
+                .setIcebergCatalogName("iceberg")
                 .setSizeBasedSplitWeightsEnabled(false)
                 .setMinimumAssignedSplitWeight(1.0);
 

testing/trino-product-tests-launcher/src/main/java/io/trino/tests/product/launcher/env/environment/EnvSinglenodeHiveIcebergRedirections.java

@@ -0,0 +1,52 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.trino.tests.product.launcher.env.environment;
+
+import com.google.common.collect.ImmutableList;
+import io.trino.tests.product.launcher.docker.DockerFiles;
+import io.trino.tests.product.launcher.docker.DockerFiles.ResourceProvider;
+import io.trino.tests.product.launcher.env.Environment;
+import io.trino.tests.product.launcher.env.EnvironmentProvider;
+import io.trino.tests.product.launcher.env.common.Hadoop;
+import io.trino.tests.product.launcher.env.common.Standard;
+import io.trino.tests.product.launcher.env.common.TestsEnvironment;
+
+import javax.inject.Inject;
+
+import static io.trino.tests.product.launcher.env.EnvironmentContainers.COORDINATOR;
+import static io.trino.tests.product.launcher.env.common.Hadoop.CONTAINER_PRESTO_HIVE_PROPERTIES;
+import static io.trino.tests.product.launcher.env.common.Hadoop.CONTAINER_PRESTO_ICEBERG_PROPERTIES;
+import static org.testcontainers.utility.MountableFile.forHostPath;
+
+@TestsEnvironment
+public class EnvSinglenodeHiveIcebergRedirections
+        extends EnvironmentProvider
+{
+    private final ResourceProvider configDir;
+
+    @Inject
+    public EnvSinglenodeHiveIcebergRedirections(DockerFiles dockerFiles, Standard standard, Hadoop hadoop)
+    {
+        super(ImmutableList.of(standard, hadoop));
+        configDir = dockerFiles.getDockerFilesHostDirectory("conf/environment/singlenode-hive-iceberg-redirections");
+    }
+
+    @Override
+    public void extendEnvironment(Environment.Builder builder)
+    {
+        builder.configureContainer(COORDINATOR, container -> container
+                .withCopyFileToContainer(forHostPath(configDir.getPath("hive.properties")), CONTAINER_PRESTO_HIVE_PROPERTIES)
+                .withCopyFileToContainer(forHostPath(configDir.getPath("iceberg.properties")), CONTAINER_PRESTO_ICEBERG_PROPERTIES));
+    }
+}

testing/trino-product-tests-launcher/src/main/java/io/trino/tests/product/launcher/suite/suites/Suite7NonGeneric.java

@@ -16,6 +16,7 @@
 import com.google.common.collect.ImmutableList;
 import io.trino.tests.product.launcher.env.EnvironmentConfig;
 import io.trino.tests.product.launcher.env.EnvironmentDefaults;
+import io.trino.tests.product.launcher.env.environment.EnvSinglenodeHiveIcebergRedirections;
 import io.trino.tests.product.launcher.env.environment.EnvSinglenodeKerberosHdfsImpersonationCrossRealm;
 import io.trino.tests.product.launcher.env.environment.EnvSinglenodeMysql;
 import io.trino.tests.product.launcher.env.environment.EnvSinglenodePostgresql;
@@ -46,6 +47,7 @@ public List<SuiteTestRun> getTestRuns(EnvironmentConfig config)
                 testOnEnvironment(EnvSinglenodeSqlserver.class).withGroups("sqlserver").build(),
                 testOnEnvironment(EnvSinglenodeSparkHive.class).withGroups("hive_spark").build(),
                 testOnEnvironment(EnvSinglenodeSparkIceberg.class).withGroups("iceberg").withExcludedGroups("storage_formats").build(),
+                testOnEnvironment(EnvSinglenodeHiveIcebergRedirections.class).withGroups("hive_iceberg_redirections").build(),
                 testOnEnvironment(EnvSinglenodeKerberosHdfsImpersonationCrossRealm.class).withGroups("storage_formats", "cli", "hdfs_impersonation").build(),
                 testOnEnvironment(EnvTwoMixedHives.class).withGroups("two_hives").build(),
                 testOnEnvironment(EnvTwoKerberosHives.class).withGroups("two_hives").build());

testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/conf/environment/singlenode-hive-iceberg-redirections/hive.properties

@@ -0,0 +1,10 @@
+connector.name=hive-hadoop2
+hive.metastore.uri=thrift://hadoop-master:9083
+hive.config.resources=/docker/presto-product-tests/conf/presto/etc/hive-default-fs-site.xml
+hive.allow-add-column=true
+hive.allow-drop-column=true
+hive.allow-rename-column=true
+hive.allow-comment-table=true
+hive.allow-drop-table=true
+hive.allow-rename-table=true
+hive.iceberg-catalog-name=iceberg

testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/conf/environment/singlenode-hive-iceberg-redirections/iceberg.properties

@@ -0,0 +1,3 @@
+connector.name=iceberg
+hive.metastore.uri=thrift://hadoop-master:9083
+hive.config.resources=/docker/presto-product-tests/conf/presto/etc/hive-default-fs-site.xml

testing/trino-product-tests/src/main/java/io/trino/tests/product/TestGroups.java

@@ -47,6 +47,7 @@ public final class TestGroups
     public static final String HIVE_VIEWS = "hive_views";
     public static final String HIVE_VIEW_COMPATIBILITY = "hive_view_compatibility";
     public static final String HIVE_CACHING = "hive_caching";
+    public static final String HIVE_ICEBERG_REDIRECTIONS = "hive_iceberg_redirections";
     public static final String AUTHORIZATION = "authorization";
     public static final String HIVE_COERCION = "hive_coercion";
     public static final String CASSANDRA = "cassandra";