Remove CursorProcessorCompiler

PageFunctionCompiler is already capable of handling cursor based source
through the RecordPageSource. Switching to it completely is more efficient
and significantly reduces code complexity

BenchmarkPageProcessor2
(columnCount)  (dictionaryBlocks)   (type)   Mode  Cnt  columnOriented          rowOriented            Units
            2               false  varchar  thrpt   10   24094.594 ±   298.176  23104.763 ±  1703.015  ops/s
            2               false   bigint  thrpt   10  227586.727 ±  2931.301  98167.939 ±  1298.626  ops/s
            2                true  varchar  thrpt   10  522379.404 ± 17311.851  22438.570 ±   293.261  ops/s
            2                true   bigint  thrpt   10  531809.004 ±  3907.024  80766.463 ±  2149.091  ops/s
            8               false  varchar  thrpt   10   10905.569 ±    74.438   7881.082 ±   309.976  ops/s
            8               false   bigint  thrpt   10   77246.491 ±   715.483  35992.972 ±   868.800  ops/s
            8                true  varchar  thrpt   10  202403.928 ±  5096.054   7619.441 ±   254.538  ops/s
            8                true   bigint  thrpt   10  210286.557 ±  9252.181  30140.612 ±   579.285  ops/s
           32               false  varchar  thrpt   10    3265.821 ±    39.931   2532.048 ±    95.691  ops/s
           32               false   bigint  thrpt   10   21298.000 ±   141.566   8812.520 ±   193.922  ops/s
           32                true  varchar  thrpt   10   62201.438 ±   883.989   2225.182 ±    37.580  ops/s
           32                true   bigint  thrpt   10   62372.831 ±   452.341   7131.056 ±    98.403  ops/s
          200               false  varchar  thrpt   10     546.173 ±    13.762    300.797 ±    11.186  ops/s
          200               false   bigint  thrpt   10    3428.100 ±    48.497   1189.737 ±     6.415  ops/s
          200                true  varchar  thrpt   10   10243.644 ±    36.385    231.297 ±     4.015  ops/s
          200                true   bigint  thrpt   10   10014.793 ±   366.954    843.738 ±     9.865  ops/s

Author: raunaqmorarka

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

@@ -26,19 +26,14 @@
 import io.trino.metadata.TableHandle;
 import io.trino.operator.WorkProcessor.ProcessState;
 import io.trino.operator.WorkProcessor.TransformationState;
-import io.trino.operator.project.CursorProcessor;
-import io.trino.operator.project.CursorProcessorOutput;
 import io.trino.operator.project.PageProcessor;
 import io.trino.operator.project.PageProcessorMetrics;
 import io.trino.spi.Page;
-import io.trino.spi.PageBuilder;
 import io.trino.spi.connector.ColumnHandle;
 import io.trino.spi.connector.ConnectorPageSource;
 import io.trino.spi.connector.ConnectorSession;
 import io.trino.spi.connector.DynamicFilter;
 import io.trino.spi.connector.EmptyPageSource;
-import io.trino.spi.connector.RecordCursor;
-import io.trino.spi.connector.RecordPageSource;
 import io.trino.spi.connector.SourcePage;
 import io.trino.spi.metrics.Metrics;
 import io.trino.spi.type.Type;
@@ -56,7 +51,6 @@
 import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.function.LongConsumer;
-import java.util.function.Supplier;
 
 import static com.google.common.base.Preconditions.checkState;
 import static com.google.common.util.concurrent.MoreExecutors.directExecutor;
@@ -74,8 +68,6 @@ public class ScanFilterAndProjectOperator
     private final WorkProcessor<Page> pages;
     private final PageProcessorMetrics pageProcessorMetrics = new PageProcessorMetrics();
 
-    @Nullable
-    private RecordCursor cursor;
     @Nullable
     private ConnectorPageSource pageSource;
 
@@ -93,7 +85,6 @@ private ScanFilterAndProjectOperator(
             DriverYieldSignal yieldSignal,
             WorkProcessor<Split> split,
             PageSourceProvider pageSourceProvider,
-            CursorProcessor cursorProcessor,
             PageProcessor pageProcessor,
             TableHandle table,
             Iterable<ColumnHandle> columns,
@@ -107,7 +98,6 @@ private ScanFilterAndProjectOperator(
                         session,
                         yieldSignal,
                         pageSourceProvider,
-                        cursorProcessor,
                         pageProcessor,
                         table,
                         columns,
@@ -163,9 +153,6 @@ public Metrics getConnectorMetrics()
     @Override
     public Metrics getMetrics()
     {
-        if (cursor != null) {
-            return Metrics.EMPTY;
-        }
         return pageProcessorMetrics.getMetrics();
     }
 
@@ -187,9 +174,6 @@ public void close()
                 throw new UncheckedIOException(e);
             }
         }
