diff --git a/api/src/main/java/org/apache/iceberg/BatchScan.java b/api/src/main/java/org/apache/iceberg/BatchScan.java
index bd53fe97a316..76577eee0261 100644
--- a/api/src/main/java/org/apache/iceberg/BatchScan.java
+++ b/api/src/main/java/org/apache/iceberg/BatchScan.java
@@ -18,6 +18,7 @@
  */
 package org.apache.iceberg;
 
+import java.util.function.Supplier;
 import org.apache.iceberg.io.FileIO;
 
 /** API for configuring a batch scan. */
@@ -73,7 +74,7 @@ public interface BatchScan extends Scan<BatchScan, ScanTask, ScanTaskGroup<ScanT
 
   /** Returns the {@link FileIO} instance to use when reading data files for this scan. */
   @Override
-  default FileIO io() {
-    return table().io();
+  default Supplier<FileIO> fileIO() {
+    return table()::io;
   }
 }
diff --git a/api/src/main/java/org/apache/iceberg/BatchScanAdapter.java b/api/src/main/java/org/apache/iceberg/BatchScanAdapter.java
index cccd1cd16253..8af1144decc5 100644
--- a/api/src/main/java/org/apache/iceberg/BatchScanAdapter.java
+++ b/api/src/main/java/org/apache/iceberg/BatchScanAdapter.java
@@ -20,6 +20,7 @@
 
 import java.util.Collection;
 import java.util.concurrent.ExecutorService;
+import java.util.function.Supplier;
 import org.apache.iceberg.expressions.Expression;
 import org.apache.iceberg.io.CloseableIterable;
 import org.apache.iceberg.io.FileIO;
@@ -40,8 +41,8 @@ public Table table() {
   }
 
   @Override
