srielau commented on code in PR #58317:
URL: https://github.com/apache/spark/pull/58317#discussion_r4084317229
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/CacheManager.scala:
##########
@@ -348,6 +349,47 @@ 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 = {
+ val unboundRelation = relation.copy(charVarcharScanMode = None)
+ recacheByCondition(spark, cd => cd.plan.exists {
+ case cached: DataSourceV2Relation =>
+ cached.copy(charVarcharScanMode = None).sameResult(unboundRelation)
+ case _ => false
+ })
+ }
+
+ /**
+ * Looks up direct cache entries for a V2 table mutation while ignoring only
their analyzed
+ * CHAR/VARCHAR scan mode. Normal cache substitution remains mode-sensitive.
+ */
+ def lookupCachedDataByV2Relation(relation: DataSourceV2Relation):
Seq[CachedData] = {
+ val unboundRelation = relation.copy(charVarcharScanMode = None)
+ cachedData.filter { cd =>
+ EliminateSubqueryAliases(cd.plan) match {
Review Comment:
Addressed at the current head. Direct named caches retain the analyzer-added
Project, and lookupCacheDescriptorsByV2Relation matches the contained V2
relation with only scan-mode identity ignored. The VARCHAR rename regression
materializes both variants and verifies their storage levels.
##########
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:
Corrected in 312f1646026. The earlier recache response was wrong because
micro-batch writes have an established cascading invalidation contract. The
callback now invalidates by catalog name or by mode-insensitive catalog-less
relation identity. Both SPARK-34947 and the two-mode CHAR/VARCHAR micro-batch
regression pass.
##########
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:
I am fencing SPARK-58814 at first-class format/ORC behavior and regressions
directly introduced by its analyzed scan mode. I propose handling explicit
Legacy policy and complete cross-operation cache preservation in a follow-up
subtask under SPARK-58794. That follow-up will define the three-state lifecycle
contract across V1, V2, and Hive before changing None in this PR.
##########
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:
Fixed in 312f1646026. The FileSourceScanExec parameter documentation now
describes generic mode-aware versus legacy overload dispatch first and
qualifies constrained versus physical STRING decoding as ORC-specific.
##########
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:
The current head dispatches V1 refresh by catalog identity or BaseRelation
and Hive refresh by qualified table name before falling back to mode-sensitive
plan matching. This rebuilds all matching variants while preserving ordinary
cache identity. The focused V1 coexistence regression is present; the broader
three-policy V1/V2/Hive lifecycle matrix is fenced into the follow-up
cache-policy subtask.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/RenameTableExec.scala:
##########
@@ -49,9 +50,11 @@ case class RenameTableExec(
} else newIdent
catalog.renameTable(oldIdent, qualifiedNewIdent)
Review Comment:
Fixed in 312f1646026. RenameTableExec now loads the destination table once
when cached variants exist and reuses it for every restored descriptor. The
focused RENAME TABLE cache tests pass.
##########
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:
The Boolean has since been replaced by Option[CharVarcharScanMode]. Updated
in 312f1646026: the parameter documentation now states that SparkStandard
requests physical ORC STRING, while PreserveNative and the unbound fallback
retain native constrained decoding.
--
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]
