Merge branch 'main' into regex-transpiler-bugfixes

Signed-off-by: Igor Peshansky <ipeshansky@nvidia.com>

Author: igorpeshansky

.github/workflows/blossom-ci.yml

@@ -34,51 +34,52 @@ jobs:
 
     # This job only runs for pull request comments
     if: |
+      github.event.issue.pull_request &&
       github.event.comment.body == 'build' &&
       (
         github.actor == 'abellina' ||
+        github.actor == 'amahussein' ||
         github.actor == 'anfeng' ||
+        github.actor == 'binmahone' ||
+        github.actor == 'cindyyuanjiang' ||
         github.actor == 'firestarman' ||
         github.actor == 'GaryShen2008' ||
+        github.actor == 'gerashegalov' ||
+        github.actor == 'HaoYang670' ||
+        github.actor == 'hyperbolic2346' ||
+        github.actor == 'igorpeshansky' ||
+        github.actor == 'jihoonson' ||
+        github.actor == 'knoguchi22' ||
         github.actor == 'kuhushukla' ||
+        github.actor == 'liurenjie1024' ||
+        github.actor == 'mattahrens' ||
         github.actor == 'mythrocks' ||
         github.actor == 'nartal1' ||
+        github.actor == 'nvliyuan' ||
+        github.actor == 'NVnavkumar' ||
         github.actor == 'NvTimLiu' ||
-        github.actor == 'razajafri' ||
+        github.actor == 'parthosa' ||
+        github.actor == 'patilkishorv' ||
+        github.actor == 'pmattione-nvidia' ||
+        github.actor == 'pxLi' ||
+        github.actor == 'res-life' ||
         github.actor == 'revans2' ||
+        github.actor == 'rishic3' ||
         github.actor == 'sameerz' ||
+        github.actor == 'sdrp713' ||
+        github.actor == 'sperlingxx' ||
+        github.actor == 'SurajAralihalli' ||
         github.actor == 'tgravescs' ||
+        github.actor == 'thirtiseven' ||
+        github.actor == 'ttnghia' ||
+        github.actor == 'viadea' ||
         github.actor == 'wbo4958' ||
+        github.actor == 'winningsix' ||
         github.actor == 'wjxiz1992' ||
-        github.actor == 'sperlingxx' ||
-        github.actor == 'hyperbolic2346' ||
-        github.actor == 'gerashegalov' ||
-        github.actor == 'ttnghia' ||
-        github.actor == 'nvliyuan' ||
-        github.actor == 'res-life' ||
-        github.actor == 'HaoYang670' ||
-        github.actor == 'NVnavkumar' ||
-        github.actor == 'amahussein' ||
-        github.actor == 'mattahrens' ||
         github.actor == 'YanxuanLiu' ||
-        github.actor == 'cindyyuanjiang' ||
-        github.actor == 'thirtiseven' ||
-        github.actor == 'winningsix' ||
-        github.actor == 'viadea' ||
         github.actor == 'yinqingh' ||
-        github.actor == 'parthosa' ||
-        github.actor == 'liurenjie1024' ||
-        github.actor == 'binmahone' ||
         github.actor == 'zpuller' ||
-        github.actor == 'pxLi'  ||
-        github.actor == 'SurajAralihalli' ||
-        github.actor == 'jihoonson' ||
-        github.actor == 'knoguchi22' ||
-        github.actor == 'patilkishorv' ||
-        github.actor == 'pmattione-nvidia' ||
-        github.actor == 'rishic3' ||
-        github.actor == 'sdrp713' ||
-        github.actor == 'igorpeshansky'
+        false
       )
     steps:
       - name: Check if comment is issued by authorized person

datagen/src/main/spark400/scala/org/apache/spark/sql/tests/datagen/DataGenExprShims.scala

@@ -22,6 +22,7 @@
 {"spark": "403"}
 {"spark": "411"}
 {"spark": "412"}
+{"spark": "420"}
 spark-rapids-shim-json-lines ***/
 package org.apache.spark.sql.tests.datagen
 

