[Core] support write to the external path

docs/layouts/shortcodes/generated/core_configuration.html

@@ -1014,5 +1014,23 @@
             <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-paths</h5></td>
+            <td style="word-wrap: break-word;">(none)</td>
+            <td>String</td>
+            <td>The external paths where the data of this table will be written, multiple elements separated by commas.</td>
+        </tr>
+        <tr>
+            <td><h5>data-file.external-paths.strategy</h5></td>
+            <td style="word-wrap: break-word;">none</td>
+            <td><p>Enum</p></td>
+            <td>The strategy of selecting an external path when writing data.<br /><br />Possible values:<ul><li>"none": Do not choose any external storage, data will still be written to the default warehouse path.</li><li>"specific-fs": Select a specific file system as the external path. Currently supported are S3 and OSS.</li><li>"round-robin": When writing a new file, a path is chosen from data-file.external-paths in turn.</li></ul></td>
+        </tr>
+        <tr>
+            <td><h5>data-file.external-paths.specific-fs</h5></td>
+            <td style="word-wrap: break-word;">(none)</td>
+            <td>String</td>
+            <td>The specific file system of the external path when data-file.external-paths.strategy is set to specific-fs, should be the prefix scheme of the external path, now supported are s3 and oss.</td>
+        </tr>
     </tbody>
 </table>

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

@@ -124,6 +124,33 @@ public class CoreOptions implements Serializable {
                                                     + "if there is no primary key, the full row will be used.")
                                     .build());
 
+    public static final ConfigOption<String> DATA_FILE_EXTERNAL_PATHS =
+            key("data-file.external-paths")
+                    .stringType()
+                    .noDefaultValue()
+                    .withDescription(
+                            "The external paths where the data of this table will be written, "
+                                    + "multiple elements separated by commas.");
+
+    public static final ConfigOption<ExternalPathStrategy> DATA_FILE_EXTERNAL_PATHS_STRATEGY =
+            key("data-file.external-paths.strategy")
+                    .enumType(ExternalPathStrategy.class)
+                    .defaultValue(ExternalPathStrategy.NONE)
+                    .withDescription(
+                            "The strategy of selecting an external path when writing data.");
+
+    public static final ConfigOption<String> DATA_FILE_EXTERNAL_PATHS_SPECIFIC_FS =
+            key("data-file.external-paths.specific-fs")
+                    .stringType()
+                    .noDefaultValue()
+                    .withDescription(
+                            "The specific file system of the external path when "
+                                    + DATA_FILE_EXTERNAL_PATHS_STRATEGY.key()
+                                    + " is set to "
+                                    + ExternalPathStrategy.SPECIFIC_FS
+                                    + ", should be the prefix scheme of the external path, now supported are s3 and oss.");
+
+    // todo, this path is the table schema path, the name will be changed in the later PR.
     @ExcludeFromDocumentation("Internal use only")
     public static final ConfigOption<String> PATH =
             key("path")
@@ -2181,6 +2208,21 @@ public PartitionExpireStrategy partitionExpireStrategy() {
         return options.get(PARTITION_EXPIRATION_STRATEGY);
     }
 
+    @Nullable
+    public String dataFileExternalPaths() {
+        return options.get(DATA_FILE_EXTERNAL_PATHS);
+    }
+
+    @Nullable
+    public ExternalPathStrategy externalPathStrategy() {
+        return options.get(DATA_FILE_EXTERNAL_PATHS_STRATEGY);
+    }
+
+    @Nullable
+    public String externalSpecificFS() {
+        return options.get(DATA_FILE_EXTERNAL_PATHS_SPECIFIC_FS);
+    }
+
     public String partitionTimestampFormatter() {
         return options.get(PARTITION_TIMESTAMP_FORMATTER);
     }
@@ -2988,6 +3030,40 @@ public InlineElement getDescription() {
         }
     }
 
