Added catalog name in partition object

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

@@ -35,6 +35,7 @@
 @Immutable
 public class Partition
 {
+    private final Optional<String> catalogName;
     private final String databaseName;
     private final String tableName;
     private final List<String> values;
@@ -50,6 +51,7 @@ public class Partition
 
     @JsonCreator
     public Partition(
+            @JsonProperty("catalogName") Optional<String> catalogName,
             @JsonProperty("databaseName") String databaseName,
             @JsonProperty("tableName") String tableName,
             @JsonProperty("values") List<String> values,
@@ -63,6 +65,7 @@ public Partition(
             @JsonProperty("lastDataCommitTime") long lastDataCommitTime,
             @JsonProperty("rowIdPartitionComponent") Optional<byte[]> rowIdPartitionComponent)
     {
+        this.catalogName = requireNonNull(catalogName, "catalogName is null");
         this.databaseName = requireNonNull(databaseName, "databaseName is null");
         this.tableName = requireNonNull(tableName, "tableName is null");
         this.values = ImmutableList.copyOf(requireNonNull(values, "values is null"));
@@ -77,6 +80,11 @@ public Partition(
         this.rowIdPartitionComponent = requireNonNull(rowIdPartitionComponent);
     }
 
+    @JsonProperty
+    public Optional<String> getCatalogName()
+    {
+        return catalogName;
+    }
     @JsonProperty
     public String getDatabaseName()
     {
@@ -184,7 +192,8 @@ public boolean equals(Object o)
         }
 
         Partition partition = (Partition) o;
-        return Objects.equals(databaseName, partition.databaseName) &&
+        return Objects.equals(catalogName, partition.catalogName) &&
+                Objects.equals(databaseName, partition.databaseName) &&
                 Objects.equals(tableName, partition.tableName) &&
                 Objects.equals(values, partition.values) &&
                 Objects.equals(storage, partition.storage) &&
@@ -200,7 +209,7 @@ public boolean equals(Object o)
     @Override
     public int hashCode()
     {
-        return Objects.hash(databaseName, tableName, values, storage, columns, parameters, partitionVersion, eligibleToIgnore, sealedPartition, createTime, lastDataCommitTime);
+        return Objects.hash(catalogName, databaseName, tableName, values, storage, columns, parameters, partitionVersion, eligibleToIgnore, sealedPartition, createTime, lastDataCommitTime);
     }
 
     public static Builder builder()
@@ -216,6 +225,7 @@ public static Builder builder(Partition partition)
     public static class Builder
     {
         private final Storage.Builder storageBuilder;
+        private Optional<String> catalogName;
         private String databaseName;
         private String tableName;
         private List<String> values;
@@ -236,6 +246,7 @@ private Builder()
         private Builder(Partition partition)
         {
             this.storageBuilder = Storage.builder(partition.getStorage());
+            this.catalogName = partition.getCatalogName();
             this.databaseName = partition.getDatabaseName();
             this.tableName = partition.getTableName();
             this.values = partition.getValues();
@@ -248,6 +259,11 @@ private Builder(Partition partition)
             this.rowIdPartitionComponent = partition.getRowIdPartitionComponent();
         }
 
+        public Builder setCatalogName(Optional<String> catalogName)
+        {
+            this.catalogName = catalogName;
+            return this;
+        }
         public Builder setDatabaseName(String databaseName)
         {
             this.databaseName = databaseName;
@@ -332,7 +348,7 @@ public Builder setRowIdPartitionComponent(Optional<byte[]> rowIdPartitionCompone
 
         public Partition build()
         {
-            return new Partition(databaseName, tableName, values, storageBuilder.build(), columns, parameters, partitionVersion, isEligibleToIgnore, isSealedPartition, createTime, lastDataCommitTime, rowIdPartitionComponent);
+            return new Partition(catalogName, databaseName, tableName, values, storageBuilder.build(), columns, parameters, partitionVersion, isEligibleToIgnore, isSealedPartition, createTime, lastDataCommitTime, rowIdPartitionComponent);
         }
     }
 }

presto-hive-metastore/src/main/java/com/facebook/presto/hive/metastore/file/PartitionMetadata.java

@@ -222,6 +222,7 @@ public PartitionMetadata withColumnStatistics(Map<String, HiveColumnStatistics>
     public Partition toPartition(String databaseName, String tableName, List<String> values, String location)
     {
         return new Partition(
+                Optional.empty(),
                 databaseName,
                 tableName,
                 values,

presto-hive-metastore/src/main/java/com/facebook/presto/hive/metastore/thrift/ThriftMetastoreUtil.java

@@ -373,6 +373,7 @@ public static org.apache.hadoop.hive.metastore.api.Partition toMetastoreApiParti
     public static org.apache.hadoop.hive.metastore.api.Partition toMetastoreApiPartition(Partition partition, ColumnConverter columnConverter)
     {
         org.apache.hadoop.hive.metastore.api.Partition result = new org.apache.hadoop.hive.metastore.api.Partition();
+        partition.getCatalogName().ifPresent(result::setCatName);
         result.setDbName(partition.getDatabaseName());
         result.setTableName(partition.getTableName());
         result.setValues(partition.getValues());
@@ -508,6 +509,7 @@ public static Partition fromMetastoreApiPartition(org.apache.hadoop.hive.metasto
         }
 
         Partition.Builder partitionBuilder = Partition.builder()
+                .setCatalogName(Optional.ofNullable(partition.getCatName()))
                 .setDatabaseName(partition.getDbName())
                 .setTableName(partition.getTableName())
                 .setValues(partition.getValues())

presto-hive-metastore/src/test/java/com/facebook/presto/hive/metastore/TestRecordingHiveMetastore.java

@@ -109,6 +109,7 @@ public class TestRecordingHiveMetastore
             Optional.of("original_text"),
             Optional.of("expanded_text"));
     private static final Partition PARTITION = new Partition(
+            Optional.of("catalog"),
             "database",
             "table",
             ImmutableList.of("value"),

presto-hive/src/main/java/com/facebook/presto/hive/HiveMetadata.java

@@ -1065,7 +1065,7 @@ else if (tableType.equals(MANAGED_TABLE) || tableType.equals(MATERIALIZED_VIEW))
 
         return buildTableObject(
                 session.getQueryId(),
-                Optional.of(catalogName),
+                Optional.ofNullable(catalogName),
                 schemaName,
                 tableName,
                 session.getUser(),

presto-hive/src/main/java/com/facebook/presto/hive/HivePartitionObjectBuilder.java

@@ -47,6 +47,7 @@ public Partition buildPartitionObject(
                         param -> extraParametersBuilder.put("user_supplied", param));
 
         return Partition.builder()
+                .setCatalogName(table.getCatalogName())
                 .setDatabaseName(table.getDatabaseName())
                 .setTableName(table.getTableName())
                 .setColumns(table.getDataColumns())

presto-hive/src/test/java/com/facebook/presto/hive/TestAbstractDwrfEncryptionInformationSource.java

@@ -110,8 +110,8 @@ public void testGetReadEncryptionInformationForPartitionedTableWithTableLevelEnc
                                 ImmutableList.of(new Subfield("col_struct.a"), new Subfield("col_struct.b.b2")),
                                 Optional.empty()))),
                 ImmutableMap.of(
-                        "ds=2020-01-01", new Partition("dbName", "tableName", ImmutableList.of("2020-01-01"), table.getStorage(), table.getDataColumns(), ImmutableMap.of(), Optional.empty(), false, true, 0, 0, Optional.empty()),
-                        "ds=2020-01-02", new Partition("dbName", "tableName", ImmutableList.of("2020-01-02"), table.getStorage(), table.getDataColumns(), ImmutableMap.of(), Optional.empty(), false, true, 0, 0, Optional.empty())));
+                        "ds=2020-01-01", new Partition(Optional.of("catalogname"), "dbName", "tableName", ImmutableList.of("2020-01-01"), table.getStorage(), table.getDataColumns(), ImmutableMap.of(), Optional.empty(), false, true, 0, 0, Optional.empty()),
+                        "ds=2020-01-02", new Partition(Optional.of("catalogname"), "dbName", "tableName", ImmutableList.of("2020-01-02"), table.getStorage(), table.getDataColumns(), ImmutableMap.of(), Optional.empty(), false, true, 0, 0, Optional.empty())));
 
         assertTrue(encryptionInformation.isPresent());
         assertEquals(
@@ -130,8 +130,8 @@ public void testGetReadEncryptionInformationForPartitionedTableWithTableLevelEnc
                 table,
                 Optional.of(ImmutableSet.of()),
                 ImmutableMap.of(
-                        "ds=2020-01-01", new Partition("dbName", "tableName", ImmutableList.of("2020-01-01"), table.getStorage(), table.getDataColumns(), ImmutableMap.of(), Optional.empty(), false, true, 0, 0, Optional.empty()),
-                        "ds=2020-01-02", new Partition("dbName", "tableName", ImmutableList.of("2020-01-02"), table.getStorage(), table.getDataColumns(), ImmutableMap.of(), Optional.empty(), false, true, 0, 0, Optional.empty())));
+                        "ds=2020-01-01", new Partition(Optional.of("catalogname"), "dbName", "tableName", ImmutableList.of("2020-01-01"), table.getStorage(), table.getDataColumns(), ImmutableMap.of(), Optional.empty(), false, true, 0, 0, Optional.empty()),
+                        "ds=2020-01-02", new Partition(Optional.of("catalogname"), "dbName", "tableName", ImmutableList.of("2020-01-02"), table.getStorage(), table.getDataColumns(), ImmutableMap.of(), Optional.empty(), false, true, 0, 0, Optional.empty())));
 
         assertTrue(encryptionInformation.isPresent());
         assertEquals(
@@ -162,8 +162,8 @@ public void testGetReadEncryptionInformationForPartitionedTableWithColumnLevelEn
                                 ImmutableList.of(new Subfield("col_struct.a"), new Subfield("col_struct.b.b2")),
                                 Optional.empty()))),
                 ImmutableMap.of(
-                        "ds=2020-01-01", new Partition("dbName", "tableName", ImmutableList.of("2020-01-01"), table.getStorage(), table.getDataColumns(), ImmutableMap.of(), Optional.empty(), false, true, 0, 0, Optional.empty()),
-                        "ds=2020-01-02", new Partition("dbName", "tableName", ImmutableList.of("2020-01-02"), table.getStorage(), table.getDataColumns(), ImmutableMap.of(), Optional.empty(), false, true, 0, 0, Optional.empty())));
+                        "ds=2020-01-01", new Partition(Optional.of("catalogname"), "dbName", "tableName", ImmutableList.of("2020-01-01"), table.getStorage(), table.getDataColumns(), ImmutableMap.of(), Optional.empty(), false, true, 0, 0, Optional.empty()),
+                        "ds=2020-01-02", new Partition(Optional.of("catalogname"), "dbName", "tableName", ImmutableList.of("2020-01-02"), table.getStorage(), table.getDataColumns(), ImmutableMap.of(), Optional.empty(), false, true, 0, 0, Optional.empty())));
 
         Map<String, byte[]> expectedFieldToKeyData = ImmutableMap.of("col_bigint", "key2".getBytes(), "col_struct.a", "key2".getBytes(), "col_struct.b.b2", "key1".getBytes());
         assertTrue(encryptionInformation.isPresent());

