From 40cb1b919bb420c1cd38bd688baf47e27583e6b5 Mon Sep 17 00:00:00 2001 From: Ping Zhang Date: Fri, 30 Oct 2026 14:29:19 +0000 Subject: [PATCH 1/6] fix: report make_date Date32 overflow in both ANSI modes --- docs/source/user-guide/latest/expressions.md | 5 + .../src/datetime_funcs/make_date.rs | 155 ++++++++++++++---- .../expressions/datetime/make_date.sql | 36 ++++ .../expressions/datetime/make_date_ansi.sql | 39 +++++ .../comet/CometTemporalExpressionSuite.scala | 26 ++- 5 files changed, 228 insertions(+), 33 deletions(-) diff --git a/docs/source/user-guide/latest/expressions.md b/docs/source/user-guide/latest/expressions.md index cf8baac011d..fde2d0dfaac 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 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/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 => From 6fc85ac7fe1fcb82a4e0187674dad17c3d3c312a Mon Sep 17 00:00:00 2001 From: Ping Zhang Date: Fri, 30 Oct 2026 19:21:44 +0000 Subject: [PATCH 2/6] fix: preserve make_date predicate order in Parquet scans --- native/core/src/parquet/parquet_exec.rs | 27 +++++++-- .../comet/exec/CometNativeReaderSuite.scala | 55 +++++++++++++++++++ 2 files changed, 78 insertions(+), 4 deletions(-) diff --git a/native/core/src/parquet/parquet_exec.rs b/native/core/src/parquet/parquet_exec.rs index f2e3e1e4b6c..271b3bc5e62 100644 --- a/native/core/src/parquet/parquet_exec.rs +++ b/native/core/src/parquet/parquet_exec.rs @@ -39,7 +39,7 @@ use datafusion::physical_expr_adapter::PhysicalExprAdapterFactory; use datafusion::prelude::SessionContext; use datafusion::scalar::ScalarValue; use datafusion_comet_spark_expr::jvm_udf::JvmScalarUdfExpr; -use datafusion_comet_spark_expr::EvalMode; +use datafusion_comet_spark_expr::{EvalMode, MakeDateExpr}; use datafusion_datasource::TableSchema; use parquet::variant::VariantType; use std::collections::HashMap; @@ -111,7 +111,20 @@ pub(crate) fn init_datasource_exec( // `SessionConfig` (an `Arc` plus a small extensions map); `SessionContext:: // state()` would clone the whole `SessionState`, including the function registries (~250+ // UDFs) and optimizer rules, on every scan init. - let session_config = session_ctx.copied_config(); + let mut session_config = session_ctx.copied_config(); + if data_filters + .as_ref() + .is_some_and(|filters| filters.iter().any(contains_make_date)) + { + // RowFilter can reorder predicates and flatten their Spark errors. Even pushing only + // a later sibling could hide a make_date error that Spark evaluates first. Keep all + // row filtering above this scan, while retaining metadata pruning for other predicates. + session_config + .options_mut() + .execution + .parquet + .pushdown_filters = false; + } let (table_parquet_options, mut spark_parquet_options) = get_options( session_timezone, case_sensitive, @@ -291,11 +304,12 @@ pub(crate) fn init_datasource_exec( // `propagation.parent_pushdown_result` because Spark's Filter above the // scan re-evaluates every dataFilter, so No-classified filters stay // correct without us inserting a FilterExec here. That also makes it safe - // to drop the filters that call into the JVM first; see `calls_jvm`. + // to drop the filters that call into the JVM first; see `calls_jvm`. A make_date predicate + // also stays above the scan so its evaluation and structured errors remain with the Filter. let data_filters = data_filters.map(|filters| { filters .into_iter() - .filter(|filter| !calls_jvm(filter)) + .filter(|filter| !calls_jvm(filter) && !contains_make_date(filter)) .collect::>() }); let file_source: Arc = match data_filters { @@ -351,6 +365,11 @@ fn calls_jvm(expr: &Arc) -> bool { .unwrap_or(true) } +fn contains_make_date(expr: &Arc) -> bool { + expr.exists(|e| Ok(e.downcast_ref::().is_some())) + .unwrap_or(true) +} + // Registration URLs use a reserved suffix to distinguish backend/configuration // identities. Encryption must use the physical URI that Spark registered. Match // the complete suffix, from the right, so custom Hadoop schemes are preserved. 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..d3ba8fd6cfc 100644 --- a/spark/src/test/scala/org/apache/comet/exec/CometNativeReaderSuite.scala +++ b/spark/src/test/scala/org/apache/comet/exec/CometNativeReaderSuite.scala @@ -29,6 +29,7 @@ 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.comet.CometNativeScanExec import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper @@ -1568,6 +1569,60 @@ class CometNativeReaderSuite extends CometTestBase with AdaptiveSparkPlanHelper } } + 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") + .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") + assert( + find(masked.queryExecution.executedPlan)( + _.isInstanceOf[CometNativeScanExec]).isDefined) + 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").foreach { predicate => + val df = sql(s"SELECT count(*) FROM make_date_filters WHERE $predicate") + assert( + find(df.queryExecution.executedPlan)(_.isInstanceOf[CometNativeScanExec]).isDefined) + val (sparkError, cometError) = checkSparkAnswerMaybeThrows(df) + val failures = Seq(sparkError, cometError).map { error => + val failure = error.getOrElse(fail(s"Expected overflow for $predicate")) + 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 for $predicate: $failure") + } + } + val expected = failures.head + val actual = failures(1) + assert(expected.getErrorClass == "ARITHMETIC_OVERFLOW") + assert(actual.getClass == expected.getClass) + assert(actual.getErrorClass == expected.getErrorClass) + assert(actual.getSqlState == expected.getSqlState) + } + } + } + } + } + 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 From 7d971f4b3419ac35e5dc6b9df9adbc8d91e66368 Mon Sep 17 00:00:00 2001 From: Ping Zhang Date: Fri, 30 Oct 2026 19:24:46 +0000 Subject: [PATCH 3/6] fix: preserve make_date evaluation for external row iterators --- docs/source/user-guide/latest/expressions.md | 20 ++++- .../comet/rules/MakeDateEvaluation.scala | 22 +++-- .../expressions/datetime/make_date.sql | 21 +++-- .../expressions/datetime/make_date_ansi.sql | 87 ++++++++++--------- .../apache/comet/CometExpressionSuite.scala | 3 +- .../comet/CometTemporalExpressionSuite.scala | 50 ++++++++++- .../comet/rules/CometExecRuleSuite.scala | 37 ++++++-- 7 files changed, 166 insertions(+), 74 deletions(-) diff --git a/docs/source/user-guide/latest/expressions.md b/docs/source/user-guide/latest/expressions.md index fde2d0dfaac..8571b6ec7dd 100644 --- a/docs/source/user-guide/latest/expressions.md +++ b/docs/source/user-guide/latest/expressions.md @@ -727,7 +727,8 @@ before the consuming operator decides which rows it needs. By default, Comet keeps affected operators in Spark to preserve skipped date evaluation. This applies to date projections below `LIMIT`, `SAMPLE`, `FILTER`, joins, -`mapPartitions`, and `Dataset.rdd` consumers; date expressions inside joins, aggregates, +`mapPartitions`, and external row iterators, including Scala's `Dataset.rdd` and PySpark's +`DataFrame.rdd`; date expressions inside joins, aggregates, and ordered limits; and projections below a window group limit. A parent projection can also defer a date output when an `IF`, `COALESCE`, or NULL-skipping operand does not need it. Comet preserves that evaluation across projections while allowing eager date consumers, @@ -740,6 +741,10 @@ the join rejects before producing an output row. Selecting the date eagerly does not force conditionally used sibling expressions; those computations also retain Spark evaluation when needed. +The physical plan can be reused by a later RDD consumer, so uncapped root date projections +also retain these safeguards, even when the current action collects the complete result. +Materializing exchanges, sorts, and aggregates can still allow native date evaluation below them. + With ANSI disabled, a direct date projection can remain native below these consumers when all its computations are safe to evaluate early. This includes `make_date(2024, month, day)` over input columns and years widened from `TINYINT` or `SMALLINT`. An unrestricted integer @@ -755,18 +760,25 @@ because Spark checks that argument on every input row. Filtered counts and count multiple arguments retain the safeguard when they can skip date evaluation. Only affected parts of the plan fall back. An ordinary native scan can remain native. -An adjacent collect limit and date projection can also stay native: Comet caps the projection's -input before evaluating dates, so unread rows cannot raise errors. This requires a supported +An adjacent `LIMIT 1` without `OFFSET` and date projection can also stay native: Comet caps +the projection's input before evaluating dates, so unread rows cannot raise errors. Larger +root limits retain the safeguard because an RDD caller can stop before reaching the limit. +The cap requires a supported native local limit; otherwise the Spark safeguard remains. The cap does not bypass safeguards for a lower argument-producing projection or intervening filters and joins. Exchanges materialize their inputs and partition keys independently, so an outer limit does not force an eager date partition key to Spark. Local sorts and aggregates can also separate date evaluation from an outer limit. +Predicates containing `make_date` stay above the Parquet scan. Comet disables row-level filter +pushdown for that scan so predicates cannot be reordered or hide a required date error, and +Spark's structured error is preserved. Other predicates can still use metadata to skip row groups +and pages. + 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 +that Spark would skip, including Date32 overflow with ANSI disabled. 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` still falls back when needed. 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..81d6a007379 100644 --- a/spark/src/main/scala/org/apache/comet/rules/MakeDateEvaluation.scala +++ b/spark/src/main/scala/org/apache/comet/rules/MakeDateEvaluation.scala @@ -43,6 +43,7 @@ private[rules] object MakeDateEvaluation { private val sparkEvaluation = TreeNodeTag[String]("comet.makeDateSparkEvaluation") private val collectProjectionLimit = TreeNodeTag[Int]("comet.makeDateCollectProjectionLimit") private val collectLimitReason = "make_date requires Spark evaluation below LIMIT" + private val externalRowConsumer = "an external row iterator" // Return a copy: failed cap conversion must leave the original guarded pipeline untouched. def collectLimitProjection(project: ProjectExec, limit: Int): Option[ProjectExec] = @@ -393,7 +394,9 @@ private[rules] object MakeDateEvaluation { case _ => false } val rowReason = if (evaluatesDate && !exchange && !safeProjection) { - rowDemand.map(boundary => s"make_date requires Spark evaluation below $boundary") + rowDemand + .filterNot(boundary => boundary == externalRowConsumer && localMaterialization) + .map(boundary => s"make_date requires Spark evaluation below $boundary") } else None // A safe date producer can remain native, but its NULL output can still hide a throwing // arithmetic operand in the consumer. The expression serde only sees an attribute here. @@ -464,14 +467,19 @@ private[rules] object MakeDateEvaluation { (rebuilt, fallbackReason, dateDerivedOutputs) } - // Dataset.rdd adds an object-producing root, and its caller may read only a few objects. - // EliminateSerialization can leave a typed operator, filter, or project at that root. - def producesObjects(op: SparkPlan): Boolean = op match { - case _: ObjectProducerExec => true - case _: FilterExec | _: ProjectExec => producesObjects(op.children.head) + // QueryExecution.toRdd (also used by PySpark's DataFrame.rdd) exposes this plan's row + // iterator without adding Dataset.rdd's object-producing root. Its caller can stop early, + // even if explain() or another action has already prepared the physical plan. + // A one-row collect limit can retain its native input cap: producing its only row cannot + // skip another row inside that cap. Larger limits and OFFSET do not provide that guarantee. + def isSingleRowCollect(op: SparkPlan): Boolean = op match { + case limit: CollectLimitExec => limit.limit == 1 && limit.offset == 0 + case native: CometExec => isSingleRowCollect(native.originalPlan) + case _: ColumnarToRowExec | _: CometColumnarToRowExec | _: CometNativeColumnarToRowExec => + isSingleRowCollect(op.children.head) case _ => false } - val rootDemand = if (producesObjects(plan)) Some("code reading Dataset.rdd") else None + val rootDemand = if (isSingleRowCollect(plan)) None else Some(externalRowConsumer) visit(plan, rootDemand, None)._1 } } 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 2ae30ee5589..efbdd7e06bc 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 @@ -17,6 +17,9 @@ -- Config: spark.sql.ansi.enabled=false +-- Materialize potentially throwing expression results before exposing a row iterator. +-- Bounded ANSI-off dates can still execute natively without a materialization barrier. + statement CREATE TABLE test_make_date(year int, month int, day int) USING parquet @@ -34,7 +37,7 @@ INSERT INTO test_make_date VALUES -- column arguments query -SELECT year, month, day, make_date(year, month, day) FROM test_make_date ORDER BY year, month, day +SELECT /*+ REPARTITION(1) */ year, month, day, make_date(year, month, day) FROM test_make_date ORDER BY year, month, day -- literal year, column month and day query @@ -42,11 +45,11 @@ SELECT make_date(2023, month, day) FROM test_make_date ORDER BY month, day -- column year, literal month and day query -SELECT make_date(year, 6, 15) FROM test_make_date ORDER BY year +SELECT /*+ REPARTITION(1) */ make_date(year, 6, 15) FROM test_make_date ORDER BY year -- column year and month, literal day query -SELECT make_date(year, month, 1) FROM test_make_date ORDER BY year, month +SELECT /*+ REPARTITION(1) */ make_date(year, month, 1) FROM test_make_date ORDER BY year, month -- literal values query @@ -57,7 +60,7 @@ SELECT make_date(1970, 1, 1) -- null handling with literals query -SELECT make_date(NULL, 1, 1) +SELECT /*+ REPARTITION(1) */ make_date(NULL, 1, 1) query SELECT make_date(2023, NULL, 1) @@ -190,7 +193,7 @@ INSERT INTO make_date_bounds VALUES (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 +SELECT /*+ REPARTITION(1) */ unix_date(make_date(y, m, d)) FROM make_date_bounds -- Valid calendar dates just outside Date32 must overflow, regardless of ANSI mode. statement @@ -200,16 +203,16 @@ 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 +SELECT /*+ REPARTITION(1) */ 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 +SELECT /*+ REPARTITION(1) */ 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 +SELECT /*+ REPARTITION(1) */ 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), +SELECT /*+ REPARTITION(1) */ 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 9e9d32931ac..b438d01689a 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 @@ -30,23 +30,26 @@ -- Sentinel: a valid date must still execute natively under ANSI. This guards against the -- expect_error queries passing vacuously if make_date were to fall back to Spark. +-- Materialize these expression-test results before exposing a row iterator, so native +-- evaluation is safe even when a later consumer reads only a prefix. + query -SELECT make_date(2024, 2, 28) +SELECT /*+ REPARTITION(1) */ make_date(2024, 2, 28) query -SELECT make_date(300000, 6, 15), make_date(300000, 2, 29) +SELECT /*+ REPARTITION(1) */ make_date(300000, 6, 15), make_date(300000, 2, 29) -- February 30 is not a valid date. query expect_error(Invalid date) -SELECT make_date(2024, 2, 30) +SELECT /*+ REPARTITION(1) */ make_date(2024, 2, 30) -- Month 13 is out of range. query expect_error(MonthOfYear) -SELECT make_date(2024, 13, 1) +SELECT /*+ REPARTITION(1) */ make_date(2024, 13, 1) -- Day 0 is out of range. query expect_error(DayOfMonth) -SELECT make_date(2024, 6, 0) +SELECT /*+ REPARTITION(1) */ make_date(2024, 6, 0) statement CREATE TABLE make_date_arguments(y INT, m INT, inner_m INT) USING parquet @@ -57,7 +60,7 @@ INSERT INTO make_date_arguments VALUES -- A nonempty IN evaluates its probe. Keep this two-item list as In rather than InSet. query expect_native(make_date, in) -SELECT make_date(y, m, 1) IN (DATE '2024-01-01', DATE '2024-02-01') +SELECT /*+ REPARTITION(1) */ make_date(y, m, 1) IN (DATE '2024-01-01', DATE '2024-02-01') FROM make_date_arguments statement @@ -69,83 +72,83 @@ INSERT INTO make_date_membership VALUES -- A match or NULL probe skips the throwing cast on that row, even when another row needs it. query expect_fallback(make_date requires Spark evaluation under In with computed candidates) -SELECT make_date(y, 1, 1) IN (DATE '2024-01-01', CAST(s AS DATE)) +SELECT /*+ REPARTITION(1) */ make_date(y, 1, 1) IN (DATE '2024-01-01', CAST(s AS DATE)) FROM make_date_membership -- Structs with floating-point fields use Comet's nested membership implementation. query expect_fallback(make_date requires Spark evaluation under In with computed candidates) -SELECT (make_date(y, 1, 1), 0D) IN ((DATE '2024-01-01', 0D), (CAST(s AS DATE), 0D)) +SELECT /*+ REPARTITION(1) */ (make_date(y, 1, 1), 0D) IN ((DATE '2024-01-01', 0D), (CAST(s AS DATE), 0D)) FROM make_date_membership WHERE y IS NOT NULL query expect_fallback(make_date requires Spark evaluation under In with computed candidates) -SELECT (make_date(y, 1, 1), 0D) NOT IN ((DATE '2024-01-01', 0D), (CAST(s AS DATE), 0D)) +SELECT /*+ REPARTITION(1) */ (make_date(y, 1, 1), 0D) NOT IN ((DATE '2024-01-01', 0D), (CAST(s AS DATE), 0D)) FROM make_date_membership WHERE y IS NOT NULL -- A candidate required by an unmatched probe must still fail. query expect_error(CAST_INVALID_INPUT) -SELECT make_date(y, 1, 1) IN (DATE '2023-01-01', CAST(s AS DATE)) +SELECT /*+ REPARTITION(1) */ make_date(y, 1, 1) IN (DATE '2023-01-01', CAST(s AS DATE)) FROM make_date_membership WHERE y = 2024 -- Safe structs retain the native path through the analyzer's nullability-widening casts. query expect_native(make_date, in) -SELECT (make_date(y, 1, 1), 0D) IN ((DATE '2024-01-01', 0D), (DATE '2025-01-01', 0D)) +SELECT /*+ REPARTITION(1) */ (make_date(y, 1, 1), 0D) IN ((DATE '2024-01-01', 0D), (DATE '2025-01-01', 0D)) FROM make_date_membership -- Year and month NULLs skip the invalid day expression on just those rows. query expect_native(make_date) -SELECT make_date(y, m, dayofmonth(make_date(2024, inner_m, 1))) FROM make_date_arguments +SELECT /*+ REPARTITION(1) */ make_date(y, m, dayofmonth(make_date(2024, inner_m, 1))) FROM make_date_arguments -- A NULL year also skips a computed month. query expect_native(make_date) -SELECT make_date(y, month(make_date(2024, inner_m, 1)), 1) +SELECT /*+ REPARTITION(1) */ make_date(y, month(make_date(2024, inner_m, 1)), 1) FROM make_date_arguments WHERE m IS NOT NULL -- A later NULL must not hide an error in the required first argument. query expect_error(MonthOfYear) -SELECT make_date(year(make_date(2024, inner_m, 1)), 1, y) +SELECT /*+ REPARTITION(1) */ make_date(year(make_date(2024, inner_m, 1)), 1, y) FROM make_date_arguments WHERE y IS NULL -- These date functions always evaluate their first operand, even when it returns NULL. query expect_native(make_date, date_add) -SELECT date_add(make_date(y, m, 1), 1) FROM make_date_arguments +SELECT /*+ REPARTITION(1) */ date_add(make_date(y, m, 1), 1) FROM make_date_arguments query expect_native(make_date, date_sub) -SELECT date_sub(make_date(y, m, 1), 1) FROM make_date_arguments +SELECT /*+ REPARTITION(1) */ date_sub(make_date(y, m, 1), 1) FROM make_date_arguments query expect_native(make_date, date_format) -SELECT date_format(make_date(y, m, 1), 'yyyy-MM-dd') FROM make_date_arguments +SELECT /*+ REPARTITION(1) */ date_format(make_date(y, m, 1), 'yyyy-MM-dd') FROM make_date_arguments query expect_native(make_date, datediff) -SELECT datediff(make_date(y, m, 1), DATE '2024-01-01') FROM make_date_arguments +SELECT /*+ REPARTITION(1) */ datediff(make_date(y, m, 1), DATE '2024-01-01') FROM make_date_arguments -- A NULL first operand still skips the invalid date in the second operand. query expect_fallback(make_date requires Spark evaluation under DateDiff) -SELECT datediff(make_date(y, m, 1), make_date(2024, inner_m, 1)) FROM make_date_arguments +SELECT /*+ REPARTITION(1) */ datediff(make_date(y, m, 1), make_date(2024, inner_m, 1)) FROM make_date_arguments -- Container constructors evaluate every child and do not need a parent fallback. query expect_native(make_date) -SELECT array(make_date(y, m, 1), DATE '2024-01-01'), +SELECT /*+ REPARTITION(1) */ array(make_date(y, m, 1), DATE '2024-01-01'), named_struct('date', make_date(y, m, 1)) FROM make_date_arguments -- Greatest and least evaluate every operand, including nullable dates in either position. query expect_native(make_date, greatest, least) -SELECT greatest(make_date(y, m, 1), DATE '2024-02-01'), +SELECT /*+ REPARTITION(1) */ greatest(make_date(y, m, 1), DATE '2024-02-01'), least(DATE '2024-02-01', make_date(y, m, 1)) FROM make_date_arguments -- Hashes evaluate every operand, even after a NULL, with the date in either position. query expect_native(make_date, hash, xxhash64) -SELECT hash(make_date(y, m, 1), y), xxhash64(y, make_date(y, m, 1)) +SELECT /*+ REPARTITION(1) */ hash(make_date(y, m, 1), y), xxhash64(y, make_date(y, m, 1)) FROM make_date_arguments -- Arithmetic keeps required date operands native when skipped siblings are harmless. query expect_native(make_date) -SELECT unix_date(make_date(y, m, 1)) + 1, +SELECT /*+ REPARTITION(1) */ unix_date(make_date(y, m, 1)) + 1, unix_date(make_date(y, m, 1)) - y, 2 * unix_date(make_date(y, m, 1)) FROM make_date_arguments -- A NULL date skips a throwing right operand, so the arithmetic must still fall back. query expect_fallback(make_date requires Spark evaluation under Add) -SELECT unix_date(make_date(y, 1, 1)) + CAST(s AS INT) +SELECT /*+ REPARTITION(1) */ unix_date(make_date(y, 1, 1)) + CAST(s AS INT) FROM make_date_membership WHERE y IS NULL statement @@ -156,36 +159,36 @@ INSERT INTO make_date_parents VALUES (NULL, true, 13), (1, false, 1) -- Division evaluates its right operand first and skips the date when the divisor is NULL. query expect_fallback(make_date requires Spark evaluation under Divide) -SELECT unix_date(make_date(2024, m, 1)) / lhs FROM make_date_parents +SELECT /*+ REPARTITION(1) */ unix_date(make_date(2024, m, 1)) / lhs FROM make_date_parents -- Native arithmetic/Boolean parents are still eager, so keep these in Spark. query expect_fallback(make_date requires Spark evaluation under) -SELECT lhs + unix_date(make_date(2024, m, 1)) FROM make_date_parents +SELECT /*+ REPARTITION(1) */ lhs + unix_date(make_date(2024, m, 1)) FROM make_date_parents -- An eager inner parent must not remove an outer parent's short-circuit guard. query expect_fallback(make_date requires Spark evaluation under Add) -SELECT lhs + unix_date(greatest(make_date(2024, m, 1), DATE '2024-01-01')) +SELECT /*+ REPARTITION(1) */ lhs + unix_date(greatest(make_date(2024, m, 1), DATE '2024-01-01')) FROM make_date_parents query expect_fallback(make_date requires Spark evaluation under Add) -SELECT lhs + hash(make_date(2024, m, 1), lhs) FROM make_date_parents +SELECT /*+ REPARTITION(1) */ lhs + hash(make_date(2024, m, 1), lhs) FROM make_date_parents query expect_fallback(make_date requires Spark evaluation under Add) -SELECT lhs + (unix_date(make_date(2024, m, 1)) + 1) FROM make_date_parents +SELECT /*+ REPARTITION(1) */ lhs + (unix_date(make_date(2024, m, 1)) + 1) FROM make_date_parents query expect_fallback(make_date requires Spark evaluation under) -SELECT guard OR unix_date(make_date(2024, m, 1)) > 0 FROM make_date_parents +SELECT /*+ REPARTITION(1) */ guard OR unix_date(make_date(2024, m, 1)) > 0 FROM make_date_parents -- Existing conditional expressions safely pass only selected rows to make_date. query expect_native(make_date) -SELECT IF(guard, DATE '2024-01-01', make_date(2024, m, 1)) FROM make_date_parents +SELECT /*+ REPARTITION(1) */ IF(guard, DATE '2024-01-01', make_date(2024, m, 1)) FROM make_date_parents query expect_native(make_date) -SELECT CASE WHEN guard THEN DATE '2024-01-01' ELSE make_date(2024, m, 1) END +SELECT /*+ REPARTITION(1) */ CASE WHEN guard THEN DATE '2024-01-01' ELSE make_date(2024, m, 1) END FROM make_date_parents query expect_native(make_date) -SELECT COALESCE(CASE WHEN guard THEN DATE '2024-01-01' END, make_date(2024, m, 1)) +SELECT /*+ REPARTITION(1) */ COALESCE(CASE WHEN guard THEN DATE '2024-01-01' END, make_date(2024, m, 1)) FROM make_date_parents statement @@ -196,13 +199,13 @@ INSERT INTO make_date_shared VALUES (NULL, 0), (2024, 0) -- Spark computes the shared day expression outside the constructor's NULL checks. query expect_fallback(make_date requires Spark evaluation when a computed argument) -SELECT make_date(y, 1, d + 1), d + 1 FROM make_date_shared +SELECT /*+ REPARTITION(1) */ make_date(y, 1, d + 1), d + 1 FROM make_date_shared statement SET spark.sql.subexpressionElimination.enabled=false query expect_native(make_date) -SELECT make_date(y, 1, d + 1), d + 1 FROM make_date_shared +SELECT /*+ REPARTITION(1) */ make_date(y, 1, d + 1), d + 1 FROM make_date_shared -- The Date32 endpoints remain valid, and NULL arguments skip overflow validation. statement @@ -214,7 +217,7 @@ INSERT INTO make_date_bounds VALUES (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 +SELECT /*+ REPARTITION(1) */ unix_date(make_date(y, m, d)) FROM make_date_bounds -- Valid calendar dates just outside Date32 must overflow, regardless of ANSI mode. statement @@ -224,21 +227,21 @@ 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 +SELECT /*+ REPARTITION(1) */ 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 +SELECT /*+ REPARTITION(1) */ 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 +SELECT /*+ REPARTITION(1) */ 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) +SELECT /*+ REPARTITION(1) */ make_date(6000000, 13, 32) query expect_error(Invalid date) -SELECT make_date(6000000, 2, 30) +SELECT /*+ REPARTITION(1) */ make_date(6000000, 2, 30) query expect_error(Year) -SELECT make_date(1000000000, 13, 32) +SELECT /*+ REPARTITION(1) */ make_date(1000000000, 13, 32) diff --git a/spark/src/test/scala/org/apache/comet/CometExpressionSuite.scala b/spark/src/test/scala/org/apache/comet/CometExpressionSuite.scala index 3a4f8140506..57ba69765d0 100644 --- a/spark/src/test/scala/org/apache/comet/CometExpressionSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometExpressionSuite.scala @@ -2479,7 +2479,8 @@ class CometExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelper { // the other. The fallback reason also proves the optimizer retained the empty predicate. Seq("In" -> col("d").isin(), "InSet" -> getColumnFromExpression(emptySet)).foreach { case (parent, predicate) => - val df = input.select(predicate) + // Materialize the result to isolate the parent guard from root row consumption. + val df = input.select(predicate).repartition(1) val _ = checkSparkAnswerAndFallbackReason( df, s"make_date requires Spark evaluation under $parent") diff --git a/spark/src/test/scala/org/apache/comet/CometTemporalExpressionSuite.scala b/spark/src/test/scala/org/apache/comet/CometTemporalExpressionSuite.scala index 63b0e06724f..5f13ebf1a19 100644 --- a/spark/src/test/scala/org/apache/comet/CometTemporalExpressionSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometTemporalExpressionSuite.scala @@ -22,6 +22,7 @@ package org.apache.comet import scala.util.Random import org.apache.spark.SparkThrowable +import org.apache.spark.api.java.JavaRDD import org.apache.spark.sql.{CometTestBase, DataFrame, Row, SaveMode} import org.apache.spark.sql.catalyst.analysis.UnresolvedAttribute import org.apache.spark.sql.catalyst.expressions.{Days, Hours, Literal} @@ -59,6 +60,50 @@ class CometTemporalExpressionSuite extends CometTestBase with AdaptiveSparkPlanH .getOrElse( fail(s"No SparkThrowable in cause chain: ${causeChain(error).map(_.getClass.getName)}")) + test("make_date preserves partial consumption through the PySpark RDD bridge") { + withSQLConf( + SQLConf.ANSI_ENABLED.key -> "false", + SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false", + CometConf.COMET_BATCH_SIZE.key -> "8192") { + withTempPath { path => + withSQLConf(CometConf.COMET_ENABLED.key -> "false") { + spark + .range(0L, 4096L, 1L, 1) + .selectExpr("CAST(IF(id = 4095, 6000000, 2024) AS INT) AS y", "id") + .write + .parquet(path.getCanonicalPath) + } + def frame: DataFrame = spark.read + .parquet(path.getCanonicalPath) + .selectExpr( + "unix_date(make_date(y, 1, 1)) AS d", + "repeat(sha2(CAST(id AS STRING), 512), 512) AS padding") + + def firstPythonBatch(df: DataFrame): Seq[Byte] = { + // PySpark calls this JVM method, which uses QueryExecution.toRdd rather than + // Dataset.rdd's object-producing plan. Its Scala visibility is private[sql]. + df.getClass + .getMethod("javaToPython") + .invoke(df) + .asInstanceOf[JavaRDD[Array[Byte]]] + .take(1) + .get(0) + .toSeq + } + + val expected = withSQLConf(CometConf.COMET_ENABLED.key -> "false") { + firstPythonBatch(frame) + } + assert(expected.nonEmpty) + for (df <- Seq(frame, frame.limit(4096))) { + // Preparing a plan (for example with explain()) must not change a later RDD action. + val _ = df.queryExecution.executedPlan + assert(firstPythonBatch(df) == expected) + } + } + } + } + test("make_date does not evaluate errors beyond a partial consumer") { import testImplicits._ @@ -178,8 +223,9 @@ class CometTemporalExpressionSuite extends CometTestBase with AdaptiveSparkPlanH joinedYearPlan.collect { case join: BroadcastHashJoinExec => join }.size == 1, joinedYearPlan.treeString) + // An exchange materializes the date before any external row consumer can stop. // A full collection must still evaluate the second row in the native expression. - val full = sql(unlimited) + val full = sql(unlimited).repartition(1) checkCometOperators(stripAQEPlan(full.queryExecution.executedPlan)) val (sparkError, cometError) = checkSparkAnswerMaybeThrows(full) Seq("Spark" -> sparkError, "Comet" -> cometError).foreach { case (engine, error) => @@ -207,7 +253,7 @@ class CometTemporalExpressionSuite extends CometTestBase with AdaptiveSparkPlanH "SELECT make_date(1000000000, 1, 1)", "SELECT make_date(1000000000, 13, 0)") .foreach { query => - val df = sql(query) + val df = sql(query).repartition(1) checkCometOperators(stripAQEPlan(df.queryExecution.executedPlan)) val (sparkError, cometError) = checkSparkAnswerMaybeThrows(df) 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..717b4930285 100644 --- a/spark/src/test/scala/org/apache/comet/rules/CometExecRuleSuite.scala +++ b/spark/src/test/scala/org/apache/comet/rules/CometExecRuleSuite.scala @@ -183,12 +183,27 @@ class CometExecRuleSuite extends CometTestBase { ansi), "date")()), input) - val native = applyCometExecRule(project) - assert(countOperators(native, classOf[CometProjectExec]) == 1, native.treeString) + val root = applyCometExecRule(project) + val requiresRows = ansi || !boundedYear + assert( + countOperators(root, classOf[CometProjectExec]) == (if (requiresRows) 0 else 1), + root.treeString) + assert( + countOperators(root, classOf[ProjectExec]) == (if (requiresRows) 1 else 0), + root.treeString) + def materialize(child: SparkPlan): SparkPlan = + SortExec(Seq(SortOrder(child.output.head, Ascending)), false, child) + def nativeProject: SparkPlan = applyCometExecRule(materialize(project)).collectFirst { + case native: CometProjectExec => native + }.get for ((boundary, consume) <- consumers; reused <- Seq(false, true)) { - val child = if (reused) applyCometExecRule(project) else project + // Reuse a native projection that was safe below a materializing consumer. + val child = if (reused) nativeProject else project val consumer = consume(child) - val converted = applyCometExecRule(consumer) + val root = + if (consumer.isInstanceOf[CollectLimitExec]) consumer + else materialize(consumer) + val converted = applyCometExecRule(root) for (result <- Seq(converted, applyCometExecRule(converted))) { // A reused local-table sink has no native child operator for a new cap. val capped = consumer.isInstanceOf[CollectLimitExec] && !reused @@ -240,11 +255,12 @@ class CometExecRuleSuite extends CometTestBase { checkSparkAnswerAndOperator(df, includeClasses = Seq(classOf[CometProjectExec])) } + // Materialize results to isolate expression masks from external row demand. val deferred = sql(""" |SELECT r, IF(r < 0, dt, DATE '2024-01-01'), r + 1 FROM ( | SELECT rand(7L) AS r, make_date(2024, _2, 1) AS dt FROM make_date_non_ansi |) - |""".stripMargin) + |""".stripMargin).repartition(1) checkSparkAnswerAndOperator(deferred, includeClasses = Seq(classOf[CometProjectExec])) assert( countOperators(deferred.queryExecution.executedPlan, classOf[CometProjectExec]) == 2) @@ -264,7 +280,7 @@ class CometExecRuleSuite extends CometTestBase { | SELECT rand(7L) AS r, make_date(2024, _2, 1) AS dt, _3 AS i | FROM make_date_non_ansi |) - |""".stripMargin) + |""".stripMargin).repartition(1) checkSparkAnswer(df) val plan = df.queryExecution.executedPlan withClue(s"consumer=$consumer\n${plan.treeString}") { @@ -430,7 +446,7 @@ class CometExecRuleSuite extends CometTestBase { Seq(right.output.head), Inner, None, - sort(applyCometExecRule(projection)), + applyCometExecRule(sort(projection)), sort(right)) val converted = applyCometExecRule(join) for (result <- Seq(converted, applyCometExecRule(converted))) { @@ -459,7 +475,9 @@ class CometExecRuleSuite extends CometTestBase { expected = sql(query).collect().toSeq } checkSparkRows(expected) - val df = sql(query) + // Materialize the complete result so this tests expression demand within the query, + // independently of an external RDD consumer stopping at the root. + val df = sql(query).repartition(1) checkCometAnswer(df, expected) stripAQEPlan(df.queryExecution.executedPlan) } @@ -657,7 +675,8 @@ class CometExecRuleSuite extends CometTestBase { expected = sql(query).collect().toSeq } assert(expected.map(_.get(dateIndex)).toSet == expectedDates) - val df = sql(query) + // Keep the root's external row demand separate from argument demand inside the query. + val df = sql(query).repartition(1) checkCometAnswer(df, expected) val plan = stripAQEPlan(df.queryExecution.executedPlan) assert(countOperators(plan, classOf[CometNativeScanExec]) == 1, plan.treeString) From 5dc6a0a4887d59de8ba3b6c3fd4ab6f1847efeaa Mon Sep 17 00:00:00 2001 From: Ping Zhang Date: Fri, 30 Oct 2026 22:11:37 +0000 Subject: [PATCH 4/6] fix: honor Parquet pushdown setting for scalar filters --- .../spark/sql/comet/CometNativeScanExec.scala | 8 ++- .../comet/exec/CometNativeReaderSuite.scala | 53 ++++++++++++++++++- 2 files changed, 58 insertions(+), 3 deletions(-) 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..9a015bc731e 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 @@ -30,6 +30,7 @@ import org.apache.spark.sql.comet.shims.ShimStreamSourceAwareSparkPlan import org.apache.spark.sql.execution.{ScalarSubquery => ExecScalarSubquery, _} import org.apache.spark.sql.execution.datasources._ import org.apache.spark.sql.execution.metric.SQLMetric +import org.apache.spark.sql.internal.SQLConf import org.apache.spark.sql.types._ import org.apache.spark.sql.vectorized.ColumnarBatch import org.apache.spark.util.SerializableConfiguration @@ -182,8 +183,11 @@ case class CometNativeScanExec( // (DataSourceScanExec.scala), which resolves ScalarSubquery -> Literal at execution time. val commonBytes = { val base = nativeOp.getNativeScan.getCommon - val scalarSubqueryFilters = dataFilters - .filter(_.exists(_.isInstanceOf[ExecScalarSubquery])) + val scalarSubqueryFilters = if (conf.getConf(SQLConf.PARQUET_FILTER_PUSHDOWN_ENABLED)) { + dataFilters.filter(_.exists(_.isInstanceOf[ExecScalarSubquery])) + } else { + Seq.empty + } scalarSubqueryFilters.foreach { f => f.foreach { case s: ExecScalarSubquery => 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 d3ba8fd6cfc..1a40aa04060 100644 --- a/spark/src/test/scala/org/apache/comet/exec/CometNativeReaderSuite.scala +++ b/spark/src/test/scala/org/apache/comet/exec/CometNativeReaderSuite.scala @@ -32,13 +32,14 @@ import org.apache.parquet.schema.MessageTypeParser import org.apache.spark.SparkThrowable import org.apache.spark.sql.{CometTestBase, DataFrame, Row} import org.apache.spark.sql.comet.CometNativeScanExec +import org.apache.spark.sql.execution.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 { @@ -1623,6 +1624,56 @@ class CometNativeReaderSuite extends CometTestBase with AdaptiveSparkPlanHelper } } + test("resolved scalar data 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 = find(df.queryExecution.executedPlan)(_.isInstanceOf[CometNativeScanExec]) + .getOrElse(fail("Expected a native Parquet scan")) + .asInstanceOf[CometNativeScanExec] + 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. + val (sparkError, cometError) = checkSparkAnswerMaybeThrows(df) + val failures = Seq(sparkError, cometError).map { error => + val failure = error.getOrElse(fail("Expected overflow before the keep predicate")) + 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("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 From fb4d82805f18a04adb76cf856cc3b73286494960 Mon Sep 17 00:00:00 2001 From: Ping Zhang Date: Fri, 30 Oct 2026 22:15:08 +0000 Subject: [PATCH 5/6] fix: preserve scan guards for unsupported date predicates --- .../contrib/delta/CometDeltaNativeScan.scala | 4 +++- docs/source/user-guide/latest/expressions.md | 6 ++++-- native/core/src/execution/delta_dv.rs | 1 + .../operators/dynamic_filter/join/tests.rs | 2 ++ .../dynamic_filter/join/tests/schema_errors.rs | 1 + .../tests/schema_errors/partition_columns.rs | 1 + .../join/tests/timestamp_errors.rs | 1 + .../operators/dynamic_filter/topk/tests.rs | 1 + .../dynamic_filter/topk/tests/timestamp.rs | 1 + native/core/src/execution/planner.rs | 1 + native/core/src/parquet/parquet_exec.rs | 18 +++++++++++++++--- .../src/parquet/parquet_exec/variant_tests.rs | 3 +++ native/proto/src/proto/operator.proto | 3 +++ .../comet/serde/operator/CometNativeScan.scala | 6 +++++- .../comet/exec/CometNativeReaderSuite.scala | 10 ++++++++-- 15 files changed, 50 insertions(+), 9 deletions(-) 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..14c87cded61 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 @@ -23,7 +23,7 @@ import scala.jdk.CollectionConverters._ import org.apache.hadoop.fs.Path import org.apache.spark.internal.Logging -import org.apache.spark.sql.catalyst.expressions.Literal +import org.apache.spark.sql.catalyst.expressions.{Literal, MakeDate} import org.apache.spark.sql.comet.{CometScanExec, DeltaPlanDataInjector} import org.apache.spark.sql.delta.DeltaParquetFileFormat import org.apache.spark.sql.delta.RowIndexFilterType @@ -166,6 +166,8 @@ object CometDeltaNativeScan } commonOpt.map { commonBuilder => + commonBuilder.setDisableRowFilterPushdown( + scanExec.dataFilters.exists(_.exists(_.isInstanceOf[MakeDate]))) // Already forced by declineReason on this claim; reused rather than deserialized again. val dvDescriptors = memo.dvDescriptors // Union object-store options over every authority a partition of this scan may need a diff --git a/docs/source/user-guide/latest/expressions.md b/docs/source/user-guide/latest/expressions.md index 8571b6ec7dd..1aa05bdcf8d 100644 --- a/docs/source/user-guide/latest/expressions.md +++ b/docs/source/user-guide/latest/expressions.md @@ -772,8 +772,10 @@ date evaluation from an outer limit. Predicates containing `make_date` stay above the Parquet scan. Comet disables row-level filter pushdown for that scan so predicates cannot be reordered or hide a required date error, and -Spark's structured error is preserved. Other predicates can still use metadata to skip row groups -and pages. +Spark's structured error is preserved. This safeguard uses Spark's original predicates, including +those that cannot be serialized for native execution. Other predicates can still use metadata to +skip row groups and pages when `spark.sql.parquet.filterPushdown` is enabled. That setting also +applies to filters resolved from scalar subqueries at execution time. 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 diff --git a/native/core/src/execution/delta_dv.rs b/native/core/src/execution/delta_dv.rs index a99409d2ed1..f005f67252d 100644 --- a/native/core/src/execution/delta_dv.rs +++ b/native/core/src/execution/delta_dv.rs @@ -1592,6 +1592,7 @@ mod tests { false, "", "", + false, ) .unwrap(); let batches: Vec<_> = scan diff --git a/native/core/src/execution/operators/dynamic_filter/join/tests.rs b/native/core/src/execution/operators/dynamic_filter/join/tests.rs index 54ae6737edf..7e545b919af 100644 --- a/native/core/src/execution/operators/dynamic_filter/join/tests.rs +++ b/native/core/src/execution/operators/dynamic_filter/join/tests.rs @@ -673,6 +673,7 @@ fn parquet_probe( false, "CORRECTED", "CORRECTED", + false, ) .unwrap(); (file, scan) @@ -827,6 +828,7 @@ async fn reader_filter_crosses_null_check_conjunction_and_retains_residual() { false, "CORRECTED", "CORRECTED", + false, ) .unwrap(); let checks = [("key", 0), ("payload", 1), ("other", 2)].map(|(name, index)| { diff --git a/native/core/src/execution/operators/dynamic_filter/join/tests/schema_errors.rs b/native/core/src/execution/operators/dynamic_filter/join/tests/schema_errors.rs index 8734bca6e29..f43f3aeeb42 100644 --- a/native/core/src/execution/operators/dynamic_filter/join/tests/schema_errors.rs +++ b/native/core/src/execution/operators/dynamic_filter/join/tests/schema_errors.rs @@ -89,6 +89,7 @@ fn scan( false, "CORRECTED", "CORRECTED", + false, ) .unwrap() } diff --git a/native/core/src/execution/operators/dynamic_filter/join/tests/schema_errors/partition_columns.rs b/native/core/src/execution/operators/dynamic_filter/join/tests/schema_errors/partition_columns.rs index f6b00114f10..63df42cc026 100644 --- a/native/core/src/execution/operators/dynamic_filter/join/tests/schema_errors/partition_columns.rs +++ b/native/core/src/execution/operators/dynamic_filter/join/tests/schema_errors/partition_columns.rs @@ -74,6 +74,7 @@ fn partitioned_scan( false, "CORRECTED", "CORRECTED", + false, ) .unwrap() } diff --git a/native/core/src/execution/operators/dynamic_filter/join/tests/timestamp_errors.rs b/native/core/src/execution/operators/dynamic_filter/join/tests/timestamp_errors.rs index aa982280517..757d1a0bc7c 100644 --- a/native/core/src/execution/operators/dynamic_filter/join/tests/timestamp_errors.rs +++ b/native/core/src/execution/operators/dynamic_filter/join/tests/timestamp_errors.rs @@ -108,6 +108,7 @@ async fn assert_timestamp_overflow_preserved(nested: bool) { false, "CORRECTED", "CORRECTED", + false, ) .unwrap(); let join = single_key_join_plans( diff --git a/native/core/src/execution/operators/dynamic_filter/topk/tests.rs b/native/core/src/execution/operators/dynamic_filter/topk/tests.rs index 58a9866bca0..394cddb6885 100644 --- a/native/core/src/execution/operators/dynamic_filter/topk/tests.rs +++ b/native/core/src/execution/operators/dynamic_filter/topk/tests.rs @@ -199,6 +199,7 @@ fn parquet_scan( false, "CORRECTED", "CORRECTED", + false, ) .unwrap() } diff --git a/native/core/src/execution/operators/dynamic_filter/topk/tests/timestamp.rs b/native/core/src/execution/operators/dynamic_filter/topk/tests/timestamp.rs index ed5b7267e0a..15a13d37fb3 100644 --- a/native/core/src/execution/operators/dynamic_filter/topk/tests/timestamp.rs +++ b/native/core/src/execution/operators/dynamic_filter/topk/tests/timestamp.rs @@ -98,6 +98,7 @@ fn timestamp_input( false, "CORRECTED", "CORRECTED", + false, ) .unwrap(); (file, scan) diff --git a/native/core/src/execution/planner.rs b/native/core/src/execution/planner.rs index 10cef2334f1..9750b4b21e6 100644 --- a/native/core/src/execution/planner.rs +++ b/native/core/src/execution/planner.rs @@ -494,6 +494,7 @@ impl PhysicalPlanner { rebase_from_file_metadata, datetime_rebase_mode_in_read, int96_rebase_mode_in_read, + common.disable_row_filter_pushdown, )?; Ok(Arc::new(SparkPlan::new(plan_id, scan, vec![]))) } diff --git a/native/core/src/parquet/parquet_exec.rs b/native/core/src/parquet/parquet_exec.rs index 271b3bc5e62..9bbf61827c9 100644 --- a/native/core/src/parquet/parquet_exec.rs +++ b/native/core/src/parquet/parquet_exec.rs @@ -83,6 +83,9 @@ pub(crate) struct CometParquetScanOptions { /// `data_filters`: Any predicate that must be applied to the data returned by the scan. An empty /// `Vec` means Spark supplied filters that Comet could not serialize. If specified, then /// `data_schema` must also be specified. +/// +/// `disable_row_filter_pushdown`: Keep row filtering above the scan, including when the +/// throwing Spark predicate is absent from `data_filters` because it could not be serialized. #[allow(clippy::too_many_arguments)] pub(crate) fn init_datasource_exec( required_schema: SchemaRef, @@ -106,15 +109,17 @@ pub(crate) fn init_datasource_exec( rebase_from_file_metadata: bool, datetime_rebase_mode_in_read: &str, int96_rebase_mode_in_read: &str, + disable_row_filter_pushdown: bool, ) -> Result, ExecutionError> { // Computed once and reused below for `try_pushdown_filters`. `copied_config()` clones only // `SessionConfig` (an `Arc` plus a small extensions map); `SessionContext:: // state()` would clone the whole `SessionState`, including the function registries (~250+ // UDFs) and optimizer rules, on every scan init. let mut session_config = session_ctx.copied_config(); - if data_filters - .as_ref() - .is_some_and(|filters| filters.iter().any(contains_make_date)) + if disable_row_filter_pushdown + || data_filters + .as_ref() + .is_some_and(|filters| filters.iter().any(contains_make_date)) { // RowFilter can reorder predicates and flatten their Spark errors. Even pushing only // a later sibling could hide a make_date error that Spark evaluates first. Keep all @@ -597,6 +602,7 @@ mod tests { false, "", "", + false, ) .unwrap() } @@ -734,6 +740,7 @@ mod tests { rebase_from_file_metadata, "", "", + false, ) .unwrap(); @@ -810,6 +817,7 @@ mod tests { true, datetime_rebase_mode, datetime_rebase_mode, + false, ) .unwrap(); @@ -919,6 +927,7 @@ mod tests { true, "CORRECTED", "CORRECTED", + false, ) .unwrap(); @@ -1015,6 +1024,7 @@ mod tests { true, datetime_rebase_mode, int96_rebase_mode, + false, ) .unwrap(); @@ -1162,6 +1172,7 @@ mod tests { true, datetime_rebase_mode, int96_rebase_mode, + false, ) .unwrap(); @@ -1431,6 +1442,7 @@ mod tests { false, "", "", + false, ) .unwrap(); diff --git a/native/core/src/parquet/parquet_exec/variant_tests.rs b/native/core/src/parquet/parquet_exec/variant_tests.rs index 0bfdcecd222..4d8e799a659 100644 --- a/native/core/src/parquet/parquet_exec/variant_tests.rs +++ b/native/core/src/parquet/parquet_exec/variant_tests.rs @@ -166,6 +166,7 @@ async fn scan_variant_file(filename: PathBuf) -> VariantArray { false, "", "", + false, ) .unwrap(); let mut stream = scan.execute(0, session_ctx.task_ctx()).unwrap(); @@ -234,6 +235,7 @@ async fn unread_variant_does_not_override_arrow_schema_hint() { false, "", "", + false, ) .unwrap(); let mut stream = scan.execute(0, session.task_ctx()).unwrap(); @@ -267,6 +269,7 @@ fn encrypted_projected_variant_is_rejected_before_reader_creation() { false, "", "", + false, ); assert!(result .unwrap_err() diff --git a/native/proto/src/proto/operator.proto b/native/proto/src/proto/operator.proto index 28d94e78bf3..aa4cee0a330 100644 --- a/native/proto/src/proto/operator.proto +++ b/native/proto/src/proto/operator.proto @@ -185,6 +185,9 @@ message NativeScanCommon { // True when Spark supplied a Parquet data filter, including when Comet could // not serialize any of those filters into data_filters. bool has_data_filters = 19; + // Keep row filtering above this scan when an original Spark data filter can + // throw, even if that filter could not be serialized into data_filters. + bool disable_row_filter_pushdown = 20; } message NativeScan { 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..2127e0b3719 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} @@ -308,6 +308,10 @@ object CometNativeScan extends CometOperatorSerde[CometScanExec] with CometTypeS hadoopConf = hadoopConf, conf = scan.conf) match { case Some(commonBuilder) => + // Original filters include unsupported expressions and unresolved subqueries omitted + // from native predicates. A pushed sibling must not hide their make_date errors. + commonBuilder.setDisableRowFilterPushdown( + scan.dataFilters.exists(_.exists(_.isInstanceOf[MakeDate]))) // Sink operators don't have children builder.clearChildren() val nativeScanBuilder = OperatorOuterClass.NativeScan.newBuilder() 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 1a40aa04060..78d5b66ae76 100644 --- a/spark/src/test/scala/org/apache/comet/exec/CometNativeReaderSuite.scala +++ b/spark/src/test/scala/org/apache/comet/exec/CometNativeReaderSuite.scala @@ -1582,7 +1582,8 @@ class CometNativeReaderSuite extends CometTestBase with AdaptiveSparkPlanHelper .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, '', sha2(CAST(id AS STRING), 512)) AS mask", + "IF(id = 1023, 0, 1) AS keep") .write .parquet(path.toString) } @@ -1599,7 +1600,12 @@ class CometNativeReaderSuite extends CometTestBase with AdaptiveSparkPlanHelper // 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").foreach { predicate => + 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") assert( find(df.queryExecution.executedPlan)(_.isInstanceOf[CometNativeScanExec]).isDefined) From c5c6793d7bb6b54a2ee9f7821c00af672a8d58d6 Mon Sep 17 00:00:00 2001 From: Ping Zhang Date: Sat, 31 Oct 2026 00:09:24 +0000 Subject: [PATCH 6/6] test: keep Python bridge regression compatible with Spark 3 --- .../org/apache/comet/CometTemporalExpressionSuite.scala | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/spark/src/test/scala/org/apache/comet/CometTemporalExpressionSuite.scala b/spark/src/test/scala/org/apache/comet/CometTemporalExpressionSuite.scala index 5f13ebf1a19..d5333a5c34f 100644 --- a/spark/src/test/scala/org/apache/comet/CometTemporalExpressionSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometTemporalExpressionSuite.scala @@ -91,8 +91,9 @@ class CometTemporalExpressionSuite extends CometTestBase with AdaptiveSparkPlanH .toSeq } - val expected = withSQLConf(CometConf.COMET_ENABLED.key -> "false") { - firstPythonBatch(frame) + var expected = Seq.empty[Byte] + withSQLConf(CometConf.COMET_ENABLED.key -> "false") { + expected = firstPythonBatch(frame) } assert(expected.nonEmpty) for (df <- Seq(frame, frame.limit(4096))) {