JingsongLi commented on code in PR #8334:
URL: https://github.com/apache/paimon/pull/8334#discussion_r3788779285
##########
paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonDataEvolutionTable.scala:
##########
@@ -541,7 +611,12 @@ case class MergeIntoPaimonDataEvolutionTable(
if (rawBlobUpdateColumns.exists(_.sameRef(attr))) {
Literal(null, attr.dataType)
} else {
- attr
+ prunedByExprId.get(attr.exprId) match {
+ case Some((paths, _)) =>
+ val st = attr.dataType.asInstanceOf[StructType]
+ buildPrunedStruct(st, Nil, paths, p => passthroughExpr(attr,
st, p))
Review Comment:
[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.
##########
paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonDataEvolutionTable.scala:
##########
@@ -472,7 +472,64 @@ case class MergeIntoPaimonDataEvolutionTable(
AttributeReference(MERGE_DELETED_NAME, BooleanType, nullable = false)()
// Row metadata, _FIRST_ROW_ID added by addFirstRowId, and the delete
marker.
val fixedMergeOutputColumnCount = metadataColumns.size + 2
- val mergeOutput = (updateColumnsSorted ++ metadataColumns ++
rawBlobMarkerAttributes) :+
+
+ // Sub-field-level pruning: for a struct column whose SET only touches
some sub-fields, only the
+ // changed leaves are written (an incremental column-group file containing
the partial struct);
+ // the rest are copied from the target. Falls back to whole-column write
when the changed leaves
+ // cannot be safely determined, so behaviour never regresses.
+ val matchedUpdateActions = matchedActions.collect { case ua: UpdateAction
=> ua }
+ // Gated by data-evolution.nested-field.enabled (default off): when
disabled, no column is
+ // pruned, so every struct column is rewritten whole (behaviour identical
to before this
+ // feature). When enabled, struct columns whose SET only touches some
sub-fields are pruned.
+ val nestedFieldEnabled =
table.coreOptions().dataEvolutionNestedFieldEnabled()
+ val prunedByExprId: Map[ExprId, (Seq[Seq[String]], StructType)] =
+ if (!nestedFieldEnabled) Map.empty
+ else
+ updateColumnsSorted.flatMap {
+ attr =>
+ if (rawBlobUpdateColumns.exists(_.sameRef(attr))) {
+ None
+ } else {
+ attr.dataType match {
+ case st: StructType =>
+ val perAction = matchedUpdateActions.flatMap {
+ ua =>
+ ua.assignments
+ .find(
+ a => isModifiedAssignment(a) &&
assignmentKeyAttribute(a).sameRef(attr))
+ .map(a => changedLeaves(a.value, st, attr))
+ }
+ if (perAction.isEmpty || perAction.exists(_.isEmpty)) {
+ None
+ } else {
+ val union = perAction.flatten.flatten.map(_._1).distinct
Review Comment:
[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.
##########
paimon-api/src/main/java/org/apache/paimon/types/RowType.java:
##########
@@ -333,6 +334,166 @@ public RowType project(String... names) {
return project(Arrays.asList(names));
}
+ /**
+ * Project this row type by a list of (possibly nested) dotted paths, e.g.
{@code ["f0",
+ * "nest.a"]}. A path without a dot selects the whole top-level field
(same as {@link
+ * #project(List)}); a dotted path selects only the addressed sub-field of
a nested {@link
+ * RowType}, preserving field ids and nullability of every level. Fields
are emitted in the
+ * order the paths are given (exactly like {@link #project(List)}), not in
schema declaration
+ * order. This is used by data evolution to reconstruct the partial nested
schema of a
+ * column-group file from its {@code writeCols}.
+ */
+ public RowType projectByPaths(List<String> paths) {
+ return projectTypeByPaths(this, paths);
+ }
+
+ private static RowType projectTypeByPaths(RowType type, List<String>
paths) {
+ // group paths by their immediate child name, keeping the order in
which the paths are
+ // given; a child appearing without a tail (or also with a tail) is
selected as a whole
+ // field
+ Map<String, List<String>> childToSubPaths = new LinkedHashMap<>();
+ Set<String> wholeChildren = new HashSet<>();
+ Map<String, DataField> fieldByName = new LinkedHashMap<>();
+ for (DataField field : type.getFields()) {
+ fieldByName.put(field.name(), field);
+ }
+ for (String path : paths) {
+ int dot = path.indexOf('.');
+ // Prefer an exact field-name match so a column whose name itself
contains a dot (and
+ // any
+ // plain top-level name) is selected whole; only split into
head.tail for genuine nested
+ // sub-field paths that do not name a field directly. This keeps
backward compatibility
+ // with the legacy exact-name project(List).
+ if (dot < 0 || fieldByName.containsKey(path)) {
Review Comment:
[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.
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/DataEvolutionMergeIntoAction.java:
##########
@@ -319,13 +308,178 @@ public Tuple2<DataStream<RowData>, RowType>
buildSource() {
Table source = batchTEnv.sqlQuery(query);
checkSchema(source);
- RowType sourceType =
- SpecialFields.rowTypeWithRowId(table.rowType())
- .project(source.getResolvedSchema().getColumnNames());
+
+ RowType sourceType;
+ if (updateAll) {
+ List<String> columnNames =
source.getResolvedSchema().getColumnNames();
+ sourceType =
SpecialFields.rowTypeWithRowId(table.rowType()).project(columnNames);
+ writePaths =
+ columnNames.stream()
+ .filter(name ->
!SpecialFields.ROW_ID.name().equals(name))
+ .collect(Collectors.toList());
+ } else {
+ // build the source type manually so _ROW_ID is first and the
column order matches the
+ // SQL projection order; for nested columns the field is the
partial (pruned) struct.
+ RowType pruned = table.rowType().projectByPaths(writePaths);
+ List<DataField> srcFields = new ArrayList<>();
+ srcFields.add(SpecialFields.ROW_ID);
+ for (String topCol : explicitTopColumnOrder(writePaths)) {
+
srcFields.add(pruned.getField(table.rowType().getField(topCol).id()));
+ }
+ sourceType = new RowType(srcFields);
+ }
return Tuple2.of(toDataStream(source), sourceType);
}
+ /**
+ * Validate the SET targets and build the SQL projection list. A target
may address a top-level
+ * column ({@code col} / {@code T.col}) or, for sub-field-level data
evolution, a nested
+ * sub-field ({@code nest.a} / {@code T.nest.a}). A partially-updated
struct column is rebuilt
+ * as a partial {@code CAST(ROW(...) AS ROW<...>)} so only the touched
sub-fields are written.
+ * Also sets {@link #writePaths}.
+ */
+ private List<String> buildExplicitProject() {
+ Map<String, String> changes =
parseCommaSeparatedKeyValues(matchedUpdateSet);
Review Comment:
[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.
##########
paimon-api/src/main/java/org/apache/paimon/types/RowType.java:
##########
@@ -333,6 +334,166 @@ public RowType project(String... names) {
return project(Arrays.asList(names));
}
+ /**
+ * Project this row type by a list of (possibly nested) dotted paths, e.g.
{@code ["f0",
+ * "nest.a"]}. A path without a dot selects the whole top-level field
(same as {@link
+ * #project(List)}); a dotted path selects only the addressed sub-field of
a nested {@link
+ * RowType}, preserving field ids and nullability of every level. Fields
are emitted in the
+ * order the paths are given (exactly like {@link #project(List)}), not in
schema declaration
+ * order. This is used by data evolution to reconstruct the partial nested
schema of a
+ * column-group file from its {@code writeCols}.
+ */
+ public RowType projectByPaths(List<String> paths) {
+ return projectTypeByPaths(this, paths);
+ }
+
+ private static RowType projectTypeByPaths(RowType type, List<String>
paths) {
+ // group paths by their immediate child name, keeping the order in
which the paths are
+ // given; a child appearing without a tail (or also with a tail) is
selected as a whole
+ // field
+ Map<String, List<String>> childToSubPaths = new LinkedHashMap<>();
+ Set<String> wholeChildren = new HashSet<>();
+ Map<String, DataField> fieldByName = new LinkedHashMap<>();
+ for (DataField field : type.getFields()) {
+ fieldByName.put(field.name(), field);
+ }
+ for (String path : paths) {
+ int dot = path.indexOf('.');
+ // Prefer an exact field-name match so a column whose name itself
contains a dot (and
+ // any
+ // plain top-level name) is selected whole; only split into
head.tail for genuine nested
+ // sub-field paths that do not name a field directly. This keeps
backward compatibility
+ // with the legacy exact-name project(List).
+ if (dot < 0 || fieldByName.containsKey(path)) {
+ childToSubPaths.computeIfAbsent(path, k -> new ArrayList<>());
+ wholeChildren.add(path);
+ } else {
+ String head = path.substring(0, dot);
+ String tail = path.substring(dot + 1);
+ childToSubPaths.computeIfAbsent(head, k -> new
ArrayList<>()).add(tail);
+ }
+ }
+
+ // Emit fields in the order the paths were given, exactly like
project(List). Callers such
+ // as TableSchema.project(writeCols) rebuild the physical layout of a
data file from its
+ // writeCols, and that order is not necessarily the schema declaration
order; reordering
+ // here would silently describe the file with the columns permuted.
+ List<DataField> result = new ArrayList<>();
+ for (Map.Entry<String, List<String>> entry :
childToSubPaths.entrySet()) {
+ String name = entry.getKey();
+ DataField field = fieldByName.get(name);
+ if (field == null) {
+ throw new IllegalArgumentException(
+ "Cannot project by paths, unknown field '" + name + "'
in " + type);
+ }
+ List<String> subPaths = entry.getValue();
+ if (wholeChildren.contains(name) || subPaths.isEmpty()) {
+ result.add(field);
+ } else if (field.type() instanceof RowType) {
+ RowType prunedChild =
+ projectTypeByPaths((RowType) field.type(), subPaths)
+ .copy(field.type().isNullable());
+ result.add(field.newType(prunedChild));
+ } else {
+ // a dotted path addresses a sub-field, but this field is not
a ROW; reject rather
+ // than silently selecting the whole field, so invalid dotted
paths surface early
+ throw new IllegalArgumentException(
+ "Cannot project sub-field(s) "
+ + subPaths
+ + " of non-ROW field '"
+ + name
+ + "' in "
+ + type);
+ }
+ }
+ return new RowType(type.isNullable(), result);
+ }
+
+ /**
+ * Compute the dotted paths describing this (possibly partially nested)
write type relative to a
+ * full row type. A top-level field, or a nested field whose structure
fully covers the
+ * corresponding field in {@code fullType}, is emitted by its name; a
nested field that only
+ * covers some sub-fields is expanded into dotted leaf paths. This is the
inverse of {@link
+ * #projectByPaths(List)} and is used to derive {@code writeCols}.
+ */
+ public List<String> leafPaths(RowType fullType) {
+ List<String> result = new ArrayList<>();
+ collectLeafPaths(getFields(), fullType, "", result);
+ return result;
+ }
+
+ private static void collectLeafPaths(
+ List<DataField> writeFields, RowType fullType, String prefix,
List<String> out) {
+ for (DataField writeField : writeFields) {
+ String path = prefix.isEmpty() ? writeField.name() : prefix + "."
+ writeField.name();
+ // A field absent from the reference type (e.g. the _ROW_ID /
_SEQUENCE_NUMBER special
+ // fields added by row tracking, which are not part of the table's
logical row type) has
+ // no sub-field split: emit it whole by name, matching the legacy
getFieldNames()
+ // output.
+ if (!fullType.containsField(writeField.id())) {
+ out.add(path);
+ continue;
+ }
+ DataField fullField = fullType.getField(writeField.id());
+ boolean willExpand =
+ writeField.type() instanceof RowType
+ && fullField.type() instanceof RowType
+ && !coversFully(
+ (RowType) writeField.type(), (RowType)
fullField.type());
+ // A dotted path is only unambiguous if no name segment contains a
literal '.'. A name
+ // with a dot is fine when emitted whole at top level
(projectByPaths matches it
+ // exactly),
+ // but not when it participates in a multi-segment nested path.
+ if (writeField.name().indexOf('.') >= 0 && (!prefix.isEmpty() ||
willExpand)) {
+ throw new UnsupportedOperationException(
+ "Sub-field-level data evolution does not support a
nested field whose name "
+ + "contains '.': "
+ + path);
+ }
+ if (willExpand) {
+ // A partial struct nested inside another partial struct (a
path deeper than one
+ // level, e.g. nest.sub.x) cannot be composed back on read —
the data-evolution read
+ // path only assembles one nested level. Reject it here so
such a file is never
+ // written/committed and later breaks full-table reads.
+ if (!prefix.isEmpty()) {
+ throw new UnsupportedOperationException(
+ "Sub-field-level data evolution supports only one
level of partial "
+ + "nesting; the nested sub-field '"
+ + path
+ + "' cannot be partially written. Write
the whole '"
+ + path
+ + "' sub-field instead.");
+ }
+ collectLeafPaths(
+ ((RowType) writeField.type()).getFields(),
+ (RowType) fullField.type(),
+ path,
+ out);
+ } else {
+ out.add(path);
+ }
+ }
+ }
+
+ /** Whether {@code part} contains every (recursively nested) field of
{@code full}. */
+ private static boolean coversFully(RowType part, RowType full) {
Review Comment:
[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.
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/DataEvolutionMergeIntoAction.java:
##########
@@ -527,7 +681,23 @@ private void checkSchema(Table source) {
.getTypeRoot()
.getFamilies()
.contains(DataTypeFamily.BINARY_STRING);
+ // Struct columns need a structural compatibility check:
DataTypeCasts does not
+ // support ROW-to-ROW casts. For a sub-field write (dotted
paths like nest.a) the
+ // source is a partial (subset) struct carrying only the
updated sub-fields, so a
+ // subset check is correct. For a whole-column assignment
(e.g. T.nest=S.nest) the
+ // source must fully cover the target struct, so a narrower
source is rejected
+ // instead of being written as an incomplete whole-struct file.
+ boolean structCompatible = false;
+ if (paimonType instanceof RowType && targetField.type()
instanceof RowType) {
+ RowType sourceStruct = (RowType) paimonType;
+ RowType targetStruct = (RowType) targetField.type();
+ structCompatible =
+ isSubFieldWrite(flinkColumn.getName())
+ ? isCompatiblePartialStruct(sourceStruct,
targetStruct)
+ : isFullyCompatibleStruct(sourceStruct,
targetStruct);
Review Comment:
[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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
