[core][spark][flink] Support sub-field-level data evolution for nested columns

[zhuxiangyi]

Motivation

Local, high-frequency updates on a wide nested struct are expensive under today's data evolution: because the smallest evolvable unit is a top-level column, changing one sub-field (nest.a) rewrites the entire nest column — including all the unchanged sub-fields — into a new column-group file. This causes significant write amplification and storage waste exactly in the workloads that update structs most often. This PR lowers the column-group granularity to the leaf field, so "update one sub-field" only incrementally writes that sub-field, eliminating this class of write amplification at the root.

Purpose

This PR pushes column-group granularity down to the leaf field: updating a single sub-field writes an incremental file containing only that leaf (a dotted write column like nest.a), aligned by row id; on read the sub-fields scattered across files are reassembled into the full struct.

Use cases — "wide nested struct + frequent local updates":

1. Local update of a user/entity profile. A row holds a wide profile STRUCT<age, city, tags, last_login, score, ...>, but each operation only updates one or two sub-fields (login updates last_login, risk-control updates score).

   • Without this: every update rewrites the whole profile (dozens of unchanged sub-fields).
   • With this: only a profile.last_login incremental file is written, aligned by row id; the full profile is reassembled on read.
   • Benefit: write amplification drops sharply, especially for wide structs.

2. Different pipelines/teams own different sub-fields of one struct. Pipeline A owns nest.a, pipeline B owns nest.b.

   • Each only incrementally writes its own part without rewriting the other's, and the full struct is merged back by row id on read.
   • Fits wide tables where a row is assembled by multiple owners.

Gated by a new table option data-evolution.nested-field.enabled (default false); when disabled the behavior is identical to before (whole-column rewrite). Engine entries: Spark MERGE INTO and Flink data_evolution_merge_into action.

Design (high level)

• Encode writeCols as dotted paths (nest.a) instead of only top-level names — no DataFileMeta serialization change. New RowType.projectByPaths / leafPaths convert between a (partial) nested type and its dotted paths, preserving field ids.
• Write: a partial-struct write records its real sub-field content as dotted writeCols.
• Read (DataEvolutionSplitRead): match files at leaf field-id granularity and assemble a struct split across files sub-field by sub-field (latest-wins per leaf). DataEvolutionRow composes the struct from several source files.
• Spark (MergeIntoPaimonDataEvolutionTable): prune the aligned update to only the changed leaves; fall back to whole-column write when not safely determinable.
• Flink (DataEvolutionMergeIntoAction): parse dotted SET targets, rebuild a partial struct as CAST(ROW(...) AS ROW<...>), and write via projectByPaths. Reuses the existing top-level pipeline (row-id assign / shuffle / partial-write operator / commit).
• Compaction works through the merged read unchanged.

Tests

• core: NestedDataEvolutionTableTest (5), NestedSubfieldDataEvolutionTableTest (3) — sub-field groups assembled, late overwrite, projection, compaction merges sub-fields.
• spark: NestedSubfieldMergeIntoTest — single sub-field incremental write, whole-struct write, flag-off fallback.
• flink: NestedSubfieldMergeIntoActionITCase (5) — single/multiple sub-fields (asserting dotted writeCols), whole-struct, flag-off rejection, deeper-than-one-level rejection.

API and Format

• New table option: data-evolution.nested-field.enabled (Boolean, default false).
• No change to DataFileMeta / manifest format — writeCols semantics extended (a dotted entry means a written sub-field; a plain entry still means the whole column). Backward compatible with existing files.

Documentation

• Regenerated docs/generated/core_configuration.html for the new option.

Limitations (follow-ups)

• Cross-file struct assembly supports one level of ROW only; deeper splits are rejected (or fall back to whole-column write).
• Global index on nested sub-fields is out of scope.
• Predicate stats are skipped for partially-written nested struct files (correctness-safe; loses file skipping).
• Columnar fast-path and escaping for column names containing . are follow-ups.

[JingsongLi]

