[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.