Skip to content

[SPARK-32858][SQL] UnwrapCastInBinaryComparison: support other numeric types #29792

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 5 commits into from
Closed

[SPARK-32858][SQL] UnwrapCastInBinaryComparison: support other numeric types #29792

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
65833e7
[SPARK-32858][SQL] UnwrapCastInBinaryComparison: support other numeri…
sunchao Sep 14, 2020
3fcc396
More test coverage
sunchao Sep 17, 2020
3340de4
Minor refactoring
sunchao Oct 1, 2020
94942bf
Add end-to-end test suite
sunchao Oct 6, 2020
76b9f73
Address comments
sunchao Oct 12, 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
202 changes: 126 additions & 76 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 @@ -35,18 +35,32 @@ import org.apache.spark.sql.types._
* to be optimized away later and pushed down to data sources.
*
* Currently this only handles cases where:
* 1). `fromType` (of `fromExp`) and `toType` are of integral types (i.e., byte, short, int and
* long)
* 1). `fromType` (of `fromExp`) and `toType` are of numeric types (i.e., short, int, float,
* decimal, etc)
* 2). `fromType` can be safely coerced to `toType` without precision loss (e.g., short to int,
* int to long, but not long to int)
*
* If the above conditions are satisfied, the rule checks to see if the literal `value` is within
* range `(min, max)`, where `min` and `max` are the minimum and maximum value of `fromType`,
* respectively. If this is true then it means we can safely cast `value` to `fromType` and thus
* respectively. If this is true then it means we may safely cast `value` to `fromType` and thus
* able to move the cast to the literal side. That is:
*
* `cast(fromExp, toType) op value` ==> `fromExp op cast(value, fromType)`
*
* Note there are some exceptions to the above: if casting from `value` to `fromType` causes
* rounding up or down, the above conversion will no longer be valid. Instead, the rule does the
* following:
*
* if casting `value` to `fromType` causes rounding up:
* - `cast(fromExp, toType) > value` ==> `fromExp >= cast(value, fromType)`
* - `cast(fromExp, toType) >= value` ==> `fromExp >= cast(value, fromType)`
* - `cast(fromExp, toType) === value` ==> if(isnull(fromExp), null, false)
* - `cast(fromExp, toType) <=> value` ==> false (if `fromExp` is deterministic)
* - `cast(fromExp, toType) <= value` ==> `fromExp < cast(value, fromType)`
* - `cast(fromExp, toType) < value` ==> `fromExp < cast(value, fromType)`
*
* Similarly for the case when casting `value` to `fromType` causes rounding down.

@maropu maropu Oct 9, 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.

nit: wrong indent.

