cloud-fan commented on code in PR #58317: URL: https://github.com/apache/spark/pull/58317#discussion_r3962201885
########## sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/CharVarcharScanMode.scala: ########## @@ -0,0 +1,52 @@ +/* + * 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.catalyst.util + +/** + * The CHAR/VARCHAR scan mode bound to a relation (and its scan) during analysis. + * + * A relation carries `Option[CharVarcharScanMode]`: `None` means the relation was not analyzed + * under first-class CHAR/VARCHAR types (native reader behavior), while a `Some` value pins the + * mode so that `sameResult` / cache reuse keep the two variants distinct. + */ +sealed trait CharVarcharScanMode + +object CharVarcharScanMode { + /** + * Preserve the native, constrained CHAR/VARCHAR types of the source (e.g. native ORC + * padding/truncation). Corresponds to preserve-only semantics. + */ + case object PreserveNative extends CharVarcharScanMode + + /** + * Request physical STRING from the source so Spark observes the original value and applies + * standard CHAR/VARCHAR length checks. Corresponds to standard semantics. + */ + case object SparkStandard extends CharVarcharScanMode + + /** Maps the boolean `spark.sql.charVarcharStandardSemantics` value to the typed mode. */ + def apply(standardSemantics: Boolean): CharVarcharScanMode = + if (standardSemantics) SparkStandard else PreserveNative + + /** Parses a mode from its `toString` name; the inverse of [[CharVarcharScanMode.toString]]. */ + def fromName(name: String): CharVarcharScanMode = name match { + case "PreserveNative" => PreserveNative + case "SparkStandard" => SparkStandard + case other => throw new IllegalArgumentException(s"Unknown CharVarcharScanMode: $other") Review Comment: **Nit (P3):** This names a configuration that does not exist. The registered key is `spark.sql.charVarchar.standardSemantics.enabled`; please use that exact name here. ########## sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Relation.scala: ########## @@ -116,7 +116,10 @@ case class DataSourceV2Relation( catalog: Option[CatalogPlugin], identifier: Option[Identifier], options: CaseInsensitiveStringMap, - timeTravelSpec: Option[TimeTravelSpec] = None) + timeTravelSpec: Option[TimeTravelSpec] = None, + // Bound at analysis so sameResult / cache reuse distinguish preserve-only vs standard + // CHAR/VARCHAR scans. None means the relation was not analyzed under first-class types. + charVarcharScanMode: Option[CharVarcharScanMode] = None) Review Comment: **Blocking (P1):** Because this mode is analysis-bound, retaining it on a relation returned by the SharedState relation cache lets a fresh analysis inherit the mode that populated the cache. For example, a relation cached under `PreserveNative` can remain preserve-bound when a later session analyzes it under `SparkStandard`, allowing an over-length ORC value to appear truncated rather than raising `EXCEED_LIMIT_LENGTH`. Please clear the mode specifically when adapting a shared cached table relation so the current analysis rebinds it, while preserving the mode stored in persisted view plans, and add a same-SharedState mode-switch regression. **Recommended change:** Rebind relations obtained from the shared table-relation cache to the current analysis mode. **Why this works:** Clear charVarcharScanMode only on RelationResolution's adaptation of a SharedState cache hit, before ApplyCharTypePadding binds the current SQLConf mode. **Scope:** The shared V2 table-relation cache adaptation and a focused cross-session or cross-configuration cache regression. **Compatibility:** Keep ordinary cached-plan equality mode-sensitive and preserve analysis-bound mode state embedded in persisted view plans. **Risks:** Clearing the field on persisted view plans would regress their caller-independent semantics. Making normal cache equality mode-insensitive would permit reuse of incompatible cached results. **Constraints:** Limit rebinding to relations obtained from the SharedState table-relation cache. Exercise two analyses sharing SharedState with different modes and an over-length ORC value. **Success:** A fresh SparkStandard analysis cannot inherit PreserveNative from an earlier shared cache entry and raises EXCEED_LIMIT_LENGTH, while persisted views retain their analyzed mode. ########## sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/ApplyCharTypePadding.scala: ########## @@ -51,34 +51,73 @@ object ApplyCharTypePadding extends Rule[LogicalPlan] { } override def apply(plan: LogicalPlan): LogicalPlan = { + val standardSemantics = conf.charVarcharStandardSemantics + val scanMode = CharVarcharScanMode(standardSemantics) + + // Bind into case-class state, not a TreeNodeTag: tags do not participate in structural plan + // equality / sameResult, so cache lookup and scan reuse would treat preserve-only and standard + // scans as the same plan. A case-class field does participate. Keep an already-bound value + // (views, catalog-cached relations) unchanged. + def bindStandardSemantics(p: LogicalPlan): LogicalPlan = p match { + case relation: LogicalRelation if relation.charVarcharScanMode.isEmpty => + val bound = relation.copy(charVarcharScanMode = Some(scanMode)) + bound.copyTagsFrom(relation) + bound + case relation: DataSourceV2Relation if relation.charVarcharScanMode.isEmpty => + val bound = relation.copy(charVarcharScanMode = Some(scanMode)) + bound.copyTagsFrom(relation) + bound + case relation: HiveTableRelation if relation.charVarcharScanMode.isEmpty => + val bound = relation.copy(charVarcharScanMode = Some(scanMode)) + bound.copyTagsFrom(relation) + bound + case _ => p + } + + val boundPlan = if (conf.charVarcharFirstClassTypes) { + plan.resolveOperatorsUp { + case relation: LogicalRelation => bindStandardSemantics(relation) + case relation: DataSourceV2Relation => bindStandardSemantics(relation) + case relation: HiveTableRelation => bindStandardSemantics(relation) + } + } else { + plan + } + // standardSemantics takes precedence over legacy charVarcharAsString. - if (conf.charVarcharAsString && !conf.charVarcharStandardSemantics) { - return plan + if (conf.charVarcharAsString && !standardSemantics) { + return boundPlan } - if (conf.charVarcharStandardSemantics && !conf.readSideCharPadding) { + if (standardSemantics && !conf.readSideCharPadding) { warnReadSidePaddingOverride() } - if (conf.readSideCharPadding || conf.charVarcharStandardSemantics) { - val newPlan = plan.resolveOperatorsUpWithNewOutput { + if (conf.readSideCharPadding || standardSemantics) { + val newPlan = boundPlan.resolveOperatorsUpWithNewOutput { case r: LogicalRelation => + bindStandardSemantics(r) Review Comment: **Nit (P3):** The result of this pure helper is discarded, so an unbound relation is copied and has its tags copied without affecting the plan. The same applies to the standalone calls in the V2 and Hive cases below. Please remove all three; the initial binding traversal and the cleaned-relation thunks already retain the returned nodes that are needed. ########## sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/InsertIntoDataSourceCommand.scala: ########## @@ -45,7 +45,8 @@ case class InsertIntoDataSourceCommand( // Re-cache all cached plans(including this relation itself, if it's cached) that refer to this // data source relation. - sparkSession.sharedState.cacheManager.recacheByPlan(sparkSession, logicalRelation) + sparkSession.sharedState.cacheManager.recacheByV1Relation( Review Comment: **Non-blocking (P2):** Please exercise this changed caller with a materialized cache under both `PreserveNative` and `SparkStandard`. `InsertSuite` currently tests this command only with an unbound relation, while the new two-mode regression covers the sibling `SaveIntoDataSourceCommand`. The focused test should perform the overwrite, verify that the query remains cached, and verify that it returns only the replacement rows; that will fail if this path regresses to mode-sensitive plan matching. ########## sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileFormat.scala: ########## @@ -165,6 +166,34 @@ trait FileFormat { } } + /** + * Same as [[buildReaderWithPartitionValues]] but also carries the analyzed CHAR/VARCHAR scan + * mode. [[FileSourceScanExec]] calls this overload whenever the relation has a bound mode, + * regardless of the concrete file format. + * + * The default implementation bridges the mode across the legacy seven-argument signature: it + * clones the per-call Hadoop configuration, writes an engine-private entry with the explicit + * mode, then invokes the seven-argument method virtually. A format that honors first-class + * CHAR/VARCHAR types (e.g. [[org.apache.spark.sql.execution.datasources.orc.OrcFileFormat]]) + * reads that entry in its seven-argument override, so existing subclasses keep their override + * and a call to `super` retains the bound mode. The Hadoop entry is only a transport across the + * legacy signature; the authoritative state remains the typed plan field and this parameter. + */ + def buildReaderWithPartitionValues( Review Comment: **Non-blocking (P2):** This overload is an internal planner bridge, but leaving it public enlarges the supported custom `FileFormat` surface and can make an existing untyped eta-expansion of `buildReaderWithPartitionValues` ambiguous. Please make this overload, `CHAR_VARCHAR_SCAN_MODE`, and `CharVarcharScanMode` package-private to Spark SQL. Internal callers retain access, and the default implementation can still dispatch through a third-party subclass's existing seven-argument override. ########## sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcUtils.scala: ########## @@ -528,18 +541,24 @@ object OrcUtils extends Logging { * @param resultSchema Result data schema created after pruning cols. * @param partitionSchema Schema of partitions. * @param conf Hadoop Configuration. + * @param charVarcharStandardSemantics When true, request physical ORC STRING so Spark can + * apply CHAR/VARCHAR length checks. When false, keep native + * constrained ORC CHAR/VARCHAR types. * @return Returns the result schema as string. */ def orcResultSchemaString( canPruneCols: Boolean, dataSchema: StructType, resultSchema: StructType, partitionSchema: StructType, - conf: Configuration): String = { + conf: Configuration, + charVarcharStandardSemantics: Boolean): String = { val resultSchemaString = if (canPruneCols) { - OrcUtils.getOrcSchemaString(resultSchema) + OrcUtils.getOrcSchemaString(resultSchema, charVarcharStandardSemantics) } else { - OrcUtils.getOrcSchemaString(StructType(dataSchema.fields ++ partitionSchema.fields)) + OrcUtils.getOrcSchemaString( Review Comment: **Non-blocking (P2):** The new CHAR/VARCHAR matrix reaches only the named-field (`canPruneCols=true`) path, so this changed full-schema branch has no standard-semantics regression. Please add an `OrcSourceSuite` case using forced positional evolution or an all-`_col` physical schema. It should prove that an in-range CHAR is padded and an over-length positional STRING raises `EXCEED_LIMIT_LENGTH` rather than being truncated natively by ORC. ########## sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Strategy.scala: ########## @@ -72,7 +72,7 @@ class DataSourceV2Strategy(session: SparkSession) extends Strategy with Predicat val nameParts = ident.toQualifiedNameParts(catalog) cacheManager.recacheTableOrView(session, nameParts, includeTimeTravel = false) case _ => - cacheManager.recacheByPlan(session, r) + cacheManager.recacheByV2Relation(session, r) Review Comment: **Non-blocking (P2):** This rename path still discovers the old cache entry with normal `sameResult` semantics. Under either first-class mode the cached relation is bound to `Some(mode)`, while `invalidateTableCache` constructs an unbound probe, so the lookup returns no storage level and the renamed table is left uncached. Please use the mutation-specific comparison that ignores only `charVarcharScanMode` here, retain the existing table/time-travel restrictions, and run the rename-cache regression under both bound modes. **Recommended change:** Use mode-insensitive relation matching only for V2 rename cache discovery. **Why this works:** Route invalidateTableCache's lookup through a mutation-specific relation comparison that ignores charVarcharScanMode while retaining table identity and time-travel exclusions. **Scope:** DataSourceV2Strategy's rename invalidation path and the existing V2 rename-cache regression under both bound modes. **Compatibility:** Preserve ordinary mode-sensitive cache equality and preserve the cached table's prior storage level across rename. **Risks:** A matcher that ignores more than the scan mode could select an unrelated or time-travel cache entry. Changing normal lookupCachedData semantics would allow incompatible read caches to match. **Constraints:** Keep the relaxed comparison local to mutation-specific rename discovery. Retain current table identity and time-travel filtering. **Success:** Renaming a cached V2 table under PreserveNative or SparkStandard restores it under the new identifier at the prior storage level. -- 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]
