srielau commented on code in PR #58317:
URL: https://github.com/apache/spark/pull/58317#discussion_r4134828967
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala:
##########
@@ -1171,14 +1171,34 @@ class Analyzer(
// Resolve V2TableReference nodes inside temp view plans. These are
created by
// V2TableReference.createForTempView. We only need to resolve it when
returning
- // the plan of temp views (in resolveViews and unwrapRelationPlan).
+ // the plan of temp views (in resolveViews and unwrapRelationPlan).
Discard a
+ // stored CHAR/VARCHAR policy Project so ApplyCharTypePadding can rebind
both
+ // the scan mode and the generated expressions under the current SQLConf.
private def resolveTableReferencesInTempView(plan: LogicalPlan):
LogicalPlan = {
- plan.resolveOperatorsUp {
- case r: V2TableReference if
r.context.isInstanceOf[V2TableReference.TemporaryViewContext] =>
+ plan.transformDown {
Review Comment:
Fixed. V2 temp-view table refs now store `charVarcharScanMode` and lookup
keeps that mode plus any creation-time policy Project.
`CHAR_VARCHAR_STANDARD_SEMANTICS` stays the creation-time policy: a
SparkStandard view still raises `EXCEED_LIMIT_LENGTH` after the session
switches to PreserveNative.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/CacheManager.scala:
##########
@@ -389,20 +580,69 @@ class CacheManager extends Logging with
AdaptiveSparkPlanHelper {
}
private def tryRebuildCacheEntry(spark: SparkSession, cd: CachedData):
Option[CachedData] = {
- val sessionWithConfigsOff = getOrCloneSessionWithConfigsOff(spark)
- sessionWithConfigsOff.withActive {
- tryRefreshPlan(sessionWithConfigsOff, cd.plan).map { refreshedPlan =>
- val qe = QueryExecution.create(
- sessionWithConfigsOff,
- refreshedPlan,
- refreshPhaseEnabled = false)
- val newKey = qe.normalized
- val newCache = InMemoryRelation(cd.cachedRepresentation.cacheBuilder,
qe)
- cd.copy(plan = newKey, cachedRepresentation = newCache)
+ val mode = capturedCharVarcharScanMode(cd.plan)
+ val rebuildSession = sessionForCacheRebuild(spark, mode, cd.plan)
+ rebuildSession.withActive {
+ try {
+ tryRefreshPlan(rebuildSession, cd.plan).map { refreshedPlan =>
+ val qe = QueryExecution.create(
+ rebuildSession,
+ refreshedPlan,
+ refreshPhaseEnabled = false)
+ val newKey = qe.normalized
+ val newCache =
InMemoryRelation(cd.cachedRepresentation.cacheBuilder, qe)
+ cd.copy(plan = newKey, cachedRepresentation = newCache)
+ }
+ } catch {
+ case NonFatal(e) =>
+ logWarning(log"Failed to rebuild cache entry while attempting to
recache", e)
+ None
}
}
}
+ // Clone only the configs-off session when a mode must be applied, so
catalog plugins stay
+ // shared with the caller. CHAR/VARCHAR confs are set on that clone, never
on `spark`.
+ private def sessionForCacheRebuild(
+ spark: SparkSession,
+ mode: Option[CharVarcharScanMode],
+ plan: LogicalPlan): SparkSession = {
+ val base = getOrCloneSessionWithConfigsOff(spark)
+ mode match {
+ case None => base
+ case Some(m) =>
+ val session = if (base eq spark) spark.cloneSession() else base
+ CharVarcharScanMode.configure(
+ session.sessionState.conf,
+ m,
+ nativeCharVarcharTypes = hasNativeCharVarcharTypes(plan))
+ session
+ }
+ }
+
+ private def hasNativeCharVarcharTypes(plan: LogicalPlan): Boolean = {
+ plan.exists {
+ case r: DataSourceV2Relation =>
+ r.output.exists(a => CharVarcharUtils.hasCharVarchar(a.dataType))
+ case r: LogicalRelation =>
+ r.output.exists(a => CharVarcharUtils.hasCharVarchar(a.dataType))
+ case r: HiveTableRelation =>
+ r.output.exists(a => CharVarcharUtils.hasCharVarchar(a.dataType))
+ case _ => false
+ }
+ }
+
+ private def capturedCharVarcharScanMode(plan: LogicalPlan):
Option[CharVarcharScanMode] = {
+ plan.collectFirst {
Review Comment:
Fixed. Rebuild walks the plan once into `CacheRebuildPolicy`. If bound modes
mix, we skip `configure()` and only refresh `Table` metadata so each relation
keeps the mode already stored on that node.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/CacheManager.scala:
##########
@@ -420,12 +660,14 @@ class CacheManager extends Logging with
AdaptiveSparkPlanHelper {
try {
EliminateSubqueryAliases(plan) match {
case r @ ExtractV2CatalogAndIdentifier(catalog, ident) if
r.timeTravelSpec.isEmpty =>
- val table = CatalogV2Util.getTable(catalog, ident, options =
r.options)
- if (r.table.id == table.id) {
- Some(DataSourceV2Relation.create(table, Some(catalog),
Some(ident), r.options))
- } else {
- None
- }
+ refreshV2RelationTable(r, catalog, ident)
+ case project @ Project(_, r @ ExtractV2CatalogAndIdentifier(catalog,
ident))
+ if r.timeTravelSpec.isEmpty &&
+ r.charVarcharScanMode.exists(
+ ApplyCharTypePaddingHelper.isReadSidePaddingProject(project,
r, _)) =>
+ // Drop the analyzer-generated padding Project.
QueryExecution.create re-analyzes
+ // the refreshed relation so the Project matches the new schema and
scan mode.
+ refreshV2RelationTable(r, catalog, ident)
Review Comment:
Fixed. The rebuild session now sets `READ_SIDE_CHAR_PADDING` from the cached
plan together with the scan mode, so a PreserveNative padded key cannot
collapse into the unpadded relation key when mutation runs with padding
disabled.
##########
sql/core/src/test/scala/org/apache/spark/sql/CachedTableSuite.scala:
##########
@@ -2347,14 +2348,36 @@ class CachedTableSuite extends SharedSparkSession
}
}
+ // SPARK-58814 CHAR/VARCHAR cache invariant coverage (scope freeze):
+ // In scope: two bound identities (PreserveNative, SparkStandard) and the
analyzer
+ // padding Project. This covers capture, identity, mutation matching, and
replay for those cases.
Review Comment:
Added `PreserveNative recache keeps independent read-side padding`. It
caches under each padding setting, recaches with the other setting, then checks
cache use and the padding Project through a fresh DataFrame after restoring the
original setting.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/CacheManager.scala:
##########
@@ -389,20 +580,69 @@ class CacheManager extends Logging with
AdaptiveSparkPlanHelper {
}
private def tryRebuildCacheEntry(spark: SparkSession, cd: CachedData):
Option[CachedData] = {
- val sessionWithConfigsOff = getOrCloneSessionWithConfigsOff(spark)
- sessionWithConfigsOff.withActive {
- tryRefreshPlan(sessionWithConfigsOff, cd.plan).map { refreshedPlan =>
- val qe = QueryExecution.create(
- sessionWithConfigsOff,
- refreshedPlan,
- refreshPhaseEnabled = false)
- val newKey = qe.normalized
- val newCache = InMemoryRelation(cd.cachedRepresentation.cacheBuilder,
qe)
- cd.copy(plan = newKey, cachedRepresentation = newCache)
+ val mode = capturedCharVarcharScanMode(cd.plan)
Review Comment:
Fixed. Mode, mixed-mode, native CHAR/VARCHAR types, and read-side padding
are collected in one `cacheRebuildPolicy` walk.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/CharVarcharScanMode.scala:
##########
@@ -0,0 +1,89 @@
+/*
+ * 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
+
+import org.apache.spark.sql.internal.SQLConf
+
+/**
+ * The CHAR/VARCHAR scan mode bound to a relation (and its scan) during
analysis.
+ *
+ * A relation carries `Option[CharVarcharScanMode]`: `None` means no mode was
bound, including for
+ * a relation with no CHAR/VARCHAR columns. A `Some` value pins the mode so
that `sameResult`
+ * comparisons and cache reuse keep the two variants distinct.
+ */
+private[sql] sealed trait CharVarcharScanMode
+
+/**
+ * A scan builder that accepts the CHAR/VARCHAR mode captured during relation
analysis.
+ *
+ * For example, if `SELECT c FROM t` is analyzed with standard semantics
enabled, the relation
+ * binds [[CharVarcharScanMode.SparkStandard]] before the builder creates the
physical scan.
+ */
+private[sql] trait SupportsCharVarcharScanMode {
+ def bindCharVarcharScanMode(mode: CharVarcharScanMode): Unit
+}
+
+private[sql] 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 readers that honor this mode so Spark
observes the original
+ * value and applies standard CHAR/VARCHAR length checks. Corresponds to
standard semantics.
+ * Native Hive ORC with CONVERT_METASTORE_ORC=false does not implement this
contract and still
+ * applies native CHAR/VARCHAR truncation.
+ */
+ case object SparkStandard extends CharVarcharScanMode
+
+ /**
+ * Maps the boolean `spark.sql.charVarchar.standardSemantics.enabled` value
to the typed mode.
+ */
+ def apply(standardSemantics: Boolean): CharVarcharScanMode =
+ if (standardSemantics) SparkStandard else PreserveNative
+
+ /**
+ * Configures `conf` so analysis binds `mode` and generates the matching
read-side Project.
+ *
+ * Both flags are set so a cloned session cannot inherit the caller's
PRESERVE value.
+ * SparkStandard defaults to PRESERVE=false; pass `nativeCharVarcharTypes =
true` when the
+ * cached plan kept CHAR/VARCHAR data types (created with PRESERVE=true and
STANDARD=true).
+ */
+ def configure(
+ conf: SQLConf,
+ mode: CharVarcharScanMode,
+ nativeCharVarcharTypes: Boolean = false): Unit = mode match {
+ case SparkStandard =>
+ conf.setConfString(SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key, "true")
+ conf.setConfString(
+ SQLConf.PRESERVE_CHAR_VARCHAR_TYPE_INFO.key,
+ nativeCharVarcharTypes.toString)
+ case PreserveNative =>
+ conf.setConfString(SQLConf.PRESERVE_CHAR_VARCHAR_TYPE_INFO.key, "true")
+ conf.setConfString(SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key, "false")
+ }
+
+ /** Parses a mode from its `toString` name; the inverse of
[[CharVarcharScanMode.toString]]. */
Review Comment:
Updated the scaladoc to name `PreserveNative.toString` and
`SparkStandard.toString` as the accepted `fromName` inputs.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/ApplyCharTypePadding.scala:
##########
@@ -51,17 +51,55 @@ object ApplyCharTypePadding extends Rule[LogicalPlan] {
}
override def apply(plan: LogicalPlan): LogicalPlan = {
+ val standardSemantics = conf.charVarcharStandardSemantics
+ val scanMode = CharVarcharScanMode(standardSemantics)
+
+ /**
+ * Binds the mode captured while analyzing a first-class CHAR/VARCHAR
relation.
+ *
+ * For example, `SELECT c FROM t` retains its analyzed mode if the session
setting changes
+ * before the physical scan is built. Relations without CHAR/VARCHAR
columns remain unbound.
+ */
+ def bindCharVarcharScanMode(p: LogicalPlan): LogicalPlan = p match {
+ case relation: LogicalRelation
+ if relation.charVarcharScanMode.isEmpty && relation.hasCharVarchar =>
+ val bound = relation.copy(charVarcharScanMode = Some(scanMode))
+ bound.copyTagsFrom(relation)
+ bound
+ case relation: DataSourceV2Relation
+ if relation.charVarcharScanMode.isEmpty && relation.hasCharVarchar =>
+ val bound = relation.copy(charVarcharScanMode = Some(scanMode))
+ bound.copyTagsFrom(relation)
+ bound
+ case relation: HiveTableRelation
+ if relation.charVarcharScanMode.isEmpty && relation.hasCharVarchar =>
+ val bound = relation.copy(charVarcharScanMode = Some(scanMode))
+ bound.copyTagsFrom(relation)
+ bound
+ case _ => p
+ }
+
+ val boundPlan = if (conf.charVarcharFirstClassTypes) {
Review Comment:
Fixed. Binding now happens on the same bottom-up padding traversal,
immediately before that node's padding decision. The extra plan-wide bind pass
is gone.
--
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]
