yyanyy commented on code in PR #58298: URL: https://github.com/apache/spark/pull/58298#discussion_r4009399413
########## sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/CapturedSchemaProjection.scala: ########## @@ -0,0 +1,310 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.spark.sql.execution.datasources.v2 + +import org.apache.spark.SparkException +import org.apache.spark.sql.catalyst.SQLConfHelper +import org.apache.spark.sql.catalyst.analysis.Resolver +import org.apache.spark.sql.catalyst.expressions.{Alias, ArrayTransform, AttributeReference, CreateNamedStruct, Expression, GetStructField, If, IsNull, KnownNotNull, LambdaFunction, Literal, MetadataAttributeWithLogicalName, NamedLambdaVariable, TaggingExpression, TransformKeys, TransformValues, UnresolvedNamedLambdaVariable} +import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, Project} +import org.apache.spark.sql.catalyst.util.MetadataColumnHelper +import org.apache.spark.sql.types.{ArrayType, DataType, MapType, Metadata, StructType} + +/** + * Rebinds a relation that reads a current table schema to output attributes captured from an + * earlier compatible schema. The current schema is exposed by the relation so its output remains + * aligned with the physical scan, while a projection recreates the captured output for the + * already-analyzed parent plan. + */ +private[sql] object CapturedSchemaProjection extends SQLConfHelper { + + /** + * Prevents [[CreateNamedStruct]] from inheriting metadata from a field value while leaving the + * value's type, nullability, evaluation, and code generation unchanged. + */ + private case class MetadataPropagationBarrier(child: Expression) extends TaggingExpression { + override protected def withNewChildInternal( + newChild: Expression): MetadataPropagationBarrier = copy(child = newChild) + } + + def rebindToCapturedSchema(relation: DataSourceV2Relation): LogicalPlan = { + // The relation still carries the output captured at analysis time; only its table has been + // swapped for the current one. + val capturedOutput = relation.output + val resolver = conf.resolver + val current = DataSourceV2Relation.create( + relation.table, + relation.catalog, + relation.identifier, + relation.options, + relation.timeTravelSpec) + val currentMetadataOutput = current.metadataOutput + val currentMetadata = capturedOutput.filter(_.isMetadataCol).map { captured => Review Comment: I'll drop the stronger claim I made earlier: you're right that the contract doesn't literally forbid a partially-pruning source from retaining metadata it wasn't asked for, though it would have to report a metadata column nobody requested, since metadata columns are not in `table.schema()`. Two reasons I'd still leave this out. First, it isn't a regression. Metadata attributes reach `relation.output` only when the query referenced them, and that is true both before and after this change — the refreshed relation carries the captured (now re-matched) metadata attributes, exactly the set the pre-change relation carried. So a read schema containing an unrequested metadata column fails in `toOutputAttrs` identically either way. This differs from the data-column case the PR does fix, where the captured output genuinely could not describe a newly added column. Second, exposing all of `current.metadataOutput` breaks the unchanged-schema fast path. `currentOutput` is compared against `capturedOutput` at the top of `rebindToCapturedSchema` to return the relation untouched when nothing changed. `capturedOutput` holds only the referenced metadata attributes while `current.metadataOutput` holds every declared one, so for a table declaring, say, `_partition`, `_file` and `_index` and a query referencing none of them, the two would differ by three attributes even when the schema is identical — and every refresh would insert a projection it does not need. That changes plan shape, and therefore cache keys, for every table with metadata columns, including SPARK-54424 reuse. Making this correct means reworking that comparison as well (data columns by shape, captured metadata separately), which is no longer a small change. Given that and the fact that nothing here regresses, hardening `toOutputAttrs` itself for all DSv2 relations still looks like the right separate change. -- 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]
