diff --git a/presto-benchmark/src/main/java/com/facebook/presto/benchmark/AbstractOperatorBenchmark.java b/presto-benchmark/src/main/java/com/facebook/presto/benchmark/AbstractOperatorBenchmark.java
index c874a69f59e8b..e7462a95b2054 100644
--- a/presto-benchmark/src/main/java/com/facebook/presto/benchmark/AbstractOperatorBenchmark.java
+++ b/presto-benchmark/src/main/java/com/facebook/presto/benchmark/AbstractOperatorBenchmark.java
@@ -177,7 +177,7 @@ protected final OperatorFactory createTableScanOperator(int operatorId, PlanNode
             public Operator createOperator(DriverContext driverContext)
             {
                 OperatorContext operatorContext = driverContext.addOperatorContext(operatorId, planNodeId, "BenchmarkSource");
-                ConnectorPageSource pageSource = localQueryRunner.getPageSourceManager().createPageSource(session, split, tableHandle.withDynamicFilter(TupleDomain::all), columnHandles);
+                ConnectorPageSource pageSource = localQueryRunner.getPageSourceManager().createPageSource(session, split, tableHandle.withDynamicFilter(TupleDomain::all), columnHandles, operatorContext.getRuntimeStats());
                 return new PageSourceOperator(pageSource, operatorContext);
             }
 
diff --git a/presto-bigquery/src/main/java/com/facebook/presto/plugin/bigquery/BigQueryPageSourceProvider.java b/presto-bigquery/src/main/java/com/facebook/presto/plugin/bigquery/BigQueryPageSourceProvider.java
index b2a514f64f511..65dd0886560b1 100644
--- a/presto-bigquery/src/main/java/com/facebook/presto/plugin/bigquery/BigQueryPageSourceProvider.java
+++ b/presto-bigquery/src/main/java/com/facebook/presto/plugin/bigquery/BigQueryPageSourceProvider.java
@@ -14,6 +14,7 @@
 package com.facebook.presto.plugin.bigquery;
 
 import com.facebook.airlift.log.Logger;
+import com.facebook.presto.common.RuntimeStats;
 import com.facebook.presto.spi.ColumnHandle;
 import com.facebook.presto.spi.ConnectorPageSource;
 import com.facebook.presto.spi.ConnectorSession;
@@ -51,7 +52,8 @@ public ConnectorPageSource createPageSource(
             ConnectorSplit split,
             ConnectorTableLayoutHandle layout,
             List<ColumnHandle> columns,
-            SplitContext splitContext)
+            SplitContext splitContext,
+            RuntimeStats runtimeStats)
     {
         BigQuerySplit bigQuerySplit = (BigQuerySplit) split;
         if (bigQuerySplit.representsEmptyProjection()) {
diff --git a/presto-delta/src/main/java/com/facebook/presto/delta/DeltaPageSourceProvider.java b/presto-delta/src/main/java/com/facebook/presto/delta/DeltaPageSourceProvider.java
index 1355d6256337b..8d218275d6c6d 100644
--- a/presto-delta/src/main/java/com/facebook/presto/delta/DeltaPageSourceProvider.java
+++ b/presto-delta/src/main/java/com/facebook/presto/delta/DeltaPageSourceProvider.java
@@ -139,7 +139,8 @@ public ConnectorPageSource createPageSource(
             ConnectorSplit split,
             ConnectorTableLayoutHandle layout,
             List<ColumnHandle> columns,
-            SplitContext splitContext)
+            SplitContext splitContext,
+            RuntimeStats runtimeStats)
     {
         DeltaSplit deltaSplit = (DeltaSplit) split;
         DeltaTableLayoutHandle deltaTableLayoutHandle = (DeltaTableLayoutHandle) layout;
diff --git a/presto-elasticsearch/src/main/java/com/facebook/presto/elasticsearch/ElasticsearchPageSourceProvider.java b/presto-elasticsearch/src/main/java/com/facebook/presto/elasticsearch/ElasticsearchPageSourceProvider.java
index 82c8ce7e5a696..c1b67b4178712 100644
--- a/presto-elasticsearch/src/main/java/com/facebook/presto/elasticsearch/ElasticsearchPageSourceProvider.java
+++ b/presto-elasticsearch/src/main/java/com/facebook/presto/elasticsearch/ElasticsearchPageSourceProvider.java
@@ -13,6 +13,7 @@
  */
 package com.facebook.presto.elasticsearch;
 
+import com.facebook.presto.common.RuntimeStats;
 import com.facebook.presto.common.type.StandardTypes;
 import com.facebook.presto.common.type.Type;
 import com.facebook.presto.common.type.TypeManager;
@@ -55,7 +56,8 @@ public ConnectorPageSource createPageSource(
             ConnectorSplit split,
             ConnectorTableLayoutHandle layout,
             List<ColumnHandle> columns,
-            SplitContext splitContext)
+            SplitContext splitContext,
+            RuntimeStats runtimeStats)
     {
         requireNonNull(split, "split is null");
         requireNonNull(layout, "layout is null");
diff --git a/presto-hdfs-core/src/main/java/com/facebook/presto/hive/HiveFileContext.java b/presto-hdfs-core/src/main/java/com/facebook/presto/hive/HiveFileContext.java
index 345bc0716bd9d..e41c91690030f 100644
--- a/presto-hdfs-core/src/main/java/com/facebook/presto/hive/HiveFileContext.java
+++ b/presto-hdfs-core/src/main/java/com/facebook/presto/hive/HiveFileContext.java
@@ -48,7 +48,7 @@ public class HiveFileContext
     private final long modificationTime;
     private final boolean verboseRuntimeStatsEnabled;
 
-    private final RuntimeStats stats = new RuntimeStats();
+    private final RuntimeStats stats;
 
     public HiveFileContext(
             boolean cacheable,
@@ -59,6 +59,20 @@ public HiveFileContext(
             OptionalLong length,
             long modificationTime,
             boolean verboseRuntimeStatsEnabled)
+    {
+        this(cacheable, cacheQuota, extraFileInfo, fileSize, startOffset, length, modificationTime, verboseRuntimeStatsEnabled, new RuntimeStats());
+    }
+
+    public HiveFileContext(
+            boolean cacheable,
+            CacheQuota cacheQuota,
+            Optional<ExtraHiveFileInfo<?>> extraFileInfo,
+            OptionalLong fileSize,
+            OptionalLong startOffset,
+            OptionalLong length,
+            long modificationTime,
+            boolean verboseRuntimeStatsEnabled,
+            RuntimeStats runtimeStats)
     {
         this.cacheable = cacheable;
         this.cacheQuota = requireNonNull(cacheQuota, "cacheQuota is null");
@@ -68,6 +82,7 @@ public HiveFileContext(
         this.length = requireNonNull(length, "length is null");
         this.modificationTime = modificationTime;
         this.verboseRuntimeStatsEnabled = verboseRuntimeStatsEnabled;
+        this.stats = requireNonNull(runtimeStats, "runtimeStats is null");
     }
 
     /**
diff --git a/presto-hive/src/main/java/com/facebook/presto/hive/HivePageSourceProvider.java b/presto-hive/src/main/java/com/facebook/presto/hive/HivePageSourceProvider.java
index a99bcbcffddd1..9be6f7109b942 100644
--- a/presto-hive/src/main/java/com/facebook/presto/hive/HivePageSourceProvider.java
+++ b/presto-hive/src/main/java/com/facebook/presto/hive/HivePageSourceProvider.java
@@ -13,6 +13,7 @@
  */
 package com.facebook.presto.hive;
 
+import com.facebook.presto.common.RuntimeStats;
 import com.facebook.presto.common.Subfield;
 import com.facebook.presto.common.Subfield.NestedField;
 import com.facebook.presto.common.Subfield.PathElement;
@@ -137,7 +138,8 @@ public ConnectorPageSource createPageSource(
             ConnectorSplit split,
             ConnectorTableLayoutHandle layout,
             List<ColumnHandle> columns,
-            SplitContext splitContext)
+            SplitContext splitContext,
+            RuntimeStats runtimeStats)
     {
         HiveTableLayoutHandle hiveLayout = (HiveTableLayoutHandle) layout;
 
@@ -167,7 +169,8 @@ public ConnectorPageSource createPageSource(
                 OptionalLong.of(hiveSplit.getFileSplit().getStart()),
                 OptionalLong.of(hiveSplit.getFileSplit().getLength()),
                 hiveSplit.getFileSplit().getFileModifiedTime(),
-                HiveSessionProperties.isVerboseRuntimeStatsEnabled(session));
+                HiveSessionProperties.isVerboseRuntimeStatsEnabled(session),
+                runtimeStats);
 
         if (columns.stream().anyMatch(columnHandle -> ((HiveColumnHandle) columnHandle).getColumnType().equals(AGGREGATED))) {
             checkArgument(columns.stream().allMatch(columnHandle -> ((HiveColumnHandle) columnHandle).getColumnType().equals(AGGREGATED)), "Not all columns are of 'AGGREGATED' type");
diff --git a/presto-hive/src/test/java/com/facebook/presto/hive/AbstractTestHiveClient.java b/presto-hive/src/test/java/com/facebook/presto/hive/AbstractTestHiveClient.java
index 1eea4e3e6da7c..6d36b4863cf35 100644
--- a/presto-hive/src/test/java/com/facebook/presto/hive/AbstractTestHiveClient.java
+++ b/presto-hive/src/test/java/com/facebook/presto/hive/AbstractTestHiveClient.java
@@ -2196,7 +2196,7 @@ private void doTestBucketedTableEvolutionWithDifferentReadCount(HiveStorageForma
 
             ImmutableList.Builder<MaterializedRow> allRows = ImmutableList.builder();
             for (ConnectorSplit split : splits) {
-                try (ConnectorPageSource pageSource = pageSourceProvider.createPageSource(transaction.getTransactionHandle(), session, split, tableHandle.getLayout().get(), columnHandles, NON_CACHEABLE)) {
+                try (ConnectorPageSource pageSource = pageSourceProvider.createPageSource(transaction.getTransactionHandle(), session, split, tableHandle.getLayout().get(), columnHandles, NON_CACHEABLE, new RuntimeStats())) {
                     MaterializedResult intermediateResult = materializeSourceDataStream(session, pageSource, getTypes(columnHandles));
                     allRows.addAll(intermediateResult.getMaterializedRows());
                 }
@@ -2441,7 +2441,7 @@ public void testGetRecords()
 
                 long rowNumber = 0;
                 long completedBytes = 0;
-                try (ConnectorPageSource pageSource = pageSourceProvider.createPageSource(transaction.getTransactionHandle(), session, hiveSplit, layoutHandle, columnHandles, NON_CACHEABLE)) {
+                try (ConnectorPageSource pageSource = pageSourceProvider.createPageSource(transaction.getTransactionHandle(), session, hiveSplit, layoutHandle, columnHandles, NON_CACHEABLE, new RuntimeStats())) {
                     MaterializedResult result = materializeSourceDataStream(session, pageSource, getTypes(columnHandles));
 
                     assertPageSourceType(pageSource, fileType);
@@ -2532,7 +2532,7 @@ public void testGetPartialRecords()
                 int dummyPartition = Integer.parseInt(partitionKeys.get(2).getValue().orElse(null));
 
                 long rowNumber = 0;
-                try (ConnectorPageSource pageSource = pageSourceProvider.createPageSource(transaction.getTransactionHandle(), session, hiveSplit, layoutHandle, columnHandles, NON_CACHEABLE)) {
+                try (ConnectorPageSource pageSource = pageSourceProvider.createPageSource(transaction.getTransactionHandle(), session, hiveSplit, layoutHandle, columnHandles, NON_CACHEABLE, new RuntimeStats())) {
                     assertPageSourceType(pageSource, fileType);
                     MaterializedResult result = materializeSourceDataStream(session, pageSource, getTypes(columnHandles));
                     for (MaterializedRow row : result) {
@@ -2571,7 +2571,7 @@ public void testGetRecordsUnpartitioned()
                 assertEquals(hiveSplit.getPartitionKeys(), ImmutableList.of());
 
                 long rowNumber = 0;
-                try (ConnectorPageSource pageSource = pageSourceProvider.createPageSource(transaction.getTransactionHandle(), session, split, layoutHandle, columnHandles, NON_CACHEABLE)) {
+                try (ConnectorPageSource pageSource = pageSourceProvider.createPageSource(transaction.getTransactionHandle(), session, split, layoutHandle, columnHandles, NON_CACHEABLE, new RuntimeStats())) {
                     assertPageSourceType(pageSource, TEXTFILE);
                     MaterializedResult result = materializeSourceDataStream(session, pageSource, getTypes(columnHandles));
 
@@ -2639,7 +2639,7 @@ public void testPartitionSchemaNonCanonical()
             ConnectorSplit split = getOnlyElement(getAllSplits(splitSource));
 
             ImmutableList<ColumnHandle> columnHandles = ImmutableList.of(column);
-            try (ConnectorPageSource ignored = pageSourceProvider.createPageSource(transaction.getTransactionHandle(), session, split, layoutHandle, columnHandles, NON_CACHEABLE)) {
+            try (ConnectorPageSource ignored = pageSourceProvider.createPageSource(transaction.getTransactionHandle(), session, split, layoutHandle, columnHandles, NON_CACHEABLE, new RuntimeStats())) {
                 fail("expected exception");
             }
             catch (PrestoException e) {
@@ -3149,7 +3149,7 @@ private void doTestBucketSortedTables(SchemaTableName table, boolean useTempPath
 
             int actualRowCount = 0;
             for (ConnectorSplit split : splits) {
-                try (ConnectorPageSource pageSource = pageSourceProvider.createPageSource(transaction.getTransactionHandle(), session, split, layoutHandle, columnHandles, NON_CACHEABLE)) {
+                try (ConnectorPageSource pageSource = pageSourceProvider.createPageSource(transaction.getTransactionHandle(), session, split, layoutHandle, columnHandles, NON_CACHEABLE, new RuntimeStats())) {
                     String lastValueAsc = null;
                     long lastValueDesc = -1;
 
@@ -4971,7 +4971,7 @@ protected void assertGetRecords(String tableName, HiveStorageFormat hiveStorageF
 
             List<ColumnHandle> columnHandles = ImmutableList.copyOf(metadata.getColumnHandles(session, tableHandle).values());
 
-            ConnectorPageSource pageSource = pageSourceProvider.createPageSource(transaction.getTransactionHandle(), session, hiveSplit, layoutHandle, columnHandles, NON_CACHEABLE);
+            ConnectorPageSource pageSource = pageSourceProvider.createPageSource(transaction.getTransactionHandle(), session, hiveSplit, layoutHandle, columnHandles, NON_CACHEABLE, new RuntimeStats());
             assertGetRecords(hiveStorageFormat, tableMetadata, hiveSplit, pageSource, columnHandles);
         }
     }
@@ -5261,7 +5261,7 @@ protected MaterializedResult readTable(
 
         ImmutableList.Builder<MaterializedRow> allRows = ImmutableList.builder();
         for (ConnectorSplit split : splits) {
-            try (ConnectorPageSource pageSource = pageSourceProvider.createPageSource(transaction.getTransactionHandle(), session, split, tableHandle.getLayout().get(), columnHandles, NON_CACHEABLE)) {
+            try (ConnectorPageSource pageSource = pageSourceProvider.createPageSource(transaction.getTransactionHandle(), session, split, tableHandle.getLayout().get(), columnHandles, NON_CACHEABLE, new RuntimeStats())) {
                 expectedStorageFormat.ifPresent(format -> assertPageSourceType(pageSource, format));
                 MaterializedResult result = materializeSourceDataStream(session, pageSource, getTypes(columnHandles));
                 allRows.addAll(result.getMaterializedRows());
diff --git a/presto-hive/src/test/java/com/facebook/presto/hive/AbstractTestHiveFileSystem.java b/presto-hive/src/test/java/com/facebook/presto/hive/AbstractTestHiveFileSystem.java
index 865e770156124..a692ba3d9ee44 100644
--- a/presto-hive/src/test/java/com/facebook/presto/hive/AbstractTestHiveFileSystem.java
+++ b/presto-hive/src/test/java/com/facebook/presto/hive/AbstractTestHiveFileSystem.java
@@ -17,6 +17,7 @@
 import com.facebook.airlift.stats.CounterStat;
 import com.facebook.presto.GroupByHashPageIndexerFactory;
 import com.facebook.presto.cache.CacheConfig;
+import com.facebook.presto.common.RuntimeStats;
 import com.facebook.presto.hive.AbstractTestHiveClient.HiveTransaction;
 import com.facebook.presto.hive.AbstractTestHiveClient.Transaction;
 import com.facebook.presto.hive.authentication.NoHdfsAuthentication;
@@ -480,7 +481,8 @@ private void createTable(MetastoreContext metastoreContext, SchemaTableName tabl
                     split,
                     tableHandle.getLayout().get(),
                     columnHandles,
-                    NON_CACHEABLE)) {
+                    NON_CACHEABLE,
+                    new RuntimeStats())) {
                 MaterializedResult result = materializeSourceDataStream(session, pageSource, getTypes(columnHandles));
                 assertEqualsIgnoreOrder(result.getMaterializedRows(), data.getMaterializedRows());
             }
diff --git a/presto-hive/src/test/java/com/facebook/presto/hive/HiveFileSystemTestUtils.java b/presto-hive/src/test/java/com/facebook/presto/hive/HiveFileSystemTestUtils.java
index 12f62c1e3bf30..0d509285ed16a 100644
--- a/presto-hive/src/test/java/com/facebook/presto/hive/HiveFileSystemTestUtils.java
+++ b/presto-hive/src/test/java/com/facebook/presto/hive/HiveFileSystemTestUtils.java
@@ -13,6 +13,7 @@
  */
 package com.facebook.presto.hive;
 
+import com.facebook.presto.common.RuntimeStats;
 import com.facebook.presto.common.type.Type;
 import com.facebook.presto.hive.AbstractTestHiveClient.HiveTransaction;
 import com.facebook.presto.hive.AbstractTestHiveClient.Transaction;
@@ -95,7 +96,8 @@ public static MaterializedResult readTable(SchemaTableName tableName,
                         split,
                         tableHandle.getLayout().get(),
                         columnHandles,
-                        NON_CACHEABLE)) {
+                        NON_CACHEABLE,
+                        new RuntimeStats())) {
                     MaterializedResult pageSourceResult = materializeSourceDataStream(session, pageSource, allTypes);
                     for (MaterializedRow row : pageSourceResult.getMaterializedRows()) {
                         Object[] dataValues = IntStream.range(0, row.getFieldCount())
@@ -153,7 +155,8 @@ public static MaterializedResult filterTable(SchemaTableName tableName,
                         split,
                         tableHandle.getLayout().get(),
                         projectedColumns,
-                        NON_CACHEABLE)) {
+                        NON_CACHEABLE,
+                        new RuntimeStats())) {
                     MaterializedResult pageSourceResult = materializeSourceDataStream(session, pageSource, allTypes);
                     for (MaterializedRow row : pageSourceResult.getMaterializedRows()) {
                         Object[] dataValues = IntStream.range(0, row.getFieldCount())
diff --git a/presto-hive/src/test/java/com/facebook/presto/hive/TestDynamicPruning.java b/presto-hive/src/test/java/com/facebook/presto/hive/TestDynamicPruning.java
index 6890a7e3c316c..5ed9ae8f568cc 100644
--- a/presto-hive/src/test/java/com/facebook/presto/hive/TestDynamicPruning.java
+++ b/presto-hive/src/test/java/com/facebook/presto/hive/TestDynamicPruning.java
@@ -15,6 +15,7 @@
 
 import com.facebook.airlift.testing.TempFile;
 import com.facebook.presto.cache.CacheConfig;
+import com.facebook.presto.common.RuntimeStats;
 import com.facebook.presto.common.predicate.Domain;
 import com.facebook.presto.common.predicate.TupleDomain;
 import com.facebook.presto.common.type.StandardTypes;
@@ -198,7 +199,7 @@ config, createTestHdfsEnvironment(config, metastoreClientConfig),
                 getDefaultHiveAggregatedPageSourceFactories(config, metastoreClientConfig),
                 FUNCTION_AND_TYPE_MANAGER,
                 ROW_EXPRESSION_SERVICE);
-        return provider.createPageSource(transaction, getSession(config), split, tableHandle.getLayout().get(), ImmutableList.copyOf(getColumnHandles()), splitContext);
+        return provider.createPageSource(transaction, getSession(config), split, tableHandle.getLayout().get(), ImmutableList.copyOf(getColumnHandles()), splitContext, new RuntimeStats());
     }
 
     private static TupleDomain<ColumnHandle> getToSkipTupleDomain()
diff --git a/presto-hive/src/test/java/com/facebook/presto/hive/TestHivePageSink.java b/presto-hive/src/test/java/com/facebook/presto/hive/TestHivePageSink.java
index a6965fabcae80..c1e11a4338b96 100644
--- a/presto-hive/src/test/java/com/facebook/presto/hive/TestHivePageSink.java
+++ b/presto-hive/src/test/java/com/facebook/presto/hive/TestHivePageSink.java
@@ -17,6 +17,7 @@
 import com.facebook.presto.cache.CacheConfig;
 import com.facebook.presto.common.Page;
 import com.facebook.presto.common.PageBuilder;
+import com.facebook.presto.common.RuntimeStats;
 import com.facebook.presto.common.block.BlockBuilder;
 import com.facebook.presto.common.predicate.TupleDomain;
 import com.facebook.presto.common.type.Type;
@@ -304,7 +305,7 @@ private static ConnectorPageSource createPageSource(HiveTransactionHandle transa
                 getDefaultHiveAggregatedPageSourceFactories(config, metastoreClientConfig),
                 FUNCTION_AND_TYPE_MANAGER,
                 ROW_EXPRESSION_SERVICE);
-        return provider.createPageSource(transaction, getSession(config, new HiveCommonClientConfig()), split, tableHandle.getLayout().get(), ImmutableList.copyOf(getColumnHandles()), NON_CACHEABLE);
+        return provider.createPageSource(transaction, getSession(config, new HiveCommonClientConfig()), split, tableHandle.getLayout().get(), ImmutableList.copyOf(getColumnHandles()), NON_CACHEABLE, new RuntimeStats());
     }
 
     private static ConnectorPageSink createPageSink(HiveTransactionHandle transaction, HiveClientConfig config, MetastoreClientConfig metastoreClientConfig, ExtendedHiveMetastore metastore, Path outputPath, HiveWriterStats stats)
diff --git a/presto-hive/src/test/java/com/facebook/presto/hive/TestHivePageSourceProvider.java b/presto-hive/src/test/java/com/facebook/presto/hive/TestHivePageSourceProvider.java
index 80877ccc8e2ca..42a5a02d919f3 100644
--- a/presto-hive/src/test/java/com/facebook/presto/hive/TestHivePageSourceProvider.java
+++ b/presto-hive/src/test/java/com/facebook/presto/hive/TestHivePageSourceProvider.java
@@ -15,6 +15,7 @@
 
 import com.facebook.presto.cache.CacheConfig;
 import com.facebook.presto.common.Page;
+import com.facebook.presto.common.RuntimeStats;
 import com.facebook.presto.common.Subfield;
 import com.facebook.presto.common.predicate.TupleDomain;
 import com.facebook.presto.common.type.TestingTypeManager;
@@ -317,13 +318,15 @@ public void testNotUseRecordReaderWithInputFormatAnnotationWithoutCustomSplit()
     public void testUsesPageSourceForPartition()
     {
         HivePageSourceProvider pageSourceProvider = createPageSourceProvider();
+        RuntimeStats runtimeStats = new RuntimeStats();
         ConnectorPageSource pageSource = pageSourceProvider.createPageSource(
                 new HiveTransactionHandle(),
                 SESSION,
                 getHiveSplit(ORC),
                 getHiveTableLayout(false, false, false),
                 ImmutableList.of(LONG_COLUMN),
-                new SplitContext(false));
+                new SplitContext(false),
+                runtimeStats);
         assertTrue(pageSource instanceof HivePageSource, format("pageSource was %s", pageSource.getClass().getSimpleName()));
         assertTrue(((HivePageSource) pageSource).getPageSource() instanceof MockOrcBatchPageSource,
                 format("pageSoruce was %s", ((HivePageSource) pageSource).getPageSource().getClass().getSimpleName()));
@@ -334,7 +337,8 @@ public void testUsesPageSourceForPartition()
                 getHiveSplit(RCBINARY),
                 getHiveTableLayout(false, false, false),
                 ImmutableList.of(LONG_COLUMN),
-                new SplitContext(false));
+                new SplitContext(false),
+                runtimeStats);
         assertTrue(pageSource instanceof HivePageSource, format("pageSource  was %s", pageSource.getClass().getSimpleName()));
         assertTrue(((HivePageSource) pageSource).getPageSource() instanceof MockRcBinaryBatchPageSource,
                 format("pageSource  was %s", ((HivePageSource) pageSource).getPageSource().getClass().getSimpleName()));
@@ -345,7 +349,8 @@ public void testUsesPageSourceForPartition()
                 getHiveSplit(ORC),
                 getHiveTableLayout(true, false, false),
                 ImmutableList.of(LONG_COLUMN),
-                new SplitContext(false));
+                new SplitContext(false),
+                runtimeStats);
         assertTrue(pageSource instanceof MockOrcSelectivePageSource, format("pageSource  was %s", pageSource.getClass().getSimpleName()));
     }
 
@@ -359,7 +364,8 @@ public void testWrapsInFilteringPageSourceWhenNoSelectivePageSource()
                 getHiveSplit(RCBINARY),
                 getHiveTableLayout(true, false, false),
                 ImmutableList.of(),
-                new SplitContext(false));
+                new SplitContext(false),
+                new RuntimeStats());
         assertTrue(pageSource instanceof FilteringPageSource, format("pageSource was %s", pageSource.getClass().getSimpleName()));
     }
 
@@ -373,7 +379,8 @@ public void testAggregatedPageSource()
                 getHiveSplit(ORC),
                 getHiveTableLayout(true, true, false),
                 ImmutableList.of(LONG_AGGREGATED_COLUMN),
-                new SplitContext(false));
+                new SplitContext(false),
+                new RuntimeStats());
         assertTrue(pageSource instanceof MockOrcAggregatedPageSource, format("pageSource %s", pageSource.getClass().getSimpleName()));
     }
 
@@ -389,7 +396,8 @@ public void testFailsWhenNoAggregatedPageSourceAvailable()
                 getHiveSplit(RCBINARY),
                 getHiveTableLayout(false, true, false),
                 ImmutableList.of(LONG_AGGREGATED_COLUMN),
-                new SplitContext(false));
+                new SplitContext(false),
+                new RuntimeStats());
     }
 
     @Test(expectedExceptions = PrestoException.class,
@@ -405,7 +413,8 @@ public void testFailsWhenFooterStatsUnreliable()
                 getHiveSplit(ORC),
                 getHiveTableLayout(false, true, true),
                 ImmutableList.of(LONG_AGGREGATED_COLUMN),
-                new SplitContext(false));
+                new SplitContext(false),
+                new RuntimeStats());
     }
 
     @Test(expectedExceptions = IllegalArgumentException.class, expectedExceptionsMessageRegExp = "Not all columns are of 'AGGREGATED' type")
@@ -418,7 +427,8 @@ public void testFailsWhenMixOfAggregatedAndRegularColumns()
                 getHiveSplit(ORC),
                 getHiveTableLayout(false, true, false),
                 ImmutableList.of(LONG_COLUMN, LONG_AGGREGATED_COLUMN),
-                new SplitContext(false));
+                new SplitContext(false),
+                new RuntimeStats());
     }
 
     @Test
