cloud-fan commented on code in PR #58317:
URL: https://github.com/apache/spark/pull/58317#discussion_r4009049859
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/InsertIntoDataSourceCommand.scala:
##########
@@ -45,7 +45,8 @@ case class InsertIntoDataSourceCommand(
// Re-cache all cached plans(including this relation itself, if it's
cached) that refer to this
// data source relation.
- sparkSession.sharedState.cacheManager.recacheByPlan(sparkSession,
logicalRelation)
+ sparkSession.sharedState.cacheManager.recacheByV1Relation(
Review Comment:
Confirmed: InsertSuite now materializes the cache, overwrites under both
bound modes, and checks both cache retention and replacement rows. Thanks.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:3962201895","thread_id":"inline:3962201895","verdict_sha256":"9e8299073fe3a3fd317d35d14ddb819ddbbaf0c580a01f233a538f0836de3d9e"}
-->
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/ApplyCharTypePadding.scala:
##########
@@ -51,34 +51,73 @@ object ApplyCharTypePadding extends Rule[LogicalPlan] {
}
override def apply(plan: LogicalPlan): LogicalPlan = {
+ val standardSemantics = conf.charVarcharStandardSemantics
+ val scanMode = CharVarcharScanMode(standardSemantics)
+
+ // Bind into case-class state, not a TreeNodeTag: tags do not participate
in structural plan
+ // equality / sameResult, so cache lookup and scan reuse would treat
preserve-only and standard
+ // scans as the same plan. A case-class field does participate. Keep an
already-bound value
+ // (views, catalog-cached relations) unchanged.
+ def bindStandardSemantics(p: LogicalPlan): LogicalPlan = p match {
+ case relation: LogicalRelation if relation.charVarcharScanMode.isEmpty =>
+ val bound = relation.copy(charVarcharScanMode = Some(scanMode))
+ bound.copyTagsFrom(relation)
+ bound
+ case relation: DataSourceV2Relation if
relation.charVarcharScanMode.isEmpty =>
+ val bound = relation.copy(charVarcharScanMode = Some(scanMode))
+ bound.copyTagsFrom(relation)
+ bound
+ case relation: HiveTableRelation if relation.charVarcharScanMode.isEmpty
=>
+ val bound = relation.copy(charVarcharScanMode = Some(scanMode))
+ bound.copyTagsFrom(relation)
+ bound
+ case _ => p
+ }
+
+ val boundPlan = if (conf.charVarcharFirstClassTypes) {
+ plan.resolveOperatorsUp {
+ case relation: LogicalRelation => bindStandardSemantics(relation)
+ case relation: DataSourceV2Relation => bindStandardSemantics(relation)
+ case relation: HiveTableRelation => bindStandardSemantics(relation)
+ }
+ } else {
+ plan
+ }
+
// standardSemantics takes precedence over legacy charVarcharAsString.
- if (conf.charVarcharAsString && !conf.charVarcharStandardSemantics) {
- return plan
+ if (conf.charVarcharAsString && !standardSemantics) {
+ return boundPlan
}
- if (conf.charVarcharStandardSemantics && !conf.readSideCharPadding) {
+ if (standardSemantics && !conf.readSideCharPadding) {
warnReadSidePaddingOverride()
}
- if (conf.readSideCharPadding || conf.charVarcharStandardSemantics) {
- val newPlan = plan.resolveOperatorsUpWithNewOutput {
+ if (conf.readSideCharPadding || standardSemantics) {
+ val newPlan = boundPlan.resolveOperatorsUpWithNewOutput {
case r: LogicalRelation =>
+ bindStandardSemantics(r)
Review Comment:
Confirmed: the three discarded binding calls are gone and only retained
traversal results remain. Thanks.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:3962201904","thread_id":"inline:3962201904","verdict_sha256":"9e8299073fe3a3fd317d35d14ddb819ddbbaf0c580a01f233a538f0836de3d9e"}
-->
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Strategy.scala:
##########
@@ -72,7 +72,7 @@ class DataSourceV2Strategy(session: SparkSession) extends
Strategy with Predicat
val nameParts = ident.toQualifiedNameParts(catalog)
cacheManager.recacheTableOrView(session, nameParts, includeTimeTravel =
false)
case _ =>
- cacheManager.recacheByPlan(session, r)
+ cacheManager.recacheByV2Relation(session, r)
Review Comment:
Confirmed that the mutation-only lookup fixes this thread's
unbound-versus-bound scan-mode mismatch. I found a separate direct-cache
Project shape that is still omitted and will report it in the new review.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:3962201912","thread_id":"inline:3962201912","verdict_sha256":"9e8299073fe3a3fd317d35d14ddb819ddbbaf0c580a01f233a538f0836de3d9e"}
-->
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcUtils.scala:
##########
@@ -528,18 +541,24 @@ object OrcUtils extends Logging {
* @param resultSchema Result data schema created after pruning cols.
* @param partitionSchema Schema of partitions.
* @param conf Hadoop Configuration.
+ * @param charVarcharStandardSemantics When true, request physical ORC
STRING so Spark can
+ * apply CHAR/VARCHAR length checks. When
false, keep native
+ * constrained ORC CHAR/VARCHAR types.
* @return Returns the result schema as string.
*/
def orcResultSchemaString(
canPruneCols: Boolean,
dataSchema: StructType,
resultSchema: StructType,
partitionSchema: StructType,
- conf: Configuration): String = {
+ conf: Configuration,
+ charVarcharStandardSemantics: Boolean): String = {
val resultSchemaString = if (canPruneCols) {
- OrcUtils.getOrcSchemaString(resultSchema)
+ OrcUtils.getOrcSchemaString(resultSchema, charVarcharStandardSemantics)
} else {
- OrcUtils.getOrcSchemaString(StructType(dataSchema.fields ++
partitionSchema.fields))
+ OrcUtils.getOrcSchemaString(
Review Comment:
Confirmed: both ORC source suites now cover forced positional evolution,
in-range padding, and over-length EXCEED_LIMIT_LENGTH behavior. Thanks.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:3962201917","thread_id":"inline:3962201917","verdict_sha256":"9e8299073fe3a3fd317d35d14ddb819ddbbaf0c580a01f233a538f0836de3d9e"}
-->
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/CacheManager.scala:
##########
@@ -348,6 +349,45 @@ 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 a cache entry for a V2 table mutation while ignoring only its
analyzed CHAR/VARCHAR
+ * scan mode. Normal cache substitution remains mode-sensitive.
+ */
+ def lookupCachedDataByV2Relation(relation: DataSourceV2Relation):
Option[CachedData] = {
+ val unboundRelation = relation.copy(charVarcharScanMode = None)
+ cachedData.find(_.plan.exists {
Review Comment:
Confirmed that the new lookup excludes dependent and time-travel keys and
restores all matching bare-root variants. I found a separate direct
CHAR/VARCHAR Project shape that is still omitted and will report it in the new
review.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:3992302185","thread_id":"inline:3992302185","verdict_sha256":"9e8299073fe3a3fd317d35d14ddb819ddbbaf0c580a01f233a538f0836de3d9e"}
-->
--
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]
