[core][common] Fix format tables reading and deleting files staged by a committer#8861
Open
sundapeng wants to merge 2 commits into
Open
[core][common] Fix format tables reading and deleting files staged by a committer#8861sundapeng wants to merge 2 commits into
sundapeng wants to merge 2 commits into
Conversation
sundapeng
force-pushed
the
fix/format-table-hidden-dir-listing
branch
from
July 27, 2026 09:16
8c84e6b to
9420b55
Compare
This was referenced Jul 27, 2026
sundapeng
force-pushed
the
fix/format-table-hidden-dir-listing
branch
from
July 27, 2026 09:40
9420b55 to
44f32aa
Compare
…iles A format table lists a partition recursively but decides what is data from the leaf file name only, so files staged by a committer inside the partition (__magic_job-<id>/tasks/attempt_*/__base/part-*, _temporary/..., .hive-staging_*) are read as data and deleted by INSERT OVERWRITE. A 0-byte magic-committer placeholder fails the whole query. List the directory by directory and skip hidden entries before descending, the way Hive, Trino, Spark and Hadoop MapReduce exclude hidden path components from a recursive listing. Pruning rather than post-filtering also keeps the staging tree from being enumerated at all, which matters because leftover staging trees are a normal steady state; for the flat layout a partition normally has, both cost the same single listing. Two things the walk must not confuse with staging. Entries above the listed directory are never judged, so a table under a warehouse path such as oss://bucket/_warehouse/db/t still reads. And a partition value may itself start with '_' - the default partition name does, in the value-only layout - so the levels that hold partition directories are descended whatever they are named: an INSERT OVERWRITE naming only the leading partition keys deletes under a prefix, and skipping the default partition there would leave its rows behind.
sundapeng
force-pushed
the
fix/format-table-hidden-dir-listing
branch
from
July 27, 2026 14:01
44f32aa to
f3c5eb6
Compare
…ge there TempFileCommitter#clean deleted the whole '_temporary' directory of the target directory after committing one file. That directory is shared with every other writer: two concurrent writers into the same directory wiped each other's pending files, and a Hadoop FileOutputCommitter job staging there lost its output. Delete this stream's own temporary file, then ask whether the directory can go by attempting a non-recursive delete: it succeeds when nothing else is staged there and fails while a concurrent writer still has files in it. Asking that way costs no listing and leaves no window between the check and the delete.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Purpose
A committer stages its output inside the partition it writes, under a hidden directory, and the staged files carry ordinary data file names:
A format table lists the partition recursively and then filters on the leaf file name only, so all three become splits. A zero-length placeholder then fails the query with
... is not a Parquet file (length is too low: 0), andINSERT OVERWRITE, driven by the same listing, deletes the other job's files. From the other side,TempFileCommitter#cleandeletes the whole_temporaryafter every commit, and that directory is shared with everyone else writing there.This PR skips hidden entries while walking instead of filtering names afterwards — what
PartitionPathUtils#isHiddenFilealready does for partition discovery (#6522) — and makesclean()drop only its own temporary file, removing the shared directory only when a non-recursive delete succeeds. The two halves are needed together: keeping a staging tree alive while the reader still treats it as data would put uncommitted rows into query results.partitionLevelsmarks the levels that hold partition directories. They are descended whatever they are named, because a partition value may start with_— the default partition name does, in the value-only layout — andINSERT OVERWRITE PARTITION (year = '2025')deletes under a prefix.Brief change log
FormatTableScan#listDataFiles(fileIO, root, partitionLevels): new; walks withlistStatusand prunes hidden subtrees. Used bySplitEnumerator#createSplitsandFormatTableCommit#deletePreviousDataFile.RenamingTwoPhaseOutputStream.TempFileCommitter#clean: delete this stream's temporary file; remove the shared directory only if a non-recursive delete succeeds.PartitionPathUtils#isHiddenName: extracted fromisHiddenFile.Tests
FormatTableScanTest/FormatTableCommitTest: staged files produce no splits in either partition layout; a table under a_-prefixed path still reads;INSERT OVERWRITEleaves both staging trees intact yet still clears the default partition directory.RenamingTwoPhaseOutputStreamTest: a peer's pending file survivesclean(); an uncontended commit leaves no_temporarybehind.mvn -pl paimon-common,paimon-core testis green.API and Format
PartitionPathUtilsgains one public helper (isHiddenName).Documentation