Pass catalog name to metastore for lakehouse connectors

Author: AnuragKDwivedi

presto-docs/src/main/sphinx/connector/deltalake.rst

@@ -49,6 +49,7 @@ Property Name                                   Description
                                                 ``true``.
 ``delta.case-sensitive-partitions-enabled``     Allows matching the names of partitioned columns in a     ``true``
                                                 case-sensitive manner.
+``hive.metastore.catalog.name``                 Specifies the catalog name to be passed to the metastore.
 =============================================== ========================================================= ============
 
 Delta Lake connector reuses many of the modules existing in Hive connector.

presto-docs/src/main/sphinx/connector/hive.rst

@@ -206,6 +206,8 @@ Property Name                                      Description
                                                    error iterating through empty files.
 
  ``hive.file-status-cache.max-retained-size``      Maximum size in bytes of the directory listing cache          ``0KB``
+
+ ``hive.metastore.catalog.name``                   Specifies the catalog name to be passed to the metastore.
 ================================================== ============================================================ ============
 
 Metastore Configuration Properties

presto-docs/src/main/sphinx/connector/hudi.rst

@@ -38,6 +38,8 @@ Property Name                           Description
 ======================================= ============================================= ===========
 ``hudi.metadata-table-enabled``         Fetch the list of file names and sizes from   false
                                         Hudi's metadata table rather than storage.
+``hive.metastore.catalog.name``         Specifies the catalog name to be passed to
+                                        the metastore.
 ======================================= ============================================= ===========
 
 File-Based Metastore

presto-docs/src/main/sphinx/connector/iceberg.rst

@@ -91,6 +91,7 @@ Property Name                                            Description
 
 ``iceberg.hive.table-refresh.backoff-scale-factor``      The multiple used to scale subsequent wait time between       4.0
                                                          retries.
+``hive.metastore.catalog.name``                          Specifies the catalog name to be passed to the metastore.
 ======================================================== ============================================================= ============
 
 Nessie catalog

presto-hive-common/src/main/java/com/facebook/presto/hive/HiveCommonClientConfig.java

@@ -46,6 +46,7 @@ public class HiveCommonClientConfig
     private boolean readNullMaskedParquetEncryptedValueEnabled;
     private boolean useParquetColumnNames;
     private boolean zstdJniDecompressionEnabled;
+    private String catalogName;
     private DataSize affinitySchedulingFileSectionSize = new DataSize(256, MEGABYTE);
 
     public NodeSelectionStrategy getNodeSelectionStrategy()
@@ -286,6 +287,19 @@ public HiveCommonClientConfig setZstdJniDecompressionEnabled(boolean zstdJniDeco
         return this;
     }
 
+    public String getCatalogName()
+    {
+        return catalogName;
+    }
+
+    @Config("hive.metastore.catalog.name")
+    @ConfigDescription("Specified property to store the metastore catalog name.")
+    public HiveCommonClientConfig setCatalogName(String catalogName)
+    {
+        this.catalogName = catalogName;
+        return this;
+    }
+
     @NotNull
     public DataSize getAffinitySchedulingFileSectionSize()
     {

presto-hive-common/src/main/java/com/facebook/presto/hive/MetadataUtils.java

@@ -29,6 +29,8 @@
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.Iterables;
 
+import javax.annotation.Nullable;
+
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
@@ -49,6 +51,10 @@
 
 public final class MetadataUtils
 {
+    private static final String CATALOG_DB_SEPARATOR = "#";
+    private static final String CATALOG_DB_THRIFT_NAME_MARKER = "@";
+    private static final String DB_EMPTY_MARKER = "!";
+    private static final String DEFAULT_DATABASE = "default";
     private MetadataUtils() {}
 
     public static Optional<DiscretePredicates> getDiscretePredicates(List<ColumnHandle> partitionColumns, List<HivePartition> partitions)
@@ -155,4 +161,26 @@ private static Domain buildColumnDomain(ColumnHandle column, List<HivePartition>
 
         return Domain.onlyNull(type);
     }
+
+    /**
+     * Constructs the schema name, including catalog name if applicable.
+     *
+     * @param schemaName the original schema name
+     * @return the formatted schema name (Example - @catalog_name#schema_name)
+     */
+    public static String constructSchemaName(Optional<String> catalogName, @Nullable String schemaName)
+    {
+        if (!catalogName.isPresent() || DEFAULT_DATABASE.equals(schemaName) ||
+                (schemaName != null && schemaName.contains(CATALOG_DB_SEPARATOR))) {
+            return schemaName;
+        }
+
+        StringBuilder catalogDatabaseName = new StringBuilder()
+                .append(CATALOG_DB_THRIFT_NAME_MARKER)
+                .append(catalogName.get()) // Safe since we checked isPresent()
+                .append(CATALOG_DB_SEPARATOR)
+                .append(schemaName == null ? "" : schemaName.isEmpty() ? DB_EMPTY_MARKER : schemaName);
+
+        return catalogDatabaseName.toString();
+    }
 }

presto-hive-common/src/test/java/com/facebook/presto/hive/TestHiveCommonClientConfig.java

@@ -49,6 +49,7 @@ public void testDefaults()
                 .setParquetBatchReaderVerificationEnabled(false)
                 .setParquetBatchReadOptimizationEnabled(false)
                 .setReadNullMaskedParquetEncryptedValue(false)
+                .setCatalogName(null)
                 .setAffinitySchedulingFileSectionSize(new DataSize(256, MEGABYTE)));
     }
 
