diff --git a/contrib/delta-spark/src/main/scala/org/apache/comet/contrib/delta/CometDeltaNativeScan.scala b/contrib/delta-spark/src/main/scala/org/apache/comet/contrib/delta/CometDeltaNativeScan.scala index 32a1b4a00c6..d50f226ca61 100644 --- a/contrib/delta-spark/src/main/scala/org/apache/comet/contrib/delta/CometDeltaNativeScan.scala +++ b/contrib/delta-spark/src/main/scala/org/apache/comet/contrib/delta/CometDeltaNativeScan.scala @@ -353,11 +353,13 @@ object CometDeltaNativeScan case s: ExecScalarSubquery => Literal.create(s.eval(null), s.dataType) } - val proto = exprToProto(resolved, output) - if (proto.isEmpty) { - logWarning(s"Could not serialize resolved scalar subquery filter: $resolved") + CometNativeScan.filterForParquetPushdown(resolved).flatMap { pushable => + val proto = exprToProto(pushable, output) + if (proto.isEmpty) { + logWarning(s"Could not serialize resolved scalar subquery filter: $pushable") + } + proto } - proto } ResolvedSubqueryFilters(hasResolvedFilters = true, protos) } @@ -432,6 +434,7 @@ object CometDeltaNativeScan val dataColIds = output.take(strippedLen).map(_.exprId).toSet val filterProtos = scanHelper.supportedDataFilters .filter(_.references.forall(r => dataColIds.contains(r.exprId))) + .flatMap(CometNativeScan.filterForParquetPushdown) .flatMap(f => exprToProto(f, output)) commonBuilder.addAllDataFilters(filterProtos.asJava) } diff --git a/contrib/delta-spark/src/test/scala/org/apache/comet/contrib/delta/DeltaScanContribSuite.scala b/contrib/delta-spark/src/test/scala/org/apache/comet/contrib/delta/DeltaScanContribSuite.scala index 1d240ba3e22..f0b077fedb8 100644 --- a/contrib/delta-spark/src/test/scala/org/apache/comet/contrib/delta/DeltaScanContribSuite.scala +++ b/contrib/delta-spark/src/test/scala/org/apache/comet/contrib/delta/DeltaScanContribSuite.scala @@ -32,6 +32,7 @@ import org.apache.hadoop.fs.{Path, RawLocalFileSystem} import org.apache.hadoop.fs.s3a.{S3AFileSystem, S3AUtils} import org.apache.hadoop.security.alias.CredentialProviderFactory import org.apache.spark.sql.comet.CometDeltaNativeScanExec +import org.apache.spark.sql.delta.DeltaLog import org.apache.spark.sql.delta.actions.DeletionVectorDescriptor import org.apache.spark.sql.execution.SparkPlan @@ -92,6 +93,47 @@ class DeltaScanContribSuite extends CometDeltaTestBase { } } + test("make_date filters preserve Spark's reader before the Delta scan claim") { + withSQLConf( + "spark.sql.ansi.enabled" -> "false", + "spark.sql.adaptive.enabled" -> "false", + CometConf.COMET_PARQUET_ROW_FILTER_PUSHDOWN_ENABLED.key -> "true") { + withTempPath { dir => + val path = dir.getAbsolutePath + spark + .range(0, 3, 1, 1) + .selectExpr("id", "CAST(id + 2024 AS INT) AS y", "IF(id = 1, 13, 1) AS m") + .write + .format("delta") + .save(path) + + for (deletionVectors <- Seq(false, true)) { + if (deletionVectors) { + spark.sql( + s"ALTER TABLE delta.`$path` SET TBLPROPERTIES ('delta.enableDeletionVectors' = 'true')") + spark.sql(s"DELETE FROM delta.`$path` WHERE id = 2") + assert( + DeltaLog + .forTable(spark, path) + .update() + .allFiles + .collect() + .exists(_.deletionVector != null)) + } + val input = spark.read.format("delta").load(path) + val guarded = input.filter("unix_date(make_date(y, m, 1)) > 0") + checkSparkAnswer(guarded) + assert(deltaNativeScans(guarded).isEmpty) + assert( + collectByName(guarded.queryExecution.executedPlan, "FileSourceScanExec").nonEmpty) + + // A bounded ANSI-off year is safe, including the invalid month that returns NULL. + checkDeltaNativeScanAnswer(input.filter("unix_date(make_date(2024, m, 1)) > 0")) + } + } + } + } + test("declined scan carries the contrib's fallback reason, not core's generic one") { // Disabling Spark's vectorized Parquet reader is a scan the contrib recognizes // (DeltaScanSupport.isDeltaScan) but explicitly declines (DeltaScanSupport.declineReason, diff --git a/docs/source/user-guide/latest/expressions.md b/docs/source/user-guide/latest/expressions.md index cf8baac011d..d5ca1c02db4 100644 --- a/docs/source/user-guide/latest/expressions.md +++ b/docs/source/user-guide/latest/expressions.md @@ -715,6 +715,11 @@ This list is illustrative, not exhaustive: the per-function tables are not the c ## make_date evaluation +A valid calendar date outside Date32's range raises `ARITHMETIC_OVERFLOW` in both ANSI +modes. For example, `make_date(6000000, 1, 1)` raises an error even with ANSI disabled. +Invalid calendar fields still return NULL with ANSI disabled and raise a date error with +ANSI enabled; a NULL argument returns NULL before date validation. + With ANSI mode enabled, `make_date(2024, 13, 1)` raises an invalid-date error. If an input iterator yields a valid row followed by that invalid row, a `LIMIT 1` can stop before evaluating the second date. A native projection can evaluate a whole batch @@ -758,12 +763,20 @@ Exchanges materialize their inputs and partition keys independently, so an outer not force an eager date partition key to Spark. Local sorts and aggregates can also separate date evaluation from an outer limit. +Parquet scans below a filter whose `make_date` can throw use Spark's reader. This preserves +Spark's dictionary and metadata pruning, which determine whether an overflowing date is read, +as well as its predicate order and structured errors. The safeguard inspects complete filters, +including projected date arguments and predicates absent from the native scan's data filters. +ANSI-off dates with bounded years and safe arguments can retain the native reader and sibling +row filtering. Predicates containing `make_date` remain above a native scan without weakening +their Boolean structure. Other native scans retain their existing pruning behavior. + Users who accept eagerly evaluating dates, projected arguments, and sibling expressions can set `spark.comet.exec.makeDate.operatorFallback.enabled=false` to allow native operators for these shapes. The default is `true`. Disabling it can expose invalid-date or argument errors -that Spark would skip under ANSI mode. This setting does not change -`make_date`'s invalid-date handling or the expression-level safeguards for skipped -arguments and operands. For example, an empty `IN` predicate containing `make_date` +that Spark would skip, including Date32 overflow with ANSI disabled. This setting does not change +`make_date`'s invalid-date handling, the Spark-reader safeguard, or the expression-level safeguards +for skipped arguments and operands. For example, an empty `IN` predicate containing `make_date` still falls back when needed. ## See also diff --git a/native/spark-expr/src/datetime_funcs/make_date.rs b/native/spark-expr/src/datetime_funcs/make_date.rs index bd7b906a4f4..1d7d876f600 100644 --- a/native/spark-expr/src/datetime_funcs/make_date.rs +++ b/native/spark-expr/src/datetime_funcs/make_date.rs @@ -28,7 +28,7 @@ use std::fmt::{Display, Formatter}; use std::hash::{Hash, Hasher}; use std::sync::Arc; -use crate::{conversion_funcs::ymd_to_epoch_day, SparkError}; +use crate::{arithmetic_overflow_error, conversion_funcs::ymd_to_epoch_day, SparkError}; /// Skips the entire date expression on empty input, before any argument can raise an error. /// `inner` retains the planner's argument evaluation, including NULL short-circuiting when enabled. @@ -107,6 +107,7 @@ impl PhysicalExpr for MakeDateExpr { /// Creates a date from year, month, and day columns. /// For an invalid `(year, month, day)` triple Spark returns NULL when `spark.sql.ansi.enabled` is /// false, and throws otherwise. The ANSI flag is carried here as `fail_on_error`. +/// Valid calendar dates outside Date32 raise arithmetic overflow in either mode. #[derive(Debug, PartialEq, Eq, Hash)] pub struct SparkMakeDate { signature: Signature, @@ -174,10 +175,17 @@ fn cast_to_int32(arr: &Arc) -> Result> { } /// Convert year, month, day to days since Unix epoch (1970-01-01). -/// Returns None if the date is invalid or its epoch day does not fit Date32. -fn make_date(year: i32, month: i32, day: i32) -> Option { - ymd_to_epoch_day(year.into(), month.into(), day.into()) - .and_then(|days| i32::try_from(days).ok()) +/// Invalid Java calendar dates return None; valid dates outside Date32 raise arithmetic overflow. +fn make_date(year: i32, month: i32, day: i32) -> Result> { + if !(-999_999_999..=999_999_999).contains(&year) { + return Ok(None); + } + let Some(days) = ymd_to_epoch_day(year.into(), month.into(), day.into()) else { + return Ok(None); + }; + i32::try_from(days) + .map(Some) + .map_err(|_| arithmetic_overflow_error("integer").into()) } impl ScalarUDFImpl for SparkMakeDate { @@ -247,7 +255,7 @@ impl ScalarUDFImpl for SparkMakeDate { let m = month_array.value(i); let d = day_array.value(i); - match make_date(y, m, d) { + match make_date(y, m, d)? { Some(days) => builder.append_value(days), None => { if self.fail_on_error { @@ -278,6 +286,92 @@ mod tests { use datafusion::physical_expr::expressions::{col, lit, Literal}; use datafusion::physical_expr::{PhysicalExpr, ScalarFunctionExpr}; + fn invoke_make_date(values: [Option; 3], fail_on_error: bool) -> Result { + SparkMakeDate::new(fail_on_error).invoke_with_args(ScalarFunctionArgs { + args: values + .into_iter() + .map(|value| ColumnarValue::Array(Arc::new(Int32Array::from(vec![value])))) + .collect(), + arg_fields: vec![], + number_rows: 1, + return_field: Arc::new(Field::new("make_date", DataType::Date32, true)), + config_options: Arc::new(ConfigOptions::default()), + }) + } + + #[test] + fn valid_dates_outside_date32_overflow_in_both_modes() { + for values in [ + [5_881_580, 7, 12], + [-5_877_641, 6, 22], + [999_999_999, 1, 1], + [-999_999_999, 1, 1], + ] { + for fail_on_error in [false, true] { + let error = invoke_make_date(values.map(Some), fail_on_error).unwrap_err(); + let DataFusionError::External(error) = error else { + panic!("Expected typed SparkError, got {error:?}"); + }; + assert!(matches!( + error.downcast_ref::(), + Some(SparkError::ArithmeticOverflow { from_type, function_name }) + if from_type == "integer" && function_name.is_empty() + )); + } + } + } + + #[test] + fn invalid_calendar_fields_take_precedence_over_date32_overflow() { + for (values, expected) in [ + ( + [1_000_000_000, 13, 32], + "Invalid value for Year (valid values -999999999 - 999999999): 1000000000", + ), + ( + [-1_000_000_000, 1, 1], + "Invalid value for Year (valid values -999999999 - 999999999): -1000000000", + ), + ( + [6_000_000, 13, 32], + "Invalid value for MonthOfYear (valid values 1 - 12): 13", + ), + ([6_000_000, 2, 30], "Invalid date 'FEBRUARY 30'"), + ] { + let result = invoke_make_date(values.map(Some), false) + .unwrap() + .into_array(1) + .unwrap(); + assert!(result.is_null(0)); + let error = invoke_make_date(values.map(Some), true).unwrap_err(); + let DataFusionError::External(error) = error else { + panic!("Expected typed SparkError, got {error:?}"); + }; + assert!(matches!( + error.downcast_ref::(), + Some(SparkError::DatetimeFieldOutOfBounds { range_message }) + if range_message == expected + )); + } + } + + #[test] + fn null_arguments_skip_date32_overflow() { + for values in [ + [None, Some(13), Some(32)], + [Some(6_000_000), None, Some(1)], + [Some(6_000_000), Some(1), None], + ] { + for fail_on_error in [false, true] { + let result = invoke_make_date(values, fail_on_error) + .unwrap() + .into_array(1) + .unwrap(); + assert!(result.is_null(0)); + } + } + } + #[test] fn scalar_arguments_do_not_validate_dates_on_empty_batches() { let expression = MakeDateExpr::new(Arc::new(ScalarFunctionExpr::new( @@ -359,54 +453,55 @@ mod tests { #[test] fn test_make_date_valid() { // Unix epoch - assert_eq!(make_date(1970, 1, 1), Some(0)); + assert_eq!(make_date(1970, 1, 1).unwrap(), Some(0)); // Day after epoch - assert_eq!(make_date(1970, 1, 2), Some(1)); + assert_eq!(make_date(1970, 1, 2).unwrap(), Some(1)); // Day before epoch - assert_eq!(make_date(1969, 12, 31), Some(-1)); + assert_eq!(make_date(1969, 12, 31).unwrap(), Some(-1)); // Leap years - just verify they return Some (valid dates) - assert!(make_date(2000, 2, 29).is_some()); // 2000 is a leap year - assert!(make_date(2004, 2, 29).is_some()); // 2004 is a leap year - // Regular date - assert!(make_date(2023, 6, 15).is_some()); + assert!(make_date(2000, 2, 29).unwrap().is_some()); // 2000 is a leap year + assert!(make_date(2004, 2, 29).unwrap().is_some()); // 2004 is a leap year + + // Regular date + assert!(make_date(2023, 6, 15).unwrap().is_some()); } #[test] fn test_make_date_invalid_month() { - assert_eq!(make_date(2023, 0, 15), None); - assert_eq!(make_date(2023, 13, 15), None); - assert_eq!(make_date(2023, -1, 15), None); + assert_eq!(make_date(2023, 0, 15).unwrap(), None); + assert_eq!(make_date(2023, 13, 15).unwrap(), None); + assert_eq!(make_date(2023, -1, 15).unwrap(), None); } #[test] fn test_make_date_invalid_day() { - assert_eq!(make_date(2023, 6, 0), None); - assert_eq!(make_date(2023, 6, 32), None); - assert_eq!(make_date(2023, 6, -1), None); + assert_eq!(make_date(2023, 6, 0).unwrap(), None); + assert_eq!(make_date(2023, 6, 32).unwrap(), None); + assert_eq!(make_date(2023, 6, -1).unwrap(), None); } #[test] fn test_make_date_invalid_dates() { // Feb 30 never exists - assert_eq!(make_date(2023, 2, 30), None); + assert_eq!(make_date(2023, 2, 30).unwrap(), None); // Feb 29 on non-leap year - assert_eq!(make_date(2023, 2, 29), None); + assert_eq!(make_date(2023, 2, 29).unwrap(), None); // 1900 is not a leap year (divisible by 100 but not 400) - assert_eq!(make_date(1900, 2, 29), None); + assert_eq!(make_date(1900, 2, 29).unwrap(), None); // 2100 will not be a leap year - assert_eq!(make_date(2100, 2, 29), None); + assert_eq!(make_date(2100, 2, 29).unwrap(), None); // April has 30 days - assert_eq!(make_date(2023, 4, 31), None); + assert_eq!(make_date(2023, 4, 31).unwrap(), None); } #[test] fn test_make_date_wide_year_range() { - assert_eq!(make_date(0, 1, 1), Some(-719_528)); - assert_eq!(make_date(-1, 1, 1), Some(-719_893)); - assert_eq!(make_date(300_000, 6, 15), Some(108_853_388)); - assert_eq!(make_date(300_000, 2, 29), Some(108_853_281)); + assert_eq!(make_date(0, 1, 1).unwrap(), Some(-719_528)); + assert_eq!(make_date(-1, 1, 1).unwrap(), Some(-719_893)); + assert_eq!(make_date(300_000, 6, 15).unwrap(), Some(108_853_388)); + assert_eq!(make_date(300_000, 2, 29).unwrap(), Some(108_853_281)); - assert_eq!(make_date(5_881_580, 7, 11), Some(i32::MAX)); - assert_eq!(make_date(-5_877_641, 6, 23), Some(i32::MIN)); + assert_eq!(make_date(5_881_580, 7, 11).unwrap(), Some(i32::MAX)); + assert_eq!(make_date(-5_877_641, 6, 23).unwrap(), Some(i32::MIN)); } } diff --git a/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala b/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala index 955db40ce42..582c3aa4873 100644 --- a/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala +++ b/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala @@ -32,24 +32,26 @@ import scala.jdk.CollectionConverters._ import org.apache.hadoop.conf.Configuration import org.apache.spark.internal.Logging import org.apache.spark.sql.SparkSession -import org.apache.spark.sql.catalyst.expressions.{Attribute, DynamicPruningExpression, Expression, GenericInternalRow, InputFileBlockLength, InputFileBlockStart, InputFileName} +import org.apache.spark.sql.catalyst.expressions.{Alias, Attribute, DynamicPruningExpression, Expression, ExprId, GenericInternalRow, InputFileBlockLength, InputFileBlockStart, InputFileName, MakeDate, NamedExpression} import org.apache.spark.sql.catalyst.rules.Rule import org.apache.spark.sql.catalyst.util.{sideBySide, ArrayBasedMapData, DateTimeUtils, GenericArrayData, MetadataColumnHelper} import org.apache.spark.sql.catalyst.util.ResolveDefaultColumns.getExistenceDefaultValues -import org.apache.spark.sql.comet.{CometBatchScanExec, CometScanExec, CometScanUtils} -import org.apache.spark.sql.execution.{FileSourceScanExec, InSubqueryExec, SparkPlan, SubqueryAdaptiveBroadcastExec} +import org.apache.spark.sql.comet.{CometBatchScanExec, CometBroadcastExchangeExec, CometExec, CometFilterExec, CometProjectExec, CometScanExec, CometScanUtils, CometSinkPlaceHolder} +import org.apache.spark.sql.comet.execution.shuffle.CometShuffleExchangeExec +import org.apache.spark.sql.execution.{FileSourceScanExec, FilterExec, InSubqueryExec, ProjectExec, SparkPlan, SubqueryAdaptiveBroadcastExec} import org.apache.spark.sql.execution.datasources.HadoopFsRelation import org.apache.spark.sql.execution.datasources.v2.BatchScanExec import org.apache.spark.sql.execution.datasources.v2.csv.CSVScan import org.apache.spark.sql.internal.SQLConf import org.apache.spark.sql.types._ -import org.apache.comet.{CometConf, DataTypeSupport, NativeBase} +import org.apache.comet.{CometConf, CometExplainInfo, DataTypeSupport, NativeBase} import org.apache.comet.CometConf._ import org.apache.comet.CometSparkSessionExtensions.{isCometLoaded, isSpark35Plus, withFallbackReason, withFallbackReasons} import org.apache.comet.iceberg.{CometIcebergNativeScanMetadata, IcebergReflection, IcebergStorageSchemes} import org.apache.comet.objectstore.NativeConfig import org.apache.comet.parquet.CometParquetUtils.{encryptionEnabled, isEncryptionConfigSupported, readFieldId} +import org.apache.comet.serde.CometMakeDate import org.apache.comet.serde.operator.{CometIcebergNativeScan, CometNativeScan} import org.apache.comet.shims.{CometTypeShim, ShimCometStreaming, ShimFileFormat, ShimSubqueryBroadcast} @@ -62,6 +64,7 @@ case class CometScanRule(session: SparkSession) with ShimSubqueryBroadcast { private lazy val showTransformations = CometConf.COMET_EXPLAIN_TRANSFORMATIONS.get() + private val makeDateScanReason = "make_date filters require Spark's reader to preserve pruning" override def apply(plan: SparkPlan): SparkPlan = { val newPlan = _apply(plan) @@ -82,6 +85,8 @@ case class CometScanRule(session: SparkSession) // would still rewrite scans to CometScanExec in a streaming plan. if (ShimCometStreaming.isStreamingPlan(plan)) return plan + val sparkScans = scansRequiringSparkForMakeDate(plan) + def isSupportedScanNode(plan: SparkPlan): Boolean = plan match { case _: FileSourceScanExec => true case _: BatchScanExec => true @@ -122,9 +127,123 @@ case class CometScanRule(session: SparkSession) transformV2Scan(fullPlan, scanExec) } - plan.transform { - case scan if isSupportedScanNode(scan) => transformScan(scan) + def transform(node: SparkPlan): SparkPlan = sparkScans.get(node.id) match { + // This guard precedes the contrib hook so Delta keeps its Spark reader too. + case Some(scan) => withFallbackReason(scan, makeDateScanReason) + case None => + val children = node.children.map(transform) + val rebuilt = + if (children == node.children) node + else { + // Re-plan native parents when a restored reader changes their serialized input. + val restored = node match { + case _: CometSinkPlaceHolder => children.head + case native: CometExec => native.sparkFallback(children) + case exchange: CometBroadcastExchangeExec => + exchange.originalPlan.withNewChildren(children) + case exchange: CometShuffleExchangeExec => + exchange.originalPlan.withNewChildren(children) + case _ => node.withNewChildren(children) + } + restored.copyTagsFrom(node) + restored + } + if (isSupportedScanNode(rebuilt)) transformScan(rebuilt) else rebuilt } + transform(plan) + } + + /** + * Spark omits partition-column and nondeterministic predicates from a V1 scan's data filters. + * Keep its original reader when those filters can throw: native metadata pruning need not visit + * the same rows as Spark's per-file statistics and dictionary filters. + */ + private def scansRequiringSparkForMakeDate(plan: SparkPlan): Map[Int, FileSourceScanExec] = { + val scans = mutable.Map.empty[Int, FileSourceScanExec] + + def inspectScan( + node: SparkPlan, + scan: FileSourceScanExec, + filters: Seq[Expression], + aliases: Map[ExprId, Expression]): Unit = { + // This pass owns only its own reason; an earlier parent context must not prevent conversion. + scan.getTagValue(CometExplainInfo.FALLBACK_REASONS).foreach { reasons => + val remaining = reasons - makeDateScanReason + if (remaining.isEmpty) scan.unsetTagValue(CometExplainInfo.FALLBACK_REASONS) + else scan.setTagValue(CometExplainInfo.FALLBACK_REASONS, remaining) + } + // Follow each producer once instead of expanding repeated references through projections. + // These caches belong to this scan's filter/project pipeline, not to the rule instance. + val resolvedAliases = mutable.Map.empty[ExprId, Expression] + val argumentSafety = mutable.Map.empty[ExprId, Boolean] + val throwingDates = mutable.Map.empty[ExprId, Boolean] + + def resolveAlias(expression: Expression): Expression = expression match { + case attribute: Attribute if aliases.contains(attribute.exprId) => + resolvedAliases.getOrElseUpdate( + attribute.exprId, + resolveAlias(aliases(attribute.exprId))) + case _ => expression + } + + def canEvaluateArgument(expression: Expression): Boolean = + CometMakeDate.canEvaluateEagerly(expression) && expression.references.forall { + attribute => + aliases.get(attribute.exprId).forall { producer => + argumentSafety.getOrElseUpdate(attribute.exprId, canEvaluateArgument(producer)) + } + } + + def containsThrowingDate(expression: Expression): Boolean = expression match { + case attribute: Attribute => + aliases.get(attribute.exprId).exists { producer => + throwingDates.getOrElseUpdate(attribute.exprId, containsThrowingDate(producer)) + } + case date: MakeDate => + val resolved = date.copy( + year = resolveAlias(date.year), + month = resolveAlias(date.month), + day = resolveAlias(date.day)) + !MakeDateEvaluation.canEvaluateMakeDateEagerly(resolved, canEvaluateArgument) + case _ => expression.children.exists(containsThrowingDate) + } + + if ((filters ++ scan.dataFilters).exists(containsThrowingDate)) { + scans(node.id) = scan + } + } + + def projectAliases(expressions: Seq[NamedExpression]): Map[ExprId, Expression] = + expressions.collect { case alias: Alias => alias.exprId -> alias.child }.toMap + + def visit(node: SparkPlan, filters: Seq[Expression], aliases: Map[ExprId, Expression]): Unit = + node match { + case scan: FileSourceScanExec => inspectScan(scan, scan, filters, aliases) + case scan: CometScanExec => + val restored = scan.wrapped.copy( + partitionFilters = scan.partitionFilters, + dataFilters = scan.dataFilters) + restored.copyTagsFrom(scan.wrapped) + inspectScan(scan, restored, filters, aliases) + case native: CometExec if native.originalPlan.isInstanceOf[FileSourceScanExec] => + inspectScan( + native, + native.sparkFallback(Nil).asInstanceOf[FileSourceScanExec], + filters, + aliases) + case filter: FilterExec => visit(filter.child, filters :+ filter.condition, aliases) + case filter: CometFilterExec => visit(filter.child, filters :+ filter.condition, aliases) + case project: ProjectExec => + visit(project.child, filters, aliases ++ projectAliases(project.projectList)) + case project: CometProjectExec => + visit(project.child, filters, aliases ++ projectAliases(project.projectList)) + // Only the same filter/project pipeline shares predicate evaluation order. Joins and + // materializing operators must not propagate this guard into unrelated scans. + case _ => node.children.foreach(visit(_, Nil, Map.empty)) + } + + visit(plan, Nil, Map.empty) + scans.toMap } /** diff --git a/spark/src/main/scala/org/apache/comet/rules/MakeDateEvaluation.scala b/spark/src/main/scala/org/apache/comet/rules/MakeDateEvaluation.scala index 3a6f0d6c196..91b196d9452 100644 --- a/spark/src/main/scala/org/apache/comet/rules/MakeDateEvaluation.scala +++ b/spark/src/main/scala/org/apache/comet/rules/MakeDateEvaluation.scala @@ -90,22 +90,30 @@ private[rules] object MakeDateEvaluation { cast.dataType == IntegerType && (cast.child.dataType == ByteType || cast.child.dataType == ShortType) + // An attribute can name a deferred producer; callers must also preserve its evaluation. + def canEvaluateMakeDateEagerly( + date: MakeDate, + canEvaluateArgument: Expression => Boolean = CometMakeDate.canEvaluateEagerly): Boolean = { + val boundedYear = date.year match { + case Literal(year: Int, IntegerType) => + year >= firstCompleteYear && year <= lastCompleteYear + case cast: Cast if isSmallIntegerWidening(cast) => true + case _ => false + } + !date.failOnError && boundedYear && date.children.forall { + case cast: Cast if isSmallIntegerWidening(cast) => + canEvaluateArgument(cast.child) + case child => canEvaluateArgument(child) + } + } + // This only describes computations over their input values. An attribute may still name a // deferred producer, so argument lineage and sibling provenance must be tracked separately. private def canEvaluateProjectionEagerly(expression: Expression): Boolean = expression match { case Alias(child, _) => canEvaluateProjectionEagerly(child) - case date: MakeDate if !date.failOnError => - val boundedYear = date.year match { - case Literal(year: Int, IntegerType) => - year >= firstCompleteYear && year <= lastCompleteYear - case cast: Cast if isSmallIntegerWidening(cast) => true - case _ => false - } - boundedYear && date.children.forall { - case cast: Cast if isSmallIntegerWidening(cast) => - CometMakeDate.canEvaluateEagerly(cast.child) - case child => CometMakeDate.canEvaluateEagerly(child) - } + case _: UnixDate | _: IsNull | _: IsNotNull => + expression.children.forall(canEvaluateProjectionEagerly) + case date: MakeDate => canEvaluateMakeDateEagerly(date) case _ => CometMakeDate.canEvaluateEagerly(expression) } @@ -240,7 +248,12 @@ private[rules] object MakeDateEvaluation { case broadcast: CometBroadcastExchangeExec => broadcast.originalPlan case _ => node } - val evaluatesDate = containsMakeDate(original.expressions) + val evaluatedExpressions = original match { + // Residual data filters run in the enclosing FilterExec, not in the reader. + case scan: FileSourceScanExec => scan.partitionFilters + case _ => original.expressions + } + val evaluatesDate = containsMakeDate(evaluatedExpressions) val exchange = original.isInstanceOf[Exchange] val join = original.isInstanceOf[BaseJoinExec] // Broadcast joins can inline streamed expressions into their consumer. Their build side diff --git a/spark/src/main/scala/org/apache/comet/serde/operator/CometNativeScan.scala b/spark/src/main/scala/org/apache/comet/serde/operator/CometNativeScan.scala index e8efa37202c..59755ef513f 100644 --- a/spark/src/main/scala/org/apache/comet/serde/operator/CometNativeScan.scala +++ b/spark/src/main/scala/org/apache/comet/serde/operator/CometNativeScan.scala @@ -25,7 +25,7 @@ import scala.jdk.CollectionConverters._ import org.apache.hadoop.conf.Configuration import org.apache.spark.internal.Logging import org.apache.spark.sql.catalyst.analysis.Resolver -import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeReference, Expression, Literal} +import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeReference, Expression, Literal, MakeDate} import org.apache.spark.sql.catalyst.util.ResolveDefaultColumns.getExistenceDefaultValues import org.apache.spark.sql.comet.{CometNativeExec, CometNativeScanExec, CometScanExec, CometScanUtils} import org.apache.spark.sql.execution.{FileSourceScanExec, InSubqueryExec, SubqueryAdaptiveBroadcastExec} @@ -327,6 +327,10 @@ object CometNativeScan extends CometOperatorSerde[CometScanExec] with CometTypeS } } + /** Keep each make_date predicate above the scan, without weakening its Boolean structure. */ + def filterForParquetPushdown(filter: Expression): Option[Expression] = + if (filter.exists(_.isInstanceOf[MakeDate])) None else Some(filter) + /** * Build the `NativeScanCommon` proto shared by the core parquet scan and contrib scans that * delegate to the same native parquet machinery (e.g. a Delta scan contrib, which passes @@ -366,7 +370,7 @@ object CometNativeScan extends CometOperatorSerde[CometScanExec] with CometTypeS if (conf.getConf(SQLConf.PARQUET_FILTER_PUSHDOWN_ENABLED)) { commonBuilder.setHasDataFilters(dataFilters.nonEmpty) val filterProtos = new ListBuffer[Expr]() - for (filter <- dataFilters) { + for (filter <- dataFilters.flatMap(filterForParquetPushdown)) { exprToProto(filter, output) match { case Some(proto) => filterProtos += proto case _ => diff --git a/spark/src/main/scala/org/apache/spark/sql/comet/CometNativeScanExec.scala b/spark/src/main/scala/org/apache/spark/sql/comet/CometNativeScanExec.scala index 8c8456b074a..f5c160c216a 100644 --- a/spark/src/main/scala/org/apache/spark/sql/comet/CometNativeScanExec.scala +++ b/spark/src/main/scala/org/apache/spark/sql/comet/CometNativeScanExec.scala @@ -40,6 +40,7 @@ import com.google.common.base.Objects import org.apache.comet.parquet.CometParquetUtils import org.apache.comet.serde.OperatorOuterClass.Operator import org.apache.comet.serde.QueryPlanSerde.exprToProto +import org.apache.comet.serde.operator.CometNativeScan import org.apache.comet.shims.ShimFileFormat /** @@ -198,7 +199,7 @@ case class CometNativeScanExec( if (resolvedFilters.nonEmpty) { val commonBuilder = base.toBuilder commonBuilder.setHasDataFilters(true) - for (filter <- resolvedFilters) { + for (filter <- resolvedFilters.flatMap(CometNativeScan.filterForParquetPushdown)) { exprToProto(filter, output) match { case Some(proto) => commonBuilder.addDataFilters(proto) case _ => diff --git a/spark/src/test/resources/sql-tests/expressions/datetime/make_date.sql b/spark/src/test/resources/sql-tests/expressions/datetime/make_date.sql index bd2577cd81e..2ae30ee5589 100644 --- a/spark/src/test/resources/sql-tests/expressions/datetime/make_date.sql +++ b/spark/src/test/resources/sql-tests/expressions/datetime/make_date.sql @@ -15,6 +15,8 @@ -- specific language governing permissions and limitations -- under the License. +-- Config: spark.sql.ansi.enabled=false + statement CREATE TABLE test_make_date(year int, month int, day int) USING parquet @@ -177,3 +179,37 @@ SELECT make_date(2024, 2, 29) query SELECT make_date(2023, 2, 28) + +-- The Date32 endpoints remain valid, and NULL arguments skip overflow validation. +statement +CREATE TABLE make_date_bounds(y INT, m INT, d INT) USING parquet + +statement +INSERT INTO make_date_bounds VALUES + (5881580, 7, 11), (-5877641, 6, 23), + (NULL, 13, 32), (6000000, NULL, 1), (6000000, 1, NULL) + +query expect_native(make_date) +SELECT unix_date(make_date(y, m, d)) FROM make_date_bounds + +-- Valid calendar dates just outside Date32 must overflow, regardless of ANSI mode. +statement +CREATE TABLE make_date_overflow(y INT, m INT, d INT) USING parquet + +statement +INSERT INTO make_date_overflow VALUES (5881580, 7, 12), (-5877641, 6, 22) + +query expect_error_class(ARITHMETIC_OVERFLOW) +SELECT make_date(y, m, d) FROM make_date_overflow WHERE y > 0 + +query expect_error_class(ARITHMETIC_OVERFLOW) +SELECT make_date(y, m, d) FROM make_date_overflow WHERE y < 0 + +-- An empty input must not evaluate overflowing scalar arguments. +query expect_native(make_date) +SELECT make_date(6000000, 1, 1) FROM make_date_overflow WHERE y = 0 + +-- Calendar validation precedes the checked epoch-day conversion. +query +SELECT make_date(6000000, 13, 32), make_date(6000000, 2, 30), + make_date(1000000000, 1, 1), make_date(-1000000000, 1, 1) diff --git a/spark/src/test/resources/sql-tests/expressions/datetime/make_date_ansi.sql b/spark/src/test/resources/sql-tests/expressions/datetime/make_date_ansi.sql index 74bcce98bf0..9e9d32931ac 100644 --- a/spark/src/test/resources/sql-tests/expressions/datetime/make_date_ansi.sql +++ b/spark/src/test/resources/sql-tests/expressions/datetime/make_date_ansi.sql @@ -203,3 +203,42 @@ SET spark.sql.subexpressionElimination.enabled=false query expect_native(make_date) SELECT make_date(y, 1, d + 1), d + 1 FROM make_date_shared + +-- The Date32 endpoints remain valid, and NULL arguments skip overflow validation. +statement +CREATE TABLE make_date_bounds(y INT, m INT, d INT) USING parquet + +statement +INSERT INTO make_date_bounds VALUES + (5881580, 7, 11), (-5877641, 6, 23), + (NULL, 13, 32), (6000000, NULL, 1), (6000000, 1, NULL) + +query expect_native(make_date) +SELECT unix_date(make_date(y, m, d)) FROM make_date_bounds + +-- Valid calendar dates just outside Date32 must overflow, regardless of ANSI mode. +statement +CREATE TABLE make_date_overflow(y INT, m INT, d INT) USING parquet + +statement +INSERT INTO make_date_overflow VALUES (5881580, 7, 12), (-5877641, 6, 22) + +query expect_error_class(ARITHMETIC_OVERFLOW) +SELECT make_date(y, m, d) FROM make_date_overflow WHERE y > 0 + +query expect_error_class(ARITHMETIC_OVERFLOW) +SELECT make_date(y, m, d) FROM make_date_overflow WHERE y < 0 + +-- An empty input must not evaluate overflowing scalar arguments. +query expect_native(make_date) +SELECT make_date(6000000, 1, 1) FROM make_date_overflow WHERE y = 0 + +-- Invalid calendar fields must report their own error, not Date32 overflow. +query expect_error(MonthOfYear) +SELECT make_date(6000000, 13, 32) + +query expect_error(Invalid date) +SELECT make_date(6000000, 2, 30) + +query expect_error(Year) +SELECT make_date(1000000000, 13, 32) diff --git a/spark/src/test/scala/org/apache/comet/CometTemporalExpressionSuite.scala b/spark/src/test/scala/org/apache/comet/CometTemporalExpressionSuite.scala index c9e97bbfddd..63b0e06724f 100644 --- a/spark/src/test/scala/org/apache/comet/CometTemporalExpressionSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometTemporalExpressionSuite.scala @@ -59,7 +59,7 @@ class CometTemporalExpressionSuite extends CometTestBase with AdaptiveSparkPlanH .getOrElse( fail(s"No SparkThrowable in cause chain: ${causeChain(error).map(_.getClass.getName)}")) - test("make_date does not evaluate an invalid row beyond a partial consumer") { + test("make_date does not evaluate errors beyond a partial consumer") { import testImplicits._ withSQLConf( @@ -75,7 +75,8 @@ class CometTemporalExpressionSuite extends CometTestBase with AdaptiveSparkPlanH .coalesce(1) .selectExpr( "CAST(CASE WHEN id = 0 THEN 1 ELSE 13 END AS INT) AS m", - "CASE WHEN id = 0 THEN '2024' ELSE 'bad' END AS year_text") + "CASE WHEN id = 0 THEN '2024' ELSE 'bad' END AS year_text", + "CAST(CASE WHEN id = 0 THEN 2024 ELSE 6000000 END AS INT) AS y") .write .parquet(path.getCanonicalPath) } @@ -117,6 +118,25 @@ class CometTemporalExpressionSuite extends CometTestBase with AdaptiveSparkPlanH } } + // Date32 overflow throws even with ANSI off, but only on rows Spark consumes. + for (ansi <- Seq(false, true)) { + withSQLConf(SQLConf.ANSI_ENABLED.key -> ansi.toString) { + val overflow = "SELECT unix_date(make_date(y, 1, 1)) FROM make_date_limit" + checkSparkAnswerAndOperator( + sql(overflow + " LIMIT 1"), + includeClasses = Seq(classOf[CometProjectExec], classOf[CometLocalLimitExec])) + partialConsumers.foreach { case (name, consume) => + withClue(s"$name ansi=$ansi codegen=$codegen: ") { + withSQLConf(CometConf.COMET_ENABLED.key -> "false") { + expected = consume(sql(overflow)) + } + assert(expected == Seq(Row(19723))) + assert(consume(sql(overflow)) == expected) + } + } + } + } + // LIMIT can also skip an eager date argument produced by an earlier projection. val yearQuery = """ |SELECT r, unix_date(make_date(y, 1, 1)), r + 1 FROM ( @@ -183,7 +203,7 @@ class CometTemporalExpressionSuite extends CometTestBase with AdaptiveSparkPlanH Seq( "SELECT next_day(date('2024-01-01'), 'NOT_A_DAY')", "SELECT make_date(2024, 13, 1)", - // 999999999 instead overflows the epoch-day conversion as a plain ArithmeticException. + // 999999999 instead raises ARITHMETIC_OVERFLOW during the epoch-day conversion. "SELECT make_date(1000000000, 1, 1)", "SELECT make_date(1000000000, 13, 0)") .foreach { query => diff --git a/spark/src/test/scala/org/apache/comet/exec/CometNativeReaderSuite.scala b/spark/src/test/scala/org/apache/comet/exec/CometNativeReaderSuite.scala index 7e35d22002d..3cd3b1b1167 100644 --- a/spark/src/test/scala/org/apache/comet/exec/CometNativeReaderSuite.scala +++ b/spark/src/test/scala/org/apache/comet/exec/CometNativeReaderSuite.scala @@ -29,15 +29,18 @@ import org.apache.parquet.hadoop.api.WriteSupport import org.apache.parquet.hadoop.api.WriteSupport.WriteContext import org.apache.parquet.io.api.RecordConsumer import org.apache.parquet.schema.MessageTypeParser +import org.apache.spark.SparkThrowable import org.apache.spark.sql.{CometTestBase, DataFrame, Row} +import org.apache.spark.sql.catalyst.expressions.{MakeDate, Or} import org.apache.spark.sql.comet.CometNativeScanExec +import org.apache.spark.sql.execution.{FileSourceScanExec, ScalarSubquery} import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper import org.apache.spark.sql.functions.{array, col} import org.apache.spark.sql.internal.SQLConf import org.apache.spark.sql.types._ import org.apache.comet.{CometConf, CometNativeException, ExtendedExplainInfo} -import org.apache.comet.CometSparkSessionExtensions.isSpark41Plus +import org.apache.comet.CometSparkSessionExtensions.{isSpark40Plus, isSpark41Plus} class CometNativeReaderSuite extends CometTestBase with AdaptiveSparkPlanHelper { @@ -1568,6 +1571,296 @@ class CometNativeReaderSuite extends CometTestBase with AdaptiveSparkPlanHelper } } + private def assertSparkParquetReader(df: DataFrame): FileSourceScanExec = { + val plan = df.queryExecution.executedPlan + assert(find(plan)(_.isInstanceOf[CometNativeScanExec]).isEmpty, plan.treeString) + find(plan)(_.isInstanceOf[FileSourceScanExec]) + .getOrElse(fail(s"Expected Spark's Parquet reader:\n${plan.treeString}")) + .asInstanceOf[FileSourceScanExec] + } + + private def checkMakeDateOverflow(df: DataFrame): Unit = { + assertSparkParquetReader(df) + val (sparkError, cometError) = checkSparkAnswerMaybeThrows(df) + val failures = Seq(sparkError, cometError).map { error => + val failure = error.getOrElse(fail("Expected make_date overflow")) + val chain = causeChain(failure) + assert(!chain.exists(_.isInstanceOf[CometNativeException]), failure.toString) + chain.collect { case e: SparkThrowable with Throwable => e }.lastOption.getOrElse { + fail(s"Expected a SparkThrowable: $failure") + } + } + assert(failures.head.getErrorClass == "ARITHMETIC_OVERFLOW") + assert(failures(1).getClass == failures.head.getClass) + assert(failures(1).getErrorClass == failures.head.getErrorClass) + assert(failures(1).getSqlState == failures.head.getSqlState) + } + + test("make_date predicates preserve evaluation order and errors with row-level pushdown") { + withSQLConf( + SQLConf.ANSI_ENABLED.key -> "false", + SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false", + CometConf.COMET_PARQUET_ROW_FILTER_PUSHDOWN_ENABLED.key -> "true") { + withTempPath { path => + withSQLConf(CometConf.COMET_ENABLED.key -> "false") { + spark + .range(1024) + .coalesce(1) + .selectExpr( + "CAST(IF(id = 1023, 6000000, 2024) AS INT) AS y", + "IF(id = 1023, '', sha2(CAST(id AS STRING), 512)) AS mask", + "IF(id = 1023, 0, 1) AS keep") + .write + .parquet(path.toString) + } + withTempView("make_date_filters") { + spark.read.parquet(path.toString).createOrReplaceTempView("make_date_filters") + val datePredicate = "unix_date(make_date(y, 1, 1)) > 0" + val masked = sql( + s"SELECT count(*) FROM make_date_filters WHERE length(mask) > 0 AND $datePredicate") + assertSparkParquetReader(masked) + checkSparkAnswer(masked) + assert(masked.collect().toSeq == Seq(Row(1023L))) + + // A later mask must not hide an overflow that Spark evaluates first. The unmasked + // predicate must also preserve Spark's structured error outside Parquet's RowFilter. + Seq( + datePredicate, + s"$datePredicate AND length(mask) > 0", + // OR keeps the date predicate in Spark; its serializable sibling must not + // become a row filter that hides an error before Spark evaluates the OR. + s"(y = 0 OR $datePredicate) AND keep = 1").foreach { predicate => + val df = sql(s"SELECT count(*) FROM make_date_filters WHERE $predicate") + checkMakeDateOverflow(df) + } + } + } + } + } + + Seq( + "partition column" -> "unix_date(make_date(y, m, 1)) > 0", + "nondeterministic bound" -> "unix_date(make_date(y, 1, 1)) > rand(42)").foreach { + case (shape, datePredicate) => + test(s"make_date predicates outside scan data filters preserve errors: $shape") { + withSQLConf( + SQLConf.ANSI_ENABLED.key -> "false", + SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false", + CometConf.COMET_PARQUET_ROW_FILTER_PUSHDOWN_ENABLED.key -> "true") { + withTempPath { path => + withSQLConf(CometConf.COMET_ENABLED.key -> "false") { + spark + .range(0L, 1024L, 1L, 1) + .selectExpr( + "CAST(IF(id = 1023, 6000000, 2024) AS INT) AS y", + "IF(id = 1023, '', sha2(CAST(id AS STRING), 512)) AS mask", + "1 AS m") + .write + .partitionBy("m") + .parquet(path.toString) + } + withTempView("make_date_residual_filters") { + spark.read + .parquet(path.toString) + .createOrReplaceTempView("make_date_residual_filters") + val df = sql( + "SELECT count(*) FROM make_date_residual_filters " + + s"WHERE $datePredicate AND length(mask) > 0") + val scan = assertSparkParquetReader(df) + // Spark omitted the date predicate before Comet saw the scan's data filters. + assert(!scan.dataFilters.exists(_.exists(_.isInstanceOf[MakeDate]))) + checkMakeDateOverflow(df) + } + } + } + } + } + + test("make_date filters preserve Spark Parquet pruning and safe native scans") { + withSQLConf( + SQLConf.ANSI_ENABLED.key -> "false", + SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false", + "parquet.filter.dictionary.enabled" -> "true") { + withTempPath { path => + withSQLConf(CometConf.COMET_ENABLED.key -> "false") { + spark + .range(0L, 128L, 1L, 1) + .selectExpr( + "CAST(IF(id = 127, 6000000, 2024) AS INT) AS y", + "IF(id = 127, 13, 1) AS m", + "IF(id % 2 = 0, 1, 3) AS k", + "0 AS a", + "0 AS b", + "42D AS f", + "DATE '2020-01-01' AS d", + "TIMESTAMP_NTZ '2020-01-01' AS nt", + "CAST(1 AS DECIMAL(10, 2)) AS amount") + .write + .option("parquet.enable.dictionary", "true") + .parquet(path.toString) + } + withTempView("make_date_metadata") { + spark.read.parquet(path.toString).createOrReplaceTempView("make_date_metadata") + def filtered(predicate: String): DataFrame = sql( + "SELECT count(*) FROM make_date_metadata " + + s"WHERE unix_date(make_date(y, 1, 1)) > 0 AND $predicate") + def checkPruned(predicate: String): Unit = { + val df = filtered(predicate) + val scan = assertSparkParquetReader(df) + checkSparkAnswer(df) + assert(df.collect().toSeq == Seq(Row(0L))) + assert(scan.metrics("numOutputRows").value == 0) + } + val literalPredicates = Seq( + SQLConf.PARQUET_FILTER_PUSHDOWN_DATE_ENABLED.key -> "d < DATE '1900-01-01'", + SQLConf.PARQUET_FILTER_PUSHDOWN_DECIMAL_ENABLED.key -> + "amount < CAST(0 AS DECIMAL(10, 2))") + // Spark 4 also pushes scalar subqueries, which are resolved after planning. + val predicates = literalPredicates ++ (if (isSpark40Plus) { + Seq( + SQLConf.PARQUET_FILTER_PUSHDOWN_DATE_ENABLED.key -> + "d < (SELECT DATE '1900-01-01' + CAST(min(id) AS INT) FROM range(1))") + } else { + Seq.empty + }) + for (rowPushdown <- Seq(false, true)) { + withSQLConf( + CometConf.COMET_PARQUET_ROW_FILTER_PUSHDOWN_ENABLED.key -> rowPushdown.toString) { + for ((setting, predicate) <- predicates) { + withSQLConf(setting -> "false") { + val df = filtered(predicate) + val scan = assertSparkParquetReader(df) + if (predicate.contains("SELECT")) { + assert(scan.dataFilters.exists(_.exists(_.isInstanceOf[ScalarSubquery]))) + } + checkMakeDateOverflow(df) + } + withSQLConf(setting -> "true") { + checkPruned(predicate) + } + } + + // Spark rejects this entire OR during V1 translation. Dropping months_between + // first would expose a = 1 OR b = 1 and incorrectly discard the row group. + withSQLConf(SQLConf.PARQUET_FILTER_PUSHDOWN_DATE_ENABLED.key -> "false") { + checkMakeDateOverflow( + filtered("((months_between(d, DATE '2020-01-01') > 0 AND a = 1) OR b = 1)")) + } + // NTZ files have isAdjustedToUTC=false, so Spark cannot push this comparison. + checkMakeDateOverflow(filtered("nt < TIMESTAMP_NTZ '2000-01-01'")) + + // Preserve both float statistics and dictionary pruning. The dictionary excludes + // k = 2 even though it lies within the row group's min/max range [1, 3]. + checkPruned("f < 0D") + checkPruned("k = 2") + // Session SQLConf entries are copied into Hadoop conf without stripping a prefix. + withSQLConf("parquet.filter.dictionary.enabled" -> "false") { + checkMakeDateOverflow(filtered("k = 2")) + // This independent predicate still prunes through min/max statistics. + checkPruned("k = 5") + } + } + } + + // A bounded ANSI-off date cannot throw. Keep the native reader and sibling row + // filtering, including when its invalid month makes the final date NULL. + withSQLConf(CometConf.COMET_PARQUET_ROW_FILTER_PUSHDOWN_ENABLED.key -> "true") { + val safe = sql("""SELECT count(*) FROM make_date_metadata + |WHERE unix_date(make_date(2024, m, 1)) > 0 AND y = 6000000""".stripMargin) + val (_, plan) = checkSparkAnswer(safe) + val scan = find(plan)(_.isInstanceOf[CometNativeScanExec]) + .getOrElse(fail("Expected a native Parquet scan for the bounded date")) + .asInstanceOf[CometNativeScanExec] + assert(scan.metrics("pushdown_rows_pruned").value > 0) + } + } + } + } + } + + test("make_date disjunctions are not weakened before Parquet metadata pruning") { + withSQLConf( + SQLConf.ANSI_ENABLED.key -> "false", + SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false", + SQLConf.PARQUET_FILTER_PUSHDOWN_DATE_ENABLED.key -> "false") { + withTempPath { path => + withSQLConf(CometConf.COMET_ENABLED.key -> "false") { + spark + .range(0L, 1024L, 1L, 1) + .selectExpr( + "CAST(IF(id = 1023, 6000000, 2024) AS INT) AS y", + "0 AS a", + "0 AS b", + "0 AS keep") + .write + .parquet(path.toString) + } + withTempView("make_date_disjunction") { + spark.read.parquet(path.toString).createOrReplaceTempView("make_date_disjunction") + val query = """SELECT count(*) FROM make_date_disjunction + |WHERE ((unix_date(make_date(y, 1, 1)) > 0 AND a = 1) OR b = 1)""".stripMargin + for (rowPushdown <- Seq(false, true)) { + withSQLConf( + CometConf.COMET_PARQUET_ROW_FILTER_PUSHDOWN_ENABLED.key -> rowPushdown.toString) { + val df = sql(query) + // Removing make_date would leave a = 1 OR b = 1, whose statistics discard the + // entire row group before Spark can observe its overflowing year. + checkMakeDateOverflow(df) + + // An independent predicate may still prune the group. Keep the original OR + // in the plan so this exercises metadata pruning rather than simplification. + val pruned = sql(s"$query AND keep = 1") + checkSparkAnswer(pruned) + val scan = assertSparkParquetReader(pruned) + assert(scan.dataFilters.exists { + case predicate: Or => predicate.exists(_.isInstanceOf[MakeDate]) + case _ => false + }) + assert(pruned.collect().toSeq == Seq(Row(0L))) + assert(scan.metrics("numOutputRows").value == 0) + } + } + } + } + } + } + + test("make_date scalar filters respect disabled Parquet filter pushdown") { + assume(isSpark40Plus, "SPARK-43402 scalar subquery pushdown is Spark 4.0+ only") + withSQLConf( + SQLConf.ANSI_ENABLED.key -> "false", + SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false", + SQLConf.PARQUET_FILTER_PUSHDOWN_ENABLED.key -> "false") { + withTempPath { path => + withSQLConf(CometConf.COMET_ENABLED.key -> "false") { + spark + .range(0L, 1024L, 1L, 1) + .selectExpr("CAST(IF(id = 1023, 6000000, 2024) AS INT) AS y", "0 AS keep") + .write + .parquet(path.toString) + } + withTempView("make_date_subquery", "make_date_bounds") { + spark.read.parquet(path.toString).createOrReplaceTempView("make_date_subquery") + spark + .range(1) + .selectExpr("CAST(id AS INT) AS v") + .createOrReplaceTempView("make_date_bounds") + val df = sql(""" + |SELECT count(*) FROM make_date_subquery + |WHERE unix_date(make_date(y, 1, 1)) > (SELECT max(v) FROM make_date_bounds) + | AND keep > (SELECT max(v) FROM make_date_bounds) + |""".stripMargin) + val scan = assertSparkParquetReader(df) + assert(scan.dataFilters.exists(_.exists(_.isInstanceOf[ScalarSubquery]))) + + // keep is zero throughout the file. Statistics must not prune its row group before + // Spark evaluates the earlier date predicate, because pushdown was explicitly disabled. + checkMakeDateOverflow(df) + } + } + } + } + test("datafusion escape hatch override wins over rowFilterPushdown.enabled") { // Regression test for the precedence fix: an explicit `spark.comet.datafusion. // execution.parquet.pushdown_filters=false` (behind respectDataFusionConfigs) must not be diff --git a/spark/src/test/scala/org/apache/comet/rules/CometExecRuleSuite.scala b/spark/src/test/scala/org/apache/comet/rules/CometExecRuleSuite.scala index ffbe1d0e1b0..c0592598387 100644 --- a/spark/src/test/scala/org/apache/comet/rules/CometExecRuleSuite.scala +++ b/spark/src/test/scala/org/apache/comet/rules/CometExecRuleSuite.scala @@ -29,7 +29,7 @@ import org.scalatest.PrivateMethodTester._ import org.apache.logging.log4j.Level import org.apache.spark.sql._ import org.apache.spark.sql.catalyst.{FunctionIdentifier, InternalRow} -import org.apache.spark.sql.catalyst.expressions.{Alias, Ascending, Attribute, AttributeReference, BoundReference, Descending, Expression, ExpressionInfo, GreaterThan, In, InSet, KnownFloatingPointNormalized, Literal, MakeDate, Not, RuntimeReplaceable, SortOrder} +import org.apache.spark.sql.catalyst.expressions.{Add, Alias, Ascending, Attribute, AttributeReference, BoundReference, Cast, Cos, CreateArray, Descending, ElementAt, Expression, ExpressionInfo, GreaterThan, In, InSet, KnownFloatingPointNormalized, Literal, MakeDate, Not, Rand, RuntimeReplaceable, Sin, SortOrder, Unevaluable, UnixDate} import org.apache.spark.sql.catalyst.expressions.aggregate.{AggregateExpression, BloomFilterAggregate, Count, Final, Min, Partial, PartialMerge} import org.apache.spark.sql.catalyst.expressions.objects.{Invoke, StaticInvoke} import org.apache.spark.sql.catalyst.optimizer.{BuildLeft, BuildRight, NormalizeNaNAndZero} @@ -48,7 +48,7 @@ import org.apache.spark.sql.execution.exchange.{BroadcastExchangeExec, ReusedExc import org.apache.spark.sql.execution.joins.{BaseJoinExec, BroadcastHashJoinExec, BroadcastNestedLoopJoinExec, HashedRelationBroadcastMode, SortMergeJoinExec} import org.apache.spark.sql.functions.expr import org.apache.spark.sql.internal.SQLConf -import org.apache.spark.sql.types.{DataTypes, DoubleType, FloatType, StructField, StructType} +import org.apache.spark.sql.types.{DataType, DataTypes, DoubleType, FloatType, StructField, StructType} import org.apache.comet.{CometConf, CometCoverageStats, CometExplainInfo, CometSparkSessionExtensions, ConfigEntry, ExtendedExplainInfo} import org.apache.comet.CometSparkSessionExtensions.{isSpark35Plus, isSpark40Plus, isSpark42Plus, withFallbackReason} @@ -235,8 +235,12 @@ class CometExecRuleSuite extends CometTestBase { withParquetTable(path.getCanonicalPath, "make_date_non_ansi") { // A literal or widened small-integer year cannot overflow the Date32 representation. // The invalid month is harmless with ANSI off, including below a lazy row consumer. - for (year <- Seq("2024", "_1")) { - val df = sql(s"SELECT make_date($year, _2, 1) FROM make_date_non_ansi LIMIT 10") + for (year <- Seq("2024", "_1"); + projection <- Seq( + s"make_date($year, _2, 1)", + s"unix_date(make_date($year, _2, 1))", + s"make_date($year, _2, 1) IS NULL")) { + val df = sql(s"SELECT $projection FROM make_date_non_ansi LIMIT 10") checkSparkAnswerAndOperator(df, includeClasses = Seq(classOf[CometProjectExec])) } @@ -312,6 +316,163 @@ class CometExecRuleSuite extends CometTestBase { } } + test("make_date safe residual filters keep the native Parquet reader") { + withSQLConf( + SQLConf.ANSI_ENABLED.key -> "false", + SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") { + withParquetTable(Seq((1, 1, 1), (13, 1, 2)), "make_date_residual") { + val df = sql(""" + |SELECT _3 FROM make_date_residual + |WHERE unix_date(make_date(2024, _1, _2)) > 0 + |""".stripMargin) + checkSparkAnswer(df) + val plan = df.queryExecution.executedPlan + val scan = plan + .collectFirst { case scan: CometNativeScanExec => scan } + .getOrElse(fail(s"Expected a native Parquet reader:\n${plan.treeString}")) + assert(scan.dataFilters.exists(_.exists(_.isInstanceOf[MakeDate]))) + assert(countOperators(plan, classOf[FileSourceScanExec]) == 0, plan.treeString) + } + } + } + + test("make_date scan fallback follows projected arguments and current native children") { + withSQLConf( + SQLConf.ANSI_ENABLED.key -> "false", + SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false", + CometConf.COMET_CONVERT_FROM_PARQUET_ENABLED.key -> "false") { + withParquetTable(Seq((2024, 1, 1)), "make_date_scan_arguments") { + def input: FileSourceScanExec = + createSparkPlan(spark, "SELECT * FROM make_date_scan_arguments").collectFirst { + case scan: FileSourceScanExec => scan + }.get + val scanRule = CometScanRule(spark) + def filterDate(child: SparkPlan, date: Expression): FilterExec = + FilterExec(GreaterThan(UnixDate(date), Literal(0)), child) + def assertReader(plan: SparkPlan, sparkReader: Boolean): Unit = { + assert( + countOperators(plan, classOf[FileSourceScanExec]) == (if (sparkReader) 1 else 0), + plan.treeString) + assert( + countOperators(plan, classOf[CometScanExec]) + + countOperators(plan, classOf[CometNativeScanExec]) == (if (sparkReader) 0 else 1), + plan.treeString) + } + + // The upper date looks safe until its month alias is traced to the lower producer. + for (computedMonth <- Seq(false, true); converted <- Seq(false, true)) { + val source = if (converted) CometRule(spark).apply(input) else input + val month = + if (computedMonth) ElementAt(CreateArray(Seq(Literal(1))), Literal(0)) + else source.output(1) + val project = ProjectExec( + Seq(Alias(Literal(2024), "y")(), Alias(month, "m")(), source.output(2)), + source) + val date = + MakeDate(project.output(0), project.output(1), project.output(2), failOnError = false) + val result = scanRule(filterDate(project, date)) + assertReader(result, sparkReader = computedMonth) + assertReader(scanRule(result), sparkReader = computedMonth) + } + + // A projected date can be hidden behind an attribute in the enclosing filter. + for (converted <- Seq(false, true)) { + val source = input + val date = MakeDate(source.output.head, Literal(1), Literal(1), failOnError = false) + val original = ProjectExec(Seq(Alias(date, "d")()), source) + val project = if (converted) CometRule(spark).apply(original) else original + val result = scanRule(filterDate(project, project.output.head)) + assertReader(result, sparkReader = true) + assert(countOperators(result, classOf[CometProjectExec]) == 0, result.treeString) + } + + // A materializing boundary prevents an outer filter from changing its source scan. + val source = input + val exchange = ShuffleExchangeExec(SinglePartition, source) + val date = MakeDate(exchange.output.head, Literal(1), Literal(1), failOnError = false) + assertReader(scanRule(filterDate(exchange, date)), sparkReader = false) + + // Re-planning a guarded scan without that parent must release only our own reason. + val guarded = scanRule(filterDate(source, date)).collectFirst { + case scan: FileSourceScanExec => scan + }.get + assertReader(scanRule(guarded), sparkReader = false) + } + } + } + + test("make_date scan fallback visits shared projection dependencies without expanding them") { + withSQLConf( + SQLConf.ANSI_ENABLED.key -> "false", + SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") { + withParquetTable(Seq(Tuple1(2024)), "make_date_scan_dependencies") { + val source = + createSparkPlan(spark, "SELECT * FROM make_date_scan_dependencies").collectFirst { + case scan: FileSourceScanExec => scan + }.get + var expressionVisits = 0 + case class CountedExpression(child: Expression) extends Expression with Unevaluable { + override def children: Seq[Expression] = { + expressionVisits += 1 + assert( + expressionVisits <= 10000, + "Scan analysis expanded shared projection dependencies") + Seq(child) + } + override def nullable: Boolean = true + override def dataType: DataType = DoubleType + override protected def withNewChildrenInternal( + newChildren: IndexedSeq[Expression]): Expression = copy(child = newChildren.head) + } + val findScans = + PrivateMethod[Map[Int, FileSourceScanExec]](Symbol("scansRequiringSparkForMakeDate")) + val scanRule = CometScanRule(spark) + for (throwingDate <- Seq(false, true)) { + val input = if (throwingDate) { + UnixDate(MakeDate(source.output.head, Literal(1), Literal(1), failOnError = false)) + } else { + source.output.head + } + var project = + ProjectExec(Seq(Alias(CountedExpression(Cast(input, DoubleType)), "x")()), source) + for (level <- 0 until 20) { + val previous = project.output.head + val expression = Add(Add(Sin(previous), Cos(previous)), Rand(Literal(level.toLong))) + // Count every layer so alias expansion fails before building an exponential tree. + project = ProjectExec(Seq(Alias(CountedExpression(expression), "x")()), project) + } + val filter = FilterExec(GreaterThan(project.output.head, Literal(0.0)), project) + expressionVisits = 0 + withClue(s"throwingDate=$throwingDate: ") { + val scans = scanRule invokePrivate findScans(filter) + assert(scans.keySet == (if (throwingDate) Set(source.id) else Set.empty[Int])) + } + } + } + } + } + + test("make_date scan fallback preserves current native partition filters") { + withSQLConf( + SQLConf.ANSI_ENABLED.key -> "false", + SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") { + withParquetTable(Seq(Tuple1(2024)), "make_date_scan_reuse") { + val source = createSparkPlan(spark, "SELECT * FROM make_date_scan_reuse").collectFirst { + case scan: FileSourceScanExec => scan + }.get + val native = CometRule(spark).apply(source).asInstanceOf[CometNativeScanExec] + val current = native.copy(partitionFilters = Seq(Literal(true))) + val date = MakeDate(current.output.head, Literal(1), Literal(1), failOnError = false) + val filtered = FilterExec(GreaterThan(UnixDate(date), Literal(0)), current) + val result = CometScanRule(spark).apply(filtered) + val restored = result.collectFirst { case scan: FileSourceScanExec => scan }.get + assert(restored.partitionFilters == current.partitionFilters) + assert(restored.dataFilters == current.dataFilters) + assert(countOperators(result, classOf[CometNativeScanExec]) == 0, result.treeString) + } + } + } + test("make_date collect limit reuses the native projection input cap") { withSQLConf( SQLConf.ANSI_ENABLED.key -> "true", diff --git a/spark/src/test/scala/org/apache/comet/serde/operator/CometNativeScanSuite.scala b/spark/src/test/scala/org/apache/comet/serde/operator/CometNativeScanSuite.scala index a2bcde5d3c8..fd335c539ea 100644 --- a/spark/src/test/scala/org/apache/comet/serde/operator/CometNativeScanSuite.scala +++ b/spark/src/test/scala/org/apache/comet/serde/operator/CometNativeScanSuite.scala @@ -24,7 +24,7 @@ import scala.jdk.CollectionConverters._ import org.scalatest.funsuite.AnyFunSuite import org.apache.hadoop.conf.Configuration -import org.apache.spark.sql.catalyst.expressions.AttributeReference +import org.apache.spark.sql.catalyst.expressions.{And, AttributeReference, EqualTo, GreaterThan, Literal, MakeDate, Or, UnixDate} import org.apache.spark.sql.execution.datasources.parquet.ParquetUtils import org.apache.spark.sql.internal.SQLConf import org.apache.spark.sql.types.{IntegerType, LongType, MetadataBuilder, StringType, StructField, StructType} @@ -109,4 +109,18 @@ class CometNativeScanSuite extends AnyFunSuite { Set("id", colliding, colliding + "_")) assert(fields.map(_.name) == Seq(colliding + "__")) } + + test("Parquet pushdown excludes whole make_date predicates and preserves independent filters") { + val year = AttributeReference("year", IntegerType)() + val a = AttributeReference("a", IntegerType)() + val b = AttributeReference("b", IntegerType)() + val date = GreaterThan( + UnixDate(MakeDate(year, Literal(1), Literal(1), failOnError = false)), + Literal(0)) + val predicate = Or(And(date, EqualTo(a, Literal(1))), EqualTo(b, Literal(1))) + val independent = EqualTo(AttributeReference("keep", IntegerType)(), Literal(1)) + assert( + Seq(predicate, independent).flatMap(CometNativeScan.filterForParquetPushdown) == + Seq(independent)) + } }