-        else if (cursor != null) {
-            cursor.close();
-        }
     }
 
     private class SplitToPages
@@ -198,7 +182,6 @@ private class SplitToPages
         final Session session;
         final DriverYieldSignal yieldSignal;
         final PageSourceProvider pageSourceProvider;
-        final CursorProcessor cursorProcessor;
         final PageProcessor pageProcessor;
         final TableHandle table;
         final List<ColumnHandle> columns;
@@ -215,7 +198,6 @@ private class SplitToPages
                 Session session,
                 DriverYieldSignal yieldSignal,
                 PageSourceProvider pageSourceProvider,
-                CursorProcessor cursorProcessor,
                 PageProcessor pageProcessor,
                 TableHandle table,
                 Iterable<ColumnHandle> columns,
@@ -228,7 +210,6 @@ private class SplitToPages
             this.session = requireNonNull(session, "session is null");
             this.yieldSignal = requireNonNull(yieldSignal, "yieldSignal is null");
             this.pageSourceProvider = requireNonNull(pageSourceProvider, "pageSourceProvider is null");
-            this.cursorProcessor = requireNonNull(cursorProcessor, "cursorProcessor is null");
             this.pageProcessor = requireNonNull(pageProcessor, "pageProcessor is null");
             this.table = requireNonNull(table, "table is null");
             this.columns = ImmutableList.copyOf(requireNonNull(columns, "columns is null"));
@@ -250,7 +231,7 @@ public TransformationState<WorkProcessor<Page>> process(Split split)
                 return finished();
             }
 
-            checkState(cursor == null && pageSource == null, "Table scan split already set");
+            checkState(pageSource == null, "Table scan split already set");
 
             if (!dynamicFilter.getCurrentPredicate().isAll()) {
                 dynamicFilterSplitsProcessed++;
@@ -264,22 +245,10 @@ public TransformationState<WorkProcessor<Page>> process(Split split)
                 source = pageSourceProvider.createPageSource(session, split, table, columns, dynamicFilter);
             }
 
-            if (source instanceof RecordPageSource recordPageSource) {
-                cursor = recordPageSource.getCursor();
-                return ofResult(processColumnSource());
-            }
             pageSource = source;
             return ofResult(processPageSource());
         }
 
-        WorkProcessor<Page> processColumnSource()
-        {
-            return WorkProcessor
-                    .create(new RecordCursorToPages(session, yieldSignal, cursorProcessor, types, pageSourceMemoryContext, outputMemoryContext))
-                    .yielding(yieldSignal::isSet)
-                    .blocking(() -> memoryContext.setBytes(localAggregatedMemoryContext.getBytes()));
-        }
-
         WorkProcessor<Page> processPageSource()
         {
             ConnectorSession connectorSession = session.toConnectorSession();
@@ -301,68 +270,6 @@ WorkProcessor<Page> processPageSource()
         }
     }
 
