srielau commented on code in PR #58317:
URL: https://github.com/apache/spark/pull/58317#discussion_r4106191657
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RelationResolution.scala:
##########
@@ -344,7 +344,10 @@ class RelationResolution(
writePrivileges == null && !u.isStreaming
cached <- lookupSharedRelationCache(catalog, ident, t,
tableKey.stateOptions)
} yield {
- val updatedRelation = cached.copy(options = finalOptions)
+ // A shared cache entry may have been analyzed under another
session's CHAR/VARCHAR
+ // policy. Rebind it in this analysis instead of inheriting that
session's scan mode.
+ val updatedRelation =
+ cached.copy(options = finalOptions, charVarcharScanMode = None)
Review Comment:
Addressed in 08dc28c57d7.
The `V2TableReference` cache adaptation now clears `charVarcharScanMode`
while retaining the reference output and options, so `ApplyCharTypePadding`
binds the current analysis mode. The temporary-view shared-cache test seeds a
`PreserveNative` cached relation and verifies both the initial shared-cache
result and subsequent query-cache hits are unbound for re-analysis.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/LogicalRelation.scala:
##########
@@ -77,6 +80,16 @@ case class LogicalRelation(
case _ => // Do nothing.
}
+ def hasCharVarchar: Boolean = output.exists { attr =>
+ CharVarcharUtils.hasCharVarchar(attr.dataType) ||
+
CharVarcharUtils.getRawType(attr.metadata).exists(CharVarcharUtils.hasCharVarchar)
+ }
+
+ def sameResultWithUnboundCharVarcharScanMode(other: LogicalRelation):
Boolean = {
Review Comment:
Addressed in 08dc28c57d7. The unused
`LogicalRelation.sameResultWithUnboundCharVarcharScanMode` helper has been
removed; V1 mutation matching continues to use the underlying `BaseRelation`.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/CacheManager.scala:
##########
@@ -348,6 +358,92 @@ 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 = {
+ recacheByCondition(spark, cd => cd.plan.exists {
+ case cached: DataSourceV2Relation =>
+ cached.sameResultWithUnboundCharVarcharScanMode(relation)
+ case _ => false
+ })
+ }
+
+ /**
+ * Removes cache entries whose plans contain the given catalog-less
+ * [[DataSourceV2Relation]]. The scan mode is ignored only for this
mutation-specific match.
+ */
+ def uncacheByV2Relation(
+ spark: SparkSession,
+ relation: DataSourceV2Relation,
+ cascade: Boolean,
+ blocking: Boolean = false): Unit = {
+ uncacheByCondition(
+ spark,
+ {
+ case cached: DataSourceV2Relation =>
+ cached.sameResultWithUnboundCharVarcharScanMode(relation)
+ case _ => false
+ },
+ cascade,
+ blocking)
+ }
+
+ /**
+ * Describes cache entries containing the given V2 relation, ignoring scan
mode for the relation
Review Comment:
Addressed in 08dc28c57d7.
The default descriptor lookup now uses `plan.exists` and compares every
`DataSourceV2Relation`, emitting at most one descriptor per cache entry. The
direct-only branch remains strict. A cached join regression verifies lookup
finds the requested second relation without treating the dependent query as a
direct table cache.
##########
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.
Review Comment:
Addressed in 08dc28c57d7. Both `LogicalRelation` and `DataSourceV2Relation`
now document `None` only as "no scan mode was bound," without inferring whether
first-class analysis was enabled.
##########
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:
Completed the missing catalog-less coverage in 08dc28c57d7.
`CharVarcharWritableDataSource` now supports committed micro-batch writes
for this fixture. The new regression materializes both bound cache modes on a
catalog-less relation, commits a micro-batch through `.format(...)`, verifies
both fresh reads observe the appended row, and verifies
`lookupCacheDescriptorsByV2Relation` finds no stale cache entry. This reaches
the `uncacheByV2Relation` branch rather than the catalog-name branch.
--
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]
