feat: support batch incremental diff reads

[Pandas886]

Purpose

Linked issue: close#505

Depends on:#508 and #509 (stacked). Against main you'll see the whole stack; the new work is Diff planning + before/after comparison.

For tables with changelog-producer=none, Diff is how you still get an audit-style change stream without materializing changelog during write.

Brief change log

• Diff planning: before/after snapshot state → DiffPair per (partition, bucket)
• Reject bucket rescale between snapshots (explicit Unsupported, no silent wrong results)
• Read path: PK-sorted merge of before/after images, emit +I / -U / +U / -D, skip equal rows
• Phase-1 limits (fail loud):
  • deduplicate only
  • no deletion-vector support yet
  • nested / decimal types rejected for now
• diff.parallelism option (default 4) for pair execution
• After-image mode on plain incremental read (non-audit) returns the post-change rows only

I used input-delete rows for delete cases in tests so we don't need compact/lookup generation in this PR.

Tests

cargo test -p paimon --test incremental_batch_scan_test
cargo test -p paimon --test audit_log_table_test
cargo fmt --all -- --check
cargo clippy -p paimon --all-targets --features fulltext,vortex -- -D warnings

API and Format

• Diff mode becomes usable on the existing incremental / AuditLog APIs
• No storage format change

Documentation

Still holding docs for the DataFusion PR so users see one end-to-end story.

Known gaps (intentionally)

• Real DV-backed tables: unit-level fail-loud only, no full e2e with DV files yet
• Non-deduplicate engines: hard reject

[JingsongLi]

There are multiple correctness issues:

• When comparing the null Arrow value without checking its validity, 'NULL ->0' will be judged as unchanged; The minimum regression I added actually returns' [] 'and expects' (1,0)'.
• Diff only compares projection columns; When the value changes without projection, it will be missed, and if the projection does not include PK, it will directly report an error. The second minimum regression also returns an empty result.
• Claiming to support types larger than the actual support range of comparator, runtime failures such as timestamp/time/financial; NaN will also be treated as equal.
• Using serialized 'Binary Row' bytes for key range lexicographic comparison and pruning files based on it has unreliable logical order and a risk of data leakage.

[Pandas886]

Thanks, I addressed these in 7b47806 and rebased onto latest main. Diff now compares the full schema before projection (including NULLs), rejects unsupported types, and no longer uses BinaryRow bytes for range pruning. NaN follows Java Float.compare. I also added the NULL/projection regressions. Could you take another look?

[JingsongLi]

Can you add API doc in docs?

[Pandas886]

Added the API guide in docs/src/incremental-reading.md and linked it from the docs navigation. I also tightened the Diff guards and updated the regression tests. Thanks for the reminder!

[JingsongLi]

validate_diff_bucket_layout compares only the partitions that exist in both the before and after snapshots. If the table is empty before and after a bucket rescale, or if the partitions do not overlap at all, the validation will be skipped and the Diff will continue to run, which may produce incorrect results. The added regression test for “different partitions, bucket 1→2” has been confirmed to fail. We recommend validating the bucket configuration directly based on the schemas of the two snapshots, similar to Java’s TimeTravelUtil.

[Pandas886]

Good catch — fixed in 56cf37b. Diff now compares the bucket count from both snapshot schemas before reading manifests, following Java's TimeTravelUtil. I also added an empty-snapshot regression test. Thanks!

[JingsongLi]

+1