Skip to content

[SPARK-24994][SQL][FOLLOW-UP] Handle foldable, timezone and cleanup #29775

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
sunchao wants to merge 2 commits into from
Closed

[SPARK-24994][SQL][FOLLOW-UP] Handle foldable, timezone and cleanup #29775

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
4291ec2
[SPARK-24994][SQL][FOLLOW-UP] Handle foldable, timezone and cleanup
sunchao Sep 16, 2020
3a0f4b0
Replace cast with literal value
sunchao Sep 17, 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
14 changes: 9 additions & 5 deletions ...src/main/scala/org/apache/spark/sql/catalyst/optimizer/UnwrapCastInBinaryComparison.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -103,9 +103,9 @@ object UnwrapCastInBinaryComparison extends Rule[LogicalPlan] {
// In case both sides have integral type, optimize the comparison by removing casts or
// moving cast to the literal side.
case be @ BinaryComparison(
Cast(fromExp, toType: IntegralType, _), Literal(value, literalType))
Cast(fromExp, toType: IntegralType, tz), Literal(value, literalType))
if canImplicitlyCast(fromExp, toType, literalType) =>
simplifyIntegralComparison(be, fromExp, toType, value)
simplifyIntegralComparison(be, fromExp, toType, value, tz)

@dongjoon-hyun dongjoon-hyun Sep 16, 2020 •
edited
Loading

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

This looks irrelevant as a follow-up.
Is there a comment on this? pass timezone info to the generated cast on the literal value?

@dongjoon-hyun dongjoon-hyun Sep 16, 2020

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

It's hard for me to think a valid test case for this in the scope of SPARK-24994 (Integral Types). Although Spark handles some time-related types with the integral types, I believe we need to revisit this on new PR which aims to support time.

@sunchao sunchao Sep 16, 2020

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

Thanks @dongjoon-hyun . I discovered this in tests - without this some of the existing tests will fail because of timezone mismatch (although I think this shouldn't affect correctness given we only handle integral types?).

In the follow-up, the current plan is to only extend to numeric types (e.g., float/double/decimal). I don't know whether support for time-related types is important or not.

@dongjoon-hyun dongjoon-hyun Sep 16, 2020

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

Could you give me some pointer for the failures? Or, can we have a reproducer? Usually, every code change had better have a test coverage. If there is a test case for this, it sounds reasonable.

@sunchao sunchao Sep 17, 2020

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

（oops just found out my comment was not sent out successfully)

This is because ResolveTimeZone will try to add timezone info to all expressions that don't have it during query analysis. However, since the Cast expr was generated at optimization phase, it will not have the timezone info. As result, PlanTest.comparePlans will fail because of mismatch. I can try to come up with a test if necessary.

On the other hand, I think instead of using Cast, we may just directly use the value, since the Cast will be optimized away by ConstantFolding anyways later. What do you think?

@cloud-fan cloud-fan Sep 17, 2020

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

yea this also works: just evaluate the cast and make a literal.

@sunchao sunchao Sep 17, 2020

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

Thanks. I'll do that then. I'm not totally sure aboutCast.canonicalize - somehow it is not used in query analysis and optimization and the lazy val is not initialized before the plan comparison.


