Skip to content

fix(spark): reject order-breaking casts and negative factors in data ... #19475

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

Open
voonhous wants to merge 1 commit into
base: master
Choose a base branch
from
Open

fix(spark): reject order-breaking casts and negative factors in data ... #19475

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
29 changes: 24 additions & 5 deletions ...-spark-common/src/main/scala/org/apache/spark/sql/BaseHoodieCatalystExpressionUtils.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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, Lower, Multiply, PredicateHelper, ShiftLeft, ShiftRight, ToUnixTimestamp, ToUTCTimestamp, Upper}
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, Lower, Multiply, PredicateHelper, ShiftLeft, ShiftRight, ToUnixTimestamp, ToUTCTimestamp, Upper}
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
Expand Down Expand Up @@ -82,9 +82,9 @@ abstract class BaseHoodieCatalystExpressionUtils extends HoodieCatalystExpressio
// 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)
case Multiply(OrderPreservingTransformation(attrRef), factor, _) if isPositiveNumericLiteral(factor) => Some(attrRef)

@cshuo cshuo Aug 4, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

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

A strictly positive factor is not sufficient to make multiplication order-preserving because integral arithmetic wraps in non-ANSI mode. For a bigint file containing {51, Long.MaxValue}, A * 2L > 100 is accepted here and translated using the transformed max. Spark evaluates Long.MaxValue * 2L as -2, so the file is pruned, although 51 * 2L = 102 matches the original predicate.

Please restrict multiplication to analyzed types/factors for which overflow cannot break monotonicity, or conservatively reject the unsafe integral/decimal cases. This exact file-stats scenario should be added as a regression test.

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
Expand All @@ -110,5 +110,24 @@ abstract class BaseHoodieCatalystExpressionUtils extends HoodieCatalystExpressio
}
}
}

// Multiplying or dividing by a constant preserves ordering only when the constant is a
// strictly positive numeric literal: negative factors reverse the ordering, zero collapses
// it (and makes division undefined), and non-literal operands cannot be validated
// statically (the optimizer folds constant factors to literals before data skipping runs).
// Typed null literals carry a null value and fail the value match
private def isPositiveNumericLiteral(expr: Expression): Boolean = expr match {
case Literal(value, _) => value match {

@hudi-agent hudi-agent Aug 3, 2026

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.

🤖 nit: the block comment above could be trimmed — the method name already says "positive numeric literal", and the case _ => false arm makes the null/non-literal behavior self-evident. The most non-obvious part (why non-literals are excluded) could stand on its own as a one-liner: // non-literal operands can't be evaluated statically; the optimizer folds constant expressions before data skipping runs.

⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.

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
case d: Double => d > 0
case dec: Decimal => dec.toBigDecimal.signum > 0
case _ => false
}
case _ => false
}
}
}
19 changes: 15 additions & 4 deletions ...tasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/HoodieSparkTypeUtils.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@

package org.apache.spark.sql

import org.apache.spark.sql.catalyst.expressions.Cast
import org.apache.spark.sql.types.{DataType, DecimalType, NumericType, StringType}

