This is an automated email from the ASF dual-hosted git repository.
voonhous pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new c58c54e068b5 fix(spark): reject negative factors in data skipping
multiply/divide (#20457)
c58c54e068b5 is described below
commit c58c54e068b59df35d0d25e22873b3c57636b557
Author: voonhous <[email protected]>
AuthorDate: Wed Nov 4 10:42:13 2026 +0800
fix(spark): reject negative factors in data skipping multiply/divide
(#20457)
* fix(spark): reject negative factors in data skipping multiply/divide
The Multiply and Divide arms of the order-preserving matcher accepted
any operand as the factor. A negative constant reverses the ordering,
so the translated min/max bound could prune files holding matching
rows (A in [-10, 0] with A * -1.0 < 5.0 translates to 10 < 5).
Multiply and Divide now require a strictly positive numeric literal
factor; anything else falls back to no pruning for that predicate.
Refs #19445
* fix(spark): reject Add operands that reference the column
The Add arms matched the order-preserving side without inspecting the
other operand, so A + A * -2 (= -A) slipped a negative factor past the
Multiply check and pruned files holding matching rows. Require the
other operand to be reference-free.
* fix(spark): reject non-constant and infinite data skipping operands
- Add arms now require a foldable other operand instead of a
reference-free one: data skipping re-evaluates it against the
min/max bounds, so A + rand() could prune files holding matching
rows
- Multiply/Divide factors must be finite: 0 * Infinity and
Infinity / Infinity give NaN, which sorts above every number
Refs #19445
---
.../sql/BaseHoodieCatalystExpressionUtils.scala | 34 +++++++++---
.../org/apache/hudi/TestDataSkippingUtils.scala | 63 ++++++++++++++++++++--
.../hudi/TestHoodieCatalystExpressionUtils.scala | 40 +++++++++++++-
3 files changed, 125 insertions(+), 12 deletions(-)
diff --git
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/BaseHoodieCatalystExpressionUtils.scala
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/BaseHoodieCatalystExpressionUtils.scala
index 4ea289bf496c..e929fabaf94f 100644
---
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/BaseHoodieCatalystExpressionUtils.scala
+++
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/BaseHoodieCatalystExpressionUtils.scala
@@ -18,9 +18,9 @@
package org.apache.spark.sql
import org.apache.spark.sql.HoodieSparkTypeUtils.isCastPreservingOrdering
-import org.apache.spark.sql.catalyst.expressions.{Add, Attribute,
AttributeReference, AttributeSet, BitwiseOr, Cast, DateAdd, DateDiff,
DateFormatClass, DateSub, Divide, Exp, Expm1, Expression, FromUnixTime,
FromUTCTimestamp, Log, Log10, Log1p, Log2, Multiply, PredicateHelper,
ShiftLeft, ShiftRight, ToUnixTimestamp, ToUTCTimestamp}
+import org.apache.spark.sql.catalyst.expressions.{Add, Attribute,
AttributeReference, AttributeSet, BitwiseOr, Cast, DateAdd, DateDiff,
DateFormatClass, DateSub, Divide, Exp, Expm1, Expression, FromUnixTime,
FromUTCTimestamp, Literal, Log, Log10, Log1p, Log2, Multiply, PredicateHelper,
ShiftLeft, ShiftRight, ToUnixTimestamp, ToUTCTimestamp}
import org.apache.spark.sql.execution.datasources.DataSourceStrategy
-import org.apache.spark.sql.types.DataType
+import org.apache.spark.sql.types.{DataType, Decimal}
/**
* Base implementation of [[HoodieCatalystExpressionUtils]] carrying the
method bodies that are
@@ -81,11 +81,14 @@ abstract class BaseHoodieCatalystExpressionUtils extends
HoodieCatalystExpressio
// Math Expressions
// Binary
- case Add(OrderPreservingTransformation(attrRef), _, _) => Some(attrRef)
- case Add(_, OrderPreservingTransformation(attrRef), _) => Some(attrRef)
- case Multiply(OrderPreservingTransformation(attrRef), _, _) =>
Some(attrRef)
- case Multiply(_, OrderPreservingTransformation(attrRef), _) =>
Some(attrRef)
- case Divide(OrderPreservingTransformation(attrRef), _, _) =>
Some(attrRef)
+ // NOTE: The other operand must be a constant. Referencing the column
could reverse the
+ // ordering (for ex, A + A * -2 = -A), and a non-deterministic
operand is re-evaluated
+ // against the min/max bounds (for ex, A + rand())
+ case Add(OrderPreservingTransformation(attrRef), other, _) if
other.foldable => Some(attrRef)
+ case Add(other, OrderPreservingTransformation(attrRef), _) if
other.foldable => Some(attrRef)
+ case Multiply(OrderPreservingTransformation(attrRef), factor, _) if
isPositiveNumericLiteral(factor) => Some(attrRef)
+ case Multiply(factor, OrderPreservingTransformation(attrRef), _) if
isPositiveNumericLiteral(factor) => Some(attrRef)
+ case Divide(OrderPreservingTransformation(attrRef), divisor, _) if
isPositiveNumericLiteral(divisor) => Some(attrRef)
case BitwiseOr(OrderPreservingTransformation(attrRef), _) =>
Some(attrRef)
case BitwiseOr(_, OrderPreservingTransformation(attrRef)) =>
Some(attrRef)
// Unary
@@ -111,5 +114,22 @@ abstract class BaseHoodieCatalystExpressionUtils extends
HoodieCatalystExpressio
}
}
}
+
+ // Negative factors reverse the ordering and zero collapses it. Infinite
factors can yield NaN
+ // (0 * Infinity, Infinity / Infinity), which sorts above every number.
Non-literal factors are
+ // not matched since the optimizer folds constant factors to literals
before data skipping runs
+ private def isPositiveNumericLiteral(expr: Expression): Boolean = expr
match {
+ case Literal(value, _) => value match {
+ case b: Byte => b > 0
+ case s: Short => s > 0
+ case i: Int => i > 0
+ case l: Long => l > 0
+ case f: Float => f > 0 && !f.isInfinite
+ case d: Double => d > 0 && !d.isInfinite
+ case dec: Decimal => dec.toBigDecimal.signum > 0
+ case _ => false
+ }
+ case _ => false
+ }
}
}
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestDataSkippingUtils.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestDataSkippingUtils.scala
index 41c04681fd54..4ac23be2174b 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestDataSkippingUtils.scala
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestDataSkippingUtils.scala
@@ -29,7 +29,7 @@ import org.apache.spark.sql.catalyst.expressions.{Expression,
InSet, Not}
import org.apache.spark.sql.catalyst.optimizer.OptimizeIn
import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
import org.apache.spark.sql.catalyst.rules.Rule
-import org.apache.spark.sql.functions.{col, date_format, lower, upper}
+import org.apache.spark.sql.functions.{col, date_format, lit, lower, upper}
import org.apache.spark.sql.hudi.DataSkippingUtils
import org.apache.spark.sql.internal.SQLConf.SESSION_LOCAL_TIMEZONE
import org.apache.spark.sql.types._
@@ -95,7 +95,8 @@ class TestDataSkippingUtils extends HoodieSparkClientTestBase
with SparkAdapterS
"testAdvancedLookupFilterExpressionsSource",
"testCompositeFilterExpressionsSource",
"testSupportedAndUnsupportedDataSkippingColumnsSource",
- "testNonOrderPreservingTransformationsSource"
+ "testNonOrderPreservingTransformationsSource",
+ "testNegativeFactorSource"
))
def testLookupFilterExpressions(sourceFilterExprStr: String, input:
Seq[IndexRow], expectedOutput: Seq[String]): Unit = {
// We have to fix the timezone to make sure all date-bound utilities output
@@ -335,7 +336,18 @@ object TestDataSkippingUtils {
C_maxValue = new Timestamp(1648785048000L), // 04/01/2022
C_nullCount = 0)
),
- Seq("file_2"))
+ Seq("file_2")),
+ arguments(
+ // An infinite factor turns the zero min into NaN (0 * Infinity),
which sorts above every
+ // number, so the translated bound A_minValue * Infinity <= Infinity
is false and would
+ // wrongly prune file_1, although A = 1 gives Infinity <= Infinity.
Built with lit() since
+ // SQL has no infinite double literal. Must fall back to always-true,
keeping the file
+ sparkAdapter.getExpressionFromColumn(
+ col("A").cast(DoubleType) * lit(Double.PositiveInfinity) <=
lit(Double.PositiveInfinity)),
+ Seq(
+ IndexRow("file_1", valueCount = 2, A_minValue = 0, A_maxValue = 1,
A_nullCount = 0)
+ ),
+ Seq("file_1"))
)
}
@@ -576,6 +588,51 @@ object TestDataSkippingUtils {
)
}
+ def testNegativeFactorSource(): java.util.stream.Stream[Arguments] = {
+ java.util.stream.Stream.of(
+ // Multiplying by a negative constant reverses ordering: the translated
bound
+ // A_minValue * -1.0 < 5.0 is 10 < 5 and would wrongly prune file_1,
although A = 0 gives
+ // 0 < 5. Must fall back to always-true, keeping the file
+ arguments(
+ "CAST(A AS DOUBLE) * -1.0D < 5.0D",
+ Seq(
+ IndexRow("file_1", valueCount = 2, A_minValue = -10, A_maxValue = 0,
A_nullCount = 0)
+ ),
+ Seq("file_1")),
+ // Dividing by a negative constant reverses ordering the same way
+ arguments(
+ "CAST(A AS DOUBLE) / -2.0D < 5.0D",
+ Seq(
+ IndexRow("file_1", valueCount = 2, A_minValue = -20, A_maxValue = 0,
A_nullCount = 0)
+ ),
+ Seq("file_1")),
+ // Adding a term that references the column can carry a negative factor
past the Multiply
+ // check: A + A * -2 = -A, so the translated bound A_minValue +
A_minValue * -2 < -5 is
+ // 0 < -5 and would wrongly prune file_1, although A = 10 gives -10 < -5
+ arguments(
+ "A + A * -2 < -5",
+ Seq(
+ IndexRow("file_1", valueCount = 2, A_minValue = 0, A_maxValue = 10,
A_nullCount = 0)
+ ),
+ Seq("file_1")),
+ // Multiplying and dividing by a positive constant still translate and
prune
+ arguments(
+ "CAST(A AS DOUBLE) * 2.0D < 5.0D",
+ Seq(
+ IndexRow("file_1", valueCount = 2, A_minValue = 1, A_maxValue = 2,
A_nullCount = 0),
+ IndexRow("file_2", valueCount = 2, A_minValue = 3, A_maxValue = 4,
A_nullCount = 0)
+ ),
+ Seq("file_1")),
+ arguments(
+ "CAST(A AS DOUBLE) / 2.0D > 50.0D",
+ Seq(
+ IndexRow("file_1", valueCount = 2, A_minValue = 1, A_maxValue = 100,
A_nullCount = 0),
+ IndexRow("file_2", valueCount = 2, A_minValue = 101, A_maxValue =
200, A_nullCount = 0)
+ ),
+ Seq("file_2"))
+ )
+ }
+
def testCompositeFilterExpressionsSource():
java.util.stream.Stream[Arguments] = {
// NOTE: all timestamps in UTC
java.util.stream.Stream.of(
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestHoodieCatalystExpressionUtils.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestHoodieCatalystExpressionUtils.scala
index d937ed51f7ca..fd063bac95ba 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestHoodieCatalystExpressionUtils.scala
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestHoodieCatalystExpressionUtils.scala
@@ -17,8 +17,8 @@
package org.apache.hudi
-import org.apache.spark.sql.catalyst.expressions.{Add, AttributeReference,
BitwiseOr, Cast, DateAdd, DateDiff, DateSub, Divide, Exp, Expression, Literal,
Log, Lower, Multiply, ParseToDate, ShiftLeft, Sqrt, Upper}
-import org.apache.spark.sql.types.{BinaryType, BooleanType, ByteType,
DataType, DateType, DecimalType, DoubleType, FloatType, IntegerType, LongType,
ShortType, StringType, TimestampType}
+import org.apache.spark.sql.catalyst.expressions.{Add, AttributeReference,
BitwiseOr, Cast, DateAdd, DateDiff, DateSub, Divide, Exp, Expression, Literal,
Log, Lower, Multiply, ParseToDate, Rand, ShiftLeft, Sqrt, Upper}
+import org.apache.spark.sql.types.{BinaryType, BooleanType, ByteType,
DataType, DateType, Decimal, DecimalType, DoubleType, FloatType, IntegerType,
LongType, ShortType, StringType, TimestampType}
import org.junit.jupiter.api.Assertions.assertEquals
import org.junit.jupiter.api.Assumptions.assumeTrue
import org.junit.jupiter.api.Test
@@ -61,6 +61,42 @@ class TestHoodieCatalystExpressionUtils extends
SparkAdapterSupport {
assertEquals(Some(intAttr), matched(ShiftLeft(intAttr, Literal(1))))
}
+ @Test
+ def testMultiplyDivideRequirePositiveLiteralFactor(): Unit = {
+ // Negative factors reverse ordering.
+ assertEquals(None, matched(Multiply(dblAttr, Literal(-1.0d))))
+ assertEquals(None, matched(Multiply(Literal(-1), intAttr)))
+ assertEquals(None, matched(Divide(dblAttr, Literal(Decimal(-2),
DecimalType(1, 0)))))
+ // Zero collapses ordering and makes division undefined.
+ assertEquals(None, matched(Multiply(intAttr, Literal(0L))))
+ assertEquals(None, matched(Divide(dblAttr, Literal(0.0d))))
+ // NaN fails the positive check; typed null literals carry a null value.
+ assertEquals(None, matched(Multiply(dblAttr, Literal(Double.NaN))))
+ // Infinite factors can yield NaN (0 * Infinity, Infinity / Infinity).
+ assertEquals(None, matched(Multiply(dblAttr,
Literal(Double.PositiveInfinity))))
+ assertEquals(None, matched(Multiply(Literal(Float.PositiveInfinity),
dblAttr)))
+ assertEquals(None, matched(Divide(dblAttr,
Literal(Double.PositiveInfinity))))
+ assertEquals(None, matched(Multiply(intAttr, Literal(null, IntegerType))))
+ // Non-literal factors are not matched, even when reference-free or the
column itself.
+ assertEquals(None, matched(Multiply(intAttr, Add(Literal(1), Literal(1)))))
+ assertEquals(None, matched(Multiply(intAttr, intAttr)))
+ // Strictly positive literals of any numeric type match in the supported
operand positions.
+ assertEquals(Some(dblAttr), matched(Multiply(dblAttr, Literal(2.5d))))
+ assertEquals(Some(longAttr), matched(Multiply(Literal(3L), longAttr)))
+ assertEquals(Some(dblAttr), matched(Divide(dblAttr, Literal(Decimal(2),
DecimalType(1, 0)))))
+ }
+
+ @Test
+ def testAddRequiresConstantOtherOperand(): Unit = {
+ // A + A * -2 = -A reverses ordering, so an operand referencing the column
must not match.
+ assertEquals(None, matched(Add(dblAttr, Multiply(dblAttr,
Literal(-2.0d)))))
+ assertEquals(None, matched(Add(Multiply(dblAttr, Literal(-2.0d)),
dblAttr)))
+ assertEquals(None, matched(Add(intAttr, intAttr)))
+ // Data skipping re-evaluates the operand against the min/max bounds, so
it must be a constant.
+ assertEquals(None, matched(Add(dblAttr, Rand(Literal(0L)))))
+ assertEquals(None, matched(Add(Rand(Literal(0L)), dblAttr)))
+ }
+
@Test
def testUnaryMathTransformationsPreserveOrdering(): Unit = {
assertEquals(Some(dblAttr), matched(Exp(dblAttr)))
