Merge branch 'main' into regex-inline-flags

Author: igorpeshansky

iceberg/common/src/main/java/org/apache/iceberg/spark/source/GpuStructInternalRow.java

@@ -19,14 +19,12 @@
 import com.nvidia.spark.rapids.RapidsConf;
 import com.nvidia.spark.rapids.iceberg.IcebergShimUtils;
 import org.apache.iceberg.*;
-import org.apache.iceberg.data.IdentityPartitionConverters;
 import org.apache.iceberg.io.FileIO;
 import org.apache.iceberg.io.StorageCredential;
 import org.apache.iceberg.io.SupportsStorageCredentials;
+import org.apache.iceberg.spark.SparkUtil;
 import org.apache.iceberg.spark.source.GpuSparkCopyOnWriteV1Scan;
 import org.apache.iceberg.spark.source.GpuSparkScan;
-import org.apache.iceberg.spark.source.GpuStructInternalRow;
-import org.apache.iceberg.types.Type;
 import org.apache.iceberg.types.Types;
 import org.apache.iceberg.util.PartitionUtil;
 import org.apache.spark.sql.connector.read.Scan;
@@ -35,33 +33,23 @@
 import java.util.HashMap;
 import java.util.Map;
 
-/** Iceberg 1.9.x shim: uses {@code IdentityPartitionConverters::convertConstant}. */
+/** Iceberg 1.9.x shim: uses {@code SparkUtil::internalToSpark}. */
 public class ShimUtilsImpl implements IcebergShimUtils {
     @Override
     public String locationOf(ContentFile<?> f) {
         return f.location();
     }
 
-    private static Object convertConstant(Type type, Object value) {
-        Object converted = IdentityPartitionConverters.convertConstant(type, value);
-        if (converted instanceof StructLike && type instanceof Types.StructType) {
-            GpuStructInternalRow row = new GpuStructInternalRow((Types.StructType) type);
-            row.setStruct((StructLike) converted);
-            return row;
-        }
-        return converted;
-    }
-
     @Override
     public Map<Integer, ?> constantsMap(FileScanTask task, Schema readSchema, Table table) {
         if (readSchema.findField(MetadataColumns.PARTITION_COLUMN_ID) != null) {
             Types.StructType partitionType = Partitioning.partitionType(table);
             return PartitionUtil.constantsMap(task,
                     partitionType,
-                    ShimUtilsImpl::convertConstant);
+                    SparkUtil::internalToSpark);
         } else {
             return PartitionUtil.constantsMap(task,
-                    ShimUtilsImpl::convertConstant);
+                    SparkUtil::internalToSpark);
         }
     }
 

iceberg/iceberg-1-9-x/src/main/java/com/nvidia/spark/rapids/iceberg/iceberg19x/ShimUtilsImpl.java

@@ -24,7 +24,9 @@
 from pyspark.sql.types import *
 import pyspark.sql.utils
 import pyspark.sql.functions as f
-from spark_session import with_cpu_session, with_gpu_session, is_databricks104_or_later, is_databricks_version_or_later, is_before_spark_320, is_spark_400_or_later, is_before_spark_340
+from spark_session import with_cpu_session, with_gpu_session, is_databricks104_or_later, \
+    is_databricks_version_or_later, is_before_spark_320, is_spark_400_or_later, \
+    is_before_spark_340, spark_version
 
 _regexp_conf = { 'spark.rapids.sql.regexp.enabled': 'true' }
 
