[spark] Support V2 delta-based row-level operations for deletion-vector append tables#8539
Open
kerwin-zk wants to merge 1 commit into
Open
[spark] Support V2 delta-based row-level operations for deletion-vector append tables#8539kerwin-zk wants to merge 1 commit into
kerwin-zk wants to merge 1 commit into
Conversation
093992c to
c5b2740
Compare
c5b2740 to
f246867
Compare
f246867 to
1319082
Compare
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
This PR adds delta-based row-level operations (Spark's
SupportsDelta/WriteDeltaAPI) for deletion-vector enabled unaware-bucket append tables. DELETE, UPDATE and MERGE INTO no longer fall back to the V1 commands: deleted rows are marked in deletion vectors keyed by the(__paimon_file_path, __paimon_row_index)row ID, and inserted rows / new versions of updated rows go through the regular append writer — no data file is rewritten.Design notes:
PaimonSparkDeltaOperationimplementsSupportsDeltawithrepresentUpdateAsDeleteAndInsert = false: Spark plans a single UPDATE operation per row (noExpanddoubling) and the delta writer decomposes it into a deletion-vector mark plus an appended row.BaseAppendDeleteFileMaintainer, and commits index-only messages together with the appended data messages. The driver-side merge and the manifest read that maps touched files to partitions scale with the number of touched files, which should be acceptable for typical row-level workloads; a distributed persist (like the V1 commands'mapGroups) can be introduced later if huge deletes turn out to need it.scan.snapshot-id) and initializes the deletion-vector maintainer from the same snapshot — the same contract as the V1 commands'readSnapshot. The scan, the maintainer and the commit-time file mapping therefore observe one version, and any concurrent deletion-vector change or compaction of the same files surfaces as a commit conflict (retryable) instead of being silently merged, which could otherwise resurrect a concurrently deleted row as the new version of an UPDATE.IndexManifestFileHandleroverwrites the whole bucket's index without conflict checking. PK, data-evolution, row-tracking and CHAR tables are unchanged.analyzed=trueUPDATE/MERGE subtrees (same issue as the existing pure-append rules), soSpark41UpdateTableRewrite/Spark41MergeIntoRewritetranscribe thebuildWriteDeltaPlanbranches, reusing the protected helpers ofRewriteRowLevelCommand.deletedRecords/updatedRecords/insertedRecordstask metrics for all three commands — numbers the copy-on-write path cannot derive from file statistics (see [spark] Support deleted record metrics for V2 delete #8486). The commit-levelinsertedRecordsgauge is filtered out on the delta path becauseCommitStats.deltaRecordsAppendedmixes ADD and DELETE row counts and would overwrite the exact values.__paimon_file_path/__paimon_row_indexas non-nullable metadata columns (Spark rejects nullable row ID columns); all other tables keep them nullable because the V1 commands project these columns through outer joins where the target side can be null (e.g. the not-matched rows of MERGE INTO).Tests
DeltaRowLevelOpsTest: DELETE (partial file / whole file / repeated deletes merging vectors / zero rows / partitioned), UPDATE (values, partition-moving rows, NULL, repeated), MERGE INTO (all clause combinations, NOT MATCHED BY SOURCE, cardinality violation), a concurrency case (a delta commit planned against a stale snapshot fails with a conflict when a concurrent commit adds a deletion vector to the same file, and the failed commit stays invisible), plus negative gating (metadata-only DELETE keeps the V1 truncate fast path; bucketed deletion-vector tables, plain append tables anduse-v2-write=falsekeep their current paths). Passed under both the spark3 (3.5) and spark4 (4.1) profiles.PaimonMetricTestextended with exact record metrics assertions for the three delta commands.