yyanyy commented on code in PR #58298: URL: https://github.com/apache/spark/pull/58298#discussion_r4009332955
########## 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: You're right that there shouldn't be two rules in this codebase for deciding whether two names refer to the same column, and that duplication is what produced the INTERNAL_ERROR you found in the thread below. I've fixed that half. Name matching now goes through a shared `SchemaUtils.foldName`, the fold that `AttributeSeq` resolution keys on (`expressions/package.scala` looks attributes up by `name.toLowerCase(Locale.ROOT)` and only then filters candidates with the resolver) and that `validateSchemaCompatibility` and `V2TableUtil` already used. That change deleted the exact-name preference and the ambiguity branch rather than adding to them. I don't think reusing `Project.matchSchema` for the recursive part is the right direction, though. Its only caller is `Dataset.to(schema)`, and `reorderFields`/`reconcileColumnType` are deliberately lenient in ways that are each wrong here: a missing nullable field becomes a NULL literal, a type mismatch becomes an ANSI `Cast`, metadata is merged from both sides, a non-attribute expression gets a fresh `exprId`, and non-nullable to nullable widening is accepted. Rebinding needs the opposite of every one of those, because validation has already guaranteed compatibility — so any mismatch is a bug that must surface, and the captured `exprId`, qualifier and exact metadata have to be reproduced or the parent plan breaks. Extracting a "shared strict" helper means adding a strictness mode to a Catalyst helper with a single consumer, and once every divergent leaf is parameterised the shared residue is the recursive walk itself. For the record the recursive logic is 111 lines, not ~200 . -- 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]
