[AURON #2434] Support full-data-file Iceberg changelog deletes - #2435
[AURON #2434] Support full-data-file Iceberg changelog deletes#2435weimingdiit wants to merge 3 commits into
Conversation
Allow native Iceberg changelog scan to handle DeletedDataFileScanTask when the task represents a DELETE operation and has no existing delete files. Keep existing support for insert changelog tasks and continue falling back for row-level deletes, position/equality deletes, update changelog tasks, mixed file formats, and delete tasks with existing deletes. Add tests for full-data-file delete changelog scans and mixed insert/delete changelog ranges. Signed-off-by: weimingdiit <weimingdiit@gmail.com>
merrily01
left a comment
merrily01
left a comment
There was a problem hiding this comment.
@weimingdiit Thanks for the contribution. The implementation looks correct to me, just a few comments inline.
| test("iceberg changelog scan falls back when delete changes exist") { | ||
| withTable("local.db.t_changelog_delete") { | ||
| withTempView("t_changelog_delete_changes") { | ||
| test("iceberg native scan supports mixed insert and full-data-file delete changelog scan") { |
There was a problem hiding this comment.
This replaces the previous falls back when delete changes exist test, which covered row-level deletes. Could we keep a dedicated fallback test (e.g. non-partitioned table with a position delete, or a DeletedDataFileScanTask with non-empty existingDeletes())? It would help guard against regressions on the row-level delete / update paths.
There was a problem hiding this comment.
Thanks for pointing this out. I rechecked the behavior of the replaced test and Iceberg 1.10.1.
The replaced test did not actually exercise a row-level delete. In the CI run, the insert produced two data files and the DELETE removed one complete data file, so it covered a DeletedDataFileScanTask with empty existingDeletes(), which is the case this PR now makes native.
A genuine position-delete changelog case cannot currently reach Auron's fallback guard. BaseIncrementalChangelogScan rejects any snapshot in the changelog range that contains delete manifests before Auron receives the planned tasks. In addition, Iceberg 1.10.1's CreateDataFileChangeTasks constructs both AddedRowsScanTask and DeletedDataFileScanTask with NO_DELETES, so a DeletedDataFileScanTask with non-empty existingDeletes() cannot be produced through the public changelog planner.
The existing "iceberg scan falls back when delete files exist" test still covers position-delete fallback for regular Iceberg scans. It explicitly creates a position delete, verifies that the planned FileScanTask contains deletes, and checks that NativeIcebergTableScan is not used.
Given these limitations, I have not added a changelog integration test that would either fail inside Iceberg before reaching Auron or not exercise the intended guard. If useful, I can add a focused synthetic task-level test for the existingDeletes() guard instead.
| deletesEmpty(added.deletes()) => | ||
| Some(NativeChangelogDataFileTask(added.file(), added.start(), added.length(), added)) | ||
| case deleted: DeletedDataFileScanTask | ||
| if deleted.operation() == ChangelogOperation.DELETE && |
There was a problem hiding this comment.
Nit: deleted.operation() == ChangelogOperation.DELETE is always true. How about dropping it or adding a comment for clarity?
There was a problem hiding this comment.
Agreed. DeletedDataFileScanTask already defines the operation as DELETE, so I removed the redundant operation check and kept only the existingDeletes() guard.
Signed-off-by: weimingdiit <weimingdiit@gmail.com>
Signed-off-by: weimingdiit <weimingdiit@gmail.com>
There was a problem hiding this comment.
🟡 Not ready to approve
The native-acceptance guard for DeletedDataFileScanTask does not enforce the PR’s stated operation == DELETE constraint, risking incorrect native execution for non-DELETE changelog tasks.
Once you've addressed the issues Copilot identified, you can request another Copilot review.
This review doesn't count toward merge requirements. Sign up for the private preview to control whether Copilot approvals count.
Pull request overview
This PR expands Auron’s native Iceberg changelog scan support to include full-data-file delete changelog tasks (while keeping row-level delete semantics on Spark’s reader), improving native coverage for simple DELETE changelog ranges.
Changes:
- Extend native changelog planning to accept
DeletedDataFileScanTaskwhen it can be handled by scanning the data file directly. - Refactor changelog task handling to a unified “native changelog data-file task” representation used for native scan task construction.
- Add integration tests covering full-data-file delete changelog scans and mixed insert + full-data-file delete ranges.
File summaries
| File | Description |
|---|---|
| thirdparty/auron-iceberg/src/main/scala/org/apache/spark/sql/auron/iceberg/IcebergScanSupport.scala | Adds native handling for full-data-file delete changelog tasks by mapping supported changelog tasks into native scan tasks. |
| thirdparty/auron-iceberg/src/test/scala/org/apache/auron/iceberg/AuronIcebergIntegrationSuite.scala | Adds integration coverage for native execution of delete-only and mixed insert+delete changelog ranges. |
Review details
- Files reviewed: 2/2 changed files
- Comments generated: 2
- Review effort level: Lite
We're testing this review assessment. Please use 👍 or 👎 to tell us if it's correct.
| private def toNativeChangelogDataFileTask( | ||
| task: ChangelogScanTask): Option[NativeChangelogDataFileTask] = { | ||
| task match { | ||
| case added: AddedRowsScanTask | ||
| if added.operation() == ChangelogOperation.INSERT && | ||
| deletesEmpty(added.deletes()) => | ||
| Some(NativeChangelogDataFileTask(added.file(), added.start(), added.length(), added)) | ||
| case deleted: DeletedDataFileScanTask if deletesEmpty(deleted.existingDeletes()) => | ||
| Some( | ||
| NativeChangelogDataFileTask(deleted.file(), deleted.start(), deleted.length(), deleted)) | ||
| case _ => | ||
| None | ||
| } | ||
| } |
| test("iceberg native scan supports mixed insert and full-data-file delete changelog scan") { | ||
| withTable("local.db.t_changelog_mixed_delete") { | ||
| withTempView("t_changelog_mixed_delete_changes") { | ||
| sql(""" | ||
| |create table local.db.t_changelog_delete (id int, v string) | ||
| |create table local.db.t_changelog_mixed_delete (id int, v string, p int) |
Which issue does this PR close?
Closes #2434
Rationale for this change
Auron native Iceberg changelog scan currently supports insert-only changelog tasks.
Full-data-file delete changelog tasks can be executed natively without row-level delete-file handling. Auron can scan the deleted data file directly and materialize changelog metadata from the Iceberg changelog task.
Supporting this case improves native Iceberg changelog scan coverage while keeping more complex delete and update semantics on Spark's reader.
What changes are included in this PR?
Allows native Iceberg changelog scan to accept
DeletedDataFileScanTaskwhen:DELETEexistingDeletes()is emptyKeeps existing native support for
AddedRowsScanTaskwhen:INSERTdeletes()is emptyKeeps fallback behavior for unsupported changelog tasks, including row-level deletes, position/equality deletes, update changelog tasks, mixed file formats, and delete tasks with existing delete files.
Adds coverage for:
Are there any user-facing changes?
No user-facing API changes. More Iceberg changelog scans can now be executed natively by Auron.
How was this patch tested?
UT.