cloud-fan commented on code in PR #58317:
URL: https://github.com/apache/spark/pull/58317#discussion_r4129485024
##########
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:
**Non-blocking (P2):** This refresh arm discards the analyzer-generated
padding Project, but the rebuild session restores only CharVarcharScanMode. For
PreserveNative, READ_SIDE_CHAR_PADDING independently decides whether that
Project is recreated. If padded and unpadded variants are cached and mutation
runs with padding disabled, both rebuild to the bare relation key; one entry is
then treated as already cached and disappears. Capture and replay the padding
policy with the relation mode so each variant keeps its key and storage level.
See **Shared repair plan 1** in the review body.
##########
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:
**Nit (P3):** The rebuild path has just traversed cd.plan to capture its
mode, then hasNativeCharVarchar walks the same plan again. Both properties can
be accumulated in one traversal while preserving the existing first-mode
ordering and replay eligibility.
**Verification:**
- **Inspection:** Confirm the rebuild path has one plan-wide property
collection and no fallback path repeats either full traversal.
- **Behavior:** Verify cache replay still selects the same eligible entries
and preserves their results.
##########
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:
**Nit (P3):** This link denotes the companion object's inherited toString,
not the names accepted by fromName. Please name or link PreserveNative.toString
and SparkStandard.toString explicitly so generated documentation points to the
actual contract.
**Verification:**
- **Inspection:** Confirm generated documentation resolves each referenced
symbol to a concrete CharVarcharScanMode value and accurately states the
accepted names.
##########
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:
**Nit (P3):** This preliminary transform is used only by the padding
transform that follows it. Bind each relation when the existing bottom-up
traversal visits that node, preserving child-before-parent ordering while
avoiding the additional full-plan pass.
**Verification:**
- **Inspection:** Confirm the rule contains no separate plan-wide binding
pass and binds relations before their padding decision is consumed.
- **Behavior:** Verify first-class CHAR/VARCHAR analysis produces the same
bound modes and padded outputs with padding enabled and disabled.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala:
##########
@@ -202,19 +203,31 @@ case class AlterTableRenameCommand(
} else {
val table = catalog.getTableMetadata(oldName)
DDLUtils.verifyAlterTableType(catalog, table, isView)
- // If `optStorageLevel` is defined, the old table was cached.
- val optCachedData =
sparkSession.sharedState.cacheManager.lookupCachedData(
- sparkSession.table(oldName.unquotedString))
- val optStorageLevel =
optCachedData.map(_.cachedRepresentation.cacheBuilder.storageLevel)
- if (optStorageLevel.isDefined) {
+ val cacheManager = sparkSession.sharedState.cacheManager
+ val resolver = sparkSession.sessionState.conf.resolver
+ val lookedUp = (
+ cacheManager.lookupDirectCacheDescriptorsByName(oldName.nameParts,
resolver) ++
Review Comment:
**Nit (P3):** These adjacent lookups inspect the same cache snapshot and
differ only by the name predicate. Evaluate both predicates in one ordered fold
and retain separate result groups so ordering, deduplication, storage levels,
and restoration behavior stay unchanged.
**Verification:**
- **Inspection:** Confirm rename descriptor discovery evaluates qualified
and unqualified predicates within one registry traversal.
- **Behavior:** Verify direct caches addressed by either spelling are
restored once with their original storage levels and dependent caches remain
excluded.
##########
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:
**Blocking (P1):** This rewrite makes a resolved V2 temporary view adopt the
querying session's current CHAR/VARCHAR policy, even though
CHAR_VARCHAR_STANDARD_SEMANTICS is declared PERSISTED. A view created under
SparkStandard can therefore be queried under PreserveNative and silently
truncate an overlength VARCHAR instead of retaining its creation-time error.
Refresh the table metadata while carrying forward the stored mode and matching
policy Project.
**Recommended change:** Retain the creation-time CHAR/VARCHAR scan mode and
padding semantics when refreshing V2 temporary-view references, with symmetric
cross-configuration regression coverage.
**Why this works:** Refresh the underlying V2 table reference and output
metadata while carrying forward the stored mode and its analyzer-generated
policy structure, rather than clearing the mode and reapplying the query
session's current configuration.
**Scope:** Align V2 temporary-view restoration with persisted configuration
semantics and the V1/permanent-view behavior.
**Compatibility:** Preserve catalog metadata refresh and the declared
persisted configuration contract.
**Risks:** Keeping stale output attributes while preserving policy could
break schema refresh for a changed V2 table.
**Constraints:** Table and schema refresh must continue to observe current
catalog metadata. V1 and permanent-view persistence behavior must remain
unchanged.
**Success:** A V2 temporary view returns the same CHAR/VARCHAR result or
length error regardless of later caller configuration. V2 temporary views
continue to refresh current table metadata without losing creation-time policy.
**Verification:**
- **Regression:** Create V2 temporary views under each bound mode, query
under the opposite mode, and verify creation-time truncation, padding, or
length-check behavior in both directions.
- **Compatibility:** Verify schema and table-reference refresh still works
while V1, V2, and permanent-view persisted-policy behavior remains aligned.
##########
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:
**Non-blocking (P2):** The claimed replay coverage never captures a
PreserveNative cache with read-side padding disabled, rebuilds it while padding
is enabled, and then checks cache use and results through a fresh DataFrame
after restoring the original setting. Add that inversion in both directions so
changing or collapsing the padding key necessarily fails the test.
See **Shared repair plan 1** in the review body.
##########
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:
**Blocking (P1):** A cached composite plan can contain relations already
bound under different modes, but collectFirst selects one mode for the entire
rebuild session. Refreshing the remaining relations under that global setting
can turn PreserveNative behavior into SparkStandard length checks, or remove
those checks in the reverse order. Preserve mode per relation or subtree during
refresh instead of globalizing the first one.
See **Shared repair plan 1** in the review body.
--
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]
