cloud-fan commented on code in PR #58317:
URL: https://github.com/apache/spark/pull/58317#discussion_r4107563513
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2Suite.scala:
##########
@@ -505,6 +506,51 @@ class DataSourceV2Suite extends SharedSparkSession with
AdaptiveSparkPlanHelper
}
}
+ test("SPARK-58814: catalog-less V2 writes recache both CHAR/VARCHAR scan
modes") {
+ val format = classOf[CharVarcharWritableDataSource].getName
+ val preserveConf = Seq(
+ SQLConf.PRESERVE_CHAR_VARCHAR_TYPE_INFO.key -> "true",
+ SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "false")
+ val standardConf = Seq(SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key ->
"true")
+
+ withTempPath { file =>
+ val path = file.getCanonicalPath
+ def readData: DataFrame = spark.read.format(format).option("path",
path).load()
+ def appendData(c: String, v: String): Unit = {
+ Seq((c, v)).toDF("c", "v").write
+ .format(format).option("path", path).mode("append").save()
+ }
+
+ appendData("a", "x")
+ val preserveRead = withSQLConf(preserveConf: _*) {
+ val read = readData.persist(MEMORY_ONLY)
+ checkAnswer(read, Row("a", "x"))
+ read
+ }
+ val standardRead = withSQLConf(standardConf: _*) {
+ val read = readData.persist(DISK_ONLY)
+ checkAnswer(read, Row("a", "x"))
+ read
+ }
+ try {
+ appendData("b", "y")
+ withSQLConf(preserveConf: _*) {
+ checkAnswer(readData, Seq(Row("a", "x"), Row("b", "y")))
Review Comment:
**Non-blocking (P2):** These assertions execute a newly constructed
`readData`, not the cached `preserveRead` and `standardRead` DataFrames.
Because `getTable` returns a fresh identity-based `CharVarcharTable`, a broken
`recacheByV2Relation` can match no retained entry while the new read still sees
the appended files; the storage-level lookup also does not prove that the
cached contents were refreshed. Please run `checkAnswer` on the original cached
DataFrames after the append, and do the same in the catalog-less streaming
test, so stale entries make the regression fail.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/CacheManager.scala:
##########
@@ -422,7 +523,9 @@ class CacheManager extends Logging with
AdaptiveSparkPlanHelper {
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))
+ Some(DataSourceV2Relation
Review Comment:
**Non-blocking (P2):** This restores only the relation's
`charVarcharScanMode`; the refreshed plan is still analyzed under the mutating
session's SQLConf. If a PreserveNative cache is rebuilt by a SparkStandard
write or refresh, the child remains PreserveNative while `ApplyCharTypePadding`
can generate a SparkStandard read-side Project, leaving a plan that matches
neither intended cache variant. Please replay the captured policy as a unit,
for example by analyzing the full variant in an isolated session configured for
that mode, so the relation and generated Project stay coherent.
See **Shared repair plan 1** in the review body.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RelationResolution.scala:
##########
@@ -591,7 +594,10 @@ class RelationResolution(
cached transform {
case r: DataSourceV2Relation if matchesReference(r, ref) =>
V2TableReferenceUtils.validateLoadedTable(r.table, ref)
- r.copy(output = ref.output, options = ref.options)
+ r.copy(
+ output = ref.output,
+ options = ref.options,
+ charVarcharScanMode = None)
Review Comment:
**Blocking (P1):** Clearing only the reloaded child's mode does not rebind
the analyzer-generated read-side `Project` retained by the temporary view. A
view created under PreserveNative can keep its preserve projection while a
SparkStandard lookup rebinds the child to a physical STRING scan; because the
first projection already cleaned the child metadata, `ApplyCharTypePadding`
cannot synthesize the missing `varcharTypeReadSideCheck`, and `abcdef` can
escape from VARCHAR(4) instead of raising `EXCEED_LIMIT_LENGTH`. Please
reconstruct the relation and generated Project together under the current
policy, or preserve both together as one captured policy.
See **Shared repair plan 1** in the review body.
##########
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:
Confirmed: the insert regression now uses a CHAR/VARCHAR relation, binds
both modes, retains both caches through overwrite, and checks their replacement
rows.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:3962201895","thread_id":"inline:3962201895","verdict_sha256":"1821e60da6a15b8042b8aae53c3feec24e342c5cf8056261204a976d216fdfa6"}
-->
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/CharVarcharScanMode.scala:
##########
@@ -0,0 +1,64 @@
+/*
+ * 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 no mode was
bound, including for
+ * a relation with no CHAR/VARCHAR columns. A `Some` value pins the mode so
that `sameResult` /
+ * 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 the source so Spark observes the original
value and applies
Review Comment:
**Nit (P3):** `SparkStandard` is also bound to `HiveTableRelation`, but with
`CONVERT_METASTORE_ORC=false` the native reader does not consume it; the new
Hive test explicitly expects native truncation to `"abcd"`. This comment
therefore overstates what binding the mode guarantees. Please qualify the
physical-STRING and Spark-side length-check behavior to mode-aware readers and
document the native Hive exception.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/CacheManager.scala:
##########
@@ -348,6 +349,47 @@ class CacheManager extends Logging with
AdaptiveSparkPlanHelper {
recacheByCondition(spark, _.plan.exists(_.sameResult(normalized)))
}
+ /**
+ * Re-caches every entry whose plan contains a [[LogicalRelation]] for
`relation`.
+ * Unlike [[recacheByPlan]], this ignores CHAR/VARCHAR scan-mode identity so
a V1 write
+ * invalidates preserve-only, standard, and unbound cache entries for that
BaseRelation.
+ */
+ def recacheByV1Relation(spark: SparkSession, relation: BaseRelation): Unit =
{
+ recacheByCondition(spark, cd => cd.plan.exists {
+ case logical: LogicalRelation => logical.relation == relation
+ case _ => false
+ })
+ }
+
+ /**
+ * Re-caches every entry whose plan contains the given catalog-less
[[DataSourceV2Relation]].
+ * The scan mode is ignored only for this mutation-specific match so an
unbound write target
+ * invalidates preserve-native and standard cache entries without weakening
normal cache identity.
+ */
+ def recacheByV2Relation(spark: SparkSession, relation:
DataSourceV2Relation): Unit = {
+ val unboundRelation = relation.copy(charVarcharScanMode = None)
+ recacheByCondition(spark, cd => cd.plan.exists {
+ case cached: DataSourceV2Relation =>
+ cached.copy(charVarcharScanMode = None).sameResult(unboundRelation)
+ case _ => false
+ })
+ }
+
+ /**
+ * Looks up direct cache entries for a V2 table mutation while ignoring only
their analyzed
+ * CHAR/VARCHAR scan mode. Normal cache substitution remains mode-sensitive.
+ */
+ def lookupCachedDataByV2Relation(relation: DataSourceV2Relation):
Seq[CachedData] = {
+ val unboundRelation = relation.copy(charVarcharScanMode = None)
+ cachedData.filter { cd =>
+ EliminateSubqueryAliases(cd.plan) match {
Review Comment:
Confirmed: direct named V2 cache discovery now recognizes the
analyzer-generated CHAR/VARCHAR Project and matches the contained relation
without depending on the cache display name.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:4009041874","thread_id":"inline:4009041874","verdict_sha256":"1821e60da6a15b8042b8aae53c3feec24e342c5cf8056261204a976d216fdfa6"}
-->
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/RenameTableExec.scala:
##########
@@ -49,9 +50,11 @@ case class RenameTableExec(
} else newIdent
catalog.renameTable(oldIdent, qualifiedNewIdent)
Review Comment:
Confirmed: rename now loads the destination table once when descriptors
exist and reuses that table for every restored cache variant.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:4009041896","thread_id":"inline:4009041896","verdict_sha256":"1821e60da6a15b8042b8aae53c3feec24e342c5cf8056261204a976d216fdfa6"}
-->
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/CacheManager.scala:
##########
@@ -422,7 +464,9 @@ class CacheManager extends Logging with
AdaptiveSparkPlanHelper {
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))
+ Some(DataSourceV2Relation
+ .create(table, Some(catalog), Some(ident), r.options)
+ .copy(charVarcharScanMode = r.charVarcharScanMode))
Review Comment:
Thanks; SPARK-59751 records the explicit Legacy identity and three-state
lifecycle work. This review is treating that baseline behavior separately from
PR-attributable defects in the two bound modes.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:4015284288","thread_id":"inline:4015284288","verdict_sha256":"1821e60da6a15b8042b8aae53c3feec24e342c5cf8056261204a976d216fdfa6"}
-->
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/DataSourceScanExec.scala:
##########
@@ -715,6 +715,9 @@ trait FileSourceScanLike extends DataSourceScanExec with
SessionStateHelper {
* @param tableIdentifier Identifier for the table in the metastore.
* @param disableBucketedScan Disable bucketed scan based on physical query
plan, see rule
* [[DisableUnnecessaryBucketedScan]] for details.
+ * @param charVarcharScanMode Analyzed CHAR/VARCHAR scan mode. Compared by
+ * sameResult so preserve-only and standard scans
are
+ * not reused. None means unbound (native ORC
types).
Review Comment:
Confirmed: the scan documentation now states the generic overload contract
first and limits native constrained versus physical STRING decoding to ORC.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:4015284297","thread_id":"inline:4015284297","verdict_sha256":"1821e60da6a15b8042b8aae53c3feec24e342c5cf8056261204a976d216fdfa6"}
-->
--
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]
