[Feature] Support managed BLOBs in primary-key tables and normal-file compaction across data-evolution field groups

[SteNicholas]

Search before asking

• I searched in the issues and found nothing similar.

Motivation

This tracks the roadmap item "Extend Data Evolution to primary-key tables and support compaction across evolved field groups" in #186.

One clarification on the wording first: neither Java nor C++ allows a row-tracking/data-evolution table to define primary keys (SchemaValidation.validateRowTracking on both sides). What the roadmap item maps to in Java is two capabilities:

1. Managed BLOBs in primary-key tables — [core][flink][spark] Support managed BLOBs in primary-key tables (apache/paimon#8617), extended by [core] Allow first-row merge engine for primary-key managed BLOB tables (apache/paimon#9201). BLOB payloads are externalized to shared .managed.blob pack files before entering the MergeTree write buffer, data files store small descriptors plus a .blobref reference sidecar, and compaction rewrites descriptors verbatim while rebuilding the exact per-data-file reference set.
2. Compaction across data-evolution field groups — [core] Introducer compaction for data-evolution table (apache/paimon#6828), with the planning rules of [core] Optimize data evolution compaction planning (apache/paimon#9177). Files sharing the same row-id range form one evolved field group; compaction merges the field groups of a contiguous row-id run into a single normal file without changing any row id.

paimon-cpp has neither today:

• A primary-key table with a BLOB column cannot even be created: SchemaValidation::ValidateRowTracking requires data-evolution.enabled for BLOB columns, which in turn forbids primary keys — an unsatisfiable pair.
• A data-evolution table is never compacted: auto compaction cannot run on bucket = -1, and AppendCompactCoordinator::Run only implements the plain append rewrite, which would reorder rows and break row ids.

Solution

Port the two Java capabilities:

1. Managed BLOB storage for primary-key tables (apache/paimon#8617 + #9201)

• PrimaryKeyBlobExternalizer: externalize non-null blob values of insert rows into rolling .managed.blob packs (sealed by blob.target-file-size, copied through a blob.copy-buffer-size buffer) before they enter the write buffer; retract rows drop the payload. Uncommitted packs are deleted on abort/close; PrepareCommit seals and hands them over.
• ManagedBlobReferenceFile: the .blobref sidecar of each data file, byte-compatible with Java (magic/version/count/writeUTF/CRC32, sorted and deduplicated), plus ManagedBlobReferenceCollector on the key-value write path so a compacted file lists exactly the packs its surviving rows still reference.
• Lifecycle: the sidecar travels in the data file's extra files and dies with it (writer abort, uncommitted-file cleanup, snapshot expiration); pack files are shared and never deleted by table maintenance (orphan-files clean skips .managed.blob).
• Read path: ManagedBlobResolvingBatchReader resolves descriptors to payload bytes with one ranged read per surviving value after merging; blob-as-descriptor returns the serialized descriptors instead.
• Schema validation: only deduplicate / partial-update / first-row merge engines, changelog-producer must stay none, no data-file.external-paths (rejected on the raw option, even an empty string), managed blob fields cannot be primary/bucket/sequence keys or order a sequence group, and sequence-group-protected managed blob fields reject aggregate functions that need the retracted payload.
• Postpone-bucket writers externalize the same way, so bucket = -2 tables hold descriptors too.

2. Compaction across data-evolution field groups (apache/paimon#6828 → #9177 planning rules)

• DataEvolutionCompactCoordinator: group files by exact row-id range into evolved field groups; bin weight is sum(max(file_size, source.split.open-file-cost)); a bin becomes a task once its weight strictly exceeds target-file-size; a heavier-than-target group is compacted alone; a row-id gap always cuts the bin; a bin needs at least compaction.min.file-num files.
• DataEvolutionNormalCompactTask: read the group through DataEvolutionSplitRead (newest field group wins per column), rewrite into exactly one output file, keep the input's first row id and the merged [min, max] sequence-number range so _ROW_ID stays stable.
• Entry point: AppendCompactCoordinator::Run plans with the data-evolution coordinator when data-evolution.enabled is set (scanning every live file instead of only small ones), and rejects the removed legacy mode data-evolution.compaction.rewrite-row-ids=true like Java does.

Anything else?

Known scope cuts against current Java, disclosed in the user docs (primary_key_table.rst, compaction.rst):

• Only top-level scalar BLOB columns are managed; ARRAY<BLOB> / MAP<K, BLOB> and blob-descriptor.source-table (source-table FileIO credentials) are not supported yet.
• Compaction does not yet include: deletion-vector rewrite/materialization (tables with deletion-vectors.enabled keep being rejected, see [Feature] Support deletion vectors for data-evolution tables #169 for the read side), the projected-manifest candidate planning with ~100k-file batches and multi-round commits of [core] Optimize data evolution compaction planning paimon#9177, dedicated BLOB pack compaction, and vector-store file planning (tables holding vector-store files are rejected until a VECTOR type lands).

Follow-up regression tests that need the runtime to pin expectations: concurrent partial-update/append commits racing a planned compaction (Java DataEvolutionTableTest), and compaction across real schema ids after ALTER TABLE.

Are you willing to submit a PR?

• I'm willing to submit a PR!