-  public FileIO io() {
-    return scan.io();
+  public Supplier<FileIO> fileIO() {
+    return scan.fileIO();
   }
 
   @Override
diff --git a/api/src/main/java/org/apache/iceberg/Scan.java b/api/src/main/java/org/apache/iceberg/Scan.java
index 4bd7717fd1d4..e3c6849501a2 100644
--- a/api/src/main/java/org/apache/iceberg/Scan.java
+++ b/api/src/main/java/org/apache/iceberg/Scan.java
@@ -20,6 +20,7 @@
 
 import java.util.Collection;
 import java.util.concurrent.ExecutorService;
+import java.util.function.Supplier;
 import org.apache.iceberg.expressions.Expression;
 import org.apache.iceberg.io.CloseableIterable;
 import org.apache.iceberg.io.FileIO;
@@ -211,7 +212,7 @@ default ThisT minRowsRequested(long numRows) {
   }
 
   /** Returns the {@link FileIO} instance to use when reading data files for this scan. */
-  default FileIO io() {
-    throw new UnsupportedOperationException("io() is not implemented: added in 1.11.0");
+  default Supplier<FileIO> fileIO() {
+    throw new UnsupportedOperationException("fileIO() is not implemented: added in 1.11.0");
   }
 }
diff --git a/core/src/main/java/org/apache/iceberg/BaseScan.java b/core/src/main/java/org/apache/iceberg/BaseScan.java
index 53f3782b38ca..242a5aaacc09 100644
--- a/core/src/main/java/org/apache/iceberg/BaseScan.java
+++ b/core/src/main/java/org/apache/iceberg/BaseScan.java
@@ -24,6 +24,7 @@
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.ExecutorService;
+import java.util.function.Supplier;
 import java.util.stream.Collectors;
 import org.apache.iceberg.expressions.Binder;
 import org.apache.iceberg.expressions.Expression;
@@ -102,11 +103,19 @@ public Table table() {
     return table;
   }
 
-  @Override
-  public FileIO io() {
+  /**
+   * @deprecated since 1.11.0, will be removed in 1.12.0; use {@link BaseScan#fileIO()} instead.
+   */
+  @Deprecated
+  protected FileIO io() {
     return table.io();
   }
 
+  @Override
+  public Supplier<FileIO> fileIO() {
+    return table::io;
+  }
+
   protected Schema tableSchema() {
     return schema;
   }
diff --git a/core/src/main/java/org/apache/iceberg/rest/RESTTableScan.java b/core/src/main/java/org/apache/iceberg/rest/RESTTableScan.java
index a78628ed87ed..64d31431b9ba 100644
--- a/core/src/main/java/org/apache/iceberg/rest/RESTTableScan.java
+++ b/core/src/main/java/org/apache/iceberg/rest/RESTTableScan.java
@@ -25,6 +25,7 @@
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Supplier;
 import java.util.stream.Collectors;
 import org.apache.iceberg.CatalogProperties;
 import org.apache.iceberg.CatalogUtil;
@@ -129,10 +130,12 @@ protected TableScan newRefinedScan(
   }
 
   @Override
-  public FileIO io() {
-    Preconditions.checkState(
-        null != scanFileIO, "FileIO is not available: planFiles() must be called first");
-    return scanFileIO;
+  public Supplier<FileIO> fileIO() {
+    return () -> {
+      Preconditions.checkState(
+          null != scanFileIO, "FileIO is not available: planFiles() must be called first");
+      return scanFileIO;
+    };
   }
 
   @Override
diff --git a/core/src/test/java/org/apache/iceberg/rest/TestRESTScanPlanning.java b/core/src/test/java/org/apache/iceberg/rest/TestRESTScanPlanning.java
index 734eaf485cd2..d42db7246226 100644
--- a/core/src/test/java/org/apache/iceberg/rest/TestRESTScanPlanning.java
+++ b/core/src/test/java/org/apache/iceberg/rest/TestRESTScanPlanning.java
@@ -996,18 +996,20 @@ public <T extends RESTResponse> T execute(
     assertThat(table.io().properties()).doesNotContainKey(RESTCatalogProperties.REST_SCAN_PLAN_ID);
 
     TableScan tableScan = table.newScan();
-    assertThatThrownBy(tableScan::io)
+    assertThatThrownBy(() -> tableScan.fileIO().get())
         .isInstanceOf(IllegalStateException.class)
         .hasMessage("FileIO is not available: planFiles() must be called first");
 
     // make sure remote scan planning is called and FileIO gets the planId
     assertThat(tableScan.planFiles()).hasSize(1);
     assertThat(table.io().properties()).doesNotContainKey(RESTCatalogProperties.REST_SCAN_PLAN_ID);
-    assertThat(tableScan.io().properties()).containsKey(RESTCatalogProperties.REST_SCAN_PLAN_ID);
-    String planId = tableScan.io().properties().get(RESTCatalogProperties.REST_SCAN_PLAN_ID);
+    assertThat(tableScan.fileIO().get().properties())
+        .containsKey(RESTCatalogProperties.REST_SCAN_PLAN_ID);
+    String planId =
+        tableScan.fileIO().get().properties().get(RESTCatalogProperties.REST_SCAN_PLAN_ID);
 
     TableScan newScan = table.newScan();
-    assertThatThrownBy(newScan::io)
+    assertThatThrownBy(() -> newScan.fileIO().get())
         .isInstanceOf(IllegalStateException.class)
         .hasMessage("FileIO is not available: planFiles() must be called first");
 
@@ -1016,8 +1018,9 @@ public <T extends RESTResponse> T execute(
     assertThat(table.io().properties()).doesNotContainKey(RESTCatalogProperties.REST_SCAN_PLAN_ID);
 
     // make sure planIds are different for each scan
-    assertThat(newScan.io().properties()).containsKey(RESTCatalogProperties.REST_SCAN_PLAN_ID);
-    assertThat(newScan.io().properties().get(RESTCatalogProperties.REST_SCAN_PLAN_ID))
+    assertThat(newScan.fileIO().get().properties())
+        .containsKey(RESTCatalogProperties.REST_SCAN_PLAN_ID);
+    assertThat(newScan.fileIO().get().properties().get(RESTCatalogProperties.REST_SCAN_PLAN_ID))
         .isNotEqualTo(planId);
   }