Zoltan Borok-Nagy has posted comments on this change. ( http://gerrit.cloudera.org:8080/24137 )
Change subject: IMPALA-13810: Generate lineage records for UPDATE and MERGE statements ...................................................................... Patch Set 12: (12 comments) http://gerrit.cloudera.org:8080/#/c/24137/12/fe/src/main/java/org/apache/impala/planner/Planner.java File fe/src/main/java/org/apache/impala/planner/Planner.java: http://gerrit.cloudera.org:8080/#/c/24137/12/fe/src/main/java/org/apache/impala/planner/Planner.java@245 PS12, Line 245: private static final class LineageComputationState { : private List<Integer> kuduUpdateColIdxs; : private List<Column> kuduUpdateColumns; : // For non-Kudu UPDATE statements, stores the indices of columns being updated : private Set<Integer> updateColIdxs; : } The current two-step logic seems fragile to me (and hard to read), consider the following design: /** * Target columns paired by index with the source exprs feeding each, plus predicate-only * deps. For MERGE a target may have several source exprs (one per non-DELETE case) unioned * into one PROJECTION edge; INSERT/UPDATE/SELECT use singleton lists; an empty inner list * (delete-only MERGE) yields no projection edge. */ private static final class LineageProjection { final List<ColumnLabel> targets; final List<List<Expr>> sourcesPerTarget; // parallel to targets final List<Expr> extraPredicates; // MERGE ON/WHEN filters; empty otherwise LineageProjection(List<ColumnLabel> t, List<List<Expr>> s, List<Expr> p) { Preconditions.checkState(t.size() == s.size()); targets = t; sourcesPerTarget = s; extraPredicates = p; } } Then in computeColumnLineage(): private void computeColumnLineage(PlanFragment rootFragment) { Analyzer analyzer = ctx_.getRootAnalyzer(); ColumnLineageGraph graph = analyzer.getColumnLineageGraph(); LineageProjection p = buildLineageProjection(rootFragment); graph.addTargetColumnLabels(p.targets); graph.addDependencyPredicates(p.extraPredicates); // no-op for non-MERGE graph.computeProjectionEdges(p.sourcesPerTarget, analyzer, ctx_.getAnalysisResult().getParsedStmt().getOperationType()); if (LOG.isTraceEnabled()) LOG.trace("lineage: " + graph.debugString()); ctx_.getTimeline().markEvent("Lineage info computed"); } private LineageProjection buildLineageProjection(PlanFragment rootFragment) { if (ctx_.isInsertOrCtas()) return projectionForInsertOrCtas(rootFragment); if (ctx_.isUpdate()) return projectionForUpdate(rootFragment); if (ctx_.isMerge()) return projectionForMerge(); ... } http://gerrit.cloudera.org:8080/#/c/24137/12/fe/src/main/java/org/apache/impala/planner/Planner.java@368 PS12, Line 368: targetTable Add precondition check that it is an Iceberg table. http://gerrit.cloudera.org:8080/#/c/24137/12/fe/src/main/java/org/apache/impala/planner/Planner.java@371 PS12, Line 371: (String columnName: targetTable.getColumnNames()) This labels all columns, while addTargetColumnLabelsForUpdate() correctly only labels assigned columns. http://gerrit.cloudera.org:8080/#/c/24137/12/fe/src/main/java/org/apache/impala/planner/Planner.java@418 PS12, Line 418: colIdx < outputExprs.size() Should be a precondition check? Or add comment please. http://gerrit.cloudera.org:8080/#/c/24137/11/testdata/workloads/functional-query/queries/QueryTest/lineage-iceberg-update-merge.test File testdata/workloads/functional-query/queries/QueryTest/lineage-iceberg-update-merge.test: http://gerrit.cloudera.org:8080/#/c/24137/11/testdata/workloads/functional-query/queries/QueryTest/lineage-iceberg-update-merge.test@353 PS11, Line 353: { : "sources": [ : 0 : ], : "targets": [ : 0 : ], : "edgeType": "PROJECTION" : } > We're updating event_time based on its own value, so that's why the project Ack. We should also have a similar self-assignment for a MERGE statement, and only keep the PROJECTION edge for that. http://gerrit.cloudera.org:8080/#/c/24137/11/testdata/workloads/functional-query/queries/QueryTest/lineage-iceberg-update-merge.test@810 PS11, Line 810: { : "sources": [ : 0 : ], : "targets": [ : 0 : ], : "edgeType": "PROJECTION" : } > I think in the update case, it reinserts the modified rows and only changes Regular updates don't produce these self-assignments. These generate noise, would it be possible to filter them out? http://gerrit.cloudera.org:8080/#/c/24137/11/testdata/workloads/functional-query/queries/QueryTest/lineage-iceberg-update-merge.test@826 PS11, Line 826: { : "sources": [ : 2 : ], : "targets": [ : 2 : ], : "edgeType": "PROJECTION" : }, : { : "sources": [ : 3 : ], : "targets": [ : 3 : ], : "edgeType": "PROJECTION" These are also noise only. http://gerrit.cloudera.org:8080/#/c/24137/12/testdata/workloads/functional-query/queries/QueryTest/lineage-iceberg-update-merge.test File testdata/workloads/functional-query/queries/QueryTest/lineage-iceberg-update-merge.test: http://gerrit.cloudera.org:8080/#/c/24137/12/testdata/workloads/functional-query/queries/QueryTest/lineage-iceberg-update-merge.test@704 PS12, Line 704: update set t.event_time = HOURS_ADD(s.event_time, 1) Can we have MERGE test with - multiple UPDATE case - more columns, different assignment order - multi-source updates, e.g. set c = a + b http://gerrit.cloudera.org:8080/#/c/24137/12/testdata/workloads/functional-query/queries/QueryTest/lineage-iceberg-update-merge.test@705 PS12, Line 705: when not matched then insert Can we have tests for MERGE statement with multiple INSERT branches with different column permutations? http://gerrit.cloudera.org:8080/#/c/24137/12/testdata/workloads/functional-query/queries/QueryTest/lineage-iceberg-update-merge.test@1467 PS12, Line 1467: user = r.new_user, action = r.new_action Please add tests for even more columns, and using different column order in the assignments. http://gerrit.cloudera.org:8080/#/c/24137/11/testdata/workloads/functional-query/queries/QueryTest/lineage-kudu-update.test File testdata/workloads/functional-query/queries/QueryTest/lineage-kudu-update.test: http://gerrit.cloudera.org:8080/#/c/24137/11/testdata/workloads/functional-query/queries/QueryTest/lineage-kudu-update.test@418 PS11, Line 418: "sources": [ : 0 : ], : "targets": [ : 0 : ] Is it possible to filter this out? http://gerrit.cloudera.org:8080/#/c/24137/12/testdata/workloads/functional-query/queries/QueryTest/lineage-kudu-update.test File testdata/workloads/functional-query/queries/QueryTest/lineage-kudu-update.test: http://gerrit.cloudera.org:8080/#/c/24137/12/testdata/workloads/functional-query/queries/QueryTest/lineage-kudu-update.test@623 PS12, Line 623: set string_col = r.new_string, bigint_col = r.new_bigint Add tests for updating multiple columns, in various column order. -- To view, visit http://gerrit.cloudera.org:8080/24137 To unsubscribe, visit http://gerrit.cloudera.org:8080/settings Gerrit-Project: Impala-ASF Gerrit-Branch: master Gerrit-MessageType: comment Gerrit-Change-Id: Icba79f756509438455bfbf3067733f6f29284220 Gerrit-Change-Number: 24137 Gerrit-PatchSet: 12 Gerrit-Owner: Daniel Vanko <[email protected]> Gerrit-Reviewer: Daniel Vanko <[email protected]> Gerrit-Reviewer: Fang-Yu Rao <[email protected]> Gerrit-Reviewer: Impala Public Jenkins <[email protected]> Gerrit-Reviewer: Peter Rozsa <[email protected]> Gerrit-Reviewer: Zoltan Borok-Nagy <[email protected]> Gerrit-Comment-Date: Mon, 29 Jun 2026 14:32:09 +0000 Gerrit-HasComments: Yes
