[DRAFT] Add changes to populate the data source metadata details

presto-base-jdbc/src/main/java/com/facebook/presto/plugin/jdbc/JdbcInputInfo.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 com.facebook.presto.plugin.jdbc;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+import static java.util.Objects.requireNonNull;
+
+public class JdbcInputInfo
+{
+    private final String tableLocation;
+    private final String tableName;
+
+    public JdbcInputInfo(
+            @JsonProperty("tableLocation") String tableLocation,
+            @JsonProperty("tableName") String tableName)
+
+    {
+        this.tableLocation = requireNonNull(tableLocation, "tableLocation is null");
+        this.tableName = requireNonNull(tableName, "tableName is null");
+    }
+
+    @JsonProperty
+    public String getTableLocation()
+    {
+        return tableLocation;
+    }
+}

presto-base-jdbc/src/main/java/com/facebook/presto/plugin/jdbc/JdbcMetadata.java

@@ -55,14 +55,16 @@ public class JdbcMetadata
     private final JdbcMetadataCache jdbcMetadataCache;
     private final JdbcClient jdbcClient;
     private final boolean allowDropTable;
+    private final BaseJdbcConfig baseJdbcConfig;
 
     private final AtomicReference<Runnable> rollbackAction = new AtomicReference<>();
 
-    public JdbcMetadata(JdbcMetadataCache jdbcMetadataCache, JdbcClient jdbcClient, boolean allowDropTable)
+    public JdbcMetadata(JdbcMetadataCache jdbcMetadataCache, JdbcClient jdbcClient, boolean allowDropTable, BaseJdbcConfig baseJdbcConfig)
     {
         this.jdbcMetadataCache = requireNonNull(jdbcMetadataCache, "jdbcMetadataCache is null");
         this.jdbcClient = requireNonNull(jdbcClient, "client is null");
         this.allowDropTable = allowDropTable;
+        this.baseJdbcConfig = requireNonNull(baseJdbcConfig, "baseJdbcConfig is null");
     }
 
     @Override
@@ -188,8 +190,9 @@ public Optional<ConnectorOutputMetadata> finishCreateTable(ConnectorSession sess
     {
         JdbcOutputTableHandle handle = (JdbcOutputTableHandle) tableHandle;
         jdbcClient.commitCreateTable(session, JdbcIdentity.from(session), handle);
+        String url = baseJdbcConfig.getConnectionUrl();
         clearRollback();
-        return Optional.empty();
+        return Optional.of(new JdbcOutputMetadata(url));
     }
 
     private void setRollback(Runnable action)
@@ -267,4 +270,12 @@ public TableStatistics getTableStatistics(ConnectorSession session, ConnectorTab
         List<JdbcColumnHandle> columns = columnHandles.stream().map(JdbcColumnHandle.class::cast).collect(Collectors.toList());
         return jdbcClient.getTableStatistics(session, handle, columns, constraint.getSummary());
     }
+
+    @Override
+    public Optional<Object> getInfo(ConnectorTableLayoutHandle tableHandle)
+    {
+        JdbcTableLayoutHandle jdbcTableLayoutHandle = (JdbcTableLayoutHandle) tableHandle;
+        String tableLocation = baseJdbcConfig.getConnectionUrl();
+        return Optional.of(new JdbcInputInfo(tableLocation, jdbcTableLayoutHandle.getTable().getTableName()));
+    }
 }

presto-base-jdbc/src/main/java/com/facebook/presto/plugin/jdbc/JdbcMetadataFactory.java

@@ -22,18 +22,20 @@ public class JdbcMetadataFactory
     private final JdbcMetadataCache jdbcMetadataCache;
     private final JdbcClient jdbcClient;
     private final boolean allowDropTable;
+    private final BaseJdbcConfig baseJdbcConfig;
 
     @Inject
-    public JdbcMetadataFactory(JdbcMetadataCache jdbcMetadataCache, JdbcClient jdbcClient, JdbcMetadataConfig config)
+    public JdbcMetadataFactory(JdbcMetadataCache jdbcMetadataCache, JdbcClient jdbcClient, JdbcMetadataConfig config, BaseJdbcConfig baseJdbcConfig)
     {
         this.jdbcMetadataCache = requireNonNull(jdbcMetadataCache, "jdbcMetadataCache is null");
         this.jdbcClient = requireNonNull(jdbcClient, "jdbcClient is null");
         requireNonNull(config, "config is null");
         this.allowDropTable = config.isAllowDropTable();
+        this.baseJdbcConfig = requireNonNull(baseJdbcConfig, "baseJdbcConfig is null");
     }
 
     public JdbcMetadata create()
     {
-        return new JdbcMetadata(jdbcMetadataCache, jdbcClient, allowDropTable);
+        return new JdbcMetadata(jdbcMetadataCache, jdbcClient, allowDropTable, baseJdbcConfig);
     }
 }

