[SPARK-59251][SQL] Reject incompatible Parquet row reader conversions

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetRowConverter.scala

@@ -40,7 +40,7 @@ import org.apache.spark.sql.catalyst.util.RebaseDateTime.RebaseSpec
 import org.apache.spark.sql.catalyst.util.ResolveDefaultColumns._
 import org.apache.spark.sql.errors.QueryCompilationErrors
 import org.apache.spark.sql.errors.QueryExecutionErrors
-import org.apache.spark.sql.execution.datasources.{DataSourceUtils, VariantMetadata}
+import org.apache.spark.sql.execution.datasources.{DataSourceUtils, SchemaColumnConvertNotSupportedException, VariantMetadata}
 import org.apache.spark.sql.execution.datasources.parquet.types.ops.ParquetTypeOps
 import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.types._
@@ -329,6 +329,15 @@ private[parquet] class ParquetRowConverter(
       }
     }
 
+    def canReadAsDecimal: Boolean = {
+      parquetType.getLogicalTypeAnnotation match {
+        case _: DecimalLogicalTypeAnnotation => true
+        case null => true
+        case i: IntLogicalTypeAnnotation => i.isSigned
+        case _ => false
+      }
+    }
+
     catalystType match {
       case NullType
         if parquetType.getLogicalTypeAnnotation.isInstanceOf[UnknownLogicalTypeAnnotation] =>
@@ -374,7 +383,8 @@ private[parquet] class ParquetRowConverter(
         }
 
       // For INT32 backed decimals
-      case _: DecimalType if parquetType.asPrimitiveType().getPrimitiveTypeName == INT32 =>
+      case _: DecimalType
+          if parquetType.asPrimitiveType().getPrimitiveTypeName == INT32 && canReadAsDecimal =>
         parquetType.asPrimitiveType().getLogicalTypeAnnotation match {
           case decimalType: DecimalLogicalTypeAnnotation =>
             new ParquetIntDictionaryAwareDecimalConverter(
@@ -395,7 +405,8 @@ private[parquet] class ParquetRowConverter(
         }
 
       // For INT64 backed decimals
-      case t: DecimalType if parquetType.asPrimitiveType().getPrimitiveTypeName == INT64 =>
+      case t: DecimalType
+          if parquetType.asPrimitiveType().getPrimitiveTypeName == INT64 && canReadAsDecimal =>
         parquetType.asPrimitiveType().getLogicalTypeAnnotation match {
           case decimalType: DecimalLogicalTypeAnnotation =>
             new ParquetLongDictionaryAwareDecimalConverter(
@@ -421,12 +432,26 @@ private[parquet] class ParquetRowConverter(
         }
 
       case t: DecimalType =>
-        throw QueryExecutionErrors.cannotCreateParquetConverterForDecimalTypeError(
-          t, parquetType.toString)
+        parquetType.asPrimitiveType().getPrimitiveTypeName match {
+          case INT32 | INT64 =>
+            throw new SchemaColumnConvertNotSupportedException(
+              parquetType.getName,
+              parquetType.asPrimitiveType().getPrimitiveTypeName.toString,
+              t.catalogString)
+          case _ =>
+            throw QueryExecutionErrors.cannotCreateParquetConverterForDecimalTypeError(
+              t, parquetType.toString)
+        }
 
-      case _: StringType =>
+      case _: StringType if parquetType.asPrimitiveType().getPrimitiveTypeName == BINARY =>
         new ParquetStringConverter(updater)
 
+      case t: StringType =>
+        throw new SchemaColumnConvertNotSupportedException(
+          parquetType.getName,
+          parquetType.asPrimitiveType().getPrimitiveTypeName.toString,
+          t.catalogString)
+
       case geom: GeometryType =>
         new ParquetGeometryConverter(geom.srid, updater)
 

sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetIOSuite.scala

@@ -2009,6 +2009,37 @@ class ParquetIOSuite extends ParquetTest with SharedSparkSession {
     }
   }
 
+  test("SPARK-59251: Parquet readers reject incompatible primitive type conversions consistently") {
+    val cases = Seq(
+      ("required int32 c (DATE);", DecimalType(10, 0),
+        (record: SimpleGroup) => record.add(0, 1)),
+      ("required fixed_len_byte_array(4) c;", StringType,
+        (record: SimpleGroup) =>
+          record.add(0, Binary.fromConstantByteArray(Array[Byte](1, 2, 3, 4)))))
+
+    cases.foreach { case (column, readType, writeValue) =>
+      val parquetSchema = MessageTypeParser.parseMessageType(
+        s"message root {\n  $column\n}")
+      val readSchema = new StructType().add("c", readType)
+
+      withTempDir { dir =>
+        val path = new Path(s"${dir.getCanonicalPath}/incompatible.parquet")
+        val writer = createParquetWriter(parquetSchema, path)
+        val record = new SimpleGroup(parquetSchema)
+        writeValue(record)
+        writer.write(record)
+        writer.close()
+
+        withAllParquetReaders {
+          val error = intercept[SparkException] {
+            spark.read.schema(readSchema).parquet(path.toString).collect()
+          }
+          assert(error.getCondition === "FAILED_READ_FILE.PARQUET_COLUMN_DATA_TYPE_MISMATCH")
+        }
+      }
+    }
+  }
+
   test("SPARK-55444: vectorized read rejects an incompatible encoding requested as TimeType") {
     // TimeTypeParquetOps.getVectorUpdater returns None for any encoding other than INT64
     // TIME(MICROS/NANOS), so the vectorized factory falls through to a clean

sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetSchemaSuite.scala

@@ -21,7 +21,6 @@ import scala.reflect.ClassTag
 import scala.reflect.runtime.universe.TypeTag
 
 import org.apache.parquet.column.ColumnDescriptor
-import org.apache.parquet.io.ParquetDecodingException
 import org.apache.parquet.schema._
 import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName
 import org.apache.parquet.schema.Type._
@@ -1047,7 +1046,10 @@ class ParquetSchemaSuite extends ParquetSchemaTest {
     withTempPath { dir =>
       val e = testSchemaMismatch(dir.getCanonicalPath, vectorizedReaderEnabled = false)
       val expectedMessage = "Encountered error while reading file"
-      assert(e.getCause.isInstanceOf[ParquetDecodingException])
+      // SPARK-59251: the row (non-vectorized) reader now rejects incompatible conversions with the
+      // same SchemaColumnConvertNotSupportedException as the vectorized reader, instead of the
+      // reader-internal ParquetDecodingException.
+      assert(e.getCause.isInstanceOf[SchemaColumnConvertNotSupportedException])
       assert(e.getMessage.contains(expectedMessage))
     }
   }