diff --git a/paimon-common/src/main/java/org/apache/paimon/fs/RenamingTwoPhaseOutputStream.java b/paimon-common/src/main/java/org/apache/paimon/fs/RenamingTwoPhaseOutputStream.java index f5dcdf7a1b28..14f3dc482e8e 100644 --- a/paimon-common/src/main/java/org/apache/paimon/fs/RenamingTwoPhaseOutputStream.java +++ b/paimon-common/src/main/java/org/apache/paimon/fs/RenamingTwoPhaseOutputStream.java @@ -20,6 +20,9 @@ import org.apache.paimon.annotation.Public; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + import java.io.IOException; import java.util.UUID; @@ -29,6 +32,9 @@ */ @Public public class RenamingTwoPhaseOutputStream extends TwoPhaseOutputStream { + + private static final Logger LOG = LoggerFactory.getLogger(RenamingTwoPhaseOutputStream.class); + private static final String TEMP_DIR_NAME = "_temporary"; private final Path targetPath; @@ -136,7 +142,21 @@ public Path targetPath() { @Override public void clean(FileIO fileIO) { - fileIO.deleteDirectoryQuietly(tempPath.getParent()); + fileIO.deleteQuietly(tempPath); + Path stagingDir = tempPath.getParent(); + if (stagingDir == null) { + return; + } + try { + // '_temporary' is shared with every other writer of this directory, Paimon or not, + // so it may only go away once nothing is staged there any more. A non-recursive + // delete is exactly that question, asked without a listing and without a window + // between the check and the delete: it succeeds when the directory is empty and + // fails while a concurrent writer still has files in it. + fileIO.delete(stagingDir, false); + } catch (IOException e) { + LOG.debug("Staging directory {} is still in use or already gone", stagingDir, e); + } } } } diff --git a/paimon-common/src/test/java/org/apache/paimon/fs/RenamingTwoPhaseOutputStreamTest.java b/paimon-common/src/test/java/org/apache/paimon/fs/RenamingTwoPhaseOutputStreamTest.java index 2b929f18ca1b..d47bbbe4eeae 100644 --- a/paimon-common/src/test/java/org/apache/paimon/fs/RenamingTwoPhaseOutputStreamTest.java +++ b/paimon-common/src/test/java/org/apache/paimon/fs/RenamingTwoPhaseOutputStreamTest.java @@ -71,6 +71,41 @@ void testSuccessfulCommit() throws IOException { assertThat(new String(content)).isEqualTo(testData); } + @Test + void testCleanKeepsTheSharedStagingDirectory() throws IOException { + RenamingTwoPhaseOutputStream stream = + new RenamingTwoPhaseOutputStream(fileIO, targetPath, false); + stream.write("Some data".getBytes()); + TwoPhaseOutputStream.Committer committer = stream.closeForCommit(); + + // A MapReduce-style writer with a task attempt still pending in the same directory. + Path otherWriterPending = + new Path(targetPath.getParent(), "_temporary/attempt_0001_m_000010_15/part-00010"); + fileIO.writeFile(otherWriterPending, "concurrent", false); + + committer.commit(fileIO); + committer.clean(fileIO); + + assertThat(fileIO.exists(targetPath)).isTrue(); + assertThat(fileIO.exists(otherWriterPending)).isTrue(); + assertThat(fileIO.exists(new Path(targetPath.getParent(), "_temporary"))).isTrue(); + } + + @Test + void testCleanRemovesTheStagingDirectoryOnceEmpty() throws IOException { + RenamingTwoPhaseOutputStream stream = + new RenamingTwoPhaseOutputStream(fileIO, targetPath, false); + stream.write("Some data".getBytes()); + TwoPhaseOutputStream.Committer committer = stream.closeForCommit(); + + committer.commit(fileIO); + committer.clean(fileIO); + + // Nothing else was staged, so the writer leaves no trace beyond its committed file. + assertThat(fileIO.exists(targetPath)).isTrue(); + assertThat(fileIO.exists(new Path(targetPath.getParent(), "_temporary"))).isFalse(); + } + @Test void testDiscard() throws IOException { RenamingTwoPhaseOutputStream stream = 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 c0f356fe226d..33dba6757744 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 @@ -128,7 +128,10 @@ public void commit(List commitMessages) { 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 it. + deletePreviousDataFile( + partitionPath, partitionKeys.size() - staticPartitions.size()); } if (!fileIO.exists(partitionPath)) { fileIO.mkdirs(partitionPath); @@ -139,7 +142,8 @@ public void commit(List commitMessages) { partitionPaths.add(c.targetPath().getParent()); } for (Path p : partitionPaths) { - deletePreviousDataFile(p); + // The parent of a written file is always a complete partition directory. + deletePreviousDataFile(p, 0); } } @@ -267,17 +271,18 @@ public void abort(List commitMessages) { @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 belongs to a writer + // that is still running. + for (FileStatus file : + FormatTableScan.listDataFiles(fileIO, partitionPath, partitionLevels)) { + 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 72fdff618291..eb79b4cba3f6 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.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 javax.annotation.Nullable; +import java.io.FileNotFoundException; import java.io.IOException; import java.util.ArrayList; import java.util.HashMap; @@ -99,7 +102,69 @@ public InnerTableScan withFilter(Predicate predicate) { } 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-/}, {@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. + * + *

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}. + * + * @param partitionLevels how many directory levels below {@code listedRoot} hold partition + * directories rather than table content. Those levels are descended whatever they are + * named: a partition value may start with {@code '_'} — the default partition name does, in + * the value-only layout, and {@link PartitionPathUtils} lets the scan read that partition — + * so treating it as staging would leave its files behind. Pass 0 when {@code listedRoot} is + * a single partition or an unpartitioned table. + * @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 listDataFiles(FileIO fileIO, Path listedRoot, int partitionLevels) + throws IOException { + List dataFiles = new ArrayList<>(); + List 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), partitionLevels >= 1, dataFiles, level); + for (int depth = 1; !level.isEmpty(); depth++) { + boolean childrenArePartitions = partitionLevels >= depth + 1; + List next = new ArrayList<>(); + for (Path directory : level) { + try { + collectDataFiles( + fileIO.listStatus(directory), childrenArePartitions, 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, + boolean childrenArePartitions, + List dataFiles, + List directories) { + if (children == null) { + return; + } + for (FileStatus child : children) { + boolean hidden = PartitionPathUtils.isHiddenName(child.getPath().getName()); + if (child.isDir()) { + if (childrenArePartitions || !hidden) { + directories.add(child.getPath()); + } + } else if (!hidden) { + dataFiles.add(child); + } + } } BinaryRow toPartitionRow(LinkedHashMap 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 4bfea41a4459..20a8b6776e3d 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 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,12 @@ BinaryRow toPartitionRow(LinkedHashMap partitionSpec) { List createSplits(FileIO fileIO, Path path, @Nullable BinaryRow partition) throws IOException { List 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: + // no partition directories below it. + List files = FormatTableScan.listDataFiles(fileIO, path, 0); + 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 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 44ded39b6ab5..9e979a98a9da 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 @@ private static boolean isHiddenFile( 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 be7d5e4584da..176dbf829145 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 @@ void testListPartitionEntriesUsesCatalogVisibility() throws Exception { 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 @@ void testMissingCatalogRegisteredPartitionReadsAsEmpty() throws Exception { 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 @@ private static List plannedFiles(List splits) { private static class TrackingLocalFileIO extends LocalFileIO { private final List listedPaths = new ArrayList<>(); - private final List 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 @@ void testParallelListingSkipsMissingCatalogRegisteredPartition() throws Exceptio 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 files = plannedFiles( @@ -569,7 +561,7 @@ void testCatalogPartitionListingIoErrorFailsWholeScan() throws Exception { 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 @@ void testFilesystemDiscoveredMissingDirectoryStillFailsScan() throws Exception { 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 @@ private static class ParallelTrackingLocalFileIO extends LocalFileIO { 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 @@ public FileStatus[] listFiles(Path path, boolean recursive) throws IOException { 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 @@ private static class SerialTrackingLocalFileIO extends LocalFileIO { 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 @@ private static class InjectingLocalFileIO extends LocalFileIO { private final Map 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 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/FormatTableCommitTest.java b/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableCommitTest.java index c7f8c8dac1af..09484501b20d 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.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; @@ -210,6 +211,113 @@ void testValueOnlyStaticPartitionCannotEscapeTableLocation() throws Exception { assertThat(fileIO.exists(siblingPath)).isTrue(); } + @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); + // Two concurrent jobs are mid-write in this partition, one under a magic committer's + // tree and one under the '_temporary' that Paimon's own writer also stages in. Their + // files carry ordinary data file names; only the directories above them say otherwise. + List stagingFiles = + Arrays.asList( + new Path( + partitionPath, + "__magic_job-6e7f/tasks/attempt_202607271200_0001_m_000010_15" + + "/__base/part-00010.csv"), + new Path( + partitionPath, + "_temporary/0/_temporary/attempt_202607271200_0001_m_000011_16" + + "/part-00011.csv")); + for (Path stagingFile : stagingFiles) { + 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 staticPartition = new LinkedHashMap<>(); + staticPartition.put("year", "2025"); + staticPartition.put("month", "10"); + FormatTableCommit commit = + new FormatTableCommit( + tablePath.toString(), + Arrays.asList("year", "month"), + fileIO, + false, + 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(); + 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 + // (PartitionPathUtils.isHiddenFile exempts 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); + + FormatTableCommit commit = + new FormatTableCommit( + tablePath.toString(), + Arrays.asList("year", "month"), + fileIO, + true, + 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(); + } + + @Test + void testCommitLeavesNoStagingDirectoryBehind() throws Exception { + LocalFileIO fileIO = LocalFileIO.create(); + Path tablePath = new Path(tempDir.toUri()); + Path partitionPath = new Path(tablePath, "year=2025/month=10"); + + commitPartitionedFile(tablePath, false, "year=2025/month=10"); + + // Nothing else was staged, so a plain commit leaves only its data file: callers that list + // the directory (and only skip '.'-prefixed names) must not trip over a leftover + // '_temporary'. + assertThat(fileIO.listStatus(partitionPath)) + .extracting(status -> status.getPath().getName()) + .containsExactly("data-1.csv"); + } + private FormatTablePartitionManager commitPartitionedFile( Path tableLocation, boolean onlyValueInPath, String partitionDir) throws Exception { LocalFileIO fileIO = LocalFileIO.create(); 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 3a93233f8936..2f7c2349a14a 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 @@ -833,6 +833,146 @@ public void testCreateSplitsWithParquetFile() throws IOException { 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 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 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 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 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());