docs/dev/adaptive-query.md

@@ -50,8 +50,8 @@ On startup, the `SQLExecPlugin` plugin registers two distinct sets of
 optimizer rules:
 
 ```scala
-extensions.injectColumnar(_ => ColumnarOverrideRules())
-extensions.injectQueryStagePrepRule(_ => GpuQueryStagePrepOverrides())
+extensions.injectColumnar(session => ColumnarOverrideRules(session))
+extensions.injectQueryStagePrepRule(session => GpuQueryStagePrepOverrides(session))
 ```
 
 The `ColumnarOverrideRules` are used whether AQE is enabled or not, and the

integration_tests/src/main/python/ast_test.py

@@ -17,8 +17,8 @@
 from asserts import assert_cpu_and_gpu_are_equal_collect_with_capture, assert_gpu_and_cpu_are_equal_collect
 from data_gen import *
 from marks import approximate_float, datagen_overrides, ignore_order, disable_ansi_mode
-from spark_session import with_cpu_session, is_spark_403, is_spark_412_or_later
-
+from spark_session import with_cpu_session, is_spark_403, is_spark_412_or_later, \
+    is_spark_420_or_later
 import pyspark.sql.functions as f
 
 # Each descriptor contains a list of data generators and a corresponding boolean
@@ -213,9 +213,10 @@ def test_acos(data_descr):
 def test_atan(data_descr):
     assert_unary_ast(data_descr, lambda df: df.selectExpr('atan(a)'))
 
-# AST is not expressive enough to support the ASINH Spark emulation expression
+# AST is not expressive enough to support the ASINH Spark emulation expression.
+# Spark 4.2 uses the improved ASINH path by default, which is AST-compatible.
 @approximate_float
-@pytest.mark.parametrize('data_descr', [(double_gen, False)], ids=idfn)
+@pytest.mark.parametrize('data_descr', [(double_gen, is_spark_420_or_later())], ids=idfn)
 def test_asinh(data_descr):
     assert_unary_ast(data_descr, lambda df: df.selectExpr('asinh(a)'))
 

integration_tests/src/main/python/date_time_test.py

@@ -760,7 +760,7 @@ def test_formats_for_legacy_mode_other_formats_tz_rules():
          'spark.rapids.sql.incompatibleDateFormats.enabled': True})
 
 @disable_ansi_mode