case _ => exp
}
Expand All @@ -120,7 +120,8 @@ object UnwrapCastInBinaryComparison extends Rule[LogicalPlan] {
exp: BinaryComparison,
fromExp: Expression,
toType: IntegralType,
value: Any): Expression = {
value: Any,
tz: Option[String]): Expression = {

val fromType = fromExp.dataType
val (min, max) = getRange(fromType)
Expand Down Expand Up @@ -184,7 +185,7 @@ object UnwrapCastInBinaryComparison extends Rule[LogicalPlan] {
} else {
// This means `value` is within range `(min, max)`. Optimize this by moving the cast to the
// literal side.
val lit = Cast(Literal(value), fromType)
val lit = Cast(Literal(value), fromType, tz)

@sunchao sunchao Sep 16, 2020

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

We can also evaluate this and pass the value. It will be optimized away by ConstantFolding anyway.

exp match {
case GreaterThan(_, _) => GreaterThan(fromExp, lit)
case GreaterThanOrEqual(_, _) => GreaterThanOrEqual(fromExp, lit)
Expand All @@ -202,9 +203,12 @@ object UnwrapCastInBinaryComparison extends Rule[LogicalPlan] {
* i.e., the conversion is injective. Note this only handles the case when both sides are of
* integral type.
*/
private def canImplicitlyCast(fromExp: Expression, toType: DataType,
private def canImplicitlyCast(
fromExp: Expression,
toType: DataType,
literalType: DataType): Boolean = {
toType.sameType(literalType) &&
!fromExp.foldable &&
fromExp.dataType.isInstanceOf[IntegralType] &&
toType.isInstanceOf[IntegralType] &&
Cast.canUpCast(fromExp.dataType, toType)
Expand Down
24 changes: 14 additions & 10 deletions ...est/scala/org/apache/spark/sql/catalyst/optimizer/UnwrapCastInBinaryComparisonSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,14 +26,14 @@ import org.apache.spark.sql.catalyst.optimizer.UnwrapCastInBinaryComparison._
import org.apache.spark.sql.catalyst.plans.PlanTest
import org.apache.spark.sql.catalyst.plans.logical._
import org.apache.spark.sql.catalyst.rules.RuleExecutor
import org.apache.spark.sql.types.{BooleanType, ByteType, DoubleType, IntegerType}
import org.apache.spark.sql.types._

class UnwrapCastInBinaryComparisonSuite extends PlanTest with ExpressionEvalHelper {

object Optimize extends RuleExecutor[LogicalPlan] {
val batches: List[Batch] =
Batch("Unwrap casts in binary comparison", FixedPoint(10),
NullPropagation, ConstantFolding, UnwrapCastInBinaryComparison) :: Nil
NullPropagation, UnwrapCastInBinaryComparison) :: Nil
}

val testRelation: LocalRelation = LocalRelation('a.short, 'b.float)
Expand Down Expand Up @@ -80,12 +80,12 @@ class UnwrapCastInBinaryComparisonSuite extends PlanTest with ExpressionEvalHelp
}

test("unwrap casts when literal is within range (min, max)") {
assertEquivalent(castInt(f) > 300, f > 300.toShort)
assertEquivalent(castInt(f) >= 500, f >= 500.toShort)
assertEquivalent(castInt(f) === 32766, f === 32766.toShort)
assertEquivalent(castInt(f) <=> 32766, f <=> 32766.toShort)
assertEquivalent(castInt(f) <= -6000, f <= -6000.toShort)
assertEquivalent(castInt(f) < -32767, f < -32767.toShort)
assertEquivalent(castInt(f) > 300, f > castShort(300))
assertEquivalent(castInt(f) >= 500, f >= castShort(500))
assertEquivalent(castInt(f) === 32766, f === castShort(32766))
assertEquivalent(castInt(f) <=> 32766, f <=> castShort(32766))
assertEquivalent(castInt(f) <= -6000, f <= castShort(-6000))
assertEquivalent(castInt(f) < -32767, f < castShort(-32767))
}

test("unwrap casts when cast is on rhs") {
Expand All @@ -97,7 +97,7 @@ class UnwrapCastInBinaryComparisonSuite extends PlanTest with ExpressionEvalHelp
assertEquivalent(Literal(v.toInt) >= castInt(f), trueIfNotNull(f))
assertEquivalent(Literal(v.toInt) > castInt(f), f =!= v)

assertEquivalent(Literal(30) <= castInt(f), Literal(30.toShort) <= f)
assertEquivalent(Literal(30) <= castInt(f), Cast(Literal(30), ShortType) <= f)
}

test("unwrap cast should have no effect when input is not integral type") {
Expand All @@ -119,10 +119,12 @@ class UnwrapCastInBinaryComparisonSuite extends PlanTest with ExpressionEvalHelp
)
}

test("unwrap cast should skip when expression is non-deterministic") {
test("unwrap cast should skip when expression is non-deterministic or foldable") {
Seq(positiveInt, negativeInt).foreach (v => {
val e = Cast(First(f, ignoreNulls = true), IntegerType) <=> v
assertEquivalent(e, e, evaluate = false)
val e2 = Cast(Literal(30.toShort), IntegerType) >= v
assertEquivalent(e2, e2, evaluate = false)
})
}

Expand All @@ -143,6 +145,8 @@ class UnwrapCastInBinaryComparisonSuite extends PlanTest with ExpressionEvalHelp

private def castInt(e: Expression): Expression = Cast(e, IntegerType)

private def castShort(e: Expression): Expression = Cast(e, ShortType)

private def castDouble(e: Expression): Expression = Cast(e, DoubleType)

private def assertEquivalent(e1: Expression, e2: Expression, evaluate: Boolean = true): Unit = {
Expand Down
5 changes: 2 additions & 3 deletions sql/core/src/test/scala/org/apache/spark/sql/FileBasedDataSourceSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,14 +32,13 @@ import org.apache.spark.scheduler.{SparkListener, SparkListenerTaskEnd}
import org.apache.spark.sql.TestingUDT.{IntervalUDT, NullData, NullUDT}
import org.apache.spark.sql.catalyst.expressions.AttributeReference
import org.apache.spark.sql.catalyst.expressions.IntegralLiteralTestUtils.{negativeInt, positiveInt}
import org.apache.spark.sql.catalyst.planning.PhysicalOperation
import org.apache.spark.sql.catalyst.plans.logical.Filter
import org.apache.spark.sql.execution.SimpleMode
import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper
import org.apache.spark.sql.execution.datasources.FilePartition
import org.apache.spark.sql.execution.datasources.v2.{BatchScanExec, DataSourceV2ScanRelation, FileScan}
import org.apache.spark.sql.execution.datasources.v2.{BatchScanExec, FileScan}
import org.apache.spark.sql.execution.datasources.v2.orc.OrcScan
import org.apache.spark.sql.execution.datasources.v2.parquet.{ParquetScan, ParquetTable}
import org.apache.spark.sql.execution.datasources.v2.parquet.ParquetScan
import org.apache.spark.sql.execution.joins.{BroadcastHashJoinExec, SortMergeJoinExec}
import org.apache.spark.sql.functions._
import org.apache.spark.sql.internal.SQLConf
Expand Down