Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[SPARK-42851][SQL] Guard EquivalentExpressions.addExpr() with supportedExpression() #40473

Closed
wants to merge 3 commits into from
Closed

[SPARK-42851][SQL] Guard EquivalentExpressions.addExpr() with supportedExpression() #40473

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
08309e7
SPARK-42851: guard EquivalentExpressions.addExpr() with supportedExpr…
rednaxelafx Mar 18, 2023
28d101e
adjust the test case
rednaxelafx Mar 21, 2023
d9f8bbd
address comments, adjust test case to use LambdaVariable instead of N…
rednaxelafx Mar 21, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 5 additions & 1 deletion ...lyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/EquivalentExpressions.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,7 +40,11 @@ class EquivalentExpressions {
* Returns true if there was already a matching expression.
*/
def addExpr(expr: Expression): Boolean = {
updateExprInMap(expr, equivalenceMap)
if (supportedExpression(expr)) {
updateExprInMap(expr, equivalenceMap)
} else {
false
}
}

/**
Expand Down
16 changes: 15 additions & 1 deletion .../test/scala/org/apache/spark/sql/catalyst/expressions/SubexpressionEliminationSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,7 @@ import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.expressions.codegen._
import org.apache.spark.sql.catalyst.plans.logical.LocalRelation
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types.{BinaryType, DataType, IntegerType}
import org.apache.spark.sql.types.{ArrayType, BinaryType, DataType, IntegerType}

class SubexpressionEliminationSuite extends SparkFunSuite with ExpressionEvalHelper {
test("Semantic equals and hash") {
Expand Down Expand Up @@ -449,6 +449,20 @@ class SubexpressionEliminationSuite extends SparkFunSuite with ExpressionEvalHel
assert(e2.getCommonSubexpressions.size == 1)
assert(e2.getCommonSubexpressions.head == add)
}

test("SPARK-42851: Handle supportExpressions consistently across add and get") {
val tx = {
val arr = Literal(Array(1, 2))
val ArrayType(et, cn) = arr.dataType
val lv = NamedLambdaVariable("x", et, cn)
val lambda = LambdaFunction(lv, Seq(lv))
ArrayTransform(arr, lambda)
}
val equivalence = new EquivalentExpressions
val isNewExpr = equivalence.addExpr(tx)
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

.addExpr()'s boolean result is counter-intuitive to other collection's .add() methods so IMO isNewExpr should be !equivalence.addExpr(tx) here. This is the reasons why I think getting rid of .addExpr() is probably the most straightforward fix here: #40488

val cseState = equivalence.getExprState(tx)
assert(isNewExpr == cseState.isDefined)
}
}

case class CodegenFallbackExpression(child: Expression)
Expand Down
7 changes: 7 additions & 0 deletions sql/core/src/test/scala/org/apache/spark/sql/DataFrameAggregateSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1538,6 +1538,13 @@ class DataFrameAggregateSuite extends QueryTest
)
checkAnswer(res, Row(1, 1, 1) :: Row(4, 1, 2) :: Nil)
}

test("SPARK-42851: common subexpression should consistently handle aggregate and result exprs") {
val res = sql(
"select max(transform(array(id), x -> x)), max(transform(array(id), x -> x)) from range(2)"
)
checkAnswer(res, Row(Array(1), Array(1)))
}
}

case class B(c: Option[Double])
Expand Down