unigof commented on code in PR #37630: URL: https://github.com/apache/spark/pull/37630#discussion_r1427878186
########## sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/MergeScalarSubqueries.scala: ########## @@ -223,87 +224,383 @@ object MergeScalarSubqueries extends Rule[LogicalPlan] { } } - // Recursively traverse down and try merging 2 plans. If merge is possible then return the merged - // plan with the attribute mapping from the new to the merged version. - // Please note that merging arbitrary plans can be complicated, the current version supports only - // some of the most important nodes. + /** + * Recursively traverse down and try merging 2 plans. + * + * Please note that merging arbitrary plans can be complicated, the current version supports only + * some of the most important nodes. + * + * @param newPlan a new plan that we want to merge to an already processed plan + * @param cachedPlan a plan that we already processed, it can be either an original plan or a + * merged version of 2 or more plans + * @param filterPropagationSupported a boolean flag that we propagate down to signal we have seen + * an `Aggregate` node where propagated filters can be merged + * @return A tuple of: + * - the merged plan, + * - the attribute mapping from the new to the merged version, + * - the 2 optional filters of both plans that we need to propagate up and merge in + * an ancestor `Aggregate` node if possible, + * - the optional accumulated extra cost of merge that we need to propagate up and + * check in the ancestor `Aggregate` node. + * The cost is optional to signal if the cost needs to be taken into account up in the + * `Aggregate` node to decide about merge. + */ private def tryMergePlans( newPlan: LogicalPlan, - cachedPlan: LogicalPlan): Option[(LogicalPlan, AttributeMap[Attribute])] = { - checkIdenticalPlans(newPlan, cachedPlan).map(cachedPlan -> _).orElse( + cachedPlan: LogicalPlan, + filterPropagationSupported: Boolean): + Option[(LogicalPlan, AttributeMap[Attribute], Option[Expression], Option[Expression], + Option[Double])] = { + checkIdenticalPlans(newPlan, cachedPlan).map { outputMap => + // Currently the cost is always propagated up when `filterPropagationSupported` is true but + // later we can address cases when we don't need to take cost into account. Please find the + // details at the `Filter` node handling. + val mergeCost = if (filterPropagationSupported) Some(0d) else None + + (cachedPlan, outputMap, None, None, mergeCost) + }.orElse( (newPlan, cachedPlan) match { Review Comment: Could you add DSv2 support(especially parquet) for this pr? I can test it's performance in our production env, thank you very much -- 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: reviews-unsubscr...@spark.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org --------------------------------------------------------------------- To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For additional commands, e-mail: reviews-h...@spark.apache.org