[JingsongLi]

Purpose

Follow-up to #1255 and the remaining incremental merge-engine work from #1251. Extend incremental primary-key compaction to non-DV first-row and stateful lookup tables, plus plain partial-update/aggregation configurations whose states can safely be regrouped. Small writes can be compacted while large historical runs stay in place.

Brief change log

• Reuse the existing universal size/run policy. First-row supports none and lookup; partial-update and aggregation support lookup, and a conservative subset of none/input configurations.
• Reuse the actual catalog/FileIO Lookup File cache for all lookup engines. First-row requires it with either producer. A first-row lookup hit remains in retained history and emits neither data nor changelog; selected positive-level rows still participate in rewritten output. Validate inputs before applying this shortcut.
• Follow Java KeyValue.schemaWithLevel: append the non-null INT _LEVEL to internal KV input batches, using Java's field ID. Read each row's level directly; first-row distinguishes retained lookup hits by levels above the task's output level. Remove the private file-level and lookup-base batch metadata markers.
• Stateful lookup inserts the nearest retained state in sequence order and materializes the result. Existing lookup queries use only the nearest positive-level state, so older retained states are not aggregated twice. Preserve sparse columns, before/after images, first-add metadata and delete/reset state through the existing reducers.
• Plain none/input merges only the selected states, leaving history to ordinary query merging. Enable incremental selection only when all value fields use min, max, bool_and, bool_or, or integer sum without checked overflow. Both engines must have no global sequence fields, and aggregation must not reset on DELETE. Partial update must ignore deletes and aggregate all value fields protected by sequence groups; ordering fields retain each group's winning sequence and cannot also be protected fields. These plain modes need no local-cache.
• Preserve the previous conservative selection for other plain configurations, including sparse overwrite, first-value aggregation, floating sums and checked overflow. Keep existing retract/state restrictions. Allow first-row to retire L0 with empty data/changelog outputs when all keys already exist.

Tests

Thirteen new integration regressions plus strengthened existing cases cover fixed/dynamic buckets, both first-row producers, stateful lookup and plain safe configurations, out-of-order global/group sequences, selected positive runs, positive-only maintenance, retained history, old snapshots, cold/warm generated-file consumption with the source object removed, unchanged before/after metadata, empty output, omitted task output-level metadata, concurrent L0 and idempotent retries. Two unit regressions cover mixed levels within one batch and the trailing _LEVEL field's type, ID and preservation across concatenation and IPC serialization.

Plain tests also run with actual local-cache disabled. Counterexamples protect first-value ordering, sparse per-column overwrite, finite floating cancellation, checked overflow and reset-on-delete. Regressions verify that global sequence fields cleared by DELETE/UPDATE_BEFORE remain NULL and that intersecting ordering/protected sequence groups cannot lose an aggregate when replayed; both counterexamples failed before the safety conditions were tightened. A lookup DELETE/reinsert case preserves tombstones and their payload. Mutation checks fail as expected when first-row copies lookup hits, drops selected positive rows, the plain safety check is bypassed (20 instead of 10), or stateful lookup caching is disabled; all mutations were restored.

4,676 library unit tests and 206 related integration tests passed with --test-threads=4; six existing library tests were ignored. All 83 PK compaction tests passed. cargo fmt --all -- --check, git diff --check, and cargo clippy --locked -p paimon --lib --test pk_compact_test -- -D warnings -A clippy::nonminimal_bool passed. The allowance covers existing repository findings with the local toolchain.

Compared with Java source at Apache Paimon commit 5be7e2dd1f: LookupStrategy, FirstRowMergeFunctionWrapper, LookupMergeFunction, LookupChangelogMergeFunctionWrapper, AggregateMergeFunction and PartialUpdateMergeFunction. Plain selection remains conservative where regrouping would change the existing read result.

API and Format

No new public API, result fields, data format or Lookup File format. First-row now requires actual FileIO local-cache even with changelog-producer=none; lookup tables continue to require it. Newly enabled incremental tasks carry output_level; older executable tasks without it must be replanned. Prepared results with matching file changes can omit task output-level metadata and recover idempotently, including empty first-row results. No task transport, whole-history fence or full-compaction API is added.

Documentation

Updated the compaction guide with per-engine selection, local-cache requirements, retained/materialized state behavior, first-row empty outputs and the plain-state safety conditions.

[leaves12138]

LGTM