mxm opened a new pull request, #16844: URL: https://github.com/apache/iceberg/pull/16844
This is the commit factored out of #15996, after the merged data model and key serialization (#16831). It adds two parallel operators: a Reader turns files into per-row index commands, and a PK index resolves equality deletes against those rows and emits the positions to mark deleted. The planner, DV writer, and committer that complete the graph follow in later PRs. EqualityConvertReader reads each file format-agnostically via FormatModelRegistry and sends a task downstream: a data FileScanTask emits ADD_DATA_ROW or ADD_STAGING_DATA_ROW per row, an EqualityDeleteFileScanTask emits RESOLVE_DELETE per row. It serializes each row's PK with StructLikeSerializer. Rows already covered by an attached DV are skipped. A V2 positional delete on a main data file throws (the target must be V3 DV-only), as does an attached equality delete unless staging and target are the same branch. Any failure goes to both the aggregator ERROR_STREAM and the READER_ABORT_STREAM, so the conversion aborts. EqualityConvertPKIndex maintains a shard of the index and emits a DVPosition per resolved row. Main data is indexed on arrival. A staging snapshot's new rows and every RESOLVE_DELETE instead register an event-time timer at their phase timestamp. Timers fire in timestamp order, so a delete resolves before the cycle's own later-phase staging rows join the index, and a re-inserted key survives a same-cycle delete regardless of the order the parallel readers deliver records. On a shared staging/target branch resolution is sequence-number aware (a delete with sequence S removes only rows below S). On a separate branch the committer reassigns sequences, so every match is deleted and event-time ordering prevents incorrect deletions. Stale-index state is prevented on two levels, both via the main sequence number: a per-key check at the top of processElement clears a key whose stored sequence differs from the incoming command (the keyed and broadcast streams have no mutual ordering), and a CLEAR_INDEX broadcast evicts every key older than it across the subtask, dropping PKs that an external copy-on-write commit removed from main. Metrics: - resolvedDeleteNum - indexedKeyNum - eagerlyEvictedKeyNum. -- 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]
