Improve performance for Equality Delete files in Iceberg connector

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

@@ -34,7 +34,7 @@
 import io.trino.spi.connector.ConnectorIndexProvider;
 import io.trino.spi.connector.ConnectorNodePartitioningProvider;
 import io.trino.spi.connector.ConnectorPageSinkProvider;
-import io.trino.spi.connector.ConnectorPageSourceProvider;
+import io.trino.spi.connector.ConnectorPageSourceProviderFactory;
 import io.trino.spi.connector.ConnectorSplitManager;
 import io.trino.spi.function.FunctionProvider;
 
@@ -59,9 +59,9 @@ public static CatalogServiceProvider<ConnectorSplitManager> createSplitManagerPr
 
     @Provides
     @Singleton
-    public static CatalogServiceProvider<ConnectorPageSourceProvider> createPageSourceProvider(ConnectorServicesProvider connectorServicesProvider)
+    public static CatalogServiceProvider<ConnectorPageSourceProviderFactory> createPageSourceProviderFactory(ConnectorServicesProvider connectorServicesProvider)
     {
-        return new ConnectorCatalogServiceProvider<>("page source provider", connectorServicesProvider, connector -> connector.getPageSourceProvider().orElse(null));
+        return new ConnectorCatalogServiceProvider<>("page source provider factory", connectorServicesProvider, connector -> connector.getPageSourceProviderFactory().orElse(null));
     }
 
     @Provides

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

@@ -30,7 +30,7 @@
 import io.trino.spi.connector.ConnectorIndexProvider;
 import io.trino.spi.connector.ConnectorNodePartitioningProvider;
 import io.trino.spi.connector.ConnectorPageSinkProvider;
-import io.trino.spi.connector.ConnectorPageSourceProvider;
+import io.trino.spi.connector.ConnectorPageSourceProviderFactory;
 import io.trino.spi.connector.ConnectorRecordSetProvider;
 import io.trino.spi.connector.ConnectorSecurityContext;
 import io.trino.spi.connector.ConnectorSplitManager;
@@ -78,7 +78,7 @@ public class ConnectorServices
     private final Optional<FunctionProvider> functionProvider;
     private final CatalogTableFunctions tableFunctions;
     private final Optional<ConnectorSplitManager> splitManager;
-    private final Optional<ConnectorPageSourceProvider> pageSourceProvider;
+    private final Optional<ConnectorPageSourceProviderFactory> pageSourceProviderFactory;
     private final Optional<ConnectorPageSinkProvider> pageSinkProvider;
     private final Optional<ConnectorIndexProvider> indexProvider;
     private final Optional<ConnectorNodePartitioningProvider> partitioningProvider;
@@ -128,23 +128,24 @@ public ConnectorServices(Tracer tracer, CatalogHandle catalogHandle, Connector c
         }
         this.splitManager = Optional.ofNullable(splitManager);
 
-        ConnectorPageSourceProvider connectorPageSourceProvider = null;
+        ConnectorPageSourceProviderFactory connectorPageSourceProviderFactory = null;
         try {
-            connectorPageSourceProvider = connector.getPageSourceProvider();
-            requireNonNull(connectorPageSourceProvider, format("Connector '%s' returned a null page source provider", catalogHandle));
+            connectorPageSourceProviderFactory = connector.getPageSourceProviderFactory();
+            requireNonNull(connectorPageSourceProviderFactory, format("Connector '%s' returned a null page source provider factory", catalogHandle));
         }
         catch (UnsupportedOperationException ignored) {
         }
 
         try {
             ConnectorRecordSetProvider connectorRecordSetProvider = connector.getRecordSetProvider();
             requireNonNull(connectorRecordSetProvider, format("Connector '%s' returned a null record set provider", catalogHandle));
-            verify(connectorPageSourceProvider == null, "Connector '%s' returned both page source and record set providers", catalogHandle);
-            connectorPageSourceProvider = new RecordPageSourceProvider(connectorRecordSetProvider);
+            verify(connectorPageSourceProviderFactory == null, "Connector '%s' returned both page source and record set providers", catalogHandle);
+            var pageSourceProvider = new RecordPageSourceProvider(connectorRecordSetProvider);
+            connectorPageSourceProviderFactory = () -> pageSourceProvider;
         }
         catch (UnsupportedOperationException ignored) {
         }
