[spark] Add point-delete fast path for primary key DELETE to skip the target table scan

[ZZZxDong]

Purpose

Closes #8836.

Deleting a set of primary keys from a large primary-key table currently costs a full table scan.

Concrete case that motivated this: a ~10-billion-row primary-key table and a ~10-million-row key table, DELETE FROM big WHERE (pk) IN (SELECT pk FROM keys). Today the IN subquery is rewritten into a left-semi join, so the whole target table is read (all columns, because the -D rows are copied from the matched rows) just to find the 10M rows to delete. Runtime filtering does not help either: PaimonSupportsRuntimeFiltering only pushes down partition-level predicates, and the keys are spread over every partition and bucket.

Where this fits in the existing DELETE paths on a primary-key table:

Condition	Today
partition columns only (WHERE dt = 'p1')	OptimizeMetadataOnlyDeleteFromPaimonTable -> metadata-only truncate, no scan
all primary key columns pinned	full scan <- this PR
anything else	scan-based COW rewrite / deletion vectors

When the condition pins every primary key column, the matched keys are already fully described by the condition itself, so the -D records can be built without reading the target table at all:

Condition	Fast path
pk = literal / pk IN (literals) covering all pk columns	build -D rows from the condition on the driver, capped by delete.point-delete.max-rows (default 1M, falls back to scan beyond that)
pk IN (subquery) covering all pk columns	use the subquery result as the key DataFrame, fully distributed - this is the case that matters for a large key table
anything else	existing scan-based path, unchanged

Keys that do not exist in the table only add a -D record that compaction removes later (note it can materialize a partition that did not exist before).

Writing -D rows that carry only the primary key is not a new semantics in Paimon: with rowkind.field a user can already INSERT rows holding just the primary keys plus '-D' and get exactly the same records, without touching the target table. This PR turns that into an automatic optimization for DELETE, instead of requiring a schema change and hand-written SQL.

Safety

The -D rows carry NULL for non-key columns, so the fast path is opt-in (delete.point-delete.enabled, default false) and additionally falls back to the scan path unless all of the following hold:

• merge-engine = deduplicate: the merge replaces the whole row and never reads field values of a delete row. partial-update (remove-record-on-sequence-group) and aggregation do read them.
• no sequence.field: the user sequence fields are part of the write buffer sort key (SortBufferWriteBuffer) and NULL sorts smallest, i.e. oldest, so a NULL-filled -D row would lose the merge against the existing row and the delete would be silently dropped.
• no NOT NULL non-key column: TableWriteImpl.checkNullability rejects the write.
• no primary key index: those are built from real field values, so a NULL-filled -D row would leave the deletion invisible to index lookups.
• no cross-partition update (pk not covering the partition keys): the -D row would carry a NULL partition.

Two more cases fall back inside the subquery path:

• correlated subqueries (the plan cannot be executed on its own),
• NULL keys produced by the subquery, and NULL literals in the condition - NULL never matches under SQL three-valued logic, so no -D row may be written (otherwise a NULL primary key would be stored).

Known trade-off, documented on the option: until compaction merges them away, incremental / audit_log reads see NULL instead of the old field values of the deleted rows. For the same reason the option should not be enabled on tables whose changelog is consumed by streaming jobs relying on the field values of retract records.

Tests

DeletePointFastPathTest, 18 cases. Besides asserting the query results, the suite asserts which path ran: deleting a key that does not exist writes exactly one -D record on the fast path and nothing at all on the scan path, so a fallback case cannot pass by accident.

• fast path: literal equality, literal IN, composite pk (partition + id), subquery with single and composite pk (including positionally renamed columns), delete-then-reinsert, "really skips the scan"
• fallback: non-pk column in the condition, pk not fully pinned, NULL literal (id = NULL, id IN (1, NULL)), NULL keys from a subquery, correlated subquery, sequence.field, NOT NULL non-key column, cross-partition table, partial-update with sequence-group, over max-rows, option disabled (default)

The existing DeleteFromTableTest suite (29 tests) passes unchanged.

API and Format

Two new Spark connector options, no format change:

• delete.point-delete.enabled (default false)
• delete.point-delete.max-rows (default 1000000)

Documentation

docs/generated/spark_connector_configuration.html regenerated for the two options; the enabled description spells out when the fast path applies and the incremental-read caveat.

Follow-up (not in this PR)

The same idea generalizes one level further: for any DELETE condition, project only the primary key columns plus sequence.field instead of all columns before writing the -D rows. That keeps the scan but cuts the IO to a few columns, works for sequence.field tables, and covers conditions that do not pin the primary keys. Happy to send that as a separate PR if the direction looks good.

[JingsongLi]

• For fast paths, only fill in the primary key (PK); set all non-PK fields to NULL; however, the validation also allows partial updates and aggregates.
• Non-null, non-PK fields will cause an error when the merge function initializes the DELETE row.
• partial-update.remove-record-on-sequence-group also requires values for the sequence-group field; leaving all fields blank will not preserve the original delete semantics.
• Additionally, the literal extraction does not exclude NULL values, posing a risk of generating delete rows with empty primary keys.

[ZZZxDong]

Good catch, all four points are valid — fixed. The fast path is now gated to DEDUPLICATE tables without NOT NULL non-pk columns (anything else falls back to the scan path), and NULL literals in the condition also fall back so SQL three-valued logic is preserved. Added tests for each case, and rebased onto master to resolve the conflict.

[ZZZxDong]

CI caught another case of the same pattern: tables with primary-key indexes (btree/bitmap etc.) build index entries from real field values, so NULL-filled -D rows would leave the deletion invisible to index lookups. Added a third gate — tables with any pk index definition also fall back to the scan path. All tests green now.

[ZZZxDong]

The remaining CI failure surfaced a real semantic difference: -D rows from the fast path carry NULL for non-key columns, so incremental / audit_log reads would not see the old field values of deleted rows (the TVF test asserts [-D, 1, 11] but got [-D, 1, null]). Since this affects the changelog contract for any deduplicate table, I made the fast path opt-in: it is now off by default and enabled via spark.paimon.delete.point-delete.enabled, with the trade-off documented on the option. Default behavior is completely unchanged. Happy to adjust if you prefer a different approach.

[JingsongLi]

Why not just introducing this by default just like Flink SQL?