presto-iceberg/src/main/java/com/facebook/presto/iceberg/IcebergWrittenPartitions.java → presto-base-jdbc/src/main/java/com/facebook/presto/plugin/jdbc/JdbcOutputMetadata.java

@@ -11,31 +11,30 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package com.facebook.presto.iceberg;
+
+package com.facebook.presto.plugin.jdbc;
 
 import com.facebook.presto.spi.connector.ConnectorOutputMetadata;
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonProperty;
-import com.google.common.collect.ImmutableList;
-
-import java.util.List;
 
 import static java.util.Objects.requireNonNull;
 
-public class IcebergWrittenPartitions
+public class JdbcOutputMetadata
         implements ConnectorOutputMetadata
 {
-    private final List<String> partitionNames;
+    private final String tableLocation;
 
     @JsonCreator
-    public IcebergWrittenPartitions(@JsonProperty("partitionNames") List<String> partitionNames)
+    public JdbcOutputMetadata(@JsonProperty("tableLocation") String tableLocation)
     {
-        this.partitionNames = ImmutableList.copyOf(requireNonNull(partitionNames, "partitionNames is null"));
+        this.tableLocation = requireNonNull(tableLocation, "tableLocation is null");
     }
 
     @JsonProperty
-    public List<String> getInfo()
+    @Override
+    public String getInfo()
     {
-        return partitionNames;
+        return tableLocation;
     }
 }

presto-base-jdbc/src/test/java/com/facebook/presto/plugin/jdbc/TestJdbcMetadata.java

@@ -66,7 +66,7 @@ public void setUp()
         database = new TestingDatabase();
         ListeningExecutorService executor = listeningDecorator(newCachedThreadPool(daemonThreadsNamed("test-%s")));
         jdbcMetadataCache = new JdbcMetadataCache(executor, database.getJdbcClient(), new JdbcMetadataCacheStats(), OptionalLong.of(0), OptionalLong.of(0), 100);
-        metadata = new JdbcMetadata(jdbcMetadataCache, database.getJdbcClient(), false);
+        metadata = new JdbcMetadata(jdbcMetadataCache, database.getJdbcClient(), false, new BaseJdbcConfig());
         tableHandle = metadata.getTableHandle(SESSION, new SchemaTableName("example", "numbers"));
     }
 
@@ -261,7 +261,7 @@ public void testDropTableTable()
             assertEquals(e.getErrorCode(), PERMISSION_DENIED.toErrorCode());
         }
 
