[python] Implement partial-update merge engine in pypaimon

[TheR1sing3un]

Purpose

pypaimon exposes MergeEngine.PARTIAL_UPDATE in core_options.py and accepts merge-engine: partial-update as a table option, but the read path never reads that option — pypaimon/read/reader/sort_merge_reader.py hardcodes DeduplicateMergeFunction(). So a user who:

1. Creates a PK table with merge-engine: partial-update,
2. Writes one batch with [{id: 1, a: 'A', b: null}], then another with [{id: 1, a: null, b: 'B'}],
3. Reads back,

gets [{id: 1, a: null, b: 'B'}] (silently deduplicated to the latest row) instead of the expected [{id: 1, a: 'A', b: 'B'}] (per-field merge of non-null values). No error, no warning, just wrong data. The same is true today for merge-engine: aggregation and merge-engine: first-row — both are silently degraded to dedupe.

This PR ports the core PartialUpdateMergeFunction semantics from Java (paimon-core/.../mergetree/compact/PartialUpdateMergeFunction.java) and wires the Python read path to dispatch on merge-engine. End state:

• merge-engine: partial-update works correctly for the common case (no DELETE rows, no sequence-group config, no per-field aggregator overrides).
• merge-engine: deduplicate is unchanged.
• merge-engine: aggregation / first-row raise an explicit NotImplementedError instead of silently behaving as dedupe — fail loud now that we have an obvious place for users to escalate from.

Changes

New: pypaimon/read/reader/partial_update_merge_function.py

class PartialUpdateMergeFunction:
    def __init__(self, key_arity, value_arity): ...
    def reset(self): ...
    def add(self, kv):
        if not RowKind.is_add_byte(kv.value_row_kind_byte):
            raise NotImplementedError(...)  # DELETE / UPDATE_BEFORE
        # accumulator[i] = kv.value.get_field(i) if non-null
    def get_result(self) -> Optional[KeyValue]:
        # Build a fresh KeyValue from a fresh tuple — never aliased to
        # upstream's reused input kv.

The reset / add / get_result protocol matches DeduplicateMergeFunction exactly so SortMergeReader doesn't change.

Modified: pypaimon/read/reader/sort_merge_reader.py

SortMergeReaderWithMinHeap.__init__ gains an optional merge_function kwarg (defaults to DeduplicateMergeFunction() so any direct callers are unchanged).

Modified: pypaimon/read/split_read.py

MergeFileSplitRead.section_reader_supplier now picks the merge function based on self.table.options.merge_engine():

DEDUPLICATE     -> DeduplicateMergeFunction()
PARTIAL_UPDATE  -> PartialUpdateMergeFunction(key_arity, value_arity)
AGGREGATE / FIRST_ROW -> NotImplementedError(...)

Out of scope (deliberate, called out in code comments)

• Per-field aggregator overrides (fields.<name>.aggregate-function=...) — needs a FieldAggregator framework.
• Sequence-group support (fields.<name>.sequence-group=...) — needs a UserDefinedSeqComparator-equivalent.
• ignore-delete, partial-update.remove-record-on-delete, partial-update.remove-record-on-sequence-group — depend on the above.
• AGGREGATE / FIRST_ROW merge engines — separate PRs each. This PR just upgrades them from "silent dedupe" to "explicit error".
• DELETE / UPDATE_BEFORE row kinds — raise NotImplementedError so we never silently corrupt data with a half-implemented contract.

Linked issue

N/A — surfaced when verifying that merge-engine: partial-update actually does what the option name implies in pypaimon.

Tests

• pypaimon/tests/test_partial_update_merge_function.py — 11 unit cases driving the merge function with synthetic KeyValue instances. Covers: single insert, two-way overlapping merge (overwrite vs fill-null), three-way merge composition, later-null-does-not-clobber, reset between keys, get_result before any add, UPDATE_AFTER acceptance, DELETE / UPDATE_BEFORE refusal, and result decoupling from input kv (proves the result is built into a fresh tuple so upstream's reused-KeyValue pattern doesn't corrupt us).
• pypaimon/tests/test_partial_update_e2e.py — 8 end-to-end cases on real PK tables. Covers: two-write merge (A,_ + _,B → A,B), three-write left-to-right composition, disjoint keys unaffected, later-non-null wins over earlier non-null, later-null preserves earlier value, deduplicate-engine-unchanged (regression), aggregation / first-row raise NotImplementedError.

Master-vs-fix verification — checked out origin/master's sort_merge_reader.py / split_read.py over the new tests:

$ git show origin/master:.../sort_merge_reader.py > .../sort_merge_reader.py
$ git show origin/master:.../split_read.py > .../split_read.py
$ pytest pypaimon/tests/test_partial_update_e2e.py
... 6 failed, 2 passed
   FAILED test_partial_update_two_writes_merges_non_null
   FAILED test_partial_update_three_writes_merges_left_to_right
   FAILED test_partial_update_later_value_wins_over_earlier_non_null
   FAILED test_partial_update_later_null_does_not_clobber_earlier_value
   FAILED test_aggregation_engine_raises_not_implemented
   FAILED test_first_row_engine_raises_not_implemented
