[python] Extend commit protocol for compaction (DataIncrement/CompactIncrement)

[TheR1sing3un]

Purpose

Align CommitMessage with Java's CommitMessageImpl shape and add a JSON-safe wire format, so later compaction work has somewhere to plug compact_before / compact_after files and a serializer to ship them through Ray workers.

Foundation only — read / write / commit produce the same snapshots.

Split from #7771.

Changes

• DataIncrement / CompactIncrement value objects; CommitMessage now holds (partition, bucket, total_buckets, data_increment, compact_increment, check_from_snapshot). Convenience properties preserve msg.new_files / msg.compact_before ergonomics.
• FileStoreCommit emits ADD for compact_after, DELETE for compact_before, auto-picks commit_kind=COMPACT when only compact increments are present. New commit_compact() skips row-id assignment.
• DataFileMeta.to_dict / from_dict with tagged encoding for bytes / Decimal / datetime / date / time / Timestamp; encode_value / decode_value public for CommitMessage.partition round-trip.
• CommitMessageSerializer (VERSION=1) covers the full DataIncrement + CompactIncrement shape including IndexFileMeta.

Tests

• commit_message_serializer_test — round-trip with non-JSON-native partition values + index files + version rejection.
• file_store_commit_compact_test — compact_before → DELETE, compact_after → ADD, auto-COMPACT kind.
• Existing file_store_commit_test / partition_predicate_test / table_commit_test adapted to the new CommitMessage signature.

[TheR1sing3un]

Ready for review, 1st pr of the entire compaction feature.

[JingsongLi]

Review: [python] Extend commit protocol for compaction (DataIncrement/CompactIncrement)

Overall this is a well-structured PR that cleanly separates write-side and compaction-side semantics by introducing DataIncrement / CompactIncrement value objects, aligning pypaimon's CommitMessage with Java's CommitMessageImpl. The code is clean, well-documented, and has good test coverage. A few observations:

Correctness

1. encode_value does not handle list / tuple values. If a GenericRow or partition column ever carries an ARRAY-typed value, serialization will raise TypeError. This may be fine for the current scope (partition keys are typically scalar), but worth a brief comment or a graceful error message mentioning "ARRAY" if it's intentionally out of scope.

2. _index_file_to_dict drops dv_ranges and global_index_meta — the comment explains this is deferred to Phase 6/7. Just want to flag: if any code path populates those fields before Phase 6/7 lands, a round-trip through the serializer will silently lose them. The IndexFileMeta.__eq__ only checks the four scalar fields so tests won't catch this. Consider adding an assertion/warning in _index_file_to_dict if those fields are non-None, similar to how _build_commit_entries rejects un-wired increment slots.

Design

3. _build_commit_entries entry ordering. For a message with both new_files and compact_before/compact_after (currently rejected at the API level but structurally possible), the method emits ADD(new_files), then DELETE(compact_before), then ADD(compact_after). The explicit separation of commit() vs commit_compact() makes this a non-issue today, but the ordering assumption should be documented in a comment on _build_commit_entries for future maintainers who may relax the validation.

4. total_buckets fallback logic is good. Using the message's total_buckets (captured at plan time) over the table's current value correctly handles bucket rescale races. The test test_build_entries_uses_message_total_buckets_when_set explicitly covers this, which is great.

5. Convenience properties on CommitMessage (new_files, compact_before, etc.) preserve ergonomics nicely. These delegate cleanly and avoid breaking existing call sites.

Minor / Nits

6. In _generic_row_to_dict, the fallback path [row.get_field(i) for i in range(len(row))] assumes the row implements __len__. A brief type annotation or protocol check (e.g., InternalRow) would improve clarity for readers unfamiliar with the row hierarchy.

7. The CommitMessageSerializer uses separators=(",", ":") for compact JSON output — good choice for a wire format since it minimizes payload size without sacrificing readability when pretty-printed for debugging.

8. DataIncrement.empty_increment() and CompactIncrement.empty_increment() class methods duplicate what DataIncrement() / CompactIncrement() already do with default factories. Unless these are intended as semantic markers for readability in call sites, they could be removed to reduce API surface.

Summary

Solid foundation work. The separation of concerns between write and compaction paths is well-motivated, the serializer is defensively versioned, and the fail-loud approach for un-wired slots (NotImplementedError) is the right call for incremental development. The test suite covers the key round-trip and rejection scenarios thoroughly.

[JingsongLi]

I think the new CommitMessageSerializer drops one existing part of CommitMessage.

CommitMessage has index_deletes, and is_empty() treats it as meaningful content. The global-index update path can build messages that only contain index_deletes (build_index_delete_msgs(...)). But CommitMessageSerializer.to_dict/from_dict only round-trips partition, bucket, total_buckets, data/compact increments, and check_from_snapshot; it does not serialize index_deletes.

If such a message is shipped through this serializer, the deserialized message will lose the index manifest deletions and the commit can leave stale global-index entries behind. Could we include index_deletes in the wire format and add a round-trip test for a message whose only content is index_deletes?

[TheR1sing3un]

I think the new CommitMessageSerializer drops one existing part of CommitMessage.

CommitMessage has index_deletes, and is_empty() treats it as meaningful content. The global-index update path can build messages that only contain index_deletes (build_index_delete_msgs(...)). But CommitMessageSerializer.to_dict/from_dict only round-trips partition, bucket, total_buckets, data/compact increments, and check_from_snapshot; it does not serialize index_deletes.

If such a message is shipped through this serializer, the deserialized message will lose the index manifest deletions and the commit can leave stale global-index entries behind. Could we include index_deletes in the wire format and add a round-trip test for a message whose only content is index_deletes?

Nice catch, fixed~

[TheR1sing3un]

@JingsongLi @XiaoHongbo-Hope Hi, could me help to review it? Hope to land it!

[JingsongLi]

Thank you for your continuous updates, but I still feel concerned about this ability because Compaction is a very inefficient task in Python. I would prefer Compaction to be completed by the Rust layer.

[TheR1sing3un]

Thank you for your continuous updates, but I still feel concerned about this ability because Compaction is a very inefficient task in Python. I would prefer Compaction to be completed by the Rust layer.

Agreed！ Lets close it，thanks for your kindly reviews