This is an automated email from the ASF dual-hosted git repository.
andygrove pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
The following commit(s) were added to refs/heads/main by this push:
new 671a8dd9ee fix: make native cast from float/double to decimal
compatible with Spark (#5136)
671a8dd9ee is described below
commit 671a8dd9ee4c4f5bf44f3096ab83b2b30801fcb8
Author: Andy Grove <[email protected]>
AuthorDate: Sat Aug 1 14:03:33 2026 -0600
fix: make native cast from float/double to decimal compatible with Spark
(#5136)
* fix: make cast from float/double to decimal compatible with Spark
Spark casts float/double to decimal by rounding the shortest decimal
string form of the value (Double.toString followed by
BigDecimal.setScale with HALF_UP), while Comet rounded the binary
value, producing different results for values such as 0.5153125 whose
binary form sits just below the rounding tie that the string form
lands on.
Convert through the shortest round-trip decimal form (via ryu) with
integer rescaling, keeping the vectorized binary path for values
provably far from a rounding tie. Also match Spark by returning null
for NaN and infinity even in ANSI mode.
Closes #1371
* refactor: simplify float to decimal cast internals
Reuse the existing pow10_i128 lookup table instead of computing powers
of ten per element, which also removes the hand-maintained exponent
bound and the redundant zero-mantissa branch. Collapse repeated test
boilerplate behind helpers on both the Rust and Scala sides, and
clarify that the overflow pre-check is a perf-only guard and that
high-scale targets commonly take the exact path.
* test: benchmark float and double to decimal casts
Add FLOAT/DOUBLE to DECIMAL(10,2) and DECIMAL(38,18) cases so the
numeric cast benchmark covers both the guarded binary fast path and the
shortest-decimal-string path of the native float-to-decimal cast.
* test: regenerate TPC-DS plan stability golden files
Casting float/double to decimal is now compatible, so aggregates over
decimal columns no longer report a Cast fallback in the plan output.
* review: share the HALF_UP and digit-scan helpers, note the JDK 19 caveat
Addresses review feedback on #5136:
- `float_to_decimal128` uses a plain `shift as u32` cast instead of a
fallible `try_into().ok()?`; the branch already knows `shift >= 0`, so
the conversion can never fail and the sibling `else` branch and every
other `pow10_i128` caller already spell it this way.
- Factor the HALF_UP-by-power-of-ten integer division into
`div_round_half_up_i128` next to `pow10_i128` in `string.rs` and call
it from both `float_to_decimal128` and `parse_string_to_decimal`,
instead of keeping two copies of the same quotient/remainder rounding.
The `checked_div` guard the string path had is dropped: the divisor is
always `10^n` with `n >= 1`, so it can never be zero.
- Rewrite `parse_decimal_notation` on top of the existing
`digits_to_i128` helper, splitting off the sign and the `.` the way
`parse_string_to_decimal` does, rather than hand-rolling a second digit
accumulator. It now returns `Option` so the (unreachable) overflow case
propagates instead of wrapping. A new test round-trips every shape ryu
can emit, including `f64::MAX` and the smallest subnormal.
- Float/double to decimal reports `Compatible(Some(...))` with the JDK 19
caveat: `Double.toString` only emits the shortest round-trip form on
JDK 19 and later, so on older JDKs Spark's own result can differ for a
value whose shortest form lands exactly on a rounding tie. The cast
audit doc records the same note.
* test: use CometColumnarToRow in q13 and q44 plan stability goldens
Native columnar-to-row conversion is now disabled by default (#5114), so
plans render CometColumnarToRow. The merge from main kept the branch side
of the adjacent COMET-INFO edits in these four files, leaving the stale
CometNativeColumnarToRow name behind.
---
.../expression-audits/conversion_funcs.md | 2 +-
native/Cargo.lock | 1 +
native/spark-expr/Cargo.toml | 3 +
native/spark-expr/src/conversion_funcs/numeric.rs | 317 +++++++++++++++++----
native/spark-expr/src/conversion_funcs/string.rs | 45 ++-
.../org/apache/comet/expressions/CometCast.scala | 15 +-
.../approved-plans-v1_4-spark3_5/q44/extended.txt | 6 +-
.../approved-plans-v1_4-spark4_0/q44/extended.txt | 6 +-
.../approved-plans-v1_4-spark4_0/q6/extended.txt | 2 +-
.../approved-plans-v1_4/q13/extended.txt | 2 +-
.../approved-plans-v1_4/q26/extended.txt | 2 +-
.../approved-plans-v1_4/q27/extended.txt | 2 +-
.../approved-plans-v1_4/q28/extended.txt | 12 +-
.../approved-plans-v1_4/q32/extended.txt | 2 +-
.../approved-plans-v1_4/q44/extended.txt | 8 +-
.../approved-plans-v1_4/q6/extended.txt | 2 +-
.../approved-plans-v1_4/q7/extended.txt | 2 +-
.../approved-plans-v1_4/q85/extended.txt | 2 +-
.../approved-plans-v1_4/q9/extended.txt | 10 +-
.../approved-plans-v1_4/q92/extended.txt | 2 +-
.../approved-plans-v2_7-spark4_0/q6/extended.txt | 2 +-
.../approved-plans-v2_7/q27a/extended.txt | 6 +-
.../approved-plans-v2_7/q6/extended.txt | 2 +-
.../scala/org/apache/comet/CometCastSuite.scala | 56 ++--
.../apache/comet/exec/CometAggregateSuite.scala | 8 +-
.../CometCastNumericToNumericBenchmark.scala | 12 +-
26 files changed, 392 insertions(+), 137 deletions(-)
diff --git
a/docs/source/contributor-guide/expression-audits/conversion_funcs.md
b/docs/source/contributor-guide/expression-audits/conversion_funcs.md
index c11d441a23..64e01822dc 100644
--- a/docs/source/contributor-guide/expression-audits/conversion_funcs.md
+++ b/docs/source/contributor-guide/expression-audits/conversion_funcs.md
@@ -33,7 +33,7 @@
- Spark 4.1 `TimeType` casts have no explicit `Unsupported` arm; they fall
back implicitly but do not appear in the auto-generated compatibility doc
([#4490](https://github.com/apache/datafusion-comet/issues/4490)).
- `CAST(<map> AS <map>)` runs natively via `cast_map_to_map`;
`CometCast.isSupported` recurses into the key and value casts, so the map arm
inherits whichever support level the inner casts report.
- `spark.sql.legacy.castComplexTypesToString.enabled=true` is `Unsupported`
for any array/map/struct-to-string cast, so those casts fall back to Spark
while the flag is on. Comet only implements the default (`{}`-wrapped,
NULL-rendering) formatting. The flag is internal from Spark 4.0 onward and
defaults to false.
- - `CAST(<float|double> AS DECIMAL)` rounding may differ from Spark
(`Incompatible`, gated by `spark.comet.expression.Cast.allowIncompatible`,
tracked at [#1371](https://github.com/apache/datafusion-comet/issues/1371)).
+ - `CAST(<float|double> AS DECIMAL)` is `Compatible`: it rounds the shortest
decimal string form of the value (matching Spark's `Double.toString` +
`BigDecimal.setScale(HALF_UP)` path) rather than the binary value, taking a
guarded binary fast path for values provably far from a rounding tie. NaN and
infinity are null even in ANSI mode, matching Spark, which swallows the
`NumberFormatException` from `Decimal(double)` before the ANSI overflow check.
The support level carries a note: `Dou [...]
- Spark registers the type-name conversion functions (`bigint`, `binary`,
`boolean`, `date`, `decimal`, `double`, `float`, `int`, `smallint`, `string`,
`timestamp`, `tinyint`) as cast aliases. Each lowers to the same `Cast` node,
so Comet handles it via the `cast` implementation with the same compatibility
profile.
- Performance (tuned 2026-07-14, PR
[#4920](https://github.com/apache/datafusion-comet/pull/4920)): narrowing
integer casts (`spark_cast_int_to_int`) map the values buffer in a single pass
with Arrow `unary`/`try_unary` and carry the null buffer over untouched,
replacing an element-by-element `Option`/`Result` iterator-collect. Up to 100x
faster on narrowing casts. Benchmark: `benches/cast_numeric.rs`.
- Performance (tuned 2026-07-15, PR
[#4940](https://github.com/apache/datafusion-comet/pull/4940)):
float/double-to-decimal casts (`cast_floating_point_to_decimal128`) now convert
in a single vectorized `unary_opt` pass that maps out-of-range values (NaN,
infinity, precision overflow) to null, replacing the per-element
`Decimal128Builder` loop. ANSI raises via an O(1) null-count check plus a rare
element-wise rescan. 15-36% faster with no regression on any shape. Benchmark:
`benches/cast [...]
diff --git a/native/Cargo.lock b/native/Cargo.lock
index a920991922..54b748cd9b 100644
--- a/native/Cargo.lock
+++ b/native/Cargo.lock
@@ -2071,6 +2071,7 @@ dependencies = [
"num",
"rand 0.10.2",
"regex",
+ "ryu",
"serde",
"serde_json",
"tokio",
diff --git a/native/spark-expr/Cargo.toml b/native/spark-expr/Cargo.toml
index 477de86fe2..775311935c 100644
--- a/native/spark-expr/Cargo.toml
+++ b/native/spark-expr/Cargo.toml
@@ -43,6 +43,9 @@ futures = { workspace = true }
twox-hash = "2.1.2"
rand = { workspace = true }
base64 = "0.23.0"
+# shortest round-trip float formatting, used to match Spark's
Double.toString-based
+# float/double to decimal conversion
+ryu = "1.0"
uuid = "1.23.3"
[dev-dependencies]
diff --git a/native/spark-expr/src/conversion_funcs/numeric.rs
b/native/spark-expr/src/conversion_funcs/numeric.rs
index cb141880d0..90fc5ffc5b 100644
--- a/native/spark-expr/src/conversion_funcs/numeric.rs
+++ b/native/spark-expr/src/conversion_funcs/numeric.rs
@@ -15,6 +15,7 @@
// specific language governing permissions and limitations
// under the License.
+use crate::conversion_funcs::string::{digits_to_i128, div_round_half_up_i128,
pow10_i128};
use crate::conversion_funcs::utils::cast_overflow;
use crate::conversion_funcs::utils::MICROS_PER_SECOND;
use crate::{EvalMode, SparkError, SparkResult};
@@ -899,33 +900,54 @@ where
{
let input = array.as_any().downcast_ref::<PrimitiveArray<T>>().unwrap();
let mul = 10_f64.powi(scale as i32);
-
- // Single vectorized pass: a value with no in-range integer form (NaN /
infinity) or that
- // does not fit the output precision maps to null. `unary_opt` only
applies the closure to
- // non-null slots and carries the input null buffer over, so it replaces
the per-element
- // builder loop without a second pass.
+ // Perf-only early out: both paths below already map overflow to null, but
this skips
+ // the exact-path string formatting for values (including ±infinity) whose
scaled
+ // magnitude clearly exceeds the precision limit no matter what their
exact decimal
+ // digits are. The slack generously covers the same scaling error that
TIE_GUARD below
+ // bounds, so borderline values still take the exact path.
+ let overflow_bound = 10_f64.powi(precision as i32) * (1.0 + 1e-12);
+
+ // A value with no decimal form (NaN / infinity) or that does not fit the
output
+ // precision maps to null. `unary_opt` only applies the closure to
non-null slots and
+ // carries the input null buffer over. Floats are widened to double first
because Spark
+ // casts through `x.toDouble` before converting to a string.
let result: Decimal128Array = input.unary_opt::<_, Decimal128Type>(|v| {
let f: f64 = v.as_();
- (f * mul)
- .round()
- .to_i128()
- .filter(|x| is_validate_decimal_precision(*x, precision))
+ let x = f * mul;
+ if x.abs() > overflow_bound {
+ // also catches ±infinity inputs, which Spark nulls
+ return None;
+ }
+ // Rounding the binary product agrees with Spark's string-based result
whenever
+ // every value within the combined error of the shortest-string
representation,
+ // the multiplication, and `powi` (a few ulp, at most |x| * 2^-50)
rounds to the
+ // same integer: |x - round(x)| + |x| * TIE_GUARD < 0.5, with
TIE_GUARD at 4x the
+ // error bound. For |x| > 2^47 the guard exceeds 0.5, so large values,
whose
+ // integer neighborhoods are wider than 1, always take the exact path;
for
+ // high-scale targets (e.g. Decimal(38,18)) that makes the exact path
the common
+ // path, not a rare fallback. NaN fails the comparison and takes the
exact path,
+ // which nulls it.
+ const TIE_GUARD: f64 = 3.552713678800501e-15; // 2^-48
+ let r = x.round();
+ if (x - r).abs() + x.abs() * TIE_GUARD < 0.5 {
+ return r
+ .to_i128()
+ .filter(|u| is_validate_decimal_precision(*u, precision));
+ }
+ float_to_decimal128(f, precision, scale)
});
- // ANSI must raise on out-of-range values instead of nulling them.
`unary_opt` only nulls
- // non-null inputs that overflow, so a null count beyond the input's
signals an overflow to
- // report. This check is O(1); the element-wise rescan runs only on the
rare error path and
- // reports the first offending value with Spark's exact error.
+ // ANSI must raise on out-of-range values instead of nulling them, but NaN
/ infinity
+ // stay null even in ANSI mode: Spark catches the NumberFormatException
thrown by
+ // `Decimal(double)` and returns null before the ANSI overflow check runs.
`unary_opt`
+ // only nulls non-null inputs, so a null count beyond the input's signals
a potential
+ // overflow; the element-wise rescan runs only on that rare path and
reports the first
+ // finite offending value with Spark's exact error.
if eval_mode == EvalMode::Ansi && result.null_count() > input.null_count()
{
for i in 0..input.len() {
- if !input.is_null(i) {
+ if !input.is_null(i) && result.is_null(i) {
let input_value: f64 = input.value(i).as_();
- let fits = (input_value * mul)
- .round()
- .to_i128()
- .map(|x| is_validate_decimal_precision(x, precision))
- .unwrap_or(false);
- if !fits {
+ if input_value.is_finite() {
return Err(SparkError::NumericValueOutOfRange {
value: input_value.to_string(),
precision,
@@ -939,6 +961,71 @@ where
Ok(Arc::new(result.with_precision_and_scale(precision, scale)?))
}
+/// Convert a double to a decimal unscaled value with Spark semantics.
+///
+/// Spark converts through `BigDecimal(Double.toString(d)).setScale(scale,
HALF_UP)`: it
+/// rounds the shortest decimal string form of the value, not its exact binary
expansion.
+/// The two disagree for values like 0.5153125 whose binary value
(0.51531249999...) sits
+/// just below the rounding tie that the string form lands on, so a plain
+/// `(f * 10^scale).round()` produces results that differ from Spark.
+///
+/// Returns `None` for NaN / infinity and for results that do not fit
`precision`.
+fn float_to_decimal128(f: f64, precision: u8, scale: i8) -> Option<i128> {
+ if !f.is_finite() {
+ return None;
+ }
+
+ // Shortest round-trip decimal form, same digits as Java's Double.toString
+ let mut buf = ryu::Buffer::new();
+ let (mantissa, exp10) = parse_decimal_notation(buf.format_finite(f))?;
+
+ // value = mantissa * 10^exp10, so unscaled = round(mantissa * 10^(exp10 +
scale))
+ let shift = exp10 + scale as i32;
+ let unscaled = if shift >= 0 {
+ // Overflowing i128 here means the result cannot fit any decimal
precision
+ mantissa.checked_mul(pow10_i128(shift as u32)?)?
+ } else {
+ match pow10_i128(-shift as u32) {
+ // The mantissa has at most 17 significant digits, so dividing by
a power of
+ // ten too large for i128 always rounds to zero
+ None => 0,
+ // Divide with HALF_UP rounding (away from zero on a tie, matching
BigDecimal)
+ Some(div) => div_round_half_up_i128(mantissa, div),
+ }
+ };
+
+ is_validate_decimal_precision(unscaled, precision).then_some(unscaled)
+}
+
+/// Parse ryu's `[-]digits[.digits][e[-]digits]` output into an integer
mantissa and a
+/// base-10 exponent such that the value equals `mantissa * 10^exp10`.
+///
+/// Returns `None` if the mantissa does not fit an `i128`, which cannot happen
for ryu
+/// output: a shortest-form double carries at most 17 significant digits.
+fn parse_decimal_notation(s: &str) -> Option<(i128, i32)> {
+ let (digits, exp10) = match s.split_once('e') {
+ Some((digits, exp)) => (digits, exp.parse::<i32>().expect("exponent
from ryu")),
+ None => (s, 0),
+ };
+ let (digits, negative) = match digits.strip_prefix('-') {
+ Some(unsigned) => (unsigned, true),
+ None => (digits, false),
+ };
+ let (integral, fractional) = match digits.split_once('.') {
+ Some((integral, fractional)) => (integral, fractional),
+ None => (digits, ""),
+ };
+
+ // value = (integral * 10^frac_digits + fractional) * 10^(exp10 -
frac_digits)
+ let frac_digits = fractional.len() as i32;
+ let mantissa = pow10_i128(frac_digits as u32)
+ .and_then(|p| digits_to_i128(integral.as_bytes())?.checked_mul(p))
+ .and_then(|v| v.checked_add(digits_to_i128(fractional.as_bytes())?))?;
+
+ let mantissa = if negative { -mantissa } else { mantissa };
+ Some((mantissa, exp10 - frac_digits))
+}
+
pub(crate) fn spark_cast_nonintegral_numeric_to_integral(
array: &dyn Array,
eval_mode: EvalMode,
@@ -1325,13 +1412,6 @@ mod tests {
}
}
#[test]
- // Currently the cast function depending on `f64::powi`, which has
unspecified precision according to the doc
- // https://doc.rust-lang.org/std/primitive.f64.html#unspecified-precision.
- // Miri deliberately apply random floating-point errors to these
operations to expose bugs
- // https://github.com/rust-lang/miri/issues/4395.
- // The random errors may interfere with test cases at rounding edge, so we
ignore it on miri for now.
- // Once https://github.com/apache/datafusion-comet/issues/1371 is fixed,
this should no longer be an issue.
- #[cfg_attr(miri, ignore)]
fn test_cast_float_to_decimal() {
let a: ArrayRef = Arc::new(Float64Array::from(vec![
Some(42.),
@@ -1350,8 +1430,7 @@ mod tests {
assert_eq!(b.len(), a.len());
let casted = b.as_primitive::<Decimal128Type>();
assert_eq!(casted.value(0), 42000000);
- // https://github.com/apache/datafusion-comet/issues/1371
- // assert_eq!(casted.value(1), 515313);
+ assert_eq!(casted.value(1), 515313);
assert_eq!(casted.value(2), -42424242);
assert_eq!(casted.value(3), 0);
assert_eq!(casted.value(4), 0);
@@ -1363,7 +1442,6 @@ mod tests {
}
#[test]
- #[cfg_attr(miri, ignore)]
fn test_cast_float_to_decimal_no_overflow_fast_path() {
// All values fit precision 10, scale 2, so the vectorized fast path
is taken and the
// input null is preserved.
@@ -1384,7 +1462,6 @@ mod tests {
}
#[test]
- #[cfg_attr(miri, ignore)]
fn test_cast_float_to_decimal_ansi_no_overflow() {
let a: ArrayRef = Arc::new(Float64Array::from(vec![Some(42.0), None,
Some(-1.5)]));
let b =
@@ -1396,7 +1473,6 @@ mod tests {
}
#[test]
- #[cfg_attr(miri, ignore)]
fn test_cast_float_to_decimal_ansi_overflow_errors() {
// Precision overflow: 4242.42 * 10^2 = 424242 does not fit precision
4. Two overflow
// rows with distinct string forms pin that the rescan reports the
*first* offender.
@@ -1419,34 +1495,177 @@ mod tests {
}
}
+ /// Cast an f64 array to Decimal128(precision, scale), panicking on error
+ fn cast_f64_to_decimal128(
+ values: Vec<Option<f64>>,
+ precision: u8,
+ scale: i8,
+ eval_mode: EvalMode,
+ ) -> Decimal128Array {
+ let a: ArrayRef = Arc::new(Float64Array::from(values));
+ cast_floating_point_to_decimal128::<Float64Type>(&a, precision, scale,
eval_mode)
+ .unwrap()
+ .as_primitive::<Decimal128Type>()
+ .clone()
+ }
+
#[test]
+ fn test_cast_float_to_decimal_ansi_nan_and_infinity_are_null() {
+ // Spark converts through Decimal(double) which throws
NumberFormatException for
+ // NaN/infinity; the cast catches it and produces null before the ANSI
overflow
+ // check runs, so these are null even in ANSI mode.
+ let d = cast_f64_to_decimal128(
+ vec![
+ Some(1.0),
+ Some(f64::NAN),
+ Some(f64::INFINITY),
+ Some(f64::NEG_INFINITY),
+ ],
+ 10,
+ 2,
+ EvalMode::Ansi,
+ );
+ assert_eq!(d.value(0), 100);
+ assert!(d.is_null(1));
+ assert!(d.is_null(2));
+ assert!(d.is_null(3));
+ }
+
+ #[test]
+ fn test_parse_decimal_notation_round_trips_ryu_output() {
+ // Every shape ryu can emit: plain integral ("1.0"), plain fractional
("12.34"),
+ // leading-zero fractional ("0.001234", up to 21 fractional digits),
single-digit
+ // exponent form ("1e30") and full exponent form ("1.234e33"). The
extremes pin
+ // down that the mantissa always fits an i128, so the `None` arm stays
unreachable.
+ let mut buf = ryu::Buffer::new();
+ for v in [
+ 0.0,
+ 1.0,
+ -1.0,
+ 12.34,
+ -0.001234,
+ 0.5153125,
+ 1e16,
+ 1e17,
+ 1.5e10,
+ 1e30,
+ 1.2345678901234567e-5,
+ 1.2345678901234567e300,
+ f64::MAX,
+ f64::MIN,
+ f64::MIN_POSITIVE,
+ f64::from_bits(1), // smallest subnormal
+ ] {
+ let formatted = buf.format_finite(v);
+ let (mantissa, exp10) =
+ parse_decimal_notation(formatted).expect("ryu mantissa fits
i128");
+ let reconstructed: f64 =
format!("{mantissa}e{exp10}").parse().unwrap();
+ assert_eq!(reconstructed, v, "{formatted} -> {mantissa}e{exp10}");
+ }
+ }
+
+ #[test]
+ fn test_cast_float_to_decimal_spark_rounding_parity() {
+ // Spark rounds the shortest decimal string form of the double
(Double.toString ->
+ // BigDecimal.setScale(HALF_UP)), not the binary value. These values
sit exactly on
+ // a rounding tie in string form while the binary value is just below
it, so binary
+ // rounding gives a different (wrong) result.
+ let d = cast_f64_to_decimal128(
+ vec![Some(0.5153125), Some(-0.5153125), Some(0.5203125)],
+ 8,
+ 6,
+ EvalMode::Legacy,
+ );
+ assert_eq!(d.value(0), 515313);
+ assert_eq!(d.value(1), -515313);
+ assert_eq!(d.value(2), 520313);
+
+ // HALF_UP rounds away from zero for negative values too
+ let d = cast_f64_to_decimal128(vec![Some(1.005), Some(-1.005)], 10, 2,
EvalMode::Legacy);
+ assert_eq!(d.value(0), 101);
+ assert_eq!(d.value(1), -101);
+
+ // values whose shortest form uses exponent notation
+ let d = cast_f64_to_decimal128(vec![Some(1.5e10)], 20, 2,
EvalMode::Legacy);
+ assert_eq!(d.value(0), 1_500_000_000_000);
+
+ let d = cast_f64_to_decimal128(vec![Some(1e30)], 10, 2,
EvalMode::Legacy);
+ assert!(d.is_null(0));
+ }
+
+ #[test]
+ fn test_cast_float32_to_decimal_widens_to_double_first() {
+ // Spark widens float to double before converting to string, so
1.005f32 becomes
+ // 1.0049999952316284 and rounds down; using the shortest f32 form
("1.005") would
+ // incorrectly round up.
+ let a: ArrayRef = Arc::new(Float32Array::from(vec![Some(1.005f32)]));
+ let b =
+ cast_floating_point_to_decimal128::<Float32Type>(&a, 10, 2,
EvalMode::Legacy).unwrap();
+ let d = b.as_primitive::<Decimal128Type>();
+ assert_eq!(d.value(0), 100);
+
+ // 0.5153125f32 widens to 0.5153124928474426 which rounds down, unlike
the f64 case
+ let a: ArrayRef =
Arc::new(Float32Array::from(vec![Some(0.5153125f32)]));
+ let b =
+ cast_floating_point_to_decimal128::<Float32Type>(&a, 8, 6,
EvalMode::Legacy).unwrap();
+ let d = b.as_primitive::<Decimal128Type>();
+ assert_eq!(d.value(0), 515312);
+ }
+
+ #[test]
+ // Not a correctness concern under miri, just far too slow: 240k values
through the
+ // interpreter. The fixed-value tests above cover both paths under miri.
#[cfg_attr(miri, ignore)]
- fn test_cast_float_to_decimal_ansi_nan_errors() {
- // NaN / infinity have no in-range integer form: `to_i128()` returns
`None`, so the
- // closure nulls the slot and the rescan must recompute `fits ==
false` and raise.
- // This is a distinct path from finite precision overflow — assert
both here to lock
- // the reported value string.
- let a: ArrayRef = Arc::new(Float64Array::from(vec![Some(1.0),
Some(f64::NAN)]));
- match cast_floating_point_to_decimal128::<Float64Type>(&a, 10, 2,
EvalMode::Ansi) {
+ fn test_cast_float_to_decimal_fast_path_matches_exact_path() {
+ use rand::rngs::StdRng;
+ use rand::{RngExt, SeedableRng};
+ let mut rng = StdRng::seed_from_u64(42);
+ let mut values: Vec<f64> = Vec::new();
+ // random bit patterns cover the full range including subnormals, NaN
and infinity
+ for _ in 0..20_000 {
+ values.push(f64::from_bits(rng.random::<u64>()));
+ }
+ // decimal-looking values, which land on or near rounding ties far
more often
+ for _ in 0..20_000 {
+ let unscaled = rng.random_range(-10_000_000_i64..10_000_000);
+ let scale = rng.random_range(0_i32..8);
+ values.push(unscaled as f64 / 10_f64.powi(scale));
+ }
+ for (precision, scale) in [(38_u8, 18_i8), (18, 6), (10, 2), (8, 6),
(4, 2), (38, 0)] {
+ let d = cast_f64_to_decimal128(
+ values.iter().copied().map(Some).collect(),
+ precision,
+ scale,
+ EvalMode::Legacy,
+ );
+ for (i, &v) in values.iter().enumerate() {
+ let expected = float_to_decimal128(v, precision, scale);
+ let actual = (!d.is_null(i)).then(|| d.value(i));
+ assert_eq!(actual, expected, "value {v:?} at
({precision},{scale})");
+ }
+ }
+ }
+
+ #[test]
+ fn test_cast_float_to_decimal_round_up_overflow() {
+ // 99.995 rounds up to 100.00 in string form, which no longer fits
precision 4.
+ // Binary rounding of 99.99499999999999886... would incorrectly keep
99.99.
+ let d = cast_f64_to_decimal128(vec![Some(99.995)], 4, 2,
EvalMode::Legacy);
+ assert!(d.is_null(0));
+
+ let a: ArrayRef = Arc::new(Float64Array::from(vec![Some(99.995)]));
+ match cast_floating_point_to_decimal128::<Float64Type>(&a, 4, 2,
EvalMode::Ansi) {
Err(SparkError::NumericValueOutOfRange {
value,
precision,
scale,
}) => {
- assert_eq!(value, "NaN");
- assert_eq!(precision, 10);
+ assert_eq!(value, "99.995");
+ assert_eq!(precision, 4);
assert_eq!(scale, 2);
}
other => panic!("expected NumericValueOutOfRange, got {other:?}"),
}
-
- let a: ArrayRef =
Arc::new(Float64Array::from(vec![Some(f64::INFINITY)]));
- match cast_floating_point_to_decimal128::<Float64Type>(&a, 10, 2,
EvalMode::Ansi) {
- Err(SparkError::NumericValueOutOfRange { value, .. }) => {
- assert_eq!(value, "inf");
- }
- other => panic!("expected NumericValueOutOfRange, got {other:?}"),
- }
}
#[test]
diff --git a/native/spark-expr/src/conversion_funcs/string.rs
b/native/spark-expr/src/conversion_funcs/string.rs
index 091c6544d5..46495ed981 100644
--- a/native/spark-expr/src/conversion_funcs/string.rs
+++ b/native/spark-expr/src/conversion_funcs/string.rs
@@ -507,16 +507,33 @@ const POW10_I128: [i128; 39] = {
/// `10^exp`, or `None` when the exponent overflows an `i128` (exp >= 39).
#[inline]
-fn pow10_i128(exp: u32) -> Option<i128> {
+pub(crate) fn pow10_i128(exp: u32) -> Option<i128> {
POW10_I128.get(exp as usize).copied()
}
+/// Divide by a power of ten with HALF_UP rounding, matching
`BigDecimal.setScale`: a tie
+/// rounds away from zero.
+///
+/// `divisor` must be `10^n` for `n >= 1`, so `divisor / 2` is exact and a
zero remainder
+/// can never be mistaken for a tie.
+#[inline]
+pub(crate) fn div_round_half_up_i128(numerator: i128, divisor: i128) -> i128 {
+ debug_assert!(divisor >= 10);
+ let quotient = numerator / divisor;
+ let remainder = numerator % divisor;
+ if remainder.abs() >= divisor / 2 {
+ quotient + numerator.signum()
+ } else {
+ quotient
+ }
+}
+
/// Accumulate an ASCII-digit slice into an `i128`, returning `None` on
overflow.
///
/// The first 38 digits always fit (`i128::MAX` is ~1.7e38), so only the
digits past
/// them need the per-digit overflow checks.
#[inline]
-fn digits_to_i128(digits: &[u8]) -> Option<i128> {
+pub(crate) fn digits_to_i128(digits: &[u8]) -> Option<i128> {
let (head, tail) = digits.split_at(digits.len().min(38));
let mut value: i128 = 0;
for &d in head {
@@ -616,28 +633,10 @@ fn parse_string_to_decimal(input_str: &str, precision:
u8, scale: i8) -> SparkRe
return Ok(Some(0));
}
- // Bounded above, so pow10_i128 always returns Some.
+ // Bounded above, so pow10_i128 always returns Some. The adjustment is
at least 1
+ // here, so the divisor is a power of ten no smaller than 10.
let divisor = pow10_i128(abs_scale_adjustment).unwrap();
- let quotient_opt = mantissa.checked_div(divisor);
- // Check if divisor is 0
- if quotient_opt.is_none() {
- return Ok(None);
- }
- let quotient = quotient_opt.unwrap();
- let remainder = mantissa % divisor;
-
- // Round half up: if abs(remainder) >= divisor/2, round away from zero
- let half_divisor = divisor / 2;
- let rounded = if remainder.abs() >= half_divisor {
- if mantissa >= 0 {
- quotient + 1
- } else {
- quotient - 1
- }
- } else {
- quotient
- };
- Some(rounded)
+ Some(div_round_half_up_i128(mantissa, divisor))
};
match scaled_value {
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 eba1e4b152..78abf54c8d 100644
--- a/spark/src/main/scala/org/apache/comet/expressions/CometCast.scala
+++ b/spark/src/main/scala/org/apache/comet/expressions/CometCast.scala
@@ -55,6 +55,15 @@ object CometCast
.getConfString("spark.sql.legacy.castComplexTypesToString.enabled",
"false")
.toBoolean
+ // Spark rounds the shortest decimal string form of the value
(`Double.toString`) rather
+ // than its binary expansion, and Comet's native cast reproduces that.
`Double.toString`
+ // only emits the shortest round-trip form on JDK 19 and later
(JDK-4511638); older JDKs
+ // can emit an extra digit, so a value whose shortest form lands exactly on
a rounding
+ // tie at the target scale can round differently there.
+ private val floatToDecimalJdkNote: String =
+ "Rounding matches Spark on JDK 19 and later. On older JDKs, Spark's own
result may " +
+ "differ for values whose shortest decimal form falls exactly on a
rounding tie"
+
def supportedTypes: Seq[DataType] =
Seq(
DataTypes.BooleanType,
@@ -425,8 +434,7 @@ object CometCast
DataTypes.IntegerType | DataTypes.LongType | DataTypes.TimestampType =>
Compatible()
case _: DecimalType =>
- // https://github.com/apache/datafusion-comet/issues/1371
- Incompatible(Some("There can be rounding differences"))
+ Compatible(Some(floatToDecimalJdkNote))
case _ =>
unsupported(DataTypes.FloatType, toType)
}
@@ -436,8 +444,7 @@ object CometCast
DataTypes.IntegerType | DataTypes.LongType | DataTypes.TimestampType =>
Compatible()
case _: DecimalType =>
- // https://github.com/apache/datafusion-comet/issues/1371
- Incompatible(Some("There can be rounding differences"))
+ Compatible(Some(floatToDecimalJdkNote))
case _ => unsupported(DataTypes.DoubleType, toType)
}
diff --git
a/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4-spark3_5/q44/extended.txt
b/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4-spark3_5/q44/extended.txt
index eb162b967a..236d51688a 100644
---
a/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4-spark3_5/q44/extended.txt
+++
b/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4-spark3_5/q44/extended.txt
@@ -19,13 +19,13 @@ TakeOrderedAndProject
: : : +- CometFilter
: : : : +- Subquery
: : : : +-
CometColumnarToRow
- : : : : +-
CometHashAggregate [COMET-INFO: A native implementation of Cast is available.
Set spark.comet.expression.Cast.allowIncompatible=true to enable it. For more
information, refer to the Comet Compatibility Guide
(https://datafusion.apache.org/comet/user-guide/latest/compatibility/index.html)]
+ : : : : +-
CometHashAggregate
: : : : +-
CometExchange
: : : : +-
CometHashAggregate
: : : : +-
CometProject
: : : : +-
CometFilter
: : : :
+- CometNativeScan parquet spark_catalog.default.store_sales
- : : : +- CometHashAggregate
[COMET-INFO: A native implementation of Cast is available. Set
spark.comet.expression.Cast.allowIncompatible=true to enable it. For more
information, refer to the Comet Compatibility Guide
(https://datafusion.apache.org/comet/user-guide/latest/compatibility/index.html)]
+ : : : +- CometHashAggregate
: : : +- CometExchange
: : : +-
CometHashAggregate
: : : +- CometProject
@@ -44,7 +44,7 @@ TakeOrderedAndProject
: : +- CometSort
: : +- CometFilter
: : : +- ReusedSubquery
- : : +- CometHashAggregate
[COMET-INFO: A native implementation of Cast is available. Set
spark.comet.expression.Cast.allowIncompatible=true to enable it. For more
information, refer to the Comet Compatibility Guide
(https://datafusion.apache.org/comet/user-guide/latest/compatibility/index.html)]
+ : : +- CometHashAggregate
: : +- CometExchange
: : +-
CometHashAggregate
: : +- CometProject
diff --git
a/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4-spark4_0/q44/extended.txt
b/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4-spark4_0/q44/extended.txt
index 48071ad123..6595583fa5 100644
---
a/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4-spark4_0/q44/extended.txt
+++
b/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4-spark4_0/q44/extended.txt
@@ -21,13 +21,13 @@ CometColumnarToRow
: : : +- CometFilter
: : : : +- Subquery
: : : : +-
CometColumnarToRow
- : : : : +-
CometHashAggregate [COMET-INFO: A native implementation of Cast is available.
Set spark.comet.expression.Cast.allowIncompatible=true to enable it. For more
information, refer to the Comet Compatibility Guide
(https://datafusion.apache.org/comet/user-guide/latest/compatibility/index.html)]
+ : : : : +-
CometHashAggregate
: : : : +-
CometExchange
: : : : +-
CometHashAggregate
: : : :
+- CometProject
: : : :
+- CometFilter
: : : :
+- CometNativeScan parquet spark_catalog.default.store_sales
- : : : +-
CometHashAggregate [COMET-INFO: A native implementation of Cast is available.
Set spark.comet.expression.Cast.allowIncompatible=true to enable it. For more
information, refer to the Comet Compatibility Guide
(https://datafusion.apache.org/comet/user-guide/latest/compatibility/index.html)]
+ : : : +-
CometHashAggregate
: : : +- CometExchange
: : : +-
CometHashAggregate
: : : +-
CometProject
@@ -47,7 +47,7 @@ CometColumnarToRow
: : +- CometSort
: : +- CometFilter
: : : +-
ReusedSubquery
- : : +-
CometHashAggregate [COMET-INFO: A native implementation of Cast is available.
Set spark.comet.expression.Cast.allowIncompatible=true to enable it. For more
information, refer to the Comet Compatibility Guide
(https://datafusion.apache.org/comet/user-guide/latest/compatibility/index.html)]
+ : : +-
CometHashAggregate
: : +- CometExchange
: : +-
CometHashAggregate
: : +-
CometProject
diff --git
a/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4-spark4_0/q6/extended.txt
b/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4-spark4_0/q6/extended.txt
index a737cf153d..d452f76572 100644
---
a/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4-spark4_0/q6/extended.txt
+++
b/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4-spark4_0/q6/extended.txt
@@ -55,7 +55,7 @@ CometColumnarToRow
: +- CometNativeScan parquet
spark_catalog.default.item
+- CometBroadcastExchange
+- CometFilter
- +- CometHashAggregate [COMET-INFO: A
native implementation of Cast is available. Set
spark.comet.expression.Cast.allowIncompatible=true to enable it. For more
information, refer to the Comet Compatibility Guide
(https://datafusion.apache.org/comet/user-guide/latest/compatibility/index.html)]
+ +- CometHashAggregate
+- CometExchange
+- CometHashAggregate
+- CometProject
diff --git
a/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q13/extended.txt
b/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q13/extended.txt
index 443dd08d35..80dfee9629 100644
---
a/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q13/extended.txt
+++
b/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q13/extended.txt
@@ -1,5 +1,5 @@
CometColumnarToRow
-+- CometHashAggregate [COMET-INFO: A native implementation of Cast is
available. Set spark.comet.expression.Cast.allowIncompatible=true to enable it.
For more information, refer to the Comet Compatibility Guide
(https://datafusion.apache.org/comet/user-guide/latest/compatibility/index.html)]
++- CometHashAggregate
+- CometExchange
+- CometHashAggregate
+- CometProject
diff --git
a/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q26/extended.txt
b/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q26/extended.txt
index 5c41b12c08..764e2b9c71 100644
---
a/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q26/extended.txt
+++
b/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q26/extended.txt
@@ -1,6 +1,6 @@
CometColumnarToRow
+- CometTakeOrderedAndProject
- +- CometHashAggregate [COMET-INFO: A native implementation of Cast is
available. Set spark.comet.expression.Cast.allowIncompatible=true to enable it.
For more information, refer to the Comet Compatibility Guide
(https://datafusion.apache.org/comet/user-guide/latest/compatibility/index.html)]
+ +- CometHashAggregate
+- CometExchange
+- CometHashAggregate
+- CometProject
diff --git
a/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q27/extended.txt
b/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q27/extended.txt
index c6dfd3e7ef..7f2df4754b 100644
---
a/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q27/extended.txt
+++
b/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q27/extended.txt
@@ -1,6 +1,6 @@
CometColumnarToRow
+- CometTakeOrderedAndProject
- +- CometHashAggregate [COMET-INFO: A native implementation of Cast is
available. Set spark.comet.expression.Cast.allowIncompatible=true to enable it.
For more information, refer to the Comet Compatibility Guide
(https://datafusion.apache.org/comet/user-guide/latest/compatibility/index.html)]
+ +- CometHashAggregate
+- CometExchange
+- CometHashAggregate
+- CometExpand
diff --git
a/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q28/extended.txt
b/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q28/extended.txt
index bb5b1f95a2..b63d8a2b9d 100644
---
a/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q28/extended.txt
+++
b/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q28/extended.txt
@@ -4,7 +4,7 @@ CometColumnarToRow
: :- CometBroadcastNestedLoopJoin
: : :- CometBroadcastNestedLoopJoin
: : : :- CometBroadcastNestedLoopJoin
- : : : : :- CometHashAggregate [COMET-INFO: A native implementation of
Cast is available. Set spark.comet.expression.Cast.allowIncompatible=true to
enable it. For more information, refer to the Comet Compatibility Guide
(https://datafusion.apache.org/comet/user-guide/latest/compatibility/index.html)]
+ : : : : :- CometHashAggregate
: : : : : +- CometExchange
: : : : : +- CometHashAggregate
: : : : : +- CometHashAggregate
@@ -14,7 +14,7 @@ CometColumnarToRow
: : : : : +- CometFilter
: : : : : +- CometNativeScan parquet
spark_catalog.default.store_sales
: : : : +- CometBroadcastExchange
- : : : : +- CometHashAggregate [COMET-INFO: A native implementation
of Cast is available. Set spark.comet.expression.Cast.allowIncompatible=true to
enable it. For more information, refer to the Comet Compatibility Guide
(https://datafusion.apache.org/comet/user-guide/latest/compatibility/index.html)]
+ : : : : +- CometHashAggregate
: : : : +- CometExchange
: : : : +- CometHashAggregate
: : : : +- CometHashAggregate
@@ -24,7 +24,7 @@ CometColumnarToRow
: : : : +- CometFilter
: : : : +- CometNativeScan parquet
spark_catalog.default.store_sales
: : : +- CometBroadcastExchange
- : : : +- CometHashAggregate [COMET-INFO: A native implementation of
Cast is available. Set spark.comet.expression.Cast.allowIncompatible=true to
enable it. For more information, refer to the Comet Compatibility Guide
(https://datafusion.apache.org/comet/user-guide/latest/compatibility/index.html)]
+ : : : +- CometHashAggregate
: : : +- CometExchange
: : : +- CometHashAggregate
: : : +- CometHashAggregate
@@ -34,7 +34,7 @@ CometColumnarToRow
: : : +- CometFilter
: : : +- CometNativeScan parquet
spark_catalog.default.store_sales
: : +- CometBroadcastExchange
- : : +- CometHashAggregate [COMET-INFO: A native implementation of
Cast is available. Set spark.comet.expression.Cast.allowIncompatible=true to
enable it. For more information, refer to the Comet Compatibility Guide
(https://datafusion.apache.org/comet/user-guide/latest/compatibility/index.html)]
+ : : +- CometHashAggregate
: : +- CometExchange
: : +- CometHashAggregate
: : +- CometHashAggregate
@@ -44,7 +44,7 @@ CometColumnarToRow
: : +- CometFilter
: : +- CometNativeScan parquet
spark_catalog.default.store_sales
: +- CometBroadcastExchange
- : +- CometHashAggregate [COMET-INFO: A native implementation of Cast
is available. Set spark.comet.expression.Cast.allowIncompatible=true to enable
it. For more information, refer to the Comet Compatibility Guide
(https://datafusion.apache.org/comet/user-guide/latest/compatibility/index.html)]
+ : +- CometHashAggregate
: +- CometExchange
: +- CometHashAggregate
: +- CometHashAggregate
@@ -54,7 +54,7 @@ CometColumnarToRow
: +- CometFilter
: +- CometNativeScan parquet
spark_catalog.default.store_sales
+- CometBroadcastExchange
- +- CometHashAggregate [COMET-INFO: A native implementation of Cast is
available. Set spark.comet.expression.Cast.allowIncompatible=true to enable it.
For more information, refer to the Comet Compatibility Guide
(https://datafusion.apache.org/comet/user-guide/latest/compatibility/index.html)]
+ +- CometHashAggregate
+- CometExchange
+- CometHashAggregate
+- CometHashAggregate
diff --git
a/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q32/extended.txt
b/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q32/extended.txt
index e4e3cb9529..88b18a4e60 100644
---
a/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q32/extended.txt
+++
b/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q32/extended.txt
@@ -21,7 +21,7 @@ CometColumnarToRow
: : +- CometNativeScan parquet
spark_catalog.default.item
: +- CometBroadcastExchange
: +- CometFilter
- : +- CometHashAggregate [COMET-INFO: A native
implementation of Cast is available. Set
spark.comet.expression.Cast.allowIncompatible=true to enable it. For more
information, refer to the Comet Compatibility Guide
(https://datafusion.apache.org/comet/user-guide/latest/compatibility/index.html)]
+ : +- CometHashAggregate
: +- CometExchange
: +- CometHashAggregate
: +- CometProject
diff --git
a/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q44/extended.txt
b/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q44/extended.txt
index 9d32a5f0f1..d606397281 100644
---
a/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q44/extended.txt
+++
b/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q44/extended.txt
@@ -15,13 +15,13 @@ CometColumnarToRow
: : : +- CometFilter
: : : : +- Subquery
: : : : +- CometColumnarToRow
- : : : : +- CometHashAggregate
[COMET-INFO: A native implementation of Cast is available. Set
spark.comet.expression.Cast.allowIncompatible=true to enable it. For more
information, refer to the Comet Compatibility Guide
(https://datafusion.apache.org/comet/user-guide/latest/compatibility/index.html)]
+ : : : : +- CometHashAggregate
: : : : +- CometExchange
: : : : +- CometHashAggregate
: : : : +- CometProject
: : : : +- CometFilter
: : : : +-
CometNativeScan parquet spark_catalog.default.store_sales
- : : : +- CometHashAggregate [COMET-INFO:
A native implementation of Cast is available. Set
spark.comet.expression.Cast.allowIncompatible=true to enable it. For more
information, refer to the Comet Compatibility Guide
(https://datafusion.apache.org/comet/user-guide/latest/compatibility/index.html)]
+ : : : +- CometHashAggregate
: : : +- CometExchange
: : : +- CometHashAggregate
: : : +- CometProject
@@ -36,13 +36,13 @@ CometColumnarToRow
: : +- CometFilter
: : : +- Subquery
: : : +- CometColumnarToRow
- : : : +- CometHashAggregate
[COMET-INFO: A native implementation of Cast is available. Set
spark.comet.expression.Cast.allowIncompatible=true to enable it. For more
information, refer to the Comet Compatibility Guide
(https://datafusion.apache.org/comet/user-guide/latest/compatibility/index.html)]
+ : : : +- CometHashAggregate
: : : +- CometExchange
: : : +- CometHashAggregate
: : : +- CometProject
: : : +- CometFilter
: : : +-
CometNativeScan parquet spark_catalog.default.store_sales
- : : +- CometHashAggregate [COMET-INFO:
A native implementation of Cast is available. Set
spark.comet.expression.Cast.allowIncompatible=true to enable it. For more
information, refer to the Comet Compatibility Guide
(https://datafusion.apache.org/comet/user-guide/latest/compatibility/index.html)]
+ : : +- CometHashAggregate
: : +- CometExchange
: : +- CometHashAggregate
: : +- CometProject
diff --git
a/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q6/extended.txt
b/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q6/extended.txt
index e847952604..c9c8cfb975 100644
---
a/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q6/extended.txt
+++
b/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q6/extended.txt
@@ -53,7 +53,7 @@ CometColumnarToRow
: +- CometNativeScan parquet
spark_catalog.default.item
+- CometBroadcastExchange
+- CometFilter
- +- CometHashAggregate [COMET-INFO: A
native implementation of Cast is available. Set
spark.comet.expression.Cast.allowIncompatible=true to enable it. For more
information, refer to the Comet Compatibility Guide
(https://datafusion.apache.org/comet/user-guide/latest/compatibility/index.html)]
+ +- CometHashAggregate
+- CometExchange
+- CometHashAggregate
+- CometProject
diff --git
a/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q7/extended.txt
b/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q7/extended.txt
index 7b60b3e7c7..ff85af4ff4 100644
---
a/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q7/extended.txt
+++
b/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q7/extended.txt
@@ -1,6 +1,6 @@
CometColumnarToRow
+- CometTakeOrderedAndProject
- +- CometHashAggregate [COMET-INFO: A native implementation of Cast is
available. Set spark.comet.expression.Cast.allowIncompatible=true to enable it.
For more information, refer to the Comet Compatibility Guide
(https://datafusion.apache.org/comet/user-guide/latest/compatibility/index.html)]
+ +- CometHashAggregate
+- CometExchange
+- CometHashAggregate
+- CometProject
diff --git
a/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q85/extended.txt
b/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q85/extended.txt
index 5b3d8e7c4d..7ea084adf4 100644
---
a/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q85/extended.txt
+++
b/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q85/extended.txt
@@ -1,6 +1,6 @@
CometColumnarToRow
+- CometTakeOrderedAndProject
- +- CometHashAggregate [COMET-INFO: A native implementation of Cast is
available. Set spark.comet.expression.Cast.allowIncompatible=true to enable it.
For more information, refer to the Comet Compatibility Guide
(https://datafusion.apache.org/comet/user-guide/latest/compatibility/index.html)]
+ +- CometHashAggregate
+- CometExchange
+- CometHashAggregate
+- CometProject
diff --git
a/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q9/extended.txt
b/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q9/extended.txt
index 82ae15542c..53ba3252cf 100644
---
a/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q9/extended.txt
+++
b/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q9/extended.txt
@@ -2,7 +2,7 @@
: :- Subquery
: : +- CometColumnarToRow
: : +- CometProject
-: : +- CometHashAggregate [COMET-INFO: A native implementation of
Cast is available. Set spark.comet.expression.Cast.allowIncompatible=true to
enable it. For more information, refer to the Comet Compatibility Guide
(https://datafusion.apache.org/comet/user-guide/latest/compatibility/index.html)]
+: : +- CometHashAggregate
: : +- CometExchange
: : +- CometHashAggregate
: : +- CometProject
@@ -13,7 +13,7 @@
: :- Subquery
: : +- CometColumnarToRow
: : +- CometProject
-: : +- CometHashAggregate [COMET-INFO: A native implementation of
Cast is available. Set spark.comet.expression.Cast.allowIncompatible=true to
enable it. For more information, refer to the Comet Compatibility Guide
(https://datafusion.apache.org/comet/user-guide/latest/compatibility/index.html)]
+: : +- CometHashAggregate
: : +- CometExchange
: : +- CometHashAggregate
: : +- CometProject
@@ -24,7 +24,7 @@
: :- Subquery
: : +- CometColumnarToRow
: : +- CometProject
-: : +- CometHashAggregate [COMET-INFO: A native implementation of
Cast is available. Set spark.comet.expression.Cast.allowIncompatible=true to
enable it. For more information, refer to the Comet Compatibility Guide
(https://datafusion.apache.org/comet/user-guide/latest/compatibility/index.html)]
+: : +- CometHashAggregate
: : +- CometExchange
: : +- CometHashAggregate
: : +- CometProject
@@ -35,7 +35,7 @@
: :- Subquery
: : +- CometColumnarToRow
: : +- CometProject
-: : +- CometHashAggregate [COMET-INFO: A native implementation of
Cast is available. Set spark.comet.expression.Cast.allowIncompatible=true to
enable it. For more information, refer to the Comet Compatibility Guide
(https://datafusion.apache.org/comet/user-guide/latest/compatibility/index.html)]
+: : +- CometHashAggregate
: : +- CometExchange
: : +- CometHashAggregate
: : +- CometProject
@@ -46,7 +46,7 @@
: :- Subquery
: : +- CometColumnarToRow
: : +- CometProject
-: : +- CometHashAggregate [COMET-INFO: A native implementation of
Cast is available. Set spark.comet.expression.Cast.allowIncompatible=true to
enable it. For more information, refer to the Comet Compatibility Guide
(https://datafusion.apache.org/comet/user-guide/latest/compatibility/index.html)]
+: : +- CometHashAggregate
: : +- CometExchange
: : +- CometHashAggregate
: : +- CometProject
diff --git
a/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q92/extended.txt
b/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q92/extended.txt
index b00960a1b9..eb03ca3a22 100644
---
a/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q92/extended.txt
+++
b/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q92/extended.txt
@@ -21,7 +21,7 @@ CometColumnarToRow
: : +- CometNativeScan parquet
spark_catalog.default.item
: +- CometBroadcastExchange
: +- CometFilter
- : +- CometHashAggregate [COMET-INFO: A native
implementation of Cast is available. Set
spark.comet.expression.Cast.allowIncompatible=true to enable it. For more
information, refer to the Comet Compatibility Guide
(https://datafusion.apache.org/comet/user-guide/latest/compatibility/index.html)]
+ : +- CometHashAggregate
: +- CometExchange
: +- CometHashAggregate
: +- CometProject
diff --git
a/spark/src/test/resources/tpcds-plan-stability/approved-plans-v2_7-spark4_0/q6/extended.txt
b/spark/src/test/resources/tpcds-plan-stability/approved-plans-v2_7-spark4_0/q6/extended.txt
index a737cf153d..d452f76572 100644
---
a/spark/src/test/resources/tpcds-plan-stability/approved-plans-v2_7-spark4_0/q6/extended.txt
+++
b/spark/src/test/resources/tpcds-plan-stability/approved-plans-v2_7-spark4_0/q6/extended.txt
@@ -55,7 +55,7 @@ CometColumnarToRow
: +- CometNativeScan parquet
spark_catalog.default.item
+- CometBroadcastExchange
+- CometFilter
- +- CometHashAggregate [COMET-INFO: A
native implementation of Cast is available. Set
spark.comet.expression.Cast.allowIncompatible=true to enable it. For more
information, refer to the Comet Compatibility Guide
(https://datafusion.apache.org/comet/user-guide/latest/compatibility/index.html)]
+ +- CometHashAggregate
+- CometExchange
+- CometHashAggregate
+- CometProject
diff --git
a/spark/src/test/resources/tpcds-plan-stability/approved-plans-v2_7/q27a/extended.txt
b/spark/src/test/resources/tpcds-plan-stability/approved-plans-v2_7/q27a/extended.txt
index 97c22e816f..05101798b2 100644
---
a/spark/src/test/resources/tpcds-plan-stability/approved-plans-v2_7/q27a/extended.txt
+++
b/spark/src/test/resources/tpcds-plan-stability/approved-plans-v2_7/q27a/extended.txt
@@ -1,7 +1,7 @@
CometColumnarToRow
+- CometTakeOrderedAndProject
+- CometUnion
- :- CometHashAggregate [COMET-INFO: A native implementation of Cast is
available. Set spark.comet.expression.Cast.allowIncompatible=true to enable it.
For more information, refer to the Comet Compatibility Guide
(https://datafusion.apache.org/comet/user-guide/latest/compatibility/index.html)]
+ :- CometHashAggregate
: +- CometExchange
: +- CometHashAggregate
: +- CometProject
@@ -35,7 +35,7 @@ CometColumnarToRow
: +- CometProject
: +- CometFilter
: +- CometNativeScan parquet
spark_catalog.default.item
- :- CometHashAggregate [COMET-INFO: A native implementation of Cast is
available. Set spark.comet.expression.Cast.allowIncompatible=true to enable it.
For more information, refer to the Comet Compatibility Guide
(https://datafusion.apache.org/comet/user-guide/latest/compatibility/index.html)]
+ :- CometHashAggregate
: +- CometExchange
: +- CometHashAggregate
: +- CometProject
@@ -65,7 +65,7 @@ CometColumnarToRow
: +- CometProject
: +- CometFilter
: +- CometNativeScan parquet
spark_catalog.default.item
- +- CometHashAggregate [COMET-INFO: A native implementation of Cast is
available. Set spark.comet.expression.Cast.allowIncompatible=true to enable it.
For more information, refer to the Comet Compatibility Guide
(https://datafusion.apache.org/comet/user-guide/latest/compatibility/index.html)]
+ +- CometHashAggregate
+- CometExchange
+- CometHashAggregate
+- CometProject
diff --git
a/spark/src/test/resources/tpcds-plan-stability/approved-plans-v2_7/q6/extended.txt
b/spark/src/test/resources/tpcds-plan-stability/approved-plans-v2_7/q6/extended.txt
index e847952604..c9c8cfb975 100644
---
a/spark/src/test/resources/tpcds-plan-stability/approved-plans-v2_7/q6/extended.txt
+++
b/spark/src/test/resources/tpcds-plan-stability/approved-plans-v2_7/q6/extended.txt
@@ -53,7 +53,7 @@ CometColumnarToRow
: +- CometNativeScan parquet
spark_catalog.default.item
+- CometBroadcastExchange
+- CometFilter
- +- CometHashAggregate [COMET-INFO: A
native implementation of Cast is available. Set
spark.comet.expression.Cast.allowIncompatible=true to enable it. For more
information, refer to the Comet Compatibility Guide
(https://datafusion.apache.org/comet/user-guide/latest/compatibility/index.html)]
+ +- CometHashAggregate
+- CometExchange
+- CometHashAggregate
+- CometProject
diff --git a/spark/src/test/scala/org/apache/comet/CometCastSuite.scala
b/spark/src/test/scala/org/apache/comet/CometCastSuite.scala
index 4bcc918516..6e5b03ee35 100644
--- a/spark/src/test/scala/org/apache/comet/CometCastSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/CometCastSuite.scala
@@ -501,17 +501,29 @@ class CometCastSuite extends CometTestBase with
AdaptiveSparkPlanHelper {
castTest(generateFloats(), DataTypes.DoubleType)
}
- ignore("cast FloatType to DecimalType(10,2)") {
- // // https://github.com/apache/datafusion-comet/issues/1371
+ test("cast FloatType to DecimalType(10,2)") {
castTest(generateFloats(), DataTypes.createDecimalType(10, 2))
}
- test("cast FloatType to DecimalType(10,2) - allow incompat") {
- withSQLConf(CometConf.getExprAllowIncompatConfigKey(classOf[Cast]) ->
"true") {
- castTest(generateFloats(), DataTypes.createDecimalType(10, 2))
+ // Values whose shortest decimal form sits exactly on a rounding tie while
the binary
+ // value is just below it, so rounding the binary value disagrees with
Spark's
+ // Double.toString-based rounding
(https://github.com/apache/datafusion-comet/issues/1371)
+ private val decimalRoundingEdgeCases =
+ Seq(0.5153125d, -0.5153125d, 0.5203125d, 1.005d, -1.005d, 99.995d,
-99.995d, 0.999995d)
+
+ private def castRoundingEdgeCasesTest(input: DataFrame): Unit = {
+ Seq(
+ DataTypes.createDecimalType(10, 2),
+ DataTypes.createDecimalType(8, 6),
+ DataTypes.createDecimalType(4, 2)).foreach { toType =>
+ castTest(input, toType)
}
}
+ test("cast FloatType to DecimalType - rounding edge cases") {
+
castRoundingEdgeCasesTest(withNulls(decimalRoundingEdgeCases.map(_.toFloat)).toDF("a"))
+ }
+
test("cast FloatType to StringType") {
// https://github.com/apache/datafusion-comet/issues/312
val r = new Random(0)
@@ -566,15 +578,12 @@ class CometCastSuite extends CometTestBase with
AdaptiveSparkPlanHelper {
castTest(generateDoubles(), DataTypes.FloatType)
}
- ignore("cast DoubleType to DecimalType(10,2)") {
- // https://github.com/apache/datafusion-comet/issues/1371
+ test("cast DoubleType to DecimalType(10,2)") {
castTest(generateDoubles(), DataTypes.createDecimalType(10, 2))
}
- test("cast DoubleType to DecimalType(10,2) - allow incompat") {
- withSQLConf(CometConf.getExprAllowIncompatConfigKey(classOf[Cast]) ->
"true") {
- castTest(generateDoubles(), DataTypes.createDecimalType(10, 2))
- }
+ test("cast DoubleType to DecimalType - rounding edge cases") {
+ castRoundingEdgeCasesTest(withNulls(decimalRoundingEdgeCases).toDF("a"))
}
test("cast DoubleType to StringType") {
@@ -1807,16 +1816,25 @@ class CometCastSuite extends CometTestBase with
AdaptiveSparkPlanHelper {
}
test("cast MapType propagates Incompatible from inner value cast") {
- // Float → Decimal is Incompatible due to rounding (see canCastFromFloat).
+ // Negative-scale Decimal → String is Incompatible when
+ // spark.sql.legacy.allowNegativeScaleOfDecimal is disabled (see
canCastToString).
// The Map arm must propagate that Incompatible up rather than silently
// marking the whole Map → Map cast Compatible.
- assert(
- CometCast.isSupported(
- MapType(IntegerType, FloatType),
- MapType(IntegerType, DecimalType(10, 2)),
- None,
- CometEvalMode.LEGACY) ==
- Incompatible(Some("There can be rounding differences")))
+ // Note: DecimalType(7, -2) must be constructed while the config is
enabled, because
+ // the constructor itself checks the config and throws if negative scale
is disallowed.
+ var negScaleType: DecimalType = null
+ withSQLConf("spark.sql.legacy.allowNegativeScaleOfDecimal" -> "true") {
+ negScaleType = DecimalType(7, -2)
+ }
+ withSQLConf("spark.sql.legacy.allowNegativeScaleOfDecimal" -> "false") {
+ assert(
+ CometCast.isSupported(
+ MapType(IntegerType, negScaleType),
+ MapType(IntegerType, StringType),
+ None,
+ CometEvalMode.LEGACY) ==
+ Incompatible(Some(CometCast.negativeScaleDecimalToStringReason)))
+ }
}
test("cast MapType propagates Unsupported from nested value cast") {
diff --git
a/spark/src/test/scala/org/apache/comet/exec/CometAggregateSuite.scala
b/spark/src/test/scala/org/apache/comet/exec/CometAggregateSuite.scala
index 2a3e19f1fa..2aaf3eb829 100644
--- a/spark/src/test/scala/org/apache/comet/exec/CometAggregateSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/exec/CometAggregateSuite.scala
@@ -158,11 +158,9 @@ class CometAggregateSuite extends CometTestBase with
AdaptiveSparkPlanHelper {
withTable(tableName) {
val table = spark.read.parquet(filename).coalesce(1)
table.createOrReplaceTempView(tableName)
- // we fall back to Spark for avg on decimal due to the following issue
- // https://github.com/apache/datafusion-comet/issues/1371
- // once this is fixed, we should change this test to
- // checkSparkAnswerAndNumOfAggregates
- checkSparkAnswer(s"SELECT c1, avg(c7) FROM $tableName GROUP BY c1
ORDER BY c1")
+ checkSparkAnswerAndNumOfAggregates(
+ s"SELECT c1, avg(c7) FROM $tableName GROUP BY c1 ORDER BY c1",
+ 2)
}
}
}
diff --git
a/spark/src/test/scala/org/apache/spark/sql/benchmark/CometCastNumericToNumericBenchmark.scala
b/spark/src/test/scala/org/apache/spark/sql/benchmark/CometCastNumericToNumericBenchmark.scala
index a9ea19a0e9..2141b1bea0 100644
---
a/spark/src/test/scala/org/apache/spark/sql/benchmark/CometCastNumericToNumericBenchmark.scala
+++
b/spark/src/test/scala/org/apache/spark/sql/benchmark/CometCastNumericToNumericBenchmark.scala
@@ -81,6 +81,15 @@ object CometCastNumericToNumericBenchmark extends
CometBenchmarkBase {
("DECIMAL(10,2)", "c_decimal", "LONG"),
("DECIMAL(10,2)", "c_decimal", "DOUBLE"))
+ // Float and double to decimal. The native cast rounds the binary product
when that provably
+ // agrees with Spark and otherwise formats the shortest decimal string, so
low-scale targets
+ // exercise the arithmetic path and high-scale targets exercise the string
path.
+ private val floatToDecimalPairs = Seq(
+ ("FLOAT", "c_float", "DECIMAL(10,2)"),
+ ("FLOAT", "c_float", "DECIMAL(38,18)"),
+ ("DOUBLE", "c_double", "DECIMAL(10,2)"),
+ ("DOUBLE", "c_double", "DECIMAL(38,18)"))
+
private def generateConfigs(
pairs: Seq[(String, String, String)]): Seq[CastNumericToNumericConfig] =
{
for {
@@ -137,7 +146,8 @@ object CometCastNumericToNumericBenchmark extends
CometBenchmarkBase {
generateConfigs(floatPairs) ++
generateConfigs(intToFloatPairs) ++
generateConfigs(floatToIntPairs) ++
- generateConfigs(decimalPairs)).foreach { config =>
+ generateConfigs(decimalPairs) ++
+ generateConfigs(floatToDecimalPairs)).foreach { config =>
runExpressionBenchmark(config.name, v, config.query,
config.extraCometConfigs)
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