// TODO unify w/ DataTypeUtils
Expand All @@ -35,10 +36,20 @@ object HoodieSparkTypeUtils {
*/
def isCastPreservingOrdering(from: DataType, to: DataType): Boolean =
(from, to) match {
// NOTE: In the casting rules defined by Spark, only casting from String to Numeric
// (and vice versa) are the only casts that might break the ordering of the elements after casting
case (StringType, _: NumericType) => false
case (_: NumericType, StringType) => false
// NOTE: Casting between String and Numeric types re-orders elements (for ex, "10" < "9"
// lexicographically). These arms must stay ahead of the numeric arm below, since
// Cast.canUpCast treats atomic-to-string casts as legal up-casts
case (_: StringType, _: NumericType) => false
case (_: NumericType, _: StringType) => false
// NOTE: On Spark 4 StringType carries a collation (and constraint) and its equals compares
// both; casting to a different collation changes the sort order. On Spark 3
// StringType is a singleton, making this arm trivially true
case (fromStr: StringType, toStr: StringType) => fromStr == toStr
// NOTE: Narrowing numeric casts (for ex, bigint to int) overflow and wrap around in
// non-ANSI mode, breaking ordering; only up-casts are guaranteed order-preserving.
// This is conservative (for ex, double to float rounds monotonically but is
// rejected), which only costs pruning opportunity, never correctness
case (fromNum: NumericType, toNum: NumericType) => Cast.canUpCast(fromNum, toNum)

case _ => true
}
Expand Down
42 changes: 41 additions & 1 deletion hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestDataSkippingUtils.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -93,7 +93,8 @@ class TestDataSkippingUtils extends HoodieSparkClientTestBase with SparkAdapterS
"testBasicLookupFilterExpressionsSource",
"testAdvancedLookupFilterExpressionsSource",
"testCompositeFilterExpressionsSource",
"testSupportedAndUnsupportedDataSkippingColumnsSource"
"testSupportedAndUnsupportedDataSkippingColumnsSource",
"testNonOrderPreservingTransformationsSource"
))
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
Expand Down Expand Up @@ -689,4 +690,43 @@ object TestDataSkippingUtils {

)
}

def testNonOrderPreservingTransformationsSource(): java.util.stream.Stream[Arguments] = {
java.util.stream.Stream.of(
// Issue #19445 repro: cast(4294967297L as int) wraps around to 1 in non-ANSI mode, so
// re-applying the cast over min/max stats would wrongly prune both files; a narrowing
// cast must instead fall back to always-true, keeping every file
arguments(
"CAST(A AS INT) > 100",
Seq(
IndexRow("file_1", valueCount = 2, A_minValue = 1, A_maxValue = 100, A_nullCount = 0),
IndexRow("file_2", valueCount = 3, A_minValue = 1, A_maxValue = 4294967297L, A_nullCount = 0)
),
Seq("file_1", "file_2")),
// A widening (up) cast still translates and prunes
arguments(
"CAST(A AS DOUBLE) > 100.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")),
// Multiplying by a negative constant reverses ordering: min/max swap places, so the
// translated bound would wrongly prune file_1; must fall back to always-true
arguments(
"A * -1 < 5",
Seq(
IndexRow("file_1", valueCount = 2, A_minValue = -10, A_maxValue = 0, A_nullCount = 0)
),
Seq("file_1")),
// Multiplying by a positive constant still translates and prunes
arguments(
"A * 2L < 5",
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"))
)
}
}
63 changes: 58 additions & 5 deletions ...asource/hudi-spark/src/test/scala/org/apache/hudi/TestHoodieCatalystExpressionUtils.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,8 +18,9 @@
package org.apache.hudi

import org.apache.spark.sql.catalyst.expressions.{Add, AttributeReference, BitwiseOr, Cast, DateAdd, DateSub, Divide, Exp, Expression, Literal, Log, Lower, Multiply, ParseToDate, ShiftLeft, Sqrt, Upper}
import org.apache.spark.sql.types.{DateType, DoubleType, IntegerType, LongType, StringType}
import org.apache.spark.sql.types.{DataType, DateType, Decimal, DecimalType, DoubleType, FloatType, IntegerType, LongType, StringType}
import org.junit.jupiter.api.Assertions.assertEquals
import org.junit.jupiter.api.Assumptions.assumeTrue
import org.junit.jupiter.api.Test

