upmerge

andygrove · andygrove · commit a72d27742743 · 2026-02-12T06:59:40.000-07:00
diff --git a/.github/workflows/iceberg_spark_test.yml b/.github/workflows/iceberg_spark_test.yml
@@ -23,6 +23,8 @@ concurrency:
 
 on:
   push:
+    branches:
+      - main
     paths-ignore:
       - "doc/**"
       - "docs/**"
diff --git a/.github/workflows/miri.yml b/.github/workflows/miri.yml
@@ -19,6 +19,8 @@ name: Run Miri Safety Checks
 
 on:
   push:
+    branches:
+      - main
     paths-ignore:
       - "doc/**"
       - "docs/**"
diff --git a/.github/workflows/pr_benchmark_check.yml b/.github/workflows/pr_benchmark_check.yml
@@ -26,6 +26,8 @@ concurrency:
 
 on:
   push:
+    branches:
+      - main
     paths:
       - "native/core/benches/**"
       - "native/spark-expr/benches/**"
diff --git a/.github/workflows/pr_build_linux.yml b/.github/workflows/pr_build_linux.yml
@@ -23,6 +23,8 @@ concurrency:
 
 on:
   push:
+    branches:
+      - main
     paths-ignore:
       - "doc/**"
       - "docs/**"
diff --git a/.github/workflows/pr_build_macos.yml b/.github/workflows/pr_build_macos.yml
@@ -23,6 +23,8 @@ concurrency:
 
 on:
   push:
+    branches:
+      - main
     paths-ignore:
       - "doc/**"
       - "docs/**"
diff --git a/.github/workflows/spark_sql_test.yml b/.github/workflows/spark_sql_test.yml
@@ -23,6 +23,8 @@ concurrency:
 
 on:
   push:
+    branches:
+      - main
     paths-ignore:
       - "doc/**"
       - "docs/**"
diff --git a/.github/workflows/stale.yml b/.github/workflows/stale.yml
@@ -0,0 +1,38 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+name: "Close stale PRs"
+on:
+  schedule:
+    - cron: "30 1 * * *"
+
+jobs:
+  close-stale-prs:
+    runs-on: ubuntu-latest
+    permissions:
+      issues: write
+      pull-requests: write
+    steps:
+      - uses: actions/stale@997185467fa4f803885201cee163a9f38240193d  # v10.1.1
+        with:
+          stale-pr-message: "Thank you for your contribution. Unfortunately, this pull request is stale because it has been open 60 days with no activity. Please remove the stale label or comment or this will be closed in 7 days."
+          days-before-pr-stale: 60
+          days-before-pr-close: 7
+          # do not close stale issues
+          days-before-issue-stale: -1
+          days-before-issue-close: -1
+          repo-token: ${{ secrets.GITHUB_TOKEN }}
diff --git a/native/spark-expr/src/conversion_funcs/cast.rs b/native/spark-expr/src/conversion_funcs/cast.rs
@@ -16,11 +16,12 @@
 // under the License.
 
 use crate::utils::array_with_timezone;
+use crate::EvalMode::Legacy;
 use crate::{timezone, BinaryOutputStyle};
 use crate::{EvalMode, SparkError, SparkResult};
 use arrow::array::builder::StringBuilder;
 use arrow::array::{
-    BooleanBuilder, Decimal128Builder, DictionaryArray, GenericByteArray, ListArray, MapArray,
+    BinaryBuilder, BooleanBuilder, Decimal128Builder, DictionaryArray, GenericByteArray, ListArray, MapArray,
     PrimitiveBuilder, StringArray, StructArray, TimestampMicrosecondBuilder,
 };
 use arrow::compute::can_cast_types;