@@ -1150,3 +1152,57 @@ def test_multi_contains_conditional(ansi):
                     ELSE CAST(a AS LONG)
                 END as result'''),
             conf = conf)
+
+
+# SPARK-57507 / #15383: reverse must clamp truncated trailing UTF-8 to the row boundary.
+# Neighbor sentinel rows make an over-read observable.
+def _cpu_reverse_has_spark_57507():
+    version = spark_version()
+    # The fix was backported to the 4.0.4, 4.1.3, and 4.2.0 release lines.
+    return ('4.0.4' <= version < '4.1.0'
+            or '4.1.3' <= version < '4.2.0'
+            or version >= '4.2.0')
+
+
+@ignore_order(local=True)
+@pytest.mark.parametrize('hex_in,expected_hex', [
+    ('41CE', 'CE41'),                 # A + truncated 2-byte lead
+    ('41E4B8', 'E4B841'),             # A + truncated 3-byte lead
+    ('41F090', 'F09041'),             # A + truncated 4-byte lead
+    ('E4B896CE', 'CEE4B896'),         # complete 世 + orphan 2-byte lead
+    ('41C3A9', 'C3A941'),             # well-formed 2-byte control
+    ('41E4B896', 'E4B89641'),         # well-formed 3-byte control
+    ('41F0908D88', 'F0908D8841'),     # well-formed 4-byte control
+], ids=idfn)
+def test_reverse_truncated_trailing_utf8(hex_in, expected_hex):
+    def do_it(spark):
+        # Keep target + sentinel neighbors adjacent in one partition so an over-read
+        # into the next row is observable. Avoid ORDER BY (CPU shuffle) and unhex
+        # (not allowed in GpuCpuBridge); feed raw bytes via BINARY then cast to STRING.
+        df = spark.createDataFrame(
+            [
+                (0, bytearray.fromhex(hex_in)),
+                (1, bytearray.fromhex('FFFFFFFF')),
+                (2, bytearray.fromhex('414243')),
+            ],
+            'row_id INT, b BINARY'
+        ).coalesce(1).selectExpr('row_id', 'CAST(b AS STRING) AS v')
+        return df.selectExpr('row_id', 'hex(reverse(v)) as h')
+
+    if _cpu_reverse_has_spark_57507():
+        assert_cpu_and_gpu_are_equal_collect_with_capture(
+            do_it, exist_classes='GpuReverse', require_non_empty=True)
+    else:
+        # Older CPU builds over-read malformed trailing UTF-8, so validate the GPU result
+        # against Spark's corrected semantics instead of asserting CPU/GPU equality.
+        def collect_gpu(spark):
+            out = do_it(spark)
+            rows = out.collect()
+            plan = out._jdf.queryExecution().executedPlan().toString()
+            # Expression trees render as lowercase "gpureverse(...)", not the class name.
+            assert 'gpureverse' in plan.lower(), 'expected GpuReverse in plan, got:\n' + plan
+            return rows
+
+        rows = with_gpu_session(collect_gpu)
+        rows_by_id = {row['row_id']: row['h'] for row in rows}
+        assert rows_by_id[0] == expected_hex

integration_tests/src/main/python/string_test.py

@@ -26,7 +26,7 @@ import com.nvidia.spark.rapids.ArrayIndexUtils.firstIndexAndNumElementUnchecked
 import com.nvidia.spark.rapids.BoolUtils.isAllValidTrue
 import com.nvidia.spark.rapids.GpuListUtils
 import com.nvidia.spark.rapids.RapidsPluginImplicits._
-import com.nvidia.spark.rapids.jni.{GpuListSliceUtils, MapUtils}
+import com.nvidia.spark.rapids.jni.{GpuListSliceUtils, MapUtils, StringUtils}
 import com.nvidia.spark.rapids.shims.{GetSequenceSize, NullIntolerantShim, ShimExpression}
 
 import org.apache.spark.sql.catalyst.analysis.{TypeCheckResult, TypeCoercion}
@@ -674,7 +674,14 @@ case class GpuReverse(child: Expression) extends GpuUnaryExpression {
   override def dataType: DataType = child.dataType
 
   override protected def doColumnar(input: GpuColumnVector): ColumnVector = {
-    input.getBase.reverseStringsOrLists()
+    // Strings use Spark UTF8String.reverse semantics (SPARK-57507): clamp truncated trailing
+    // multi-byte UTF-8 widths to the bytes remaining in each row. libcudf reverse can
+    // over-read into the next row for malformed Spark StringType values.
+    if (child.dataType.isInstanceOf[StringType]) {
+      StringUtils.reverseStrings(input.getBase)
+    } else {
+      input.getBase.reverseStringsOrLists()
+    }
   }
 }