srielau commented on code in PR #58317:
URL: https://github.com/apache/spark/pull/58317#discussion_r4140745575
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/CacheManager.scala:
##########
@@ -436,6 +776,71 @@ class CacheManager extends Logging with
AdaptiveSparkPlanHelper {
}
}
+ // 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.
+ // Rebuild output from the fresh table (SPARK-54424). create() rewrites
CHAR/VARCHAR to
+ // annotated STRING; restore first-class types from the catalog schema so
re-analysis
+ // matches a fresh spark.table() cache key.
+ private def refreshV2RelationTable(
+ relation: DataSourceV2Relation,
+ catalog: TableCatalog,
+ ident: Identifier): Option[DataSourceV2Relation] = {
+ val table = CatalogV2Util.getTable(catalog, ident, options =
relation.options)
+ if (relation.table.id != table.id) {
+ None
+ } else {
+ val rebuilt = DataSourceV2Relation
+ .create(table, Some(catalog), Some(ident), relation.options)
+ .copy(charVarcharScanMode = relation.charVarcharScanMode)
+ val restored = restoreFirstClassCharVarcharTypes(rebuilt, table,
relation)
+ // Same column names: keep captured attributes so CHAR/VARCHAR cache
keys still match.
+ // Different names: use the rebuilt output so SPARK-54424 can add or
drop columns.
+ if (relation.output.map(_.name) == restored.output.map(_.name)) {
Review Comment:
Captured attributes are kept only when name, nullability, and logical
CHAR/VARCHAR types match (including annotated STRING metadata). A same-name
widen such as VARCHAR(4) to VARCHAR(8) now rebuilds output from the catalog.
Added `SPARK-54424: refresh CHAR/VARCHAR table cache on type change`, which
inserts `abcdef` after the widen and checks VARCHAR(8) plus the stored value.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/CacheManager.scala:
##########
@@ -436,6 +776,71 @@ class CacheManager extends Logging with
AdaptiveSparkPlanHelper {
}
}
+ // 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.
+ // Rebuild output from the fresh table (SPARK-54424). create() rewrites
CHAR/VARCHAR to
+ // annotated STRING; restore first-class types from the catalog schema so
re-analysis
+ // matches a fresh spark.table() cache key.
+ private def refreshV2RelationTable(
+ relation: DataSourceV2Relation,
+ catalog: TableCatalog,
+ ident: Identifier): Option[DataSourceV2Relation] = {
+ val table = CatalogV2Util.getTable(catalog, ident, options =
relation.options)
+ if (relation.table.id != table.id) {
+ None
+ } else {
+ val rebuilt = DataSourceV2Relation
+ .create(table, Some(catalog), Some(ident), relation.options)
+ .copy(charVarcharScanMode = relation.charVarcharScanMode)
+ val restored = restoreFirstClassCharVarcharTypes(rebuilt, table,
relation)
+ // Same column names: keep captured attributes so CHAR/VARCHAR cache
keys still match.
+ // Different names: use the rebuilt output so SPARK-54424 can add or
drop columns.
+ if (relation.output.map(_.name) == restored.output.map(_.name)) {
Review Comment:
Captured attributes are kept only when name, nullability, and logical
CHAR/VARCHAR types match (including annotated STRING metadata). A same-name
widen such as VARCHAR(4) to VARCHAR(8) now rebuilds output from the catalog.
Added `SPARK-54424: refresh CHAR/VARCHAR table cache on type change`, which
inserts `abcdef` after the widen and checks VARCHAR(8) plus the stored value.
--
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]
