Skip to content

[SPARK-32931][SQL] Unevaluable Expressions are not Foldable #29798

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

Closed
gatorsmile wants to merge 4 commits into from
Closed

[SPARK-32931][SQL] Unevaluable Expressions are not Foldable #29798

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
043ba63
fix
gatorsmile Sep 18, 2020
ca4b371
fix
gatorsmile Sep 18, 2020
590eef5
address comments
gatorsmile Sep 23, 2020
7ecb1e6
fix
gatorsmile Sep 23, 2020
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
5 changes: 0 additions & 5 deletions sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/unresolved.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -255,7 +255,6 @@ case class UnresolvedFunction(
override def children: Seq[Expression] = arguments ++ filter.toSeq

override def dataType: DataType = throw new UnresolvedException(this, "dataType")
override def foldable: Boolean = throw new UnresolvedException(this, "foldable")
override def nullable: Boolean = throw new UnresolvedException(this, "nullable")
override lazy val resolved = false

Expand Down Expand Up @@ -443,7 +442,6 @@ case class UnresolvedExtractValue(child: Expression, extraction: Expression)
override def right: Expression = extraction

override def dataType: DataType = throw new UnresolvedException(this, "dataType")
override def foldable: Boolean = throw new UnresolvedException(this, "foldable")
override def nullable: Boolean = throw new UnresolvedException(this, "nullable")
override lazy val resolved = false

Expand Down Expand Up @@ -513,14 +511,12 @@ case class UnresolvedDeserializer(deserializer: Expression, inputAttributes: Seq

override def child: Expression = deserializer
override def dataType: DataType = throw new UnresolvedException(this, "dataType")
override def foldable: Boolean = throw new UnresolvedException(this, "foldable")
override def nullable: Boolean = throw new UnresolvedException(this, "nullable")
override lazy val resolved = false
}

case class GetColumnByOrdinal(ordinal: Int, dataType: DataType) extends LeafExpression
with Unevaluable with NonSQLExpression {
override def foldable: Boolean = throw new UnresolvedException(this, "foldable")
override def nullable: Boolean = throw new UnresolvedException(this, "nullable")
override lazy val resolved = false
}
Expand All @@ -538,7 +534,6 @@ case class GetColumnByOrdinal(ordinal: Int, dataType: DataType) extends LeafExpr
case class UnresolvedOrdinal(ordinal: Int)
extends LeafExpression with Unevaluable with NonSQLExpression {
override def dataType: DataType = throw new UnresolvedException(this, "dataType")
override def foldable: Boolean = throw new UnresolvedException(this, "foldable")
override def nullable: Boolean = throw new UnresolvedException(this, "nullable")
override lazy val resolved = false
}
Expand Down
4 changes: 3 additions & 1 deletion sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/Expression.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -298,6 +298,9 @@ abstract class Expression extends TreeNode[Expression] {
*/
trait Unevaluable extends Expression {

/** Unevaluable is not foldable because we don't have an eval for it. */
final override def foldable: Boolean = false

final override def eval(input: InternalRow = null): Any =
throw new UnsupportedOperationException(s"Cannot evaluate expression: $this")

Expand All @@ -318,7 +321,6 @@ trait Unevaluable extends Expression {
*/
trait RuntimeReplaceable extends UnaryExpression with Unevaluable {
override def nullable: Boolean = child.nullable
override def foldable: Boolean = child.foldable
Copy link
Contributor

@rednaxelafx rednaxelafx Sep 18, 2020

Choose a reason for hiding this comment

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

This change is not what you'd actually want.
I'm actually really curious why RuntimeReplaceable has to implement Unevaluable in the first place. It doesn't really make sense. We could just turn RuntimeReplaceable into a proper wrapper that delegates everything, right?

Copy link
Contributor

@cloud-fan cloud-fan Sep 21, 2020

Choose a reason for hiding this comment

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

This change is noop in practice, because we replace RuntimeReplaceable at the beginning of the optimizer, so foldable or not doesn't really matter.

Semantically, it might be clearer to just make RuntimeReplaceable a pure delegator like Alias, but it complicates the expression tree. Anyway, this is out of the scope of this PR.

Copy link
Contributor

@rednaxelafx rednaxelafx Sep 21, 2020

Choose a reason for hiding this comment

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

but it complicates the expression tree

Why would it complicate the expression tree? There's a TaggingExpression trait (https://github.com/apache/spark/blob/master/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/constraintExpressions.scala#L24) that'd handle this perfectly. Making RuntimeReplaceable implement TaggingExpression instead of Unevaluable makes much more sense to me. It'll still retain the same semantics of being replaced early.

I agree that we can do it in a separate PR, but I'd really like to make sure the refactoring done while we're at it.

Copy link
Contributor

@cloud-fan cloud-fan Sep 22, 2020

Choose a reason for hiding this comment

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

Ah TaggingExpression sounds like a good idea, we can try it later.

override def dataType: DataType = child.dataType
// As this expression gets replaced at optimization with its `child" expression,
// two `RuntimeReplaceable` are considered to be semantically equal if their "child" expressions
Expand Down
3 changes: 0 additions & 3 deletions sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/SortOrder.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -66,9 +66,6 @@ case class SortOrder(
sameOrderExpressions: Set[Expression])
extends UnaryExpression with Unevaluable {

/** Sort order is not foldable because we don't have an eval for it. */
override def foldable: Boolean = false

override def checkInputDataTypes(): TypeCheckResult = {
if (RowOrdering.isOrderable(dataType)) {
TypeCheckResult.TypeCheckSuccess
Expand Down
12 changes: 8 additions & 4 deletions ...alyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/interfaces.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,7 @@ package org.apache.spark.sql.catalyst.expressions.aggregate
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.analysis.UnresolvedAttribute
import org.apache.spark.sql.catalyst.expressions._
import org.apache.spark.sql.catalyst.expressions.codegen.CodegenFallback
import org.apache.spark.sql.catalyst.expressions.codegen.{CodegenContext, CodegenFallback, ExprCode}
import org.apache.spark.sql.types._

/** The mode of an [[AggregateFunction]]. */
Expand Down Expand Up @@ -133,7 +133,6 @@ case class AggregateExpression(
override def children: Seq[Expression] = aggregateFunction +: filter.toSeq

override def dataType: DataType = aggregateFunction.dataType
override def foldable: Boolean = false
override def nullable: Boolean = aggregateFunction.nullable

@transient
Expand Down Expand Up @@ -374,8 +373,7 @@ abstract class ImperativeAggregate extends AggregateFunction with CodegenFallbac
*/
abstract class DeclarativeAggregate
extends AggregateFunction
with Serializable
with Unevaluable {
with Serializable {

/**
* Expressions for initializing empty aggregation buffers.
Expand Down Expand Up @@ -421,6 +419,12 @@ abstract class DeclarativeAggregate
/** Represents this attribute at the input buffer side (the data value is read-only). */
def right: AttributeReference = inputAggBufferAttributes(aggBufferAttributes.indexOf(a))
}

final override def eval(input: InternalRow = null): Any =
throw new UnsupportedOperationException(s"Cannot evaluate expression: $this")

final override protected def doGenCode(ctx: CodegenContext, ev: ExprCode): ExprCode =
throw new UnsupportedOperationException(s"Cannot generate code for expression: $this")
}


Expand Down
1 change: 0 additions & 1 deletion ...atalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/complexTypeCreator.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -301,7 +301,6 @@ case class MapFromArrays(left: Expression, right: Expression)
*/
case object NamePlaceholder extends LeafExpression with Unevaluable {
override lazy val resolved: Boolean = false
override def foldable: Boolean = false
override def nullable: Boolean = false
override def dataType: DataType = StringType
override def prettyName: String = "NamePlaceholder"
Expand Down
2 changes: 0 additions & 2 deletions sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/misc.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -111,7 +111,6 @@ case class AssertTrue(child: Expression) extends UnaryExpression with ImplicitCa
""")
case class CurrentDatabase() extends LeafExpression with Unevaluable {
override def dataType: DataType = StringType
override def foldable: Boolean = true
override def nullable: Boolean = false
override def prettyName: String = "current_database"
}
Expand All @@ -129,7 +128,6 @@ case class CurrentDatabase() extends LeafExpression with Unevaluable {
since = "3.1.0")
case class CurrentCatalog() extends LeafExpression with Unevaluable {
override def dataType: DataType = StringType
override def foldable: Boolean = true
override def nullable: Boolean = false
override def prettyName: String = "current_catalog"
}
Expand Down
1 change: 0 additions & 1 deletion sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/predicates.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -327,7 +327,6 @@ case class InSubquery(values: Seq[Expression], query: ListQuery)

override def children: Seq[Expression] = values :+ query
override def nullable: Boolean = children.exists(_.nullable)
override def foldable: Boolean = children.forall(_.foldable)
Copy link
Contributor

@rednaxelafx rednaxelafx Sep 18, 2020

Choose a reason for hiding this comment

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

Let's have some more confirmation of why it was written this way originally, and whether or not we're losing anything by just making folable always false here.

Copy link
Contributor

@cloud-fan cloud-fan Sep 21, 2020

Choose a reason for hiding this comment

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

seems there was no specific reason: #21403 . We just followed In at that time.

It's obviously not foldable as it needs to evaluate a subquery.

Copy link
Contributor

@rednaxelafx rednaxelafx Sep 21, 2020

Choose a reason for hiding this comment

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

OK +1 with @cloud-fan 's comment.

override def toString: String = s"$value IN ($query)"
override def sql: String = s"(${value.sql} IN (${query.sql}))"
}
Expand Down
10 changes: 4 additions & 6 deletions ...catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/windowExpressions.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -50,7 +50,6 @@ case class WindowSpecDefinition(
frameSpecification.isInstanceOf[SpecifiedWindowFrame]

override def nullable: Boolean = true
override def foldable: Boolean = false
override def dataType: DataType = throw new UnsupportedOperationException("dataType")

override def checkInputDataTypes(): TypeCheckResult = {
Expand Down Expand Up @@ -144,7 +143,6 @@ case object RangeFrame extends FrameType {
sealed trait SpecialFrameBoundary extends Expression with Unevaluable {
override def children: Seq[Expression] = Nil
override def dataType: DataType = NullType
override def foldable: Boolean = false
override def nullable: Boolean = false
}

Expand All @@ -168,7 +166,6 @@ case object CurrentRow extends SpecialFrameBoundary {
sealed trait WindowFrame extends Expression with Unevaluable {
override def children: Seq[Expression] = Nil
override def dataType: DataType = throw new UnsupportedOperationException("dataType")
override def foldable: Boolean = false
override def nullable: Boolean = false
}

Expand Down Expand Up @@ -275,7 +272,6 @@ case class UnresolvedWindowExpression(
windowSpec: WindowSpecReference) extends UnaryExpression with Unevaluable {

override def dataType: DataType = throw new UnresolvedException(this, "dataType")
override def foldable: Boolean = throw new UnresolvedException(this, "foldable")
override def nullable: Boolean = throw new UnresolvedException(this, "nullable")
override lazy val resolved = false
}
Expand All @@ -287,7 +283,6 @@ case class WindowExpression(
override def children: Seq[Expression] = windowFunction :: windowSpec :: Nil

override def dataType: DataType = windowFunction.dataType
override def foldable: Boolean = windowFunction.foldable
override def nullable: Boolean = windowFunction.nullable

override def toString: String = s"$windowFunction $windowSpec"
Expand Down Expand Up @@ -370,8 +365,11 @@ abstract class OffsetWindowFunction
* OffsetWindowFunction is executed, the input expression and the default expression. Even when
* both the input and the default expression are foldable, the result is still not foldable due to
* the frame.
*
* Note, the value of foldable is set to false in the trait Unevaluable
*
* override def foldable: Boolean = false
*/
override def foldable: Boolean = false

override def nullable: Boolean = default == null || default.nullable || input.nullable

Expand Down
2 changes: 0 additions & 2 deletions sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/v2Commands.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -348,7 +348,6 @@ case class MergeIntoTable(

sealed abstract class MergeAction extends Expression with Unevaluable {
def condition: Option[Expression]
override def foldable: Boolean = false
override def nullable: Boolean = false
override def dataType: DataType = throw new UnresolvedException(this, "nullable")
override def children: Seq[Expression] = condition.toSeq
Expand All @@ -369,7 +368,6 @@ case class InsertAction(
}

case class Assignment(key: Expression, value: Expression) extends Expression with Unevaluable {
override def foldable: Boolean = false
override def nullable: Boolean = false
override def dataType: DataType = throw new UnresolvedException(this, "nullable")
override def children: Seq[Expression] = key :: value :: Nil
Expand Down