srielau commented on code in PR #58317:
URL: https://github.com/apache/spark/pull/58317#discussion_r4140330785
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/CacheManager.scala:
##########
@@ -416,26 +703,30 @@ class CacheManager extends Logging with
AdaptiveSparkPlanHelper {
*
* @return the refreshed plan if refresh succeeds, None otherwise
*/
- private def tryRefreshPlan(spark: SparkSession, plan: LogicalPlan):
Option[LogicalPlan] = {
+ private def tryRefreshPlan(plan: LogicalPlan): Option[LogicalPlan] = {
try {
- EliminateSubqueryAliases(plan) match {
- 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))
- } else {
- None
- }
- case _ =>
- Some(V2TableRefreshUtil.refresh(spark, plan))
- }
+ // Keep scan mode and any analyzer padding Project. Load Table from the
catalog rather
+ // than the shared relation cache so a recache after write sees
committed rows.
+ Some(refreshV2RelationTablesFromCatalog(plan))
} catch {
case NonFatal(e) =>
logWarning(log"Failed to refresh plan while attempting to recache", e)
None
}
}
+ // Load from the catalog, not the shared relation cache, so a recache after
write sees
+ // the committed rows instead of the copy pinned when the cache was created.
+ // Keep output types, scan mode, and padding Projects:
DataSourceV2Relation.create rewrites
+ // CHAR/VARCHAR to annotated STRING, so a later analysis pass cannot restore
the cache key.
+ private def refreshV2RelationTablesFromCatalog(plan: LogicalPlan):
LogicalPlan = {
+ plan.transformWithSubqueries {
+ case r @ ExtractV2CatalogAndIdentifier(catalog, ident) if
r.timeTravelSpec.isEmpty =>
+ val table = CatalogV2Util.getTable(catalog, ident, options = r.options)
+ if (r.table.id == table.id) r.copy(table = table) else r
Review Comment:
Direct caches now reconstruct output from `DataSourceV2Relation.create` when
the catalog column set changes, including CHAR/VARCHAR tables: first-class
types are restored from the catalog schema so the cache key still matches
`spark.table()`. Unchanged column names still copy captured attributes (and any
padding Project) so recache identity is preserved. Table-id mismatch returns
`None`. Derived plans still go through `V2TableRefreshUtil`.
Added SPARK-54424 CHAR/VARCHAR add-column and remove-column regressions.
--
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]
