srielau commented on code in PR #58317:
URL: https://github.com/apache/spark/pull/58317#discussion_r4112272174
##########
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:
Addressed in 85be94c8b98.
The catalog-less recache test now persists both bound DataFrames, appends,
and `checkAnswer`s those retained frames (first collect after recache, under
the matching scan-mode conf so the VARCHAR encoder is valid). Storage-level
lookups remain. Mutation matching ignores Table identity and extra write
options and uses catalog/identifier or catalog-less name plus path, so
`getTable()` writes hit the cached read relation.
##########
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:
Addressed in 85be94c8b98.
`tryRebuildCacheEntry` clones a session, applies
`CharVarcharScanMode.configure` for the captured mode, refreshes the plan, and
creates a new `QueryExecution` with `refreshPhaseEnabled = false` under that
session. Rename restoration uses the same helper, so a SparkStandard
write/refresh cannot attach a SparkStandard Project to a PreserveNative cache
key.
##########
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:
Addressed in 85be94c8b98.
Temp-view reanalysis now drops a stored analyzer-generated CHAR/VARCHAR
Project around `V2TableReference` before `resolveReference`, so
`ApplyCharTypePadding` rebinds both the scan mode and the read-side expressions
under the current SQLConf. Covered by `SPARK-58814: V2 temp view rebinds
CHAR/VARCHAR policy on lookup`, which stores a PreserveNative view over an
over-length VARCHAR and expects `EXCEED_LIMIT_LENGTH` on a SparkStandard lookup.
##########
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:
Addressed in 85be94c8b98.
`SparkStandard` Scaladoc now says physical STRING and Spark-side length
checks apply to readers that honor the mode, and that native Hive ORC with
`CONVERT_METASTORE_ORC=false` still truncates natively.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Relation.scala:
##########
@@ -115,7 +115,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:
Addressed in 85be94c8b98.
The catalog-less micro-batch test now `checkAnswer`s the retained
`preserveRead` and `standardRead` DataFrames after commit (first collect under
the matching scan-mode conf) and still asserts
`lookupCacheDescriptorsByV2Relation` is empty. Streaming write targets that
omit `path` match the cached read by table name so `uncacheByV2Relation`
actually invalidates both bound entries.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/LogicalRelation.scala:
##########
@@ -41,7 +41,10 @@ case class LogicalRelation(
output: Seq[AttributeReference],
catalogTable: Option[CatalogTable],
override val isStreaming: Boolean,
- @transient stream: Option[SparkDataStream])
+ @transient stream: Option[SparkDataStream],
+ // 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])
Review Comment:
Addressed in 85be94c8b98.
The Hive ORC test no longer calls `clearCache` before `REFRESH TABLE`. Both
CONVERT_METASTORE_ORC paths keep the PreserveNative cache through refresh.
Converted `LogicalRelation` caches are checked for the rewritten padded value;
native `HiveTableRelation` still asserts the PreserveNative cache remains and
SparkStandard stays uncached.
--
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]
