Unified variable names

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

@@ -2381,7 +2381,7 @@ public String getWarehouseRootPath() {
         return options.get(WAREHOUSE_ROOT_PATH);
     }
 
-    public String getDataWriteRootPath() {
+    public String getDataRootLocation() {
         if (getDataFileExternalPath() == null || getDataFileExternalPath().isEmpty()) {
             return getWarehouseRootPath();
         }

paimon-core/src/main/java/org/apache/paimon/io/DataFileMeta.java

@@ -82,7 +82,8 @@ public class DataFileMeta {
                             new DataField(
                                     16,
                                     "_VALUE_STATS_COLS",
-                                    DataTypes.ARRAY(DataTypes.STRING().notNull()))));
+                                    DataTypes.ARRAY(DataTypes.STRING().notNull())),
+                            new DataField(17, "_DATA_ROOT_LOCATION", newStringType(true))));
 
     public static final BinaryRow EMPTY_MIN_KEY = EMPTY_ROW;
     public static final BinaryRow EMPTY_MAX_KEY = EMPTY_ROW;
@@ -120,9 +121,12 @@ public class DataFileMeta {
 
     private final @Nullable List<String> valueStatsCols;
 
-    // the external path that the file resides in, if it is null,
-    // the file is in the default warehouse path
-    private final @Nullable Path dataRootLocation;
+    /**
+     * the data root location that the file resides in, if it is null, the file is in the default
+     * warehouse path, when {@link CoreOptions#DATA_FILE_PATH_DIRECTORY} is set, new writen files
+     * will be persisted in {@link CoreOptions#DATA_FILE_PATH_DIRECTORY}.
+     */
+    private final @Nullable String dataRootLocation;
 
     public static DataFileMeta forAppend(
             String fileName,
@@ -136,7 +140,7 @@ public static DataFileMeta forAppend(
             @Nullable byte[] embeddedIndex,
             @Nullable FileSource fileSource,
             @Nullable List<String> valueStatsCols,
-            @Nullable Path dataRootLocation) {
+            @Nullable String dataRootLocation) {
         return new DataFileMeta(
                 fileName,
                 fileSize,
@@ -213,7 +217,7 @@ public DataFileMeta(
             @Nullable byte[] embeddedIndex,
             @Nullable FileSource fileSource,
             @Nullable List<String> valueStatsCols,
-            Path dataRootLocation) {
+            String dataRootLocation) {
         this(
                 fileName,
                 fileSize,
@@ -329,7 +333,7 @@ public DataFileMeta(
             @Nullable byte[] embeddedIndex,
             @Nullable FileSource fileSource,
             @Nullable List<String> valueStatsCols,
-            @Nullable Path dataRootLocation) {
+            @Nullable String dataRootLocation) {
         this.fileName = fileName;
         this.fileSize = fileSize;
 
@@ -367,10 +371,18 @@ public long rowCount() {
     }
 
     @Nullable
-    public Path getDataRootLocation() {
+    public String getDataRootLocationString() {
         return dataRootLocation;
     }
 
+    @Nullable
+    public Path getDataRootLocation() {
+        if (dataRootLocation == null) {
+            return null;
+        }
+        return new Path(dataRootLocation);
+    }
+
     public Optional<Long> addRowCount() {
         return Optional.ofNullable(deleteRowCount).map(c -> rowCount - c);
     }
@@ -528,8 +540,8 @@ public DataFileMeta copyWithoutStats() {
 
     public List<Path> collectFiles(DataFilePathFactory pathFactory) {
         List<Path> paths = new ArrayList<>();
-        paths.add(pathFactory.toPath(dataRootLocation, fileName));
-        extraFiles.forEach(f -> paths.add(pathFactory.toPath(dataRootLocation, f)));
+        paths.add(pathFactory.toPath(getDataRootLocation(), fileName));
+        extraFiles.forEach(f -> paths.add(pathFactory.toPath(getDataRootLocation(), f)));
         return paths;
     }
 
@@ -602,7 +614,8 @@ public boolean equals(Object o) {
                 && Objects.equals(creationTime, that.creationTime)
                 && Objects.equals(deleteRowCount, that.deleteRowCount)
                 && Objects.equals(fileSource, that.fileSource)
-                && Objects.equals(valueStatsCols, that.valueStatsCols);
+                && Objects.equals(valueStatsCols, that.valueStatsCols)
+                && Objects.equals(dataRootLocation, that.dataRootLocation);
     }
 
     @Override
@@ -624,7 +637,8 @@ public int hashCode() {
                 creationTime,
                 deleteRowCount,
                 fileSource,
-                valueStatsCols);
+                valueStatsCols,
+                dataRootLocation);
     }
 
     @Override
@@ -634,7 +648,7 @@ public String toString() {
                         + "minKey: %s, maxKey: %s, keyStats: %s, valueStats: %s, "
                         + "minSequenceNumber: %d, maxSequenceNumber: %d, "
                         + "schemaId: %d, level: %d, extraFiles: %s, creationTime: %s, "
-                        + "deleteRowCount: %d, fileSource: %s, valueStatsCols: %s}",
+                        + "deleteRowCount: %d, fileSource: %s, valueStatsCols: %s, dataRootLocation: %s}",
                 fileName,
                 fileSize,
                 rowCount,
@@ -651,7 +665,8 @@ public String toString() {
                 creationTime,
                 deleteRowCount,
                 fileSource,
-                valueStatsCols);
+                valueStatsCols,
+                dataRootLocation);
     }
 
     public static long getMaxSequenceNumber(List<DataFileMeta> fileMetas) {

paimon-core/src/main/java/org/apache/paimon/io/DataFileMetaSerializer.java

@@ -58,7 +58,8 @@ public InternalRow toRow(DataFileMeta meta) {
                 meta.deleteRowCount().orElse(null),
                 meta.embeddedIndex(),
                 meta.fileSource().map(FileSource::toByteValue).orElse(null),
-                toStringArrayData(meta.valueStatsCols()));
+                toStringArrayData(meta.valueStatsCols()),
+                meta.getDataRootLocationString());
     }
 
     @Override
@@ -81,6 +82,6 @@ public DataFileMeta fromRow(InternalRow row) {
                 row.isNullAt(14) ? null : row.getBinary(14),
                 row.isNullAt(15) ? null : FileSource.fromByteValue(row.getByte(15)),
                 row.isNullAt(16) ? null : fromStringArrayData(row.getArray(16)),
-                null);
+                row.isNullAt(17) ? null : row.getString(17).toString());
     }
 }

paimon-core/src/main/java/org/apache/paimon/io/DataFilePathFactory.java

@@ -32,7 +32,7 @@ public class DataFilePathFactory {
 
     public static final String INDEX_PATH_SUFFIX = ".index";
 
-    private final Path defaultWriteRootPath;
+    private final Path dataRootLocation;
     private final Path relativeDataFilePath;
     private final Path parent;
     private final String uuid;
@@ -45,14 +45,14 @@ public class DataFilePathFactory {
     private final String fileCompression;
 
     public DataFilePathFactory(
-            Path defaultWriteRootPath,
+            Path dataRootLocation,
             Path relativeDataFilePath,
             String formatIdentifier,
             String dataFilePrefix,
             String changelogFilePrefix,
             boolean fileSuffixIncludeCompression,
             String fileCompression) {
-        this.defaultWriteRootPath = defaultWriteRootPath;
+        this.dataRootLocation = dataRootLocation;
         this.relativeDataFilePath = relativeDataFilePath;
         this.uuid = UUID.randomUUID().toString();
         this.pathCount = new AtomicInteger(0);
@@ -61,7 +61,7 @@ public DataFilePathFactory(
         this.changelogFilePrefix = changelogFilePrefix;
         this.fileSuffixIncludeCompression = fileSuffixIncludeCompression;
         this.fileCompression = fileCompression;
-        this.parent = new Path(this.defaultWriteRootPath, this.relativeDataFilePath);
+        this.parent = new Path(this.dataRootLocation, this.relativeDataFilePath);
     }
 
     public Path newPath() {
@@ -137,7 +137,7 @@ public static String formatIdentifier(String fileName) {
         return fileName.substring(index + 1);
     }
 
-    public Path getDefaultWriteRootPath() {
-        return defaultWriteRootPath;
+    public Path getDataRootLocation() {
+        return dataRootLocation;
     }
 }

paimon-core/src/main/java/org/apache/paimon/io/KeyValueDataFileWriter.java

@@ -76,7 +76,7 @@ public abstract class KeyValueDataFileWriter
     private long minSeqNumber = Long.MAX_VALUE;
     private long maxSeqNumber = Long.MIN_VALUE;
     private long deleteRecordCount = 0;
-    private final Path defaultWriteRootPath;
+    private final Path dataRootLocation;
 
     public KeyValueDataFileWriter(
             FileIO fileIO,
@@ -93,7 +93,7 @@ public KeyValueDataFileWriter(
             CoreOptions options,
             FileSource fileSource,
             FileIndexOptions fileIndexOptions,
-            Path defaultWriteRootPath) {
+            Path dataRootLocation) {
         super(
                 fileIO,
                 factory,
@@ -118,7 +118,7 @@ public KeyValueDataFileWriter(
         this.dataFileIndexWriter =
                 DataFileIndexWriter.create(
                         fileIO, dataFileToFileIndexPath(path), valueType, fileIndexOptions);
-        this.defaultWriteRootPath = defaultWriteRootPath;
+        this.dataRootLocation = dataRootLocation;
     }
 
     @Override
@@ -199,7 +199,7 @@ public DataFileMeta result() throws IOException {
                 indexResult.embeddedIndexBytes(),
                 fileSource,
                 valueStatsPair.getKey(),
-                defaultWriteRootPath);
+                dataRootLocation.toString());
     }
 
     abstract Pair<SimpleColStats[], SimpleColStats[]> fetchKeyValueStats(SimpleColStats[] rowStats);

paimon-core/src/main/java/org/apache/paimon/io/KeyValueDataFileWriterImpl.java

@@ -53,7 +53,7 @@ public KeyValueDataFileWriterImpl(
             CoreOptions options,
             FileSource fileSource,
             FileIndexOptions fileIndexOptions,
-            Path defaultWriteRootPath) {
+            Path dataRootLocation) {
         super(
                 fileIO,
                 factory,
@@ -69,7 +69,7 @@ public KeyValueDataFileWriterImpl(
                 options,
                 fileSource,
                 fileIndexOptions,
-                defaultWriteRootPath);
+                dataRootLocation);
     }
 
     @Override

paimon-core/src/main/java/org/apache/paimon/io/KeyValueFileWriterFactory.java

@@ -126,7 +126,7 @@ private KeyValueDataFileWriter createDataFileWriter(
                         options,
                         fileSource,
                         fileIndexOptions,
-                        formatContext.pathFactory(level).getDefaultWriteRootPath())
+                        formatContext.pathFactory(level).getDataRootLocation())
                 : new KeyValueDataFileWriterImpl(
                         fileIO,
                         formatContext.writerFactory(level),
@@ -141,7 +141,7 @@ private KeyValueDataFileWriter createDataFileWriter(
                         options,
                         fileSource,
                         fileIndexOptions,
-                        formatContext.pathFactory(level).getDefaultWriteRootPath());
+                        formatContext.pathFactory(level).getDataRootLocation());
     }
 
     public void deleteFile(String filename, int level) {

paimon-core/src/main/java/org/apache/paimon/io/KeyValueThinDataFileWriterImpl.java

@@ -78,7 +78,7 @@ public KeyValueThinDataFileWriterImpl(
             CoreOptions options,
             FileSource fileSource,
             FileIndexOptions fileIndexOptions,
-            Path defaultWriteRootPath) {
+            Path dataRootLocation) {
         super(
                 fileIO,
                 factory,
@@ -94,7 +94,7 @@ public KeyValueThinDataFileWriterImpl(
                 options,
                 fileSource,
                 fileIndexOptions,
-                defaultWriteRootPath);
+                dataRootLocation);
         Map<Integer, Integer> idToIndex = new HashMap<>(valueType.getFieldCount());
         for (int i = 0; i < valueType.getFieldCount(); i++) {
             idToIndex.put(valueType.getFields().get(i).id(), i);

paimon-core/src/main/java/org/apache/paimon/io/RowDataFileWriter.java

@@ -52,7 +52,7 @@ public class RowDataFileWriter extends StatsCollectingSingleFileWriter<InternalR
     private final SimpleStatsConverter statsArraySerializer;
     @Nullable private final DataFileIndexWriter dataFileIndexWriter;
     private final FileSource fileSource;
-    private final Path defaultWriteRootPath;
+    private final Path dataRootLocation;
 
     public RowDataFileWriter(
             FileIO fileIO,
@@ -68,7 +68,7 @@ public RowDataFileWriter(
             FileSource fileSource,
             boolean asyncFileWrite,
             boolean statsDenseStore,
-            Path defaultWriteRootPath) {
+            Path dataRootLocation) {
         super(
                 fileIO,
                 factory,
@@ -86,7 +86,7 @@ public RowDataFileWriter(
                 DataFileIndexWriter.create(
                         fileIO, dataFileToFileIndexPath(path), writeSchema, fileIndexOptions);
         this.fileSource = fileSource;
-        this.defaultWriteRootPath = defaultWriteRootPath;
+        this.dataRootLocation = dataRootLocation;
     }
 
     @Override
@@ -128,6 +128,6 @@ public DataFileMeta result() throws IOException {
                 indexResult.embeddedIndexBytes(),
                 fileSource,
                 statsPair.getKey(),
-                defaultWriteRootPath);
+                dataRootLocation.toString());
     }
 }

paimon-core/src/main/java/org/apache/paimon/io/RowDataRollingFileWriter.java

@@ -65,7 +65,7 @@ public RowDataRollingFileWriter(
                                 fileSource,
                                 asyncFileWrite,
                                 statsDenseStore,
-                                pathFactory.getDefaultWriteRootPath()),
+                                pathFactory.getDataRootLocation()),
                 targetFileSize);
     }
 }

paimon-core/src/main/java/org/apache/paimon/io/TablePathProvider.java

@@ -66,25 +66,24 @@ public TablePathProvider(
         this.tableName = tableName;
     }
 
-    public String getTableWritePathString() {
-        return getTableWritePath().toString();
+    public Path getTableWriteDataPath() {
+        return new Path(getDataRootLocation(), new Path(databaseName + "/" + tableName));
     }
 
-    public Path getTableWritePath() {
-        Path location = dataFileExternalPath != null ? dataFileExternalPath : warehouseRootPath;
-        return new Path(location, new Path(databaseName + "/" + tableName));
+    public Path getTableSchemaPath() {
+        return new Path(warehouseRootPath, new Path(databaseName + "/" + tableName));
     }
 
     public Path getReleativeTableWritePath() {
         return new Path(databaseName + "/" + tableName);
     }
 
-    public Path getDataFileExternalPath() {
+    public Path getDataRootLocation() {
         return dataFileExternalPath != null ? dataFileExternalPath : warehouseRootPath;
     }
 
-    public String getWarehouseRootPathString() {
-        return warehouseRootPath.toString();
+    public @NotNull Path getWarehouseRootPath() {
+        return warehouseRootPath;
     }
 
     public String getDatabaseName() {

paimon-core/src/main/java/org/apache/paimon/migrate/FileMetaUtils.java

@@ -158,7 +158,7 @@ private static DataFileMeta constructFileMeta(
                 simpleStatsExtractor.extractWithFileInfo(fileIO, path);
         SimpleStats stats = statsArraySerializer.toBinaryAllMode(fileInfo.getLeft());
 
-        String dataRootPath = ((FileStoreTable) table).coreOptions().getDataWriteRootPath();
+        String dataRootLocation = ((FileStoreTable) table).coreOptions().getDataRootLocation();
         return DataFileMeta.forAppend(
                 fileName,
                 fileSize,
@@ -171,7 +171,7 @@ private static DataFileMeta constructFileMeta(
                 null,
                 FileSource.APPEND,
                 null,
-                new Path(dataRootPath));
+                dataRootLocation);
     }
 
     public static BinaryRow writePartitionValue(

paimon-core/src/main/java/org/apache/paimon/schema/SchemaManager.java

@@ -237,6 +237,7 @@ public TableSchema createTable(Schema schema, boolean externalTable) throws Exce
                             primaryKeys,
                             options,
                             schema.comment());
+
             // validate table from creating table
             FileStoreTableFactory.create(fileIO, tableRoot, newSchema).store();