Implement distributed processing for iceberg $files system table

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

@@ -19,8 +19,6 @@
 import io.opentelemetry.api.OpenTelemetry;
 import io.opentelemetry.api.trace.Tracer;
 import io.trino.connector.informationschema.InformationSchemaConnector;
-import io.trino.connector.system.CoordinatorSystemTablesProvider;
-import io.trino.connector.system.StaticSystemTablesProvider;
 import io.trino.connector.system.SystemConnector;
 import io.trino.connector.system.SystemTablesProvider;
 import io.trino.execution.scheduler.NodeSchedulerConfig;
@@ -156,17 +154,11 @@ private CatalogConnector createCatalog(CatalogHandle catalogHandle, ConnectorNam
                         accessControl,
                         maxPrefetchedInformationSchemaPrefixes));
 
-        SystemTablesProvider systemTablesProvider;
-        if (currentNode.isCoordinator()) {
-            systemTablesProvider = new CoordinatorSystemTablesProvider(
-                    transactionManager,
-                    metadata,
-                    catalogHandle.getCatalogName().toString(),
-                    new StaticSystemTablesProvider(catalogConnector.getSystemTables()));
-        }
-        else {
-            systemTablesProvider = new StaticSystemTablesProvider(catalogConnector.getSystemTables());
-        }
+        SystemTablesProvider systemTablesProvider = new SystemTablesProvider(
+                transactionManager,
+                metadata,
+                catalogHandle.getCatalogName().toString(),
+                catalogConnector.getSystemTables());
 
         ConnectorServices systemConnector = new ConnectorServices(
                 tracer,
@@ -177,7 +169,8 @@ private CatalogConnector createCatalog(CatalogHandle catalogHandle, ConnectorNam
                         systemTablesProvider,
                         transactionId -> transactionManager.getConnectorTransaction(transactionId, catalogHandle),
                         accessControl,
-                        catalogHandle.getCatalogName().toString()));
+                        catalogHandle.getCatalogName().toString(),
+                        catalogConnector.getPageSourceProviderFactory()));
 
         return new CatalogConnector(
                 catalogHandle,

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

@@ -18,13 +18,15 @@
 import io.trino.security.AccessControl;
 import io.trino.spi.connector.ConnectorMetadata;
 import io.trino.spi.connector.ConnectorPageSourceProvider;
+import io.trino.spi.connector.ConnectorPageSourceProviderFactory;
 import io.trino.spi.connector.ConnectorSession;
 import io.trino.spi.connector.ConnectorSplitManager;
 import io.trino.spi.connector.ConnectorTransactionHandle;
 import io.trino.spi.transaction.IsolationLevel;
 import io.trino.transaction.InternalConnector;
 import io.trino.transaction.TransactionId;
 
+import java.util.Optional;
 import java.util.function.Function;
 
 import static java.util.Objects.requireNonNull;
@@ -43,7 +45,8 @@ public SystemConnector(
             SystemTablesProvider tables,
             Function<TransactionId, ConnectorTransactionHandle> transactionHandleFunction,
             AccessControl accessControl,
-            String catalogName)
+            String catalogName,
+            Optional<ConnectorPageSourceProviderFactory> pageSourceProviderFactory)
     {
         requireNonNull(currentNode, "currentNode is null");
         requireNonNull(nodeManager, "nodeManager is null");
@@ -54,7 +57,7 @@ public SystemConnector(
 
         this.metadata = new SystemTablesMetadata(tables);
         this.splitManager = new SystemSplitManager(currentNode, nodeManager, tables);
-        this.pageSourceProvider = new SystemPageSourceProvider(tables, accessControl, catalogName);
+        this.pageSourceProvider = new SystemPageSourceProvider(tables, accessControl, catalogName, pageSourceProviderFactory);
         this.transactionHandleFunction = transactionHandleFunction;
     }
 

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

@@ -28,6 +28,7 @@
 import io.trino.spi.connector.ConnectorAccessControl;
 import io.trino.spi.connector.ConnectorPageSource;
 import io.trino.spi.connector.ConnectorPageSourceProvider;
+import io.trino.spi.connector.ConnectorPageSourceProviderFactory;
 import io.trino.spi.connector.ConnectorSession;
 import io.trino.spi.connector.ConnectorSplit;
 import io.trino.spi.connector.ConnectorTableHandle;
@@ -38,13 +39,15 @@
 import io.trino.spi.connector.RecordPageSource;
 import io.trino.spi.connector.RecordSet;
 import io.trino.spi.connector.SchemaTableName;
+import io.trino.spi.connector.SystemColumnHandle;
 import io.trino.spi.connector.SystemTable;
 import io.trino.spi.predicate.TupleDomain;
 import io.trino.spi.type.Type;
 
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import java.util.Set;
 
 import static com.google.common.collect.ImmutableList.toImmutableList;
@@ -60,12 +63,15 @@ public class SystemPageSourceProvider
     private final SystemTablesProvider tables;
     private final AccessControl accessControl;
     private final String catalogName;
+    private final Optional<ConnectorPageSourceProvider> connectorPageSourceProvider;
 
-    public SystemPageSourceProvider(SystemTablesProvider tables, AccessControl accessControl, String catalogName)
+    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.connectorPageSourceProvider = requireNonNull(pageSourceProviderFactory, "pageSourceProviderFactory is null")
+                .map(ConnectorPageSourceProviderFactory::createPageSourceProvider);
     }
 
     @Override
