Re: [PR] fix(spark): keep to_date/to_timestamp data skipping working after ReplaceExpressions [hudi]

["wombatu-kun (via GitHub)" <gi...@apache.org>]

wombatu-kun commented on code in PR #19474:
URL: https://github.com/apache/hudi/pull/19474#discussion_r4214341194


##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/DataSkippingUtils.scala:
##########
@@ -390,6 +391,24 @@ object DataSkippingUtils extends Logging {
     }
   }
 
+  /**
+   * Wraps a translated index-lookup bound in a null-tolerant guard when the source expression is
+   * an actual transformation of the column (not a bare attribute reference).
+   *
+   * Whitelisted transformations may be partial functions of the source column (for example
+   * GetTimestamp parsing a string): re-applied on top of a min/max stat value that the function
+   * is undefined for, they return null, in which case f(max(col)) != max(f(col)) and the bound
+   * cannot be decided. A null bound must keep the file (hence Coalesce to true) rather than
+   * silently prune it. A false bound may still prune: each one-sided bound is independently
+   * valid whenever its own stat value is defined (for example "null AND false" evaluates to
+   * false, which is still sound).
+   */
+  private def wrapTransformedBoundNullSafe(sourceExpr: Expression, bound: Expression): Expression =
+    sourceExpr match {
+      case _: AttributeReference => bound
+      case _ => Coalesce(Seq(bound, TrueLiteral))

Review Comment:
   `Coalesce(bound, true)` also fires when the stat itself is null, so for a column that is all-null in a file `lower(col) IN ('a', 'b')` now keeps the file while `col IN ('a', 'b')` still prunes it. Not a blocker, but could the fallback be `IsNotNull(genColMinValueExpr(colName))` instead of `true`, so only a defined stat whose transformation yields null keeps the file?



##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestBucketIndexSupport.scala:
##########
@@ -31,13 +31,12 @@ import org.apache.hudi.index.bucket.BucketIdentifier
 import org.apache.hudi.index.bucket.partition.NumBucketsFunction
 import org.apache.hudi.keygen.{ComplexKeyGenerator, NonpartitionedKeyGenerator}
 import org.apache.hudi.keygen.constant.KeyGeneratorOptions
-import org.apache.hudi.testutils.HoodieSparkClientTestBase
+import org.apache.hudi.testutils.{HoodieDummyExpressionHolder, HoodieSparkClientTestBase}
 import org.apache.hudi.timeline.service.TimelineService
 
 import org.apache.avro.generic.GenericData
 import org.apache.spark.sql.{BucketPartitionUtils, HoodieCatalystExpressionUtils, SparkSession}
-import org.apache.spark.sql.catalyst.expressions.{Attribute, Expression, PredicateHelper}
-import org.apache.spark.sql.catalyst.plans.logical.LeafNode
+import org.apache.spark.sql.catalyst.expressions.PredicateHelper

Review Comment:
   This import drops `Expression`, but current master's `optimizeResolvedExpr` in this class returns `Expression`, so the textually clean merge with master no longer compiles. Could you rebase and keep `{Expression, PredicateHelper}` here?



##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/BaseHoodieCatalystExpressionUtils.scala:
##########
@@ -70,6 +89,21 @@ abstract class BaseHoodieCatalystExpressionUtils extends HoodieCatalystExpressio
         case FromUTCTimestamp(OrderPreservingTransformation(attrRef), _) => Some(attrRef)
         case ToUnixTimestamp(OrderPreservingTransformation(attrRef), _, _, _) => Some(attrRef)
         case ToUTCTimestamp(OrderPreservingTransformation(attrRef), _) => Some(attrRef)
+        // Post-ReplaceExpressions shape of to_date/to_timestamp with an explicit format: since
+        // SPARK-38240 the optimizer rewrites those RuntimeReplaceable nodes into
+        // GetTimestamp(source, fmt) (Cast-wrapped to date for to_date) before filters reach
+        // file pruning. Matched only when
+        //   (a) the format is a literal, fixed-width, year-first pattern: the translation
+        //       re-applies the parse on top of string min/max column stats, which is only
+        //       sound when lexicographic ordering of parseable strings agrees with
+        //       chronological ordering, and
+        //   (b) failOnError is false (non-ANSI), so probing a stat value that does not parse
+        //       can never throw from the index lookup; it yields null instead, which the
+        //       null-tolerant bounds in DataSkippingUtils turn into "keep the file".
+        // GetTimestamp's constructor arity differs across Spark versions but its left()/right()
+        // accessors are stable, hence the typed match
+        case gt: GetTimestamp if !gt.failOnError && isOrderPreservingDateFormat(gt.right) =>

Review Comment:
   Under `spark.sql.legacy.timeParserPolicy=LEGACY` this parse accepts variable-width input, so the `file_slop` fixture queried with `to_date(B, 'yyyy-MM-dd') = '2022-11-05'` gets pruned even though its `'2022-11-5'` row matches, and under the Spark 3.x default `EXCEPTION` that stat throws `SparkUpgradeException` with `failOnError` off, uncaught on the partition-stats path this arm now reaches. Should the arm also require the policy to be `CORRECTED`, like the ANSI gate?



-- 
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: commits-unsubscribe@hudi.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org