diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetRowConverter.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetRowConverter.scala index 1b633d63dc22b..8dae84a8ecc64 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetRowConverter.scala +++ b/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) diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetIOSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetIOSuite.scala index b9ac9df8e31d3..d166cb7d8776d 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetIOSuite.scala +++ b/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 diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetSchemaSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetSchemaSuite.scala index e468a1449783b..92b7cb5d89a9d 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetSchemaSuite.scala +++ b/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)) } }