-        this.pageSourceProvider = Optional.ofNullable(connectorPageSourceProvider);
+        this.pageSourceProviderFactory = Optional.ofNullable(connectorPageSourceProviderFactory);
 
         ConnectorPageSinkProvider connectorPageSinkProvider = null;
         try {
@@ -261,9 +262,9 @@ public Optional<ConnectorSplitManager> getSplitManager()
         return splitManager;
     }
 
-    public Optional<ConnectorPageSourceProvider> getPageSourceProvider()
+    public Optional<ConnectorPageSourceProviderFactory> getPageSourceProviderFactory()
     {
-        return pageSourceProvider;
+        return pageSourceProviderFactory;
     }
 
     public Optional<ConnectorPageSinkProvider> getPageSinkProvider()

core/trino-main/src/main/java/io/trino/metadata/FunctionManager.java

@@ -172,7 +172,7 @@ public TableFunctionProcessorProvider getTableFunctionProcessorProvider(TableFun
             checkArgument(provider != null, "No function provider for catalog: '%s'", catalogHandle);
         }
 
-        return provider.getTableFunctionProcessorProvider(tableFunctionHandle.getFunctionHandle());
+        return provider.getTableFunctionProcessorProviderFactory(tableFunctionHandle.getFunctionHandle()).createTableFunctionProcessorProvider();
     }
 
     private FunctionDependencies getFunctionDependencies(ResolvedFunction resolvedFunction)

core/trino-main/src/main/java/io/trino/operator/ScanFilterAndProjectOperator.java

@@ -43,6 +43,7 @@
 import io.trino.spi.type.Type;
 import io.trino.split.EmptySplit;
 import io.trino.split.PageSourceProvider;
+import io.trino.split.PageSourceProviderFactory;
 import io.trino.sql.planner.plan.PlanNodeId;
 import jakarta.annotation.Nullable;
 