-        metadata = new JdbcMetadata(jdbcMetadataCache, database.getJdbcClient(), true);
+        metadata = new JdbcMetadata(jdbcMetadataCache, database.getJdbcClient(), true, new BaseJdbcConfig());
         metadata.dropTable(SESSION, tableHandle);
 
         try {

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

@@ -0,0 +1,76 @@
+/*
+ * 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 com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.google.common.collect.ImmutableList;
+
+import java.util.List;
+import java.util.Objects;
+
+import static java.util.Objects.requireNonNull;
+
+public class HiveOutputInfo
+{
+    private final List<String> partitionNames;
+    private final String tableLocation;
+
+    @JsonCreator
+    public HiveOutputInfo(
+            @JsonProperty("partitionNames") List<String> partitionNames,
+            @JsonProperty("tableLocation") String tableLocation)
+    {
+        this.partitionNames = ImmutableList.copyOf(requireNonNull(partitionNames, "partitionNames is null"));
+        this.tableLocation = requireNonNull(tableLocation, "tableLocation is null");
+    }
+
+    @JsonProperty
+    public List<String> getPartitionNames()
+    {
+        return partitionNames;
+    }
+
+    @JsonProperty
+    public String getTableLocation()
+    {
+        return tableLocation;
+    }
+
+    @Override
+    public boolean equals(Object o)
+    {
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        HiveOutputInfo that = (HiveOutputInfo) o;
+        return Objects.equals(partitionNames, that.partitionNames) && Objects.equals(tableLocation, that.tableLocation);
+    }
+
+    @Override
+    public int hashCode()
+    {
+        return Objects.hash(partitionNames, tableLocation);
+    }
+
+    @Override
+    public String toString()
+    {
+        return "HiveOutputInfo{" +
+                "partitionNames=" + partitionNames +
+                ", tableLocation='" + tableLocation + '\'' +
+                '}';
+    }
+}

presto-hive/src/main/java/com/facebook/presto/hive/HiveWrittenPartitions.java → presto-hive-common/src/main/java/com/facebook/presto/hive/HiveWrittenPartitions.java

@@ -16,26 +16,24 @@
 import com.facebook.presto.spi.connector.ConnectorOutputMetadata;
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonProperty;
-import com.google.common.collect.ImmutableList;
-
-import java.util.List;
 
 import static java.util.Objects.requireNonNull;
 
 public class HiveWrittenPartitions
         implements ConnectorOutputMetadata
 {
-    private final List<String> partitionNames;
+    private final HiveOutputInfo hiveOutputInfo;
 
     @JsonCreator
-    public HiveWrittenPartitions(@JsonProperty("partitionNames") List<String> partitionNames)
+    public HiveWrittenPartitions(@JsonProperty("hiveOutputInfo") HiveOutputInfo hiveOutputInfo)
     {
-        this.partitionNames = ImmutableList.copyOf(requireNonNull(partitionNames, "partitionNames is null"));
+        this.hiveOutputInfo = requireNonNull(hiveOutputInfo, "hiveOutputInfo is null");
     }
 
     @JsonProperty
-    public List<String> getInfo()
+    @Override
+    public HiveOutputInfo getInfo()
     {
-        return partitionNames;
+        return hiveOutputInfo;
     }
 }

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

@@ -24,14 +24,17 @@ public class HiveInputInfo
     // Code that serialize HiveInputInfo into log would often need the ability to limit the length of log entries.
     // This boolean field allows such code to mark the log entry as length limited.
     private final boolean truncated;
+    private final String tableLocation;
 
     @JsonCreator
     public HiveInputInfo(
             @JsonProperty("partitionIds") List<String> partitionIds,
-            @JsonProperty("truncated") boolean truncated)
+            @JsonProperty("truncated") boolean truncated,
+            @JsonProperty("tableLocation") String tableLocation)
     {
         this.partitionIds = partitionIds;
         this.truncated = truncated;
+        this.tableLocation = tableLocation;
     }
 
     @JsonProperty
@@ -45,4 +48,10 @@ public boolean isTruncated()
     {
         return truncated;
     }
+
+    @JsonProperty
+    public String getTableLocation()
+    {
+        return tableLocation;
+    }
 }

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

@@ -808,7 +808,7 @@ public Optional<Object> getInfo(ConnectorTableLayoutHandle layoutHandle)
                     tableLayoutHandle.getPartitions().get().stream()
                             .map(hivePartition -> hivePartition.getPartitionId().getPartitionName())
                             .collect(toList()),
-                    false));
+                    false, tableLayoutHandle.getTablePath()));
         }
 
         return Optional.empty();
@@ -1756,7 +1756,7 @@ public Optional<ConnectorOutputMetadata> finishCreateTable(ConnectorSession sess
         metastore.createTable(session, table, principalPrivileges, Optional.of(writeInfo.getWritePath()), false, tableStatistics, emptyList());
 
         if (handle.getPartitionedBy().isEmpty()) {
-            return Optional.of(new HiveWrittenPartitions(ImmutableList.of(UNPARTITIONED_ID.getPartitionName())));
+            return Optional.of(new HiveWrittenPartitions(new HiveOutputInfo(ImmutableList.of(UNPARTITIONED_ID.getPartitionName()), writeInfo.getTargetPath().toString())));
         }
 
         if (isRespectTableFormat(session)) {
@@ -1785,10 +1785,10 @@ public Optional<ConnectorOutputMetadata> finishCreateTable(ConnectorSession sess
                     partitionStatistics);
         }
 
-        return Optional.of(new HiveWrittenPartitions(
+        return Optional.of(new HiveWrittenPartitions(new HiveOutputInfo(
                 partitionUpdates.stream()
                         .map(PartitionUpdate::getName)
-                        .collect(toList())));
+                        .collect(toList()), writeInfo.getTargetPath().toString())));
     }
 
     public static boolean shouldCreateFilesForMissingBuckets(Table table, ConnectorSession session)
@@ -2196,11 +2196,11 @@ else if (partitionUpdate.getUpdateMode() == NEW || partitionUpdate.getUpdateMode
             }
         }
 
-        return Optional.of(new HiveWrittenPartitions(
+        return Optional.of(new HiveWrittenPartitions(new HiveOutputInfo(
                 partitionUpdates.stream()
                         .map(PartitionUpdate::getName)
                         .map(name -> name.isEmpty() ? UNPARTITIONED_ID.getPartitionName() : name)
-                        .collect(toList())));
+                        .collect(toList()), table.getStorage().getLocation())));
     }
 
     /**