zhuxiangyi commented on code in PR #8334:
URL: https://github.com/apache/paimon/pull/8334#discussion_r3795051332
##########
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:
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.
--
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]
