cloud-fan commented on a change in pull request #35147:
URL: https://github.com/apache/spark/pull/35147#discussion_r783941191
##########
File path:
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/SchemaPruning.scala
##########
@@ -31,58 +31,69 @@ import org.apache.spark.sql.util.SchemaUtils._
* By "physical column", we mean a column as defined in the data source format
like Parquet format
* or ORC format. For example, in Spark SQL, a root-level Parquet column
corresponds to a SQL
* column, and a nested Parquet column corresponds to a [[StructField]].
+ *
+ * Also prunes the unnecessary metadata columns if any for all file formats.
*/
object SchemaPruning extends Rule[LogicalPlan] {
import org.apache.spark.sql.catalyst.expressions.SchemaPruning._
- override def apply(plan: LogicalPlan): LogicalPlan =
- if (conf.nestedSchemaPruningEnabled) {
- apply0(plan)
- } else {
- plan
- }
+ override def apply(plan: LogicalPlan): LogicalPlan = apply0(plan)
private def apply0(plan: LogicalPlan): LogicalPlan =
plan transformDown {
case op @ PhysicalOperation(projects, filters,
- l @ LogicalRelation(hadoopFsRelation: HadoopFsRelation, _, _, _))
- if canPruneRelation(hadoopFsRelation) =>
-
- prunePhysicalColumns(l.output, projects, filters,
hadoopFsRelation.dataSchema,
- prunedDataSchema => {
+ l @ LogicalRelation(hadoopFsRelation: HadoopFsRelation, _, _, _)) =>
+ prunePhysicalColumns(l, projects, filters, hadoopFsRelation,
+ (prunedDataSchema, prunedMetadataSchema) => {
val prunedHadoopRelation =
hadoopFsRelation.copy(dataSchema =
prunedDataSchema)(hadoopFsRelation.sparkSession)
- buildPrunedRelation(l, prunedHadoopRelation)
+ buildPrunedRelation(l, prunedHadoopRelation, prunedMetadataSchema)
}).getOrElse(op)
}
/**
* This method returns optional logical plan. `None` is returned if no
nested field is required or
* all nested fields are required.
+ *
+ * This method will prune both the data schema and the metadata schema
*/
private def prunePhysicalColumns(
- output: Seq[AttributeReference],
+ relation: LogicalRelation,
projects: Seq[NamedExpression],
filters: Seq[Expression],
- dataSchema: StructType,
- leafNodeBuilder: StructType => LeafNode): Option[LogicalPlan] = {
+ hadoopFsRelation: HadoopFsRelation,
+ leafNodeBuilder: (StructType, StructType) => LeafNode):
Option[LogicalPlan] = {
+
val (normalizedProjects, normalizedFilters) =
- normalizeAttributeRefNames(output, projects, filters)
+ normalizeAttributeRefNames(relation.output, projects, filters)
val requestedRootFields = identifyRootFields(normalizedProjects,
normalizedFilters)
// If requestedRootFields includes a nested field, continue. Otherwise,
// return op
if (requestedRootFields.exists { root: RootField => !root.derivedFromAtt
}) {
- val prunedDataSchema = pruneDataSchema(dataSchema, requestedRootFields)
- // If the data schema is different from the pruned data schema,
continue. Otherwise,
- // return op. We effect this comparison by counting the number of "leaf"
fields in
- // each schemata, assuming the fields in prunedDataSchema are a subset
of the fields
- // in dataSchema.
- if (countLeaves(dataSchema) > countLeaves(prunedDataSchema)) {
- val prunedRelation = leafNodeBuilder(prunedDataSchema)
- val projectionOverSchema = ProjectionOverSchema(prunedDataSchema)
+ val prunedDataSchema = if (canPruneDataSchema(hadoopFsRelation)) {
+ pruneDataSchema(hadoopFsRelation.dataSchema, requestedRootFields)
+ } else {
+ hadoopFsRelation.dataSchema
+ }
+
+ val metadataSchema =
+ relation.output.collect { case MetadataAttribute(attr) => attr
}.toStructType
+ val prunedMetadataSchema = if (canPruneMetadataSchema(relation)) {
+ pruneDataSchema(metadataSchema, requestedRootFields)
+ } else {
+ metadataSchema
+ }
+ // If the data schema + metadata schema is different from
+ // the pruned data schema + pruned metadata schema, continue.
+ // Otherwise, return None.
+ if (countLeaves(hadoopFsRelation.dataSchema.merge(metadataSchema)) >
+ countLeaves(prunedDataSchema.merge(prunedMetadataSchema))) {
Review comment:
`countLeaves(hadoopFsRelation.dataSchema) >
countLeaves(prunedDataSchema) || countLeaves(metadataSchema) > ...`
--
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]
