sunchao commented on code in PR #59997:
URL: https://github.com/apache/spark/pull/59997#discussion_r4437973588
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetReadSupport.scala:
##########
@@ -106,13 +106,48 @@ class ParquetReadSupport(
new ParquetRecordMaterializer(
parquetRequestedSchema,
ParquetReadSupport.expandUDT(catalystRequestedSchema),
- new ParquetToSparkSchemaConverter(conf),
+ new ParquetReadSchemaConverter(conf, parquetRequestedSchema, fileSchema),
convertTz,
datetimeRebaseSpec,
int96RebaseSpec)
}
}
+/**
+ * Interpret LISTs using the file schema. Clipping a repeated group to one
field can make a
+ * two-level LIST indistinguishable from a three-level LIST in the requested
schema.
+ */
+private[parquet] class ParquetReadSchemaConverter(
+ conf: Configuration,
+ requestedSchema: MessageType,
+ fileSchema: MessageType) extends ParquetToSparkSchemaConverter(conf) {
+
+ private lazy val fileRepeatedTypes = {
+ // Equal requested types at different paths can have different layouts in
the file.
+ val types = new util.IdentityHashMap[Type, Type]()
+ def visit(requested: GroupType, file: GroupType): Unit = {
+ if
(file.getLogicalTypeAnnotation.isInstanceOf[ListLogicalTypeAnnotation] &&
+ requested.getFieldCount == 1 && file.getFieldCount == 1) {
+ types.put(requested.getType(0), file.getType(0))
+ }
+ requested.getFields.asScala.foreach {
Review Comment:
Added the nested-LIST case in d28cfa1113f. With no field IDs,
`outer[0].inner[1].a` returns 3 on both readers in V1 and V2. On the parent,
the row reader returns NULL and the vectorized reader fails. I also disabled
descent through LIST nodes locally: the new test failed in V1 and V2 while the
five existing cases still passed. This covers the recursive lookup without a
second Map fixture. I updated the description with the no-ID example and the
repeated-group names that already worked.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetSchemaPruningSuite.scala:
##########
@@ -35,6 +41,89 @@ abstract class ParquetSchemaPruningSuite extends
SchemaPruningSuite with Adaptiv
override protected val vectorizedReaderNestedEnabledKey: String =
SQLConf.PARQUET_VECTORIZED_READER_NESTED_COLUMN_ENABLED.key
+ test("a pruned two-level Parquet LIST preserves its element type and field
IDs") {
Review Comment:
Added the `SPARK-60788:` prefix to all the regression tests for this change,
including the new nested-LIST case, in d28cfa1113f.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetReadSupport.scala:
##########
@@ -106,13 +106,48 @@ class ParquetReadSupport(
new ParquetRecordMaterializer(
parquetRequestedSchema,
ParquetReadSupport.expandUDT(catalystRequestedSchema),
- new ParquetToSparkSchemaConverter(conf),
+ new ParquetReadSchemaConverter(conf, parquetRequestedSchema, fileSchema),
convertTz,
datetimeRebaseSpec,
int96RebaseSpec)
}
}
+/**
+ * Interpret LISTs using the file schema. Clipping a repeated group to one
field can make a
+ * two-level LIST indistinguishable from a three-level LIST in the requested
schema.
+ */
+private[parquet] class ParquetReadSchemaConverter(
Review Comment:
Moved `ParquetReadSchemaConverter` below the companion object in d28cfa1113f
so the class and its companion sit together.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetReadSupport.scala:
##########
@@ -106,13 +106,48 @@ class ParquetReadSupport(
new ParquetRecordMaterializer(
parquetRequestedSchema,
ParquetReadSupport.expandUDT(catalystRequestedSchema),
- new ParquetToSparkSchemaConverter(conf),
+ new ParquetReadSchemaConverter(conf, parquetRequestedSchema, fileSchema),
convertTz,
datetimeRebaseSpec,
int96RebaseSpec)
}
}
+/**
+ * Interpret LISTs using the file schema. Clipping a repeated group to one
field can make a
+ * two-level LIST indistinguishable from a three-level LIST in the requested
schema.
+ */
+private[parquet] class ParquetReadSchemaConverter(
+ conf: Configuration,
+ requestedSchema: MessageType,
+ fileSchema: MessageType) extends ParquetToSparkSchemaConverter(conf) {
+
+ private lazy val fileRepeatedTypes = {
Review Comment:
I kept this lazy. The converter is used immediately, but only
`isElementType` consults this map, and that method is reached when interpreting
a LIST. An eager val would walk the projected schema and allocate an identity
map on scalar-only or struct-only reads too.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetRowConverter.scala:
##########
@@ -254,19 +255,20 @@ private[parquet] class ParquetRowConverter(
// field has id, try to match by id first before falling back to match
by name
catalystFieldIdxByFieldId.get(fieldId.intValue())
}.orElse {
- // field doesn't have id, just match by name
- catalystFieldIdxByName.get(parquetField.getName)
+ // A field read only for struct nullness can share a name with a
missing field ID.
Review Comment:
Updated the comment in d28cfa1113f to state the field-ID rule and point to
`clipParquetGroupFields`. I also reproduced the example with no LIST and no IDs
in the file: with `ignoreMissing=true`, the parent row reader returned 42 and
this PR returns NULL. That example is now in the description.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetSchemaPruningSuite.scala:
##########
@@ -70,6 +159,23 @@ class ParquetV2SchemaPruningSuite extends
ParquetSchemaPruningSuite {
.sparkConf
.set(SQLConf.USE_V1_SOURCE_LIST, "")
+ test("row reader uses the scan's field ID setting") {
Review Comment:
Yes. V1 uses the session setting for both clipping and conversion. V2 honors
the per-read option, which is where the settings could diverge. I added a
comment explaining why this test is V2-only in d28cfa1113f.
--
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]