@@ -432,7 +442,8 @@ public void testCreatePageSource_withRowID()
                 hiveSplit,
                 getHiveTableLayout(false, true, false),
                 ImmutableList.of(LONG_COLUMN, HiveColumnHandle.rowIdColumnHandle()),
-                new SplitContext(false))) {
+                new SplitContext(false),
+                new RuntimeStats())) {
             assertEquals(0, pageSource.getCompletedBytes());
         }
     }
@@ -447,7 +458,8 @@ public void testCreatePageSource_withRowIDMissingPartitionComponent()
                 hiveSplit,
                 getHiveTableLayout(false, true, false),
                 ImmutableList.of(LONG_COLUMN, HiveColumnHandle.rowIdColumnHandle()),
-                new SplitContext(false));
+                new SplitContext(false),
+                new RuntimeStats());
     }
 
     private static ConnectorTableLayoutHandle getHiveTableLayout(boolean pushdownFilterEnabled, boolean partialAggregationsPushedDown, boolean footerStatsUnreliable)
diff --git a/presto-hive/src/test/java/com/facebook/presto/hive/TestOrcBatchPageSourceMemoryTracking.java b/presto-hive/src/test/java/com/facebook/presto/hive/TestOrcBatchPageSourceMemoryTracking.java
index fb7d52b43bcc0..30f02bb1a0477 100644
--- a/presto-hive/src/test/java/com/facebook/presto/hive/TestOrcBatchPageSourceMemoryTracking.java
+++ b/presto-hive/src/test/java/com/facebook/presto/hive/TestOrcBatchPageSourceMemoryTracking.java
@@ -499,7 +499,7 @@ public SourceOperator newTableScanOperator(DriverContext driverContext)
             SourceOperatorFactory sourceOperatorFactory = new TableScanOperatorFactory(
                     0,
                     new PlanNodeId("0"),
-                    (session, split, table, columnHandles) -> pageSource,
+                    (session, split, table, columnHandles, runtimeStats) -> pageSource,
                     table,
                     columns.stream().map(columnHandle -> (ColumnHandle) columnHandle).collect(toList()));
             SourceOperator operator = sourceOperatorFactory.createOperator(driverContext);
