cloud-fan commented on code in PR #58317:
URL: https://github.com/apache/spark/pull/58317#discussion_r3960197735
##########
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):** Adding this mode to case-class state fixes cross-mode
cache reuse, but it also makes `sameResult` distinguish the unbound target used
after a catalog-less V2 write from every bound read of the same table.
`DataSourceV2Strategy.refreshCache` still calls `recacheByPlan(session, r)` for
this path, and `V2WriteCommand.table` is a non-child slot, so analysis leaves
the write target at `None` while cached reads carry `Some(PreserveNative)` or
`Some(SparkStandard)`. A successful append can therefore leave a materialized
path read stale. Please add a V2 write-invalidation path that matches the
written TableProvider relation while ignoring only this mode, and cover
append/recache under both bound modes without weakening ordinary cross-mode
cache identity.
**Recommended change:** Add a dedicated catalog-less V2
mutation-invalidation path that deliberately ignores only charVarcharScanMode
when locating cache entries for the written relation.
**Why this works:** Have CacheManager match cached DataSourceV2Relation
nodes to the written TableProvider relation using the existing table identity
while excluding charVarcharScanMode from this mutation-only comparison, then
use that path from DataSourceV2Strategy.refreshCache.
**Scope:** CacheManager, the catalog-less TableProvider branch of
DataSourceV2Strategy.refreshCache, and focused V2 append/recache coverage.
**Compatibility:** Keep charVarcharScanMode in ordinary sameResult identity
so preserve-native and standard reads cannot reuse each other's cached results.
**Risks:** A matcher broader than the written TableProvider relation could
refresh unrelated cache entries. Covering only one bound mode could leave the
symmetric mismatch undiscovered.
**Constraints:** Ignore the mode only for mutation invalidation, not for
normal cache substitution. Invalidate every cached descendant that reads the
successfully mutated relation.
**Success:** After a catalog-less V2 append or overwrite, materialized
cached reads observe the written rows under PreserveNative and SparkStandard,
while cross-mode read plans remain distinct for normal cache reuse.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/ApplyCharTypePadding.scala:
##########
@@ -51,34 +51,74 @@ 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: `TreeNode.makeCopy`
calls `copyTagsFrom`,
+ // so a tag survives canonicalization, but it does not participate in
structural plan
Review Comment:
**Nit (P3):** `TreeNode.makeCopy` is not a general guarantee that tags
survive canonicalization: `LogicalRelation.doCanonicalize` and
`HiveTableRelation.doCanonicalize` return direct `copy(...)` results, so
QueryPlan never invokes the tag-copying fallback and the tags are dropped.
Please avoid the blanket claim here and state the stable reason for using a
field: `TreeNodeTag` values do not participate in structural equality /
`sameResult`.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileFormat.scala:
##########
@@ -266,6 +295,25 @@ object FileFormat {
*/
val OPTION_RETURNING_BATCH = "returning_batch"
+ /**
+ * Engine-private Hadoop configuration entry that transports the analyzed
CHAR/VARCHAR scan mode
+ * across the legacy [[FileFormat.buildReaderWithPartitionValues]]
signature. It is written by the
+ * mode-aware overload and read by formats that honor first-class
CHAR/VARCHAR types. This is not
+ * a public option; the authoritative state is the typed plan field and
overload parameter.
+ */
+ val CHAR_VARCHAR_SCAN_MODE = "__spark_sql_char_varchar_scan_mode"
+
+ /** Writes the CHAR/VARCHAR scan mode into `conf` under
[[CHAR_VARCHAR_SCAN_MODE]]. */
+ private[sql] def setCharVarcharScanMode(
+ conf: Configuration, mode: CharVarcharScanMode): Unit = {
+ conf.set(CHAR_VARCHAR_SCAN_MODE, mode.toString)
+ }
+
+ /** Reads the CHAR/VARCHAR scan mode from `conf`, or `None` if no mode was
bridged in. */
+ private[sql] def charVarcharScanMode(conf: Configuration):
Option[CharVarcharScanMode] = {
+ Option(conf.get(CHAR_VARCHAR_SCAN_MODE)).map(CharVarcharScanMode.fromName)
Review Comment:
**Non-blocking (P2):** `FileSourceScanExec` builds this Hadoop configuration
from relation options. When the relation mode is `None`, the legacy overload
passes it unchanged, and ORC now calls this strict parser on any value under
this key. Thus `spark.read.option("__spark_sql_char_varchar_scan_mode",
"not-a-mode")...` can fail an otherwise ordinary unbound ORC read; a valid
value can forge analyzed semantics. Please remove the private key before legacy
dispatch and let only the typed overload add it to its cloned per-call
configuration, with a collision regression.
**Recommended change:** Ensure that only the mode-aware FileFormat overload
can author the private CHAR/VARCHAR bridge entry.
**Why this works:** Strip the reserved bridge key from Hadoop configurations
derived from relation options before legacy dispatch, then add it only to the
cloned configuration created by the typed overload.
**Scope:** V1 FileSourceScanExec/FileFormat configuration dispatch and an
unbound ORC read regression with a colliding caller option.
**Compatibility:** Existing callers of the seven-argument overload continue
to use legacy preserve-native behavior, and arbitrary data-source options
remain opaque to ORC scan-mode selection.
**Risks:** Removing the key after the typed overload adds it would break
delegating OrcFileFormat subclasses. Sanitizing only direct DataFrameReader
options could miss persisted table options.
**Constraints:** Preserve virtual dispatch through the existing
seven-argument overload. Do not let caller-controlled option maps select or
invalidate typed plan state.
**Success:** An unbound ORC read ignores both valid and invalid caller
values under the reserved key, while bound PreserveNative and SparkStandard
calls still cross delegating legacy overrides with their analyzed mode intact.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/SaveIntoDataSourceCommandSuite.scala:
##########
@@ -69,7 +70,29 @@ class SaveIntoDataSourceCommandSuite extends
SharedSparkSession {
saveIntoDataSource(2)
checkAnswer(loadData, Row(0) :: Row(1) :: Nil)
+ spark.catalog.clearCache()
FakeV1DataSource.data = null
+
+ // Bound modes store Some(false)/Some(true) on LogicalRelation, so recache
must match
Review Comment:
**Nit (P3):** `charVarcharScanMode` is now `Option[CharVarcharScanMode]`, so
this comment's `Some(false)`/`Some(true)` representation no longer exists.
Please name `Some(PreserveNative)` and `Some(SparkStandard)` (or simply say
"the two bound modes") so the cache-invalidation rationale matches the state
under test.
--
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]