presto-hive/src/test/java/com/facebook/presto/hive/TestBackgroundHiveSplitLoader.java

@@ -259,6 +259,7 @@ public void testUnsupportedTableFormat()
     private static Partition orcPartition()
     {
         return new Partition(
+                Optional.of("catalogname"),
                 "test_db",
                 "test_table",
                 PARTITION_COLUMNS.stream().map(x -> "Partition_Value").collect(Collectors.toList()),

presto-hive/src/test/java/com/facebook/presto/hive/TestHiveSplitManager.java

@@ -475,6 +475,7 @@ private void assertRedundantColumnDomains(Range predicateRange, PartitionStatist
         // Prepare partition with stats
         PartitionWithStatistics partitionWithStatistics = new PartitionWithStatistics(
                 new Partition(
+                        Optional.of("catalogname"),
                         "test_db",
                         "test_table",
                         ImmutableList.of(PARTITION_VALUE),
@@ -621,6 +622,7 @@ public void testEncryptionInformation()
                         DWRF_ENCRYPTION_PROVIDER_KEY, testEncryptionProvider));
         PartitionWithStatistics partitionWithStatistics = new PartitionWithStatistics(
                 new Partition(
+                        Optional.of("catalogname"),
                         "test_db",
                         "test_table",
                         ImmutableList.of(PARTITION_VALUE),

presto-hive/src/test/java/com/facebook/presto/hive/s3select/TestS3SelectPushdown.java

@@ -75,7 +75,7 @@ public void setUp()
                 .setLocation("location")
                 .build();
 
-        partition = new Partition("db", "table", emptyList(), storage, singletonList(column), emptyMap(), Optional.empty(), false, false, 1234, 4567L, Optional.empty());
+        partition = new Partition(Optional.of("catalogname"), "db", "table", emptyList(), storage, singletonList(column), emptyMap(), Optional.empty(), false, false, 1234, 4567L, Optional.empty());
 
         table = new Table(
                 Optional.of("catalog"),
@@ -148,7 +148,7 @@ public void testShouldNotEnableSelectPushdownWhenIsNotSupportedSerde()
 
         assertFalse(shouldEnablePushdownForTable(session, newTable, "s3://fakeBucket/fakeObject", Optional.empty()));
 
-        Partition newPartition = new Partition("db", "table", emptyList(), newStorage, singletonList(column), emptyMap(), Optional.empty(), false, false, 1234, 4567L, Optional.empty());
+        Partition newPartition = new Partition(Optional.of("catalogname"), "db", "table", emptyList(), newStorage, singletonList(column), emptyMap(), Optional.empty(), false, false, 1234, 4567L, Optional.empty());
         assertFalse(shouldEnablePushdownForTable(session, newTable, "s3://fakeBucket/fakeObject", Optional.of(newPartition)));
     }
 
@@ -192,7 +192,7 @@ public void testShouldNotEnableSelectPushdownWhenColumnTypesAreNotSupported()
                 Optional.empty());
         assertFalse(shouldEnablePushdownForTable(session, newTable, "s3://fakeBucket/fakeObject", Optional.empty()));
 
-        Partition newPartition = new Partition("db", "table", emptyList(), storage, singletonList(column), emptyMap(), Optional.empty(), false, false, 1234, 4567L, Optional.empty());
+        Partition newPartition = new Partition(Optional.of("catalogname"), "db", "table", emptyList(), storage, singletonList(column), emptyMap(), Optional.empty(), false, false, 1234, 4567L, Optional.empty());
         assertFalse(shouldEnablePushdownForTable(session, newTable, "s3://fakeBucket/fakeObject", Optional.of(newPartition)));
     }
 

presto-hive/src/test/java/com/facebook/presto/hive/statistics/TestQuickStatsProvider.java

@@ -148,6 +148,7 @@ public void setUp()
                 ImmutableMap.of(),
                 ImmutableMap.of());
         Partition mockPartition = new Partition(
+                Optional.of("catalogname"),
                 TEST_SCHEMA,
                 TEST_TABLE,
                 ImmutableList.of(),

presto-iceberg/src/main/java/com/facebook/presto/iceberg/IcebergHiveMetadata.java

@@ -227,7 +227,7 @@ private Optional<Table> getHiveTable(ConnectorSession session, SchemaTableName s
         IcebergTableName name = IcebergTableName.from(schemaTableName.getTableName());
         try {
             return tableCache.get(schemaTableName, () ->
-                    metastore.getTable(getMetastoreContext(session), schemaTableName.getSchemaName(), name.getTableName()));
+                    metastore.getTable(getMetastoreContext(session), constructSchemaName(schemaTableName.getSchemaName()), name.getTableName()));
         }
         catch (UncheckedExecutionException e) {
             throwIfInstanceOf(e.getCause(), PrestoException.class);
@@ -637,7 +637,7 @@ public void registerTable(ConnectorSession clientSession, SchemaTableName schema
         }
 
         Table.Builder builder = Table.builder()
-                .setCatalogName(Optional.of(catalogName))
+                .setCatalogName(Optional.ofNullable(catalogName))
                 .setDatabaseName(schemaTableName.getSchemaName())
                 .setTableName(schemaTableName.getTableName())
                 .setOwner(clientSession.getUser())

presto-iceberg/src/test/java/com/facebook/presto/iceberg/hive/TestIcebergHiveStatistics.java

@@ -518,7 +518,7 @@ private Table loadTable(String tableName)
                 getHdfsEnvironment(),
                 new IcebergHiveTableOperationsConfig(),
                 getQueryRunner().getDefaultSession().toConnectorSession(connectorId),
-                SchemaTableName.valueOf("tpch." + tableName));
+                SchemaTableName.valueOf("tpch." + tableName), "hive");
     }
 
     protected ExtendedHiveMetastore getFileHiveMetastore()