Merge branch 'release/26.08' into cudf-spark-rename-1

Author: GaryShen2008

integration_tests/src/main/python/parquet_test.py

@@ -22,6 +22,7 @@
 from parquet_write_test import parquet_datetime_gen_simple, parquet_nested_datetime_gen, parquet_ts_write_options
 from marks import *
 import pyarrow as pa
+import pyarrow.parquet as pq
 from parquet_test_utils import parquet_row_group_midpoints
 from pyspark.sql.types import *
 from pyspark.sql.functions import *
@@ -1991,3 +1992,185 @@ def setup_table(spark):
     with_cpu_session(lambda spark: setup_table(spark))
     assert_gpu_and_cpu_are_equal_collect(lambda spark: spark.read.parquet(data_path).select("p"),
                                          conf={"spark.rapids.sql.columnSizeBytes": "100"})
+
+
+def _write_parquet_unknown_null_table(
+        data_path, with_list=False, with_map=False, field_id=None):
+    """Write INT32 physical + UNKNOWN/Null logical annotation (Spark void_in_parquet shape)."""
+    if with_list:
+        table = pa.table({
+            'list_void': pa.array([[None, None], [None], None], type=pa.list_(pa.null())),
+        })
+    elif with_map:
+        table = pa.table({
+            'map_void': pa.array(
+                [{1: None, 2: None}, {3: None}, None],
+                type=pa.map_(pa.int32(), pa.null())),
+        })
+    elif field_id is not None:
+        arrow_schema = pa.schema([
+            pa.field('void_col', pa.null(), metadata={b'PARQUET:field_id': str(field_id).encode()}),
+        ])
+        table = pa.Table.from_arrays(
+            [pa.array([None, None, None], type=pa.null())], schema=arrow_schema)
+    else:
+        table = pa.table({
+            'id': pa.array([1, 2, 3], type=pa.int32()),
+            'void_col': pa.array([None, None, None], type=pa.null()),
+        })
+    pq.write_table(table, data_path)
+
+
+# SPARK-56045 / SPARK-54220: Parquet UNKNOWN logical type annotation. PyArrow null columns are
+# written as INT32 physical + UNKNOWN/Null logical annotation.
+
+@pytest.mark.skipif(is_spark_411_or_later(),
+                    reason='pre-SPARK-54220 physical-type behavior')
+@pytest.mark.parametrize('reader_confs', reader_opt_confs)
+def test_parquet_unknown_type_annotation_pre_411_physical(spark_tmp_path, reader_confs):
+    data_path = spark_tmp_path + '/PARQUET_UNKNOWN_PRE_411'
+    _write_parquet_unknown_null_table(data_path)
+
+    def read_and_check_schema(spark):
+        df = spark.read.parquet(data_path)
+        assert df.schema['void_col'].dataType == IntegerType(), \
+            f"expected void_col=IntegerType, got {df.schema['void_col'].dataType}"
+        return df
+
+    assert_gpu_and_cpu_are_equal_collect(read_and_check_schema, conf=reader_confs)
+
+
+@pytest.mark.skipif(not is_spark_412_or_later(),
+                    reason='SPARK-56045 requires Spark 4.1.2+')
+@pytest.mark.parametrize('reader_confs', reader_opt_confs)
+def test_parquet_unknown_type_annotation_default_physical(spark_tmp_path, reader_confs):
+    data_path = spark_tmp_path + '/PARQUET_UNKNOWN'
+    _write_parquet_unknown_null_table(data_path)
+
+    conf = copy_and_update(reader_confs, {
+        'spark.sql.parquet.reader.respectUnknownTypeAnnotation.enabled': 'false',
+    })
+
+    def read_and_check_schema(spark):
+        df = spark.read.parquet(data_path)
+        assert df.schema['void_col'].dataType == IntegerType(), \
+            f"expected void_col=IntegerType, got {df.schema['void_col'].dataType}"
+        return df
+
+    assert_gpu_and_cpu_are_equal_collect(read_and_check_schema, conf=conf)
+
+
+@pytest.mark.skipif(not is_spark_411_or_later(),
+                    reason='SPARK-54220 requires Spark 4.1.1+')
+@pytest.mark.parametrize('reader_confs', reader_opt_confs)
+@allow_non_gpu('FileSourceScanExec', 'ColumnarToRowExec')
+def test_parquet_unknown_type_annotation_respect_nulltype(spark_tmp_path, reader_confs):
+    data_path = spark_tmp_path + '/PARQUET_UNKNOWN'
+    _write_parquet_unknown_null_table(data_path)
+
+    # Spark 4.1.1 always maps UNKNOWN to NullType; 4.1.2+ needs the conf enabled.
+    conf = reader_confs
+    if is_spark_412_or_later():
+        conf = copy_and_update(reader_confs, {
+            'spark.sql.parquet.reader.respectUnknownTypeAnnotation.enabled': 'true',
+        })
+
+    def read_and_check_schema(spark):
+        df = spark.read.parquet(data_path)
+        assert df.schema['void_col'].dataType == NullType(), \
+            f"expected void_col=NullType, got {df.schema['void_col'].dataType}"
+        return df
+
+    # GPU Parquet scan does not support NullType yet; expect CPU fallback.
+    assert_gpu_fallback_collect(read_and_check_schema, 'FileSourceScanExec', conf=conf)
+
+
+@pytest.mark.skipif(not is_spark_412_or_later(),
+                    reason='SPARK-56045 requires Spark 4.1.2+')
+@pytest.mark.parametrize('reader_confs', reader_opt_confs)
+def test_parquet_unknown_type_annotation_explicit_int_schema(spark_tmp_path, reader_confs):
+    """Explicit non-Null schema should strip UNKNOWN even when respect conf is true."""
+    data_path = spark_tmp_path + '/PARQUET_UNKNOWN_EXPLICIT'
+    _write_parquet_unknown_null_table(data_path)
+
+    read_schema = StructType([
+        StructField('id', IntegerType(), True),
+        StructField('void_col', IntegerType(), True),
+    ])
+    conf = copy_and_update(reader_confs, {
+        'spark.sql.parquet.reader.respectUnknownTypeAnnotation.enabled': 'true',
+    })
+
+    def read_and_check_schema(spark):
+        df = spark.read.schema(read_schema).parquet(data_path)
+        assert df.schema['void_col'].dataType == IntegerType(), \
+            f"expected void_col=IntegerType, got {df.schema['void_col'].dataType}"
+        return df
+
+    assert_gpu_and_cpu_are_equal_collect(read_and_check_schema, conf=conf)
+
+
+@pytest.mark.skipif(not is_spark_412_or_later(),
+                    reason='SPARK-56045 requires Spark 4.1.2+')
+@pytest.mark.parametrize('reader_confs', reader_opt_confs)
+def test_parquet_unknown_type_annotation_preserves_field_id(spark_tmp_path, reader_confs):
+    data_path = spark_tmp_path + '/PARQUET_UNKNOWN_FIELD_ID'
+    _write_parquet_unknown_null_table(data_path, field_id=7)
+
+    read_schema = StructType([
+        StructField('renamed_void', IntegerType(), True, metadata=with_id(7)),
+    ])
+    conf = copy_and_update(
+        reader_confs,
+        enable_parquet_field_id_read,
+        {'spark.sql.parquet.reader.respectUnknownTypeAnnotation.enabled': 'false'})
+
+    def read_and_check_schema(spark):
+        df = spark.read.schema(read_schema).parquet(data_path)
+        assert df.schema['renamed_void'].dataType == IntegerType(), \
+            f"expected renamed_void=IntegerType, got {df.schema['renamed_void'].dataType}"
+        return df
+
+    assert_gpu_and_cpu_are_equal_collect(read_and_check_schema, conf=conf)
+
+
+@pytest.mark.skipif(not is_spark_412_or_later(),
+                    reason='SPARK-56045 requires Spark 4.1.2+')
+@pytest.mark.parametrize('reader_confs', reader_opt_confs)
+def test_parquet_unknown_type_annotation_list_physical(spark_tmp_path, reader_confs):
+    """Primitive-element lists bypass structural clipping; UNKNOWN must still be stripped."""
+    data_path = spark_tmp_path + '/PARQUET_UNKNOWN_LIST'
+    _write_parquet_unknown_null_table(data_path, with_list=True)
+
+    conf = copy_and_update(reader_confs, {
+        'spark.sql.parquet.reader.respectUnknownTypeAnnotation.enabled': 'false',
+    })
+
+    def read_and_check_schema(spark):
+        df = spark.read.parquet(data_path)
+        assert df.schema['list_void'].dataType == ArrayType(IntegerType()), \
+            f"expected ArrayType(IntegerType), got {df.schema['list_void'].dataType}"
+        return df
+
+    assert_gpu_and_cpu_are_equal_collect(read_and_check_schema, conf=conf)
+
+
+@pytest.mark.skipif(not is_spark_412_or_later(),
+                    reason='SPARK-56045 requires Spark 4.1.2+')
+@pytest.mark.parametrize('reader_confs', reader_opt_confs)
+def test_parquet_unknown_type_annotation_map_physical(spark_tmp_path, reader_confs):
+    """Primitive-value maps bypass structural clipping; UNKNOWN must still be stripped."""
+    data_path = spark_tmp_path + '/PARQUET_UNKNOWN_MAP'
+    _write_parquet_unknown_null_table(data_path, with_map=True)
+
+    conf = copy_and_update(reader_confs, {
+        'spark.sql.parquet.reader.respectUnknownTypeAnnotation.enabled': 'false',
+    })
+
+    def read_and_check_schema(spark):
+        df = spark.read.parquet(data_path)
+        assert df.schema['map_void'].dataType == MapType(IntegerType(), IntegerType()), \
+            f"expected MapType(IntegerType, IntegerType), got {df.schema['map_void'].dataType}"
+        return df
+
+    assert_gpu_and_cpu_are_equal_collect(read_and_check_schema, conf=conf)

