Marouane-elfadli opened a new pull request, #18824:
URL: https://github.com/apache/iceberg/pull/18824
Related to #18713
## Summary
`SparkTable.deleteWhere` ignores `spark.wap.id`. A `DELETE` that Spark
handles as
a metadata delete is therefore committed straight to the current branch
instead
of being staged. This PR makes `deleteWhere` stage the commit and record the
`wap.id`, the same way the other two write paths already do. It touches Spark
3.5, 4.0, 4.1 and 4.2.
## How Spark reaches this code
For `DELETE FROM t WHERE <filter>`, Spark first asks the table whether the
filter
can be applied as a metadata-only delete
(`SupportsDeleteV2.canDeleteWhere`). In
Iceberg that is true when the filter matches whole partitions or whole
files, so
no data has to be rewritten. If it is true, Spark calls
`SparkTable.deleteWhere`.
If not, it plans a row-level write, which goes through
`SparkPositionDeltaWrite`
(merge-on-read) or `SparkWrite` (copy-on-write).
That gives three write paths for a DELETE:
| Path | Class | Handled WAP before this PR? |
|---|---|---|
| Merge-on-read | `SparkPositionDeltaWrite.commitOperation` | Yes |
| Copy-on-write | `SparkWrite.commitOperation` | Yes |
| Metadata delete | `SparkTable.deleteWhere` | **No** |
## Where the bug is
File:
`spark/vX.Y/spark/src/main/java/org/apache/iceberg/spark/source/SparkTable.java`
Method: `deleteWhere(Predicate[] predicates)`
This is the method before the change (Spark 4.1, lines 264 to 291):
```java
public void deleteWhere(Predicate[] predicates) {
Expression deleteExpr = SparkV2Filters.convert(predicates);
if (deleteExpr == Expressions.alwaysFalse()) {
LOG.info("Skipping the delete operation as the condition is always
false");
return;
}
DeleteFiles deleteFiles =
table()
.newDelete()
.set("spark.app.id", spark().sparkContext().applicationId())
.deleteFromRowFilter(deleteExpr);
String writeBranch =
SparkTableUtil.determineWriteBranch(
spark(), table(), branch, CaseInsensitiveStringMap.empty());
if (writeBranch != null) {
deleteFiles.toBranch(writeBranch);
}
if (!CommitMetadata.commitProperties().isEmpty()) {
CommitMetadata.commitProperties().forEach(deleteFiles::set);
}
deleteFiles.commit();
}
```
Nothing in it reads `spark.wap.id` or the `write.wap.enabled` table property.
`deleteFiles.commit()` runs without `stageOnly()`, so
`SnapshotProducer.commit()`
takes the `setBranchSnapshot` branch and moves the branch head.
For comparison, this is what the other two paths do in their
`commitOperation`:
```java
if (wapEnabled && wapId != null) {
operation.set(SnapshotSummary.STAGED_WAP_ID_PROP, wapId);
operation.stageOnly();
}
```
## What goes wrong
With `write.wap.enabled=true` and `SET spark.wap.id = 'x'`:
1. `DELETE FROM t WHERE <whole partition>` runs through `deleteWhere`.
2. The delete is committed to `main` immediately. Readers see it before
anyone
audited or published anything.
3. No snapshot has `summary['wap.id'] = 'x'`, so
`SELECT snapshot_id FROM t.snapshots WHERE summary['wap.id'] = 'x'`
returns
zero rows.
4. `cherrypick_snapshot` has no snapshot ID to work with.
The code never looks at the table format version or at deletion vectors, so
this
also happens on v2 tables.
## The fix
The block below is added in `deleteWhere`, after the branch is resolved and
before `CommitMetadata` is applied (Spark 4.1 and 4.2):
```java
// honor spark.wap.id for metadata deletes so the change is staged rather
than applied to the
// current branch
SparkWriteConf writeConf = new SparkWriteConf(spark(), table());
if (writeConf.wapEnabled() && writeConf.wapId() != null) {
deleteFiles.set(SnapshotSummary.STAGED_WAP_ID_PROP, writeConf.wapId());
deleteFiles.stageOnly();
}
```
On Spark 3.5 and 4.0 the field names differ and `SparkWriteConf` has no
two-argument constructor, so it reads:
```java
SparkWriteConf writeConf =
new SparkWriteConf(sparkSession(), icebergTable, Collections.emptyMap());
```
What each line does:
- `SparkWriteConf` provides `wapEnabled()` (the `write.wap.enabled` table
property) and `wapId()` (the `spark.wap.id` session setting). These are
the same
two values `SparkWrite` and `SparkPositionDeltaWrite` use, so all three
paths
now share one definition of "WAP is on".
- `STAGED_WAP_ID_PROP` writes `wap.id` into the snapshot summary. That is
what
makes the snapshot findable in `.snapshots`.
- `stageOnly()` adds the snapshot to the table metadata without moving the
branch. `main` stays where it was until a cherry-pick.
Imports added to each `SparkTable.java`: `SnapshotSummary`,
`SparkWriteConf`, and
`java.util.Collections` (3.5 and 4.0 only).
When WAP is off or `spark.wap.id` is not set, the new condition is false and
the
method behaves exactly as before.
## Differences between Spark versions
| Version | Table field | Session accessor | `SparkWriteConf` call |
|---|---|---|---|
| 4.1, 4.2 | `table()` | `spark()` | `new SparkWriteConf(spark(), table())` |
| 3.5, 4.0 | `icebergTable` | `sparkSession()` | `new
SparkWriteConf(sparkSession(), icebergTable, Collections.emptyMap())` |
## Files changed
For each of Spark 3.5, 4.0, 4.1 and 4.2:
-
`spark/vX.Y/spark/src/main/java/org/apache/iceberg/spark/source/SparkTable.java`
(the fix)
-
`spark/vX.Y/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestDelete.java`
(two new tests)
## Tests
Two tests are added to `TestDelete` in each Spark version. Both enable
`write.wap.enabled`, set `spark.wap.id`, run a `DELETE`, then check the
table.
- `testMetadataDeleteWithWapIdIsStaged`: partitioned table with one `hr` row
and
one `eng` row. `DELETE ... WHERE dep = 'hr'` removes a whole partition,
which
goes through `deleteWhere`. The test checks that a plain `SELECT` still
returns
both rows and that exactly one snapshot has the matching `wap.id`. Without
the
fix the `hr` row disappears from `main` and no snapshot has the `wap.id`,
so
this test should fail on the old code. I have not run it against the old
code.
- `testRowLevelDeleteWithWapIdIsStaged`: same table setup with two `hr` rows.
`DELETE ... WHERE id = 1` goes through the row-level path. This path
already
staged correctly, so the test is there to catch regressions.
Both tests are skipped when the test is parameterized with a branch, like the
existing WAP branch tests, because `spark.wap.id` only applies to a table
identifier without a branch.
I have not run the full suite yet. <!-- replace this line with what you
actually ran -->
## How to check it by hand
```sql
CREATE TABLE t (id INT, dep STRING) USING iceberg PARTITIONED BY (dep)
TBLPROPERTIES ('write.wap.enabled' = 'true');
INSERT INTO t VALUES (1, 'hr'), (5, 'eng');
SET spark.wap.id = 'wap_001';
DELETE FROM t WHERE dep = 'hr';
SELECT * FROM t; -- both
rows
SELECT snapshot_id FROM t.snapshots WHERE summary['wap.id'] = 'wap_001'; --
one row
```
Before the fix the first query returns only `(5, 'eng')` and the second
returns
nothing.
## What this does not fix
- The INSERT symptom in #18713 (staged rows showing up on `main`). I read
through `SparkWrite` and `SnapshotProducer` and found nothing that would
cause
it, and the issue has no steps to reproduce it. I am not claiming it is
fixed.
- The exact repro in the issue. It uses an unpartitioned table with all three
rows in one file, so I am not sure that `DELETE` takes the metadata path.
This
PR fixes that path, but I could not confirm it is the one the reporter hit.
- Publishing a staged DELETE. `CherryPickOperation` only handles staged
append,
dynamic overwrite and fast-forward snapshots, so a staged delete can only
be
published by fast-forward. That limit is unchanged.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
