xupefei commented on code in PR #48920:
URL: https://github.com/apache/spark/pull/48920#discussion_r1852025363
##########
sql/connect/server/src/main/scala/org/apache/spark/sql/connect/planner/SparkConnectPlanner.scala:
##########
@@ -1190,14 +1190,14 @@ class SparkConnectPlanner(
val input = transformRelation(rel.getInput)
if (input.isStreaming || executeHolderOpt.isEmpty) {
- CollectMetrics(name, metrics.map(_.named),
transformRelation(rel.getInput), planId)
+ CollectMetrics(name, metrics.map(_.named), input, planId)
Review Comment:
This fixes a bug where the input of a `CollectMetrics` can be processed two
times, once in Line 1190 and once here/below.
When the `input` contains another `CollectMetrics`, transforming it twice
will cause two `Observation` objects (in the input) to be initialised and
registered two times to the system. Since only one of them will be fulfilled
when the query finishes, the one we'll be looking at may not be the right one.
This issue is captured by the test case `Observation.get is blocked until
the query is finished ...` where we only execute the `observedObservedDf`,
which is a `CollectMetrics` having another `CollectMetrics` as input.
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