-    private class RecordCursorToPages
-            implements WorkProcessor.Process<Page>
-    {
-        final ConnectorSession session;
-        final DriverYieldSignal yieldSignal;
-        final CursorProcessor cursorProcessor;
-        final PageBuilder pageBuilder;
-        final LocalMemoryContext pageSourceMemoryContext;
-        final LocalMemoryContext outputMemoryContext;
-
-        boolean finished;
-
-        RecordCursorToPages(
-                Session session,
-                DriverYieldSignal yieldSignal,
-                CursorProcessor cursorProcessor,
-                List<Type> types,
-                LocalMemoryContext pageSourceMemoryContext,
-                LocalMemoryContext outputMemoryContext)
-        {
-            this.session = session.toConnectorSession();
-            this.yieldSignal = yieldSignal;
-            this.cursorProcessor = cursorProcessor;
-            this.pageBuilder = new PageBuilder(types);
-            this.pageSourceMemoryContext = pageSourceMemoryContext;
-            this.outputMemoryContext = outputMemoryContext;
-        }
-
-        @Override
-        public ProcessState<Page> process()
-        {
-            if (!finished) {
-                CursorProcessorOutput output = cursorProcessor.process(session, yieldSignal, cursor, pageBuilder);
-                pageSourceMemoryContext.setBytes(cursor.getMemoryUsage());
-
-                processedPositions += output.getProcessedRows();
-                // TODO: derive better values for cursors
-                processedBytes = cursor.getCompletedBytes();
-                physicalBytes = cursor.getCompletedBytes();
-                physicalPositions = processedPositions;
-                readTimeNanos = cursor.getReadTimeNanos();
-                if (output.isNoMoreRows()) {
-                    finished = true;
-                }
-            }
-
-            if (pageBuilder.isFull() || (finished && !pageBuilder.isEmpty())) {
-                // only return a page if buffer is full or cursor has finished
-                Page page = pageBuilder.build();
-                pageBuilder.reset();
-                outputMemoryContext.setBytes(pageBuilder.getRetainedSizeInBytes());
-                return ProcessState.ofResult(page);
-            }
-            if (finished) {
-                checkState(pageBuilder.isEmpty());
-                return ProcessState.finished();
-            }
-            outputMemoryContext.setBytes(pageBuilder.getRetainedSizeInBytes());
-            return ProcessState.yielded();
-        }
-    }
-
     static class ProcessedBytesMonitor
             implements Consumer<ProcessState<Page>>
     {
@@ -445,7 +352,6 @@ public static class ScanFilterAndProjectOperatorFactory
     {
         private final int operatorId;
         private final PlanNodeId planNodeId;
-        private final Supplier<CursorProcessor> cursorProcessor;
         private final Function<DynamicFilter, PageProcessor> pageProcessor;
         private final PlanNodeId sourceId;
         private final PageSourceProvider pageSourceProvider;
@@ -462,7 +368,6 @@ public ScanFilterAndProjectOperatorFactory(
                 PlanNodeId planNodeId,
                 PlanNodeId sourceId,
                 PageSourceProviderFactory pageSourceProvider,
-                Supplier<CursorProcessor> cursorProcessor,
                 Function<DynamicFilter, PageProcessor> pageProcessor,
                 TableHandle table,
                 Iterable<ColumnHandle> columns,
@@ -473,7 +378,6 @@ public ScanFilterAndProjectOperatorFactory(
         {
             this.operatorId = operatorId;
             this.planNodeId = requireNonNull(planNodeId, "planNodeId is null");
-            this.cursorProcessor = requireNonNull(cursorProcessor, "cursorProcessor is null");
             this.pageProcessor = requireNonNull(pageProcessor, "pageProcessor is null");
             this.sourceId = requireNonNull(sourceId, "sourceId is null");
             this.table = requireNonNull(table, "table is null");
@@ -530,7 +434,6 @@ public WorkProcessorSourceOperator create(
                     yieldSignal,
                     split,
                     pageSourceProvider,
-                    cursorProcessor.get(),
                     pageProcessor.apply(dynamicFilter),
                     table,
                     columns,

core/trino-main/src/main/java/io/trino/operator/project/CursorProcessor.java

@@ -127,7 +127,6 @@
 import io.trino.sql.SqlEnvironmentConfig;
 import io.trino.sql.analyzer.SessionTimeProvider;
 import io.trino.sql.analyzer.StatementAnalyzerFactory;
-import io.trino.sql.gen.CursorProcessorCompiler;
 import io.trino.sql.gen.ExpressionCompiler;
 import io.trino.sql.gen.JoinCompiler;
 import io.trino.sql.gen.JoinFilterFunctionCompiler;
@@ -276,8 +275,6 @@ protected void setup(Binder binder)
         newExporter(binder).export(PageFunctionCompiler.class).withGeneratedName();
         binder.bind(ColumnarFilterCompiler.class).in(Scopes.SINGLETON);
         newExporter(binder).export(ColumnarFilterCompiler.class).withGeneratedName();
-        binder.bind(CursorProcessorCompiler.class).in(Scopes.SINGLETON);
-        newExporter(binder).export(CursorProcessorCompiler.class).withGeneratedName();
         configBinder(binder).bindConfig(TaskManagerConfig.class);
 
         // TODO: use conditional module