sunchao commented on code in PR #59972:
URL: https://github.com/apache/spark/pull/59972#discussion_r4433554841
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetReadSupport.scala:
##########
@@ -106,20 +106,61 @@ 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:
Agreed. I split the reader fix into #59997 so it can be backported
independently. I reproduced both cases with constant projections on master. The
two-level LIST read failed in V1 and V2, and the row reader returned `[10, 20]`
for a missing field ID where the vectorized reader returned `[NULL, NULL]`. The
standalone change fixes both. Its separate JIRA is pending.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/objects.scala:
##########
@@ -221,7 +222,14 @@ object ObjectSerializerPruning extends Rule[LogicalPlan] {
val usedRefs = p.references
val prunedSerializer = s.serializer.filter(usedRefs.contains)
- val rootFields = SchemaPruning.identifyRootFields(p.projectList,
Seq.empty)
+ // Resolved references can use the spelling from the query. Both pruning
and rebuilding
+ // field accesses need the names from the serializer, including when a
root is used twice.
+ val namesById = s.output.map(a => a.exprId -> a.name).toMap
Review Comment:
Agreed. #59996 isolates the Dataset serializer fix and has regression tests
with constant selectors. On master, the constant Map query returned `(map-a,
other-a)` rather than the requested `(map-b, other-a)`, and the
case-insensitive array reference threw `ArrayIndexOutOfBoundsException`. Both
pass with the standalone fix. Its separate JIRA is pending.
##########
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.
+ catalystFieldIdxByName.get(parquetField.getName).filterNot { index =>
+ useFieldId && ParquetUtils.hasFieldId(catalystType(index))
+ }
}.map { catalystFieldIndex =>
val catalystField = catalystType(catalystFieldIndex)
// Create a RowUpdater instance for converting Parquet objects to
Catalyst rows.
val rowUpdater: RowUpdater = new RowUpdater(currentRow,
catalystFieldIndex)
// Converted field value should be set to the `fieldIndex`-th cell of
`currentRow`
newConverter(parquetField, catalystField.dataType, rowUpdater)
}.getOrElse {
- // This should only happen if we are reading an arbitrary field from a
struct for its levels
- // that is not otherwise requested.
- val catalystType =
SparkShreddingUtils.parquetTypeToSparkType(parquetField)
- newConverter(parquetField, catalystType, NoopUpdater)
+ // Only the enclosing struct's levels matter. Its clipped child may no
longer carry
+ // enough information to infer its logical type (for example, a
two-level LIST).
+ ParquetRowConverter.discardConverter(parquetField)
Review Comment:
You're right that both tests pass on unmodified master. They protect
interactions introduced by the LIST/field-ID fix, so I moved them to #59997. I
rebuilt that patch on Spark master's CI build and reverted each change
separately. Removing the discard converter made the enclosing-struct test fail
because Parquet requested a group converter and received a primitive converter.
Reading the session's field-ID setting instead of the scan's made the V2 row
reader return NULL where the query expected 0. V1 still ignores the per-read
option. The option regression test is V2-only.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/SelectedField.scala:
##########
@@ -70,23 +63,41 @@ object SelectedField {
private[catalyst] def withDataType(
expr: Expression,
dataType: DataType): Option[StructField] = {
+ prunedRootField(expr, Some(dataType))
+ }
+
+ private def prunedRootField(expr: Expression, dataType: Option[DataType]):
Option[StructField] = {
val unaliased = expr match {
case Alias(child, _) => child
- case expression => expression
+ case other => other
}
- selectField(unaliased, Some(dataType))
+ // A separate projection can read a nested default under this same root.
Preserve its
+ // declared type for dynamic lookups, including when the default is on a
sibling.
+ selectField(unaliased, dataType).filterNot { case (root, field) =>
Review Comment:
I reproduced the underlying bug on master with the nonempty default.
`arr[0].b`, `arr.b`, and the defaulted struct's `s.b` all fail with
`INVALID_DEFAULT_VALUE.DATA_TYPE`. I also tried removing the guard here. On
older files, `arr[idx].b` then changed from 22 to that same error, on V1 and V2
with both Parquet readers. I kept the guard for this feature. I agree the
general fix belongs in a separate change that resolves the default using the
declared type and then projects it. The follow-up JIRA is pending.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/SchemaPruning.scala:
##########
@@ -67,7 +67,19 @@ object SchemaPruning extends Rule[LogicalPlan] {
val normalizedProjects = normalizeAttributeRefNames(attrNameMap, projects)
.asInstanceOf[Seq[NamedExpression]]
val normalizedFilters = normalizeAttributeRefNames(attrNameMap, filters)
- val requestedRootFields = identifyRootFields(normalizedProjects,
normalizedFilters)
+ val pruneDynamicSelectors = hadoopFsRelation.fileFormat match {
+ case _: OrcFileFormat => false
Review Comment:
Confirmed both on master with row and vectorized readers. V2 positional
evolution returned the sibling's 11 for `arr[0].wanted` when the requested
value was 22. V1 raised `UNSUPPORTED_FEATURE.ORC_TYPE_CAST` for the constant
lookup of the TIMESTAMP_NTZ field. With the index supplied by a column, the
same queries currently return 22 and the requested timestamp. I kept the V1/V2
exclusions to preserve those results. I prepared separate JIRA reports for the
positional and timestamp bugs. The keys are pending.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/parquet/ParquetScanBuilder.scala:
##########
@@ -55,6 +55,9 @@ case class ParquetScanBuilder(
override protected val supportsNestedSchemaPruning: Boolean = true
+ override private[sql] def supportsDynamicSelectorPruning: Boolean =
Review Comment:
I would keep the V1/V2 check. I tested a one-element array with an index
column containing 1. Under ANSI and legacy missing-field mode,
`items[1].missing` already returns NULL, as you noted. But `items[idx].missing`
currently raises `INVALID_ARRAY_INDEX` on both V1 and V2, with both Parquet
readers. Removing the check silently turned that existing exception into NULL
in all four combinations. Keeping the collection shape for this mode preserves
the observable behavior of the query we're optimizing.
--
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]
