gengliangwang commented on code in PR #58298: URL: https://github.com/apache/spark/pull/58298#discussion_r4008007054
########## 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 => + val logicalName = metadataLogicalName(captured) + matchName(currentMetadataOutput, logicalName, resolver)(metadataLogicalName) + .map(pos => currentMetadataOutput(pos)) + .getOrElse { + // The connector still reports this metadata column, so it can only be absent here + // because a data column has taken its name and the connector suppresses rather than + // renames the conflict (`canRenameConflictingMetadataColumns`). Validation owns + // rejecting that. + unexpectedSchemaChange( + s"captured metadata column $logicalName is missing from the current relation") + } + } + + val currentOutput = current.output ++ currentMetadata + + // Refresh may visit an already rebound relation. Preserve its attributes so the projection + // above it continues to reference valid expression IDs. + // + // A further schema change on such a relation adds a second projection instead of replacing + // the first. Only the cache stores a refreshed plan, so the effect is limited to that entry: + // it stops matching the single projection a query rebuilds from its own captured output, and + // is no longer reused. Results stay correct. + if (sameOutputShape(capturedOutput, currentOutput)) { + return relation + } + + val capturedIndex = new AttributeIndex(capturedOutput, resolver) + val reboundOutput = currentOutput.map { currentAttr => + capturedIndex.get(currentAttr).filter(canReuse(_, currentAttr)).getOrElse(currentAttr) + } + val reboundRelation = relation.copy(output = reboundOutput) + + val reboundIndex = new AttributeIndex(reboundOutput, resolver) + val projectList = capturedOutput.map { capturedAttr => + val currentAttr = reboundIndex.get(capturedAttr).getOrElse { + unexpectedSchemaChange( + s"captured column ${capturedAttr.name} is missing from current table ${relation.name}") + } + if (currentAttr.exprId == capturedAttr.exprId && + sameAttributeShape(currentAttr, capturedAttr)) { + currentAttr + } else { + if (currentAttr.nullable != capturedAttr.nullable) { + unexpectedSchemaChange( + s"nullability changed for captured column ${capturedAttr.name} in ${relation.name}") + } + val projected = projectToType( + currentAttr, currentAttr.dataType, capturedAttr.dataType, resolver) + if (projected.dataType != capturedAttr.dataType || + projected.nullable != capturedAttr.nullable) { + unexpectedSchemaChange( + s"failed to recreate captured column ${capturedAttr.name} in ${relation.name}") + } + Alias(projected, capturedAttr.name)( + exprId = capturedAttr.exprId, + qualifier = capturedAttr.qualifier, + explicitMetadata = Some(capturedAttr.metadata)) + } + } + + Project(projectList, reboundRelation) + } + + private[v2] def projectToType( Review Comment: Can we avoid introducing a second, DSv2-only schema-alignment implementation here? `V2TableRefreshUtil` has already validated the captured and current schemas through `V2TableUtil.validateCapturedColumns` / `validateCapturedMetadataColumns`, which delegate to `SchemaUtils.validateSchemaCompatibility(..., ALLOW_NEW_FIELDS, checkIds = true)`. `Project.matchSchema` already contains the recursive struct/array/map narrowing needed to build a projection from the fresh current relation. It is not a direct drop-in today: it merges metadata, uses resolver-only matching, cannot distinguish logically renamed metadata columns, and does not preserve captured expression IDs or qualifiers. However, those gaps argue for extracting or extending a shared strict Catalyst schema-projection helper, then applying final aliases with the captured name, exprId, qualifier, and exact metadata. A plain attribute-only Project works only for top-level additions; nested additions still need the shared recursive narrowing. Please refactor this rather than maintaining another roughly 200 lines of recursive schema-projection logic here. -- 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]