+    /** Specifies the strategy for selecting external storage paths. */
+    public enum ExternalPathStrategy implements DescribedEnum {
+        NONE(
+                "none",
+                "Do not choose any external storage, data will still be written to the default warehouse path."),
+
+        SPECIFIC_FS(
+                "specific-fs",
+                "Select a specific file system as the external path. Currently supported are S3 and OSS."),
+
+        ROUND_ROBIN(
+                "round-robin",
+                "When writing a new file, a path is chosen from data-file.external-paths in turn.");
+
+        private final String value;
+
+        private final String description;
+
+        ExternalPathStrategy(String value, String description) {
+            this.value = value;
+            this.description = description;
+        }
+
+        @Override
+        public String toString() {
+            return value;
+        }
+
+        @Override
+        public InlineElement getDescription() {
+            return text(description);
+        }
+    }
+
     /** Specifies the local file type for lookup. */
     public enum LookupLocalFileType implements DescribedEnum {
         SORT("sort", "Construct a sorted file for lookup."),

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

@@ -0,0 +1,82 @@
+/*
+ * 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 javax.annotation.Nullable;
+
+import java.io.Serializable;
+import java.util.Objects;
+import java.util.Optional;
+
+/** Provider for external data paths. */
+public class DataFileExternalPathProvider implements Serializable {
+    @Nullable private final TableExternalPathProvider tableExternalPathProvider;
+    private final Path relativeBucketPath;
+
+    public DataFileExternalPathProvider(
+            @Nullable TableExternalPathProvider tableExternalPathProvider,
+            Path relativeBucketPath) {
+        this.tableExternalPathProvider = tableExternalPathProvider;
+        this.relativeBucketPath = relativeBucketPath;
+    }
+
+    /**
+     * Get the next external data path.
+     *
+     * @return the next external data path
+     */
+    public Optional<Path> getNextExternalDataPath() {
+        return Optional.ofNullable(tableExternalPathProvider)
+                .flatMap(TableExternalPathProvider::getNextExternalPath)
+                .map(path -> new Path(path, relativeBucketPath));
+    }
+
+    public boolean externalPathExists() {
+        return tableExternalPathProvider != null && tableExternalPathProvider.externalPathExists();
+    }
+
+    @Override
+    public final boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (!(o instanceof DataFileExternalPathProvider)) {
+            return false;
+        }
+
+        DataFileExternalPathProvider that = (DataFileExternalPathProvider) o;
+        return Objects.equals(tableExternalPathProvider, that.tableExternalPathProvider)
+                && Objects.equals(relativeBucketPath, that.relativeBucketPath);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(tableExternalPathProvider, relativeBucketPath);
+    }
+
+    @Override
+    public String toString() {
+        return "DataFileExternalPathProvider{"
+                + " externalPathProvider="
+                + tableExternalPathProvider
+                + ", relativeBucketPath="
+                + relativeBucketPath
+                + "}";
+    }
+}

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

