uros-b commented on code in PR #58496:
URL: https://github.com/apache/spark/pull/58496#discussion_r3955899159
##########
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")
Review Comment:
Tests do not actually prove reader consistency.
The new ParquetIOSuite test only asserts:
```
assert(error.getCondition ===
"FAILED_READ_FILE.PARQUET_COLUMN_DATA_TYPE_MISMATCH")
```
That hides a real payload mismatch. Vectorized uses
Arrays.toString(descriptor.getPath()) ([c]); the row converter uses
parquetType.getName (c). Nearby tests (SPARK-35640, the vectorized half of
ParquetSchemaSuite) use checkErrorMatchPVals with "column" -> "\\[c\\]". If
this test did the same, the row reader would fail.
The updated ParquetSchemaSuite row-reader case still only checks the cause
type and "Encountered error while reading file", while the vectorized sibling
already checks the full error class and [a]. Now that both throw the same
exception, those two tests should be aligned.
Please also assert expectedType / actualType (decimal(10,0) vs INT32, string
vs FIXED_LEN_BYTE_ARRAY).
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetRowConverter.scala:
##########
@@ -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)
Review Comment:
User-facing change is a bit wider than the description.
The PR description lists only:
- FIXED_LEN_BYTE_ARRAY → STRING
- INT32 DATE → DECIMAL
However, the code also changes:
- INT32/INT64/BOOLEAN/FLOAT/INT96 → STRING; Before: ParquetDecodingException
(or silent UTF-8 for INT96 / FLBA); After: PARQUET_COLUMN_DATA_TYPE_MISMATCH.
- INT64 TIMESTAMP/TIME → DECIMAL; Before: day/time ticks as decimal; After:
mismatch.
- UINT32 → DECIMAL; Before: treated as signed INT32 decimal; After: mismatch
(vectorized already rejects).
The INT32 vs STRING schema-mismatch test change is this broader STRING
guard, not the DATE/FLBA cases. Please expand the user-facing section, and add
at least INT64 TIMESTAMP → DECIMAL and INT96 → STRING (the other
silent-corruption path). ParquetTypeWideningSuite still does not list DateType
→ DecimalType as unsupported.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetRowConverter.scala:
##########
@@ -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 =>
Review Comment:
Nit regarding robustness on the asPrimitiveType() on the new StringType
match:
```
case _: StringType if parquetType.asPrimitiveType().getPrimitiveTypeName ==
BINARY =>
```
If the Parquet type is a group (struct/list requested as STRING), this
throws IllegalStateException instead of
SchemaColumnConvertNotSupportedException. Previously this path did not call
asPrimitiveType(). Guard with parquetType.isPrimitive && ....
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetRowConverter.scala:
##########
@@ -329,6 +329,15 @@ private[parquet] class ParquetRowConverter(
}
}
+ def canReadAsDecimal: Boolean = {
Review Comment:
There is a possible follow up here regarding the remaining decimal parity
gap. Please investigate more.
Vectorized isDecimalTypeMatched also requires the requested decimal to be
wide enough (integerPrecision >= IntDecimal.precision for unannotated INT32).
The new canReadAsDecimal only looks at the logical annotation, so INT32 →
Decimal(5,0) can still succeed on the row reader and fail on the vectorized
reader (ParquetTypeWideningSuite already documents that). Not a regression, but
it is the same class of “flip enableVectorizedReader and the query changes” bug
this PR is fixing. Worth a comment that this mirrors only the annotation half
of isDecimalTypeMatched and/or linking a Jira item to fully close the gap.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