sql-plugin/src/main/scala/com/nvidia/spark/rapids/parquet/ParquetSchemaUtils.scala

@@ -24,7 +24,9 @@ import ai.rapids.cudf.{ColumnView, DType, Table}
 import com.nvidia.spark.rapids.{CastOptions, GpuCast, GpuColumnVector, SchemaUtils}
 import com.nvidia.spark.rapids.Arm.{closeOnExcept, withResource}
 import com.nvidia.spark.rapids.shims.parquet.ParquetSchemaClipShims
+import com.nvidia.spark.rapids.shims.parquet.ParquetUnknownTypeAnnotationShims
 import org.apache.parquet.schema._
+import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName
 import org.apache.parquet.schema.Type.Repetition
 
 import org.apache.spark.sql.rapids.execution.TrampolineUtil
@@ -80,18 +82,93 @@ object ParquetSchemaUtils {
         clipParquetGroup(parquetType.asGroupType(), t, caseSensitive, useFieldId)
 
       case _ =>
-        // UDTs and primitive types are not clipped.  For UDTs, a clipped version might not be able
-        // to be mapped to desired user-space types.  So UDTs shouldn't participate schema merging.
-        parquetType
+        // UDTs, primitive types, and primitive-element arrays/maps are not clipped
+        // structurally. Still normalize UNKNOWN annotations so cuDF sees the physical
+        // type when Spark would ignore the annotation (SPARK-56045).
+        stripIgnoredUnknownAnnotation(parquetType, catalystType)
     }
 
