diff --git a/core/src/main/java/org/apache/iceberg/hadoop/HadoopTables.java b/core/src/main/java/org/apache/iceberg/hadoop/HadoopTables.java
index 740158741b2a..2f3f621e8103 100644
--- a/core/src/main/java/org/apache/iceberg/hadoop/HadoopTables.java
+++ b/core/src/main/java/org/apache/iceberg/hadoop/HadoopTables.java
@@ -44,6 +44,7 @@
 import org.apache.iceberg.catalog.Catalog;
 import org.apache.iceberg.exceptions.AlreadyExistsException;
 import org.apache.iceberg.exceptions.NoSuchTableException;
+import org.apache.iceberg.io.FileIO;
 import org.apache.iceberg.relocated.com.google.common.annotations.VisibleForTesting;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
@@ -68,12 +69,23 @@ public class HadoopTables implements Tables, Configurable {
 
   private Configuration conf;
 
+  private FileIO fileIO;
+
   public HadoopTables() {
     this(new Configuration());
   }
 
+  public HadoopTables(FileIO fileIO) {
+    this(new Configuration(), fileIO);
+  }
+
   public HadoopTables(Configuration conf) {
+    this(conf, null);
+  }
+
+  public HadoopTables(Configuration conf, FileIO fileIO) {
     this.conf = conf;
+    this.fileIO = fileIO;
   }
 
   /**
@@ -202,9 +214,9 @@ public boolean dropTable(String location, boolean purge) {
   @VisibleForTesting
   TableOperations newTableOps(String location) {
     if (location.contains(METADATA_JSON)) {
-      return new StaticTableOperations(location, new HadoopFileIO(conf));
+      return new StaticTableOperations(location, fileIO == null? new HadoopFileIO(conf): fileIO);
     } else {
-      return new HadoopTableOperations(new Path(location), new HadoopFileIO(conf), conf,
+      return new HadoopTableOperations(new Path(location), fileIO == null? new HadoopFileIO(conf): fileIO, conf,
           createOrGetLockManager(this));
     }
   }
diff --git a/core/src/test/java/org/apache/iceberg/hadoop/TestHadoopTables.java b/core/src/test/java/org/apache/iceberg/hadoop/TestHadoopTables.java
index 56145bb0f798..790edc9f7d56 100644
--- a/core/src/test/java/org/apache/iceberg/hadoop/TestHadoopTables.java
+++ b/core/src/test/java/org/apache/iceberg/hadoop/TestHadoopTables.java
@@ -24,6 +24,8 @@
 import java.nio.charset.StandardCharsets;
 import java.nio.file.Files;
 import java.nio.file.Paths;
+
+import org.apache.hadoop.conf.Configuration;
 import org.apache.iceberg.AppendFiles;
 import org.apache.iceberg.AssertHelpers;
 import org.apache.iceberg.DataFile;
@@ -33,6 +35,9 @@
 import org.apache.iceberg.SortOrder;
 import org.apache.iceberg.Table;
 import org.apache.iceberg.exceptions.NoSuchTableException;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.io.InputFile;
+import org.apache.iceberg.io.OutputFile;
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
 import org.apache.iceberg.relocated.com.google.common.collect.Maps;
 import org.apache.iceberg.transforms.Transform;
@@ -164,6 +169,20 @@ public void testTableName() {
     Assert.assertEquals("Name must match", location + "#snapshots", snapshotsTable.name());
   }
 
+  @Test
+  public void testDynamicFileIO(){
+    FileIO testFileIO = new HadoopFileIO(new Configuration());
+    PartitionSpec spec = PartitionSpec.builderFor(SCHEMA)
+            .bucket("data", 16)
+            .build();
+    String location = tableDir.toURI().toString();
+    HadoopTables hadoopTables = new HadoopTables(testFileIO);
+    hadoopTables.create(SCHEMA, spec, location);
+    Table table = hadoopTables.load(location);
+    Assert.assertEquals("Name must match", testFileIO, table.io());
+    Assert.assertTrue("FileIO must match", testFileIO == table.io());
+  }
+
   private static void createDummyTable(File tableDir, File dataDir) throws IOException {
     Table table = TABLES.create(SCHEMA, tableDir.toURI().toString());
     AppendFiles append = table.newAppend();