@@ -524,7 +524,7 @@ public SourceOperator newScanFilterAndProjectOperator(DriverContext driverContex
                     0,
                     new PlanNodeId("test"),
                     new PlanNodeId("0"),
-                    (session, split, table, columnHandles) -> pageSource,
+                    (session, split, table, columnHandles, runtimeStats) -> pageSource,
                     cursorProcessor,
                     pageProcessor,
                     table,
diff --git a/presto-hudi/src/main/java/com/facebook/presto/hudi/HudiPageSourceProvider.java b/presto-hudi/src/main/java/com/facebook/presto/hudi/HudiPageSourceProvider.java
index b512575978e88..a67174d8419b3 100644
--- a/presto-hudi/src/main/java/com/facebook/presto/hudi/HudiPageSourceProvider.java
+++ b/presto-hudi/src/main/java/com/facebook/presto/hudi/HudiPageSourceProvider.java
@@ -14,6 +14,7 @@
 
 package com.facebook.presto.hudi;
 
+import com.facebook.presto.common.RuntimeStats;
 import com.facebook.presto.common.predicate.TupleDomain;
 import com.facebook.presto.common.type.Type;
 import com.facebook.presto.common.type.TypeManager;
@@ -75,7 +76,8 @@ public ConnectorPageSource createPageSource(
             ConnectorSplit split,
             ConnectorTableLayoutHandle layoutHandle,
             List<ColumnHandle> columns,
-            SplitContext splitContext)
+            SplitContext splitContext,
+            RuntimeStats runtimeStats)
     {
         HudiTableLayoutHandle layout = (HudiTableLayoutHandle) layoutHandle;
         HudiSplit hudiSplit = (HudiSplit) split;
diff --git a/presto-iceberg/src/main/java/com/facebook/presto/iceberg/IcebergPageSourceProvider.java b/presto-iceberg/src/main/java/com/facebook/presto/iceberg/IcebergPageSourceProvider.java
index b99ba6d3f7859..8b62406d4d5d6 100644
--- a/presto-iceberg/src/main/java/com/facebook/presto/iceberg/IcebergPageSourceProvider.java
+++ b/presto-iceberg/src/main/java/com/facebook/presto/iceberg/IcebergPageSourceProvider.java
@@ -715,7 +715,8 @@ public ConnectorPageSource createPageSource(
             ConnectorSplit connectorSplit,
             ConnectorTableLayoutHandle layout,
             List<ColumnHandle> desiredColumns,
-            SplitContext splitContext)
+            SplitContext splitContext,
+            RuntimeStats runtimeStats)
     {
         IcebergTableLayoutHandle icebergLayout = (IcebergTableLayoutHandle) layout;
         if (icebergLayout.isPushdownFilterEnabled()) {
diff --git a/presto-main/src/main/java/com/facebook/presto/connector/system/GlobalSystemConnector.java b/presto-main/src/main/java/com/facebook/presto/connector/system/GlobalSystemConnector.java
index 631eb6b1e70b4..7a4ff87f54dec 100644
--- a/presto-main/src/main/java/com/facebook/presto/connector/system/GlobalSystemConnector.java
+++ b/presto-main/src/main/java/com/facebook/presto/connector/system/GlobalSystemConnector.java
@@ -13,6 +13,7 @@
  */
 package com.facebook.presto.connector.system;
 
+import com.facebook.presto.common.RuntimeStats;
 import com.facebook.presto.common.transaction.TransactionId;
 import com.facebook.presto.spi.ColumnHandle;
 import com.facebook.presto.spi.ColumnMetadata;
@@ -143,7 +144,7 @@ public ConnectorPageSourceProvider getPageSourceProvider()
     {
         return new ConnectorPageSourceProvider() {
             @Override
-            public ConnectorPageSource createPageSource(ConnectorTransactionHandle transactionHandle, ConnectorSession session, ConnectorSplit split, ConnectorTableLayoutHandle layout, List<ColumnHandle> columns, SplitContext splitContext)
+            public ConnectorPageSource createPageSource(ConnectorTransactionHandle transactionHandle, ConnectorSession session, ConnectorSplit split, ConnectorTableLayoutHandle layout, List<ColumnHandle> columns, SplitContext splitContext, RuntimeStats runtimeStats)
             {
                 throw new UnsupportedOperationException();
             }
diff --git a/presto-main/src/main/java/com/facebook/presto/operator/ScanFilterAndProjectOperator.java b/presto-main/src/main/java/com/facebook/presto/operator/ScanFilterAndProjectOperator.java
index c078a3491b435..56d24cdb00292 100644
--- a/presto-main/src/main/java/com/facebook/presto/operator/ScanFilterAndProjectOperator.java
+++ b/presto-main/src/main/java/com/facebook/presto/operator/ScanFilterAndProjectOperator.java
@@ -247,7 +247,7 @@ public Page getOutput()
         }
 
         if (!finishing && pageSource == null && cursor == null) {
-            ConnectorPageSource source = pageSourceProvider.createPageSource(operatorContext.getSession(), split, dynamicFilterSupplier.map(table::withDynamicFilter).orElse(table), columns);
+            ConnectorPageSource source = pageSourceProvider.createPageSource(operatorContext.getSession(), split, dynamicFilterSupplier.map(table::withDynamicFilter).orElse(table), columns, operatorContext.getRuntimeStats());
             if (source instanceof RecordPageSource) {
                 cursor = ((RecordPageSource) source).getCursor();
             }
diff --git a/presto-main/src/main/java/com/facebook/presto/operator/TableScanOperator.java b/presto-main/src/main/java/com/facebook/presto/operator/TableScanOperator.java
index 8b39f69e371d8..c97e9dc2fff60 100644
--- a/presto-main/src/main/java/com/facebook/presto/operator/TableScanOperator.java
+++ b/presto-main/src/main/java/com/facebook/presto/operator/TableScanOperator.java
@@ -260,7 +260,7 @@ public Page getOutput()
             return null;
         }
         if (source == null) {
-            source = pageSourceProvider.createPageSource(operatorContext.getSession(), split, table, columns);
+            source = pageSourceProvider.createPageSource(operatorContext.getSession(), split, table, columns, operatorContext.getRuntimeStats());
         }
 
         Page page = source.getNextPage();
diff --git a/presto-main/src/main/java/com/facebook/presto/split/PageSourceManager.java b/presto-main/src/main/java/com/facebook/presto/split/PageSourceManager.java
index bef904072a944..57f6348fbf51f 100644
--- a/presto-main/src/main/java/com/facebook/presto/split/PageSourceManager.java
+++ b/presto-main/src/main/java/com/facebook/presto/split/PageSourceManager.java
@@ -14,6 +14,7 @@
 package com.facebook.presto.split;
 
 import com.facebook.presto.Session;
+import com.facebook.presto.common.RuntimeStats;
 import com.facebook.presto.common.predicate.TupleDomain;
 import com.facebook.presto.metadata.Split;
 import com.facebook.presto.spi.ColumnHandle;
@@ -54,7 +55,7 @@ public void removeConnectorPageSourceProvider(ConnectorId connectorId)
     }
 
     @Override
-    public ConnectorPageSource createPageSource(Session session, Split split, TableHandle table, List<ColumnHandle> columns)
+    public ConnectorPageSource createPageSource(Session session, Split split, TableHandle table, List<ColumnHandle> columns, RuntimeStats runtimeStats)
     {
         requireNonNull(split, "split is null");
         requireNonNull(columns, "columns is null");
@@ -83,7 +84,8 @@ public ConnectorPageSource createPageSource(Session session, Split split, TableH
                     split.getConnectorSplit(),
                     table.getLayout().get(),
                     columns,
-                    split.getSplitContext());
+                    split.getSplitContext(),
+                    runtimeStats);
         }
         return getPageSourceProvider(split).createPageSource(split.getTransactionHandle(), connectorSession, split.getConnectorSplit(), columns, split.getSplitContext());
     }