@@ -304,29 +305,32 @@ fn can_cast_from_timestamp(to_type: &DataType, _options: &SparkCastOptions) -> b
 
 fn can_cast_from_boolean(to_type: &DataType, _: &SparkCastOptions) -> bool {
     use DataType::*;
-    matches!(to_type, Int8 | Int16 | Int32 | Int64 | Float32 | Float64)
+    matches!(
+        to_type,
+        Int8 | Int16 | Int32 | Int64 | Float32 | Float64 | Decimal128(_, _)
+    )
 }
 
 fn can_cast_from_byte(to_type: &DataType, _: &SparkCastOptions) -> bool {
     use DataType::*;
     matches!(
         to_type,
-        Boolean | Int8 | Int16 | Int32 | Int64 | Float32 | Float64 | Decimal128(_, _)
+        Boolean | Int8 | Int16 | Int32 | Int64 | Float32 | Float64 | Decimal128(_, _) | Binary
     )
 }
 
 fn can_cast_from_short(to_type: &DataType, _: &SparkCastOptions) -> bool {
     use DataType::*;
     matches!(
         to_type,
-        Boolean | Int8 | Int16 | Int32 | Int64 | Float32 | Float64 | Decimal128(_, _)
+        Boolean | Int8 | Int16 | Int32 | Int64 | Float32 | Float64 | Decimal128(_, _) | Binary
     )
 }
 
 fn can_cast_from_int(to_type: &DataType, options: &SparkCastOptions) -> bool {
     use DataType::*;
     match to_type {
-        Boolean | Int8 | Int16 | Int32 | Int64 | Float32 | Float64 | Utf8 => true,
+        Boolean | Int8 | Int16 | Int32 | Int64 | Float32 | Float64 | Utf8 | Binary => true,
         Decimal128(_, _) => {
             // incompatible: no overflow check
             options.allow_incompat
@@ -338,7 +342,7 @@ fn can_cast_from_int(to_type: &DataType, options: &SparkCastOptions) -> bool {
 fn can_cast_from_long(to_type: &DataType, options: &SparkCastOptions) -> bool {
     use DataType::*;
     match to_type {
-        Boolean | Int8 | Int16 | Int32 | Int64 | Float32 | Float64 => true,
+        Boolean | Int8 | Int16 | Int32 | Int64 | Float32 | Float64 | Binary => true,
         Decimal128(_, _) => {
             // incompatible: no overflow check
             options.allow_incompat
@@ -501,6 +505,29 @@ macro_rules! cast_float_to_string {
     }};
 }
 
+// eval mode is not needed since all ints can be implemented in binary format
+macro_rules! cast_whole_num_to_binary {
+    ($array:expr, $primitive_type:ty, $byte_size:expr) => {{
+        let input_arr = $array
+            .as_any()
+            .downcast_ref::<$primitive_type>()
+            .ok_or_else(|| SparkError::Internal("Expected numeric array".to_string()))?;
+
+        let len = input_arr.len();
+        let mut builder = BinaryBuilder::with_capacity(len, len * $byte_size);
+
+        for i in 0..input_arr.len() {
+            if input_arr.is_null(i) {
+                builder.append_null();
+            } else {
+                builder.append_value(input_arr.value(i).to_be_bytes());
+            }
+        }
+
+        Ok(Arc::new(builder.finish()) as ArrayRef)
+    }};
+}
+
 macro_rules! cast_int_to_int_macro {
     (
         $array: expr,
@@ -1128,6 +1155,19 @@ fn cast_array(
         }
         (Binary, Utf8) => Ok(cast_binary_to_string::<i32>(&array, cast_options)?),
         (Date32, Timestamp(_, tz)) => Ok(cast_date_to_timestamp(&array, cast_options, tz)?),
+        (Int8, Binary) if (eval_mode == Legacy) => cast_whole_num_to_binary!(&array, Int8Array, 1),
+        (Int16, Binary) if (eval_mode == Legacy) => {
+            cast_whole_num_to_binary!(&array, Int16Array, 2)
+        }
+        (Int32, Binary) if (eval_mode == Legacy) => {
+            cast_whole_num_to_binary!(&array, Int32Array, 4)
+        }
+        (Int64, Binary) if (eval_mode == Legacy) => {
+            cast_whole_num_to_binary!(&array, Int64Array, 8)
+        }
+        (Boolean, Decimal128(precision, scale)) => {
+            cast_boolean_to_decimal(&array, *precision, *scale)
+        }
         _ if cast_options.is_adapting_schema
             || is_datafusion_spark_compatible(&from_type, to_type) =>
         {
@@ -1202,6 +1242,16 @@ fn cast_date_to_timestamp(
     ))
 }
 
+fn cast_boolean_to_decimal(array: &ArrayRef, precision: u8, scale: i8) -> SparkResult<ArrayRef> {
+    let bool_array = array.as_boolean();
+    let scaled_val = 10_i128.pow(scale as u32);
+    let result: Decimal128Array = bool_array
+        .iter()
+        .map(|v| v.map(|b| if b { scaled_val } else { 0 }))
+        .collect();
+    Ok(Arc::new(result.with_precision_and_scale(precision, scale)?))
+}
+
 fn cast_string_to_float(
     array: &ArrayRef,
     to_type: &DataType,
diff --git a/spark/src/main/scala/org/apache/comet/expressions/CometCast.scala b/spark/src/main/scala/org/apache/comet/expressions/CometCast.scala
@@ -147,13 +147,13 @@ object CometCast extends CometExpressionSerde[Cast] with CometExprShim {
       case (DataTypes.BooleanType, _) =>
         canCastFromBoolean(toType)
       case (DataTypes.ByteType, _) =>
-        canCastFromByte(toType)
+        canCastFromByte(toType, evalMode)
       case (DataTypes.ShortType, _) =>
-        canCastFromShort(toType)
+        canCastFromShort(toType, evalMode)
       case (DataTypes.IntegerType, _) =>
-        canCastFromInt(toType)
+        canCastFromInt(toType, evalMode)
       case (DataTypes.LongType, _) =>
-        canCastFromLong(toType)
+        canCastFromLong(toType, evalMode)
       case (DataTypes.FloatType, _) =>
         canCastFromFloat(toType)
       case (DataTypes.DoubleType, _) =>
@@ -264,58 +264,68 @@ object CometCast extends CometExpressionSerde[Cast] with CometExprShim {
 
   private def canCastFromBoolean(toType: DataType): SupportLevel = toType match {
     case DataTypes.ByteType | DataTypes.ShortType | DataTypes.IntegerType | DataTypes.LongType |
-        DataTypes.FloatType | DataTypes.DoubleType =>
+        DataTypes.FloatType | DataTypes.DoubleType | _: DecimalType =>
       Compatible()
     case _ => unsupported(DataTypes.BooleanType, toType)
   }
 
-  private def canCastFromByte(toType: DataType): SupportLevel = toType match {
-    case DataTypes.BooleanType =>
-      Compatible()
-    case DataTypes.ShortType | DataTypes.IntegerType | DataTypes.LongType =>
-      Compatible()
-    case DataTypes.FloatType | DataTypes.DoubleType | _: DecimalType =>
-      Compatible()
-    case _ =>
-      unsupported(DataTypes.ByteType, toType)
-  }
+  private def canCastFromByte(toType: DataType, evalMode: CometEvalMode.Value): SupportLevel =
+    toType match {
+      case DataTypes.BooleanType =>
+        Compatible()
+      case DataTypes.ShortType | DataTypes.IntegerType | DataTypes.LongType =>
+        Compatible()
+      case DataTypes.FloatType | DataTypes.DoubleType | _: DecimalType =>
+        Compatible()
+      case DataTypes.BinaryType if (evalMode == CometEvalMode.LEGACY) =>
+        Compatible()
+      case _ =>
+        unsupported(DataTypes.ByteType, toType)
+    }
 
-  private def canCastFromShort(toType: DataType): SupportLevel = toType match {
-    case DataTypes.BooleanType =>
-      Compatible()
-    case DataTypes.ByteType | DataTypes.IntegerType | DataTypes.LongType =>
-      Compatible()
-    case DataTypes.FloatType | DataTypes.DoubleType | _: DecimalType =>
-      Compatible()
-    case _ =>
-      unsupported(DataTypes.ShortType, toType)
-  }
+  private def canCastFromShort(toType: DataType, evalMode: CometEvalMode.Value): SupportLevel =
+    toType match {
+      case DataTypes.BooleanType =>
+        Compatible()
+      case DataTypes.ByteType | DataTypes.IntegerType | DataTypes.LongType =>
+        Compatible()
+      case DataTypes.FloatType | DataTypes.DoubleType | _: DecimalType =>
+        Compatible()
+      case DataTypes.BinaryType if (evalMode == CometEvalMode.LEGACY) =>
+        Compatible()
+      case _ =>
+        unsupported(DataTypes.ShortType, toType)
+    }
 
-  private def canCastFromInt(toType: DataType): SupportLevel = toType match {
-    case DataTypes.BooleanType =>
-      Compatible()
-    case DataTypes.ByteType | DataTypes.ShortType | DataTypes.LongType =>
-      Compatible()
-    case DataTypes.FloatType | DataTypes.DoubleType =>
-      Compatible()
-    case _: DecimalType =>
-      Compatible()
-    case _ =>
-      unsupported(DataTypes.IntegerType, toType)
-  }
+  private def canCastFromInt(toType: DataType, evalMode: CometEvalMode.Value): SupportLevel =
+    toType match {
+      case DataTypes.BooleanType =>
+        Compatible()
+      case DataTypes.ByteType | DataTypes.ShortType | DataTypes.LongType =>
+        Compatible()
+      case DataTypes.FloatType | DataTypes.DoubleType =>
+        Compatible()
+      case _: DecimalType =>
+        Compatible()
+      case DataTypes.BinaryType if (evalMode == CometEvalMode.LEGACY) => Compatible()
+      case _ =>
+        unsupported(DataTypes.IntegerType, toType)
+    }
 
-  private def canCastFromLong(toType: DataType): SupportLevel = toType match {
-    case DataTypes.BooleanType =>
-      Compatible()
-    case DataTypes.ByteType | DataTypes.ShortType | DataTypes.IntegerType =>
-      Compatible()
-    case DataTypes.FloatType | DataTypes.DoubleType =>
-      Compatible()
-    case _: DecimalType =>
-      Compatible()
-    case _ =>
-      unsupported(DataTypes.LongType, toType)
-  }
+  private def canCastFromLong(toType: DataType, evalMode: CometEvalMode.Value): SupportLevel =
+    toType match {
+      case DataTypes.BooleanType =>
+        Compatible()
+      case DataTypes.ByteType | DataTypes.ShortType | DataTypes.IntegerType =>
+        Compatible()
+      case DataTypes.FloatType | DataTypes.DoubleType =>
+        Compatible()
+      case _: DecimalType =>
+        Compatible()
+      case DataTypes.BinaryType if (evalMode == CometEvalMode.LEGACY) => Compatible()
+      case _ =>
+        unsupported(DataTypes.LongType, toType)
+    }
 
   private def canCastFromFloat(toType: DataType): SupportLevel = toType match {
     case DataTypes.BooleanType | DataTypes.DoubleType | DataTypes.ByteType | DataTypes.ShortType |
diff --git a/spark/src/test/scala/org/apache/comet/CometCastSuite.scala b/spark/src/test/scala/org/apache/comet/CometCastSuite.scala
