wombatu-kun commented on code in PR #20532:
URL: https://github.com/apache/hudi/pull/20532#discussion_r4488368343
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/dataskipping/rules/ConstantOperandRule.scala:
##########
@@ -17,23 +17,33 @@
package org.apache.spark.sql.hudi.dataskipping.rules
-import org.apache.spark.sql.catalyst.expressions.{Add, BitwiseOr, DateAdd,
DateDiff, DateSub, Divide, Expression, Literal, Multiply, ShiftLeft, ShiftRight}
-import org.apache.spark.sql.hudi.dataskipping.{Increasing, RangeRule, Step}
-import org.apache.spark.sql.types.Decimal
+import org.apache.spark.sql.catalyst.expressions.{Add, And, BitwiseOr,
DateAdd, DateDiff, DateSub, Divide, Expression, GreaterThanOrEqual,
LessThanOrEqual, Literal, Multiply, ShiftLeft, ShiftRight}
+import org.apache.spark.sql.hudi.dataskipping.{Increasing, RangeRule, Step,
TermRange}
+import org.apache.spark.sql.types.{ByteType, DataType, Decimal, DoubleType,
FloatType, IntegerType, LongType, ShortType}
+
+import scala.util.Try
/**
* Binary operators with the column on one side and a constant on the other:
* `Add`, `Multiply`, `Divide`, `DateAdd`, `DateSub`, `DateDiff`, `ShiftLeft`,
`ShiftRight`, `BitwiseOr`.
*
* Shared guard: the other operand is constant. Today that is enforced per
function:
- * - `Add`: the other side is foldable (either side)
- * - `Multiply`: the other side is a positive numeric literal (either side)
+ * - `Add`: float / double result and a finite constant, or integral result
and a constant (either side)
+ * - `Multiply`: float / double or integral result, and the other side is a
positive numeric literal
+ * (either side)
* - `Divide`: the divisor is a positive numeric literal (column as dividend
only)
- * - `DateAdd`, `DateSub`, `DateDiff` (end only), `ShiftLeft`, `ShiftRight`,
`BitwiseOr`: ANY second
- * operand is accepted today, constant or not
+ * - `ShiftLeft`: the shift amount is constant
+ * - `DateAdd`, `DateSub`, `DateDiff` (end only), `ShiftRight`, `BitwiseOr`:
ANY second operand is
+ * accepted today, constant or not
*
* NOTE: A column-referencing other operand could reverse the ordering (for
ex, A + A * -2 = -A), and a
* non-deterministic one is re-evaluated against the min/max bounds (for
ex, A + rand())
+ *
+ * NOTE: Integral `Add`, `Multiply` and `ShiftLeft` wrap around on overflow in
non-ANSI mode, so the
+ * max can map below the min (for ex, Long.MaxValue * 2 = -2), and throw
in ANSI mode. Their
Review Comment:
`ShiftLeft` has no eval mode, so it wraps around on overflow with ANSI on as
well, and only `Add` and `Multiply` throw. Could this NOTE, the `IntegralType`
scaladoc, the ANSI comment in `testLookupThroughFullOptimizer` and the
description say so?
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/dataskipping/rules/ConstantOperandRule.scala:
##########
@@ -63,14 +89,112 @@ object ConstantOperandRule extends RangeRule {
Step(term, left, Increasing, fileGuard = None, partial = false),
Step(term, right, Increasing, fileGuard = None, partial = false))
- case ShiftLeft(value, _) => Seq(Step(term, value, Increasing, fileGuard =
None, partial = false))
+ // NOTE: Spark only shifts int and bigint values, by an int amount
+ case ShiftLeft(value, amount) =>
+ IntegralType.of(term).filter(t => t.dataType == IntegerType ||
t.dataType == LongType).toSeq.flatMap { t =>
+ integralConstant(amount).map(n => guardedStep(term, value,
t.shiftLeftFits(n)))
+ }
case ShiftRight(value, _) => Seq(Step(term, value, Increasing, fileGuard =
None, partial = false))
case _ => Nil
}
private def when(guard: Boolean)(step: => Step): Option[Step] = if (guard)
Some(step) else None
+ private def guardedStep(term: Expression, child: Expression, fileGuard:
TermRange => Expression): Step =
+ Step(term, child, Increasing, fileGuard = Some(fileGuard), partial = false)
+
+ // NOTE: Expression indexes skip the resolution check, and some unresolved
terms throw on `dataType`
+ private def hasFloatingPointResult(term: Expression): Boolean =
Try(term.dataType).toOption.exists {
+ case FloatType | DoubleType => true
+ case _ => false
+ }
+
+ // An infinite addend maps the opposite infinity to NaN, which sorts above
every number (for ex,
+ // -Infinity + Infinity = NaN > 1 + Infinity). Foldable operands evaluate
without input, which is
+ // how the optimizer folds them into literals
+ private def isFiniteConstant(expr: Expression): Boolean =
+ expr.foldable && Try(expr.eval()).toOption.exists {
+ case f: Float => !f.isInfinite && !f.isNaN
+ case d: Double => !d.isInfinite && !d.isNaN
+ case _ => false
+ }
+
+ // NOTE: Foldable operands evaluate without input, which is how the
optimizer folds them into literals
+ private def integralConstant(expr: Expression): Option[Long] =
+ if (expr.foldable) Try(expr.eval()).toOption.collect(integralValue) else
None
+
+ // NOTE: Literal only, as `isPositiveNumericLiteral`
+ private def positiveIntegralLiteral(expr: Expression): Option[Long] = expr
match {
+ case Literal(value, _) => integralValue.lift(value).filter(_ > 0)
+ case _ => None
+ }
+
+ private val integralValue: PartialFunction[Any, Long] = {
+ case b: Byte => b.toLong
+ case s: Short => s.toLong
+ case i: Int => i.toLong
+ case l: Long => l
+ }
+
+ /**
+ * An integral result type and its limits. Past them, integral arithmetic
wraps around in non-ANSI
+ * mode and throws in ANSI mode. Each guard holds when no value of the
file's range goes past them:
+ * the step is increasing, so checking the min and the max covers every
value in between. A file
+ * that may overflow gets a NULL bound and is kept, and a file that holds
the guard never evaluates
+ * an overflowing bound, so ANSI does not throw.
+ */
+ private case class IntegralType(dataType: DataType, min: Long, max: Long,
bits: Int) {
Review Comment:
`IntegralType` is also the name of Spark's
`org.apache.spark.sql.types.IntegralType`, the parent of exactly these four
types, while this class holds a type's limits rather than being one. Could it
be named after what it holds, for ex `IntegralLimits`?
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/dataskipping/rules/ConstantOperandRule.scala:
##########
@@ -17,23 +17,33 @@
package org.apache.spark.sql.hudi.dataskipping.rules
-import org.apache.spark.sql.catalyst.expressions.{Add, BitwiseOr, DateAdd,
DateDiff, DateSub, Divide, Expression, Literal, Multiply, ShiftLeft, ShiftRight}
-import org.apache.spark.sql.hudi.dataskipping.{Increasing, RangeRule, Step}
-import org.apache.spark.sql.types.Decimal
+import org.apache.spark.sql.catalyst.expressions.{Add, And, BitwiseOr,
DateAdd, DateDiff, DateSub, Divide, Expression, GreaterThanOrEqual,
LessThanOrEqual, Literal, Multiply, ShiftLeft, ShiftRight}
+import org.apache.spark.sql.hudi.dataskipping.{Increasing, RangeRule, Step,
TermRange}
+import org.apache.spark.sql.types.{ByteType, DataType, Decimal, DoubleType,
FloatType, IntegerType, LongType, ShortType}
+
+import scala.util.Try
/**
* Binary operators with the column on one side and a constant on the other:
* `Add`, `Multiply`, `Divide`, `DateAdd`, `DateSub`, `DateDiff`, `ShiftLeft`,
`ShiftRight`, `BitwiseOr`.
*
* Shared guard: the other operand is constant. Today that is enforced per
function:
- * - `Add`: the other side is foldable (either side)
- * - `Multiply`: the other side is a positive numeric literal (either side)
+ * - `Add`: float / double result and a finite constant, or integral result
and a constant (either side)
+ * - `Multiply`: float / double or integral result, and the other side is a
positive numeric literal
+ * (either side)
* - `Divide`: the divisor is a positive numeric literal (column as dividend
only)
- * - `DateAdd`, `DateSub`, `DateDiff` (end only), `ShiftLeft`, `ShiftRight`,
`BitwiseOr`: ANY second
- * operand is accepted today, constant or not
+ * - `ShiftLeft`: the shift amount is constant
+ * - `DateAdd`, `DateSub`, `DateDiff` (end only), `ShiftRight`, `BitwiseOr`:
ANY second operand is
+ * accepted today, constant or not
*
* NOTE: A column-referencing other operand could reverse the ordering (for
ex, A + A * -2 = -A), and a
* non-deterministic one is re-evaluated against the min/max bounds (for
ex, A + rand())
+ *
+ * NOTE: Integral `Add`, `Multiply` and `ShiftLeft` wrap around on overflow in
non-ANSI mode, so the
+ * max can map below the min (for ex, Long.MaxValue * 2 = -2), and throw
in ANSI mode. Their
+ * steps carry a file guard that the file's range does not overflow (see
[[IntegralType]]).
+ * Float / double results round monotonically and saturate to infinity
instead. Decimal results
+ * overflow into NULL and are not matched (#20460)
Review Comment:
Decimal overflow only ever yields NULL, or an error under ANSI, which a
`partial = true` step already turns into a kept file, so could decimal `Add` /
`Multiply` keep pruning that way rather than being rejected outright? `Divide`
needs the same here or in a follow-up, since a decimal quotient overflows into
NULL too (for ex, a `decimal(38, 0)` column divided by 2 has the result type
`decimal(38, 6)`).
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestDataSkippingUtils.scala:
##########
@@ -1168,4 +1169,79 @@ object TestDataSkippingUtils {
Seq("file_1"))
)
}
+
+ def testIntegralOverflowSource(): java.util.stream.Stream[Arguments] = {
+ java.util.stream.Stream.of(
+ // Issue #20460 repros: integral arithmetic wraps around on overflow in
non-ANSI mode, so
+ // re-applying it over min/max stats maps the max below the min and
would wrongly prune
+ // file_1, whose first row matches (51 * 2 = 102, 0 + 1 = 1, 1 << 1 =
2). The file may
+ // overflow, so its bounds are unknown and it is kept
+ arguments(
+ "A * 2L > 100",
+ Seq(
+ IndexRow("file_1", valueCount = 2, A_minValue = 51, A_maxValue =
Long.MaxValue, A_nullCount = 0)
+ ),
+ Seq("file_1")),
+ arguments(
+ "A + 1 > 0",
+ Seq(
+ IndexRow("file_1", valueCount = 2, A_minValue = 0, A_maxValue =
Long.MaxValue, A_nullCount = 0)
+ ),
+ Seq("file_1")),
+ arguments(
+ "shiftleft(A, 1) > 0",
+ Seq(
+ IndexRow("file_1", valueCount = 2, A_minValue = 1, A_maxValue =
Long.MaxValue, A_nullCount = 0)
+ ),
+ Seq("file_1")),
+ // Integral arithmetic still prunes files that cannot overflow. At the
limit (file_1) the
+ // bounds are exact and prune; one past it (file_2) a row wraps around
in non-ANSI mode and
+ // matches (for ex, Long.MaxValue + 1 = Long.MinValue < 0), so the file
is kept
+ arguments(
+ "A + 1 < 0",
+ Seq(
+ IndexRow("file_1", valueCount = 2, A_minValue = Long.MaxValue - 10,
A_maxValue = Long.MaxValue - 1, A_nullCount = 0),
+ IndexRow("file_2", valueCount = 2, A_minValue = Long.MaxValue - 10,
A_maxValue = Long.MaxValue, A_nullCount = 0)
+ ),
+ Seq("file_2")),
+ arguments(
+ "A + -1 > 0",
+ Seq(
+ IndexRow("file_1", valueCount = 2, A_minValue = Long.MinValue + 1,
A_maxValue = Long.MinValue + 10, A_nullCount = 0),
+ IndexRow("file_2", valueCount = 2, A_minValue = Long.MinValue,
A_maxValue = Long.MinValue + 10, A_nullCount = 0)
+ ),
+ Seq("file_2")),
+ // Long.MinValue / 3 does not divide evenly, so the min limit rounds
towards zero
+ arguments(
+ "A * 3L < 0",
+ Seq(
+ IndexRow("file_1", valueCount = 2, A_minValue = 1, A_maxValue =
Long.MaxValue / 3, A_nullCount = 0),
+ IndexRow("file_2", valueCount = 2, A_minValue = 1, A_maxValue =
Long.MaxValue / 3 + 1, A_nullCount = 0)
+ ),
+ Seq("file_2")),
+ arguments(
+ "A * 3L > 0",
+ Seq(
+ IndexRow("file_1", valueCount = 2, A_minValue = Long.MinValue / 3,
A_maxValue = -1, A_nullCount = 0),
+ IndexRow("file_2", valueCount = 2, A_minValue = Long.MinValue / 3 -
1, A_maxValue = -1, A_nullCount = 0)
+ ),
+ Seq("file_2")),
+ // A bigint shifts by the low 6 bits of the amount, so 65 shifts by 1
Review Comment:
65 shifts by 1 under any mask of 1 to 6 bits, so this pair still passes with
a wrong `bits` width, and no test reaches the tinyint / smallint limits either.
Could the pair use an amount such as 33, plus an int amount between 32 and 63
and a smallint term in the golden inputs?
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
