[GLUTEN-11355][UT] Add new Spark 4.1 tests

.github/workflows/clickhouse_be_trigger.yml

@@ -19,7 +19,6 @@ name: Clickhouse Backend Trigger
 on:
   pull_request_target:
     paths:
-      - '.github/workflows/clickhouse_be_trigger.yml'
       - 'pom.xml'
       - 'backends-clickhouse/**'
       - 'gluten-celeborn/**'
@@ -34,6 +33,10 @@ on:
       - 'gluten-ras/**'
       - 'gluten-ui/**'
       - 'gluten-kafka/**'
+      - '!shims/spark40/**'
+      - '!shims/spark41/**'
+      - '!gluten-ut/spark40/**'
+      - '!gluten-ut/spark41/**'
 
 jobs:
   add-comment:
@@ -63,3 +66,4 @@ jobs:
               issue_number: issueNumber,
               body: body
             });
+

.github/workflows/velox_backend_x86.yml

@@ -1508,7 +1508,7 @@ jobs:
         uses: actions/download-artifact@v4
         with:
           name: velox-native-lib-centos-7-${{github.sha}}
-          path: ./cpp/build/releases
+          path: ./cpp/build/
       - name: Download Arrow Jars
         uses: actions/download-artifact@v4
         with:
@@ -1565,7 +1565,7 @@ jobs:
         uses: actions/download-artifact@v4
         with:
           name: velox-native-lib-centos-7-${{github.sha}}
-          path: ./cpp/build/releases
+          path: ./cpp/build/
       - name: Download Arrow Jars
         uses: actions/download-artifact@v4
         with:

backends-velox/src/main/scala/org/apache/spark/sql/execution/ColumnarCachedBatchSerializer.scala

@@ -38,14 +38,55 @@ import org.apache.spark.sql.utils.SparkArrowUtil
 import org.apache.spark.sql.vectorized.ColumnarBatch
 import org.apache.spark.storage.StorageLevel
 
+import com.esotericsoftware.kryo.{Kryo, Serializer => KryoSerializer}
+import com.esotericsoftware.kryo.DefaultSerializer
+import com.esotericsoftware.kryo.io.{Input, Output}
 import org.apache.arrow.c.ArrowSchema
 
+/**
+ * TODO: fix on Spark-4.1 - Documentation
+ *
+ * If you encounter serialization issues, manually register this class:
+ * {{{
+ *   spark.kryo.classesToRegister=org.apache.spark.sql.execution.CachedColumnarBatch
+ * }}}
+ */
+@DefaultSerializer(classOf[CachedColumnarBatchKryoSerializer])
 case class CachedColumnarBatch(
     override val numRows: Int,
     override val sizeInBytes: Long,
     bytes: Array[Byte])
   extends CachedBatch {}
 
