1fanwang opened a new pull request, #7599: URL: https://github.com/apache/datafusion-comet/pull/7599
## Which issue does this PR close? Closes #6015. ## Rationale for this change In legacy (non-ANSI) mode, Spark negates the minimum of a signed integer type by wrapping it, so -(-2147483648) is -2147483648. In ANSI mode the same expression raises an overflow error. Comet already behaves this way when the value comes from a column. When the value is a scalar, such as a literal, legacy mode fails with "Overflow happened on: - -2147483648", so one expression gives two answers depending on how its input arrives. Spark checks for overflow in unary minus only in ANSI mode and otherwise negates and casts back, so the minimum wraps. See [Spark 4.1.3](https://github.com/apache/spark/blob/77bbf77e86ad48f58b5dfbc6ac882b3e70cf1989/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/arithmetic.scala#L62-L78), Comet's default profile, and [Spark 4.0.0](https://github.com/apache/spark/blob/fa33ea000a0bda9e5a3fa1af98e8e85b8cc5e4d4/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/arithmetic.scala#L60-L76). [Spark 3.5.1](https://github.com/apache/spark/blob/fd86f85e181fc2dc0f50a096855acf83a6cc5d9c/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/arithmetic.scala#L59-L88) differs only in how it reports ANSI byte and short overflow. ## What changes are included in this PR? In legacy mode, Int8, Int16, Int32 and Int64 scalars now use the same wrapping negation as arrays. ANSI mode and all other types are unchanged. New unit tests compare a scalar with a one element array for each minimum, with ordinary values, the value next to the minimum and typed nulls as controls. https://github.com/apache/datafusion-comet/pull/5982 has a [test](https://github.com/Smallfu666/datafusion-comet/blob/a712231719ee47dfb0b6ca2e76d65f6d3dec3cc9/native/spark-expr/src/math_funcs/negative.rs#L523-L550) that records legacy scalar negation of the minimum as an error and says it needs updating once this issue is fixed. I left it alone, so whichever pull request merges second needs that test changed to expect the wrapped value. ## How are these changes tested? ### Testing done The program projects -(MIN) for each signed integer type, once with the value read from a column and once as a literal, in legacy mode (fail_on_error=false) and in ANSI mode (fail_on_error=true). Its source is at the end of this section; save it as native/spark-expr/examples/negate_min.rs. Before, with the file from upstream main as it stands: ``` git checkout e59d93730f8eaa5ff908a7f75a379fc3cea83d14 -- native/spark-expr/src/math_funcs/negative.rs cd native && cargo run --locked -p datafusion-comet-spark-expr --example negate_min ``` In legacy mode the literal fails for all four types while the column wraps: ``` fail_on_error=false -(Int8(-128)) from a column: Int8(-128) fail_on_error=false -(Int8(-128)) from a literal: error: Arrow error: Arithmetic overflow: Overflow happened on: - -128 fail_on_error=false -(Int16(-32768)) from a column: Int16(-32768) fail_on_error=false -(Int16(-32768)) from a literal: error: Arrow error: Arithmetic overflow: Overflow happened on: - -32768 fail_on_error=false -(Int32(-2147483648)) from a column: Int32(-2147483648) fail_on_error=false -(Int32(-2147483648)) from a literal: error: Arrow error: Arithmetic overflow: Overflow happened on: - -2147483648 fail_on_error=false -(Int64(-9223372036854775808)) from a column: Int64(-9223372036854775808) fail_on_error=false -(Int64(-9223372036854775808)) from a literal: error: Arrow error: Arithmetic overflow: Overflow happened on: - -9223372036854775808 fail_on_error=true -(Int8(-128)) from a column: error: External error: [ARITHMETIC_OVERFLOW] byte overflow. If necessary set "spark.sql.ansi.enabled" to "false" to bypass this error. fail_on_error=true -(Int8(-128)) from a literal: error: External error: [ARITHMETIC_OVERFLOW] byte overflow. If necessary set "spark.sql.ansi.enabled" to "false" to bypass this error. fail_on_error=true -(Int16(-32768)) from a column: error: External error: [ARITHMETIC_OVERFLOW] short overflow. If necessary set "spark.sql.ansi.enabled" to "false" to bypass this error. fail_on_error=true -(Int16(-32768)) from a literal: error: External error: [ARITHMETIC_OVERFLOW] short overflow. If necessary set "spark.sql.ansi.enabled" to "false" to bypass this error. fail_on_error=true -(Int32(-2147483648)) from a column: error: External error: [ARITHMETIC_OVERFLOW] integer overflow. If necessary set "spark.sql.ansi.enabled" to "false" to bypass this error. fail_on_error=true -(Int32(-2147483648)) from a literal: error: External error: [ARITHMETIC_OVERFLOW] integer overflow. If necessary set "spark.sql.ansi.enabled" to "false" to bypass this error. fail_on_error=true -(Int64(-9223372036854775808)) from a column: error: External error: [ARITHMETIC_OVERFLOW] long overflow. If necessary set "spark.sql.ansi.enabled" to "false" to bypass this error. fail_on_error=true -(Int64(-9223372036854775808)) from a literal: error: External error: [ARITHMETIC_OVERFLOW] long overflow. If necessary set "spark.sql.ansi.enabled" to "false" to bypass this error. ``` After, with the file from this branch: ``` git checkout 18907ac5ab9715ff322eb3b00014a1244494555c -- native/spark-expr/src/math_funcs/negative.rs cd native && cargo run --locked -p datafusion-comet-spark-expr --example negate_min ``` The literal now wraps like the column, and the ANSI lines are the same as before: ``` fail_on_error=false -(Int8(-128)) from a column: Int8(-128) fail_on_error=false -(Int8(-128)) from a literal: Int8(-128) fail_on_error=false -(Int16(-32768)) from a column: Int16(-32768) fail_on_error=false -(Int16(-32768)) from a literal: Int16(-32768) fail_on_error=false -(Int32(-2147483648)) from a column: Int32(-2147483648) fail_on_error=false -(Int32(-2147483648)) from a literal: Int32(-2147483648) fail_on_error=false -(Int64(-9223372036854775808)) from a column: Int64(-9223372036854775808) fail_on_error=false -(Int64(-9223372036854775808)) from a literal: Int64(-9223372036854775808) fail_on_error=true -(Int8(-128)) from a column: error: External error: [ARITHMETIC_OVERFLOW] byte overflow. If necessary set "spark.sql.ansi.enabled" to "false" to bypass this error. fail_on_error=true -(Int8(-128)) from a literal: error: External error: [ARITHMETIC_OVERFLOW] byte overflow. If necessary set "spark.sql.ansi.enabled" to "false" to bypass this error. fail_on_error=true -(Int16(-32768)) from a column: error: External error: [ARITHMETIC_OVERFLOW] short overflow. If necessary set "spark.sql.ansi.enabled" to "false" to bypass this error. fail_on_error=true -(Int16(-32768)) from a literal: error: External error: [ARITHMETIC_OVERFLOW] short overflow. If necessary set "spark.sql.ansi.enabled" to "false" to bypass this error. fail_on_error=true -(Int32(-2147483648)) from a column: error: External error: [ARITHMETIC_OVERFLOW] integer overflow. If necessary set "spark.sql.ansi.enabled" to "false" to bypass this error. fail_on_error=true -(Int32(-2147483648)) from a literal: error: External error: [ARITHMETIC_OVERFLOW] integer overflow. If necessary set "spark.sql.ansi.enabled" to "false" to bypass this error. fail_on_error=true -(Int64(-9223372036854775808)) from a column: error: External error: [ARITHMETIC_OVERFLOW] long overflow. If necessary set "spark.sql.ansi.enabled" to "false" to bypass this error. fail_on_error=true -(Int64(-9223372036854775808)) from a literal: error: External error: [ARITHMETIC_OVERFLOW] long overflow. If necessary set "spark.sql.ansi.enabled" to "false" to bypass this error. ``` <details> <summary>Reproducer source: negate_min.rs</summary> ```rust use std::sync::Arc; use arrow::array::{ArrayRef, Int16Array, Int32Array, Int64Array, Int8Array, RecordBatch}; use arrow::datatypes::{Field, Schema}; use datafusion::common::ScalarValue; use datafusion::datasource::memory::MemorySourceConfig; use datafusion::execution::TaskContext; use datafusion::physical_expr::expressions::{Column, Literal}; use datafusion::physical_expr::PhysicalExpr; use datafusion::physical_plan::projection::ProjectionExec; use datafusion::physical_plan::{collect, ExecutionPlan}; use datafusion_comet_spark_expr::NegativeExpr; // Runs a plan that projects -(MIN) for each signed integer type, with the value read // from a column and given as a literal, in legacy mode and in ANSI mode. fn main() { let minimums: [(ScalarValue, ArrayRef); 4] = [ ( ScalarValue::Int8(Some(i8::MIN)), Arc::new(Int8Array::from(vec![i8::MIN])), ), ( ScalarValue::Int16(Some(i16::MIN)), Arc::new(Int16Array::from(vec![i16::MIN])), ), ( ScalarValue::Int32(Some(i32::MIN)), Arc::new(Int32Array::from(vec![i32::MIN])), ), ( ScalarValue::Int64(Some(i64::MIN)), Arc::new(Int64Array::from(vec![i64::MIN])), ), ]; let runtime = tokio::runtime::Runtime::new().unwrap(); for fail_on_error in [false, true] { for (minimum, array) in &minimums { let schema = Arc::new(Schema::new(vec![Field::new( "a", array.data_type().clone(), false, )])); let batch = RecordBatch::try_new(Arc::clone(&schema), vec![Arc::clone(array)]).unwrap(); let source: Arc<dyn ExecutionPlan> = MemorySourceConfig::try_new_exec(&[vec![batch]], schema, None).unwrap(); let arguments: [(&str, Arc<dyn PhysicalExpr>); 2] = [ ("column", Arc::new(Column::new("a", 0))), ("literal", Arc::new(Literal::new(minimum.clone()))), ]; for (kind, argument) in arguments { let negated: Arc<dyn PhysicalExpr> = Arc::new(NegativeExpr::new(argument, fail_on_error)); let plan = ProjectionExec::try_new( [(negated, "negated".to_string())], Arc::clone(&source), ) .unwrap(); let result = runtime.block_on(collect(Arc::new(plan), Arc::new(TaskContext::default()))); let shown = match result { Ok(batches) => { format!( "{:?}", ScalarValue::try_from_array(batches[0].column(0), 0).unwrap() ) } Err(e) => format!("error: {e}"), }; println!("fail_on_error={fail_on_error} -({minimum:?}) from a {kind}: {shown}"); } } } } ``` </details> This exercises the native expression through a DataFusion plan and does not run the Spark JVM suites. GitHub Actions is disabled on my fork, so the fork CI run did not happen and the upstream checks decide. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