/**
Expand All @@ -35,6 +36,7 @@ class TestHoodieCatalystExpressionUtils extends SparkAdapterSupport {
private val dblAttr = AttributeReference("d", DoubleType)()
private val dateAttr = AttributeReference("dt", DateType)()
private val longAttr = AttributeReference("l", LongType)()
private val decAttr = AttributeReference("dec", DecimalType(10, 2))()

private def matched(expr: Expression): Option[AttributeReference] =
sparkAdapter.getCatalystExpressionUtils.tryMatchAttributeOrderingPreservingTransformation(expr)
Expand Down Expand Up @@ -84,10 +86,61 @@ class TestHoodieCatalystExpressionUtils extends SparkAdapterSupport {
assertEquals(Some(intAttr), matched(Cast(intAttr, LongType)))
// Casting a numeric column to string can reorder values, so it must not match.
assertEquals(None, matched(Cast(intAttr, StringType)))
// TODO(#19445): a narrowing numeric cast wraps around in non-ANSI mode and does not preserve
// ordering, so this should be None; pinning the current (incorrect) behavior until
// isCastPreservingOrdering rejects it.
assertEquals(Some(longAttr), matched(Cast(longAttr, IntegerType)))
// A narrowing numeric cast wraps around in non-ANSI mode and does not preserve ordering.
assertEquals(None, matched(Cast(longAttr, IntegerType)))
}

@Test
def testNumericCastOrderingFollowsUpCastRules(): Unit = {
// Widening precedence casts preserve ordering, including lossy-but-monotonic long to float.
assertEquals(Some(intAttr), matched(Cast(intAttr, DoubleType)))
assertEquals(Some(longAttr), matched(Cast(longAttr, FloatType)))
// Decimal widening is an up-cast; narrowing precision is not.
assertEquals(Some(decAttr), matched(Cast(decAttr, DecimalType(12, 2))))
assertEquals(None, matched(Cast(decAttr, DecimalType(8, 2))))
// Cast.canUpCast rejects double to float even though rounding is weakly monotonic; the
// conservative answer only costs pruning opportunity.
assertEquals(None, matched(Cast(dblAttr, FloatType)))
// String to numeric is not order-preserving.
assertEquals(None, matched(Cast(strAttr, IntegerType)))
// Identity string cast keeps the ordering.
assertEquals(Some(strAttr), matched(Cast(strAttr, StringType)))
}

@Test
def testMultiplyDivideRequireStrictlyPositiveLiteral(): Unit = {
// Negative factors reverse ordering.
assertEquals(None, matched(Multiply(intAttr, Literal(-1))))
assertEquals(None, matched(Multiply(Literal(-1), intAttr)))
assertEquals(None, matched(Divide(intAttr, Literal(-2))))
// Zero collapses ordering and makes division undefined.
assertEquals(None, matched(Multiply(intAttr, Literal(0))))
assertEquals(None, matched(Divide(intAttr, Literal(0))))
// Typed null literals carry a null value.
assertEquals(None, matched(Multiply(intAttr, Literal(null, IntegerType))))
// Non-literal operands cannot be validated statically, even when reference-free.
assertEquals(None, matched(Multiply(intAttr, Add(Literal(1), Literal(1)))))
// Self-multiplication is not monotonic over negative values.
assertEquals(None, matched(Multiply(intAttr, intAttr)))
// Strictly positive literals of any numeric type match in the supported operand positions.
assertEquals(Some(intAttr), matched(Multiply(intAttr, Literal(2.5d))))
assertEquals(Some(dblAttr), matched(Multiply(Literal(Decimal(2)), dblAttr)))
assertEquals(Some(intAttr), matched(Divide(intAttr, Literal(3L))))
}

@Test
def testCollationChangingStringCastsDoNotPreserveOrdering(): Unit = {
assumeTrue(HoodieSparkUtils.gteqSpark4_0, "String collations only exist on Spark 4.x")
// StringType("UTF8_LCASE") is a Spark 4 API, so it is obtained reflectively to keep this
// file compiling against the Spark 3 profiles.
val lcase = StringType.getClass.getMethod("apply", classOf[String])
.invoke(StringType, "UTF8_LCASE").asInstanceOf[DataType]
val collatedAttr = AttributeReference("cs", lcase)()
// Changing collation changes the sort order, in either direction.
assertEquals(None, matched(Cast(collatedAttr, StringType)))
assertEquals(None, matched(Cast(strAttr, lcase)))
// A collation-preserving cast keeps the ordering.
assertEquals(Some(collatedAttr), matched(Cast(collatedAttr, lcase)))
}

@Test
Expand Down
Loading