cloud-fan commented on code in PR #58317:
URL: https://github.com/apache/spark/pull/58317#discussion_r4015284288
##########
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:
**Non-blocking (P2):** A cache created with both first-class flags disabled
legitimately carries `None`, but a refreshed relation copied with that value is
analyzed again under the mutation session. If that session enables a
first-class mode, `ApplyCharTypePadding` treats the completed legacy decision
as unbound and replaces it. This can collapse the legacy entry into a
first-class cache or make its key and cached execution disagree after rename.
Please represent legacy as an explicit bound state and preserve it across cache
rebuild and rename; the regression should coexist a legacy and first-class
cache and mutate from the opposite session.
**Recommended change:** Represent the legacy scan policy as an explicit
bound state, bind every analyzed relation to one policy, and make recache and
rename restoration preserve that state without consulting the mutation session.
**Why this works:** Replace the overloaded empty sentinel with an explicit
analysis result (or an equivalent separate bound marker), propagate it through
V1, Hive, and V2 relation copies, and construct one consistently bound plan for
both cache identity and cached execution.
**Scope:** Make scan-policy binding total and preserve the resulting state
through cache rebuild and rename paths.
**Compatibility:** Fresh relations continue to bind from the analyzing
session, while already analyzed relations and persisted views retain their
prior decision.
**Risks:** Changing relation case-class state can alter plan equality and
serialization assumptions if every constructor and copy site is not updated
together.
**Constraints:** Ordinary session analysis must still select legacy,
preserve-native, or standard behavior from the current SQLConf exactly once.
Normal cache substitution must remain mode-sensitive.
**Success:** A legacy cache entry remains a distinct legacy entry after
mutation or rename from a session using either first-class mode. Cache keys and
their InMemoryRelation execution use the same preserved policy. PreserveNative
and SparkStandard cache identity remains distinct.
##########
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:
**Nit (P3):** This node is format-agnostic, so `None` only means that Spark
selects the legacy seven-argument overload; native constrained decoding is
OrcFileFormat-specific. Likewise, the typed overload generically bridges a
private Hadoop value, but an arbitrary FileFormat subclass observes semantic
mode only if it understands that value. Could these comments state the generic
overload/virtual-dispatch contract first and qualify the preserve-native
interpretation as ORC-only?
##########
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:
**Blocking (P1):** Adding `charVarcharScanMode` to relation identity also
changes the micro-batch write path: `WriteToMicroBatchDataSource` keeps its
target outside the child tree, so V2Writes forwards this relation with `None`,
and `DataSourceV2Strategy` later calls ordinary mode-sensitive `uncacheQuery`
after commit. That probe cannot match either bound cache variant, allowing a
successfully appended batch to leave stale cached rows. Please route this
callback through the same catalog-name or catalog-less mutation-specific
identity used by batch V2 writes, and cover a committed micro-batch with both
cache modes materialized.
**Recommended change:** Route micro-batch V2 targets through the same
table-wide or catalog-less mode-insensitive cache mutation helper as batch V2
writes, preserving cascade semantics for dependent cached queries.
**Why this works:** Build the callback from the target relation's catalog
identity: invalidate by qualified table name when present and otherwise match
the underlying V2 relation with scan mode removed, rather than calling
sameResult on the unbound target.
**Scope:** Make streaming V2 writes invalidate all semantic cache variants
after each commit.
**Compatibility:** Ordinary read cache identity remains mode-sensitive and
failed writes do not publish refreshed cache state.
**Risks:** Using recache rather than invalidation at the wrong streaming
lifecycle point could expose pre-commit data or preserve stale dependent caches.
**Constraints:** The callback runs only after successful write commit.
Catalog-backed tables retain qualified-name invalidation behavior. Catalog-less
matching ignores only CHAR/VARCHAR mode identity.
**Success:** A successful V2 micro-batch write leaves no materialized cache
variant serving pre-batch rows. Both PreserveNative and SparkStandard cache
entries are invalidated even though the write target is unbound.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/orc/OrcPartitionReaderFactory.scala:
##########
@@ -49,6 +49,7 @@ import org.apache.spark.util.ArrayImplicits._
* @param readDataSchema Required data schema in the batch scan.
* @param partitionSchema Schema of partitions.
* @param options Options for parsing ORC files.
+ * @param charVarcharStandardSemantics CHAR/VARCHAR semantics bound during
analysis.
Review Comment:
**Nit (P3):** `false` is not always a mode bound during analysis:
`OrcScanBuilder` initializes it as the unbound default. Could this parameter
documentation also define the behavior of both values—`true` requests physical
ORC STRING for Spark-side checks, while `false` retains native constrained
decoding and can mean either explicit PreserveNative or the unbound default?
##########
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:
**Blocking (P1):** Making this field part of V1/Hive `sameResult` also
changes `spark.catalog.refreshTable`: the method resolves one relation under
the caller's mode, and `CommandUtils.recacheTableOrView` falls back to
`recacheByPlan` for these nodes. With PreserveNative and SparkStandard caches
in SharedState, only the caller-mode entry is rebuilt and the sibling can
continue serving pre-refresh data. Please use stable V1/Hive table or
BaseRelation identity for this table-wide operation while keeping ordinary
cache substitution mode-sensitive, and test one refresh against both coexisting
variants.
**Recommended change:** Recognize direct V1 and Hive relations in the
refresh helper and recache all entries for their stable table or BaseRelation
identity while retaining existing name-based V2 and view behavior.
**Why this works:** Dispatch refresh plans to a mode-insensitive V1/Hive
relation matcher (or qualified table matcher when available) before the
recacheByPlan fallback, and let CacheManager rebuild every selected entry with
its original semantic descriptor.
**Scope:** Restore refreshTable's table-wide contract after relation
equality becomes mode-sensitive.
**Compatibility:** The caller-visible refresh operation remains table-wide
and continues to refresh dependent cached queries according to existing policy.
**Risks:** Matching by table name too broadly could alter time-travel or
view refresh behavior.
**Constraints:** Refresh preserves current dependent-cache semantics.
Ordinary cache lookup stays mode-sensitive. V2 catalog-backed refresh remains
name-based.
**Success:** One refreshTable call refreshes every V1 or Hive cache variant
for the named table. No sibling semantic mode can continue serving stale
pre-refresh data.
--
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]
