diff --git a/core/trino-main/src/main/java/io/trino/operator/ScanFilterAndProjectOperator.java b/core/trino-main/src/main/java/io/trino/operator/ScanFilterAndProjectOperator.java
index 8d0a1e5d911c..3daf5d94d0d7 100644
--- a/core/trino-main/src/main/java/io/trino/operator/ScanFilterAndProjectOperator.java
+++ b/core/trino-main/src/main/java/io/trino/operator/ScanFilterAndProjectOperator.java
@@ -491,33 +491,33 @@ public SourceOperator createOperator(DriverContext driverContext)
 
         @Override
         public WorkProcessorSourceOperator create(
-                Session session,
+                OperatorContext operatorContext,
                 MemoryTrackingContext memoryTrackingContext,
                 DriverYieldSignal yieldSignal,
                 WorkProcessor<Split> splits)
         {
-            return create(session, memoryTrackingContext, yieldSignal, splits, true);
+            return create(operatorContext, memoryTrackingContext, yieldSignal, splits, true);
         }
 
         @Override
         public WorkProcessorSourceOperator createAdapterOperator(
-                Session session,
+                OperatorContext operatorContext,
                 MemoryTrackingContext memoryTrackingContext,
                 DriverYieldSignal yieldSignal,
                 WorkProcessor<Split> splits)
         {
-            return create(session, memoryTrackingContext, yieldSignal, splits, false);
+            return create(operatorContext, memoryTrackingContext, yieldSignal, splits, false);
         }
 
         private ScanFilterAndProjectOperator create(
-                Session session,
+                OperatorContext operatorContext,
                 MemoryTrackingContext memoryTrackingContext,
                 DriverYieldSignal yieldSignal,
                 WorkProcessor<Split> splits,
                 boolean avoidPageMaterialization)
         {
             return new ScanFilterAndProjectOperator(
-                    session,
+                    operatorContext.getSession(),
                     memoryTrackingContext,
                     yieldSignal,
                     splits,
diff --git a/core/trino-main/src/main/java/io/trino/operator/TableScanOperator.java b/core/trino-main/src/main/java/io/trino/operator/TableScanOperator.java
index 4ffdb1481701..648f6529b322 100644
--- a/core/trino-main/src/main/java/io/trino/operator/TableScanOperator.java
+++ b/core/trino-main/src/main/java/io/trino/operator/TableScanOperator.java
@@ -18,7 +18,6 @@
 import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.ListenableFuture;
 import com.google.common.util.concurrent.SettableFuture;
-import io.trino.Session;
 import io.trino.memory.context.LocalMemoryContext;
 import io.trino.memory.context.MemoryTrackingContext;
 import io.trino.metadata.Split;
@@ -114,13 +113,13 @@ public SourceOperator createOperator(DriverContext driverContext)
 
         @Override
         public WorkProcessorSourceOperator create(
-                Session session,
+                OperatorContext operatorContext,
                 MemoryTrackingContext memoryTrackingContext,
                 DriverYieldSignal yieldSignal,
                 WorkProcessor<Split> splits)
         {
             return new TableScanWorkProcessorOperator(
-                    session,
+                    operatorContext.getSession(),
                     memoryTrackingContext,
                     splits,
                     pageSourceProvider,
diff --git a/core/trino-main/src/main/java/io/trino/operator/WorkProcessorPipelineSourceOperator.java b/core/trino-main/src/main/java/io/trino/operator/WorkProcessorPipelineSourceOperator.java
index dfd2527b9cf7..267f0eaabc02 100644
--- a/core/trino-main/src/main/java/io/trino/operator/WorkProcessorPipelineSourceOperator.java
+++ b/core/trino-main/src/main/java/io/trino/operator/WorkProcessorPipelineSourceOperator.java
@@ -146,7 +146,7 @@ private WorkProcessorPipelineSourceOperator(
         WorkProcessor<Split> splits = WorkProcessor.create(new Splits());
 
         sourceOperator = sourceOperatorFactory.create(
-                operatorContext.getSession(),
+                operatorContext,
                 sourceOperatorMemoryTrackingContext,
                 operatorContext.getDriverContext().getYieldSignal(),
                 splits);
diff --git a/core/trino-main/src/main/java/io/trino/operator/WorkProcessorSourceOperatorAdapter.java b/core/trino-main/src/main/java/io/trino/operator/WorkProcessorSourceOperatorAdapter.java
index 7e8ea6af736d..7feb9a353985 100644
--- a/core/trino-main/src/main/java/io/trino/operator/WorkProcessorSourceOperatorAdapter.java
+++ b/core/trino-main/src/main/java/io/trino/operator/WorkProcessorSourceOperatorAdapter.java
@@ -16,7 +16,6 @@
 import com.google.common.base.Suppliers;
 import com.google.common.util.concurrent.ListenableFuture;
 import com.google.common.util.concurrent.SettableFuture;
-import io.trino.Session;
 import io.trino.memory.context.MemoryTrackingContext;
 import io.trino.metadata.Split;
 import io.trino.spi.Page;
@@ -56,12 +55,12 @@ public interface AdapterWorkProcessorSourceOperatorFactory
             extends WorkProcessorSourceOperatorFactory
     {
         default WorkProcessorSourceOperator createAdapterOperator(
-                Session session,
+                OperatorContext operatorContext,
                 MemoryTrackingContext memoryTrackingContext,
                 DriverYieldSignal yieldSignal,
                 WorkProcessor<Split> splits)
         {
-            return create(session, memoryTrackingContext, yieldSignal, splits);
+            return create(operatorContext, memoryTrackingContext, yieldSignal, splits);
         }
     }
 
@@ -72,7 +71,7 @@ public WorkProcessorSourceOperatorAdapter(OperatorContext operatorContext, Adapt
         this.splitBuffer = new SplitBuffer();
         this.sourceOperator = sourceOperatorFactory
                 .createAdapterOperator(
-                        operatorContext.getSession(),
+                        operatorContext,
                         new MemoryTrackingContext(
                                 operatorContext.aggregateUserMemoryContext(),
                                 operatorContext.aggregateRevocableMemoryContext()),
diff --git a/core/trino-main/src/main/java/io/trino/operator/WorkProcessorSourceOperatorFactory.java b/core/trino-main/src/main/java/io/trino/operator/WorkProcessorSourceOperatorFactory.java
index 04b4002f8d9f..6db14d8967f3 100644
--- a/core/trino-main/src/main/java/io/trino/operator/WorkProcessorSourceOperatorFactory.java
+++ b/core/trino-main/src/main/java/io/trino/operator/WorkProcessorSourceOperatorFactory.java
@@ -13,7 +13,6 @@
  */
 package io.trino.operator;
 
-import io.trino.Session;
 import io.trino.memory.context.MemoryTrackingContext;
 import io.trino.metadata.Split;
 import io.trino.sql.planner.plan.PlanNodeId;
@@ -29,7 +28,7 @@ public interface WorkProcessorSourceOperatorFactory
     String getOperatorType();
 
     WorkProcessorSourceOperator create(
-            Session session,
+            OperatorContext operatorContext,
             MemoryTrackingContext memoryTrackingContext,
             DriverYieldSignal yieldSignal,
             WorkProcessor<Split> splits);
diff --git a/core/trino-main/src/test/java/io/trino/operator/TestWorkProcessorPipelineSourceOperator.java b/core/trino-main/src/test/java/io/trino/operator/TestWorkProcessorPipelineSourceOperator.java
index ca0426372237..4f5286bf6c03 100644
--- a/core/trino-main/src/test/java/io/trino/operator/TestWorkProcessorPipelineSourceOperator.java
+++ b/core/trino-main/src/test/java/io/trino/operator/TestWorkProcessorPipelineSourceOperator.java
@@ -19,7 +19,6 @@
 import com.google.common.util.concurrent.SettableFuture;
 import io.airlift.units.DataSize;
 import io.airlift.units.Duration;
-import io.trino.Session;
 import io.trino.memory.context.MemoryTrackingContext;
 import io.trino.metadata.Split;
 import io.trino.operator.WorkProcessor.Transformation;
@@ -342,7 +341,7 @@ public String getOperatorType()
         }
 
         @Override
-        public WorkProcessorSourceOperator create(Session session, MemoryTrackingContext memoryTrackingContext, DriverYieldSignal yieldSignal, WorkProcessor<Split> splits)
+        public WorkProcessorSourceOperator create(OperatorContext operatorContext, MemoryTrackingContext memoryTrackingContext, DriverYieldSignal yieldSignal, WorkProcessor<Split> splits)
         {
             assertNull(sourceOperator, "source operator already created");
             sourceOperator = new TestWorkProcessorSourceOperator(
diff --git a/core/trino-main/src/test/java/io/trino/operator/TestWorkProcessorSourceOperatorAdapter.java b/core/trino-main/src/test/java/io/trino/operator/TestWorkProcessorSourceOperatorAdapter.java
index 07007a7b44c7..c914f1332740 100644
--- a/core/trino-main/src/test/java/io/trino/operator/TestWorkProcessorSourceOperatorAdapter.java
+++ b/core/trino-main/src/test/java/io/trino/operator/TestWorkProcessorSourceOperatorAdapter.java
@@ -16,7 +16,6 @@
 import com.google.common.collect.ImmutableMap;
 import com.google.common.util.concurrent.MoreExecutors;
 import io.airlift.units.Duration;
-import io.trino.Session;
 import io.trino.memory.context.MemoryTrackingContext;
 import io.trino.metadata.Split;
 import io.trino.operator.WorkProcessorSourceOperatorAdapter.AdapterWorkProcessorSourceOperatorFactory;
@@ -93,7 +92,7 @@ private static class TestWorkProcessorOperatorFactory
     {
         @Override
         public WorkProcessorSourceOperator create(
-                Session session,
+                OperatorContext operatorContext,
                 MemoryTrackingContext memoryTrackingContext,
                 DriverYieldSignal yieldSignal,
                 WorkProcessor<Split> splits)