yabola commented on code in PR #7127:
URL: https://github.com/apache/kyuubi/pull/7127#discussion_r2205401533
##########
extensions/spark/kyuubi-spark-lineage/src/main/scala/org/apache/kyuubi/plugin/lineage/helper/SparkSQLLineageParseHelper.scala:
##########
@@ -307,7 +310,35 @@ trait LineageParser {
extractColumnsLineage(getQuery(plan), parentColumnsLineage).map { case
(k, v) =>
k.withName(s"$table.${k.name}") -> v
}
+ case p if p.nodeName == "MergeRows" =>
+ val instructionsOutputs =
+ getField[Seq[Expression]](p, "matchedInstructions")
+ .map(extractInstructionOutputs) ++
+ getField[Seq[Expression]](p, "notMatchedInstructions")
+ .map(extractInstructionOutputs)
+ val nextColumnsLineage = ListMap(p.output.indices.map { index =>
+ val keyAttr = p.output(index)
+ val instructionOutputs = instructionsOutputs.map(_(index))
Review Comment:
@wForget I think index match should always be correct.
1. `ResolveRowLevelCommandAssignments#alignActions` will align the length
of the assignments in the merge-into actions to match the target table output.
2. `RewriteMergeIntoTable` will rewrite the merge-into operation as either
`WriteDelta` or `ReplaceData` only when it has being an already aligned plan.
3. I have reviewed all the places where `mergeRows` is referenced.
`buildReplaceDataPlan` -> target table + `metadataAttrs`
`buildWriteDeltaPlan` -> `operationTypeAttr` + target table +
`metadataAttrs` + `originalRowIdAttrs`
`RewriteMergeIntoTable` -> assignments(target table output)
--
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: notifications-unsubscr...@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscr...@kyuubi.apache.org
For additional commands, e-mail: notifications-h...@kyuubi.apache.org
