From 27044df9193f5ec3a7cfdaca15f71dc6803574c5 Mon Sep 17 00:00:00 2001 From: Chao Sun Date: Sat, 15 Aug 2026 12:24:17 -0700 Subject: [PATCH 1/6] fix: match Spark whitespace trimming in to_time and try_to_time (#5364) * fix: match Spark whitespace trimming in to_time and try_to_time * test: address to_time trim review feedback (cherry picked from commit a74839cf76ff15a7de4128b4740e8133ae8d0d29) --- .../user-guide/latest/compatibility/index.md | 8 +- native/spark-expr/src/conversion_funcs/mod.rs | 2 +- .../spark-expr/src/conversion_funcs/trim.rs | 6 +- .../spark-expr/src/datetime_funcs/to_time.rs | 245 ++++++++++++++++-- .../expressions/datetime/to_time.sql | 77 ++++++ 5 files changed, 306 insertions(+), 32 deletions(-) diff --git a/docs/source/user-guide/latest/compatibility/index.md b/docs/source/user-guide/latest/compatibility/index.md index ae29a67bc19..001ec876061 100644 --- a/docs/source/user-guide/latest/compatibility/index.md +++ b/docs/source/user-guide/latest/compatibility/index.md @@ -134,10 +134,10 @@ so users hunting an unexpected value have a single place to check: `DECIMAL(1, 1)`) throws `NUMERIC_VALUE_OUT_OF_RANGE` regardless of the eval mode. Spark returns `NULL` under legacy and try mode, and only throws under ANSI ([#5068](https://github.com/apache/datafusion-comet/issues/5068)). -- `CAST(string AS timestamp)`, `CAST(string AS timestamp_ntz)`, `to_time` and `try_to_time` trim - Unicode whitespace. Spark trims only the bytes `0x00`-`0x20` and `0x7F`, so a value padded with - an ASCII control byte parses in Spark and returns `NULL` in Comet, while a value padded with - non-ASCII whitespace such as `U+3000` returns `NULL` in Spark and parses in Comet +- `CAST(string AS timestamp)` and `CAST(string AS timestamp_ntz)` trim Unicode whitespace. + Spark trims only the bytes `0x00`-`0x20` and `0x7F`, so a value padded with an ASCII control byte + parses in Spark and returns `NULL` in Comet, while a value padded with non-ASCII whitespace such + as `U+3000` returns `NULL` in Spark and parses in Comet ([#5149](https://github.com/apache/datafusion-comet/issues/5149)). - Native `RANGE` window frames with an explicit `PRECEDING` / `FOLLOWING` offset diverge from Spark when the boundary arithmetic overflows for `DATE` or `DECIMAL` `ORDER BY` columns diff --git a/native/spark-expr/src/conversion_funcs/mod.rs b/native/spark-expr/src/conversion_funcs/mod.rs index 2f42c316b3f..7b864e193b8 100644 --- a/native/spark-expr/src/conversion_funcs/mod.rs +++ b/native/spark-expr/src/conversion_funcs/mod.rs @@ -20,5 +20,5 @@ pub mod cast; mod numeric; mod string; mod temporal; -mod trim; +pub(crate) mod trim; mod utils; diff --git a/native/spark-expr/src/conversion_funcs/trim.rs b/native/spark-expr/src/conversion_funcs/trim.rs index 983bf34fa4a..8cd71590962 100644 --- a/native/spark-expr/src/conversion_funcs/trim.rs +++ b/native/spark-expr/src/conversion_funcs/trim.rs @@ -31,9 +31,11 @@ //! `Double.parseDouble` trims before parsing, and `Decimal.stringToJavaBigDecimal` does //! `str.toString.trim`. //! +//! `to_time` and `try_to_time` also use [`trim_all`] after detecting an optional AM/PM suffix; +//! suffix detection itself first removes ASCII spaces only, matching Spark's `stringToTime`. +//! //! \* `timestamp` and `timestamp_ntz` are listed for what Spark does; the Comet parsers for those -//! two targets have not been migrated to these helpers and still use `str::trim`, as do `to_time` -//! and `try_to_time` in `datetime_funcs::to_time` +//! two targets have not been migrated to these helpers and still use `str::trim` //! (). /// True for the bytes trimmed by `org.apache.spark.unsafe.types.UTF8String.trimAll`, i.e. the diff --git a/native/spark-expr/src/datetime_funcs/to_time.rs b/native/spark-expr/src/datetime_funcs/to_time.rs index 727998fdf85..249be04a4f7 100644 --- a/native/spark-expr/src/datetime_funcs/to_time.rs +++ b/native/spark-expr/src/datetime_funcs/to_time.rs @@ -15,6 +15,7 @@ // specific language governing permissions and limitations // under the License. +use crate::conversion_funcs::trim::trim_all; use arrow::array::{Array, StringArray, Time64NanosecondArray}; use datafusion::common::{DataFusionError, Result}; use datafusion::physical_plan::ColumnarValue; @@ -80,26 +81,18 @@ pub fn spark_to_time(args: &[ColumnarValue], fail_on_error: bool) -> Result Option { - let trimmed = s.trim(); - if trimmed.is_empty() { - return None; - } - - // Spark's parseTimestampString gates the T-prefix branch on j == 0 (start of - // the trimmed string), so " T12:30" is rejected even though leading whitespace - // is trimmed: the original segment start differs from the trimmed position. - if trimmed.as_bytes()[0] == b'T' && s.as_bytes()[0].is_ascii_whitespace() { - return None; - } - - let bytes = trimmed.as_bytes(); - let num_chars = bytes.len(); - - // Detect AM/PM suffix - let (is_am, is_pm, has_suffix) = if num_chars > 2 { - let last = bytes[num_chars - 1]; + // Spark's stringToTime calls UTF8String.trimRight before looking for AM/PM. + // Unlike trimAll, trimRight removes ASCII spaces only, so a control byte + // after AM/PM prevents the suffix from being recognized. + let right_trimmed = s.trim_end_matches(' '); + let bytes = right_trimmed.as_bytes(); + let num_bytes = bytes.len(); + + // ASCII AM/PM suffix bytes cannot be UTF-8 continuation bytes, so byte indexing is safe. + let (is_am, is_pm, has_suffix) = if num_bytes > 2 { + let last = bytes[num_bytes - 1]; if last == b'M' || last == b'm' { - let second_last = bytes[num_chars - 2]; + let second_last = bytes[num_bytes - 2]; let am = second_last == b'A' || second_last == b'a'; let pm = second_last == b'P' || second_last == b'p'; (am, pm, am || pm) @@ -110,14 +103,24 @@ fn string_to_time(s: &str) -> Option { (false, false, false) }; - // Strip AM/PM suffix (and optional space before it) - let time_str = if has_suffix { - let end = num_chars - 2; - let s = &trimmed[..end]; - s.trim_end() + // Spark passes the remaining segment to parseTimestampString, which trims + // all ASCII control bytes and spaces, including DELETE, from both ends. + // Unicode whitespace is intentionally preserved and fails to parse. + let untrimmed_time = if has_suffix { + &right_trimmed[..num_bytes - 2] } else { - trimmed + right_trimmed }; + let time_str = trim_all(untrimmed_time); + if time_str.is_empty() { + return None; + } + + // parseTimestampString accepts a T-prefix only at the original segment + // start, before trimAll advances past any leading ASCII control bytes. + if time_str.starts_with('T') && !untrimmed_time.starts_with('T') { + return None; + } // Parse the time components let (hour, minute, second, micros) = parse_time_components(time_str)?; @@ -479,4 +482,196 @@ mod tests { assert_eq!(string_to_time(" T12:30:45"), None); assert_eq!(string_to_time(" T12:30"), None); } + + #[test] + fn test_spark_trim_all_control_bytes() { + for (time, after_hour) in [("12:30:45", "30:45"), ("12:30", "30")] { + let expected = string_to_time(time); + + for byte in (0_u8..=0x20).chain(std::iter::once(0x7f)) { + let padding = char::from(byte); + for input in [ + format!("{padding}{time}"), + format!("{time}{padding}"), + format!("{padding}{time}{padding}"), + ] { + assert_eq!( + string_to_time(&input), + expected, + "padding byte 0x{byte:02X} in {input:?}" + ); + } + + let interior = format!("12:{padding}{after_hour}"); + assert_eq!( + string_to_time(&interior), + None, + "interior padding byte 0x{byte:02X} in {interior:?}" + ); + } + } + } + + #[test] + fn test_unicode_whitespace_is_not_trimmed() { + for padding in [ + '\u{0085}', '\u{00a0}', '\u{1680}', '\u{2000}', '\u{2003}', '\u{2007}', '\u{2028}', + '\u{2029}', '\u{202f}', '\u{205f}', '\u{3000}', + ] { + assert!(padding.is_whitespace()); + + for input in [ + format!("{padding}12:30:45"), + format!("12:30:45{padding}"), + format!("{padding}12:30:45{padding}"), + format!("12:{padding}30:45"), + format!("{padding}1:00:00 AM"), + format!("1:00:00{padding}AM"), + format!("1:00:00 AM{padding}"), + ] { + assert_eq!( + string_to_time(&input), + None, + "Unicode whitespace U+{:04X} in {input:?}", + padding as u32 + ); + } + } + } + + #[test] + fn test_am_pm_control_byte_trimming() { + for (suffix, expected) in [ + ("AM", NANOS_PER_HOUR), + ("PM", 13 * NANOS_PER_HOUR), + ("am", NANOS_PER_HOUR), + ("pm", 13 * NANOS_PER_HOUR), + ] { + for byte in (0_u8..=0x20).chain(std::iter::once(0x7f)) { + let padding = char::from(byte); + + for input in [ + format!("{padding}1:00:00 {suffix}"), + format!("1:00:00{padding}{suffix}"), + format!("{padding}1:00:00{padding}{suffix}"), + ] { + assert_eq!( + string_to_time(&input), + Some(expected), + "padding byte 0x{byte:02X} in {input:?}" + ); + } + + let trailing = format!("1:00:00 {suffix}{padding}"); + let expected_trailing = (byte == b' ').then_some(expected); + assert_eq!( + string_to_time(&trailing), + expected_trailing, + "trailing padding byte 0x{byte:02X} in {trailing:?}" + ); + } + } + + assert_eq!(string_to_time("1:00:00 AM \t"), None); + assert_eq!(string_to_time("1:00:00 AM\t "), None); + assert_eq!(string_to_time("1:00:00 AM "), Some(NANOS_PER_HOUR)); + } + + #[test] + fn test_t_prefix_rejects_all_leading_trimmed_bytes() { + let expected = string_to_time("T12:30:45"); + + for byte in (0_u8..=0x20).chain(std::iter::once(0x7f)) { + let padding = char::from(byte); + let leading = format!("{padding}T12:30:45"); + let trailing = format!("T12:30:45{padding}"); + + assert_eq!( + string_to_time(&leading), + None, + "leading padding byte 0x{byte:02X} in {leading:?}" + ); + assert_eq!( + string_to_time(&trailing), + expected, + "trailing padding byte 0x{byte:02X} in {trailing:?}" + ); + } + } + + #[test] + fn test_t_prefix_am_pm_control_byte_trimming() { + for (time, seconds) in [("T1:30", 0), ("T1:30:45", 45)] { + for (suffix, hour) in [("AM", 1), ("PM", 13), ("am", 1), ("pm", 13)] { + let expected = + hour * NANOS_PER_HOUR + 30 * NANOS_PER_MINUTE + seconds * NANOS_PER_SECOND; + + assert_eq!(string_to_time(&format!("{time} {suffix}")), Some(expected)); + + for byte in (0_u8..=0x20).chain(std::iter::once(0x7f)) { + let padding = char::from(byte); + + for input in [ + format!("{padding}{time} {suffix}"), + format!("{padding}{time}{padding}{suffix}"), + ] { + assert_eq!( + string_to_time(&input), + None, + "leading padding byte 0x{byte:02X} in {input:?}" + ); + } + + let before_suffix = format!("{time}{padding}{suffix}"); + assert_eq!( + string_to_time(&before_suffix), + Some(expected), + "pre-suffix padding byte 0x{byte:02X} in {before_suffix:?}" + ); + + let after_suffix = format!("{time} {suffix}{padding}"); + assert_eq!( + string_to_time(&after_suffix), + (byte == b' ').then_some(expected), + "post-suffix padding byte 0x{byte:02X} in {after_suffix:?}" + ); + } + } + } + } + + #[test] + fn test_spark_to_time_whitespace_error_modes() { + let input = StringArray::from(vec![ + Some("\u{3000}12:30:45"), + Some("1:00:00 AM\t"), + Some("\u{1}12:30:45\u{7f}"), + Some("1:00:00\u{b}PM"), + None, + ]); + let args = [ColumnarValue::Array(Arc::new(input))]; + + assert!(matches!( + spark_to_time(&args, true), + Err(DataFusionError::Execution(message)) + if message.contains("cannot be parsed to a TIME value") + )); + + let ColumnarValue::Array(output) = spark_to_time(&args, false).unwrap() else { + panic!("spark_to_time should return an array"); + }; + let output = output + .as_any() + .downcast_ref::() + .unwrap(); + + assert!(output.is_null(0)); + assert!(output.is_null(1)); + assert_eq!( + output.value(2), + 12 * NANOS_PER_HOUR + 30 * NANOS_PER_MINUTE + 45 * NANOS_PER_SECOND + ); + assert_eq!(output.value(3), 13 * NANOS_PER_HOUR); + assert!(output.is_null(4)); + } } diff --git a/spark/src/test/resources/sql-tests/expressions/datetime/to_time.sql b/spark/src/test/resources/sql-tests/expressions/datetime/to_time.sql index b3ac439fd8e..5522033f5cb 100644 --- a/spark/src/test/resources/sql-tests/expressions/datetime/to_time.sql +++ b/spark/src/test/resources/sql-tests/expressions/datetime/to_time.sql @@ -242,6 +242,83 @@ SELECT try_to_time(' 12:30:45') query SELECT try_to_time(' 1:00:00 PM') +-- Spark's time parser trims ASCII control characters, spaces, and DELETE after detecting AM/PM, +-- but never trims non-ASCII whitespace. Materialize the padding so every query remains native. +statement +CREATE TABLE test_to_time_trim(name STRING, pad STRING) USING parquet + +statement +INSERT INTO test_to_time_trim VALUES + ('a_none', ''), + ('b_nul_0x00', chr(0)), + ('c_soh_0x01', chr(1)), + ('d_tab_0x09', chr(9)), + ('e_vtab_0x0b', chr(11)), + ('f_us_0x1f', chr(31)), + ('g_space_0x20', ' '), + ('h_del_0x7f', chr(127)), + ('i_nbsp_u00a0', cast(X'C2A0' as string)), + ('j_ideographic_u3000', cast(X'E38080' as string)) + +-- Leading, trailing, both-sided, and interior padding must match Spark for every codepoint. +query +SELECT + name, + try_to_time(concat(pad, '12:30:45')), + try_to_time(concat('12:30:45', pad)), + try_to_time(concat(pad, '12:30:45', pad)), + try_to_time(concat('12:', pad, '30:45')), + try_to_time(concat(pad, '12:30')), + try_to_time(concat('12:30', pad)), + try_to_time(concat(pad, '12:30', pad)), + try_to_time(concat('12:', pad, '30')) +FROM test_to_time_trim + +-- Leading trimAll padding invalidates a T prefix. Controls before AM/PM are valid, but only ASCII +-- spaces after AM/PM are trimmed before Spark checks the suffix, including with a T prefix. +query +SELECT + name, + try_to_time(concat(pad, 'T12:30:45')), + try_to_time(concat('T12:30:45', pad)), + try_to_time(concat('1:00:00', pad, 'PM')), + try_to_time(concat('1:00:00 PM', pad)), + try_to_time(concat(pad, 'T12:30:45 PM')), + try_to_time(concat('T12:30:45', pad, 'PM')), + try_to_time(concat('T12:30:45 PM', pad)), + try_to_time(concat(pad, 'T12:30 PM')), + try_to_time(concat('T12:30', pad, 'PM')), + try_to_time(concat('T12:30 PM', pad)) +FROM test_to_time_trim + +-- The throwing variant must accept the same valid ASCII controls as try_to_time. +query +SELECT + name, + to_time(concat(pad, '12:30:45', pad)), + to_time(concat('1:00:00', pad, 'PM')) +FROM test_to_time_trim +WHERE name IN ( + 'a_none', 'b_nul_0x00', 'c_soh_0x01', 'd_tab_0x09', 'e_vtab_0x0b', + 'f_us_0x1f', 'g_space_0x20', 'h_del_0x7f') + +-- The throwing variant must reject Unicode whitespace instead of silently parsing it. +query expect_error(cannot be parsed to a TIME value) +SELECT to_time(concat(pad, '12:30:45')) +FROM test_to_time_trim +WHERE name = 'i_nbsp_u00a0' + +query expect_error(cannot be parsed to a TIME value) +SELECT to_time(concat('12:30:45', pad)) +FROM test_to_time_trim +WHERE name = 'j_ideographic_u3000' + +-- Only literal ASCII spaces are removed before AM/PM suffix detection. +query expect_error(cannot be parsed to a TIME value) +SELECT to_time(concat('1:00:00 PM', pad)) +FROM test_to_time_trim +WHERE name = 'd_tab_0x09' + -- to_time with format pattern falls back to Spark (not supported natively) query expect_fallback(invoke is not supported) SELECT to_time('12:30:45', 'HH:mm:ss') From 1d785941b2a2a3f352be0a7152380f5bdd8fb2ab Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Mon, 17 Aug 2026 14:20:34 -0600 Subject: [PATCH 2/6] feat: remove native cast from boolean to decimal (#5185) * feat: remove native cast from boolean to decimal Boolean -> Decimal is an edge case that nobody uses in practice, and the native implementation has to reproduce Spark's precision/scale and overflow semantics for it. That is not worth the complexity, so mark the cast unsupported in `CometCast` and let the `CodegenDispatchFallback` mixin route it through Spark's own generated code inside the Comet pipeline. The projection still runs natively; only the cast itself is evaluated by Spark's codegen. Adds SQL file tests covering non-ANSI and ANSI behavior, including the value-dependent overflow edge cases. * docs: link issue #5186 from the ignored boolean-to-decimal cast test Explain in the test comment that the ignore is a limitation of CometCastSuite rather than of the cast itself, and point at the tracking issue for re-enabling it. (cherry picked from commit ce602ac85683ab093304af55570cfeb7768af588) --- .../spark-expr/benches/cast_from_boolean.rs | 18 +---- .../src/conversion_funcs/boolean.rs | 55 ++++--------- .../spark-expr/src/conversion_funcs/cast.rs | 5 +- .../apache/comet/expressions/CometCast.scala | 6 +- .../cast/cast_boolean_to_decimal.sql | 72 +++++++++++++++++ .../cast/cast_boolean_to_decimal_ansi.sql | 79 +++++++++++++++++++ .../org/apache/comet/CometCastSuite.scala | 21 ++++- 7 files changed, 192 insertions(+), 64 deletions(-) create mode 100644 spark/src/test/resources/sql-tests/expressions/cast/cast_boolean_to_decimal.sql create mode 100644 spark/src/test/resources/sql-tests/expressions/cast/cast_boolean_to_decimal_ansi.sql diff --git a/native/spark-expr/benches/cast_from_boolean.rs b/native/spark-expr/benches/cast_from_boolean.rs index 04bd72dc01c..caccd67e26d 100644 --- a/native/spark-expr/benches/cast_from_boolean.rs +++ b/native/spark-expr/benches/cast_from_boolean.rs @@ -69,20 +69,7 @@ fn criterion_benchmark(c: &mut Criterion) { None, None, ); - let cast_to_str = Cast::new( - expr.clone(), - DataType::Utf8, - spark_cast_options.clone(), - None, - None, - ); - let cast_to_decimal = Cast::new( - expr, - DataType::Decimal128(10, 4), - spark_cast_options, - None, - None, - ); + let cast_to_str = Cast::new(expr, DataType::Utf8, spark_cast_options, None, None); let mut group = c.benchmark_group("cast_bool".to_string()); group.bench_function("i8", |b| { @@ -106,9 +93,6 @@ fn criterion_benchmark(c: &mut Criterion) { group.bench_function("str", |b| { b.iter(|| cast_to_str.evaluate(&boolean_batch).unwrap()); }); - group.bench_function("decimal", |b| { - b.iter(|| cast_to_decimal.evaluate(&boolean_batch).unwrap()); - }); } fn create_boolean_batch() -> RecordBatch { diff --git a/native/spark-expr/src/conversion_funcs/boolean.rs b/native/spark-expr/src/conversion_funcs/boolean.rs index 3f4a86429cc..1db2746ce24 100644 --- a/native/spark-expr/src/conversion_funcs/boolean.rs +++ b/native/spark-expr/src/conversion_funcs/boolean.rs @@ -15,11 +15,13 @@ // specific language governing permissions and limitations // under the License. -use crate::{SparkError, SparkResult}; -use arrow::array::{Array, ArrayRef, AsArray, Decimal128Array, TimestampMicrosecondBuilder}; +use crate::SparkResult; +use arrow::array::{Array, ArrayRef, AsArray, TimestampMicrosecondBuilder}; use arrow::datatypes::DataType; use std::sync::Arc; +/// Boolean -> Decimal is intentionally absent: it has no native path and is routed through the +/// JVM codegen dispatcher on the Scala side (see `CometCast.canCastFromBoolean`). pub fn is_df_cast_from_bool_spark_compatible(to_type: &DataType) -> bool { use DataType::*; matches!( @@ -28,35 +30,6 @@ pub fn is_df_cast_from_bool_spark_compatible(to_type: &DataType) -> bool { ) } -pub fn cast_boolean_to_decimal( - array: &ArrayRef, - precision: u8, - scale: i8, -) -> SparkResult { - let bool_array = array.as_boolean(); - let scaled_val = 10_i128.pow(scale as u32); - let result: Decimal128Array = bool_array - .iter() - .map(|v| v.map(|b| if b { scaled_val } else { 0 })) - .collect(); - - // Convert Arrow decimal overflow errors to SparkError - let decimal_array = result - .with_precision_and_scale(precision, scale) - .map_err(|e| { - if matches!(e, arrow::error::ArrowError::InvalidArgumentError(_)) - && e.to_string().contains("too large to store in a Decimal128") - { - // Use the scaled value as it's the only non-zero value that could overflow - crate::error::decimal_overflow_error(scaled_val, precision, scale) - } else { - SparkError::Arrow(Arc::new(e)) - } - })?; - - Ok(Arc::new(decimal_array)) -} - pub(crate) fn cast_boolean_to_timestamp( array_ref: &ArrayRef, target_tz: &Option>, @@ -212,20 +185,20 @@ mod tests { } #[test] - fn test_bool_to_decimal_cast() { - let result = cast_array( + fn test_bool_to_decimal_cast_is_not_supported() { + // Boolean -> Decimal has no native path; the Scala planner routes it through the JVM + // codegen dispatcher, so reaching the native cast at all is a bug. + let err = cast_array( test_input_bool_array(), &Decimal128(10, 4), &test_input_spark_opts(), ) - .unwrap(); - let expected_arr = Decimal128Array::from(vec![10000_i128, 0_i128]) - .with_precision_and_scale(10, 4) - .unwrap(); - let arr = result.as_any().downcast_ref::().unwrap(); - assert_eq!(arr.value(0), expected_arr.value(0)); - assert_eq!(arr.value(1), expected_arr.value(1)); - assert!(arr.is_null(2)); + .expect_err("expected boolean -> decimal to be unsupported"); + assert!( + err.to_string() + .contains("Native cast invoked for unsupported cast"), + "unexpected error: {err}" + ); } #[test] diff --git a/native/spark-expr/src/conversion_funcs/cast.rs b/native/spark-expr/src/conversion_funcs/cast.rs index 37fddb8c115..504ea6695e1 100644 --- a/native/spark-expr/src/conversion_funcs/cast.rs +++ b/native/spark-expr/src/conversion_funcs/cast.rs @@ -16,7 +16,7 @@ // under the License. use crate::conversion_funcs::boolean::{ - cast_boolean_to_decimal, cast_boolean_to_timestamp, is_df_cast_from_bool_spark_compatible, + cast_boolean_to_timestamp, is_df_cast_from_bool_spark_compatible, }; use crate::conversion_funcs::numeric::{ cast_decimal128_to_utf8, cast_decimal_to_timestamp, cast_float32_to_decimal128, @@ -438,9 +438,6 @@ pub(crate) fn cast_array( (Int64, Binary) if (eval_mode == Legacy) => { cast_whole_num_to_binary!(&array, Int64Array, 8) } - (Boolean, Decimal128(precision, scale)) => { - cast_boolean_to_decimal(&array, *precision, *scale) - } (Int8 | Int16 | Int32 | Int64, Timestamp(_, tz)) => cast_int_to_timestamp(&array, tz), (Float32 | Float64, Timestamp(_, tz)) => cast_float_to_timestamp(&array, tz, eval_mode), (Boolean, Timestamp(_, tz)) => cast_boolean_to_timestamp(&array, tz), diff --git a/spark/src/main/scala/org/apache/comet/expressions/CometCast.scala b/spark/src/main/scala/org/apache/comet/expressions/CometCast.scala index 78abf54c8dc..1bd767e6f99 100644 --- a/spark/src/main/scala/org/apache/comet/expressions/CometCast.scala +++ b/spark/src/main/scala/org/apache/comet/expressions/CometCast.scala @@ -356,10 +356,14 @@ object CometCast private def canCastFromBoolean(toType: DataType, evalMode: CometEvalMode.Value): SupportLevel = toType match { case DataTypes.ByteType | DataTypes.ShortType | DataTypes.IntegerType | DataTypes.LongType | - DataTypes.FloatType | DataTypes.DoubleType | _: DecimalType => + DataTypes.FloatType | DataTypes.DoubleType => Compatible() case _: TimestampType if evalMode == CometEvalMode.LEGACY => Compatible() + // Boolean -> Decimal has no native path. It is a rare cast and getting the + // precision/scale/overflow behavior right in native code is not worth the complexity, so + // the `CodegenDispatchFallback` mixin routes it through Spark's own generated code inside + // the Comet pipeline instead. case _ => unsupported(DataTypes.BooleanType, toType) } diff --git a/spark/src/test/resources/sql-tests/expressions/cast/cast_boolean_to_decimal.sql b/spark/src/test/resources/sql-tests/expressions/cast/cast_boolean_to_decimal.sql new file mode 100644 index 00000000000..c995fb26d87 --- /dev/null +++ b/spark/src/test/resources/sql-tests/expressions/cast/cast_boolean_to_decimal.sql @@ -0,0 +1,72 @@ +-- Licensed to the Apache Software Foundation (ASF) under one +-- or more contributor license agreements. See the NOTICE file +-- distributed with this work for additional information +-- regarding copyright ownership. The ASF licenses this file +-- to you under the Apache License, Version 2.0 (the +-- "License"); you may not use this file except in compliance +-- with the License. You may obtain a copy of the License at +-- +-- http://www.apache.org/licenses/LICENSE-2.0 +-- +-- Unless required by applicable law or agreed to in writing, +-- software distributed under the License is distributed on an +-- "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +-- KIND, either express or implied. See the License for the +-- specific language governing permissions and limitations +-- under the License. + +-- Boolean -> Decimal has no native path in Comet. `CometCast` reports it as unsupported so the +-- `CodegenDispatchFallback` mixin routes it through Spark's own generated code inside the Comet +-- pipeline. The default `query` mode therefore still asserts fully native execution. + +statement +CREATE TABLE test_cast_bool_to_decimal(id int, b boolean) USING parquet + +statement +INSERT INTO test_cast_bool_to_decimal VALUES (1, true), (2, false), (3, NULL) + +-- basic precision/scale combinations, including the long fast path (precision <= 18) and the +-- BigDecimal path (precision > 18) +query +SELECT id, cast(b as decimal(10,2)), cast(b as decimal(14,4)), cast(b as decimal(30,0)) +FROM test_cast_bool_to_decimal ORDER BY id + +-- smallest precision/scale that can represent 1 +query +SELECT id, cast(b as decimal(1,0)), cast(b as decimal(2,1)) +FROM test_cast_bool_to_decimal ORDER BY id + +-- maximum precision, with and without scale +query +SELECT id, cast(b as decimal(38,0)), cast(b as decimal(38,37)) +FROM test_cast_bool_to_decimal ORDER BY id + +-- overflow: decimal(1,1) holds at most 0.9, so true does not fit and returns NULL in non-ANSI +-- mode while false and NULL are unaffected +query +SELECT id, cast(b as decimal(1,1)), cast(b as decimal(2,2)), cast(b as decimal(38,38)) +FROM test_cast_bool_to_decimal ORDER BY id + +-- literal arguments +query +SELECT cast(true as decimal(10,2)), cast(false as decimal(10,2)), + cast(cast(NULL as boolean) as decimal(10,2)) + +-- literal overflow +query +SELECT cast(true as decimal(1,1)), cast(false as decimal(1,1)) + +-- try_cast returns NULL on overflow rather than throwing +query +SELECT id, try_cast(b as decimal(10,2)), try_cast(b as decimal(1,1)) +FROM test_cast_bool_to_decimal ORDER BY id + +-- comparison predicates on the result of the cast +query +SELECT id FROM test_cast_bool_to_decimal +WHERE cast(b as decimal(10,2)) > 0.5 ORDER BY id + +-- the cast feeding an aggregate +query +SELECT sum(cast(b as decimal(10,2))), count(cast(b as decimal(1,1))) +FROM test_cast_bool_to_decimal diff --git a/spark/src/test/resources/sql-tests/expressions/cast/cast_boolean_to_decimal_ansi.sql b/spark/src/test/resources/sql-tests/expressions/cast/cast_boolean_to_decimal_ansi.sql new file mode 100644 index 00000000000..1025dbbee5f --- /dev/null +++ b/spark/src/test/resources/sql-tests/expressions/cast/cast_boolean_to_decimal_ansi.sql @@ -0,0 +1,79 @@ +-- Licensed to the Apache Software Foundation (ASF) under one +-- or more contributor license agreements. See the NOTICE file +-- distributed with this work for additional information +-- regarding copyright ownership. The ASF licenses this file +-- to you under the Apache License, Version 2.0 (the +-- "License"); you may not use this file except in compliance +-- with the License. You may obtain a copy of the License at +-- +-- http://www.apache.org/licenses/LICENSE-2.0 +-- +-- Unless required by applicable law or agreed to in writing, +-- software distributed under the License is distributed on an +-- "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +-- KIND, either express or implied. See the License for the +-- specific language governing permissions and limitations +-- under the License. + +-- ANSI edge cases for Boolean -> Decimal. Comet has no native path for this cast; the +-- `CodegenDispatchFallback` mixin runs Spark's own generated code inside the Comet pipeline, so +-- the ANSI overflow errors must match Spark exactly. The non-error queries act as sentinels +-- proving the cast really executed natively rather than falling the whole plan back to Spark. + +-- Config: spark.sql.ansi.enabled=true + +statement +CREATE TABLE test_cast_bool_to_decimal_ansi(id int, b boolean) USING parquet + +statement +INSERT INTO test_cast_bool_to_decimal_ansi VALUES (1, true), (2, false), (3, NULL) + +-- sentinel: a cast that always fits must run natively under ANSI mode +query +SELECT id, cast(b as decimal(10,2)), cast(b as decimal(38,0)) +FROM test_cast_bool_to_decimal_ansi ORDER BY id + +-- sentinel: decimal(1,0) is the tightest type that can hold 1 +query +SELECT id, cast(b as decimal(1,0)), cast(b as decimal(2,1)) +FROM test_cast_bool_to_decimal_ansi ORDER BY id + +-- decimal(1,1) holds at most 0.9, so casting true overflows and must throw under ANSI mode +query expect_error(NUMERIC_VALUE_OUT_OF_RANGE) +SELECT cast(b as decimal(1,1)) FROM test_cast_bool_to_decimal_ansi WHERE id = 1 + +-- decimal(2,2) holds at most 0.99: same overflow, larger precision +query expect_error(NUMERIC_VALUE_OUT_OF_RANGE) +SELECT cast(b as decimal(2,2)) FROM test_cast_bool_to_decimal_ansi WHERE id = 1 + +-- all-scale decimal at maximum precision still cannot represent 1 +query expect_error(NUMERIC_VALUE_OUT_OF_RANGE) +SELECT cast(b as decimal(38,38)) FROM test_cast_bool_to_decimal_ansi WHERE id = 1 + +-- literal true overflows the same way +query expect_error(NUMERIC_VALUE_OUT_OF_RANGE) +SELECT cast(true as decimal(1,1)) + +-- overflow is value dependent: false scales to 0, which fits every decimal type, so rows that +-- only contain false must not throw +query +SELECT cast(b as decimal(1,1)), cast(b as decimal(38,38)) +FROM test_cast_bool_to_decimal_ansi WHERE id = 2 + +-- NULL input short-circuits before the precision check, so it must not throw either +query +SELECT cast(b as decimal(1,1)), cast(b as decimal(38,38)) +FROM test_cast_bool_to_decimal_ansi WHERE id = 3 + +-- try_cast suppresses the ANSI overflow and yields NULL +query +SELECT id, try_cast(b as decimal(1,1)), try_cast(b as decimal(38,38)) +FROM test_cast_bool_to_decimal_ansi ORDER BY id + +-- overflow raised from inside an aggregate input +query expect_error(NUMERIC_VALUE_OUT_OF_RANGE) +SELECT sum(cast(b as decimal(1,1))) FROM test_cast_bool_to_decimal_ansi + +-- overflow raised from inside a filter predicate +query expect_error(NUMERIC_VALUE_OUT_OF_RANGE) +SELECT id FROM test_cast_bool_to_decimal_ansi WHERE cast(b as decimal(1,1)) > 0.5 diff --git a/spark/src/test/scala/org/apache/comet/CometCastSuite.scala b/spark/src/test/scala/org/apache/comet/CometCastSuite.scala index 6e5b03ee356..32864292d17 100644 --- a/spark/src/test/scala/org/apache/comet/CometCastSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometCastSuite.scala @@ -177,10 +177,29 @@ class CometCastSuite extends CometTestBase with AdaptiveSparkPlanHelper { castTest(generateBools(), DataTypes.DoubleType) } - test("cast BooleanType to DecimalType(10,2)") { + // Boolean -> Decimal has no native path and is routed through the JVM codegen dispatcher, so + // `CometCast.isSupported` reports it as unsupported and the matrix-consistency test above + // requires this one to be ignored. The test would actually pass if un-ignored; the suite + // predates codegen dispatch and cannot yet express "runs in Comet without a native kernel". + // https://github.com/apache/datafusion-comet/issues/5186 tracks teaching the suite that + // distinction, at which point this test can be re-enabled. Execution coverage in the meantime, + // including the ANSI overflow edge cases, lives in + // `sql-tests/expressions/cast/cast_boolean_to_decimal{,_ansi}.sql`. + ignore("cast BooleanType to DecimalType(10,2)") { castTest(generateBools(), DataTypes.createDecimalType(10, 2)) } + test("cast BooleanType to DecimalType is not supported natively") { + Seq( + DataTypes.createDecimalType(10, 2), + DataTypes.createDecimalType(14, 4), + DataTypes.createDecimalType(30, 0)).foreach { toType => + assert( + CometCast.isSupported(BooleanType, toType, None, CometEvalMode.LEGACY) == + Unsupported(Some(s"Cast from $BooleanType to $toType is not supported"))) + } + } + test("cast BooleanType to DecimalType(14,4)") { castTest(generateBools(), DataTypes.createDecimalType(14, 4)) } From 98e33b8219d9d1407db4942404ee3e06f1abe296 Mon Sep 17 00:00:00 2001 From: Chao Sun Date: Wed, 19 Aug 2026 09:28:31 -0700 Subject: [PATCH 3/6] fix: canonicalize NaN in flat arrays_overlap float keys (#5376) (cherry picked from commit bdd2e13f57c052b2047c2e630bc0a3b1dcd7fe6d) --- .../src/array_funcs/arrays_overlap.rs | 145 +++++++++++++++++- .../expressions/array/arrays_overlap.sql | 51 ++++++ .../comet/CometArrayExpressionSuite.scala | 27 ++++ 3 files changed, 217 insertions(+), 6 deletions(-) diff --git a/native/spark-expr/src/array_funcs/arrays_overlap.rs b/native/spark-expr/src/array_funcs/arrays_overlap.rs index bd75a6ddccb..4d74718f162 100644 --- a/native/spark-expr/src/array_funcs/arrays_overlap.rs +++ b/native/spark-expr/src/array_funcs/arrays_overlap.rs @@ -272,10 +272,9 @@ fn range_has_null(nulls: Option<&NullBuffer>, range: Range) -> bool { nulls.is_some_and(|n| n.null_count() > 0 && n.slice(range.start, range.len()).null_count() > 0) } -/// Projects a native value onto a hashable key whose equality matches the Arrow compare kernels: -/// the value itself for integral types, the bit pattern for floats. Arrow orders floats by total -/// order rather than IEEE semantics (NaN equals NaN, and 0.0 does not equal -0.0), which is -/// exactly bit equality. +/// Projects a native value onto a Spark-compatible hashable key. Floating-point keys canonicalize +/// every NaN representation while preserving the distinct bit patterns of positive and negative +/// zero. trait OverlapKey: Copy { type Key: Hash + Eq + Copy; @@ -299,7 +298,11 @@ impl OverlapKey for f32 { type Key = u32; fn overlap_key(self) -> u32 { - self.to_bits() + if self.is_nan() { + f32::NAN.to_bits() + } else { + self.to_bits() + } } } @@ -307,7 +310,11 @@ impl OverlapKey for f64 { type Key = u64; fn overlap_key(self) -> u64 { - self.to_bits() + if self.is_nan() { + f64::NAN.to_bits() + } else { + self.to_bits() + } } } @@ -554,6 +561,132 @@ mod tests { Ok(()) } + #[test] + fn test_flat_float32_nan_payloads_and_signed_zero() -> Result<()> { + let positive_nan = f32::from_bits(0x7fc0_0001); + let negative_nan = f32::from_bits(0xffc0_0002); + let signaling_nan = f32::from_bits(0x7f80_0001); + + let hash_nan_left = (1..=16) + .map(|value| Some(value as f32)) + .chain([Some(positive_nan)]) + .collect::>(); + let hash_nan_right = (17..=32) + .map(|value| Some(value as f32)) + .chain([Some(negative_nan)]) + .collect::>(); + let hash_zero_left = (1..=16) + .map(|value| Some(value as f32)) + .chain([Some(0.0)]) + .collect::>(); + let hash_zero_right = (17..=32) + .map(|value| Some(value as f32)) + .chain([Some(-0.0)]) + .collect::>(); + + let left = ListArray::from_iter_primitive::([ + Some(vec![Some(positive_nan)]), + Some(vec![Some(negative_nan)]), + Some(vec![Some(signaling_nan)]), + Some(vec![Some(0.0)]), + Some(vec![Some(-0.0)]), + Some(vec![Some(positive_nan), None]), + Some(vec![Some(0.0), None]), + Some(hash_nan_left), + Some(hash_zero_left), + ]); + let right = ListArray::from_iter_primitive::([ + Some(vec![Some(f32::NAN)]), + Some(vec![Some(positive_nan)]), + Some(vec![Some(negative_nan)]), + Some(vec![Some(-0.0)]), + Some(vec![Some(0.0)]), + Some(vec![Some(negative_nan)]), + Some(vec![Some(-0.0)]), + Some(hash_nan_right), + Some(hash_zero_right), + ]); + + let result = arrays_overlap_list::(&left, &right)?; + let result = result.as_any().downcast_ref::().unwrap(); + let expected = BooleanArray::from(vec![ + Some(true), + Some(true), + Some(true), + Some(false), + Some(false), + Some(true), + None, + Some(true), + Some(false), + ]); + assert_eq!(result, &expected); + Ok(()) + } + + #[test] + fn test_flat_float64_nan_payloads_and_signed_zero() -> Result<()> { + let positive_nan = f64::from_bits(0x7ff8_0000_0000_0001); + let negative_nan = f64::from_bits(0xfff8_0000_0000_0002); + let signaling_nan = f64::from_bits(0x7ff0_0000_0000_0001); + + let hash_nan_left = (1..=16) + .map(|value| Some(value as f64)) + .chain([Some(positive_nan)]) + .collect::>(); + let hash_nan_right = (17..=32) + .map(|value| Some(value as f64)) + .chain([Some(negative_nan)]) + .collect::>(); + let hash_zero_left = (1..=16) + .map(|value| Some(value as f64)) + .chain([Some(0.0)]) + .collect::>(); + let hash_zero_right = (17..=32) + .map(|value| Some(value as f64)) + .chain([Some(-0.0)]) + .collect::>(); + + let left = ListArray::from_iter_primitive::([ + Some(vec![Some(positive_nan)]), + Some(vec![Some(negative_nan)]), + Some(vec![Some(signaling_nan)]), + Some(vec![Some(0.0)]), + Some(vec![Some(-0.0)]), + Some(vec![Some(positive_nan), None]), + Some(vec![Some(0.0), None]), + Some(hash_nan_left), + Some(hash_zero_left), + ]); + let right = ListArray::from_iter_primitive::([ + Some(vec![Some(f64::NAN)]), + Some(vec![Some(positive_nan)]), + Some(vec![Some(negative_nan)]), + Some(vec![Some(-0.0)]), + Some(vec![Some(0.0)]), + Some(vec![Some(negative_nan)]), + Some(vec![Some(-0.0)]), + Some(hash_nan_right), + Some(hash_zero_right), + ]); + + let result = arrays_overlap_list::(&left, &right)?; + let result = result.as_any().downcast_ref::().unwrap(); + let expected = BooleanArray::from(vec![ + Some(true), + Some(true), + Some(true), + Some(false), + Some(false), + Some(true), + None, + Some(true), + Some(false), + ]); + assert_eq!(result, &expected); + Ok(()) + } + #[test] fn test_null_only_overlap() -> Result<()> { // [1, NULL] vs [NULL, 2] => null (no definite overlap, but nulls present) diff --git a/spark/src/test/resources/sql-tests/expressions/array/arrays_overlap.sql b/spark/src/test/resources/sql-tests/expressions/array/arrays_overlap.sql index f2a47a33ac1..0b3d55366e0 100644 --- a/spark/src/test/resources/sql-tests/expressions/array/arrays_overlap.sql +++ b/spark/src/test/resources/sql-tests/expressions/array/arrays_overlap.sql @@ -122,6 +122,57 @@ INSERT INTO test_overlap_dbl VALUES (array(1.0, 2.0), array(2.0, 3.0)), (array(1 query SELECT a, b, arrays_overlap(a, b) FROM test_overlap_dbl +-- Flat FLOAT/DOUBLE NaNs must compare equal regardless of their sign or payload. +-- Parquet canonicalizes NaNs, so negate a scanned column to produce a noncanonical +-- NaN at execution time. String casts preserve the sign of the zero controls. +statement +CREATE TABLE test_overlap_floating_point( + id int, fl float, fr float, dl double, dr double +) USING parquet + +statement +INSERT INTO test_overlap_floating_point VALUES + (0, CAST('NaN' AS FLOAT), CAST('NaN' AS FLOAT), CAST('NaN' AS DOUBLE), CAST('NaN' AS DOUBLE)), + (1, CAST('0.0' AS FLOAT), CAST('-0.0' AS FLOAT), CAST('0.0' AS DOUBLE), CAST('-0.0' AS DOUBLE)), + (2, CAST('-0.0' AS FLOAT), CAST('0.0' AS FLOAT), CAST('-0.0' AS DOUBLE), CAST('0.0' AS DOUBLE)), + (3, CAST('0.0' AS FLOAT), CAST('0.0' AS FLOAT), CAST('0.0' AS DOUBLE), CAST('0.0' AS DOUBLE)), + (4, CAST('-0.0' AS FLOAT), CAST('-0.0' AS FLOAT), CAST('-0.0' AS DOUBLE), CAST('-0.0' AS DOUBLE)), + (5, CAST(1 AS FLOAT), CAST(2 AS FLOAT), CAST(1 AS DOUBLE), CAST(2 AS DOUBLE)), + (6, CAST('NaN' AS FLOAT), CAST(1 AS FLOAT), CAST('NaN' AS DOUBLE), CAST(1 AS DOUBLE)), + (7, CAST(1 AS FLOAT), CAST('NaN' AS FLOAT), CAST(1 AS DOUBLE), CAST('NaN' AS DOUBLE)) + +-- Short arrays, reversed operands, and direct signed-zero controls. Spark 4.2+ +-- normalizes signed zeros before arrays_overlap (SPARK-54918); comparing with the +-- running Spark version checks the appropriate behavior without fixed expectations. +query +SELECT id, + arrays_overlap(array(fl), array(-fr)) AS float_short, + arrays_overlap(array(dl), array(-dr)) AS double_short, + arrays_overlap(array(-fr), array(fl)) AS float_reversed, + arrays_overlap(array(-dr), array(dl)) AS double_reversed, + arrays_overlap(array(fl), array(fr)) AS float_direct, + arrays_overlap(array(dl), array(dr)) AS double_direct +FROM test_overlap_floating_point + +-- Seventeen elements on each side exceed the native budget of 256 comparisons +-- and exercise the hash-probe path, including signed-zero and nonmatching rows. +query +SELECT id, + arrays_overlap(array_repeat(fl, 17), array_repeat(-fr, 17)) AS float_long, + arrays_overlap(array_repeat(dl, 17), array_repeat(-dr, 17)) AS double_long, + arrays_overlap(array_repeat(-fr, 17), array_repeat(fl, 17)) AS float_long_reversed, + arrays_overlap(array_repeat(-dr, 17), array_repeat(dl, 17)) AS double_long_reversed +FROM test_overlap_floating_point + +-- A definite NaN match wins over NULL; a nonmatch with NULL remains unknown. +query +SELECT id, + arrays_overlap(array(fl, CAST(NULL AS FLOAT)), array(-fr)), + arrays_overlap(array(dl, CAST(NULL AS DOUBLE)), array(-dr)), + arrays_overlap(array(-fr), array(fl, CAST(NULL AS FLOAT))), + arrays_overlap(array(-dr), array(dl, CAST(NULL AS DOUBLE))) +FROM test_overlap_floating_point + -- boolean arrays query SELECT arrays_overlap(array(true, false), array(false)) FROM test_overlap_dbl diff --git a/spark/src/test/scala/org/apache/comet/CometArrayExpressionSuite.scala b/spark/src/test/scala/org/apache/comet/CometArrayExpressionSuite.scala index 7377f4fe4fc..05a6e8e650d 100644 --- a/spark/src/test/scala/org/apache/comet/CometArrayExpressionSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometArrayExpressionSuite.scala @@ -560,6 +560,33 @@ class CometArrayExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelp } } + test("arrays_overlap - runtime NaN representations") { + val floatNaN = java.lang.Float.intBitsToFloat(0x7fc01234 | Int.MinValue) + val doubleNaN = java.lang.Double.longBitsToDouble(0x7ff8000000001234L | Long.MinValue) + + withParquetTable( + Seq((floatNaN, doubleNaN)), + "floating_point_overlap", + withDictionary = false) { + // The behavioral cases live in arrays_overlap.sql. SQL equality cannot distinguish NaN + // representations, so verify here that Parquet canonicalizes the inputs and that native + // runtime negation produces noncanonical NaNs after the scan. + val query = sql("SELECT _1, -_1, _2, -_2 FROM floating_point_overlap") + checkSparkAnswerAndOperator(query) + val row = query.head() + val canonicalFloatNaNBits = java.lang.Float.floatToRawIntBits(Float.NaN) + val canonicalDoubleNaNBits = java.lang.Double.doubleToRawLongBits(Double.NaN) + assert(java.lang.Float.floatToRawIntBits(row.getFloat(0)) == canonicalFloatNaNBits) + assert( + java.lang.Float.floatToRawIntBits(row.getFloat(1)) == + (canonicalFloatNaNBits | Int.MinValue)) + assert(java.lang.Double.doubleToRawLongBits(row.getDouble(2)) == canonicalDoubleNaNBits) + assert( + java.lang.Double.doubleToRawLongBits(row.getDouble(3)) == + (canonicalDoubleNaNBits | Long.MinValue)) + } + } + test("arrays_overlap - null handling behavior verification") { withSQLConf( "spark.sql.optimizer.excludedRules" -> "org.apache.spark.sql.catalyst.optimizer.ConstantFolding") { From 56d567d1eb183b1e272e8172eec1cf83cae39365 Mon Sep 17 00:00:00 2001 From: Parth Chandra Date: Thu, 20 Aug 2026 09:30:11 -0700 Subject: [PATCH 4/6] fix: Native shuffle fails with a 2GB task serialization OOM on jobs with many partitions (#5392) * fix: Native shuffle fails with a 2GB task serialization OOM on jobs with very many partitions (cherry picked from commit 7e0e5d2a26928b05b936c3f2a9abd375d6efc41f) --- .github/workflows/pr_build_linux.yml | 1 + .github/workflows/pr_build_macos.yml | 1 + .../shuffle/CometNativeShuffleInputRDD.scala | 31 ++++-- .../shuffle/CometNativeShuffleWriter.scala | 12 ++- .../shuffle/CometShuffleExchangeExec.scala | 16 ++- .../apache/spark/sql/comet/operators.scala | 7 +- .../comet/exec/CometNativeShuffleSuite.scala | 28 +++++ .../CometNativeShuffleInputRDDSuite.scala | 101 ++++++++++++++++++ 8 files changed, 181 insertions(+), 16 deletions(-) create mode 100644 spark/src/test/scala/org/apache/spark/sql/comet/execution/shuffle/CometNativeShuffleInputRDDSuite.scala diff --git a/.github/workflows/pr_build_linux.yml b/.github/workflows/pr_build_linux.yml index 7a304eeeb84..d2f50bf09b6 100644 --- a/.github/workflows/pr_build_linux.yml +++ b/.github/workflows/pr_build_linux.yml @@ -325,6 +325,7 @@ jobs: org.apache.comet.exec.CometShuffle4_0Suite org.apache.comet.exec.CometNativeColumnarToRowSuite org.apache.comet.exec.CometNativeShuffleSuite + org.apache.spark.sql.comet.execution.shuffle.CometNativeShuffleInputRDDSuite org.apache.comet.exec.CometShuffleEncryptionSuite org.apache.comet.exec.CometShuffleManagerSuite org.apache.comet.exec.CometAsyncShuffleSuite diff --git a/.github/workflows/pr_build_macos.yml b/.github/workflows/pr_build_macos.yml index 46f672a56a3..59479028e3c 100644 --- a/.github/workflows/pr_build_macos.yml +++ b/.github/workflows/pr_build_macos.yml @@ -141,6 +141,7 @@ jobs: org.apache.comet.exec.CometShuffle4_0Suite org.apache.comet.exec.CometNativeColumnarToRowSuite org.apache.comet.exec.CometNativeShuffleSuite + org.apache.spark.sql.comet.execution.shuffle.CometNativeShuffleInputRDDSuite org.apache.comet.exec.CometShuffleEncryptionSuite org.apache.comet.exec.CometShuffleManagerSuite org.apache.comet.exec.CometAsyncShuffleSuite diff --git a/spark/src/main/scala/org/apache/spark/sql/comet/execution/shuffle/CometNativeShuffleInputRDD.scala b/spark/src/main/scala/org/apache/spark/sql/comet/execution/shuffle/CometNativeShuffleInputRDD.scala index 0579e57bce1..b79fb9458c6 100644 --- a/spark/src/main/scala/org/apache/spark/sql/comet/execution/shuffle/CometNativeShuffleInputRDD.scala +++ b/spark/src/main/scala/org/apache/spark/sql/comet/execution/shuffle/CometNativeShuffleInputRDD.scala @@ -38,7 +38,8 @@ private[shuffle] class CometNativeShuffleInputRDD( sc: SparkContext, var inputRDDs: Seq[RDD[_]], numPartitionsParam: Int, - shuffleScanIndices: Set[Int]) + shuffleScanIndices: Set[Int], + @transient perPartitionByKey: Map[String, Array[Array[Byte]]] = Map.empty) extends RDD[Product2[Int, ColumnarBatch]]( sc, inputRDDs.map(rdd => new OneToOneDependency(rdd))) { @@ -50,7 +51,13 @@ private[shuffle] class CometNativeShuffleInputRDD( // `leafRdd.partitions` on the executor, which would otherwise trigger getPartitions and // hit the @transient-null trap (e.g. CometExecRDD.perPartitionByKey). val inputParts = inputRDDs.map(_.partitions(i)).toArray - new CometNativeShuffleInputPartition(i, inputParts) + // Slice this partition's plan data off the @transient full map here on the driver. Carrying + // only the per-partition slice on the Partition object (serialized per task) keeps the full + // O(numPartitions) map out of the broadcast task binary, which otherwise blows the 2GB + // ByteArrayOutputStream limit on jobs with tens of millions of partitions. Mirrors + // CometExecRDD.getPartitions. + val planDataByKey = perPartitionByKey.map { case (key, arr) => key -> arr(i) } + new CometNativeShuffleInputPartition(i, inputParts, planDataByKey) }.toArray override def compute( @@ -63,7 +70,11 @@ private[shuffle] class CometNativeShuffleInputRDD( partition.inputPartitions, shuffleScanIndices, context) - new CometNativeShuffleInputIterator(partition.index, inputObjects, shuffleBlockIters) + new CometNativeShuffleInputIterator( + partition.index, + inputObjects, + shuffleBlockIters, + partition.planDataByKey) } override def getPreferredLocations(split: Partition): Seq[String] = { @@ -84,19 +95,23 @@ private[shuffle] class CometNativeShuffleInputRDD( private[shuffle] class CometNativeShuffleInputPartition( override val index: Int, - val inputPartitions: Array[Partition]) + val inputPartitions: Array[Partition], + val planDataByKey: Map[String, Array[Byte]]) extends Partition /** * Iterator handed to [[CometNativeShuffleWriter.write]] via Spark's ShuffleMapTask. Reports no - * elements; the writer downcasts and reads `partitionIndex`, `inputObjects`, and - * `shuffleBlockIterators` directly to drive the unified native plan. `inputObjects` are the - * already-resolved native input slots (see [[CometExecRDD.resolveInputObjects]]). + * elements; the writer downcasts and reads `partitionIndex`, `inputObjects`, + * `shuffleBlockIterators`, and `planDataByKey` directly to drive the unified native plan. + * `inputObjects` are the already-resolved native input slots (see + * [[CometExecRDD.resolveInputObjects]]). `planDataByKey` is this partition's slice of the scan + * plan data (one entry per scan key); the writer injects it into the native plan. */ private[shuffle] class CometNativeShuffleInputIterator( val partitionIndex: Int, val inputObjects: Array[Object], - val shuffleBlockIterators: Map[Int, CometShuffleBlockIterator]) + val shuffleBlockIterators: Map[Int, CometShuffleBlockIterator], + val planDataByKey: Map[String, Array[Byte]]) extends Iterator[Product2[Int, ColumnarBatch]] { override def hasNext: Boolean = false diff --git a/spark/src/main/scala/org/apache/spark/sql/comet/execution/shuffle/CometNativeShuffleWriter.scala b/spark/src/main/scala/org/apache/spark/sql/comet/execution/shuffle/CometNativeShuffleWriter.scala index 5b1145e9e07..74a42124621 100644 --- a/spark/src/main/scala/org/apache/spark/sql/comet/execution/shuffle/CometNativeShuffleWriter.scala +++ b/spark/src/main/scala/org/apache/spark/sql/comet/execution/shuffle/CometNativeShuffleWriter.scala @@ -97,10 +97,14 @@ class CometNativeShuffleWriter[K, V]( val unifiedPlan = buildUnifiedPlan(tempDataFilename, tempIndexFilename) val ctx = spec.execContext val finalNativePlan = if (ctx.commonByKey.nonEmpty) { - val partitionDataByKey = ctx.perPartitionByKey.map { case (k, arr) => - k -> arr(partitionIdx) - } - PlanDataInjector.injectPlanData(unifiedPlan, ctx.commonByKey, partitionDataByKey) + // This partition's plan-data slice rides on the input iterator's Partition object (populated + // in CometNativeShuffleInputRDD.getPartitions on the driver), not on the spec. The spec's + // execContext.perPartitionByKey is emptied in prepareNativeShuffleDependency so the full + // O(numPartitions) map stays out of the broadcast task binary. + PlanDataInjector.injectPlanData( + unifiedPlan, + ctx.commonByKey, + shuffleInputIter.planDataByKey) } else { unifiedPlan } diff --git a/spark/src/main/scala/org/apache/spark/sql/comet/execution/shuffle/CometShuffleExchangeExec.scala b/spark/src/main/scala/org/apache/spark/sql/comet/execution/shuffle/CometShuffleExchangeExec.scala index 9505443fb31..84f313ad37c 100644 --- a/spark/src/main/scala/org/apache/spark/sql/comet/execution/shuffle/CometShuffleExchangeExec.scala +++ b/spark/src/main/scala/org/apache/spark/sql/comet/execution/shuffle/CometShuffleExchangeExec.scala @@ -122,7 +122,8 @@ case class CometShuffleExchangeExec( sparkContext, ctx.inputs, ctx.numPartitions, - ctx.shuffleScanIndices) + ctx.shuffleScanIndices, + ctx.perPartitionByKey) case None => // Non-native child (e.g. CometSparkToColumnarExec): no subtree to inline. The dep gets // built via the convenience overload below; we just need a real RDD of batches. @@ -786,8 +787,17 @@ object CometShuffleExchangeExec case e: Expression => e.collect { case s: ScalarSubquery => s } case _ => Nil } - val augmentedSpec = spec.copy(execContext = - spec.execContext.copy(subqueries = spec.execContext.subqueries ++ partitioningSubqueries)) + // Drop the per-partition plan-data map off the spec that lands on the (non-transient) + // CometShuffleDependency.nativeShuffleSpec. Each partition's slice now rides on the thin RDD's + // Partition objects (see CometNativeShuffleInputRDD.getPartitions), so the full + // O(numPartitions) map is dead weight here and would blow the 2GB ByteArrayOutputStream limit + // at stage submission on very-high-partition-count jobs. NativeExecContext.perPartitionByKey is + // also @transient (the structural guard against any build path), but we empty it explicitly + // here too so the map isn't retained on the driver via this dependency. commonByKey stays + // (O(#scans), not O(#partitions), and the writer still needs it). + val augmentedSpec = spec.copy(execContext = spec.execContext.copy( + subqueries = spec.execContext.subqueries ++ partitioningSubqueries, + perPartitionByKey = Map.empty)) // The code block below is mostly brought over from // ShuffleExchangeExec::prepareShuffleDependency diff --git a/spark/src/main/scala/org/apache/spark/sql/comet/operators.scala b/spark/src/main/scala/org/apache/spark/sql/comet/operators.scala index 05439a16e3a..73d50931c19 100644 --- a/spark/src/main/scala/org/apache/spark/sql/comet/operators.scala +++ b/spark/src/main/scala/org/apache/spark/sql/comet/operators.scala @@ -545,7 +545,12 @@ private[comet] case class NativeExecContext( broadcastedHadoopConfForEncryption: Option[Broadcast[SerializableConfiguration]], encryptedFilePaths: Seq[String], commonByKey: Map[String, Array[Byte]], - perPartitionByKey: Map[String, Array[Array[Byte]]], + // @transient: this holds one serialized scan-plan-data blob per partition, so at high partition + // counts it is huge. It is only read on the driver (to slice per partition onto each task's + // Partition object - see CometNativeShuffleInputRDD / CometExecRDD); the executor reads its own + // slice, never this map. Keeping it off the wire stops it from bloating the broadcast task + // binary when this context rides on the non-transient CometShuffleDependency.nativeShuffleSpec. + @transient perPartitionByKey: Map[String, Array[Array[Byte]]], shuffleScanIndices: Set[Int], hasScanInput: Boolean) { // Catch shape divergence (e.g. broadcast scans with different partition counts after DPP diff --git a/spark/src/test/scala/org/apache/comet/exec/CometNativeShuffleSuite.scala b/spark/src/test/scala/org/apache/comet/exec/CometNativeShuffleSuite.scala index ced6531cd8d..cd52f579559 100644 --- a/spark/src/test/scala/org/apache/comet/exec/CometNativeShuffleSuite.scala +++ b/spark/src/test/scala/org/apache/comet/exec/CometNativeShuffleSuite.scala @@ -108,6 +108,34 @@ class CometNativeShuffleSuite extends CometTestBase with AdaptiveSparkPlanHelper } } + test("native shuffle over a multi-partition native scan re-threads per-partition plan data") { + // End-to-end companion to CometNativeShuffleInputRDDSuite: that suite proves the per-partition + // scan plan data no longer rides the broadcast task binary; this one proves each task still + // gets its OWN slice at write time. Reading real Parquet files gives a CometNativeScanExec with + // several map partitions, so perPartitionByKey holds one file-list slice per partition and a + // correct result depends on task i seeing slice i (not partition 0's). + withTempDir { dir => + val path = new Path(dir.toURI.toString, "multi.parquet") + // Spread rows across 8 files so the native scan can yield multiple map partitions. + spark + .range(0, 10000, 1, numPartitions = 8) + .selectExpr("id AS _1", "CAST(id AS STRING) AS _2") + .write + .parquet(path.toString) + + // Force one scan partition per file split so the per-partition array has multiple distinct + // entries; otherwise Spark coalesces the tiny files into a single partition. + withSQLConf( + "spark.sql.files.maxPartitionBytes" -> "1024", + "spark.sql.files.openCostInBytes" -> "0") { + readParquetFile(path.toString) { df => + val shuffled = df.repartition(17, col("_1")) + checkShuffleAnswer(shuffled, 1, checkNativeOperators = true) + } + } + } + } + test("hash-based native shuffle") { withParquetTable((0 until 5).map(i => (i, (i + 1).toLong)), "tbl") { val df = sql("SELECT * FROM tbl").sortWithinPartitions($"_1".desc) diff --git a/spark/src/test/scala/org/apache/spark/sql/comet/execution/shuffle/CometNativeShuffleInputRDDSuite.scala b/spark/src/test/scala/org/apache/spark/sql/comet/execution/shuffle/CometNativeShuffleInputRDDSuite.scala new file mode 100644 index 00000000000..55194e80a27 --- /dev/null +++ b/spark/src/test/scala/org/apache/spark/sql/comet/execution/shuffle/CometNativeShuffleInputRDDSuite.scala @@ -0,0 +1,101 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.spark.sql.comet.execution.shuffle + +import org.apache.spark.HashPartitioner +import org.apache.spark.serializer.JavaSerializer +import org.apache.spark.sql.CometTestBase +import org.apache.spark.sql.comet.{CometMetricNode, NativeExecContext} +import org.apache.spark.sql.execution.metric.SQLMetrics +import org.apache.spark.sql.vectorized.ColumnarBatch + +import org.apache.comet.serde.OperatorOuterClass.Operator + +/** + * Ensure the serialized shuffle-map-stage task binary does not grow with the number of + * partitions. + * + * `DAGScheduler.submitMissingTasks` broadcasts the serialized `(stage.rdd, stage.shuffleDep)` + * pair, which must fit in ~2GB. On the native-shuffle path the scan plan data is one serialized + * blob per map partition; the leak lived under + * `CometShuffleDependency.nativeShuffleSpec.execContext`, not on the thin RDD. So this builds + * both carriers and serializes the pair, which catches a regression in either the RDD or the + * dependency guards without needing a real 2GB allocation. + * + * Lives in the `execution.shuffle` package so it can construct the `private[shuffle]` + * [[CometNativeShuffleInputRDD]] and the `private[comet]` [[NativeExecContext]] directly. + */ +class CometNativeShuffleInputRDDSuite extends CometTestBase { + + test("serialized (rdd, dep) task binary size is independent of partition count") { + val sc = spark.sparkContext + val ser = new JavaSerializer(sc.getConf).newInstance() + + // One scan key with a 1KB blob per map partition -- the per-partition plan data. Build both + // objects the DAGScheduler serializes for a shuffle-map stage: the thin RDD, and the + // CometShuffleDependency whose NativeShuffleSpec holds a NativeExecContext with this map. + def build(numPartitions: Int): ( + CometNativeShuffleInputRDD, + CometShuffleDependency[Int, ColumnarBatch, ColumnarBatch]) = { + val perPartitionByKey = + Map("scan-0" -> Array.fill(numPartitions)(new Array[Byte](1024))) + val rdd = new CometNativeShuffleInputRDD( + sc, + inputRDDs = Seq.empty, + numPartitionsParam = numPartitions, + shuffleScanIndices = Set.empty, + perPartitionByKey = perPartitionByKey) + val execContext = NativeExecContext( + inputs = Seq.empty, + numPartitions = numPartitions, + subqueries = Seq.empty, + broadcastedHadoopConfForEncryption = None, + encryptedFilePaths = Seq.empty, + commonByKey = Map.empty, + perPartitionByKey = perPartitionByKey, + shuffleScanIndices = Set.empty, + hasScanInput = false) + val spec = + NativeShuffleSpec(Operator.getDefaultInstance, CometMetricNode(Map.empty), execContext) + val dep = new CometShuffleDependency[Int, ColumnarBatch, ColumnarBatch]( + _rdd = rdd, + partitioner = new HashPartitioner(numPartitions), + decodeTime = SQLMetrics.createMetric(sc, "decode time"), + nativeShuffleSpec = Some(spec)) + (rdd, dep) + } + + // Pre-fix this pair grew from ~13KB to ~10MB between 10 and 10000 partitions. With the map held + // @transient on both the RDD and the NativeExecContext, the pair stays roughly constant. + val (_, smallDep) = build(10) + val (largeRdd, largeDep) = build(10000) + val smallPair = ser.serialize((smallDep.rdd, smallDep)).limit() + val largePair = ser.serialize((largeDep.rdd, largeDep)).limit() + assert( + math.abs(largePair - smallPair) < 100 * 1024, + s"serialized (rdd, dep) grew with partition count (small=$smallPair, large=$largePair); " + + "the per-partition plan-data map is leaking into the broadcast task binary") + + // Each task's Partition object still carries its own slice so the writer can inject plan data. + val part = largeRdd.partitions(7).asInstanceOf[CometNativeShuffleInputPartition] + assert(part.planDataByKey.keySet == Set("scan-0")) + assert(part.planDataByKey("scan-0").length == 1024) + } +} From 63ca4559932e3045903205ac57567812b69800cb Mon Sep 17 00:00:00 2001 From: Peter Lee Date: Mon, 24 Aug 2026 06:50:48 +0800 Subject: [PATCH 5/6] fix: support wide years in native make_date (#5443) Co-authored-by: Chao Sun (cherry picked from commit e0ab0a6fe60c05bd679654f0201ebe88319cc3a7) --- native/spark-expr/src/conversion_funcs/mod.rs | 2 + .../spark-expr/src/conversion_funcs/string.rs | 2 +- .../src/datetime_funcs/make_date.rs | 53 +++++-------------- .../org/apache/comet/serde/datetime.scala | 7 --- .../expressions/datetime/make_date.sql | 3 ++ .../expressions/datetime/make_date_ansi.sql | 3 ++ 6 files changed, 21 insertions(+), 49 deletions(-) diff --git a/native/spark-expr/src/conversion_funcs/mod.rs b/native/spark-expr/src/conversion_funcs/mod.rs index 7b864e193b8..a9da6915f97 100644 --- a/native/spark-expr/src/conversion_funcs/mod.rs +++ b/native/spark-expr/src/conversion_funcs/mod.rs @@ -22,3 +22,5 @@ mod string; mod temporal; pub(crate) mod trim; mod utils; + +pub(crate) use string::ymd_to_epoch_day; diff --git a/native/spark-expr/src/conversion_funcs/string.rs b/native/spark-expr/src/conversion_funcs/string.rs index 46495ed9819..9c7dfcd24dd 100644 --- a/native/spark-expr/src/conversion_funcs/string.rs +++ b/native/spark-expr/src/conversion_funcs/string.rs @@ -1234,7 +1234,7 @@ fn is_leap_year(year: i64) -> bool { /// Days since 1970-01-01 for a proleptic Gregorian year/month/day, or `None` when the /// combination is not a real calendar date. Unlike `NaiveDate::from_ymd_opt`, this accepts /// any year that fits in `i64`. -fn ymd_to_epoch_day(year: i64, month: i64, day: i64) -> Option { +pub(crate) fn ymd_to_epoch_day(year: i64, month: i64, day: i64) -> Option { const DAYS_IN_MONTH: [i64; 12] = [31, 28, 31, 30, 31, 30, 31, 31, 30, 31, 30, 31]; let mut max_day = *DAYS_IN_MONTH.get(usize::try_from(month.checked_sub(1)?).ok()?)?; if month == 2 && is_leap_year(year) { diff --git a/native/spark-expr/src/datetime_funcs/make_date.rs b/native/spark-expr/src/datetime_funcs/make_date.rs index b29094ba9fd..257030c75ed 100644 --- a/native/spark-expr/src/datetime_funcs/make_date.rs +++ b/native/spark-expr/src/datetime_funcs/make_date.rs @@ -18,14 +18,13 @@ use arrow::array::{Array, Date32Array, Int32Array}; use arrow::compute::cast; use arrow::datatypes::DataType; -use chrono::NaiveDate; use datafusion::common::{utils::take_function_args, DataFusionError, Result}; use datafusion::logical_expr::{ ColumnarValue, ScalarFunctionArgs, ScalarUDFImpl, Signature, Volatility, }; use std::sync::Arc; -use crate::SparkError; +use crate::{conversion_funcs::ymd_to_epoch_day, SparkError}; /// Spark-compatible make_date function. /// Creates a date from year, month, and day columns. @@ -98,18 +97,10 @@ 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. +/// 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 { - // Validate month and day ranges first - if !(1..=12).contains(&month) || !(1..=31).contains(&day) { - return None; - } - - // Try to create a valid date - NaiveDate::from_ymd_opt(year, month as u32, day as u32).map(|date| { - date.signed_duration_since(NaiveDate::from_ymd_opt(1970, 1, 1).unwrap()) - .num_days() as i32 - }) + ymd_to_epoch_day(year.into(), month.into(), day.into()) + .and_then(|days| i32::try_from(days).ok()) } impl ScalarUDFImpl for SparkMakeDate { @@ -246,33 +237,13 @@ mod tests { } #[test] - fn test_make_date_extreme_years() { - // Spark supports dates from 0001-01-01 to 9999-12-31 (Proleptic Gregorian calendar) - - // Minimum valid date in Spark: 0001-01-01 - assert!(make_date(1, 1, 1).is_some(), "Year 1 should be valid"); - - // Maximum valid date in Spark: 9999-12-31 - assert!( - make_date(9999, 12, 31).is_some(), - "Year 9999 should be valid" - ); - - // Year 0 - In Proleptic Gregorian calendar, year 0 = 1 BCE - // Spark returns NULL for year 0 in make_date - // chrono supports year 0, but we should match Spark's behavior - // For now, chrono allows it - this may need adjustment for full Spark compatibility - let year_0_result = make_date(0, 1, 1); - // chrono allows year 0 (1 BCE in proleptic Gregorian) - assert!(year_0_result.is_some(), "chrono allows year 0"); - - // Negative years - Spark returns NULL for negative years - // chrono supports negative years (BCE dates) - let negative_year_result = make_date(-1, 1, 1); - // chrono allows negative years - assert!( - negative_year_result.is_some(), - "chrono allows negative years" - ); + 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(5_881_580, 7, 11), Some(i32::MAX)); + assert_eq!(make_date(-5_877_641, 6, 23), Some(i32::MIN)); } } diff --git a/spark/src/main/scala/org/apache/comet/serde/datetime.scala b/spark/src/main/scala/org/apache/comet/serde/datetime.scala index 0835a92dd72..db62d0b368e 100644 --- a/spark/src/main/scala/org/apache/comet/serde/datetime.scala +++ b/spark/src/main/scala/org/apache/comet/serde/datetime.scala @@ -476,13 +476,6 @@ object CometMakeDate extends CometExpressionSerde[MakeDate] { * via the `ScalarFunc.fail_on_error` field. */ - override def getCompatibleNotes(): Seq[String] = Seq( - "Native `make_date` is limited to chrono's year range `[-262143, 262142]`; Spark accepts" + - " wider years (for example, `300000`), so Comet returns `NULL` or throws under ANSI mode" + - " for dates Spark accepts, and may incorrectly report valid dates as invalid (for example," + - " `300000-02-29` is falsely reported as not a leap year)" + - " ([#5208](https://github.com/apache/datafusion-comet/issues/5208)).") - override def convert(expr: MakeDate, inputs: Seq[Attribute], binding: Boolean): Option[Expr] = { val childExpr = expr.children.map(exprToProtoInternal(_, inputs, binding)) val optExpr = scalarFunctionExprToProtoWithReturnType( 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 00357662a0b..bd2577cd81e 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 @@ -135,6 +135,9 @@ SELECT make_date(0, 1, 1) query SELECT make_date(-1, 1, 1) +query +SELECT make_date(300000, 6, 15), make_date(300000, 2, 29) + -- month boundaries - last day of each month query SELECT make_date(2023, 1, 31) 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 75ca5304296..dd4a9dfde06 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 @@ -28,6 +28,9 @@ query SELECT make_date(2024, 2, 28) +query +SELECT 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) From 85e0309dcc23598c8992f3af935216ef27b311d2 Mon Sep 17 00:00:00 2001 From: Wei Yan Date: Thu, 20 Aug 2026 21:39:38 -0700 Subject: [PATCH 6/6] chore: fix clippy warnings for Rust 1.98 (#5400) * chore: fix clippy warnings for Rust 1.98 * fix: keep Clippy allow backward-compatible (cherry picked from commit 92954d7884091d2c6fa3e109d11fc1b8cb4a7325) --- native/spark-expr/src/bloom_filter/spark_bit_array.rs | 4 ++-- .../src/nondetermenistic_funcs/internal/mersenne.rs | 2 ++ 2 files changed, 4 insertions(+), 2 deletions(-) diff --git a/native/spark-expr/src/bloom_filter/spark_bit_array.rs b/native/spark-expr/src/bloom_filter/spark_bit_array.rs index 6d43bdb942f..0af1c598b03 100644 --- a/native/spark-expr/src/bloom_filter/spark_bit_array.rs +++ b/native/spark-expr/src/bloom_filter/spark_bit_array.rs @@ -77,8 +77,8 @@ impl SparkBitArray { pub fn merge_be_words(&mut self, incoming: &[u8]) { debug_assert_eq!(self.data.len() * 8, incoming.len()); let mut bit_count: usize = 0; - for (word, chunk) in self.data.iter_mut().zip(incoming.chunks_exact(8)) { - *word |= u64::from_be_bytes(chunk.try_into().unwrap()); + for (word, chunk) in self.data.iter_mut().zip(incoming.as_chunks::<8>().0) { + *word |= u64::from_be_bytes(*chunk); bit_count += word.count_ones() as usize; } self.bit_count = bit_count; diff --git a/native/spark-expr/src/nondetermenistic_funcs/internal/mersenne.rs b/native/spark-expr/src/nondetermenistic_funcs/internal/mersenne.rs index 783fd9e3600..4027c32978a 100644 --- a/native/spark-expr/src/nondetermenistic_funcs/internal/mersenne.rs +++ b/native/spark-expr/src/nondetermenistic_funcs/internal/mersenne.rs @@ -148,6 +148,8 @@ impl SparkMersenneTwister { /// Port of `BitsStreamGenerator.nextInt(int n)`. The caller always passes a /// strictly positive `n`, matching Spark's `random.nextInt(i + 1)`. + // `isolate_lowest_one` requires Rust 1.97, newer than Comet's Rust 1.88 MSRV. + #[allow(unknown_lints, clippy::manual_isolate_lowest_one)] pub(crate) fn next_int(&mut self, n: i32) -> i32 { if (n & n.wrapping_neg()) == n { // n is a power of two.