$ git checkout HEAD -- .../sort_merge_reader.py .../split_read.py
$ pytest pypaimon/tests/test_partial_update_e2e.py
... 8 passed

Regression: pytest pypaimon/tests/{reader_primary_key_test,reader_split_generator_test,reader_append_only_test,test_partial_update_*}.py → 50 passed, 7 failed. The 7 failures are all pre-existing lance / vortex environment issues unrelated to this PR; the dedupe path is unchanged.

flake8 --config=dev/cfg.ini clean.

API and format

Public API additions:

• pypaimon.read.reader.partial_update_merge_function.PartialUpdateMergeFunction — new class.
• SortMergeReaderWithMinHeap.__init__ gets an optional merge_function kwarg (back-compat default).

No public API removals or signature breaks. No file format change.

Behaviour change: tables with merge-engine: partial-update now produce per-field merged results (was: silently deduplicated). Tables with merge-engine: aggregation or first-row now raise NotImplementedError (was: silently deduplicated). Both are correctness fixes — the previous behaviour was producing wrong data with no signal to the user.

Documentation

The new module carries a docstring covering the algorithm, the out-of-scope list, and the link back to the Java reference. The dispatch in MergeFileSplitRead._build_merge_function carries an inline comment explaining why we now raise on AGGREGATE / FIRST_ROW.

Generative AI disclosure

Drafted with assistance from an AI coding tool; the algorithm follows org.apache.paimon.mergetree.compact.PartialUpdateMergeFunction and the soundness contract is exercised end-to-end by the regression tests above (which fail on master and pass post-fix).

[XiaoHongbo-Hope]

Could you add coverage for partial-update rows that land in the same data file, e.g. two write_arrow()
calls before a single prepare_commit()? I think this currently still bypasses the new merge-engine dispatch

[XiaoHongbo-Hope]

Nit: thanks for adding the coverage. I wonder if you can make it a bit more focused though — some unit and
e2e cases seem to overlap on the same last-non-null semantics, and the unsupported-option cases could probably be table-driven.

[TheR1sing3un]

Could you add coverage for partial-update rows that land in the same data file, e.g. two write_arrow() calls before a single prepare_commit()? I think this currently still bypasses the new merge-engine dispatch

Thank you for your suggestion. I will pin you again after it's done

[TheR1sing3un]

Nit: thanks for adding the coverage. I wonder if you can make it a bit more focused though — some unit and e2e cases seem to overlap on the same last-non-null semantics, and the unsupported-option cases could probably be table-driven.

Verified -- and the gap is independent of this PR.

I added two coverage cases per your suggestion (commit 0d58859):
test_partial_update_two_write_arrows_single_commit /
test_partial_update_three_write_arrows_single_commit. Both fail, so
they're marked unittest.expectedFailure for now.

To rule out anything this PR introduced, I ran the same workload on
plain origin/master (b4e54ad) using the default deduplicate
merge engine -- two write_arrow calls + one prepare_commit with
the same PK in both batches:

  w.write_arrow(pa.Table.from_pylist([{'id': 1, 'a': 'first',  'b': 'old'}], ...))
  w.write_arrow(pa.Table.from_pylist([{'id': 1, 'a': 'second', 'b': 'new'}], ...))
  c.commit(w.prepare_commit())
  # Expected: 1 row -- {'id': 1, 'a': 'second', 'b': 'new'} (dedupe to latest)
  # Actual on master: 2 rows -- {'id': 1, 'a': 'first', ...}, {'id': 1, 'a': 'second', ...}

The flushed data file contains 2 rows for the same PK; on read,
_build_split_from_pack (split_generator.py:99-100) marks any
single-file PK split as raw_convertible=True, which routes through
the raw-convertible fast path and skips SortMergeReader. Even on
master the dedupe path silently returns both rows -- so what this PR
exposes is a pre-existing user-facing data-quality issue that affects
the default PK-table configuration. It's not a regression introduced
by the partial-update dispatch.

Root cause split across the write and read paths:

• KeyValueDataWriter._merge_data only does concat + sort -- no
  merge function is applied at flush time, so the file holds duplicate
  primary keys (violates the LSM "PK unique within a file" invariant
  that Java enforces in SortBufferWriteBuffer.forEach /
  MergeTreeWriter.flushWriteBuffer).
• The read-side raw_convertible shortcut assumes that invariant
  holds, so it skips SortMergeReader -- where the merge-engine
  dispatch (this PR) and the existing dedupe both live.

Fixing it requires either (a) an in-memory merge buffer in
KeyValueDataWriter mirroring Java's SortBufferWriteBuffer (which
applies the table's MergeFunction during flush -- the proper fix,
also benefits dedupe/aggregation/first-row), or (b) a stricter
raw_convertible check that proves intra-file PK uniqueness. Both
are write-path / scan-path restructuring, well outside the read-side
scope of this PR. I'll open a separate PR for it -- the dedupe data-
quality issue is severe enough on its own to warrant a dedicated fix
rather than buried in this merge-engine port.

The expectedFailure cases here will turn into passing regressions
once that follow-up lands.

[TheR1sing3un]

@XiaoHongbo-Hope Hi, thank you for your point. However, this issue is not strictly related to this pr. It is a general problem. I would rather propose a separate pr to solve it. These two PRS are independent and can be advanced in parallel