viirya commented on code in PR #59972:
URL: https://github.com/apache/spark/pull/59972#discussion_r4433280742
##########
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:
This fixes a master bug that has nothing to do with dynamic lookups. On
master, with the two-level LIST schema from the new test, `legacy[1].renamed`
fails on both readers. With field IDs, the row reader returns `[10, 20]` for a
plain `legacy.sibling` instead of `[null, null]`. Could this go into its own
JIRA so it can be backported?
##########
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:
These two changes also fix master bugs:
- Under case-insensitive analysis, `items[0]._2` and `items._2` throw
`ArrayIndexOutOfBoundsException`.
- With a constant key, `ds.selectExpr("_1._1['k']._2", "_1._2._1")` on the
new test's data returns `map-a` instead of `map-b`.
Same request: a separate JIRA.
##########
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:
This failure is not specific to dynamic lookups. After `ADD COLUMNS (arr
ARRAY<STRUCT<a,b>> DEFAULT array(named_struct('a', 11, 'b', 22)))`, `arr[0].b`,
`arr.b` and even a struct column's `s.b` fail with
`INVALID_DEFAULT_VALUE.DATA_TYPE` on both master and this PR, because the
existence default is coerced to the pruned type.
`arr[idx].b` requests the same read schema as `arr[0].b`, so this guard
hides the bug on one path only. The constant-lookup assertions pass only
because `DEFAULT array()`/`map()` fit any pruned type.
Could the coercion be fixed for all nested pruning in a separate change, and
`hasDynamicLookupOnPath`/`hasNestedDefault` be dropped here?
##########
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:
"a two-level Parquet LIST preserves enclosing struct nullness after pruning"
and "row reader uses the scan's field ID setting" both pass on master without
this PR, so I can't find what this change, or the `schemaConverter.useFieldId`
change at L225, fixes on master. Were they needed only on the 4.1.3 base? If
so, could they be dropped? Also, V1 still ignores the per-read field-ID option.
##########
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:
In legacy missing-field mode under ANSI, the constant `items[1].missing`
already returns null, as the conf's doc warns. Is it worth the V1/V2 plumbing
to keep dynamic lookups stricter than constant ones?
##########
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:
Constant lookups already fail on ORC on master:
- V2 returns 11 instead of 22 for `arr[0].wanted` under positional evolution.
- V1 fails in the `timestamp_ntz` case.
Excluding dynamic lookups doesn't fix either. Could they be filed as bugs?
(Same for `OrcScanBuilder.supportsDynamicSelectorPruning`.)
--
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]
