Skip to content

[spark] Support V2 delta-based row-level operations for deletion-vector append tables#8539

Open
kerwin-zk wants to merge 1 commit into
apache:masterfrom
kerwin-zk:spark-v2-delta-delete
Open

[spark] Support V2 delta-based row-level operations for deletion-vector append tables#8539
kerwin-zk wants to merge 1 commit into
apache:masterfrom
kerwin-zk:spark-v2-delta-delete

Conversation

@kerwin-zk

@kerwin-zk kerwin-zk commented Jul 10, 2026

Copy link
Copy Markdown
Contributor

Purpose

This PR adds delta-based row-level operations (Spark's SupportsDelta / WriteDelta API) 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:

  • PaimonSparkDeltaOperation implements SupportsDelta with representUpdateAsDeleteAndInsert = false: Spark plans a single UPDATE operation per row (no Expand doubling) and the delta writer decomposes it into a deletion-vector mark plus an appended row.
  • Tasks accumulate per-file deletion bitmaps; the driver merges them across tasks (Spark does not cluster delta rows by file), unions them with the files' existing deletion vectors through 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.
  • The operation captures the read snapshot before the scan is planned, pins the scan to 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.
  • Scope is limited to unaware-bucket deletion-vector append tables: commit conflict detection is deletion-vector aware for them, so concurrent commits touching the same file fail and retry instead of losing deletions. Bucketed deletion-vector append tables keep the V1 path because IndexManifestFileHandler overwrites the whole bucket's index without conflict checking. PK, data-evolution, row-tracking and CHAR tables are unchanged.
  • On Spark 4.1 the built-in rewrites short-circuit on analyzed=true UPDATE/MERGE subtrees (same issue as the existing pure-append rules), so Spark41UpdateTableRewrite / Spark41MergeIntoRewrite transcribe the buildWriteDeltaPlan branches, reusing the protected helpers of RewriteRowLevelCommand.
  • The delta writer reports exact deletedRecords / updatedRecords / insertedRecords task 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-level insertedRecords gauge is filtered out on the delta path because CommitStats.deltaRecordsAppended mixes ADD and DELETE row counts and would overwrite the exact values.
  • Tables on the delta path report __paimon_file_path / __paimon_row_index as 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

  • New 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 and use-v2-write=false keep their current paths). Passed under both the spark3 (3.5) and spark4 (4.1) profiles.
  • PaimonMetricTest extended with exact record metrics assertions for the three delta commands.

@kerwin-zk kerwin-zk force-pushed the spark-v2-delta-delete branch 3 times, most recently from 093992c to c5b2740 Compare July 10, 2026 06:17
@kerwin-zk kerwin-zk changed the title [spark] Support delta-based row-level operations for deletion-vector append tables [spark] Support V2 delta-based row-level operations for deletion-vector append tables Jul 10, 2026
@kerwin-zk kerwin-zk force-pushed the spark-v2-delta-delete branch from c5b2740 to f246867 Compare July 10, 2026 09:06
@kerwin-zk kerwin-zk force-pushed the spark-v2-delta-delete branch from f246867 to 1319082 Compare July 10, 2026 09:18
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant