JingsongLi commented on code in PR #8334:
URL: https://github.com/apache/paimon/pull/8334#discussion_r3888775156
##########
paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonDataEvolutionTable.scala:
##########
@@ -689,6 +791,16 @@ case class MergeIntoPaimonDataEvolutionTable(
.sortWithinPartitions(FIRST_ROW_ID_NAME, ROW_ID_NAME)
}
+ // dotted write paths: a whole column -> its name; a pruned struct ->
"col.subfield..." leaves
+ val writePaths = updateColumnsSorted.flatMap {
+ attr =>
+ prunedByExprId.get(attr.exprId) match {
+ case Some((paths, _)) => paths.map(p => (attr.name +:
p).mkString("."))
Review Comment:
[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.
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/DataEvolutionMergeIntoAction.java:
##########
@@ -319,13 +309,206 @@ 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() {
+ checkNoDuplicateSetTargets();
+ Map<String, String> changes =
parseCommaSeparatedKeyValues(matchedUpdateSet);
+
+ // group by top-level column, preserving first-seen order
+ Map<String, String> wholeCols = new LinkedHashMap<>();
+ Map<String, LinkedHashMap<String, String>> nestedCols = new
LinkedHashMap<>();
+ List<String> order = new ArrayList<>();
+
+ for (Map.Entry<String, String> entry : changes.entrySet()) {
+ List<String> path = parseTargetPath(entry.getKey());
+ String topCol = path.get(0);
+ if (!targetFieldNames.contains(topCol)) {
+ throw new RuntimeException(
+ String.format(
+ "Invalid column reference '%s' of table '%s'
at matched-upsert action.",
+ entry.getKey(), identifier.getFullName()));
+ }
+ if (!order.contains(topCol)) {
+ order.add(topCol);
+ }
+ if (path.size() == 1) {
+ // whole top-level column
+ if (nestedCols.containsKey(topCol) ||
wholeCols.containsKey(topCol)) {
+ throw new RuntimeException(
+ "Conflicting updates for column '" + topCol + "'
in SET clause.");
+ }
+ wholeCols.put(topCol, entry.getValue());
+ } else {
+ // nested sub-field update
+ if (!coreOptions.dataEvolutionNestedFieldEnabled()) {
+ throw new UnsupportedOperationException(
+ "Updating a nested sub-field ('"
+ + entry.getKey()
+ + "') requires '"
+ +
CoreOptions.DATA_EVOLUTION_NESTED_FIELD_ENABLED.key()
+ + "=true'.");
+ }
+ if (path.size() > 2) {
+ throw new UnsupportedOperationException(
+ "Sub-field-level data evolution only supports one
level of nesting, "
+ + "but got '"
+ + entry.getKey()
+ + "'.");
+ }
+ if (wholeCols.containsKey(topCol)) {
+ throw new RuntimeException(
+ "Conflicting updates for column '" + topCol + "'
in SET clause.");
+ }
+ String subName = path.get(1);
+ LinkedHashMap<String, String> subs =
+ nestedCols.computeIfAbsent(topCol, k -> new
LinkedHashMap<>());
+ if (subs.containsKey(subName)) {
+ throw new RuntimeException(
+ "Duplicated update for sub-field '"
+ + topCol
+ + "."
+ + subName
+ + "' in SET clause.");
+ }
+ subs.put(subName, entry.getValue());
+ }
+ }
+
+ // first pass: writePaths (so projectByPaths can build the pruned
nested types)
+ writePaths = new ArrayList<>();
+ for (String topCol : order) {
+ if (wholeCols.containsKey(topCol)) {
+ writePaths.add(topCol);
+ } else {
+ // Emit sub-fields in schema declaration order rather than
SET-clause order: the
+ // write paths become the physical column layout of the
incremental file (both
+ // projectByPaths and the writeCols recorded in the manifest
keep the given order),
+ // and that layout should not depend on how the user happened
to order the SET
+ // clauses. The projection values below follow the pruned
struct, so they adapt.
+ LinkedHashMap<String, String> subs = nestedCols.get(topCol);
+ RowType topColType = (RowType)
table.rowType().getField(topCol).type();
+ for (DataField subField : topColType.getFields()) {
+ if (subs.containsKey(subField.name())) {
+ writePaths.add(topCol + "." + subField.name());
+ }
+ }
+ }
+ }
+
+ // second pass: build projection expressions
+ RowType pruned = table.rowType().projectByPaths(writePaths);
+ List<String> project = new ArrayList<>();
+ for (String topCol : order) {
+ if (wholeCols.containsKey(topCol)) {
+ project.add(String.format("%s AS `%s`", wholeCols.get(topCol),
topCol));
+ } else {
+ LinkedHashMap<String, String> subs = nestedCols.get(topCol);
+ DataType prunedColType =
+
pruned.getField(table.rowType().getField(topCol).id()).type();
+ // value order must match the pruned struct's schema field
order
+ List<String> values = new ArrayList<>();
+ for (DataField subField : ((RowType)
prunedColType).getFields()) {
+ String value = subs.get(subField.name());
+ Preconditions.checkState(
+ value != null,
+ "Missing value for sub-field '%s.%s', it's a bug.",
+ topCol,
+ subField.name());
+ values.add(value);
+ }
+ String typeStr =
+
LogicalTypeConversion.toLogicalType(prunedColType).asSerializableString();
+ project.add(
+ String.format(
+ "CAST(ROW(%s) AS %s) AS `%s`",
+ String.join(", ", values), typeStr, topCol));
+ }
+ }
+ return project;
+ }
+
+ /** The first-seen order of top-level columns present in the (dotted)
write paths. */
+ private List<String> explicitTopColumnOrder(List<String> paths) {
+ List<String> order = new ArrayList<>();
+ for (String path : paths) {
+ int dot = path.indexOf('.');
+ String topCol = dot < 0 ? path : path.substring(0, dot);
+ if (!order.contains(topCol)) {
+ order.add(topCol);
+ }
+ }
+ return order;
+ }
+
+ /**
+ * Parse a SET target into a path relative to the target table: strip an
optional leading
+ * table-qualifier segment (the target table name/alias), leaving {@code
[topColumn, sub...]}.
+ */
+ private List<String> parseTargetPath(String target) {
+ List<String> segs = new
ArrayList<>(Arrays.asList(target.split("\\.")));
+ if (segs.size() > 1 && segs.get(0).equals(targetTableName())) {
Review Comment:
[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.
--
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]
