[PIP-xxx] Introduce Table Multi-Location Management. Step 1

docs/layouts/shortcodes/generated/core_configuration.html

@@ -1008,5 +1008,11 @@
             <td>Integer</td>
             <td>The bytes of types (CHAR, VARCHAR, BINARY, VARBINARY) devote to the zorder sort.</td>
         </tr>
+        <tr>
+            <td><h5>data-file.external-path</h5></td>
+            <td style="word-wrap: break-word;">(none)</td>
+            <td>String</td>
+            <td>The location where the data of this table is currently written.</td>
+        </tr>
     </tbody>
 </table>

paimon-common/src/main/java/org/apache/paimon/CoreOptions.java

@@ -131,6 +131,13 @@ public class CoreOptions implements Serializable {
                     .noDefaultValue()
                     .withDescription("The file path of this table in the filesystem.");
 
+    @ExcludeFromDocumentation("Internal use only")
+    public static final ConfigOption<String> WAREHOUSE_ROOT_PATH =
+            key("warehouse.root-path")
+                    .stringType()
+                    .noDefaultValue()
+                    .withDescription("The file path of the warehouse in the filesystem.");
+
     public static final ConfigOption<String> BRANCH =
             key("branch").stringType().defaultValue("main").withDescription("Specify branch name.");
 
@@ -1521,6 +1528,13 @@ public class CoreOptions implements Serializable {
                     .noDefaultValue()
                     .withDescription("The serialized refresh handler of materialized table.");
 
+    public static final ConfigOption<String> DATA_FILE_EXTERNAL_PATH =
+            key("data-file.external-path")
+                    .stringType()
+                    .noDefaultValue()
+                    .withDescription(
+                            "The location where the data of this table is currently written.");
+
     private final Options options;
 
     public CoreOptions(Map<String, String> options) {
@@ -2359,6 +2373,21 @@ public boolean asyncFileWrite() {
         return options.get(ASYNC_FILE_WRITE);
     }
 
+    public String getDataFileExternalPath() {
+        return options.get(DATA_FILE_EXTERNAL_PATH);
+    }
+
+    public String getWarehouseRootPath() {
+        return options.get(WAREHOUSE_ROOT_PATH);
+    }
+
+    public String getDataRootLocation() {
+        if (getDataFileExternalPath() == null || getDataFileExternalPath().isEmpty()) {
+            return getWarehouseRootPath();
+        }
+        return getDataFileExternalPath();
+    }
+
     public boolean statsDenseStore() {
         return options.get(METADATA_STATS_DENSE_STORE);
     }

paimon-common/src/main/java/org/apache/paimon/fs/HybridFileIO.java

@@ -0,0 +1,140 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.fs;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.catalog.CatalogContext;
+import org.apache.paimon.options.Options;
+
+import java.io.IOException;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * A hybrid implementation of {@link FileIO} that supports multiple file system schemas. It
+ * dynamically selects the appropriate {@link FileIO} based on the URI scheme of the given path.
+ */
+public class HybridFileIO implements FileIO {
+
+    private static final long serialVersionUID = 1L;
+
+    protected Options options;
+
+    private Map<String, FileIO> fileIOMap;
+    private volatile FileIO fallbackFileIO;
+
+    @Override
+    public boolean isObjectStore() {
+        if (options.get(CoreOptions.DATA_FILE_EXTERNAL_PATH) != null
+                && ((options.get(CoreOptions.DATA_FILE_EXTERNAL_PATH).startsWith("oss://")
+                        || (options.get(CoreOptions.DATA_FILE_EXTERNAL_PATH)
+                                .startsWith("s3://"))))) {
+            return true;
+        }
+        return false;
+    }
+
+    @Override
+    public void configure(CatalogContext context) {
+        this.options = context.options();
+        this.fileIOMap = new ConcurrentHashMap<>();
+    }
+
+    @Override
+    public SeekableInputStream newInputStream(Path path) throws IOException {
+        return wrap(() -> fileIO(path).newInputStream(path));
+    }
+
+    @Override
+    public PositionOutputStream newOutputStream(Path path, boolean overwrite) throws IOException {
+        return wrap(() -> fileIO(path).newOutputStream(path, overwrite));
+    }
+
+    @Override
+    public FileStatus getFileStatus(Path path) throws IOException {
+        return wrap(() -> fileIO(path).getFileStatus(path));
+    }
+
+    @Override
+    public FileStatus[] listStatus(Path path) throws IOException {
+        return wrap(() -> fileIO(path).listStatus(path));
+    }
+
+    @Override
+    public boolean exists(Path path) throws IOException {
+        return wrap(() -> fileIO(path).exists(path));
+    }
+
+    @Override
+    public boolean delete(Path path, boolean recursive) throws IOException {
+        return wrap(() -> fileIO(path).delete(path, recursive));
+    }
+
+    @Override
+    public boolean mkdirs(Path path) throws IOException {
+        return wrap(() -> fileIO(path).mkdirs(path));
+    }
+
+    @Override
+    public boolean rename(Path src, Path dst) throws IOException {
+        return wrap(() -> fileIO(src).rename(src, dst));
+    }
+
+    private FileIO fileIO(Path path) throws IOException {
+        String schema = path.toUri().getScheme();
+        if (schema == null) {
+            if (fallbackFileIO == null) {
+                synchronized (this) {
+                    if (fallbackFileIO == null) {
+                        CatalogContext catalogContext = CatalogContext.create(options);
+                        fallbackFileIO = FileIO.get(path, catalogContext);
+                    }
+                }
+            }
+            return fallbackFileIO;
+        }
+
+        if (!fileIOMap.containsKey(schema)) {
+            synchronized (this) {
+                if (!fileIOMap.containsKey(schema)) {
+                    CatalogContext catalogContext = CatalogContext.create(options);
+                    FileIO fileIO = FileIO.get(path, catalogContext);
+                    fileIOMap.put(path.toUri().getScheme(), fileIO);
+                }
+            }
+        }
+        return fileIOMap.get(path.toUri().getScheme());
+    }
+
+    private <T> T wrap(Func<T> func) throws IOException {
+        ClassLoader cl = Thread.currentThread().getContextClassLoader();
+        try {
+            Thread.currentThread().setContextClassLoader(HybridFileIO.class.getClassLoader());
+            return func.apply();
+        } finally {
+            Thread.currentThread().setContextClassLoader(cl);
+        }
+    }
+
+    /** Apply function with wrapping classloader. */
+    @FunctionalInterface
+    protected interface Func<T> {
+        T apply() throws IOException;
+    }
+}

paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java

@@ -24,6 +24,7 @@
 import org.apache.paimon.fs.Path;
 import org.apache.paimon.index.HashIndexFile;
 import org.apache.paimon.index.IndexFileHandler;
+import org.apache.paimon.io.TablePathProvider;
 import org.apache.paimon.manifest.IndexManifestFile;
 import org.apache.paimon.manifest.ManifestFile;
 import org.apache.paimon.manifest.ManifestList;
@@ -82,6 +83,7 @@ abstract class AbstractFileStore<T> implements FileStore<T> {
     @Nullable private final SegmentsCache<Path> writeManifestCache;
     @Nullable private SegmentsCache<Path> readManifestCache;
     @Nullable private Cache<Path, Snapshot> snapshotCache;
+    private final TablePathProvider tablePathProvider;
 
     protected AbstractFileStore(
             FileIO fileIO,
@@ -90,7 +92,8 @@ protected AbstractFileStore(
             String tableName,
             CoreOptions options,
             RowType partitionType,
-            CatalogEnvironment catalogEnvironment) {
+            CatalogEnvironment catalogEnvironment,
+            TablePathProvider tablePathProvider) {
         this.fileIO = fileIO;
         this.schemaManager = schemaManager;
         this.schema = schema;
@@ -101,6 +104,7 @@ protected AbstractFileStore(
         this.writeManifestCache =
                 SegmentsCache.create(
                         options.pageSize(), options.writeManifestCache(), Long.MAX_VALUE);
+        this.tablePathProvider = tablePathProvider;
     }
 
     @Override
@@ -110,7 +114,7 @@ public FileStorePathFactory pathFactory() {
 
     protected FileStorePathFactory pathFactory(String format) {
         return new FileStorePathFactory(
-                options.path(),
+                tablePathProvider,
                 partitionType,
                 options.partitionDefaultName(),
                 format,

paimon-core/src/main/java/org/apache/paimon/AppendOnlyFileStore.java

@@ -22,6 +22,7 @@
 import org.apache.paimon.deletionvectors.DeletionVectorsMaintainer;
 import org.apache.paimon.format.FileFormatDiscover;
 import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.io.TablePathProvider;
 import org.apache.paimon.manifest.ManifestCacheFilter;
 import org.apache.paimon.operation.AppendOnlyFileStoreScan;
 import org.apache.paimon.operation.AppendOnlyFileStoreWrite;
@@ -59,8 +60,17 @@ public AppendOnlyFileStore(
             RowType bucketKeyType,
             RowType rowType,
             String tableName,
-            CatalogEnvironment catalogEnvironment) {
-        super(fileIO, schemaManager, schema, tableName, options, partitionType, catalogEnvironment);
+            CatalogEnvironment catalogEnvironment,
+            TablePathProvider tablePathProvider) {
+        super(
+                fileIO,
+                schemaManager,
+                schema,
+                tableName,
+                options,
+                partitionType,
+                catalogEnvironment,
+                tablePathProvider);
         this.bucketKeyType = bucketKeyType;
         this.rowType = rowType;
     }

paimon-core/src/main/java/org/apache/paimon/KeyValueFileStore.java

@@ -26,6 +26,7 @@
 import org.apache.paimon.index.HashIndexMaintainer;
 import org.apache.paimon.index.IndexMaintainer;
 import org.apache.paimon.io.KeyValueFileReaderFactory;
+import org.apache.paimon.io.TablePathProvider;
 import org.apache.paimon.manifest.ManifestCacheFilter;
 import org.apache.paimon.mergetree.compact.MergeFunctionFactory;
 import org.apache.paimon.operation.BucketSelectConverter;
@@ -84,8 +85,17 @@ public KeyValueFileStore(
             KeyValueFieldsExtractor keyValueFieldsExtractor,
             MergeFunctionFactory<KeyValue> mfFactory,
             String tableName,
-            CatalogEnvironment catalogEnvironment) {
-        super(fileIO, schemaManager, schema, tableName, options, partitionType, catalogEnvironment);
+            CatalogEnvironment catalogEnvironment,
+            TablePathProvider tablePathProvider) {
+        super(
+                fileIO,
+                schemaManager,
+                schema,
+                tableName,
+                options,
+                partitionType,
+                catalogEnvironment,
+                tablePathProvider);
         this.crossPartitionUpdate = crossPartitionUpdate;
         this.bucketKeyType = bucketKeyType;
         this.keyType = keyType;

paimon-core/src/main/java/org/apache/paimon/append/AppendOnlyWriter.java

@@ -244,7 +244,7 @@ public void close() throws Exception {
         for (DataFileMeta file : compactAfter) {
             // appendOnlyCompactManager will rewrite the file and no file upgrade will occur, so we
             // can directly delete the file in compactAfter.
-            fileIO.deleteQuietly(pathFactory.toPath(file.fileName()));
+            fileIO.deleteQuietly(pathFactory.toPath(file.getDataRootLocation(), file.fileName()));
         }
 
         sinkWriter.close();
@@ -271,7 +271,8 @@ public void toBufferedWriter() throws Exception {
             } finally {
                 // remove small files
                 for (DataFileMeta file : files) {
-                    fileIO.deleteQuietly(pathFactory.toPath(file.fileName()));
+                    fileIO.deleteQuietly(
+                            pathFactory.toPath(file.getDataRootLocation(), file.fileName()));
                 }
             }
         }

paimon-core/src/main/java/org/apache/paimon/deletionvectors/DeletionVector.java

@@ -162,5 +162,9 @@ static Factory factory(
     /** Interface to create {@link DeletionVector}. */
     interface Factory {
         Optional<DeletionVector> create(String fileName) throws IOException;
+
+        // @todo qihouliang
+        // Optional<DeletionVector> create(Path dataRootLocation, String fileName) throws
+        // IOException;
     }
 }