This override drops the @BeforeEach annotation from ActionITCaseBase.before(), so JUnit never runs the setup for this class. As a result warehouse/catalog are not initialized and ReadWriteTableTestUtil.init(warehouse) is not called; the new test class currently fails all five tests with NPE at the first sEnv.executeSql(...). Please add @BeforeEach here (as the other action ITs do) so both the base setup and init(warehouse) run before each test.

[zhuxiangyi]

Good catch, thanks! You're right — overriding before() without re-adding @BeforeEach means JUnit never runs the base setup, so warehouse/init(warehouse) were uninitialized. Fixed in 23766fd by adding @BeforeEach to the override.

[JingsongLi]

After adding the missing @BeforeEach locally to let this test class initialize, this DDL still fails before reaching the assertion: Flink's parser treats inner as a keyword (SQL parse failed. Encountered "inner" at line 1, column 41). Please quote the nested field name (and the matching CAST(ROW(... ) AS ROW<...>) below) or use a non-keyword name, otherwise testUpdateDeeplyNestedSubFieldThrows cannot exercise the intended deeper-than-one-level validation.

[zhuxiangyi]

Thanks! inner collides with the Flink SQL reserved word and breaks DDL parsing. Renamed the nested sub-field inner → sub (in the DDL, the CAST(ROW(...)) and the SET target) in 23766fd, so testUpdateDeeplyNestedSubFieldThrows now reaches and exercises the deeper-than-one-level validation.

[zhuxiangyi]

Thanks for the review @JingsongLi! Addressed both points in 23766fd:

• Added @BeforeEach to the before() override (tests were NPE-ing without base setup).
• Renamed the nested field inner → sub to avoid the Flink SQL reserved word.