@@ -79,7 +85,13 @@ public ConnectorPageSource createPageSource(
     {
         requireNonNull(columns, "columns is null");
         SystemTransactionHandle systemTransaction = (SystemTransactionHandle) transaction;
-        SystemSplit systemSplit = (SystemSplit) split;
+
+        // if the split is not a SystemSplit, we immediately delegate to the Connector to build a PageSource
+        if (!(split instanceof SystemSplit systemSplit)) {
+            return connectorPageSourceProvider.orElseThrow()
+                    .createPageSource(systemTransaction.getConnectorTransactionHandle(), session, split, table, columns, dynamicFilter);
+        }
+
         SchemaTableName tableName = ((SystemTableHandle) table).schemaTableName();
         SystemTable systemTable = tables.getSystemTable(session, tableName)
                 // table might disappear in the meantime

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

@@ -20,6 +20,7 @@
 import io.trino.spi.HostAddress;
 import io.trino.spi.connector.ColumnHandle;
 import io.trino.spi.connector.ConnectorSplit;
+import io.trino.spi.connector.SystemColumnHandle;
 import io.trino.spi.predicate.TupleDomain;
 
 import java.util.List;

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

@@ -28,6 +28,7 @@
 import io.trino.spi.connector.ConstraintApplicationResult;
 import io.trino.spi.connector.SchemaTableName;
 import io.trino.spi.connector.SchemaTablePrefix;
+import io.trino.spi.connector.SystemColumnHandle;
 import io.trino.spi.connector.SystemTable;
 import io.trino.spi.predicate.TupleDomain;
 
@@ -37,7 +38,7 @@
 
 import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.collect.ImmutableList.toImmutableList;
-import static io.trino.connector.system.SystemColumnHandle.toSystemColumnHandles;
+import static com.google.common.collect.ImmutableMap.toImmutableMap;
 import static io.trino.metadata.MetadataUtil.findColumnMetadata;
 import static io.trino.spi.StandardErrorCode.NOT_FOUND;
 import static io.trino.spi.StandardErrorCode.NOT_SUPPORTED;
@@ -111,7 +112,9 @@ public ColumnMetadata getColumnMetadata(ConnectorSession session, ConnectorTable
     public Map<String, ColumnHandle> getColumnHandles(ConnectorSession session, ConnectorTableHandle tableHandle)
     {
         ConnectorTableMetadata tableMetadata = checkAndGetTable(session, tableHandle).getTableMetadata();
-        return toSystemColumnHandles(tableMetadata);
+        return tableMetadata.getColumns().stream().collect(toImmutableMap(
+                ColumnMetadata::getName,
+                column -> new SystemColumnHandle(column.getName())));
     }
 
     private SystemTable checkAndGetTable(ConnectorSession session, ConnectorTableHandle tableHandle)