This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new cf2eb3fff6 [core] Keep staging trees out of an overwrite that clears a
prefix (#8905)
cf2eb3fff6 is described below
commit cf2eb3fff68f12bb9c54d6188865b11ca37c8799
Author: Dapeng Sun(孙大鹏) <[email protected]>
AuthorDate: Thu Jul 30 12:01:05 2026 +0800
[core] Keep staging trees out of an overwrite that clears a prefix (#8905)
---
.../table/format/FormatBatchWriteBuilder.java | 1 +
.../paimon/table/format/FormatTableCommit.java | 39 ++++--
.../paimon/table/format/FormatTableScan.java | 68 ++++++++--
.../paimon/table/format/FormatTableCommitTest.java | 145 +++++++++++++++++++++
.../paimon/table/format/FormatTableScanTest.java | 27 ----
.../paimon/utils/PartitionPathUtilsTest.java | 14 ++
6 files changed, 242 insertions(+), 52 deletions(-)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatBatchWriteBuilder.java
b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatBatchWriteBuilder.java
index 0b510c5940..b14ab00890 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatBatchWriteBuilder.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatBatchWriteBuilder.java
@@ -83,6 +83,7 @@ public class FormatBatchWriteBuilder implements
BatchWriteBuilder {
table.partitionKeys(),
table.fileIO(),
formatTablePartitionOnlyValueInPath,
+ table.defaultPartName(),
overwrite,
Identifier.fromString(table.fullName()),
staticPartition,
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableCommit.java
b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableCommit.java
index c0f356fe22..278e4c4f36 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableCommit.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableCommit.java
@@ -56,6 +56,7 @@ public class FormatTableCommit implements BatchTableCommit {
private String location;
private final boolean formatTablePartitionOnlyValueInPath;
+ private final String defaultPartName;
private FileIO fileIO;
private List<String> partitionKeys;
protected Map<String, String> staticPartitions;
@@ -69,6 +70,7 @@ public class FormatTableCommit implements BatchTableCommit {
List<String> partitionKeys,
FileIO fileIO,
boolean formatTablePartitionOnlyValueInPath,
+ String defaultPartName,
boolean overwrite,
Identifier tableIdentifier,
@Nullable Map<String, String> staticPartitions,
@@ -78,6 +80,7 @@ public class FormatTableCommit implements BatchTableCommit {
this.location = location;
this.fileIO = fileIO;
this.formatTablePartitionOnlyValueInPath =
formatTablePartitionOnlyValueInPath;
+ this.defaultPartName = defaultPartName;
validateStaticPartition(staticPartitions, partitionKeys);
this.staticPartitions = staticPartitions;
this.overwrite = overwrite;
@@ -128,7 +131,10 @@ public class FormatTableCommit implements BatchTableCommit
{
partitionSpecs.add(staticPartitions);
}
if (overwrite) {
- deletePreviousDataFile(partitionPath);
+ // A static partition may name only the leading keys, in
which case the path
+ // is a prefix and the partition directories of the
remaining keys sit below.
+ deletePreviousDataFile(
+ partitionPath, partitionKeys.size() -
staticPartitions.size());
}
if (!fileIO.exists(partitionPath)) {
fileIO.mkdirs(partitionPath);
@@ -139,7 +145,10 @@ public class FormatTableCommit implements BatchTableCommit
{
partitionPaths.add(c.targetPath().getParent());
}
for (Path p : partitionPaths) {
- deletePreviousDataFile(p);
+ // The parent of a written file is a complete partition
directory - the table
+ // directory itself when the table is unpartitioned - so
there is no partition
+ // level below it to descend.
+ deletePreviousDataFile(p, 0);
}
}
@@ -267,17 +276,23 @@ public class FormatTableCommit implements
BatchTableCommit {
@Override
public void close() throws Exception {}
- private void deletePreviousDataFile(Path partitionPath) throws IOException
{
+ private void deletePreviousDataFile(Path partitionPath, int
partitionLevels)
+ throws IOException {
if (fileIO.exists(partitionPath)) {
- FileStatus[] files = fileIO.listFiles(partitionPath, true);
- for (FileStatus file : files) {
- if (FormatTableScan.isDataFileName(file.getPath().getName())) {
- try {
- fileIO.delete(file.getPath(), false);
- } catch (FileNotFoundException ignore) {
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
+ // Committed data files only: what sits under a staging directory
is another writer's
+ // uncommitted output, whatever its name looks like.
+ for (FileStatus file :
+ FormatTableScan.listDataFiles(
+ fileIO,
+ partitionPath,
+ partitionLevels,
+ formatTablePartitionOnlyValueInPath,
+ defaultPartName)) {
+ try {
+ fileIO.delete(file.getPath(), false);
+ } catch (FileNotFoundException ignore) {
+ } catch (IOException e) {
+ throw new RuntimeException(e);
}
}
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableScan.java
b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableScan.java
index f86a3439c9..3f059a0e59 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableScan.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableScan.java
@@ -101,10 +101,6 @@ public class FormatTableScan implements InnerTableScan {
throw new UnsupportedOperationException("Filter is not supported for
FormatTable.");
}
- public static boolean isDataFileName(String fileName) {
- return fileName != null && !PartitionPathUtils.isHiddenName(fileName);
- }
-
/**
* Lists the data files under {@code listedRoot}, skipping committer
staging trees ({@code
* _temporary/}, {@code __magic_job-<id>/}, {@code .hive-staging_*})
without descending into
@@ -120,16 +116,54 @@ public class FormatTableScan implements InnerTableScan {
* disappears further down is skipped instead, leaving the rest of the
listing complete
*/
static List<FileStatus> listDataFiles(FileIO fileIO, Path listedRoot)
throws IOException {
+ return listDataFiles(fileIO, listedRoot, 0, false, null);
+ }
+
+ /**
+ * As {@link #listDataFiles(FileIO, Path)}, for a root that still has
partition directories
+ * below it: an {@code INSERT OVERWRITE} naming only the leading partition
keys clears such a
+ * prefix.
+ *
+ * <p>Being at a partition level does not exempt a directory from the
{@code '_'} / {@code '.'}
+ * rule. A job writing the same prefix with the trailing keys dynamic
stages exactly there, so
+ * {@code year=2025/_temporary} holds that job's own month directories,
not this table's. One
+ * hidden name is table content: the default partition name in the
value-only layout, where a
+ * partition directory is the bare value. That is the exemption {@link
+ * PartitionPathUtils#searchPartSpecAndPaths} already makes on the scan
side.
+ *
+ * @param partitionLevels how many directory levels below {@code
listedRoot} hold partition
+ * directories rather than table content
+ * @param onlyValueInPath whether a partition directory is named by its
value alone ({@code
+ * 2025/}) instead of {@code key=value} ({@code year=2025/})
+ * @param defaultPartName the directory name standing for a null partition
value
+ */
+ static List<FileStatus> listDataFiles(
+ FileIO fileIO,
+ Path listedRoot,
+ int partitionLevels,
+ boolean onlyValueInPath,
+ @Nullable String defaultPartName)
+ throws IOException {
+ String partitionDirExemptFromHiding = onlyValueInPath ?
defaultPartName : null;
List<FileStatus> dataFiles = new ArrayList<>();
List<Path> level = new ArrayList<>();
// A missing root is the caller's signal, e.g. a partition that the
catalog knows but whose
// directory is gone, so let it surface.
- collectDataFiles(fileIO.listStatus(listedRoot), dataFiles, level);
- while (!level.isEmpty()) {
+ collectDataFiles(
+ fileIO.listStatus(listedRoot),
+ partitionLevels >= 1 ? partitionDirExemptFromHiding : null,
+ dataFiles,
+ level);
+ for (int depth = 1; !level.isEmpty(); depth++) {
+ boolean childrenArePartitions = partitionLevels >= depth + 1;
List<Path> next = new ArrayList<>();
for (Path directory : level) {
try {
- collectDataFiles(fileIO.listStatus(directory), dataFiles,
next);
+ collectDataFiles(
+ fileIO.listStatus(directory),
+ childrenArePartitions ?
partitionDirExemptFromHiding : null,
+ dataFiles,
+ next);
} catch (FileNotFoundException e) {
// The directory vanished after its parent listed it; the
rest of the listing
// is still complete.
@@ -140,18 +174,26 @@ public class FormatTableScan implements InnerTableScan {
return dataFiles;
}
+ /**
+ * @param exemptFromHiding the one hidden directory name that holds table
content here, or null
+ * when every hidden directory is a staging tree
+ */
private static void collectDataFiles(
- @Nullable FileStatus[] children, List<FileStatus> dataFiles,
List<Path> directories) {
+ @Nullable FileStatus[] children,
+ @Nullable String exemptFromHiding,
+ List<FileStatus> dataFiles,
+ List<Path> directories) {
if (children == null) {
return;
}
for (FileStatus child : children) {
- if (PartitionPathUtils.isHiddenName(child.getPath().getName())) {
- continue;
- }
+ String name = child.getPath().getName();
+ boolean hidden = PartitionPathUtils.isHiddenName(name);
if (child.isDir()) {
- directories.add(child.getPath());
- } else {
+ if (!hidden || name.equals(exemptFromHiding)) {
+ directories.add(child.getPath());
+ }
+ } else if (!hidden) {
dataFiles.add(child);
}
}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableCommitTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableCommitTest.java
index c7f8c8dac1..c7a557b449 100644
---
a/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableCommitTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableCommitTest.java
@@ -37,6 +37,7 @@ import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
+import static org.apache.paimon.CoreOptions.PARTITION_DEFAULT_NAME;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.assertj.core.api.Assertions.entry;
@@ -72,6 +73,7 @@ class FormatTableCommitTest {
Arrays.asList("year", "month"),
fileIO,
false,
+ PARTITION_DEFAULT_NAME.defaultValue(),
false,
Identifier.create("catalog_partition_db",
"catalog_partition_table"),
null,
@@ -103,6 +105,7 @@ class FormatTableCommitTest {
Arrays.asList("year", "month"),
fileIO,
false,
+ PARTITION_DEFAULT_NAME.defaultValue(),
false,
Identifier.create("catalog_partition_db",
"catalog_partition_table"),
null,
@@ -164,6 +167,146 @@ class FormatTableCommitTest {
.containsExactly(entry("year", "2025"), entry("month", "a:b"));
}
+ @Test
+ void testOverwriteKeepsFilesOfConcurrentWritersStagingTrees() throws
Exception {
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path tablePath = new Path(tempDir.toUri());
+ Path partitionPath = new Path(tablePath, "year=2025/month=10");
+ Path previousDataFile = new Path(partitionPath, "data-old.csv");
+ fileIO.writeFile(previousDataFile, "1", false);
+ // A concurrent job is mid-write in this partition, under a magic
committer's tree. Its
+ // file carries an ordinary data file name; only the directories above
it say otherwise.
+ // ('_temporary' is the other such tree, but this writer's own clean()
still empties it -
+ // covered once that is fixed separately.)
+ Path stagingFile =
+ new Path(
+ partitionPath,
+
"__magic_job-6e7f/tasks/attempt_202607271200_0001_m_000010_15"
+ + "/__base/part-00010.csv");
+ fileIO.writeFile(stagingFile, "2", false);
+
+ Path targetPath = new Path(partitionPath, "data-new.csv");
+ RenamingTwoPhaseOutputStream outputStream =
+ new RenamingTwoPhaseOutputStream(fileIO, targetPath, false);
+ outputStream.write(1);
+ TwoPhaseOutputStream.Committer committer =
outputStream.closeForCommit();
+ Map<String, String> staticPartition = new LinkedHashMap<>();
+ staticPartition.put("year", "2025");
+ staticPartition.put("month", "10");
+ FormatTableCommit commit =
+ new FormatTableCommit(
+ tablePath.toString(),
+ Arrays.asList("year", "month"),
+ fileIO,
+ false,
+ PARTITION_DEFAULT_NAME.defaultValue(),
+ true,
+ Identifier.create("overwrite_db", "overwrite_table"),
+ staticPartition,
+ null,
+ null,
+ null);
+
+ commit.commit(Collections.singletonList(new
TwoPhaseCommitMessage(committer)));
+
+ assertThat(fileIO.exists(previousDataFile)).isFalse();
+ assertThat(fileIO.exists(targetPath)).isTrue();
+ assertThat(fileIO.exists(stagingFile)).isTrue();
+ }
+
+ @Test
+ void testOverwritingAPrefixKeepsStagingTreesSittingAtAPartitionLevel()
throws Exception {
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path tablePath = new Path(tempDir.toUri());
+ Path staticPrefix = new Path(tablePath, "year=2025");
+ Path staleFile = new Path(staticPrefix, "month=10/data-old.csv");
+ fileIO.writeFile(staleFile, "1", false);
+ // A concurrent job overwriting the same prefix with a dynamic month
stages below the
+ // prefix itself, so its staging root stands where the month
directories do rather than
+ // inside one of them. Being at a partition level does not make it
partition data: the
+ // month directories it holds are the job's own, to be moved into
place at its commit.
+ List<Path> stagingFiles =
+ Arrays.asList(
+ new Path(staticPrefix, "_temporary/attempt/part.csv"),
+ new Path(
+ staticPrefix,
+
"_temporary/0/attempt_202607271200_0001_m_000012_17"
+ + "/month=12/part-00012.csv"),
+ new Path(
+ staticPrefix,
+
"_temporary/0/_temporary/attempt_202607271200_0001_m_000011_16"
+ + "/month=11/part-00011.csv"));
+ for (Path stagingFile : stagingFiles) {
+ fileIO.writeFile(stagingFile, "2", false);
+ }
+
+ // INSERT OVERWRITE ... PARTITION (year = '2025'), month left dynamic.
+ FormatTableCommit commit =
+ new FormatTableCommit(
+ tablePath.toString(),
+ Arrays.asList("year", "month"),
+ fileIO,
+ false,
+ PARTITION_DEFAULT_NAME.defaultValue(),
+ true,
+ Identifier.create("overwrite_db", "overwrite_table"),
+ Collections.singletonMap("year", "2025"),
+ null,
+ null,
+ null);
+
+ commit.commit(Collections.emptyList());
+
+ assertThat(fileIO.exists(staleFile)).isFalse();
+ for (Path stagingFile : stagingFiles) {
+ assertThat(fileIO.exists(stagingFile)).isTrue();
+ }
+ }
+
+ @Test
+ void testOverwritingAPrefixClearsTheDefaultPartitionDirectory() throws
Exception {
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path tablePath = new Path(tempDir.toUri());
+ // Value-only layout: the directory of a null partition value is the
default partition
+ // name, which starts with '_' without being a staging directory. The
scan reads it, so an
+ // overwrite has to clear it too.
+ Path defaultPartition =
+ new Path(tablePath, "2025/" +
PARTITION_DEFAULT_NAME.defaultValue());
+ Path staleFile = new Path(defaultPartition, "data-old.csv");
+ fileIO.writeFile(staleFile, "1", false);
+ Path staleSibling = new Path(tablePath, "2025/10/data-old.csv");
+ fileIO.writeFile(staleSibling, "1", false);
+ Path stagedFile = new Path(defaultPartition,
"__magic_job-6e7f/__base/part-00010.csv");
+ fileIO.writeFile(stagedFile, "2", false);
+ // The exemption is that one directory name and no other: a staging
tree standing next to
+ // the partition directories is still a staging tree.
+ Path stagedNextToThePartitions =
+ new Path(tablePath, "2025/_temporary/attempt/part-00011.csv");
+ fileIO.writeFile(stagedNextToThePartitions, "2", false);
+
+ FormatTableCommit commit =
+ new FormatTableCommit(
+ tablePath.toString(),
+ Arrays.asList("year", "month"),
+ fileIO,
+ true,
+ PARTITION_DEFAULT_NAME.defaultValue(),
+ true,
+ Identifier.create("overwrite_db", "overwrite_table"),
+ Collections.singletonMap("year", "2025"),
+ null,
+ null,
+ null);
+
+ commit.commit(Collections.emptyList());
+
+ assertThat(fileIO.exists(staleFile)).isFalse();
+ assertThat(fileIO.exists(staleSibling)).isFalse();
+ // Inside the partition, hidden still means staging.
+ assertThat(fileIO.exists(stagedFile)).isTrue();
+ assertThat(fileIO.exists(stagedNextToThePartitions)).isTrue();
+ }
+
@Test
void testPathNotMatchingThePartitionKeysFails() {
Path tablePath = new Path(tempDir.toUri());
@@ -194,6 +337,7 @@ class FormatTableCommitTest {
Collections.singletonList("year"),
fileIO,
true,
+ PARTITION_DEFAULT_NAME.defaultValue(),
true,
Identifier.create("catalog_partition_db",
"catalog_partition_table"),
staticPartition,
@@ -225,6 +369,7 @@ class FormatTableCommitTest {
Arrays.asList("year", "month"),
fileIO,
onlyValueInPath,
+ PARTITION_DEFAULT_NAME.defaultValue(),
false,
Identifier.create("catalog_partition_db",
"catalog_partition_table"),
null,
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableScanTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableScanTest.java
index 7454f89847..53ba4cd067 100644
---
a/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableScanTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableScanTest.java
@@ -67,7 +67,6 @@ import static
org.apache.paimon.CoreOptions.SOURCE_SPLIT_TARGET_SIZE;
import static
org.apache.paimon.utils.PartitionPathUtils.searchPartSpecAndPaths;
import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
/** Test for {@link FormatTableScan}. */
@@ -99,32 +98,6 @@ public class FormatTableScanTest {
new Object[] {false, "/year=2023/month=2"}, new Object[]
{true, "/2023/2"});
}
- @TestTemplate
- void testValidDataFileNames() {
- // Test valid data file names
- String[] fileNames = {"File.txt", "file.txt", "123file.txt", "data",
"Test_file.log"};
- for (String fileName : fileNames) {
- assertTrue(
- FormatTableScan.isDataFileName(fileName),
- "Filename '" + fileName + "' should be valid");
- }
- }
-
- @TestTemplate
- void testInvalidDataFileNames() {
- String[] fileNames = {".hidden", "_file.txt"};
- for (String fileName : fileNames) {
- assertFalse(
- FormatTableScan.isDataFileName(fileName),
- "Filename '" + fileName + "' should be invalid");
- }
- }
-
- @TestTemplate
- void testNullInput() {
- assertFalse(FormatTableScan.isDataFileName(null), "Null input should
return false");
- }
-
@TestTemplate
void testComputeScanPathAndLevelNoPartitionKeys() {
List<String> partitionKeys = Collections.emptyList();
diff --git
a/paimon-core/src/test/java/org/apache/paimon/utils/PartitionPathUtilsTest.java
b/paimon-core/src/test/java/org/apache/paimon/utils/PartitionPathUtilsTest.java
index 6b29527780..6026eadc04 100644
---
a/paimon-core/src/test/java/org/apache/paimon/utils/PartitionPathUtilsTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/utils/PartitionPathUtilsTest.java
@@ -151,4 +151,18 @@ class PartitionPathUtilsTest {
assertThat(mightMatch(p, 0, 1, GenericRow.of(2025, 6))).isTrue();
assertThat(mightMatch(p, 0, 1, GenericRow.of(2025, 5))).isFalse();
}
+
+ @Test
+ void testIsHiddenName() {
+ for (String name :
+ new String[] {"File.txt", "file.txt", "123file.txt", "data",
"a_b.log"}) {
+
assertThat(PartitionPathUtils.isHiddenName(name)).as(name).isFalse();
+ }
+ for (String name : new String[] {".hidden", "_file.txt", "_temporary",
"__magic_job-1"}) {
+
assertThat(PartitionPathUtils.isHiddenName(name)).as(name).isTrue();
+ }
+ // A name that is absent or empty is not a hidden one.
+ assertThat(PartitionPathUtils.isHiddenName(null)).isFalse();
+ assertThat(PartitionPathUtils.isHiddenName("")).isFalse();
+ }
}
