Added new configuration to pass catalog name to metastore when creati…

presto-common/src/main/java/com/facebook/presto/common/Utils.java

@@ -20,6 +20,7 @@
 
 import javax.annotation.Nullable;
 
+import java.util.Optional;
 import java.util.function.Supplier;
 
 import static com.facebook.presto.common.type.TypeUtils.readNativeValue;
@@ -29,6 +30,10 @@
 
 public final class Utils
 {
+    private static final CharSequence CATALOG_DB_SEPARATOR = "#";
+    private static final Object CATALOG_DB_THRIFT_NAME_MARKER = "@";
+    private static final String DEFAULT_DATABASE = "default";
+
     private Utils()
     {
     }
@@ -126,4 +131,23 @@ public static void checkState(boolean test, String errorMessage)
             throw new IllegalStateException(errorMessage);
         }
     }
+
+    /**
+     * 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, String schemaName)
+    {
+        if (catalogName.isPresent() && schemaName != null && !schemaName.equals(DEFAULT_DATABASE) && !schemaName.contains(CATALOG_DB_SEPARATOR)) {
+            return String.format(
+                    "%s%s%s%s",
+                    CATALOG_DB_THRIFT_NAME_MARKER,
+                    catalogName.get(),
+                    CATALOG_DB_SEPARATOR,
+                    schemaName);
+        }
+        return schemaName;
+    }
 }

presto-delta/src/main/java/com/facebook/presto/delta/DeltaMetadata.java

@@ -16,6 +16,7 @@
 import com.facebook.airlift.log.Logger;
 import com.facebook.presto.common.predicate.TupleDomain;
 import com.facebook.presto.common.type.TypeManager;
+import com.facebook.presto.hive.HiveCommonClientConfig;
 import com.facebook.presto.hive.HiveStorageFormat;
 import com.facebook.presto.hive.metastore.ExtendedHiveMetastore;
 import com.facebook.presto.hive.metastore.HivePrivilegeInfo;
@@ -53,6 +54,7 @@
 import java.util.Set;
 import java.util.stream.Collectors;
 
+import static com.facebook.presto.common.Utils.constructSchemaName;
 import static com.facebook.presto.delta.DeltaColumnHandle.ColumnType.PARTITION;
 import static com.facebook.presto.delta.DeltaColumnHandle.ColumnType.REGULAR;
 import static com.facebook.presto.delta.DeltaExpressionUtils.splitPredicate;
@@ -87,27 +89,36 @@ public class DeltaMetadata
     private final ExtendedHiveMetastore metastore;
     private final TypeManager typeManager;
     private final DeltaConfig config;
+    private final String catalogName;
 
     @Inject
     public DeltaMetadata(
             DeltaConnectorId connectorId,
             DeltaClient deltaClient,
             ExtendedHiveMetastore metastore,
             TypeManager typeManager,
-            DeltaConfig config)
+            DeltaConfig config, HiveCommonClientConfig commonConfig)
     {
         this.connectorId = requireNonNull(connectorId, "connectorId is null").toString();
         this.deltaClient = requireNonNull(deltaClient, "deltaClient is null");
         this.metastore = requireNonNull(metastore, "metastore is null");
         this.typeManager = requireNonNull(typeManager, "typeManager is null");
         this.config = requireNonNull(config, "config is null");
+        this.catalogName = commonConfig.getCatalogName();
     }
 
     @Override
     public List<String> listSchemaNames(ConnectorSession session)
     {
         ArrayList<String> schemas = new ArrayList<>();
-        schemas.addAll(metastore.getAllDatabases(metastoreContext(session)));
+        List<String> schemaNames;
+        if (catalogName != null) {
+            schemaNames = metastore.getDatabases(metastoreContext(session), constructSchemaName(Optional.of(catalogName), ""));
+        }
+        else {
+            schemaNames = metastore.getAllDatabases(metastoreContext(session));
+        }
+        schemas.addAll(schemaNames);
         schemas.add(PATH_SCHEMA.toLowerCase(US));
         return schemas;
     }
@@ -131,15 +142,16 @@ public void dropTable(ConnectorSession session, ConnectorTableHandle tableHandle
     {
         DeltaTableHandle handle = (DeltaTableHandle) tableHandle;
         MetastoreContext metastoreContext = metastoreContext(session);
+        String schemaName = constructSchemaName(Optional.ofNullable(catalogName), handle.getDeltaTable().getSchemaName());
 
-        Optional<Table> target = metastore.getTable(metastoreContext, handle.getDeltaTable().getSchemaName(), handle.getDeltaTable().getTableName());
+        Optional<Table> target = metastore.getTable(metastoreContext, schemaName, handle.getDeltaTable().getTableName());
         if (!target.isPresent()) {
             throw new TableNotFoundException(handle.toSchemaTableName());
         }
 
         metastore.dropTable(
                 metastoreContext,
-                handle.getDeltaTable().getSchemaName(),
+                schemaName,
                 handle.getDeltaTable().getTableName(),
                 false);
     }
@@ -164,6 +176,7 @@ private Table prepareTable(ConnectorSession session, ConnectorTableMetadata tabl
         }
 
         Table.Builder tableBuilder = Table.builder()
+                .setCatalogName(Optional.ofNullable(catalogName))
                 .setDatabaseName(schemaName)
                 .setTableName(tableName)
                 .setOwner(session.getUser())
@@ -195,7 +208,7 @@ public DeltaTableHandle getTableHandle(ConnectorSession session, SchemaTableName
             tableLocation = deltaTableName.getTableNameOrPath();
         }
         else {
-            Optional<Table> metastoreTable = metastore.getTable(metastoreContext(session), schemaName, deltaTableName.getTableNameOrPath());
+            Optional<Table> metastoreTable = metastore.getTable(metastoreContext(session), constructSchemaName(Optional.ofNullable(catalogName), schemaName), deltaTableName.getTableNameOrPath());
             if (!metastoreTable.isPresent()) {
                 return null; // indicates table doesn't exist
             }
@@ -284,7 +297,7 @@ public List<SchemaTableName> listTables(ConnectorSession session, Optional<Strin
                 .orElseGet(() -> listSchemaNames(session));
         ImmutableList.Builder<SchemaTableName> tableNames = ImmutableList.builder();
         for (String schema : schemaNames) {
-            for (String tableName : metastore.getAllTables(metastoreContext(session), schema).orElse(emptyList())) {
+            for (String tableName : metastore.getAllTables(metastoreContext(session), constructSchemaName(Optional.ofNullable(catalogName), schema)).orElse(emptyList())) {
                 tableNames.add(new SchemaTableName(schema, tableName));
             }
         }

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

@@ -43,6 +43,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``                 Enable passing the catalog name to store the metastore.
 =============================================== ========================================================= ============
 
 Delta Lake connector reuses many of the modules existing in Hive connector.

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

@@ -200,6 +200,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``                   Enable passing the catalog name to store 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``         Enable passing the catalog name to store
+                                        the metastore.
 ======================================= ============================================= ===========
 
 SQL Support

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

@@ -85,6 +85,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``                          Enable passing the catalog name to store 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;
 
     public NodeSelectionStrategy getNodeSelectionStrategy()
     {
@@ -284,4 +285,17 @@ public HiveCommonClientConfig setZstdJniDecompressionEnabled(boolean zstdJniDeco
         this.zstdJniDecompressionEnabled = zstdJniDecompressionEnabled;
         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;
+    }
 }

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

@@ -47,7 +47,8 @@ public void testDefaults()
                 .setZstdJniDecompressionEnabled(false)
                 .setParquetBatchReaderVerificationEnabled(false)
                 .setParquetBatchReadOptimizationEnabled(false)
-                .setReadNullMaskedParquetEncryptedValue(false));
+                .setReadNullMaskedParquetEncryptedValue(false)
+                .setCatalogName(null));
     }
 
     @Test
@@ -72,6 +73,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")
                 .build();
 
         HiveCommonClientConfig expected = new HiveCommonClientConfig()
@@ -92,7 +94,8 @@ public void testExplicitPropertyMappings()
                 .setZstdJniDecompressionEnabled(true)
                 .setParquetBatchReaderVerificationEnabled(true)
                 .setParquetBatchReadOptimizationEnabled(true)
-                .setReadNullMaskedParquetEncryptedValue(true);
+                .setReadNullMaskedParquetEncryptedValue(true)
+                .setCatalogName("catalogName");
 
         ConfigAssertions.assertFullMapping(properties, expected);
     }

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;
@@ -177,7 +178,9 @@ public S3SelectTestHelper(String host,
                 new HiveFileRenamer(),
                 columnConverterProvider,
                 new QuickStatsProvider(metastoreClient, hdfsEnvironment, DO_NOTHING_DIRECTORY_LISTER, new HiveClientConfig(), new NamenodeStats(), ImmutableList.of()),
-                new HiveTableWritabilityChecker(config));
+                new HiveTableWritabilityChecker(config),
+                new HiveCommonClientConfig());
+
         transactionManager = new HiveTransactionManager();
         splitManager = new HiveSplitManager(
                 transactionManager,

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,13 @@ public Builder setParameters(Map<String, String> parameters)
             return this;
         }
 
+        public Builder setCatalogName(Optional<String> catalogName)
+        {
+            requireNonNull(catalogName, "catalogName is null");
+            this.catalogName = catalogName;
+            return this;
+        }
+
         public Database build()
         {
             return new Database(
@@ -174,7 +192,8 @@ public Database build()
                     ownerName,
                     ownerType,
                     comment,
-                    parameters);
+                    parameters,
+                    catalogName);
         }
     }
 
@@ -188,6 +207,7 @@ public String toString()
                 .add("ownerType", ownerType)
                 .add("comment", comment)
                 .add("parameters", parameters)
+                .add("catalogName", catalogName)
                 .toString();
     }
 
@@ -207,12 +227,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);
     }
 }

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

@@ -34,6 +34,11 @@
 
 public interface ExtendedHiveMetastore
 {
+    default List<String> getDatabases(MetastoreContext metastoreContext, String pattern)
+    {
+        return ImmutableList.of();
+    }
+
     Optional<Database> getDatabase(MetastoreContext metastoreContext, String databaseName);
 
     List<String> getAllDatabases(MetastoreContext metastoreContext);