+class CachedColumnarBatchKryoSerializer extends KryoSerializer[CachedColumnarBatch] {
+  override def write(kryo: Kryo, output: Output, batch: CachedColumnarBatch): Unit = {
+    output.writeInt(batch.numRows)
+    output.writeLong(batch.sizeInBytes)
+    require(
+      batch.bytes != null,
+      "The object 'CachedColumnarBatch.bytes' is invalid or malformed to " +
+        s"serialize using ${this.getClass.getName}")
+    output.writeInt(batch.bytes.length + 1) // +1 to distinguish Kryo.NULL
+    output.writeBytes(batch.bytes)
+  }
+
+  override def read(
+      kryo: Kryo,
+      input: Input,
+      cls: Class[CachedColumnarBatch]): CachedColumnarBatch = {
+    val numRows = input.readInt()
+    val sizeInBytes = input.readLong()
+    val length = input.readInt()
+    require(
+      length != Kryo.NULL,
+      "The object 'CachedColumnarBatch.bytes' is invalid or malformed to " +
+        s"deserialize using ${this.getClass.getName}")
+    val bytes = new Array[Byte](length - 1) // -1 to restore
+    input.readBytes(bytes)
+    CachedColumnarBatch(numRows, sizeInBytes, bytes)
+  }
+}
+
 // format: off
 /**
  * Feature:

dev/buildbundle-veloxbe.sh

@@ -23,7 +23,10 @@ MVN_CMD="${BASEDIR}/../build/mvn"
 
 function build_for_spark {
   spark_version=$1
-  if [ $spark_version = "4.0" ]; then
+  # Extract major version (e.g., "3.2" -> "3", "4.0" -> "4")
+  major_version=$(echo $spark_version | cut -d'.' -f1)
+
+  if [ "$major_version" -ge 4 ]; then
     ${MVN_CMD} clean install -Pbackends-velox -Pspark-$spark_version -Pjava-17 -Pscala-2.13 -DskipTests
   else
     ${MVN_CMD} clean install -Pbackends-velox -Pspark-$spark_version -DskipTests
@@ -46,10 +49,14 @@ cd $GLUTEN_DIR
 check_supported
 
 # SPARK_VERSION is defined in builddeps-veloxbe.sh
+# SUPPORTED_SPARK_VERSIONS array is also defined in builddeps-veloxbe.sh
 if [ "$SPARK_VERSION" = "ALL" ]; then
-  for spark_version in 3.2 3.3 3.4 3.5 4.0
+  # Filter out "ALL" from the supported versions list
+  for spark_version in "${SUPPORTED_SPARK_VERSIONS[@]}"
   do
-    build_for_spark $spark_version
+    if [ "$spark_version" != "ALL" ]; then
+      build_for_spark $spark_version
+    fi
   done
 else
   build_for_spark $SPARK_VERSION

dev/builddeps-veloxbe.sh

@@ -140,15 +140,15 @@ do
         shift # Remove argument name from processing
         ;;
         --build_arrow=*)
-        BUILD_ARROW=("${arg#*=}")
+        BUILD_ARROW="${arg#*=}"
         shift # Remove argument name from processing
         ;;
         --num_threads=*)
-        NUM_THREADS=("${arg#*=}")
+        NUM_THREADS="${arg#*=}"
         shift # Remove argument name from processing
         ;;
         --spark_version=*)
-        SPARK_VERSION=("${arg#*=}")
+        SPARK_VERSION="${arg#*=}"
         shift # Remove argument name from processing
         ;;
 	      *)
@@ -193,13 +193,16 @@ if [ "$ENABLE_VCPKG" = "ON" ]; then
     source ./dev/vcpkg/env.sh ${BUILD_OPTIONS}
 fi
 
-if [ "$SPARK_VERSION" = "3.2" ] || [ "$SPARK_VERSION" = "3.3" ] \
-  || [ "$SPARK_VERSION" = "3.4" ] || [ "$SPARK_VERSION" = "3.5" ] \
-  || [ "$SPARK_VERSION" = "4.0" ] \
-  || [ "$SPARK_VERSION" = "ALL" ]; then
+# Supported Spark versions
+SUPPORTED_SPARK_VERSIONS=("3.2" "3.3" "3.4" "3.5" "4.0" "4.1" "ALL")
+
+# Check if SPARK_VERSION is in the supported list
+pattern=" $SPARK_VERSION "
+if [[ " ${SUPPORTED_SPARK_VERSIONS[*]} " =~ $pattern ]]; then
   echo "Building for Spark $SPARK_VERSION"
 else
   echo "Invalid Spark version: $SPARK_VERSION"
+  echo "Supported versions: ${SUPPORTED_SPARK_VERSIONS[*]}"
   exit 1
 fi
 

gluten-ut/spark40/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala

@@ -39,6 +39,7 @@ import org.apache.spark.sql.extension.{GlutenCollapseProjectExecTransformerSuite
 import org.apache.spark.sql.gluten.{GlutenFallbackStrategiesSuite, GlutenFallbackSuite}
 import org.apache.spark.sql.hive.execution.GlutenHiveSQLQuerySuite
 import org.apache.spark.sql.sources._
+import org.apache.spark.sql.streaming.GlutenStreamingQuerySuite
 
 // Some settings' line length exceeds 100
 // scalastyle:off line.size.limit

gluten-ut/spark40/src/test/scala/org/apache/spark/sql/execution/GlutenStreamingQuerySuite.scala → gluten-ut/spark40/src/test/scala/org/apache/spark/sql/streaming/GlutenStreamingQuerySuite.scala

@@ -14,12 +14,12 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.spark.sql.execution
+package org.apache.spark.sql.streaming
 
 import org.apache.spark.sql.GlutenSQLTestsTrait
+import org.apache.spark.sql.execution.ColumnarShuffleExchangeExec
 import org.apache.spark.sql.execution.exchange.REQUIRED_BY_STATEFUL_OPERATOR
 import org.apache.spark.sql.execution.streaming.MemoryStream
-import org.apache.spark.sql.streaming._
 
 class GlutenStreamingQuerySuite extends StreamingQuerySuite with GlutenSQLTestsTrait {
 

gluten-ut/spark41/src/test/resources/backends-velox/sql-tests/inputs/alias-resolution-edge-cases.sql

@@ -0,0 +1,17 @@
+CREATE TABLE t1(col1 INT, col2 STRING);
+CREATE TABLE t2(col1 STRUCT<a: STRING>, a STRING);
+
+-- Update references to match fixed-point plan when realiasing expressions in LCA resolution
+SELECT LEN(LOWER('X')) AS a, 1 AS b, b AS c GROUP BY LOWER('X') ORDER BY LOWER('X');
+SELECT LEN(LOWER('X')) AS a, 1 AS b, b AS c GROUP BY LOWER('X') HAVING LOWER('X') = 'x';
+
+-- LCA referencing nested field
+SELECT col1.field, field FROM VALUES(named_struct('field', 1));
+SELECT col1.field, field FROM VALUES(map('field', 1));
+
+-- LCA attributes shouldn't be counted in UNRESOLVED_GROUP_BY_ALL validation
+SELECT COUNT(col1) as alias, SUM(col1) + alias FROM t1 GROUP BY ALL;
+SELECT COUNT(col1) as alias, SUM(col1) + alias, SUM(col1) + col1 FROM t1 GROUP BY ALL;
+
+DROP TABLE t1;
+DROP TABLE t2;

gluten-ut/spark41/src/test/resources/backends-velox/sql-tests/inputs/array.sql

@@ -129,6 +129,14 @@ select get(array(1, 2, 3), 0);
 select get(array(1, 2, 3), 3);
 select get(array(1, 2, 3), null);
 select get(array(1, 2, 3), -1);
+select get(1, 0);
+select get(1, -1);
+select get('1', 0);
+select get('1', -1);
+select get(null, 0);
+select get(null, -1);
+select get(null, null);
+select get(CAST (null AS string), 0);
 
 -- function array_insert()
 select array_insert(array(1, 2, 3), 3, 4);

gluten-ut/spark41/src/test/resources/backends-velox/sql-tests/inputs/binary.sql

@@ -6,3 +6,9 @@ SELECT CAST('Spark' as BINARY);
 SELECT array( X'', X'4561736F6E2059616F20323031382D31312D31373A31333A33333A3333', CAST('Spark' as BINARY));
 SELECT to_csv(named_struct('n', 1, 'info', X'4561736F6E2059616F20323031382D31312D31373A31333A33333A3333'));
 select to_xml(named_struct('name', binary('Eason'), 'birth', 2018, 'org', binary('Kindergarten Cop')));
+SELECT from_xml(
+         to_xml(named_struct('name', binary('Eason'), 'birth', 2018, 'org', binary('Kindergarten Cop'))),
+         'name STRING, birth INT, org STRING');
+SELECT from_xml(
+         to_xml(named_struct('name', binary('Eason'), 'birth', 2018, 'org', binary('Kindergarten Cop'))),
+         'name binary, birth INT, org binary');

gluten-ut/spark41/src/test/resources/backends-velox/sql-tests/inputs/cast.sql

@@ -147,6 +147,27 @@ select cast(80.654321BD as interval hour to minute);
 select cast(-10.123456BD as interval year to month);
 select cast(10.654321BD as interval month);
 
+-- cast TIME to integral types
+SELECT CAST(TIME '00:01:52' AS tinyint);
+SELECT CAST(TIME '00:01:52' AS smallint);
+SELECT CAST(TIME '00:01:52' AS int);
+SELECT CAST(TIME '00:01:52' AS bigint);
+
+-- cast TIME to integral types with potential overflow
+SELECT CAST(TIME '23:59:59' AS tinyint);
+SELECT CAST(TIME '23:59:59' AS smallint);
+SELECT CAST(TIME '23:59:59' AS int);
+SELECT CAST(TIME '23:59:59' AS bigint);
+
+-- cast TIME with fractional seconds (should floor)
+SELECT CAST(TIME '00:00:17.5' AS tinyint);
+SELECT CAST(TIME '00:00:17.5' AS int);
+SELECT CAST(TIME '00:00:17.9' AS int);
+
+-- cast TIME edge cases
+SELECT CAST(TIME '00:00:00' AS tinyint);
+SELECT CAST(TIME '00:00:00' AS int);
+
 -- cast double colon syntax tests
 SELECT '1.23' :: int;
 SELECT 'abc' :: int;
@@ -168,6 +189,12 @@ select -10L :: interval second;
 select interval '08:11:10.001' hour to second :: decimal(10, 4);
 select 10.123456BD :: interval day to second;
 
+-- cast TIME using double colon syntax
+SELECT TIME '00:01:52' :: tinyint;
+SELECT TIME '00:01:52' :: int;
+SELECT TIME '23:59:59' :: tinyint;
+SELECT TIME '23:59:59' :: int;
+
 SELECT '1.23' :: int :: long;
 SELECT '2147483648' :: long :: int;
 SELECT CAST('2147483648' :: long AS int);
@@ -177,3 +204,26 @@ SELECT map(1, '123', 2, '456')[1] :: int;
 SELECT '2147483648' :: BINT;
 SELECT '2147483648' :: SELECT;
 SELECT FALSE IS NOT NULL :: string;
+
+-- SPARK-52619: cast TIME to DECIMAL with sufficient precision and scale.
+SELECT CAST(time '00:00:00' AS decimal(1, 0));
+SELECT CAST(time '12:00:00' AS decimal(7, 2));
+SELECT CAST(time '01:30:45' AS decimal(8, 3));
+SELECT CAST(time '23:59:59' AS decimal(9, 4));
+SELECT CAST(time '01:02:03' AS decimal(15, 9));
+SELECT CAST(time '10:20:30' AS decimal(20, 10));
+SELECT CAST(time '23:59:59.001' AS decimal(8, 3));
+SELECT CAST(time '23:59:59.999999' AS decimal(11, 6));
+SELECT CAST(time '23:59:59.999999999' AS decimal(14, 9));
+SELECT CAST(time '23:59:59.999999999' AS decimal(20, 10));
+
+-- SPARK-52619: cast TIME to DECIMAL with insufficient precision.
+SELECT CAST(time '00:01:00' AS decimal(1, 0));
+SELECT CAST(time '01:00:00' AS decimal(3, 0));
+SELECT CAST(time '10:00:00' AS decimal(5, 2));
+
+-- SPARK-52619: cast TIME to DECIMAL with insufficient scale.
+SELECT CAST(time '23:59:59.9' AS decimal(6, 0));
+SELECT CAST(time '23:59:59.999' AS decimal(8, 2));
+SELECT CAST(time '23:59:59.999999' AS decimal(11, 5));
+SELECT CAST(time '23:59:59.999999999' AS decimal(14, 8));

gluten-ut/spark41/src/test/resources/backends-velox/sql-tests/inputs/collations-aliases.sql

@@ -0,0 +1,95 @@
+-- test cases for implicit aliases to collated expression trees are correctly generated
+
+create table t1(s string, utf8_binary string collate utf8_binary, utf8_lcase string collate utf8_lcase) using parquet;
+insert into t1 values ('Spark', 'Spark', 'SQL');
+insert into t1 values ('aaAaAAaA', 'aaAaAAaA', 'aaAaAAaA');
+insert into t1 values ('aaAaAAaA', 'aaAaAAaA', 'aaAaaAaA');
+insert into t1 values ('aaAaAAaA', 'aaAaAAaA', 'aaAaaAaAaaAaaAaAaaAaaAaA');
+insert into t1 values ('bbAbaAbA', 'bbAbAAbA', 'a');
+insert into t1 values ('İo', 'İo', 'İo');
+insert into t1 values ('İo', 'İo', 'İo ');
+insert into t1 values ('İo', 'İo ', 'İo');
+insert into t1 values ('İo', 'İo', 'i̇o');
+insert into t1 values ('efd2', 'efd2', 'efd2');
+insert into t1 values ('Hello, world! Nice day.', 'Hello, world! Nice day.', 'Hello, world! Nice day.');
+insert into t1 values ('Something else. Nothing here.', 'Something else. Nothing here.', 'Something else. Nothing here.');
+insert into t1 values ('kitten', 'kitten', 'sitTing');
+insert into t1 values ('abc', 'abc', 'abc');
+insert into t1 values ('abcdcba', 'abcdcba', 'aBcDCbA');
+
+-- Simple select
+select concat_ws(' ', utf8_lcase, utf8_lcase) from t1;
+
+-- Select by implicit alias
+select `concat_ws(' ' collate UTF8_LCASE, utf8_lcase, utf8_lcase)` from (
+  select concat_ws(' ', utf8_lcase, utf8_lcase) from t1
+);
+
+-- Select by star
+select * from (
+  select concat_ws(' ', utf8_lcase, utf8_lcase) from t1
+);
+
+-- Select by qualified star
+select subq1.* from (
+  select concat_ws(' ', utf8_lcase, utf8_lcase) from t1
+) AS subq1;
+
+-- Implicit alias in CTE output
+with cte as (
+  select concat_ws(' ', utf8_lcase, utf8_lcase) from t1
+)
+select * from cte;
+
+-- Implicit alias in EXISTS subquery output
+select * from values (1) where exists (
+  select concat_ws(' ', utf8_lcase, utf8_lcase) from t1
+);
+
+-- Implicit alias in scalar subquery output
+select (
+  select concat_ws(' ', utf8_lcase, utf8_lcase) from t1 limit 1
+);
+
+-- Scalar subquery with CTE with implicit alias
+select (
+  with cte as (
+    select concat_ws(' ', utf8_lcase, utf8_lcase) from t1
+  )
+  select * from cte limit 1
+);
+
+-- Outer reference to implicit alias
+select * from (
+  select concat_ws(' ', utf8_lcase, utf8_lcase) from t1 limit 1
+)
+where (
+  `concat_ws(' ' collate UTF8_LCASE, utf8_lcase, utf8_lcase)` == 'aaa'
+);
+
+-- Implicit alias reference in Sort
+select lower(`concat_ws(' ' collate UTF8_LCASE, utf8_lcase, utf8_lcase)`) from (
+  select concat_ws(' ', utf8_lcase, utf8_lcase) from t1
+  group by 1
+  order by 1
+);
+
+-- Implciit alias from aggregate in Sort
+select lower(`concat_ws(' ' collate UTF8_LCASE, utf8_lcase, utf8_lcase)`) from (
+  select concat_ws(' ', utf8_lcase, utf8_lcase) from t1
+  group by 1
+  order by max(concat_ws(' ', utf8_lcase, utf8_lcase))
+);
+
+-- Implicit alias in view schema
+create temporary view v1 as (
+  select concat_ws(' ', utf8_lcase, utf8_lcase) from t1
+);
+
+select * from v1;
+
+select `concat_ws(' ' collate UTF8_LCASE, utf8_lcase, utf8_lcase)` from v1;
+
+drop view v1;
+
+drop table t1;