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 91ffb05e62cf fix(spark): stop treating lower/upper as order-preserving
in data skipping (#20228)
91ffb05e62cf is described below
commit 91ffb05e62cf9a2f58bd10a326d07e02f69ab081
Author: Vova Kolmakov <[email protected]>
AuthorDate: Mon Oct 19 15:47:15 2026 +0700
fix(spark): stop treating lower/upper as order-preserving in data skipping
(#20228)
* fix(spark): stop treating lower/upper as order-preserving in data skipping
* addressed review comments: cover NOT startswith over date_format and trim
the lower() case
---------
Co-authored-by: Vova Kolmakov <[email protected]>
---
.../sql/BaseHoodieCatalystExpressionUtils.scala | 6 +-
.../org/apache/hudi/TestDataSkippingUtils.scala | 66 +++++++++++++++-------
.../hudi/TestHoodieCatalystExpressionUtils.scala | 7 ++-
3 files changed, 53 insertions(+), 26 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 e2a39d7c29a5..4ea289bf496c 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,7 +18,7 @@
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, 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
@@ -73,8 +73,8 @@ abstract class BaseHoodieCatalystExpressionUtils extends
HoodieCatalystExpressio
case ToUTCTimestamp(OrderPreservingTransformation(attrRef), _) =>
Some(attrRef)
// String Expressions
- case Lower(OrderPreservingTransformation(attrRef)) => Some(attrRef)
- case Upper(OrderPreservingTransformation(attrRef)) => Some(attrRef)
+ // NOTE: Lower/Upper are not matched: case folding does not preserve
the binary ordering
+ // of strings (for ex, "Zebra" < "apple", but lower("Zebra") =
"zebra" > "apple")
// Left API change: Improve RuntimeReplaceable
// https://issues.apache.org/jira/browse/SPARK-38240
case
org.apache.spark.sql.catalyst.expressions.Left(OrderPreservingTransformation(attrRef),
_) => Some(attrRef)
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 51e5aae98be6..694fb49f6df4 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, lower}
+import org.apache.spark.sql.functions.{col, date_format, 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._
@@ -189,15 +189,29 @@ object TestDataSkippingUtils {
),
Seq("file_1", "file_2", "file_3")),
arguments(
- // Composite expression
+ // NOTE: file_1 may hold "B" (sorting between "ABC" and "abc"), whose
lower-case form "b" doesn't
+ // start w/ "abc", even though lower(B_minValue) =
lower(B_maxValue) = "abc" does
Not(sparkAdapter.getExpressionFromColumn(lower(col("B")).startsWith("abc"))),
Seq(
- IndexRow("file_1", valueCount = 1, B_minValue = "ABA", B_maxValue =
"ADF", B_nullCount = 1), // may contain strings starting w/ "ABC" (after upper)
- IndexRow("file_2", valueCount = 1, B_minValue = "ADF", B_maxValue =
"AZY", B_nullCount = 0),
- IndexRow("file_3", valueCount = 1, B_minValue = "AAA", B_maxValue =
"ABA", B_nullCount = 0),
- IndexRow("file_4", valueCount = 1, B_minValue = "ABC123", B_maxValue
= "ABC345", B_nullCount = 0) // all strings start w/ "ABC" (after upper)
+ IndexRow("file_1", valueCount = 3, B_minValue = "ABC", B_maxValue =
"abc", B_nullCount = 0)
),
- Seq("file_1", "file_2", "file_3"))
+ Seq("file_1")),
+ arguments(
+ // NOTE: file_1 holds "Zebra", whose lower-case form is the queried
"zebra", even though
+ // lower(B_maxValue) = "apple" sorts before it
+ sparkAdapter.getExpressionFromColumn(lower(col("B")) === "zebra"),
+ Seq(
+ IndexRow("file_1", valueCount = 2, B_minValue = "Zebra", B_maxValue
= "apple", B_nullCount = 0)
+ ),
+ Seq("file_1")),
+ arguments(
+ // NOTE: file_1 holds "apple", whose upper-case form is the queried
"APPLE", even though
+ // upper(B_minValue) = "ZEBRA" sorts after it
+ sparkAdapter.getExpressionFromColumn(upper(col("B")) === "APPLE"),
+ Seq(
+ IndexRow("file_1", valueCount = 2, B_minValue = "Zebra", B_maxValue
= "apple", B_nullCount = 0)
+ ),
+ Seq("file_1"))
)
}
@@ -263,19 +277,31 @@ object TestDataSkippingUtils {
}
def testMiscLookupFilterExpressionsSource():
java.util.stream.Stream[Arguments] = {
- // NOTE: Have to use [[Arrays.stream]], as Scala can't resolve properly 2
overloads for [[Stream.of]]
- // (for single element)
- java.util.Arrays.stream(
- Array(
- arguments(
- InSet(UnresolvedAttribute("A"), HashSet(0, 1)),
- Seq(
- IndexRow("file_1", valueCount = 1, 1, 2, 0),
- IndexRow("file_2", valueCount = 1, -1, 1, 0),
- IndexRow("file_3", valueCount = 1, -2, -1, 0)
- ),
- Seq("file_1", "file_2"))
- )
+ java.util.stream.Stream.of(
+ arguments(
+ InSet(UnresolvedAttribute("A"), HashSet(0, 1)),
+ Seq(
+ IndexRow("file_1", valueCount = 1, 1, 2, 0),
+ IndexRow("file_2", valueCount = 1, -1, 1, 0),
+ IndexRow("file_3", valueCount = 1, -2, -1, 0)
+ ),
+ Seq("file_1", "file_2")),
+ arguments(
+ // Composite expression
+ // NOTE: date_format(C, 'yyyy-MM-dd') preserves the ordering of C, so
this filter is translated and
+ // file_1, whose dates all fall in 03/2022, is pruned
+ Not(sparkAdapter.getExpressionFromColumn(date_format(col("C"),
"yyyy-MM-dd").startsWith("2022-03"))),
+ Seq(
+ IndexRow("file_1", valueCount = 2,
+ C_minValue = new Timestamp(1646625048000L), // 03/07/2022
+ C_maxValue = new Timestamp(1646797848000L), // 03/09/2022
+ C_nullCount = 0),
+ IndexRow("file_2", valueCount = 2,
+ C_minValue = new Timestamp(1646711448000L), // 03/08/2022
+ C_maxValue = new Timestamp(1648785048000L), // 04/01/2022
+ C_nullCount = 0)
+ ),
+ Seq("file_2"))
)
}
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 a7bd52dcefd7..52f133bb6f15 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
@@ -57,11 +57,9 @@ class TestHoodieCatalystExpressionUtils extends
SparkAdapterSupport {
}
@Test
- def testUnaryMathAndStringTransformationsPreserveOrdering(): Unit = {
+ def testUnaryMathTransformationsPreserveOrdering(): Unit = {
assertEquals(Some(dblAttr), matched(Exp(dblAttr)))
assertEquals(Some(dblAttr), matched(Log(dblAttr)))
- assertEquals(Some(strAttr), matched(Upper(strAttr)))
- assertEquals(Some(strAttr), matched(Lower(strAttr)))
}
@Test
@@ -107,5 +105,8 @@ class TestHoodieCatalystExpressionUtils extends
SparkAdapterSupport {
assertEquals(None, matched(Add(Literal(1), Literal(2))))
// Sqrt is not one of the whitelisted order-preserving transformations.
assertEquals(None, matched(Sqrt(dblAttr)))
+ // Case folding breaks the binary ordering of strings: "Zebra" < "apple",
but lower("Zebra") > "apple".
+ assertEquals(None, matched(Lower(strAttr)))
+ assertEquals(None, matched(Upper(strAttr)))
}
}