-@allow_non_gpu('ProjectExec')
+@allow_non_gpu('ProjectExec', 'GetTimestamp')
 def test_to_timestamp_legacy_millisecond_format_fallback():
     conf = {
         'spark.sql.legacy.timeParserPolicy': 'LEGACY',

integration_tests/src/main/python/explain_test.py

@@ -1,4 +1,4 @@
-# Copyright (c) 2021-2025, NVIDIA CORPORATION.
+# Copyright (c) 2021-2026, NVIDIA CORPORATION.
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -19,7 +19,7 @@
 from marks import *
 from pyspark.sql.functions import *
 from pyspark.sql.types import *
-from spark_session import with_cpu_session, with_gpu_session
+from spark_session import is_spark_420_or_later, with_cpu_session, with_gpu_session
 
 # mark this test as ci_1 for mvn verify sanity check in pre-merge CI
 pytestmark = pytest.mark.premerge_ci_1
@@ -67,11 +67,20 @@ def do_explain(spark):
         df = spark.createDataFrame([(1, "John Doe", 21)], ("id", "name", "age"))
         df2 = df.select(slen("name").alias("slen(name)"), to_upper("name"), add_one("age"))
         explain_str = spark.sparkContext._jvm.com.nvidia.spark.rapids.ExplainPlan.explainPotentialGpuPlan(df2._jdf, "ALL")
-        # udf shouldn't be on GPU
-        udf_str_not = 'cannot run on GPU because GPU does not currently support the operator class org.apache.spark.sql.execution.python.BatchEvalPythonExec'
-        assert udf_str_not in explain_str
+        # udf shouldn't be fully on GPU. Spark 4.2 enables Arrow-optimized Python
+        # UDFs here, so regular udf(...) is planned as ArrowEvalPythonExec instead
+        # of the older BatchEvalPythonExec.
+        udf_exec = 'ArrowEvalPythonExec' if is_spark_420_or_later() else 'BatchEvalPythonExec'
+        udf_str_not = \
+            'cannot run on GPU because GPU does not currently support the operator class ' + \
+            f'org.apache.spark.sql.execution.python.{udf_exec}'
+        if is_spark_420_or_later():
+            assert f'Exec <{udf_exec}> will partially run on GPU' in explain_str
+        else:
+            assert udf_str_not in explain_str
         not_on_gpu_str = spark.sparkContext._jvm.com.nvidia.spark.rapids.ExplainPlan.explainPotentialGpuPlan(df2._jdf, "NOT")
-        assert udf_str_not in not_on_gpu_str
+        if not is_spark_420_or_later():
+            assert udf_str_not in not_on_gpu_str
         assert "will run on GPU" not in not_on_gpu_str
 
     with_cpu_session(do_explain)

integration_tests/src/main/python/fastparquet_compatibility_test.py

@@ -245,7 +245,9 @@ def test_reading_file_written_with_gpu(spark_tmp_path, column_gen):
     try:
         # For now, this compares the results of reading back the GPU-written data, via fastparquet and GPU.
         assert_gpu_and_cpu_are_equal_collect(read_parquet(data_path=data_path, local_data_path=local_data_path),
-                                             conf=conf)
+                                             conf=conf,
+                                             result_canonicalize_func_before_compare=
+                                             get_fastparquet_result_canonicalizer())
     finally:
         # Clean up local copy of data.
         delete_local_directory(local_base_path)

integration_tests/src/main/python/fastparquet_utils.py

@@ -1,4 +1,4 @@
-# Copyright (c) 2023, NVIDIA CORPORATION.
+# Copyright (c) 2023-2026, NVIDIA CORPORATION.
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -12,9 +12,11 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import math
 import pytest
 
 from pyspark.sql import Row
+from spark_session import is_spark_420_or_later
 
 def get_fastparquet_result_canonicalizer():
     # Get a converter like fastparquet that flattens the GPU results.
