voonhous commented on code in PR #20532:
URL: https://github.com/apache/hudi/pull/20532#discussion_r4488745095
##########
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:
Renamed to `IntegralLimits`.
##########
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:
Done: the bigint pair uses 33, and the golden inputs add `shiftleft(a, 33)`
on an int plus smallint `+ 1` and tinyint `* 3`.
--
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]
