[AURON #2474] Fix make_date null and ANSI semantics

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),

native-engine/datafusion-ext-functions/src/spark_dates.rs

@@ -276,6 +276,81 @@ pub fn spark_day(args: &[ColumnarValue]) -> Result<ColumnarValue> {
     )?))
 }
 
+pub fn spark_make_date(args: &[ColumnarValue]) -> Result<ColumnarValue> {
+    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::<Int32Array>()
+        .ok_or_else(|| {
+            DataFusionError::Execution("spark_make_date() year must be Int32".to_string())
+        })?;
+    let months = arrays[1]
+        .as_any()
+        .downcast_ref::<Int32Array>()
+        .ok_or_else(|| {
+            DataFusionError::Execution("spark_make_date() month must be Int32".to_string())
+        })?;
+    let days = arrays[2]
+        .as_any()
+        .downcast_ref::<Int32Array>()
+        .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()) {
+        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.to_epoch_days()));
+                    }
+                    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<ColumnarValue> {
     let input = resolve_local_date32(args)?;
@@ -519,6 +594,83 @@ 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_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![

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) =>

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"))
+      }
     }
   }
 

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 =>

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