[core] Introduce Table.fileIO interface to public fileIO to file operation (#5102)

Author: JingsongLi

paimon-core/src/main/java/org/apache/paimon/catalog/CatalogUtils.java

@@ -185,7 +185,7 @@ public static Table loadTable(
         TableSchema schema = metadata.schema();
         CoreOptions options = CoreOptions.fromMap(schema.options());
         if (options.type() == TableType.FORMAT_TABLE) {
-            return toFormatTable(identifier, schema);
+            return toFormatTable(identifier, schema, dataFileIO);
         }
 
         CatalogEnvironment catalogEnv =
@@ -249,7 +249,8 @@ private static Table createSystemTable(Identifier identifier, Table originTable)
         return table;
     }
 
-    private static FormatTable toFormatTable(Identifier identifier, TableSchema schema) {
+    private static FormatTable toFormatTable(
+            Identifier identifier, TableSchema schema, Function<Path, FileIO> fileIO) {
         Map<String, String> options = schema.options();
         FormatTable.Format format =
                 FormatTable.parseFormat(
@@ -258,6 +259,7 @@ private static FormatTable toFormatTable(Identifier identifier, TableSchema sche
                                 CoreOptions.FILE_FORMAT.defaultValue()));
         String location = options.get(CoreOptions.PATH.key());
         return FormatTable.builder()
+                .fileIO(fileIO.apply(new Path(location)))
                 .identifier(identifier)
                 .rowType(schema.logicalRowType())
                 .partitionKeys(schema.partitionKeys())

paimon-core/src/main/java/org/apache/paimon/table/DataTable.java

@@ -19,7 +19,6 @@
 package org.apache.paimon.table;
 
 import org.apache.paimon.CoreOptions;
-import org.apache.paimon.fs.FileIO;
 import org.apache.paimon.fs.Path;
 import org.apache.paimon.schema.SchemaManager;
 import org.apache.paimon.table.source.DataTableScan;
@@ -53,6 +52,4 @@ public interface DataTable extends InnerTable {
     DataTable switchToBranch(String branchName);
 
     Path location();
-
-    FileIO fileIO();
 }

paimon-core/src/main/java/org/apache/paimon/table/FormatTable.java

@@ -21,6 +21,7 @@
 import org.apache.paimon.Snapshot;
 import org.apache.paimon.annotation.Public;
 import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.fs.FileIO;
 import org.apache.paimon.manifest.IndexManifestEntry;
 import org.apache.paimon.manifest.ManifestEntry;
 import org.apache.paimon.manifest.ManifestFileMeta;
@@ -92,6 +93,7 @@ static FormatTable.Builder builder() {
     /** Builder for {@link FormatTable}. */
     class Builder {
 
+        private FileIO fileIO;
         private Identifier identifier;
         private RowType rowType;
         private List<String> partitionKeys;
@@ -100,6 +102,11 @@ class Builder {
         private Map<String, String> options;
         @Nullable private String comment;
 
+        public FormatTable.Builder fileIO(FileIO fileIO) {
+            this.fileIO = fileIO;
+            return this;
+        }
+
         public FormatTable.Builder identifier(Identifier identifier) {
             this.identifier = identifier;
             return this;
@@ -137,13 +144,14 @@ public FormatTable.Builder comment(@Nullable String comment) {
 
         public FormatTable build() {
             return new FormatTable.FormatTableImpl(
-                    identifier, rowType, partitionKeys, location, format, options, comment);
+                    fileIO, identifier, rowType, partitionKeys, location, format, options, comment);
         }
     }
 
     /** An implementation for {@link FormatTable}. */
     class FormatTableImpl implements FormatTable {
 
+        private final FileIO fileIO;
         private final Identifier identifier;
         private final RowType rowType;
         private final List<String> partitionKeys;
@@ -153,13 +161,15 @@ class FormatTableImpl implements FormatTable {
         @Nullable private final String comment;
 
         public FormatTableImpl(
+                FileIO fileIO,
                 Identifier identifier,
                 RowType rowType,
                 List<String> partitionKeys,
                 String location,
                 Format format,
                 Map<String, String> options,
                 @Nullable String comment) {
+            this.fileIO = fileIO;
             this.identifier = identifier;
             this.rowType = rowType;
             this.partitionKeys = partitionKeys;
@@ -214,12 +224,24 @@ public Optional<String> comment() {
             return Optional.ofNullable(comment);
         }
 
+        @Override
+        public FileIO fileIO() {
+            return fileIO;
+        }
+
         @Override
         public FormatTable copy(Map<String, String> dynamicOptions) {
             Map<String, String> newOptions = new HashMap<>(options);
             newOptions.putAll(dynamicOptions);
             return new FormatTableImpl(
-                    identifier, rowType, partitionKeys, location, format, newOptions, comment);
+                    fileIO,
+                    identifier,
+                    rowType,
+                    partitionKeys,
+                    location,
+                    format,
+                    newOptions,
+                    comment);
         }
     }
 

paimon-core/src/main/java/org/apache/paimon/table/KnownSplitsTable.java

@@ -18,6 +18,7 @@
 
 package org.apache.paimon.table;
 
+import org.apache.paimon.fs.FileIO;
 import org.apache.paimon.table.source.DataSplit;
 import org.apache.paimon.table.source.InnerTableRead;
 import org.apache.paimon.table.source.InnerTableScan;
@@ -30,6 +31,7 @@
  * A table to hold some known data splits. For now, it is only used by internal for Spark engine.
  */
 public class KnownSplitsTable implements ReadonlyTable {
+
     private final InnerTable origin;
     private final DataSplit[] splits;
 
@@ -71,6 +73,11 @@ public Map<String, String> options() {
         return origin.options();
     }
 
+    @Override
+    public FileIO fileIO() {
+        return origin.fileIO();
+    }
+
     @Override
     public InnerTableRead newRead() {
         return origin.newRead();

paimon-core/src/main/java/org/apache/paimon/table/Table.java

@@ -21,6 +21,7 @@
 import org.apache.paimon.Snapshot;
 import org.apache.paimon.annotation.Experimental;
 import org.apache.paimon.annotation.Public;
+import org.apache.paimon.fs.FileIO;
 import org.apache.paimon.manifest.IndexManifestEntry;
 import org.apache.paimon.manifest.ManifestEntry;
 import org.apache.paimon.manifest.ManifestFileMeta;
@@ -86,6 +87,9 @@ default String uuid() {
 
     // ================= Table Operations ====================
 
+    /** File io of this table. */
+    FileIO fileIO();
+
     /** Copy this table with adding dynamic options. */
     Table copy(Map<String, String> dynamicOptions);
 

paimon-core/src/main/java/org/apache/paimon/table/system/AggregationFieldsTable.java

@@ -100,6 +100,11 @@ public List<String> primaryKeys() {
         return Collections.singletonList("field_name");
     }
 
+    @Override
+    public FileIO fileIO() {
+        return fileIO;
+    }
+
     @Override
     public InnerTableScan newScan() {
         return new SchemasScan();

paimon-core/src/main/java/org/apache/paimon/table/system/AllTableOptionsTable.java

@@ -23,6 +23,8 @@
 import org.apache.paimon.data.GenericRow;
 import org.apache.paimon.data.InternalRow;
 import org.apache.paimon.disk.IOManager;
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.local.LocalFileIO;
 import org.apache.paimon.predicate.Predicate;
 import org.apache.paimon.reader.RecordReader;
 import org.apache.paimon.table.ReadonlyTable;
@@ -91,6 +93,12 @@ public List<String> primaryKeys() {
         return Collections.singletonList("table_name");
     }
 
+    @Override
+    public FileIO fileIO() {
+        // pass a useless file io, should never use this.
+        return new LocalFileIO();
+    }
+
     @Override
     public InnerTableScan newScan() {
         return new AllTableOptionsScan();

paimon-core/src/main/java/org/apache/paimon/table/system/BranchesTable.java

@@ -104,6 +104,11 @@ public List<String> primaryKeys() {
         return Collections.singletonList("branch_name");
     }
 
+    @Override
+    public FileIO fileIO() {
+        return dataTable.fileIO();
+    }
+
     @Override
     public InnerTableScan newScan() {
         return new BranchesScan();

paimon-core/src/main/java/org/apache/paimon/table/system/BucketsTable.java

@@ -25,6 +25,7 @@
 import org.apache.paimon.data.InternalRow;
 import org.apache.paimon.data.Timestamp;
 import org.apache.paimon.disk.IOManager;
+import org.apache.paimon.fs.FileIO;
 import org.apache.paimon.manifest.BucketEntry;
 import org.apache.paimon.predicate.Predicate;
 import org.apache.paimon.reader.RecordReader;
@@ -97,6 +98,11 @@ public List<String> primaryKeys() {
         return Arrays.asList("partition", "bucket");
     }
 
+    @Override
+    public FileIO fileIO() {
+        return storeTable.fileIO();
+    }
+
     @Override
     public InnerTableScan newScan() {
         return new BucketsScan();

paimon-core/src/main/java/org/apache/paimon/table/system/CatalogOptionsTable.java

@@ -22,6 +22,8 @@
 import org.apache.paimon.data.GenericRow;
 import org.apache.paimon.data.InternalRow;
 import org.apache.paimon.disk.IOManager;
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.local.LocalFileIO;
 import org.apache.paimon.options.Options;
 import org.apache.paimon.predicate.Predicate;
 import org.apache.paimon.reader.RecordReader;
@@ -93,6 +95,12 @@ public List<String> primaryKeys() {
         return Collections.singletonList("key");
     }
 
+    @Override
+    public FileIO fileIO() {
+        // pass a useless file io, should never use this.
+        return new LocalFileIO();
+    }
+
     @Override
     public Table copy(Map<String, String> dynamicOptions) {
         return new CatalogOptionsTable(catalogOptions);