koodin9 opened a new pull request, #18003:
URL: https://github.com/apache/iceberg/pull/18003
This is inspired by Flink's `ConvertEqualityDeletes` maintenance action
(#15996) and
follows the same approach: resolve the keys of equality deletes to row
positions and
commit deletion vectors instead, built on the same core pieces
(`BaseDeleteLoader`,
`BaseDVFileWriter`, `RowDelta` validations). The difference is where it
runs. Flink
converts already committed deletes of a staging branch in a separate
topology with a
keyed-state index; the Kafka Connect sink converts the deletes of the
current commit
inside its coordinator, before they are committed, with a stateless pruned
scan.
Part 1 of 4 in a series that adds CDC writes to the Kafka Connect sink
without exposing
equality deletes on the table. Each part is a separate PR, opened after the
previous one
is merged:
- [x] **Part 1 (this PR)**: `EqualityDeleteConverter`, a library class that
resolves
equality deletes to deletion vectors. No callers yet, no behavior
change.
- [ ] **Part 2**: coordinator wiring.
`iceberg.tables.convert-equality-deletes-enabled`,
conversion inside the commit with validation, retries and fallback,
runtime
dependency for Parquet reads, docs.
- [ ] **Part 3**: delta writer. `iceberg.tables.cdc-field` and
`iceberg.tables.upsert-mode-enabled`, inserts as data files, updates
and deletes as
equality deletes (deletion vectors within a batch).
- [ ] **Part 4**: integration tests for partitioned tables, keys split over
tasks,
schema evolution and concurrent writers, plus an opt-in soak test.
### Why
Equality deletes committed by a streaming sink are applied by every reader
to every older
data file until compaction runs, which is what blocked #14797. The sink has
one process,
the coordinator, that sees every delete file of a commit before the commit
exists, so the
equality deletes can be resolved to row positions right there and committed
as deletion
vectors instead. No branch, no index state, no maintenance job, and the main
branch never
carries an equality delete. This PR adds the class that does the resolving;
the
coordinator that calls it is the next PR.
```mermaid
flowchart LR
W["Worker<br/>delta writer (PR 3)"] -->|"data files +<br/>equality
delete files"| C["Coordinator (PR 2)"]
C -->|"equality delete files"| X["EqualityDeleteConverter<br/>(this PR)"]
X -->|"Result: deletion vectors,<br/>rewritten vectors, base
snapshot,<br/>conflict filter"| C
C -->|"RowDelta: data files + vectors"| T[("Iceberg table, v3<br/>no
equality deletes")]
```
### What this PR adds
| Class | Role |
|---|---|
| `EqualityDeleteConverter` (public) | Entry point. Groups the delete files,
loads their keys, asks a resolver for positions, writes the vectors, returns a
`Result`. |
| `KeyPositionResolver` (interface) | "Given a set of deleted keys, which
rows of the base snapshot hold them?" Returns matches per data file, including
the vector already attached to that file. |
| `ScanKeyPositionResolver` (default) | Answers by planning the snapshot
with pruning filters and reading only the key columns and `_pos` of the
candidate files. Keeps no state. |
| `KeyFilters` | Builds the pruning filters: partition pin through
transforms, `IN` lists or value ranges within a comparison budget. |
```mermaid
classDiagram
class EqualityDeleteConverter {
+EqualityDeleteConverter(Table, String branch)
+convert(List~DeleteFile~ eqDeleteFiles) Result
}
class Result {
+baseSnapshotId() Long
+dvFiles() List~DeleteFile~
+rewrittenDvFiles() List~DeleteFile~
+conflictFilter() Expression
}
class KeyPositionResolver {
<<interface>>
+resolve(Snapshot base, Schema keySchema, StructLikeSet keys,
PartitionSpec deleteSpec, StructLike deletePartition) Resolution
}
class FileMatches {
<<interface>>
+path() String
+spec() PartitionSpec
+partition() StructLike
+existingDeletes() PositionDeleteIndex
+forEachPosition(LongConsumer)
}
class ScanKeyPositionResolver
class KeyFilters {
+partitionFilter(Schema, PartitionSpec, StructLike) Expression
+comparisonsPerFile(Snapshot) int
+keyFilter(Schema keySchema, StructLikeSet keys, int
comparisonsPerFile) Expression
}
EqualityDeleteConverter --> KeyPositionResolver : resolve()
EqualityDeleteConverter --> Result : returns
EqualityDeleteConverter ..> KeyFilters : conflict filter
KeyPositionResolver --> FileMatches : returns per data file
ScanKeyPositionResolver ..|> KeyPositionResolver
ScanKeyPositionResolver ..> KeyFilters : plan + read filters
```
### How a conversion runs
```mermaid
sequenceDiagram
participant Caller
participant Conv as EqualityDeleteConverter
participant Loader as BaseDeleteLoader (core)
participant Res as KeyPositionResolver
participant DV as BaseDVFileWriter (core)
Caller->>Conv: convert(eqDeleteFiles)
Conv->>Conv: base = snapshot of branch (none: Result.empty())
Conv->>Conv: group files by (equality field ids, spec, partition)
loop each group
Conv->>Loader: loadEqualityDeletes(files, keySchema)
Loader-->>Conv: StructLikeSet keys
Conv->>Conv: conflictFilter OR= partitionFilter AND keyFilter
Conv->>Res: resolve(base, keySchema, keys, spec, partition)
Res-->>Conv: FileMatches per data file
Conv->>DV: delete(path, pos, spec, partition) for every position
end
Conv->>DV: close()
Note over DV: merges the vector a data file already has,<br/>writes one
Puffin file
DV-->>Conv: DeleteWriteResult (new vectors, rewritten vectors)
Conv-->>Caller: Result(baseSnapshotId, dvFiles, rewrittenDvFiles,
conflictFilter)
```
The `Result` maps one-to-one onto the `RowDelta` the next PR will build:
| `Result` | `RowDelta` |
|---|---|
| `baseSnapshotId()` | `validateFromSnapshot(id)` (also required to replace
a vector) |
| `dvFiles()` | `addDeletes(dv)` |
| `rewrittenDvFiles()` | `removeDeletes(dv)` (a data file may have only one
vector) |
| `conflictFilter()` | `conflictDetectionFilter(expr)` +
`validateNoConflictingDataFiles()` / `validateNoConflictingDeleteFiles()` |
### How positions are found (`ScanKeyPositionResolver`)
Four stages narrow the work. The first three only decide which files and row
groups are
opened; the last one decides correctness. Stages 1 and 2 happen inside
`planFiles`, stage 3
inside the file reader, stage 4 in the resolver's own loop.
```mermaid
flowchart TD
K["deleted keys of one group"] --> F["KeyFilters<br/>planFilter =
partitionFilter AND keyFilter"]
F --> P1["1. manifest level (planFiles)<br/>partition summaries of each
manifest<br/>vs. the partition pin"]
P1 --> P2["2. file level (planFiles)<br/>partition value and column
bounds<br/>of each data file entry vs. planFilter"]
P2 -->|"candidate FileScanTasks, read in parallel"| P3["3. row group
level (file reader)<br/>project key columns + _pos,<br/>filter(keyFilter) skips
row groups by their statistics"]
P3 --> P4["4. row level (resolver)<br/>skip positions the file's
existing deletion vector covers,<br/>then keys.contains(row key): exact"]
P4 --> O["FileMatches: path, spec, partition,<br/>existing vector,
positions"]
```
Only `keyFilter` is passed to the file readers: Parquet row-group filters
evaluate column
references, not partition transforms. Data files that carry v2 position
delete files are
rejected with an `IllegalStateException` asking to rewrite them into
deletion vectors
first, because a data file can have only one vector and no position delete
files next to
it. Equality delete files already attached to a candidate are left alone;
their rows may
land in the new vector too, which is harmless.
### How the filters are built (`KeyFilters`)
The filters are inclusive: a file that holds a deleted key always stays a
candidate, a file
without one may. Their size is bounded because the plan filter is evaluated
once per data
file entry.
```mermaid
flowchart TD
S["snapshot summary<br/>total-data-files = F"] --> C["budget C =
200,000,000 / F<br/>clamped to [200, 1000] comparisons per file<br/>(missing
count: 200)"]
C --> D{"keys x primitive key columns <= C ?"}
D -->|yes| L["IN lists<br/>chunks of 200 values (evaluator limit), one
list per key column,<br/>chunks OR-ed: exact, only files holding a key"]
D -->|no| G["value ranges on the leading key column<br/>at most C / 2
ranges, split at the largest gaps<br/>(integers: only gaps that skip a
value):<br/>inclusive, may keep files without a key"]
PF["partitionFilter<br/>every partition field pinned through its
transform,<br/>e.g. bucket(4, id) = 3; void / unknown transforms skipped"] -->
A["planFilter = partitionFilter AND keyFilter"]
L --> A
G --> A
```
Why 200,000,000: a fixed cap of 100 ranges would cost 200 comparisons per
file, or 200M
on a table with one million data files. The budget keeps that cost for such
tables and
lets smaller tables use a finer filter. On a 2,000-file benchmark table,
2,000 deleted
keys of which 10% were scattered opened 1,109 files with a fixed cap of 100
ranges and
202 with the budget, the files that actually hold a key.
### Notes for reviewers
- `KeyPositionResolver` is an interface so an index-backed resolver can
replace the scan
once the secondary index spec (#16961) lands; the constructor that accepts
one is
package-private for now.
- `KeyFilters.IN_PREDICATE_LIMIT` (200) mirrors the private constant in
`InclusiveEvalVisitor`; above it an `IN` list is not evaluated against
file bounds.
- Format version 3 only. The class keeps no state between calls;
`scannedDataFiles()` is
a test hook reporting the files read by the last conversion.
### Testing
- `TestEqualityDeleteConverter` (11), against a real v3 table
(`InMemoryCatalog`), read
back with `IcebergGenerics`: deletes of earlier commits become vectors and
a second
conversion merges the existing vector and reports it as rewritten;
partition pruning;
identity and `bucket` partitions pinned with more keys than one `IN` list
holds
(12 files → 3); clustered plus scattered keys open exactly the files that
hold them
(200 files → 11, and → 156); a custom resolver; conversion on a branch;
rejection of
non-equality delete files.
- `TestKeyFilters` (4), against file statistics with
`InclusiveMetricsEvaluator`: the
budget for every `total-data-files` case; `IN` lists exclude exactly the
files without
keys; ranges above the budget keep some files without keys (documented);
every key
column counts towards the budget.
- `./gradlew -DkafkaVersions=3
:iceberg-kafka-connect:iceberg-kafka-connect:check` passes
(148 tests, checkstyle, spotless).
--
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]