@@ -35,9 +37,35 @@ def _convert_fastparquet_result(fastparquet_cpu, gpu):
     :param gpu: the gpu result, it's a Spark Row list
     :return: (fastparquet, converted_gpu) tuple
     """
-    new_gpu = [_convert_gpu_row(gpu_row) for gpu_row in gpu]
+    converted_gpu = [_convert_gpu_row(gpu_row) for gpu_row in gpu]
+    if not is_spark_420_or_later():
+        return (fastparquet_cpu, converted_gpu)
+    new_gpu = [
+        _normalize_gpu_nans_like_fastparquet(
+            fastparquet_cpu[idx] if idx < len(fastparquet_cpu) else None,
+            gpu_row)
+        for idx, gpu_row in enumerate(converted_gpu)]
     return (fastparquet_cpu, new_gpu)
 
+def _normalize_gpu_nans_like_fastparquet(cpu_value, gpu_value):
+    # Spark 4.2 converts pandas NaN values from fastparquet into Spark nulls.
+    # Normalize only those matching GPU NaN values so other null mismatches still fail.
+    if cpu_value is None:
+        return None if isinstance(gpu_value, float) and math.isnan(gpu_value) else gpu_value
+    if isinstance(cpu_value, Row) and isinstance(gpu_value, Row):
+        cpu_dict = cpu_value.asDict()
+        gpu_dict = gpu_value.asDict()
+        return Row(**{
+            key: _normalize_gpu_nans_like_fastparquet(cpu_dict.get(key), gpu_dict.get(key))
+            for key in gpu_value.__fields__})
+    if isinstance(cpu_value, list) and isinstance(gpu_value, list):
+        return [
+            _normalize_gpu_nans_like_fastparquet(
+                cpu_value[idx] if idx < len(cpu_value) else None,
+                gpu_item)
+            for idx, gpu_item in enumerate(gpu_value)]
+    return gpu_value
+
 def _convert_gpu_row(gpu_row):
     converted_dict = _get_converted_dict(gpu_row)
     return Row(**converted_dict)

integration_tests/src/main/python/hash_aggregate_test.py

@@ -26,7 +26,8 @@
 from pyspark.sql.types import *
 from marks import *
 import pyspark.sql.functions as f
-from spark_session import is_databricks104_or_later, with_cpu_session, is_spark_340_or_later
+from spark_session import is_databricks104_or_later, with_cpu_session, is_spark_340_or_later, \
+    is_spark_420_or_later
 
 pytestmark = pytest.mark.nightly_resource_consuming_test
 
@@ -862,6 +863,33 @@ def doit(spark):
         doit,
         conf={'spark.sql.execution.useObjectHashAggregateExec': str(use_obj_hash_agg).lower()})
 
+
+@pytest.mark.skipif(not is_spark_420_or_later(),
+                    reason='collect_list/array_agg RESPECT NULLS is introduced in Spark 4.2')
+@allow_non_gpu("ProjectExec")
+@ignore_order(local=True)
+@pytest.mark.parametrize('use_obj_hash_agg', [True, False], ids=idfn)
+def test_hash_groupby_collect_list_respect_nulls(use_obj_hash_agg):
+    def doit(spark):
+        return spark.sql("""
+            SELECT a,
+                   sort_array(collect_list(b) RESPECT NULLS) AS respect_list,
+                   sort_array(array_agg(b) RESPECT NULLS) AS respect_array
+            FROM VALUES
+                (1, 1),
+                (1, NULL),
+                (1, 3),
+                (2, NULL),
+                (2, 5)
+            AS tab(a, b)
+            GROUP BY a
+        """)
+
+    assert_gpu_and_cpu_are_equal_collect(
+        doit,
+        conf={'spark.sql.execution.useObjectHashAggregateExec': str(use_obj_hash_agg).lower()})
+
+
 @ignore_order(local=True)
 @pytest.mark.parametrize('use_obj_hash_agg', [True, False], ids=idfn)
 def test_hash_groupby_collect_list_of_maps(use_obj_hash_agg):

integration_tests/src/main/python/hybrid_parquet_test.py

@@ -29,6 +29,13 @@
 3. BinaryType is NOT supported
 4. MapType wrapped by NestedType (Struct of Map/Array of Map/Map of Map) is NOT fully supported
 """
+
+python_udf_non_gpu_execs = ['BatchEvalPythonExec']
+if is_spark_420_or_later():
+    # Spark 4.2 can plan regular Python udf(...) as ArrowEvalPythonExec when
+    # Arrow-optimized Python UDFs are enabled.
+    python_udf_non_gpu_execs.append('ArrowEvalPythonExec')
+
 parquet_gens_list = [
     [byte_gen, short_gen, int_gen, long_gen, float_gen, double_gen,
      string_gen, boolean_gen, date_gen,
@@ -266,7 +273,7 @@ def test_hybrid_parquet_filter_pushdown_cpu(spark_tmp_path):
         lambda spark: spark.read.parquet(data_path).filter(f.col("a").startswith('1') & (f.ascii(f.col("a")) >= 50) & (f.col("a") < '1000')),
         conf=filter_split_conf)
 
-@allow_non_gpu('FilterExec', 'BatchEvalPythonExec', 'PythonUDF')
+@allow_non_gpu('FilterExec', *python_udf_non_gpu_execs, 'PythonUDF')
 @pytest.mark.skipif(is_databricks_runtime(), reason="Hybrid feature does not support Databricks currently")
 @pytest.mark.skipif(not is_hybrid_backend_loaded(), reason="HybridScan specialized tests")
 @hybrid_test