@@ -425,7 +426,7 @@ public ScanFilterAndProjectOperatorFactory(
                 int operatorId,
                 PlanNodeId planNodeId,
                 PlanNodeId sourceId,
-                PageSourceProvider pageSourceProvider,
+                PageSourceProviderFactory pageSourceProvider,
                 Supplier<CursorProcessor> cursorProcessor,
                 Supplier<PageProcessor> pageProcessor,
                 TableHandle table,
@@ -440,13 +441,13 @@ public ScanFilterAndProjectOperatorFactory(
             this.cursorProcessor = requireNonNull(cursorProcessor, "cursorProcessor is null");
             this.pageProcessor = requireNonNull(pageProcessor, "pageProcessor is null");
             this.sourceId = requireNonNull(sourceId, "sourceId is null");
-            this.pageSourceProvider = requireNonNull(pageSourceProvider, "pageSourceProvider is null");
             this.table = requireNonNull(table, "table is null");
             this.columns = ImmutableList.copyOf(requireNonNull(columns, "columns is null"));
             this.dynamicFilter = dynamicFilter;
             this.types = requireNonNull(types, "types is null");
             this.minOutputPageSize = requireNonNull(minOutputPageSize, "minOutputPageSize is null");
             this.minOutputPageRowCount = minOutputPageRowCount;
+            this.pageSourceProvider = pageSourceProvider.createPageSourceProvider(table.getCatalogHandle());
         }
 
         @Override

core/trino-main/src/main/java/io/trino/operator/TableScanOperator.java

@@ -28,6 +28,7 @@
 import io.trino.spi.connector.EmptyPageSource;
 import io.trino.split.EmptySplit;
 import io.trino.split.PageSourceProvider;
+import io.trino.split.PageSourceProviderFactory;
 import io.trino.sql.planner.plan.PlanNodeId;
 import jakarta.annotation.Nullable;
 
@@ -60,18 +61,18 @@ public TableScanOperatorFactory(
                 int operatorId,
                 PlanNodeId planNodeId,
                 PlanNodeId sourceId,
-                PageSourceProvider pageSourceProvider,
+                PageSourceProviderFactory pageSourceProvider,
                 TableHandle table,
                 Iterable<ColumnHandle> columns,
                 DynamicFilter dynamicFilter)
         {
             this.operatorId = operatorId;
             this.planNodeId = requireNonNull(planNodeId, "planNodeId is null");
             this.sourceId = requireNonNull(sourceId, "sourceId is null");
-            this.pageSourceProvider = requireNonNull(pageSourceProvider, "pageSourceProvider is null");
             this.table = requireNonNull(table, "table is null");
             this.columns = ImmutableList.copyOf(requireNonNull(columns, "columns is null"));
             this.dynamicFilter = requireNonNull(dynamicFilter, "dynamicFilter is null");
+            this.pageSourceProvider = pageSourceProvider.createPageSourceProvider(table.getCatalogHandle());
         }
 
         @Override

core/trino-main/src/main/java/io/trino/server/ServerMainModule.java

@@ -125,7 +125,7 @@
 import io.trino.split.PageSinkManager;
 import io.trino.split.PageSinkProvider;
 import io.trino.split.PageSourceManager;
-import io.trino.split.PageSourceProvider;
+import io.trino.split.PageSourceProviderFactory;
 import io.trino.split.SplitManager;
 import io.trino.sql.PlannerContext;
 import io.trino.sql.SqlEnvironmentConfig;
@@ -377,7 +377,7 @@ protected void setup(Binder binder)
 
         // data stream provider
         binder.bind(PageSourceManager.class).in(Scopes.SINGLETON);
-        binder.bind(PageSourceProvider.class).to(PageSourceManager.class).in(Scopes.SINGLETON);
+        binder.bind(PageSourceProviderFactory.class).to(PageSourceManager.class).in(Scopes.SINGLETON);
 
         // page sink provider
         binder.bind(PageSinkManager.class).in(Scopes.SINGLETON);

core/trino-main/src/main/java/io/trino/split/PageSourceManager.java

@@ -22,6 +22,7 @@
 import io.trino.spi.connector.ColumnHandle;
 import io.trino.spi.connector.ConnectorPageSource;
 import io.trino.spi.connector.ConnectorPageSourceProvider;
+import io.trino.spi.connector.ConnectorPageSourceProviderFactory;
 import io.trino.spi.connector.DynamicFilter;
 import io.trino.spi.connector.EmptyPageSource;
 import io.trino.spi.predicate.TupleDomain;
@@ -33,37 +34,56 @@
 import static java.util.Objects.requireNonNull;
 
 public class PageSourceManager
-        implements PageSourceProvider
+        implements PageSourceProviderFactory
 {
-    private final CatalogServiceProvider<ConnectorPageSourceProvider> pageSourceProvider;
+    private final CatalogServiceProvider<ConnectorPageSourceProviderFactory> pageSourceProviderFactory;
 
     @Inject
-    public PageSourceManager(CatalogServiceProvider<ConnectorPageSourceProvider> pageSourceProvider)
+    public PageSourceManager(CatalogServiceProvider<ConnectorPageSourceProviderFactory> pageSourceProviderFactory)
     {
-        this.pageSourceProvider = requireNonNull(pageSourceProvider, "pageSourceProvider is null");
+        this.pageSourceProviderFactory = requireNonNull(pageSourceProviderFactory, "pageSourceProviderFactory is null");
     }
 
     @Override
-    public ConnectorPageSource createPageSource(Session session, Split split, TableHandle table, List<ColumnHandle> columns, DynamicFilter dynamicFilter)
+    public PageSourceProvider createPageSourceProvider(CatalogHandle catalogHandle)
     {
-        requireNonNull(columns, "columns is null");
-        checkArgument(split.getCatalogHandle().equals(table.getCatalogHandle()), "mismatched split and table");
-        CatalogHandle catalogHandle = split.getCatalogHandle();
+        ConnectorPageSourceProviderFactory provider = pageSourceProviderFactory.getService(catalogHandle);
+        return new PageSourceProviderInstance(provider.createPageSourceProvider());
+    }
+
+    private static class PageSourceProviderInstance
+            implements PageSourceProvider
+    {
+        private final ConnectorPageSourceProvider pageSourceProvider;
 
-        ConnectorPageSourceProvider provider = pageSourceProvider.getService(catalogHandle);
-        TupleDomain<ColumnHandle> constraint = dynamicFilter.getCurrentPredicate();
-        if (constraint.isNone()) {
-            return new EmptyPageSource();
+        private PageSourceProviderInstance(ConnectorPageSourceProvider pageSourceProvider)
+        {
+            this.pageSourceProvider = requireNonNull(pageSourceProvider, "pageSourceProvider is null");
         }
-        if (!isAllowPushdownIntoConnectors(session)) {
-            dynamicFilter = DynamicFilter.EMPTY;
+
+        @Override
+        public ConnectorPageSource createPageSource(Session session,
+                Split split,
+                TableHandle table,
+                List<ColumnHandle> columns,
+                DynamicFilter dynamicFilter)
+        {
+            requireNonNull(columns, "columns is null");
+            checkArgument(split.getCatalogHandle().equals(table.getCatalogHandle()), "mismatched split and table");
+
+            TupleDomain<ColumnHandle> constraint = dynamicFilter.getCurrentPredicate();
+            if (constraint.isNone()) {
+                return new EmptyPageSource();
+            }
+            if (!isAllowPushdownIntoConnectors(session)) {
+                dynamicFilter = DynamicFilter.EMPTY;
+            }
+            return pageSourceProvider.createPageSource(table.getTransaction(),
+                    session.toConnectorSession(table.getCatalogHandle()),
+                    split.getConnectorSplit(),
+                    table.getConnectorHandle(),
+                    columns,
+                    dynamicFilter);
         }
-        return provider.createPageSource(
-                table.getTransaction(),
-                session.toConnectorSession(catalogHandle),
-                split.getConnectorSplit(),
-                table.getConnectorHandle(),
-                columns,
-                dynamicFilter);
     }
 }

core/trino-main/src/main/java/io/trino/split/PageSourceProviderFactory.java

@@ -0,0 +1,21 @@
+/*
+ * 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 io.trino.split;
+
+import io.trino.spi.connector.CatalogHandle;
+
+public interface PageSourceProviderFactory
+{
+    PageSourceProvider createPageSourceProvider(CatalogHandle catalogHandle);
+}

core/trino-main/src/main/java/io/trino/sql/planner/LocalExecutionPlanner.java

@@ -178,7 +178,7 @@
 import io.trino.spiller.SingleStreamSpillerFactory;
 import io.trino.spiller.SpillerFactory;
 import io.trino.split.PageSinkManager;
-import io.trino.split.PageSourceProvider;
+import io.trino.split.PageSourceManager;
 import io.trino.sql.DynamicFilters;
 import io.trino.sql.PlannerContext;
 import io.trino.sql.gen.ExpressionCompiler;
@@ -396,7 +396,7 @@ public class LocalExecutionPlanner
     private final Metadata metadata;
     private final IrTypeAnalyzer typeAnalyzer;
     private final Optional<ExplainAnalyzeContext> explainAnalyzeContext;
-    private final PageSourceProvider pageSourceProvider;
+    private final PageSourceManager pageSourceProvider;
     private final IndexManager indexManager;
     private final NodePartitioningManager nodePartitioningManager;
     private final PageSinkManager pageSinkManager;
@@ -451,7 +451,7 @@ public LocalExecutionPlanner(
             PlannerContext plannerContext,
             IrTypeAnalyzer typeAnalyzer,
             Optional<ExplainAnalyzeContext> explainAnalyzeContext,
-            PageSourceProvider pageSourceProvider,
+            PageSourceManager pageSourceProvider,
             IndexManager indexManager,
             NodePartitioningManager nodePartitioningManager,
             PageSinkManager pageSinkManager,

core/trino-main/src/main/java/io/trino/sql/planner/iterative/rule/ExtractSpatialJoins.java

@@ -41,6 +41,7 @@
 import io.trino.spi.type.TypeManager;
 import io.trino.spi.type.TypeSignature;
 import io.trino.split.PageSourceManager;
+import io.trino.split.PageSourceProvider;
 import io.trino.split.SplitManager;
 import io.trino.split.SplitSource;
 import io.trino.split.SplitSource.SplitBatch;
@@ -471,12 +472,13 @@ private static KdbTree loadKdbTree(String tableName, Session session, Metadata m
 
         Optional<KdbTree> kdbTree = Optional.empty();
         try (SplitSource splitSource = splitManager.getSplits(session, session.getQuerySpan(), tableHandle, DynamicFilter.EMPTY, alwaysTrue())) {
+            PageSourceProvider statefulPageSourceProvider = pageSourceManager.createPageSourceProvider(tableHandle.getCatalogHandle());
             while (!Thread.currentThread().isInterrupted()) {
                 SplitBatch splitBatch = getFutureValue(splitSource.getNextBatch(1000));
                 List<Split> splits = splitBatch.getSplits();
 
                 for (Split split : splits) {
-                    try (ConnectorPageSource pageSource = pageSourceManager.createPageSource(session, split, tableHandle, ImmutableList.of(kdbTreeColumn), DynamicFilter.EMPTY)) {
+                    try (ConnectorPageSource pageSource = statefulPageSourceProvider.createPageSource(session, split, tableHandle, ImmutableList.of(kdbTreeColumn), DynamicFilter.EMPTY)) {
                         do {
                             getFutureValue(pageSource.isBlocked());
                             Page page = pageSource.getNextPage();

core/trino-main/src/main/java/io/trino/testing/PlanTester.java

@@ -215,7 +215,7 @@
 import static io.trino.connector.CatalogServiceProviderModule.createMaterializedViewPropertyManager;
 import static io.trino.connector.CatalogServiceProviderModule.createNodePartitioningProvider;
 import static io.trino.connector.CatalogServiceProviderModule.createPageSinkProvider;
-import static io.trino.connector.CatalogServiceProviderModule.createPageSourceProvider;
+import static io.trino.connector.CatalogServiceProviderModule.createPageSourceProviderFactory;
 import static io.trino.connector.CatalogServiceProviderModule.createSchemaPropertyManager;
 import static io.trino.connector.CatalogServiceProviderModule.createSplitManagerProvider;
 import static io.trino.connector.CatalogServiceProviderModule.createTableFunctionProvider;
@@ -365,7 +365,7 @@ private PlanTester(Session defaultSession, int nodeCountForStats)
                 nodeSchedulerConfig,
                 optimizerConfig));
         this.splitManager = new SplitManager(createSplitManagerProvider(catalogManager), tracer, new QueryManagerConfig());
-        this.pageSourceManager = new PageSourceManager(createPageSourceProvider(catalogManager));
+        this.pageSourceManager = new PageSourceManager(createPageSourceProviderFactory(catalogManager));
         this.pageSinkManager = new PageSinkManager(createPageSinkProvider(catalogManager));
         this.indexManager = new IndexManager(createIndexProvider(catalogManager));
         NodeScheduler nodeScheduler = new NodeScheduler(new UniformNodeSelectorFactory(nodeManager, nodeSchedulerConfig, new NodeTaskMap(finalizerService)));

core/trino-main/src/test/java/io/trino/execution/TestingPageSourceProvider.java

@@ -20,6 +20,7 @@
 import io.trino.spi.connector.ColumnHandle;
 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;
@@ -34,13 +35,19 @@
 import static java.util.Objects.requireNonNull;
 
 public class TestingPageSourceProvider
-        implements ConnectorPageSourceProvider
+        implements ConnectorPageSourceProviderFactory, ConnectorPageSourceProvider
 {
     public TestingPageSourceProvider()
     {
         System.out.println();
     }
 
+    @Override
+    public ConnectorPageSourceProvider createPageSourceProvider()
+    {
+        return this;
+    }
+
     @Override
     public ConnectorPageSource createPageSource(
             ConnectorTransactionHandle transaction,

core/trino-main/src/test/java/io/trino/operator/BenchmarkScanFilterAndProjectOperator.java

@@ -168,7 +168,7 @@ private void createScanFilterAndProjectOperatorFactories(List<Page> inputPages,
                     0,
                     new PlanNodeId("test"),
                     new PlanNodeId("test_source"),
-                    (session, split, table, columns, dynamicFilter) -> new FixedPageSource(inputPages),
+                    (catalog) -> (session, split, table, columns, dynamicFilter) -> new FixedPageSource(inputPages),
                     () -> cursorProcessor,
                     () -> pageProcessor,
                     TEST_TABLE_HANDLE,