Also fixed the spotless-check failure (the spark-ut test wasn't formatted). CI is re-running.

[JingsongLi]

This relaxation should be scoped to sub-field writes. Today it also accepts whole-column assignments, e.g. --matched_update_set T.nest=S.nest where the source S.nest is ROW<a> and the target is ROW<a,b>. partialStructCompatible returns true here, but writePaths is still just nest, so sourceType is built as the full target struct and the partial RowData is sent to a whole-struct write. That can fail at runtime or create an incomplete whole-struct file. Please keep whole-struct assignments on the normal full-type compatibility check, and only allow this subset check when the column is actually being written through dotted paths such as nest.a.

[zhuxiangyi]

Agreed — fixed in ef144ac. The partial-struct check is now gated on isSubFieldWrite(column) (i.e. the column actually has dotted write paths like nest.a). Whole-column assignments such as T.nest=S.nest stay on the full-type compatibility check, so a narrower source struct is rejected instead of being written as an incomplete whole-struct file.

[JingsongLi]

Could we reject dotted paths when the selected head is not a ROW? With the current branch, projectByPaths(Collections.singletonList("id.a")) falls into this arm and returns the whole id field. That makes invalid dotted writeCols look valid to callers such as the conflict checker, and in the Flink action an invalid SET target under a scalar can pass path resolution before failing later with a less helpful error. Since dotted paths now encode physical sub-fields, this should throw unless the head field is a ROW, or the whole path matched an exact top-level field name.

[zhuxiangyi]

Good point — fixed in ef144ac. projectByPaths now throws IllegalArgumentException when a dotted path's head field is not a ROW (e.g. id.a), instead of silently returning the whole id. Exact top-level matches (including column names that themselves contain a dot) are still selected whole. Added coverage in DataTypesTest#testProjectByPaths.

[steFaiz]

Thanks for your contribution!

This class has already been marked as

TODO: Optimize implementation of this class.

I think current createUnionReader is already hard to comprehend, the modified single method have 300 rows and many complicated logic. Is there any way to extract a dedicated class for this nested-data-evolution scenario?

[zhuxiangyi]

Thanks for the review! Agreed — the nested-data-evolution assembly is what grew createUnionReader.

Plan: extract the planning logic (leaf-level matching + the tree-shaped assembly plan, the current Steps 1–4 plus the collectLeafIds/providerOf/findSubProvider helpers) into a dedicated, pure DataEvolutionReadPlanner that returns an immutable plan (rowOffsets/fieldOffsets/NestedField[] + the per-bunch read fields). createUnionReader then just resolves the bunch schemas and builds the readers from that plan, so it goes back to a thin shell. A nice side effect is that the planning logic becomes directly unit-testable instead of only through ITs.

For the broader pre-existing TODO: Optimize implementation of this class (the top-level read path, mergeRangesAndSort, etc.), I'd suggest keeping that as a separate follow-up PR so this one stays focused on the nested feature — and I'd be happy to take part in that optimization PR as well. Does this approach sound good to you?

[JingsongLi]

This can persist writeCols such as nest.sub.x for a deeper partial struct, but the read path below only supports composing one nested level and later throws when the full row is read (DataEvolutionSplitRead rejects partially-written nested sub-fields deeper than one level). That means a caller using BatchTableWrite.withWriteType(table.rowType().projectByPaths(Collections.singletonList("nest.sub.x"))) can successfully commit a file that makes normal full-table reads fail afterwards. Please reject unsupported deeper dotted paths before writing/committing them, or extend the reader to compose them recursively.

[zhuxiangyi]

Fixed in 7211c70. RowType.leafPaths now fails fast (UnsupportedOperationException) when a partial struct is nested inside another partial struct (a path deeper than one level, e.g. nest.sub.x), so withWriteType rejects it before any such file can be written/committed — a low-level BatchTableWrite.withWriteType(projectByPaths(["nest.sub.x"])) now throws up front instead of committing a file that later breaks full-table reads. One-level partial writes (nest.a, or a whole sub-struct nest.sub under a partial nest) are unaffected. Added DataTypesTest#testLeafPaths coverage.

[JingsongLi]

Please resolve conflicts.

[JingsongLi]

@zhuxiangyi This is indeed a very significant change. Can you describe in detail why your business cannot use top-level fields?

[zhuxiangyi]

@JingsongLi Thanks for the review. This is indeed a significant change, so let me describe our real use case in detail — and I'd love to hear your suggestions.

Background. We have a wide feature/data-source cache table for our risk engine. The modeling groups fields of the same kind (one data-source response / one feature family) into a single struct. Roughly: ~276 top-level columns ≈ 267 structs + 9 scalars; the number of sub-fields per struct ranges from a few up to ~2599; flattening everything into top-level columns would be ~22k columns.

Why we keep it nested instead of flattening. At this scale, ~22k top-level columns become hard to work with for us — the schema is serialized into every snapshot/manifest, columnar footer & per-column stats metadata grow (especially painful for the small incremental files data evolution produces), and engine planning/codegen cost rises noticeably; day-to-day schema evolution also gets unwieldy. Modeling "one data source = one struct" lets us manage a source as a unit and prune by group on read, which fits us better. If there's a better modeling approach here, I'm very open to it.

Read pattern. This table is only read by primary key (row id), pulling one or more whole structs to feed the risk engine — no aggregation, no filtering, no sub-field predicate pushdown. So nesting has essentially no downside for our reads, and top-level column pruning already reads only the structs actually requested.

Why we need sub-field-level updates. We backfill specific sub-fields inside a group over historical data (when a feature definition changes / data is fixed — e.g. recomputing 8 of the ~2599 features in one group), across large historical row ranges. With the existing top-level (whole-column) evolution, changing those few sub-fields forces rewriting the entire struct (up to ~2599 fields) across history — large write amplification; and when a group is maintained by multiple pipelines, whole-column rewrites also clobber each other. Sub-field-level writes aligned by row id let us write only the backfilled leaves and reassemble the rest from the original files by row id, which is exactly the pain point this PR targets.

Known trade-offs. The feature currently supports one level of nesting, and partially-written struct files don't contribute that column's stats to pushdown — which doesn't affect our "point-read only, no pushdown" usage, but it is a limitation and I've noted it in the description.

If you think there's a more suitable direction (either in modeling or in the implementation), I'm happy to discuss and adjust, and to add more docs/tests.

[JingsongLi]

This PR is super complicated. We can first perform some refactoring PRs to make the entire code path move in the direction of Field Id, so that top-level fields and nested fields are treated the same.

[zhuxiangyi]

Thanks @JingsongLi, that makes sense — moving the whole path to field id (so a nested leaf id is just another field id, and top-level vs nested are handled the same) is cleaner than the dotted-path approach here, and it also removes the name-ambiguity under rename. Happy to do this as a series of smaller PRs. Here's a concrete plan and the compatibility strategy.

Phase 1 — refactor to field id (no new feature, behavior unchanged)

• PR-1: introduce writtenFieldIds on DataFileMeta + serialization compat. Append a nullable _WRITTEN_FIELD_IDS ARRAY<INT> to DataFileMeta.SCHEMA, thread it through PojoDataFileMeta/factories, and update DataFileMetaSerializer. The write side populates it (top-level field ids for now) while still dual-writing the existing writeCols. Add a helper that resolves a file's written columns to field ids (writtenFieldIds if present, else old writeCols names → ids).
• PR-2: switch the consumers to field id. RowIdColumnConflictChecker, DataEvolutionFileStoreScan, DataEvolutionCompactCoordinator/Task, DataEvolutionRowIdReassigner, and the read path (FormatKey cache key) all resolve columns by id via that helper. Add RowType/TableSchema.projectByIds(int[]). Still top-level only; behavior identical.

Phase 2 — the nested feature on top of the id-based path

• PR-3: nested sub-field data evolution core. writtenFieldIds may now carry nested leaf ids — a whole column is recorded by its own field id, a partially-written struct by the leaf ids actually written — and the read assembly composes structs by leaf id. This lets us delete the projectByPaths/leafPaths/dotted-path layer entirely; nested and top-level become the same code.
• PR-4 / PR-5: engine entries (Spark MERGE INTO, Flink data_evolution_merge_into) producing leaf-id sets.

Compatibility strategy

• Backward (new reader, old files): writtenFieldIds is a nullable appended field, so old manifests read it as null and fall back to writeCols; for the versioned DataSplit/CommitMessage streams I'll bump the version and add a legacy serializer for the current layout (same pattern as DataFileMeta12LegacySerializer).
• Forward (old reader, new files): as long as we keep dual-writing writeCols (names) next to writtenFieldIds, an old reader simply ignores the extra field and keeps working via writeCols. We'd only drop writeCols later, once all supported versions understand writtenFieldIds. (Tables that actually use the nested feature require the new engine anyway.)

Reuse from this PR: the read-assembly (DataEvolutionReadPlanner / struct reassembly), leaf-level conflict check, the Spark/Flink entry logic and all the tests migrate into the phase-2 PRs; the dotted-path layer is dropped. So this is a re-split, not a rewrite.

Does this split and the dual-write compatibility approach look right to you? Any adjustments welcome.

[JingsongLi]

@zhuxiangyi Sounds cool to me!

[zhuxiangyi]

@JingsongLi The branch has been rebased onto the latest master to resolve a conflict with #9114 (which reworked evolutionStats's winner-selection logic around the same time as this PR). Kept #9114's restructured logic as-is and re-attached the comment explaining why a sub-field-level partial-struct file's type mismatch is intentionally treated as "no stats" there. CI is green on the rebased commits.

Marked it ready for review — would appreciate another look when you have time.

@steFaiz Following up on the createUnionReader complexity you flagged — the extraction is done. The leaf-level matching and nested assembly planning now live in a dedicated, pure DataEvolutionReadPlanner (with its own DataEvolutionReadPlannerTest), and DataEvolutionSplitRead#createUnionReader is back to being a thin shell that just resolves bunch schemas and builds readers from the plan. Would appreciate a look when you have a chance.

[JingsongLi]

Requesting changes because the current patch can silently corrupt persisted nested data. I reproduced two failures under Spark 3: reversed leaf order swaps values across fields, and a copied NULL parent struct becomes a non-NULL struct containing NULL children. The inline comments describe these blockers and the additional schema-contract issues. Before this is shipped, please also document and enforce the mixed-version barrier: old readers cannot reconstruct files whose writeCols contain entries such as nest.a; every reader, writer, compactor, and maintenance job must be upgraded before the feature is enabled, and binary rollback is unsafe after such files have been committed.

[JingsongLi]

[P1] Canonicalize the leaf order before constructing the write schema. union preserves MATCHED-action order, while prunedStructType and buildPrunedStruct emit fields in table-schema order. writePaths later reuses this action-ordered sequence, so the Spark row layout and Paimon's writeType disagree positionally. I reproduced this with nest<a,b,c> and two clauses updating c and then a: expected (10,x,100) / (200,y,40), but read back (100,x,10) / (40,y,200). This silently corrupts persisted data. Please canonicalize the paths once in schema order and use that exact sequence for the output struct, writePaths, and writeType; apply the same fix to the Spark 4 copy and add a reverse-action-order regression test.

[zhuxiangyi]

Thanks for catching this. Fixed by canonicalizing the changed-leaf set to schema declaration order once, so the output struct, writePaths and writeType all follow the same sequence. Spark 4 copy updated too.
Your repro shape mattered — a single clause with two assignments doesn't reproduce, because Spark's own assignment alignment rebuilds it in schema order first. The regression test uses two clauses as you described.

[JingsongLi]

[P1] Preserve the parent struct's nullness on copy/passthrough paths. buildPrunedStruct always returns a non-null CreateNamedStruct; when attr is NULL, this converts the copied value into a non-NULL struct whose selected children are NULL. I reproduced this by matching two source rows, conditionally updating nest.a only for row 1, and leaving row 2's nest as NULL: row 2 reads back with nest IS NULL = false. Please guard this construction with the parent-null condition (for example, an If(IsNull(attr), typedNull, prunedStruct)) and add a regression where the NULL row is included in the touched merge range. The Spark 4 copy has the same issue.

[zhuxiangyi]

Good catch. Guarded in copyOutput, and in updateOutput only for the pure-passthrough case — applying it unconditionally there would make SET t.nest.a = 5 on a NULL nest silently drop the assignment. Both have regression tests; Spark 4 copy updated.
Your repro shape mattered here too: the row must be matched but skipped by its clause, otherwise it keeps its nullness from the base file and doesn't expose the bug.

[JingsongLi]

[P1] This exact-name preference makes the persisted dotted-path encoding ambiguous. A legal schema can contain both a quoted top-level field named a.b and a struct a with child b. leafPaths serializes the nested leaf as the same string a.b, but this branch reconstructs it as the top-level field. I verified that the emitted path resolves to the wrong field ID. Readers, pruning, and conflict detection can consequently attribute a partial file to the wrong field. Please use an unambiguous escaped/versioned or field-ID-based encoding; at minimum, reject a nested write whenever its flattened path collides with a top-level name, and cover the reader and conflict-checker paths in tests.

[zhuxiangyi]

Agreed — I confirmed the emitted path resolves to the wrong field id.
This PR takes the "at minimum, reject" option: leafPaths now refuses a nested path whose fh a literal top-level name, and asks the user to rename one of the two. The read side isunchanged, so nothing on disk changes meaning.
I haven't switched to the escaped/versioned encoding here, since that changes the persisted writeCols format. Happy to do it in a follow-up, or in this PR if you'd rather not merge without it.

[JingsongLi]

[P1] coversFully must also preserve recursive physical field order. With full nest<a INT,b STRING> and projectByPaths(["nest.b", "nest.a"]), the write type is nest<b,a>, but this method returns true, so leafPaths collapses the metadata to [nest]; reconstruction then produces nest<a,b>. I verified that the two types are not equal. Row sidecars are written with the original physical write schema but read with the schema reconstructed from writeCols, so this can swap fields or decode bytes using the wrong type. Please require ordered recursive layout equality; otherwise retain the ordered dotted leaves, and add a round-trip test with different leaf types.

[zhuxiangyi]

You're right, thanks. coversFully now compares field ids positionally and recurses, so a cct keeps its ordered leaves instead of collapsing to the bare column name.
Note DataTypesTest.testProjectByPathsMatchesLegacyProjectIncludingOrder had asserted the p assertion is updated. I also added a reversed-nested-projection read test, sinceDataEvolutionSplitRead calls leafPaths for its format cache key.

[JingsongLi]

[P1] Whole-struct assignments are validated by field name here but are still written positionally. For target ROW<a INT,b INT> and source ROW<b INT,a INT>, this check accepts SET T.nest = S.nest; the projection keeps the source struct unchanged, while sourceType is rebuilt from the target schema order at lines 323-329. The nested row therefore reaches the writer in source order but is interpreted as target order, silently storing a = source.b and b = source.a (extra source fields can misalign it as well). Please recursively rebuild/cast whole structs in target order, or reject any source struct whose ordered shape differs, and add reversed-order and extra-field tests.

[zhuxiangyi]

Fixed with the second option you offered — reject rather than rebuild. isFullyCompatibleSt arity and the same field name at each position, recursing into nested rows, so a reordered or wider source is refused instead of being written under the wrong names.

[JingsongLi]

[P2] Validate duplicate SET targets before converting the clause to a map. parseCommaSeparatedKeyValues returns a map, so T.nest.a = S.x, T.nest.a = S.y loses the first entry before the duplicate check below can see it, and the last RHS silently wins. Please parse into an ordered entry list (or otherwise retain duplicate keys), reject duplicates before grouping, and test both exact duplicates and equivalent qualified/unqualified targets.

[zhuxiangyi]

Good point. matchedUpdateSet is now scanned for duplicates before parseCommaSeparatedKeyVaap, with keys normalized through the existing parseTargetPath so T.nest.a and nest.a compareequal.

[zhuxiangyi]

@JingsongLi Thanks for the detailed review — the reproductions saved a lot of time.

All six comments are addressed, and the mixed-version barrier is documented on data-evolutevery reader, writer, compactor and maintenance job must be upgraded before enabling it, anddowngrading is unsafe once a file whose write columns contain a nested sub-field path has been committed.
Two notes for the re-review:

• The ordering and null regressions are covered end to end, and I validated each test by reverting the fix and confirming it fails first. The clause shapes matter — a single WHEN MATCHED clause is normalized
  into schema order by Spark's own alignment, and an unmatched row keeps its parent-struct n, so neither reproduces the bugs.
• On the a.b ambiguity I implemented the "reject" option rather than changing the persisted writeCols format — glad to do the encoding in a follow-up or here, your call.
  Also added coverage for two previously untested paths: compaction over sub-field files, and adding a nested sub-field after such files exist.
  CI is green. Please take another look when you have time.

[JingsongLi]

[P1] Preserve deep nested schema evolution. A direct child ROW can be only partially covered because the source file predates a newly added nullable nested field, not because that child was written through an unsupported deep partial path. For example, after adding payload.inner.y to files that contain only payload.inner.x, any overlapping top-level partial update sends the group through this planner and this branch makes full reads and compaction fail. When all physically present leaves of the child come from one bunch and the remaining leaves are absent everywhere, please read the child from that provider and let the existing schema-evolution mapping null-fill the missing leaves. Reserve this exception for an actual cross-provider deep split, and add a regression covering deep ADD COLUMN plus an unrelated overlapping partial update.

[JingsongLi]

[P1] Make both Spark conflict rewriters understand these dotted write paths. A nested MERGE now persists entries such as nest.a, but DataEvolutionRowIdConflictRewriter later treats every entry as a top-level relation attribute and throws Cannot find column nest.a when concurrent compaction changes the row-id boundaries. In the reverse commit order, DataEvolutionCompactMergeConflictRewriter uses exact top-level-name containment, finds no updated field for nest.a, and cannot rebase the staged compact output. Please carry path-aware write types through both rewriters; the row-id path must overlay only the staged leaves onto the current struct so untouched siblings are not clobbered. Apply the same change to the Spark 4 copy and add nested variants of the existing concurrent-compaction tests.

[JingsongLi]

[P2] Reject ambiguous target paths before stripping the qualifier. If the target table or alias is payload and the schema contains both payload ROW<a ...> and a top-level column a, the documented unqualified nested target payload.a is normalized to [a] here and the action successfully updates the top-level a instead. Please resolve both interpretations against the target schema and reject the input when both are valid, requiring an explicit form such as payload.payload.a for the nested field or bare a for the top-level field.

[JingsongLi]

+1