diff --git a/presto-main/src/main/java/com/facebook/presto/split/PageSourceProvider.java b/presto-main/src/main/java/com/facebook/presto/split/PageSourceProvider.java
index 1fb802f239332..3fe7d3614467a 100644
--- a/presto-main/src/main/java/com/facebook/presto/split/PageSourceProvider.java
+++ b/presto-main/src/main/java/com/facebook/presto/split/PageSourceProvider.java
@@ -14,6 +14,7 @@
 package com.facebook.presto.split;
 
 import com.facebook.presto.Session;
+import com.facebook.presto.common.RuntimeStats;
 import com.facebook.presto.metadata.Split;
 import com.facebook.presto.spi.ColumnHandle;
 import com.facebook.presto.spi.ConnectorPageSource;
@@ -23,5 +24,5 @@
 
 public interface PageSourceProvider
 {
-    ConnectorPageSource createPageSource(Session session, Split split, TableHandle table, List<ColumnHandle> columns);
+    ConnectorPageSource createPageSource(Session session, Split split, TableHandle table, List<ColumnHandle> columns, RuntimeStats runtimeStats);
 }
diff --git a/presto-main/src/main/java/com/facebook/presto/sql/planner/iterative/rule/ExtractSpatialJoins.java b/presto-main/src/main/java/com/facebook/presto/sql/planner/iterative/rule/ExtractSpatialJoins.java
index fe812f9dae349..acd3c2a6ab576 100644
--- a/presto-main/src/main/java/com/facebook/presto/sql/planner/iterative/rule/ExtractSpatialJoins.java
+++ b/presto-main/src/main/java/com/facebook/presto/sql/planner/iterative/rule/ExtractSpatialJoins.java
@@ -538,7 +538,7 @@ private static KdbTree loadKdbTree(String tableName, Session session, Metadata m
                 List<Split> splits = splitBatch.getSplits();
 
                 for (Split split : splits) {
-                    try (ConnectorPageSource pageSource = pageSourceManager.createPageSource(session, split, newTableHandle, ImmutableList.of(kdbTreeColumn))) {
+                    try (ConnectorPageSource pageSource = pageSourceManager.createPageSource(session, split, newTableHandle, ImmutableList.of(kdbTreeColumn), session.getRuntimeStats())) {
                         do {
                             getFutureValue(pageSource.isBlocked());
                             Page page = pageSource.getNextPage();
diff --git a/presto-main/src/test/java/com/facebook/presto/memory/TestSystemMemoryBlocking.java b/presto-main/src/test/java/com/facebook/presto/memory/TestSystemMemoryBlocking.java
index ea611be3c1460..f49952457825f 100644
--- a/presto-main/src/test/java/com/facebook/presto/memory/TestSystemMemoryBlocking.java
+++ b/presto-main/src/test/java/com/facebook/presto/memory/TestSystemMemoryBlocking.java
@@ -103,7 +103,7 @@ public void testTableScanSystemMemoryBlocking()
         final List<Type> types = ImmutableList.of(VARCHAR);
         TableScanOperator source = new TableScanOperator(driverContext.addOperatorContext(1, new PlanNodeId("test"), "values"),
                 sourceId,
-                (session, split, table, columns) -> new FixedPageSource(rowPagesBuilder(types)
+                (session, split, table, columns, runtimeStats) -> new FixedPageSource(rowPagesBuilder(types)
                         .addSequencePage(10, 1)
                         .addSequencePage(10, 1)
                         .addSequencePage(10, 1)
diff --git a/presto-main/src/test/java/com/facebook/presto/operator/TestDriver.java b/presto-main/src/test/java/com/facebook/presto/operator/TestDriver.java
index 6e04b4a7b4da4..2468ce35a3262 100644
--- a/presto-main/src/test/java/com/facebook/presto/operator/TestDriver.java
+++ b/presto-main/src/test/java/com/facebook/presto/operator/TestDriver.java
@@ -249,7 +249,7 @@ private void testAddSourceFinish(DriverContext driverContext)
         final List<Type> types = ImmutableList.of(VARCHAR, BIGINT, BIGINT);
         TableScanOperator source = new TableScanOperator(driverContext.addOperatorContext(99, new PlanNodeId("test"), "values"),
                 sourceId,
-                (session, split, table, columns) -> new FixedPageSource(rowPagesBuilder(types)
+                (session, split, table, columns, runtimeStats) -> new FixedPageSource(rowPagesBuilder(types)
                         .addSequencePage(10, 20, 30, 40)
                         .build()),
                 TESTING_TABLE_HANDLE,
@@ -349,7 +349,7 @@ private void testMemoryRevocationRace(DriverContext driverContext)
         List<Type> types = ImmutableList.of(VARCHAR, BIGINT, BIGINT);
         TableScanOperator source = new AlwaysBlockedMemoryRevokingTableScanOperator(driverContext.addOperatorContext(99, new PlanNodeId("test"), "scan"),
                 new PlanNodeId("source"),
-                (session, split, table, columns) -> new FixedPageSource(rowPagesBuilder(types)
+                (session, split, table, columns, runtimeStats) -> new FixedPageSource(rowPagesBuilder(types)
                         .addSequencePage(10, 20, 30, 40)
                         .build()),
                 TESTING_TABLE_HANDLE,
@@ -378,7 +378,7 @@ private void testBrokenOperatorAddSource(DriverContext driverContext)
         // create a table scan operator that does not block, which will cause the driver loop to busy wait
         TableScanOperator source = new NotBlockedTableScanOperator(driverContext.addOperatorContext(99, new PlanNodeId("test"), "values"),
                 sourceId,
-                (session, split, table, columns) -> new FixedPageSource(rowPagesBuilder(types)
+                (session, split, table, columns, runtimeStats) -> new FixedPageSource(rowPagesBuilder(types)
                         .addSequencePage(10, 20, 30, 40)
                         .build()),
                 TESTING_TABLE_HANDLE,
@@ -435,7 +435,7 @@ private void processSourceDriver(DriverContext driverContext)
         final List<Type> types = ImmutableList.of(VARCHAR, BIGINT, BIGINT);
         TableScanOperator source = new TableScanOperator(driverContext.addOperatorContext(99, new PlanNodeId("test"), "values"),
                 sourceId,
-                (session, split, table, columns) -> new FixedPageSource(rowPagesBuilder(types)
+                (session, split, table, columns, runtimeStats) -> new FixedPageSource(rowPagesBuilder(types)
                         .addSequencePage(10, 20, 30, 40)
                         .build()),
                 TESTING_TABLE_HANDLE,
diff --git a/presto-main/src/test/java/com/facebook/presto/operator/TestScanFilterAndProjectOperator.java b/presto-main/src/test/java/com/facebook/presto/operator/TestScanFilterAndProjectOperator.java
index 199dd85f7bf8e..65a4cbccae0e3 100644
--- a/presto-main/src/test/java/com/facebook/presto/operator/TestScanFilterAndProjectOperator.java
+++ b/presto-main/src/test/java/com/facebook/presto/operator/TestScanFilterAndProjectOperator.java
@@ -121,7 +121,7 @@ public void testPageSource()
                 0,
                 new PlanNodeId("test"),
                 new PlanNodeId("0"),
-                (session, split, table, columns) -> new FixedPageSource(ImmutableList.of(input)),
+                (session, split, table, columns, runtimeStats) -> new FixedPageSource(ImmutableList.of(input)),
                 cursorProcessor,
                 pageProcessor,
                 TESTING_TABLE_HANDLE,
@@ -170,7 +170,7 @@ public void testPageSourceMergeOutput()
                 0,
                 new PlanNodeId("test"),
                 new PlanNodeId("0"),
-                (session, split, table, columns) -> new FixedPageSource(input),
+                (session, split, table, columns, runtimeStats) -> new FixedPageSource(input),
                 cursorProcessor,
                 pageProcessor,
                 TESTING_TABLE_HANDLE,
@@ -220,7 +220,7 @@ public void testPageSourceLazyLoad()
                 0,
                 new PlanNodeId("test"),
                 new PlanNodeId("0"),
-                (session, split, table, columns) -> new SinglePagePageSource(input),
+                (session, split, table, columns, runtimeStats) -> new SinglePagePageSource(input),
                 cursorProcessor,
                 () -> pageProcessor,
                 TESTING_TABLE_HANDLE,
@@ -263,7 +263,7 @@ public void testPageSourceLazyBlock()
                 0,
                 new PlanNodeId("test"),
                 new PlanNodeId("0"),
-                (session, split, table, columns) -> pageSource,
+                (session, split, table, columns, runtimeStats) -> pageSource,
                 cursorProcessor,
                 pageProcessor,
                 TESTING_TABLE_HANDLE,
@@ -312,7 +312,7 @@ public void testRecordCursorSource()
                 0,
                 new PlanNodeId("test"),
                 new PlanNodeId("0"),
-                (session, split, table, columns) -> new RecordPageSource(new PageRecordSet(ImmutableList.of(VARCHAR), input)),
+                (session, split, table, columns, runtimeStats) -> new RecordPageSource(new PageRecordSet(ImmutableList.of(VARCHAR), input)),
                 cursorProcessor,
                 pageProcessor,
                 TESTING_TABLE_HANDLE,
@@ -370,7 +370,7 @@ public void testPageYield()
                 0,
                 new PlanNodeId("test"),
                 new PlanNodeId("0"),
-                (session, split, table, columns) -> new FixedPageSource(ImmutableList.of(input)),
+                (session, split, table, columns, runtimeStats) -> new FixedPageSource(ImmutableList.of(input)),
                 cursorProcessor,
                 pageProcessor,
                 TESTING_TABLE_HANDLE,
@@ -443,7 +443,7 @@ public void testRecordCursorYield()
                 0,
                 new PlanNodeId("test"),
                 new PlanNodeId("0"),
-                (session, split, table, columns) -> new RecordPageSource(new PageRecordSet(ImmutableList.of(BIGINT), input)),
+                (session, split, table, columns, runtimeStats) -> new RecordPageSource(new PageRecordSet(ImmutableList.of(BIGINT), input)),
                 cursorProcessor,
                 pageProcessor,
                 TESTING_TABLE_HANDLE,
diff --git a/presto-main/src/test/java/com/facebook/presto/operator/scalar/FunctionAssertions.java b/presto-main/src/test/java/com/facebook/presto/operator/scalar/FunctionAssertions.java
index 36f27421f0660..04c939af96f08 100644
--- a/presto-main/src/test/java/com/facebook/presto/operator/scalar/FunctionAssertions.java
+++ b/presto-main/src/test/java/com/facebook/presto/operator/scalar/FunctionAssertions.java
@@ -17,6 +17,7 @@
 import com.facebook.presto.common.InvalidTypeDefinitionException;
 import com.facebook.presto.common.Page;
 import com.facebook.presto.common.PageBuilder;
+import com.facebook.presto.common.RuntimeStats;
 import com.facebook.presto.common.Utils;
 import com.facebook.presto.common.block.Block;
 import com.facebook.presto.common.function.SqlFunctionProperties;
@@ -1098,7 +1099,7 @@ private static class TestPageSourceProvider
             implements PageSourceProvider
     {
         @Override
-        public ConnectorPageSource createPageSource(Session session, Split split, TableHandle table, List<ColumnHandle> columns)
+        public ConnectorPageSource createPageSource(Session session, Split split, TableHandle table, List<ColumnHandle> columns, RuntimeStats runtimeStats)
         {
             assertInstanceOf(split.getConnectorSplit(), FunctionAssertions.TestSplit.class);
             FunctionAssertions.TestSplit testSplit = (FunctionAssertions.TestSplit) split.getConnectorSplit();
diff --git a/presto-pinot-toolkit/src/main/java/com/facebook/presto/pinot/PinotPageSourceProvider.java b/presto-pinot-toolkit/src/main/java/com/facebook/presto/pinot/PinotPageSourceProvider.java
index b87d4317ff3e3..81ae13ae533f0 100644
--- a/presto-pinot-toolkit/src/main/java/com/facebook/presto/pinot/PinotPageSourceProvider.java
+++ b/presto-pinot-toolkit/src/main/java/com/facebook/presto/pinot/PinotPageSourceProvider.java
@@ -13,6 +13,7 @@
  */
 package com.facebook.presto.pinot;
 
+import com.facebook.presto.common.RuntimeStats;
 import com.facebook.presto.pinot.auth.PinotBrokerAuthenticationProvider;
 import com.facebook.presto.spi.ColumnHandle;
 import com.facebook.presto.spi.ConnectorId;
@@ -73,7 +74,8 @@ public ConnectorPageSource createPageSource(
             ConnectorSplit split,
             ConnectorTableLayoutHandle tableLayoutHandle,
             List<ColumnHandle> columns,
-            SplitContext splitContext)
+            SplitContext splitContext,
+            RuntimeStats runtimeStats)
     {
         requireNonNull(split, "split is null");
 
diff --git a/presto-spi/src/main/java/com/facebook/presto/spi/connector/ConnectorPageSourceProvider.java b/presto-spi/src/main/java/com/facebook/presto/spi/connector/ConnectorPageSourceProvider.java
index dd30d8f901a2a..8185370cb6221 100644
--- a/presto-spi/src/main/java/com/facebook/presto/spi/connector/ConnectorPageSourceProvider.java
+++ b/presto-spi/src/main/java/com/facebook/presto/spi/connector/ConnectorPageSourceProvider.java
@@ -13,6 +13,7 @@
  */
 package com.facebook.presto.spi.connector;
 
+import com.facebook.presto.common.RuntimeStats;
 import com.facebook.presto.spi.ColumnHandle;
 import com.facebook.presto.spi.ConnectorPageSource;
 import com.facebook.presto.spi.ConnectorSession;
@@ -40,6 +41,7 @@ default ConnectorPageSource createPageSource(
 
     /**
      * @param columns columns that should show up in the output page, in this order
+     * @param runtimeStats
      */
     default ConnectorPageSource createPageSource(
             ConnectorTransactionHandle transactionHandle,
@@ -47,7 +49,8 @@ default ConnectorPageSource createPageSource(
             ConnectorSplit split,
             ConnectorTableLayoutHandle layout,
             List<ColumnHandle> columns,
-            SplitContext splitContext)
+            SplitContext splitContext,
+            RuntimeStats runtimeStats)
     {
         return createPageSource(transactionHandle, session, split, columns, splitContext);
     }
diff --git a/presto-spi/src/main/java/com/facebook/presto/spi/connector/classloader/ClassLoaderSafeConnectorPageSourceProvider.java b/presto-spi/src/main/java/com/facebook/presto/spi/connector/classloader/ClassLoaderSafeConnectorPageSourceProvider.java
index be64b36e43a5a..b44d378515c0b 100644
--- a/presto-spi/src/main/java/com/facebook/presto/spi/connector/classloader/ClassLoaderSafeConnectorPageSourceProvider.java
+++ b/presto-spi/src/main/java/com/facebook/presto/spi/connector/classloader/ClassLoaderSafeConnectorPageSourceProvider.java
@@ -13,6 +13,7 @@
  */
 package com.facebook.presto.spi.connector.classloader;
 
+import com.facebook.presto.common.RuntimeStats;
 import com.facebook.presto.spi.ColumnHandle;
 import com.facebook.presto.spi.ConnectorPageSource;
 import com.facebook.presto.spi.ConnectorSession;
@@ -57,10 +58,11 @@ public ConnectorPageSource createPageSource(
             ConnectorSplit split,
             ConnectorTableLayoutHandle layout,
             List<ColumnHandle> columns,
-            SplitContext splitContext)
+            SplitContext splitContext,
+            RuntimeStats runtimeStats)
     {
         try (ThreadContextClassLoader ignored = new ThreadContextClassLoader(classLoader)) {
-            return delegate.createPageSource(transactionHandle, session, split, layout, columns, splitContext);
+            return delegate.createPageSource(transactionHandle, session, split, layout, columns, splitContext, runtimeStats);
         }
     }
 }
diff --git a/presto-tests/src/test/java/com/facebook/presto/execution/TestBeginQuery.java b/presto-tests/src/test/java/com/facebook/presto/execution/TestBeginQuery.java
index 22a4fdc06f253..335b697bec789 100644
--- a/presto-tests/src/test/java/com/facebook/presto/execution/TestBeginQuery.java
+++ b/presto-tests/src/test/java/com/facebook/presto/execution/TestBeginQuery.java
@@ -14,6 +14,7 @@
 package com.facebook.presto.execution;
 
 import com.facebook.presto.Session;
+import com.facebook.presto.common.RuntimeStats;
 import com.facebook.presto.spi.ColumnHandle;
 import com.facebook.presto.spi.ConnectorHandleResolver;
 import com.facebook.presto.spi.ConnectorPageSource;
@@ -218,7 +219,7 @@ public ConnectorPageSourceProvider getPageSourceProvider()
         {
             return new ConnectorPageSourceProvider() {
                 @Override
-                public ConnectorPageSource createPageSource(ConnectorTransactionHandle transactionHandle, ConnectorSession session, ConnectorSplit split, ConnectorTableLayoutHandle layout, List<ColumnHandle> columns, SplitContext splitContext)
+                public ConnectorPageSource createPageSource(ConnectorTransactionHandle transactionHandle, ConnectorSession session, ConnectorSplit split, ConnectorTableLayoutHandle layout, List<ColumnHandle> columns, SplitContext splitContext, RuntimeStats runtimeStats)
                 {
                     return new FixedPageSource(ImmutableList.of());
                 }