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 6602ec9631 [core] Skip committer staging directories when reading a
format table (#8861)
6602ec9631 is described below
commit 6602ec9631a7b560b736b8e1700eb0a77145dd6d
Author: Dapeng Sun(孙大鹏) <[email protected]>
AuthorDate: Wed Jul 29 14:28:30 2026 +0800
[core] Skip committer staging directories when reading a format table
(#8861)
A format table lists a partition recursively and decides what is data
from the leaf file name alone. Files a committer staged inside the
partition carry ordinary data file names, so they are planned as data:
```
<partition>/__magic<jobid>/tasks/attempt_*/__base/part-00010-*.snappy.parquet
<partition>/_temporary/0/_temporary/attempt_*/part-*
<partition>/.hive-staging_*/-ext-10000/part-*
```
A zero-byte magic-committer placeholder then fails the whole query:
```
... is not a Parquet file (length is too low: 0)
at org.apache.paimon.format.parquet.ParquetReaderFactory.createReader
at org.apache.paimon.table.format.FormatReadBuilder.createFileReader
```
---
.../paimon/table/format/FormatTableScan.java | 57 ++++++-
.../paimon/table/format/SplitEnumerator.java | 10 +-
.../apache/paimon/utils/PartitionPathUtils.java | 11 +-
.../format/CatalogManagedPartitionScanTest.java | 32 ++--
.../paimon/table/format/FormatTableScanTest.java | 167 +++++++++++++++++++++
5 files changed, 249 insertions(+), 28 deletions(-)
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 72fdff6182..f86a3439c9 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
@@ -23,6 +23,8 @@ import org.apache.paimon.casting.CastExecutor;
import org.apache.paimon.casting.CastExecutors;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.FileStatus;
import org.apache.paimon.fs.Path;
import org.apache.paimon.manifest.PartitionEntry;
import org.apache.paimon.partition.PartitionPredicate;
@@ -48,6 +50,7 @@ import org.apache.paimon.utils.PartitionPathUtils;
import javax.annotation.Nullable;
+import java.io.FileNotFoundException;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
@@ -99,7 +102,59 @@ public class FormatTableScan implements InnerTableScan {
}
public static boolean isDataFileName(String fileName) {
- return fileName != null && !fileName.startsWith(".") &&
!fileName.startsWith("_");
+ 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
+ * them. Files staged there carry ordinary data file names, so a name
alone cannot tell them
+ * apart from committed data; the directory above them can.
+ *
+ * <p>Only entries below {@code listedRoot} are judged, never the root
itself, which may
+ * legitimately sit under a warehouse path such as {@code
oss://bucket/_warehouse/db/t}, and
+ * which is the default partition directory of the value-only layout when
a null partition value
+ * is read.
+ *
+ * @throws FileNotFoundException if {@code listedRoot} does not exist; a
directory that
+ * disappears further down is skipped instead, leaving the rest of the
listing complete
+ */
+ static List<FileStatus> listDataFiles(FileIO fileIO, Path listedRoot)
throws IOException {
+ 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()) {
+ List<Path> next = new ArrayList<>();
+ for (Path directory : level) {
+ try {
+ collectDataFiles(fileIO.listStatus(directory), dataFiles,
next);
+ } catch (FileNotFoundException e) {
+ // The directory vanished after its parent listed it; the
rest of the listing
+ // is still complete.
+ }
+ }
+ level = next;
+ }
+ return dataFiles;
+ }
+
+ private static void collectDataFiles(
+ @Nullable FileStatus[] children, List<FileStatus> dataFiles,
List<Path> directories) {
+ if (children == null) {
+ return;
+ }
+ for (FileStatus child : children) {
+ if (PartitionPathUtils.isHiddenName(child.getPath().getName())) {
+ continue;
+ }
+ if (child.isDir()) {
+ directories.add(child.getPath());
+ } else {
+ dataFiles.add(child);
+ }
+ }
}
BinaryRow toPartitionRow(LinkedHashMap<String, String> partitionSpec) {
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/format/SplitEnumerator.java
b/paimon-core/src/main/java/org/apache/paimon/table/format/SplitEnumerator.java
index 4bfea41a44..64c799dddf 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/format/SplitEnumerator.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/format/SplitEnumerator.java
@@ -40,7 +40,6 @@ import javax.annotation.Nullable;
import java.io.IOException;
import java.util.ArrayList;
-import java.util.Arrays;
import java.util.Collections;
import java.util.Comparator;
import java.util.LinkedHashMap;
@@ -109,12 +108,11 @@ abstract class SplitEnumerator {
List<Split> createSplits(FileIO fileIO, Path path, @Nullable BinaryRow
partition)
throws IOException {
List<FormatDataSplit.FileMeta> segments = new ArrayList<>();
- FileStatus[] files = fileIO.listFiles(path, true);
- Arrays.sort(files, Comparator.comparing(file ->
file.getPath().toString()));
+ // The listed directory is a single partition, or the table itself
when unpartitioned.
+ List<FileStatus> files = FormatTableScan.listDataFiles(fileIO, path);
+ files.sort(Comparator.comparing(file -> file.getPath().toString()));
for (FileStatus file : files) {
- if (FormatTableScan.isDataFileName(file.getPath().getName())) {
- segments.addAll(toSegments(file));
- }
+ segments.addAll(toSegments(file));
}
List<Split> splits = new ArrayList<>();
diff --git
a/paimon-core/src/main/java/org/apache/paimon/utils/PartitionPathUtils.java
b/paimon-core/src/main/java/org/apache/paimon/utils/PartitionPathUtils.java
index 44ded39b6a..9e979a98a9 100644
--- a/paimon-core/src/main/java/org/apache/paimon/utils/PartitionPathUtils.java
+++ b/paimon-core/src/main/java/org/apache/paimon/utils/PartitionPathUtils.java
@@ -686,6 +686,15 @@ public class PartitionPathUtils {
if (onlyValueInPath && defaultPartValue != null &&
defaultPartValue.equals(name)) {
return false;
}
- return name.startsWith("_") || name.startsWith(".");
+ return isHiddenName(name);
+ }
+
+ /** Whether a single path component is hidden by the {@code '_'} / {@code
'.'} convention. */
+ public static boolean isHiddenName(String name) {
+ return name != null && !name.isEmpty() &&
isHiddenFirstChar(name.charAt(0));
+ }
+
+ private static boolean isHiddenFirstChar(char c) {
+ return c == '_' || c == '.';
}
}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/format/CatalogManagedPartitionScanTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/format/CatalogManagedPartitionScanTest.java
index be7d5e4584..176dbf8291 100644
---
a/paimon-core/src/test/java/org/apache/paimon/table/format/CatalogManagedPartitionScanTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/table/format/CatalogManagedPartitionScanTest.java
@@ -180,7 +180,6 @@ class CatalogManagedPartitionScanTest {
assertThat(entries.get(0).lastFileCreationTime()).isEqualTo(44L);
assertThat(entries.get(0).totalBuckets()).isEqualTo(5);
assertThat(fileIO.listedPaths).isEmpty();
- assertThat(fileIO.statusListedPaths).isEmpty();
}
@Test
@@ -311,7 +310,7 @@ class CatalogManagedPartitionScanTest {
LocalFileIO fileIO =
new LocalFileIO() {
@Override
- public FileStatus[] listFiles(Path path, boolean
recursive) throws IOException {
+ public FileStatus[] listStatus(Path path) throws
IOException {
assertThat(path).isEqualTo(missingPath);
throw new FileNotFoundException(path.toString());
}
@@ -497,18 +496,11 @@ class CatalogManagedPartitionScanTest {
private static class TrackingLocalFileIO extends LocalFileIO {
private final List<Path> listedPaths = new ArrayList<>();
- private final List<Path> statusListedPaths = new ArrayList<>();
@Override
public FileStatus[] listStatus(Path path) throws IOException {
- statusListedPaths.add(path);
- return super.listStatus(path);
- }
-
- @Override
- public FileStatus[] listFiles(Path path, boolean recursive) throws
IOException {
listedPaths.add(path);
- return super.listFiles(path, recursive);
+ return super.listStatus(path);
}
}
@@ -546,7 +538,7 @@ class CatalogManagedPartitionScanTest {
partition("2025", "10"), partition("2025", "11"),
partition("2025", "12"));
Path oct = writeDataFile(fileIO, tablePath, "year=2025/month=10");
Path dec = writeDataFile(fileIO, tablePath, "year=2025/month=12");
- fileIO.failListFilesContaining("month=11", new
FileNotFoundException("missing"));
+ fileIO.failListingContaining("month=11", new
FileNotFoundException("missing"));
List<Path> files =
plannedFiles(
@@ -569,7 +561,7 @@ class CatalogManagedPartitionScanTest {
Arrays.asList(partition("2025", "10"), partition("2025",
"11"));
writeDataFile(fileIO, tablePath, "year=2025/month=10");
writeDataFile(fileIO, tablePath, "year=2025/month=11");
- fileIO.failListFilesContaining("month=11", new IOException("boom"));
+ fileIO.failListingContaining("month=11", new IOException("boom"));
FormatTable table =
stringPartitionTable(fileIO, tablePath,
recordingCatalog(partitions), 4);
@@ -585,7 +577,7 @@ class CatalogManagedPartitionScanTest {
Path tablePath = new Path(tempDir.toUri());
writeDataFile(fileIO, tablePath, "year=2025/month=10");
writeDataFile(fileIO, tablePath, "year=2025/month=11");
- fileIO.failListFilesContaining("month=11", new
FileNotFoundException("missing"));
+ fileIO.failListingContaining("month=11", new
FileNotFoundException("missing"));
FormatTable table = createStringPartitionTable(fileIO, tablePath,
null);
assertThatThrownBy(() -> new FormatTableScan(table, null,
null).plan().splits())
@@ -670,7 +662,7 @@ class CatalogManagedPartitionScanTest {
private final AtomicInteger maxConcurrentListings = new
AtomicInteger();
@Override
- public FileStatus[] listFiles(Path path, boolean recursive) throws
IOException {
+ public FileStatus[] listStatus(Path path) throws IOException {
int active = activeListings.incrementAndGet();
maxConcurrentListings.updateAndGet(current -> Math.max(current,
active));
concurrentListings.countDown();
@@ -678,7 +670,7 @@ class CatalogManagedPartitionScanTest {
if (!concurrentListings.await(10, TimeUnit.SECONDS)) {
throw new IOException("Partition file listings did not run
concurrently");
}
- return super.listFiles(path, recursive);
+ return super.listStatus(path);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new IOException("Interrupted while waiting for
concurrent listings", e);
@@ -698,12 +690,12 @@ class CatalogManagedPartitionScanTest {
private final AtomicInteger maxConcurrentListings = new
AtomicInteger();
@Override
- public FileStatus[] listFiles(Path path, boolean recursive) throws
IOException {
+ public FileStatus[] listStatus(Path path) throws IOException {
int active = activeListings.incrementAndGet();
maxConcurrentListings.updateAndGet(current -> Math.max(current,
active));
try {
Thread.sleep(50);
- return super.listFiles(path, recursive);
+ return super.listStatus(path);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new IOException("Interrupted while tracking file
listings", e);
@@ -721,18 +713,18 @@ class CatalogManagedPartitionScanTest {
private final Map<String, IOException> listFailures = new
LinkedHashMap<>();
- void failListFilesContaining(String marker, IOException failure) {
+ void failListingContaining(String marker, IOException failure) {
listFailures.put(marker, failure);
}
@Override
- public FileStatus[] listFiles(Path path, boolean recursive) throws
IOException {
+ public FileStatus[] listStatus(Path path) throws IOException {
for (Map.Entry<String, IOException> entry :
listFailures.entrySet()) {
if (path.toString().contains(entry.getKey())) {
throw entry.getValue();
}
}
- return super.listFiles(path, recursive);
+ return super.listStatus(path);
}
}
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 3a93233f89..7454f89847 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
@@ -61,6 +61,7 @@ import java.util.concurrent.atomic.AtomicInteger;
import static org.apache.paimon.CoreOptions.FILE_FORMAT;
import static
org.apache.paimon.CoreOptions.FORMAT_TABLE_PARTITION_ONLY_VALUE_IN_PATH;
+import static org.apache.paimon.CoreOptions.PARTITION_DEFAULT_NAME;
import static org.apache.paimon.CoreOptions.SOURCE_SPLIT_OPEN_FILE_COST;
import static org.apache.paimon.CoreOptions.SOURCE_SPLIT_TARGET_SIZE;
import static
org.apache.paimon.utils.PartitionPathUtils.searchPartSpecAndPaths;
@@ -833,6 +834,172 @@ public class FormatTableScanTest {
assertThat(split.files().get(0).offset()).isEqualTo(0);
}
+ @TestTemplate
+ public void testCreateSplitsSkipsCommitterStagingFiles() throws
IOException {
+ Path tableLocation = new Path(tmpPath.toUri());
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path dataFile = new Path(tableLocation, "data.csv");
+ writeTestFile(fileIO, dataFile, 100);
+ // Committer staging trees hold files whose names look exactly like
data files; only the
+ // directories above them say they are uncommitted. A recursive
listing sees them all.
+ for (String staging :
+ Arrays.asList(
+
"_temporary/0/_temporary/attempt_202607271200_0001_m_000010_15",
+
"__magic/job-6e7f/tasks/attempt_202607271200_0001_m_000010_15/__base",
+
"__magic_job-6e7f/tasks/attempt_202607271200_0001_m_000010_15/__base",
+
".hive-staging_hive_2026-07-27_12-00-00_000_1/-ext-10000")) {
+ writeTestFile(
+ fileIO, new Path(new Path(tableLocation, staging),
"part-00010.csv"), 100);
+ }
+
+ FormatTable formatTable =
+ createFormatTableWithOptions(
+ tableLocation, FormatTable.Format.CSV,
Collections.emptyMap());
+ List<Split> splits = new FormatTableScan(formatTable, null,
null).plan().splits();
+
+ assertThat(splits).hasSize(1);
+ assertThat(((FormatDataSplit) splits.get(0)).files())
+ .extracting(FormatDataSplit.FileMeta::filePath)
+ .containsExactly(dataFile);
+ }
+
+ @TestTemplate
+ public void testStagingTreesAreNotEnumerated() throws IOException {
+ Path tableLocation = new Path(tmpPath.toUri());
+ LocalFileIO setupFileIO = LocalFileIO.create();
+ writeTestFile(setupFileIO, new Path(tableLocation, "data.csv"), 100);
+ // A staging tree with many task attempts, which is what a large job
leaves behind. A
+ // recursive listing walks all of it and then throws it away.
+ for (int attempt = 0; attempt < 20; attempt++) {
+ writeTestFile(
+ setupFileIO,
+ new Path(
+ tableLocation,
+ String.format(
+
"__magic_job-6e7f/tasks/attempt_202607271200_0001_m_%06d_15"
+ + "/__base/part-%05d.csv",
+ attempt, attempt)),
+ 100);
+ }
+
+ AtomicInteger listCount = new AtomicInteger(0);
+ AtomicInteger enumerated = new AtomicInteger(0);
+ LocalFileIO fileIO =
+ new LocalFileIO() {
+ @Override
+ public FileStatus[] listStatus(Path path) throws
IOException {
+ listCount.getAndIncrement();
+ FileStatus[] statuses = super.listStatus(path);
+ enumerated.addAndGet(statuses.length);
+ return statuses;
+ }
+ };
+ FormatTable formatTable =
+ FormatTable.builder()
+ .fileIO(fileIO)
+ .identifier(Identifier.create("test_db", "test_table"))
+ .rowType(
+ RowType.builder()
+ .field("id", DataTypes.INT())
+ .field("name", DataTypes.STRING())
+ .build())
+ .partitionKeys(Collections.emptyList())
+ .location(tableLocation.toString())
+ .format(FormatTable.Format.CSV)
+ .options(Collections.emptyMap())
+ .build();
+
+ // What a recursive listing costs on the same tree, for comparison.
+ FileStatus[] recursive = fileIO.listFiles(tableLocation, true);
+ int recursiveListings = listCount.getAndSet(0);
+ int recursiveEntries = enumerated.getAndSet(0);
+
+ List<Split> splits = new FormatTableScan(formatTable, null,
null).plan().splits();
+
+ assertThat(splits).hasSize(1);
+ // The recursive listing walks the whole staging tree and returns all
of it, leaving the
+ // caller to throw 20 of the 21 files away.
+ assertEquals(21, recursive.length);
+ // Planning skips the staging tree at its root: one listing of the
table directory, and the
+ // two entries in it. The recursive baseline is only compared against,
not pinned: its
+ // exact cost belongs to FileIO, which this test does not exercise.
+ assertEquals(1, listCount.get());
+ assertEquals(2, enumerated.get());
+ assertThat(recursiveListings).isGreaterThan(listCount.get());
+ assertThat(recursiveEntries).isGreaterThan(enumerated.get());
+ }
+
+ @TestTemplate
+ public void testCreateSplitsKeepsFilesUnderAStagingLikeTableLocation()
throws IOException {
+ // The '_' rule applies below the listed directory only: a warehouse
path with a leading
+ // underscore must not make the whole table read as empty.
+ Path tableLocation = new Path(new Path(tmpPath.toUri()),
"_warehouse/db/t");
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path dataFile = new Path(tableLocation, "data.csv");
+ writeTestFile(fileIO, dataFile, 100);
+
+ FormatTable formatTable =
+ createFormatTableWithOptions(
+ tableLocation, FormatTable.Format.CSV,
Collections.emptyMap());
+ List<Split> splits = new FormatTableScan(formatTable, null,
null).plan().splits();
+
+ assertThat(splits).hasSize(1);
+ assertThat(((FormatDataSplit) splits.get(0)).files())
+ .extracting(FormatDataSplit.FileMeta::filePath)
+ .containsExactly(dataFile);
+ }
+
+ @TestTemplate
+ public void testCreateSplitsSkipsStagingFilesInsidePartitions() throws
IOException {
+ Path tableLocation = new Path(tmpPath.toUri());
+ LocalFileIO fileIO = LocalFileIO.create();
+ String partition = enablePartitionValueOnly ? "2024/1" :
"year=2024/month=1";
+ Path partitionPath = new Path(tableLocation, partition);
+ Path dataFile = new Path(partitionPath, "data.csv");
+ writeTestFile(fileIO, dataFile, 100);
+ writeTestFile(
+ fileIO,
+ new Path(
+ partitionPath,
+
"__magic_job-6e7f/tasks/attempt_202607271200_0001_m_000010_15"
+ + "/__base/part-00010.csv"),
+ 100);
+
+ FormatTable formatTable =
createYearMonthFormatTable(LocalFileIO.create(), tableLocation);
+ List<Split> splits = new FormatTableScan(formatTable, null,
null).plan().splits();
+
+ assertThat(splits).hasSize(1);
+ assertThat(((FormatDataSplit) splits.get(0)).files())
+ .extracting(FormatDataSplit.FileMeta::filePath)
+ .containsExactly(dataFile);
+ }
+
+ @TestTemplate
+ public void testCreateSplitsReadsTheDefaultPartitionDirectory() throws
IOException {
+ // In the value-only layout the directory of a null partition value is
the default
+ // partition name, which starts with '_'. It is the listed root here,
and the root is
+ // never judged, so its files are read while what is staged inside it
is not.
+ Path tableLocation = new Path(tmpPath.toUri());
+ LocalFileIO fileIO = LocalFileIO.create();
+ String defaultPartName = PARTITION_DEFAULT_NAME.defaultValue();
+ String partition =
+ enablePartitionValueOnly
+ ? "2024/" + defaultPartName
+ : "year=2024/month=" + defaultPartName;
+ Path partitionPath = new Path(tableLocation, partition);
+ Path dataFile = new Path(partitionPath, "data.csv");
+ writeTestFile(fileIO, dataFile, 100);
+ writeTestFile(fileIO, new Path(partitionPath,
"_temporary/attempt/part-00010.csv"), 100);
+
+ FormatTable formatTable =
createYearMonthFormatTable(LocalFileIO.create(), tableLocation);
+ List<Split> splits = new FormatTableScan(formatTable, null,
null).plan().splits();
+
+ assertThat(splits).hasSize(1);
+ assertThat(((FormatDataSplit) splits.get(0)).files())
+ .extracting(FormatDataSplit.FileMeta::filePath)
+ .containsExactly(dataFile);
+ }
+
@TestTemplate
public void testCreateSplitsWithEmptyDirectory() throws IOException {
Path tableLocation = new Path(tmpPath.toUri());