*
* If the `value` is not within range `(min, max)`, the rule breaks the scenario into different
* cases and try to replace each with simpler constructs.
*
Expand All @@ -55,8 +69,6 @@ import org.apache.spark.sql.types._
* - `cast(fromExp, toType) >= value` ==> if(isnull(fromExp), null, false)
* - `cast(fromExp, toType) === value` ==> if(isnull(fromExp), null, false)
* - `cast(fromExp, toType) <=> value` ==> false (if `fromExp` is deterministic)
* - `cast(fromExp, toType) <=> value` ==> cast(fromExp, toType) <=> value (if `fromExp` is
* non-deterministic)
* - `cast(fromExp, toType) <= value` ==> if(isnull(fromExp), null, true)
* - `cast(fromExp, toType) < value` ==> if(isnull(fromExp), null, true)
*
Expand Down Expand Up @@ -100,12 +112,12 @@ object UnwrapCastInBinaryComparison extends Rule[LogicalPlan] {

swap(unwrapCast(swap(exp)))

// In case both sides have integral type, optimize the comparison by removing casts or
// In case both sides have numeric 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: NumericType, _), Literal(value, literalType))
if canImplicitlyCast(fromExp, toType, literalType) =>
simplifyIntegralComparison(be, fromExp, toType, value)
simplifyNumericComparison(be, fromExp, toType, value)

case _ => exp
}
Expand All @@ -116,82 +128,118 @@ object UnwrapCastInBinaryComparison extends Rule[LogicalPlan] {
* optimizes the expression by moving the cast to the literal side. Otherwise if result is not
* true, this replaces the input binary comparison `exp` with simpler expressions.
*/
private def simplifyIntegralComparison(
private def simplifyNumericComparison(
exp: BinaryComparison,
fromExp: Expression,
toType: IntegralType,
toType: NumericType,
value: Any): Expression = {

val fromType = fromExp.dataType
val (min, max) = getRange(fromType)
val (minInToType, maxInToType) = {
(Cast(Literal(min), toType).eval(), Cast(Literal(max), toType).eval())
}
val ordering = toType.ordering.asInstanceOf[Ordering[Any]]
val minCmp = ordering.compare(value, minInToType)
val maxCmp = ordering.compare(value, maxInToType)
val range = getRange(fromType)

if (maxCmp > 0) {
exp match {
case EqualTo(_, _) | GreaterThan(_, _) | GreaterThanOrEqual(_, _) =>
falseIfNotNull(fromExp)
case LessThan(_, _) | LessThanOrEqual(_, _) =>
trueIfNotNull(fromExp)
// make sure the expression is evaluated if it is non-deterministic
case EqualNullSafe(_, _) if exp.deterministic =>
FalseLiteral
case _ => exp
if (range.isDefined) {
val (min, max) = range.get
val (minInToType, maxInToType) = {
(Cast(Literal(min), toType).eval(), Cast(Literal(max), toType).eval())
}
} else if (maxCmp == 0) {
exp match {
case GreaterThan(_, _) =>
falseIfNotNull(fromExp)
case LessThanOrEqual(_, _) =>
trueIfNotNull(fromExp)
case LessThan(_, _) =>
Not(EqualTo(fromExp, Literal(max, fromType)))
case GreaterThanOrEqual(_, _) | EqualTo(_, _) =>
EqualTo(fromExp, Literal(max, fromType))
case EqualNullSafe(_, _) =>
EqualNullSafe(fromExp, Literal(max, fromType))
case _ => exp
val minCmp = ordering.compare(value, minInToType)
val maxCmp = ordering.compare(value, maxInToType)

if (maxCmp >= 0 || minCmp <= 0) {
return if (maxCmp > 0) {
exp match {
case EqualTo(_, _) | GreaterThan(_, _) | GreaterThanOrEqual(_, _) =>
falseIfNotNull(fromExp)
case LessThan(_, _) | LessThanOrEqual(_, _) =>
trueIfNotNull(fromExp)
// make sure the expression is evaluated if it is non-deterministic
case EqualNullSafe(_, _) if exp.deterministic =>
FalseLiteral
case _ => exp
}
} else if (maxCmp == 0) {
exp match {
case GreaterThan(_, _) =>
falseIfNotNull(fromExp)
case LessThanOrEqual(_, _) =>
trueIfNotNull(fromExp)
case LessThan(_, _) =>
Not(EqualTo(fromExp, Literal(max, fromType)))
case GreaterThanOrEqual(_, _) | EqualTo(_, _) =>
EqualTo(fromExp, Literal(max, fromType))
case EqualNullSafe(_, _) =>
EqualNullSafe(fromExp, Literal(max, fromType))
case _ => exp
}
} else if (minCmp < 0) {
exp match {
case GreaterThan(_, _) | GreaterThanOrEqual(_, _) =>
trueIfNotNull(fromExp)
case LessThan(_, _) | LessThanOrEqual(_, _) | EqualTo(_, _) =>
falseIfNotNull(fromExp)
// make sure the expression is evaluated if it is non-deterministic
case EqualNullSafe(_, _) if exp.deterministic =>
FalseLiteral
case _ => exp
}
} else { // minCmp == 0
exp match {
case LessThan(_, _) =>
falseIfNotNull(fromExp)
case GreaterThanOrEqual(_, _) =>
trueIfNotNull(fromExp)
case GreaterThan(_, _) =>
Not(EqualTo(fromExp, Literal(min, fromType)))
case LessThanOrEqual(_, _) | EqualTo(_, _) =>
EqualTo(fromExp, Literal(min, fromType))
case EqualNullSafe(_, _) =>
EqualNullSafe(fromExp, Literal(min, fromType))
case _ => exp
}
}
}
} else if (minCmp < 0) {
}

// When we reach to this point, it means either there is no min/max for the `fromType` (e.g.,
// decimal type), or that the literal `value` is within range `(min, max)`. For these, we

@cloud-fan cloud-fan Sep 29, 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.

why it's safe to skip range check for decimal type?

@sunchao sunchao Sep 29, 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.

It is safe since knowing min/max for a type just gives us more opportunity for optimizations. I skipped decimal type here because (it seems) there is no min/max defined in the DecimalType, unlike other numeric types.

@cloud-fan cloud-fan Oct 6, 2020 •
edited
Loading

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.

makes sense.

// optimize by moving the cast to the literal side.

val newValue = Cast(Literal(value), fromType).eval()
if (newValue == null) {
// This means the cast failed, for instance, due to the value is not representable in the
// narrower type. In this case we simply return the original expression.

@cloud-fan cloud-fan Oct 6, 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.

can you give a real example here?

@cloud-fan cloud-fan Oct 6, 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.

I see, it's for decimal only. It's better to make the comment more explicit.

@sunchao sunchao Oct 6, 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.

yup will do - there is also a test case covering this.

return exp
}
val valueRoundTrip = Cast(Literal(newValue, fromType), toType).eval()

@cloud-fan cloud-fan Sep 29, 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.

The case I'm worried about is cast(float_col as double) cmp double_lit. It's not straightforward to me that a double -> float -> double roundtrip can tell rounding up or down. is it because float -> double can only be rounding up?

@sunchao sunchao Sep 29, 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.

So double to float can result to either rounding up or down. For instance, by casting 3.14 in double to float, even though the value is still 3.14, the binary representation is rounded up:

3.14 in double:

0 10000000000 1001 0001 1110 1011 1000 0101 0001 1110 1011 1000 0101 0001 1111

3.14 in float

0 10000000 1001 0001 1110 1011 1000 011

Here the sign bit and exponent bits (11 and 8 bits respectively for double and float) are the same for both float and double. However, in the fraction part, the last is rounded up to 1.

After casting back to double, there won't be any rounding up or down - the remaining digits are simply padded with 0:

0 10000000000 1001 0001 1110 1011 1000 0110 0000000000000000000000000000

@dbtsai dbtsai Sep 29, 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.

Is it defined as part of IEEE Standard for Floating-Point Arithmetic (IEEE 754)?

@sunchao sunchao Sep 29, 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.

Yes I think both the binary format as well as rounding rules are specified in IEEE 754. There are a few rounding rules and I think the default one is "rounding to half even".

val lit = Literal(newValue, fromType)
val cmp = ordering.compare(value, valueRoundTrip)
if (cmp == 0) {
exp match {
case GreaterThan(_, _) | GreaterThanOrEqual(_, _) =>
trueIfNotNull(fromExp)
case LessThan(_, _) | LessThanOrEqual(_, _) | EqualTo(_, _) =>
falseIfNotNull(fromExp)
// make sure the expression is evaluated if it is non-deterministic
case EqualNullSafe(_, _) if exp.deterministic =>
FalseLiteral
case GreaterThan(_, _) => GreaterThan(fromExp, lit)
case GreaterThanOrEqual(_, _) => GreaterThanOrEqual(fromExp, lit)
case EqualTo(_, _) => EqualTo(fromExp, lit)
case EqualNullSafe(_, _) => EqualNullSafe(fromExp, lit)
case LessThan(_, _) => LessThan(fromExp, lit)
case LessThanOrEqual(_, _) => LessThanOrEqual(fromExp, lit)
case _ => exp
}
} else if (minCmp == 0) {
} else if (cmp < 0) {
// This means the literal value is rounded up after casting to `fromType`
exp match {
case LessThan(_, _) =>
falseIfNotNull(fromExp)
case GreaterThanOrEqual(_, _) =>
trueIfNotNull(fromExp)
case GreaterThan(_, _) =>
Not(EqualTo(fromExp, Literal(min, fromType)))
case LessThanOrEqual(_, _) | EqualTo(_, _) =>
EqualTo(fromExp, Literal(min, fromType))
case EqualNullSafe(_, _) =>
EqualNullSafe(fromExp, Literal(min, fromType))
case EqualTo(_, _) => falseIfNotNull(fromExp)
case EqualNullSafe(_, _) if fromExp.deterministic => FalseLiteral
case GreaterThan(_, _) | GreaterThanOrEqual(_, _) => GreaterThanOrEqual(fromExp, lit)
case LessThan(_, _) | LessThanOrEqual(_, _) => LessThan(fromExp, lit)
case _ => exp
}
} else {
// This means `value` is within range `(min, max)`. Optimize this by moving the cast to the
// literal side.
val lit = Literal(Cast(Literal(value), fromType).eval(), fromType)
// This means the literal value is rounded down after casting to `fromType`
exp match {
case GreaterThan(_, _) => GreaterThan(fromExp, lit)
case GreaterThanOrEqual(_, _) => GreaterThanOrEqual(fromExp, lit)
case EqualTo(_, _) => EqualTo(fromExp, lit)
case EqualNullSafe(_, _) => EqualNullSafe(fromExp, lit)
case LessThan(_, _) => LessThan(fromExp, lit)
case LessThanOrEqual(_, _) => LessThanOrEqual(fromExp, lit)
case EqualTo(_, _) => falseIfNotNull(fromExp)
case EqualNullSafe(_, _) => FalseLiteral
case GreaterThan(_, _) | GreaterThanOrEqual(_, _) => GreaterThan(fromExp, lit)
case LessThan(_, _) | LessThanOrEqual(_, _) => LessThanOrEqual(fromExp, lit)
case _ => exp
}
}
Expand All @@ -200,25 +248,27 @@ object UnwrapCastInBinaryComparison extends Rule[LogicalPlan] {
/**
* Check if the input `fromExp` can be safely cast to `toType` without any loss of precision,
* i.e., the conversion is injective. Note this only handles the case when both sides are of
* integral type.
* numeric type.
*/
private def canImplicitlyCast(
fromExp: Expression,
toType: DataType,
literalType: DataType): Boolean = {
toType.sameType(literalType) &&
!fromExp.foldable &&
fromExp.dataType.isInstanceOf[IntegralType] &&
toType.isInstanceOf[IntegralType] &&
fromExp.dataType.isInstanceOf[NumericType] &&
toType.isInstanceOf[NumericType] &&
Cast.canUpCast(fromExp.dataType, toType)
}

private def getRange(dt: DataType): (Any, Any) = dt match {
case ByteType => (Byte.MinValue, Byte.MaxValue)
case ShortType => (Short.MinValue, Short.MaxValue)
case IntegerType => (Int.MinValue, Int.MaxValue)
case LongType => (Long.MinValue, Long.MaxValue)
case other => throw new IllegalArgumentException(s"Unsupported type: ${other.catalogString}")
private def getRange(dt: DataType): Option[(Any, Any)] = dt match {
case ByteType => Some((Byte.MinValue, Byte.MaxValue))
case ShortType => Some((Short.MinValue, Short.MaxValue))
case IntegerType => Some((Int.MinValue, Int.MaxValue))
case LongType => Some((Long.MinValue, Long.MaxValue))
case FloatType => Some((Float.NegativeInfinity, Float.NaN))

@cloud-fan cloud-fan Sep 29, 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.

why the upper bound is not PositiveInfinity?

@sunchao sunchao Sep 29, 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.

This is because PositiveInfinity is considered to be < NaN in Spark. If we treat it as the upper bound, rules handling the upper bounds will not be valid. For instance the following expr:

cast(e as double) > double('+inf')

would be converted to

e === double('+inf')

which won't be correct if e evaluates to double('NaN').

case DoubleType => Some((Double.NegativeInfinity, Double.NaN))

@maropu maropu Oct 9, 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.

Looks it does not have any test for this code path, so could you add some tests for it. (NOTE: I think byte, int, and long are not tested in UnwrapCastInBinaryComparisonSuite, too)

@sunchao sunchao Oct 9, 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.

Will add a test case (although I think it will be pretty trivial). I only added tests for short in the previous PR because the handling for other integral types is exactly the same.

case _ => None
}

/**
Expand Down
Loading