cloud-fan commented on code in PR #58317:
URL: https://github.com/apache/spark/pull/58317#discussion_r4103666679
##########
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:
**Nit (P3):** There are no callers for this new LogicalRelation helper: V1
invalidation matches the underlying BaseRelation, while only the V2 counterpart
is consumed. Keeping it creates a second mode-insensitive identity contract
with no behavior or test surface, so it should be removed.
##########
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:
**Blocking (P1):** This reset is missing from the shared-cache hit used by
V2TableReference. That path passes the cached relation to adaptCachedRelation
with its old nonempty charVarcharScanMode, so ApplyCharTypePadding will not
bind the current session. A temporary view first resolved under PreserveNative
can consequently truncate an over-length value when read under SparkStandard
instead of raising EXCEED_LIMIT_LENGTH.
##########
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:
**Nit (P3):** None means that no CHAR/VARCHAR scan mode was bound; it does
not prove first-class analysis was disabled. A first-class relation with no
CHAR/VARCHAR columns also remains None. The same wording should be corrected on
DataSourceV2Relation.
##########
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:
The callback now has the requested catalog-name/catalog-less split, but the
new micro-batch regression writes only to testcat.ns.cached_cv, so it exercises
uncacheTableOrView and cannot catch a regression in uncacheByV2Relation. Please
add a catalog-less target with both bound cache modes and assert they are
invalidated after commit.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:4015284304","thread_id":"inline:4015284304","verdict_sha256":"73cea14dc62d3c904fc325e9b5a28e808d39f74126913437982926fe0eb16528"}
-->
##########
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:
**Nit (P3):** The default branch does not satisfy the documented
containing-relation contract: collectFirst fixes one V2 leaf, so a cached join
whose requested table appears later is missed. Please test every V2 relation in
the cached plan and emit the descriptor once when any leaf matches, while
keeping directNamedCacheOnly strict.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/CacheManager.scala:
##########
@@ -348,6 +358,81 @@ 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 direct named cache entries for a V2 table. Matching the cache's
table name excludes
Review Comment:
The helper now documents both call forms, but the default branch uses
collectFirst and compares only the first DataSourceV2Relation. A cached join
whose matching relation is a later child is therefore omitted despite the new
"containing" contract; please either search all V2 relations or narrow the
contract and call surface.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:4091094262","thread_id":"inline:4091094262","verdict_sha256":"73cea14dc62d3c904fc325e9b5a28e808d39f74126913437982926fe0eb16528"}
-->
--
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]
