From ee6729b6cc830870764577fc74f5aedfe345a403 Mon Sep 17 00:00:00 2001 From: mazhengxuan Date: Mon, 24 Aug 2026 21:22:33 +0800 Subject: [PATCH 1/2] [AURON #2474] Fix make_date null and ANSI semantics Signed-off-by: mazhengxuan --- .../datafusion-ext-functions/src/lib.rs | 1 + .../src/spark_dates.rs | 131 ++++++++++++++++++ .../apache/spark/sql/auron/ShimsImpl.scala | 9 ++ .../org/apache/auron/AuronFunctionSuite.scala | 43 ++++-- .../spark/sql/auron/NativeConverters.scala | 6 +- .../org/apache/spark/sql/auron/Shims.scala | 2 + 6 files changed, 178 insertions(+), 14 deletions(-) diff --git a/native-engine/datafusion-ext-functions/src/lib.rs b/native-engine/datafusion-ext-functions/src/lib.rs index 73bdf82be..1c2170361 100644 --- a/native-engine/datafusion-ext-functions/src/lib.rs +++ b/native-engine/datafusion-ext-functions/src/lib.rs @@ -94,6 +94,7 @@ pub fn create_auron_ext_function( "Spark_DayOfWeek" => shared_function!(spark_dates::spark_dayofweek), "Spark_WeekOfYear" => shared_function!(spark_dates::spark_weekofyear), "Spark_Quarter" => shared_function!(spark_dates::spark_quarter), + "Spark_MakeDate" => shared_function!(spark_dates::spark_make_date), "Spark_Hour" => shared_function!(spark_dates::spark_hour), "Spark_Minute" => shared_function!(spark_dates::spark_minute), "Spark_Second" => shared_function!(spark_dates::spark_second), diff --git a/native-engine/datafusion-ext-functions/src/spark_dates.rs b/native-engine/datafusion-ext-functions/src/spark_dates.rs index 3bc830d87..da3235264 100644 --- a/native-engine/datafusion-ext-functions/src/spark_dates.rs +++ b/native-engine/datafusion-ext-functions/src/spark_dates.rs @@ -276,6 +276,76 @@ pub fn spark_day(args: &[ColumnarValue]) -> Result { )?)) } +pub fn spark_make_date(args: &[ColumnarValue]) -> Result { + if args.len() != 4 { + return Err(DataFusionError::Execution( + "spark_make_date() requires four arguments".to_string(), + )); + } + + let fail_on_error = match &args[3] { + ColumnarValue::Scalar(ScalarValue::Boolean(Some(value))) => *value, + _ => { + return Err(DataFusionError::Execution( + "spark_make_date() failOnError must be a boolean scalar".to_string(), + )); + } + }; + let scalar_result = args[..3] + .iter() + .all(|arg| matches!(arg, ColumnarValue::Scalar(_))); + let arrays = ColumnarValue::values_to_arrays(&args[..3])?; + let years = arrays[0] + .as_any() + .downcast_ref::() + .expect("make_date year must be Int32"); + let months = arrays[1] + .as_any() + .downcast_ref::() + .expect("make_date month must be Int32"); + let days = arrays[2] + .as_any() + .downcast_ref::() + .expect("make_date day must be Int32"); + let epoch = NaiveDate::from_ymd_opt(1970, 1, 1).expect("1970-01-01 must be a valid date"); + let mut result = Vec::with_capacity(years.len()); + + for ((year, month), day) in years.iter().zip(months.iter()).zip(days.iter()) { + match (year, month, day) { + (Some(year), Some(month), Some(day)) => { + let date = u32::try_from(month) + .ok() + .zip(u32::try_from(day).ok()) + .and_then(|(month, day)| NaiveDate::from_ymd_opt(year, month, day)); + + match date { + Some(date) => { + result.push(Some(date.signed_duration_since(epoch).num_days() as i32)); + } + None if !fail_on_error => result.push(None), + None => { + return Err(DataFusionError::Execution(format!( + "Invalid value for make_date: {year}-{month}-{day}" + ))); + } + } + } + _ => { + result.push(None); + } + } + } + + let result: ArrayRef = Arc::new(Date32Array::from(result)); + if scalar_result { + Ok(ColumnarValue::Scalar(ScalarValue::try_from_array( + &result, 0, + )?)) + } else { + Ok(ColumnarValue::Array(result)) + } +} + /// Spark `dayofweek()`: Sunday = 1, Monday = 2, ..., Saturday = 7. pub fn spark_dayofweek(args: &[ColumnarValue]) -> Result { let input = resolve_local_date32(args)?; @@ -519,6 +589,67 @@ mod tests { Ok(()) } + #[test] + fn test_spark_make_date_null_and_invalid_inputs() -> Result<()> { + let result = spark_make_date(&[ + ColumnarValue::Array(Arc::new(Int32Array::from(vec![ + Some(2025), + Some(2024), + Some(2024), + None, + Some(2024), + ]))), + ColumnarValue::Array(Arc::new(Int32Array::from(vec![ + Some(3), + None, + Some(2), + Some(7), + Some(7), + ]))), + ColumnarValue::Array(Arc::new(Int32Array::from(vec![ + Some(1), + Some(2), + Some(30), + Some(15), + None, + ]))), + ColumnarValue::Scalar(ScalarValue::Boolean(Some(false))), + ])? + .into_array(5)?; + let expected: ArrayRef = + Arc::new(Date32Array::from(vec![Some(20148), None, None, None, None])); + + assert_eq!(&result, &expected); + Ok(()) + } + + #[test] + fn test_spark_make_date_scalar_and_ansi_error() -> Result<()> { + let null_result = spark_make_date(&[ + ColumnarValue::Scalar(ScalarValue::Int32(None)), + ColumnarValue::Scalar(ScalarValue::Int32(Some(7))), + ColumnarValue::Scalar(ScalarValue::Int32(Some(15))), + ColumnarValue::Scalar(ScalarValue::Boolean(Some(true))), + ])?; + assert!(matches!( + null_result, + ColumnarValue::Scalar(ScalarValue::Date32(None)) + )); + + let error = spark_make_date(&[ + ColumnarValue::Scalar(ScalarValue::Int32(Some(2024))), + ColumnarValue::Scalar(ScalarValue::Int32(Some(13))), + ColumnarValue::Scalar(ScalarValue::Int32(Some(1))), + ColumnarValue::Scalar(ScalarValue::Boolean(Some(true))), + ]); + assert!(matches!( + error, + Err(DataFusionError::Execution(message)) + if message.contains("Invalid value for make_date") + )); + Ok(()) + } + #[test] fn test_spark_dayofweek() -> Result<()> { let input = Arc::new(Date32Array::from(vec![ diff --git a/spark-extension-shims-spark/src/main/scala/org/apache/spark/sql/auron/ShimsImpl.scala b/spark-extension-shims-spark/src/main/scala/org/apache/spark/sql/auron/ShimsImpl.scala index 2b60ba1d4..1da0b33c5 100644 --- a/spark-extension-shims-spark/src/main/scala/org/apache/spark/sql/auron/ShimsImpl.scala +++ b/spark-extension-shims-spark/src/main/scala/org/apache/spark/sql/auron/ShimsImpl.scala @@ -704,6 +704,15 @@ class ShimsImpl extends Shims with Logging { expr.asInstanceOf[Like].escapeChar } + @sparkver("3.0") + override def getMakeDateFailOnError(expr: Expression): Boolean = false + + @sparkver("3.1 / 3.2 / 3.3 / 3.4 / 3.5 / 4.0 / 4.1 / 4.2") + override def getMakeDateFailOnError(expr: Expression): Boolean = { + import org.apache.spark.sql.catalyst.expressions.MakeDate + expr.asInstanceOf[MakeDate].failOnError + } + override def convertMoreAggregateExpr(e: AggregateExpression): Option[pb.PhysicalExprNode] = { e.aggregateFunction match { case First(child, ignoresNull) => diff --git a/spark-extension-shims-spark/src/test/scala/org/apache/auron/AuronFunctionSuite.scala b/spark-extension-shims-spark/src/test/scala/org/apache/auron/AuronFunctionSuite.scala index 40f44d43f..f58e7240d 100644 --- a/spark-extension-shims-spark/src/test/scala/org/apache/auron/AuronFunctionSuite.scala +++ b/spark-extension-shims-spark/src/test/scala/org/apache/auron/AuronFunctionSuite.scala @@ -16,7 +16,6 @@ */ package org.apache.auron -import java.sql.Date import java.text.SimpleDateFormat import org.apache.spark.sql.{AuronQueryTest, Row} @@ -1135,19 +1134,37 @@ class AuronFunctionSuite extends AuronQueryTest with BaseAuronSQLSuite { } test("test function make_date") { - withTable("t1") { - sql( - "create table t1 using parquet as select '2025'" + - " as year, '03' as month, '01' as day") - val functions = - """ - |select - | make_date(year, month, day) - |from t1 - """.stripMargin + withSQLConf("spark.sql.ansi.enabled" -> "false") { + withTable("t1") { + sql("create table t1(year int, month int, day int) using parquet") + sql(""" + |insert into t1 values + | (2025, 3, 1), + | (2024, null, 2), + | (2024, 2, 30), + | (null, 7, 15), + | (2024, 7, null) + |""".stripMargin) + checkSparkAnswerAndOperator( + "select make_date(year, month, day), " + + "make_date(cast(null as int), month, day) from t1") + } + } + } - val df = sql(functions) - checkAnswer(df, Seq(Row(Date.valueOf("2025-03-01")))) + test("make_date fails for invalid input in ANSI mode") { + withSQLConf("spark.sql.ansi.enabled" -> "true") { + withTable("t1") { + sql("create table t1(year int, month int, day int) using parquet") + sql("insert into t1 values (2024, 13, 1)") + val df = sql("select make_date(year, month, day) from t1") + + val err = intercept[Exception] { + df.collect() + } + assertPlanIsNative(df) + assert(allCauseMessages(err).toLowerCase.contains("invalid value for make_date")) + } } } diff --git a/spark-extension/src/main/scala/org/apache/spark/sql/auron/NativeConverters.scala b/spark-extension/src/main/scala/org/apache/spark/sql/auron/NativeConverters.scala index 22e9db60e..10e4a836c 100644 --- a/spark-extension/src/main/scala/org/apache/spark/sql/auron/NativeConverters.scala +++ b/spark-extension/src/main/scala/org/apache/spark/sql/auron/NativeConverters.scala @@ -836,7 +836,11 @@ object NativeConverters extends Logging { case e: Acosh => buildScalarFunction(pb.ScalarFunction.Acosh, e.children, e.dataType) case e: Atan => buildScalarFunction(pb.ScalarFunction.Atan, e.children, e.dataType) case e: Exp => buildScalarFunction(pb.ScalarFunction.Exp, e.children, e.dataType) - case e: MakeDate => buildScalarFunction(pb.ScalarFunction.MakeDate, e.children, e.dataType) + case e: MakeDate => + buildExtScalarFunction( + "Spark_MakeDate", + e.children :+ Literal(Shims.get.getMakeDateFailOnError(e)), + e.dataType) case e: Log => buildScalarFunction(pb.ScalarFunction.Ln, e.children.map(nullIfNegative), e.dataType) case e: Log2 => diff --git a/spark-extension/src/main/scala/org/apache/spark/sql/auron/Shims.scala b/spark-extension/src/main/scala/org/apache/spark/sql/auron/Shims.scala index a3853f52d..6a31ea3d1 100644 --- a/spark-extension/src/main/scala/org/apache/spark/sql/auron/Shims.scala +++ b/spark-extension/src/main/scala/org/apache/spark/sql/auron/Shims.scala @@ -248,6 +248,8 @@ abstract class Shims { def getLikeEscapeChar(expr: Expression): Char + def getMakeDateFailOnError(expr: Expression): Boolean + def getAggregateExpressionFilter(expr: Expression): Option[Expression] def createFileSegment(file: File, offset: Long, length: Long, numRecords: Long): FileSegment From e0b645bfe86034dce0d6f78e038e5a8ac8acbfa9 Mon Sep 17 00:00:00 2001 From: mazhengxuan Date: Tue, 25 Aug 2026 11:16:32 +0800 Subject: [PATCH 2/2] [AURON #2474] Address make_date review comments Signed-off-by: mazhengxuan --- .../src/spark_dates.rs | 31 ++++++++++++++++--- 1 file changed, 26 insertions(+), 5 deletions(-) diff --git a/native-engine/datafusion-ext-functions/src/spark_dates.rs b/native-engine/datafusion-ext-functions/src/spark_dates.rs index da3235264..e4580baef 100644 --- a/native-engine/datafusion-ext-functions/src/spark_dates.rs +++ b/native-engine/datafusion-ext-functions/src/spark_dates.rs @@ -298,16 +298,21 @@ pub fn spark_make_date(args: &[ColumnarValue]) -> Result { let years = arrays[0] .as_any() .downcast_ref::() - .expect("make_date year must be Int32"); + .ok_or_else(|| { + DataFusionError::Execution("spark_make_date() year must be Int32".to_string()) + })?; let months = arrays[1] .as_any() .downcast_ref::() - .expect("make_date month must be Int32"); + .ok_or_else(|| { + DataFusionError::Execution("spark_make_date() month must be Int32".to_string()) + })?; let days = arrays[2] .as_any() .downcast_ref::() - .expect("make_date day must be Int32"); - let epoch = NaiveDate::from_ymd_opt(1970, 1, 1).expect("1970-01-01 must be a valid date"); + .ok_or_else(|| { + DataFusionError::Execution("spark_make_date() day must be Int32".to_string()) + })?; let mut result = Vec::with_capacity(years.len()); for ((year, month), day) in years.iter().zip(months.iter()).zip(days.iter()) { @@ -320,7 +325,7 @@ pub fn spark_make_date(args: &[ColumnarValue]) -> Result { match date { Some(date) => { - result.push(Some(date.signed_duration_since(epoch).num_days() as i32)); + result.push(Some(date.to_epoch_days())); } None if !fail_on_error => result.push(None), None => { @@ -650,6 +655,22 @@ mod tests { Ok(()) } + #[test] + fn test_spark_make_date_rejects_non_int32_inputs() { + let error = spark_make_date(&[ + ColumnarValue::Scalar(ScalarValue::Int64(Some(2024))), + ColumnarValue::Scalar(ScalarValue::Int32(Some(1))), + ColumnarValue::Scalar(ScalarValue::Int32(Some(1))), + ColumnarValue::Scalar(ScalarValue::Boolean(Some(false))), + ]); + + assert!(matches!( + error, + Err(DataFusionError::Execution(message)) + if message.contains("year must be Int32") + )); + } + #[test] fn test_spark_dayofweek() -> Result<()> { let input = Arc::new(Date32Array::from(vec![