@@ -0,0 +1,192 @@
+/*
+ * 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.ExternalPathStrategy;
+import org.apache.paimon.annotation.VisibleForTesting;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Random;
+
+/** Provider for external paths. */
+public class TableExternalPathProvider implements Serializable {
+    private final Map<String, Path> externalPathsMap;
+    private final List<Path> externalPathsList;
+
+    private final ExternalPathStrategy externalPathStrategy;
+    private final String externalSpecificFS;
+    private int currentIndex = 0;
+    private boolean externalPathExists;
+
+    public TableExternalPathProvider(
+            String externalPaths,
+            ExternalPathStrategy externalPathStrategy,
+            String externalSpecificFS) {
+        this.externalPathsMap = new HashMap<>();
+        this.externalPathsList = new ArrayList<>();
+        this.externalPathStrategy = externalPathStrategy;
+        if (externalSpecificFS != null) {
+            this.externalSpecificFS = externalSpecificFS.toLowerCase();
+        } else {
+            this.externalSpecificFS = null;
+        }
+        initExternalPaths(externalPaths);
+        if (!externalPathsList.isEmpty()) {
+            this.currentIndex = new Random().nextInt(externalPathsList.size());
+        }
+    }
+
+    private void initExternalPaths(String externalPaths) {
+        if (externalPaths == null) {
+            return;
+        }
+
+        String[] tmpArray = externalPaths.split(",");
+        for (String s : tmpArray) {
+            Path path = new Path(s.trim());
+            String scheme = path.toUri().getScheme();
+            if (scheme == null) {
+                throw new IllegalArgumentException("scheme should not be null: " + path);
+            }
+            scheme = scheme.toLowerCase();
+            externalPathsMap.put(scheme, path);
+            externalPathsList.add(path);
+        }
+
+        if (externalPathStrategy != null
+                && externalPathStrategy.equals(ExternalPathStrategy.SPECIFIC_FS)) {
+            if (externalSpecificFS == null) {
+                throw new IllegalArgumentException("external specific fs should not be null: ");
+            }
+
+            if (!externalPathsMap.containsKey(externalSpecificFS)) {
+                throw new IllegalArgumentException(
+                        "external specific fs not found: " + externalSpecificFS);
+            }
+        }
+
+        if (!externalPathsMap.isEmpty()
+                && !externalPathsList.isEmpty()
+                && externalPathStrategy != ExternalPathStrategy.NONE) {
+            externalPathExists = true;
+        }
+    }
+
+    /**
+     * Get the next external path.
+     *
+     * @return the next external path
+     */
+    public Optional<Path> getNextExternalPath() {
+        if (externalPathsMap == null || externalPathsMap.isEmpty()) {
+            return Optional.empty();
+        }
+
+        switch (externalPathStrategy) {
+            case NONE:
+                return Optional.empty();
+            case SPECIFIC_FS:
+                return getSpecificFSExternalPath();
+            case ROUND_ROBIN:
+                return getRoundRobinPath();
+            default:
+                return Optional.empty();
+        }
+    }
+
+    private Optional<Path> getSpecificFSExternalPath() {
+        if (!externalPathsMap.containsKey(externalSpecificFS)) {
+            return Optional.empty();
+        }
+        return Optional.of(externalPathsMap.get(externalSpecificFS));
+    }
+
+    private Optional<Path> getRoundRobinPath() {
+        currentIndex = (currentIndex + 1) % externalPathsList.size();
+        return Optional.of(externalPathsList.get(currentIndex));
+    }
+
+    public boolean externalPathExists() {
+        return externalPathExists;
+    }
+
+    @VisibleForTesting
+    public Map<String, Path> getExternalPathsMap() {
+        return externalPathsMap;
+    }
+
+    @VisibleForTesting
+    public List<Path> getExternalPathsList() {
+        return externalPathsList;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+
+        TableExternalPathProvider that = (TableExternalPathProvider) o;
+        return currentIndex == that.currentIndex
+                && externalPathExists == that.externalPathExists
+                && externalPathsMap.equals(that.externalPathsMap)
+                && externalPathsList.equals(that.externalPathsList)
+                && externalPathStrategy == that.externalPathStrategy
+                && Objects.equals(externalSpecificFS, that.externalSpecificFS);
+    }
+
+    @Override
+    public String toString() {
+        return "ExternalPathProvider{"
+                + " externalPathsMap="
+                + externalPathsMap
+                + ", externalPathsList="
+                + externalPathsList
+                + ", externalPathStrategy="
+                + externalPathStrategy
+                + ", externalSpecificFS='"
+                + externalSpecificFS
+                + '\''
+                + ", currentIndex="
+                + currentIndex
+                + ", externalPathExists="
+                + externalPathExists
+                + "}";
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(
+                externalPathsMap,
+                externalPathsList,
+                externalPathStrategy,
+                externalSpecificFS,
+                currentIndex,
+                externalPathExists);
+    }
+}