-    if (useFieldId && parquetType.getId != null) {
+    // Nested rebuilds may already carry field IDs; only re-apply when missing.
+    if (useFieldId && parquetType.getId != null && newParquetType.getId == null) {
       newParquetType.withId(parquetType.getId.intValue())
     } else {
       newParquetType
     }
   }
 
+  /**
+   * Leaf Catalyst type used when deciding whether an UNKNOWN annotation should be
+   * stripped for nested array/map groups that bypass structural clipping.
+   */
+  private def unknownAnnotationLeafType(catalystType: DataType): DataType = {
+    catalystType match {
+      case ArrayType(elementType, _) => unknownAnnotationLeafType(elementType)
+      case MapType(_, valueType, _) => unknownAnnotationLeafType(valueType)
+      case other => other
+    }
+  }
+
+  /**
+   * Rebuild primitives without an UNKNOWN logical annotation when Spark would not map
+   * that annotation to NullType for the requested Catalyst type. Recurses through group
+   * nodes so primitive-element lists/maps that skip structural clipping are covered.
+   *
+   * @param preserveFieldId when rebuilding nested primitives (not returning through
+   *   [[clipParquetType]]), copy the original field ID onto the rebuilt type.
+   */
+  @scala.annotation.nowarn("msg=method as in class Builder is deprecated")
+  private def stripIgnoredUnknownAnnotation(
+      parquetType: Type,
+      catalystType: DataType,
+      preserveFieldId: Boolean = false): Type = {
+    if (!parquetType.isPrimitive) {
+      val group = parquetType.asGroupType()
+      val leafType = unknownAnnotationLeafType(catalystType)
+      val fields = group.getFields.asScala
+      val strippedFields = fields.map { field =>
+        stripIgnoredUnknownAnnotation(field, leafType, preserveFieldId = true)
+      }
+      if (fields.iterator.zip(strippedFields.iterator).forall {
+            case (original, stripped) => original eq stripped
+          }) {
+        parquetType
+      } else {
+        group.withNewFields(strippedFields.asJava)
+      }
+    } else {
+      val primitive = parquetType.asPrimitiveType()
+      val rawAnnotation = primitive.getLogicalTypeAnnotation
+      val leafType = unknownAnnotationLeafType(catalystType)
+      // Respect conf for inferred/requested NullType; strip UNKNOWN when the requested
+      // type is a non-Null physical type even if respectUnknownTypeAnnotation is true.
+      val effectiveAnnotation =
+        if (leafType != NullType &&
+            ParquetUnknownTypeAnnotationShims.mapsToNullType(rawAnnotation)) {
+          null
+        } else {
+          ParquetUnknownTypeAnnotationShims.effectiveLogicalTypeAnnotation(rawAnnotation)
+        }
+      if (rawAnnotation == effectiveAnnotation) {
+        parquetType
+      } else {
+        val builder = Types.primitive(primitive.getPrimitiveTypeName, primitive.getRepetition)
+        if (primitive.getPrimitiveTypeName == PrimitiveTypeName.FIXED_LEN_BYTE_ARRAY) {
+          builder.length(primitive.getTypeLength)
+        }
+        if (effectiveAnnotation != null) {
+          builder.as(effectiveAnnotation)
+        }
+        val rebuilt = builder.named(primitive.getName)
+        if (preserveFieldId && primitive.getId != null) {
+          rebuilt.withId(primitive.getId.intValue())
+        } else {
+          // Top-level field IDs are re-applied by clipParquetType after this returns.
+          rebuilt
+        }
+      }
+    }
+  }
+
   /**
    * Whether a Catalyst DataType is primitive.  Primitive DataType is not equivalent to
    * AtomicType.  For example, CalendarIntervalType is primitive, but it's not an