Address PR comments - will rebase and clean commits after all feedback is addressed

Author: tbaeg

core/trino-main/src/main/java/io/trino/connector/DefaultCatalogFactory.java

@@ -19,7 +19,6 @@
 import io.opentelemetry.api.OpenTelemetry;
 import io.opentelemetry.api.trace.Tracer;
 import io.trino.connector.informationschema.InformationSchemaConnector;
-import io.trino.connector.system.DefaultSystemTablesProvider;
 import io.trino.connector.system.SystemConnector;
 import io.trino.connector.system.SystemTablesProvider;
 import io.trino.execution.scheduler.NodeSchedulerConfig;
@@ -155,7 +154,7 @@ private CatalogConnector createCatalog(CatalogHandle catalogHandle, ConnectorNam
                         accessControl,
                         maxPrefetchedInformationSchemaPrefixes));
 
-        SystemTablesProvider systemTablesProvider = new DefaultSystemTablesProvider(
+        SystemTablesProvider systemTablesProvider = new SystemTablesProvider(
                 transactionManager,
                 metadata,
                 catalogHandle.getCatalogName().toString(),
@@ -170,7 +169,8 @@ private CatalogConnector createCatalog(CatalogHandle catalogHandle, ConnectorNam
                         systemTablesProvider,
                         transactionId -> transactionManager.getConnectorTransaction(transactionId, catalogHandle),
                         accessControl,
-                        catalogHandle.getCatalogName().toString(), catalogConnector.getPageSourceProviderFactory()));
+                        catalogHandle.getCatalogName().toString(),
+                        catalogConnector.getPageSourceProviderFactory()));
 
         return new CatalogConnector(
                 catalogHandle,

core/trino-main/src/main/java/io/trino/connector/system/DefaultSystemTablesProvider.java

@@ -63,14 +63,15 @@ public class SystemPageSourceProvider
     private final SystemTablesProvider tables;
     private final AccessControl accessControl;
     private final String catalogName;
-    private final Optional<ConnectorPageSourceProviderFactory> pageSourceProviderFactory;
+    private final Optional<ConnectorPageSourceProvider> connectorPageSourceProvider;
 
     public SystemPageSourceProvider(SystemTablesProvider tables, AccessControl accessControl, String catalogName, Optional<ConnectorPageSourceProviderFactory> pageSourceProviderFactory)
     {
         this.tables = requireNonNull(tables, "tables is null");
         this.accessControl = requireNonNull(accessControl, "accessControl is null");
         this.catalogName = requireNonNull(catalogName, "catalogName is null");
-        this.pageSourceProviderFactory = requireNonNull(pageSourceProviderFactory, "pageSourceProviderFactory is null");
+        this.connectorPageSourceProvider = requireNonNull(pageSourceProviderFactory, "pageSourceProviderFactory is null")
+                .map(ConnectorPageSourceProviderFactory::createPageSourceProvider);
     }
 
     @Override
@@ -87,7 +88,7 @@ public ConnectorPageSource createPageSource(
 
         // if the split is not a SystemSplit, we immediately delegate to the Connector to build a PageSource
         if (!(split instanceof SystemSplit systemSplit)) {
-            return pageSourceProviderFactory.orElseThrow().createPageSourceProvider()
+            return connectorPageSourceProvider.orElseThrow()
                     .createPageSource(systemTransaction.getConnectorTransactionHandle(), session, split, table, columns, dynamicFilter);
         }
 

core/trino-main/src/main/java/io/trino/connector/system/SystemPageSourceProvider.java

@@ -11,23 +11,81 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+
 package io.trino.connector.system;
 
+import io.trino.FullConnectorSession;
+import io.trino.Session;
+import io.trino.metadata.Metadata;
+import io.trino.metadata.QualifiedObjectName;
 import io.trino.spi.connector.ConnectorSession;
 import io.trino.spi.connector.SchemaTableName;
 import io.trino.spi.connector.SystemTable;
+import io.trino.transaction.TransactionManager;
 
+import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
 
-public interface SystemTablesProvider
+import static com.google.common.collect.ImmutableMap.toImmutableMap;
+import static java.util.Objects.requireNonNull;
+import static java.util.function.Function.identity;
+
+public class SystemTablesProvider
 {
-    Set<SystemTable> listSystemTables(ConnectorSession session);
-
-    /**
-     * Resolves table name. Returns {@link Optional#empty()} if table is not found.
-     * Some tables which are not part of set returned by {@link #listSystemTables(ConnectorSession)}
-     * can still be validly resolved.
-     */
-    Optional<SystemTable> getSystemTable(ConnectorSession session, SchemaTableName tableName);
+    private final TransactionManager transactionManager;
+    private final Metadata metadata;
+    private final String catalogName;
+    private final Set<SystemTable> systemTables;
+    private final Map<SchemaTableName, SystemTable> systemTablesMap;
+
+    public SystemTablesProvider(TransactionManager transactionManager, Metadata metadata, String catalogName, Set<SystemTable> systemTables)
+    {
+        this.transactionManager = requireNonNull(transactionManager, "transactionManager is null");
+        this.metadata = requireNonNull(metadata, "metadata is null");
+        this.catalogName = requireNonNull(catalogName, "catalogName is null");
+        this.systemTables = requireNonNull(systemTables, "systemTables is null");
+        this.systemTablesMap = systemTables.stream()
+                .collect(toImmutableMap(
+                        table -> table.getTableMetadata().getTable(),
+                        identity()));
+    }
+
+    public Set<SystemTable> listSystemTables(ConnectorSession session)
+    {
+        // dynamic are not listed, so ony list static tables
+        return systemTables;
+    }
+
+    public Optional<SystemTable> getSystemTable(ConnectorSession session, SchemaTableName tableName)
+    {
+        Optional<SystemTable> staticSystemTable = Optional.ofNullable(systemTablesMap.get(tableName));
+        if (staticSystemTable.isPresent()) {
+            return staticSystemTable;
+        }
+
+        // This means there is no known static table, but that doesn't mean a dynamic table must exist.
+        // This node could have a different config that causes that table to not exist.
+        if (!isCoordinatorTransaction(session)) {
+            // this is a session from another coordinator, so there are no dynamic tables here for that session
+            return Optional.empty();
+        }
+
+        // dynamic tables that are not SINGLE_COORDINATOR mode need to implement the
+        // PageSourceProvider interface for SystemSplit through the Connector interface
+        return metadata.getSystemTable(
+                ((FullConnectorSession) session).getSession(),
+                new QualifiedObjectName(catalogName, tableName.getSchemaName(), tableName.getTableName()));
+    }
+
+    private boolean isCoordinatorTransaction(ConnectorSession connectorSession)
+    {
+        return Optional.of(connectorSession)
+                .filter(FullConnectorSession.class::isInstance)
+                .map(FullConnectorSession.class::cast)
+                .map(FullConnectorSession::getSession)
+                .flatMap(Session::getTransactionId)
+                .map(transactionManager::transactionExists)
+                .orElse(false);
+    }
 }

core/trino-main/src/main/java/io/trino/connector/system/SystemTablesProvider.java

@@ -779,7 +779,7 @@ private Optional<SystemTable> getRawSystemTable(ConnectorSession session, Schema
             case PARTITIONS -> Optional.of(new PartitionsTable(tableName, typeManager, table, getCurrentSnapshotId(table), icebergScanExecutor));
             case ALL_MANIFESTS -> Optional.of(new AllManifestsTable(tableName, table, icebergScanExecutor));
             case MANIFESTS -> Optional.of(new ManifestsTable(tableName, table, getCurrentSnapshotId(table)));
-            case FILES -> Optional.of(new FilesTable(tableName, typeManager, table, getCurrentSnapshotId(table), icebergScanExecutor));
+            case FILES -> Optional.of(new FilesTable(tableName, typeManager, table, getCurrentSnapshotId(table)));
             case ALL_ENTRIES -> Optional.of(new EntriesTable(typeManager, tableName, table, ALL_ENTRIES, icebergScanExecutor));
             case ENTRIES -> Optional.of(new EntriesTable(typeManager, tableName, table, ENTRIES, icebergScanExecutor));
             case PROPERTIES -> Optional.of(new PropertiesTable(tableName, table));

plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergMetadata.java

@@ -33,6 +33,7 @@
 import io.trino.plugin.hive.orc.OrcWriterConfig;
 import io.trino.plugin.hive.parquet.ParquetReaderConfig;
 import io.trino.plugin.hive.parquet.ParquetWriterConfig;
+import io.trino.plugin.iceberg.bean.PartitionFieldSummaryBean;
 import io.trino.plugin.iceberg.cache.IcebergCacheKeyProvider;
 import io.trino.plugin.iceberg.catalog.rest.DefaultIcebergFileSystemFactory;
 import io.trino.plugin.iceberg.fileio.ForwardingFileIoFactory;
@@ -60,10 +61,12 @@
 import io.trino.spi.function.FunctionProvider;
 import io.trino.spi.function.table.ConnectorTableFunction;
 import io.trino.spi.procedure.Procedure;
+import org.apache.iceberg.ManifestFile;
 
 import static com.google.inject.multibindings.Multibinder.newSetBinder;
 import static com.google.inject.multibindings.OptionalBinder.newOptionalBinder;
 import static io.airlift.configuration.ConfigBinder.configBinder;
+import static io.airlift.json.JsonBinder.jsonBinder;
 import static io.airlift.json.JsonCodecBinder.jsonCodecBinder;
 import static org.weakref.jmx.guice.ExportBinder.newExporter;
 
@@ -105,6 +108,7 @@ public void configure(Binder binder)
         newOptionalBinder(binder, Key.get(HiveMetastoreFactory.class, RawHiveMetastoreFactory.class));
 
         jsonCodecBinder(binder).bindJsonCodec(CommitTaskData.class);
+        jsonBinder(binder).addDeserializerBinding(ManifestFile.PartitionFieldSummary.class).to(PartitionFieldSummaryBean.Deserializer.class);
 
         binder.bind(FileFormatDataSourceStats.class).in(Scopes.SINGLETON);
         newExporter(binder).export(FileFormatDataSourceStats.class).withGeneratedName();

plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergModule.java

@@ -246,9 +246,9 @@ public ConnectorPageSource createPageSource(
         if (connectorSplit instanceof FilesTableSplit filesTableSplit) {
             return new FilesTablePageSource(
                     typeManager,
-                    fileSystemFactory.create(session.getIdentity(), filesTableSplit.fileSystemProperties()),
+                    fileSystemFactory.create(session.getIdentity(), filesTableSplit.fileIoProperties()),
                     fileIoFactory,
-                    columns.stream().map(SystemColumnHandle.class::cast).map(SystemColumnHandle::columnName).toList(),
+                    columns.stream().map(SystemColumnHandle.class::cast).map(SystemColumnHandle::columnName).collect(toImmutableList()),
                     filesTableSplit);
         }
 

plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergPageSourceProvider.java

@@ -154,16 +154,7 @@ public static List<PartitionField> getAllPartitionFields(Schema schema, Map<Inte
 
     public static List<PartitionField> getAllPartitionFields(Table icebergTable)
     {
-        Set<Integer> existingColumnsIds = TypeUtil.indexById(icebergTable.schema().asStruct()).keySet();
-
-        List<PartitionField> visiblePartitionFields = icebergTable.specs()
-                .values().stream()
-                .flatMap(partitionSpec -> partitionSpec.fields().stream())
-                // skip columns that were dropped
-                .filter(partitionField -> existingColumnsIds.contains(partitionField.sourceId()))
-                .collect(toImmutableList());
-
-        return filterOutDuplicates(visiblePartitionFields);
+        return getAllPartitionFields(icebergTable.schema(), icebergTable.specs());
     }
 
     private static List<PartitionField> filterOutDuplicates(List<PartitionField> visiblePartitionFields)