@@ -74,6 +75,7 @@ public void testExplicitPropertyMappings()
                 .put("hive.enable-parquet-batch-reader-verification", "true")
                 .put("hive.parquet-batch-read-optimization-enabled", "true")
                 .put("hive.read-null-masked-parquet-encrypted-value-enabled", "true")
+                .put("hive.metastore.catalog.name", "catalogName")
                 .put("hive.affinity-scheduling-file-section-size", "512MB")
                 .build();
 
@@ -96,6 +98,7 @@ public void testExplicitPropertyMappings()
                 .setParquetBatchReaderVerificationEnabled(true)
                 .setParquetBatchReadOptimizationEnabled(true)
                 .setReadNullMaskedParquetEncryptedValue(true)
+                .setCatalogName("catalogName")
                 .setAffinitySchedulingFileSectionSize(new DataSize(512, MEGABYTE));
 
         ConfigAssertions.assertFullMapping(properties, expected);

presto-hive-common/src/test/java/com/facebook/presto/hive/TestMetadataUtils.java

@@ -0,0 +1,74 @@
+/*
+ * 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 com.facebook.presto.hive;
+
+import org.testng.annotations.Test;
+
+import java.util.Optional;
+
+import static com.facebook.presto.hive.MetadataUtils.constructSchemaName;
+import static org.testng.Assert.assertNull;
+import static org.testng.Assert.assertTrue;
+
+public class TestMetadataUtils
+{
+    @Test
+    public void testWithCatalogAndValidSchema()
+    {
+        String result = constructSchemaName(Optional.of("testCatalog"), "testSchema");
+        assertTrue(result.equals("@testCatalog#testSchema"));
+    }
+
+    @Test
+    public void testWithCatalogAndDefaultSchema()
+    {
+        String result = constructSchemaName(Optional.of("testCatalog"), "default");
+        assertTrue(result.equals("default"));
+    }
+
+    @Test
+    public void testWithCatalogAndSchemaContainingSeparator()
+    {
+        String result = constructSchemaName(Optional.of("testCatalog"), "schema#with#dot");
+        assertTrue(result.equals("schema#with#dot"));
+    }
+
+    @Test
+    public void testWithoutCatalog()
+    {
+        String result = constructSchemaName(Optional.empty(), "testSchema");
+        assertTrue(result.equals("testSchema"));
+    }
+
+    @Test
+    public void testWithNullSchema()
+    {
+        String result = constructSchemaName(Optional.empty(), null);
+        assertNull(result);
+    }
+
+    @Test
+    public void testWithoutCatalogNameAndEmptySchemaName()
+    {
+        String result = constructSchemaName(Optional.empty(), "");
+        assertTrue(result.isEmpty());
+    }
+
+    @Test
+    public void testWithCatalogNameAndEmptySchemaName()
+    {
+        String result = constructSchemaName(Optional.of("testCatalog"), "");
+        assertTrue(result.equals("@testCatalog#!"));
+    }
+}

presto-hive-hadoop2/src/test/java/com/facebook/presto/hive/s3select/S3SelectTestHelper.java

@@ -28,6 +28,7 @@
 import com.facebook.presto.hive.HiveCoercionPolicy;
 import com.facebook.presto.hive.HiveColumnConverterProvider;
 import com.facebook.presto.hive.HiveColumnHandle;
+import com.facebook.presto.hive.HiveCommonClientConfig;
 import com.facebook.presto.hive.HiveEncryptionInformationProvider;
 import com.facebook.presto.hive.HiveFileRenamer;
 import com.facebook.presto.hive.HiveHdfsConfiguration;
@@ -137,7 +138,7 @@ public S3SelectTestHelper(String host,
             metastoreClientConfig.setMetastoreSocksProxy(HostAndPort.fromString(proxy));
         }
 
-        HiveCluster hiveCluster = new TestingHiveCluster(metastoreClientConfig, thriftHiveMetastoreConfig, host, port);
+        HiveCluster hiveCluster = new TestingHiveCluster(metastoreClientConfig, thriftHiveMetastoreConfig, host, port, new HiveCommonClientConfig());
         executor = newCachedThreadPool(daemonThreadsNamed("hive-%s"));
         HivePartitionManager hivePartitionManager = new HivePartitionManager(FUNCTION_AND_TYPE_MANAGER, config);
 

presto-hive-metastore/src/main/java/com/facebook/presto/hive/metastore/Database.java

@@ -39,6 +39,7 @@ public class Database
     private final PrincipalType ownerType;
     private final Optional<String> comment;
     private final Map<String, String> parameters;
+    private final Optional<String> catalogName;
 
     @JsonCreator
     public Database(
@@ -47,14 +48,16 @@ public Database(
             @JsonProperty("ownerName") String ownerName,
             @JsonProperty("ownerType") PrincipalType ownerType,
             @JsonProperty("comment") Optional<String> comment,
-            @JsonProperty("parameters") Map<String, String> parameters)
+            @JsonProperty("parameters") Map<String, String> parameters,
+            @JsonProperty("catalogName") Optional<String> catalogName)
     {
         this.databaseName = requireNonNull(databaseName, "databaseName is null");
         this.location = requireNonNull(location, "location is null");
         this.ownerName = requireNonNull(ownerName, "ownerName is null");
         this.ownerType = requireNonNull(ownerType, "ownerType is null");
         this.comment = requireNonNull(comment, "comment is null");
         this.parameters = ImmutableMap.copyOf(requireNonNull(parameters, "parameters is null"));
+        this.catalogName = requireNonNull(catalogName, "catalogName is null");
     }
 
     @JsonProperty
@@ -103,6 +106,12 @@ public static Builder builder(Database database)
         return new Builder(database);
     }
 
+    @JsonProperty
+    public Optional<String> getCatalogName()
+    {
+        return catalogName;
+    }
+
     public static class Builder
     {
         private String databaseName;
@@ -111,6 +120,7 @@ public static class Builder
         private PrincipalType ownerType;
         private Optional<String> comment = Optional.empty();
         private Map<String, String> parameters = new LinkedHashMap<>();
+        private Optional<String> catalogName = Optional.empty();
 
         public Builder() {}
 
@@ -122,6 +132,7 @@ public Builder(Database database)
             this.ownerType = database.ownerType;
             this.comment = database.comment;
             this.parameters = database.parameters;
+            this.catalogName = database.catalogName;
         }
 
         public Builder setDatabaseName(String databaseName)
@@ -166,6 +177,12 @@ public Builder setParameters(Map<String, String> parameters)
             return this;
         }
 
+        public Builder setCatalogName(Optional<String> catalogName)
+        {
+            this.catalogName = requireNonNull(catalogName, "catalogName is null");
+            return this;
+        }
+
         public Database build()
         {
             return new Database(
@@ -174,7 +191,8 @@ public Database build()
                     ownerName,
                     ownerType,
                     comment,
-                    parameters);
+                    parameters,
+                    catalogName);
         }
     }
 
@@ -188,6 +206,7 @@ public String toString()
                 .add("ownerType", ownerType)
                 .add("comment", comment)
                 .add("parameters", parameters)
+                .add("catalogName", catalogName)
                 .toString();
     }
 
@@ -207,12 +226,13 @@ public boolean equals(Object o)
                 Objects.equals(ownerName, database.ownerName) &&
                 ownerType == database.ownerType &&
                 Objects.equals(comment, database.comment) &&
-                Objects.equals(parameters, database.parameters);
+                Objects.equals(parameters, database.parameters) &&
+                Objects.equals(catalogName, database.catalogName);
     }
 
     @Override
     public int hashCode()
     {
-        return Objects.hash(databaseName, location, ownerName, ownerType, comment, parameters);
+        return Objects.hash(databaseName, location, ownerName, ownerType, comment, parameters, catalogName);
     }
 }