srielau commented on code in PR #58317:
URL: https://github.com/apache/spark/pull/58317#discussion_r3969892655
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Relation.scala:
##########
@@ -116,7 +116,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:
Fixed in 7c4b2a11cce. RelationResolution now clears charVarcharScanMode only
when adapting a shared relation-cache hit, so the current analysis rebinds it
while persisted-view plans remain untouched. Added a same-SharedState
regression that seeds PreserveNative and verifies a fresh SparkStandard session
raises EXCEED_LIMIT_LENGTH.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/CharVarcharScanMode.scala:
##########
@@ -0,0 +1,52 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.util
+
+/**
+ * The CHAR/VARCHAR scan mode bound to a relation (and its scan) during
analysis.
+ *
+ * A relation carries `Option[CharVarcharScanMode]`: `None` means the relation
was not analyzed
+ * under first-class CHAR/VARCHAR types (native reader behavior), while a
`Some` value pins the
+ * mode so that `sameResult` / cache reuse keep the two variants distinct.
+ */
+sealed trait CharVarcharScanMode
+
+object CharVarcharScanMode {
+ /**
+ * Preserve the native, constrained CHAR/VARCHAR types of the source (e.g.
native ORC
+ * padding/truncation). Corresponds to preserve-only semantics.
+ */
+ case object PreserveNative extends CharVarcharScanMode
+
+ /**
+ * Request physical STRING from the source so Spark observes the original
value and applies
+ * standard CHAR/VARCHAR length checks. Corresponds to standard semantics.
+ */
+ case object SparkStandard extends CharVarcharScanMode
+
+ /** Maps the boolean `spark.sql.charVarcharStandardSemantics` value to the
typed mode. */
+ def apply(standardSemantics: Boolean): CharVarcharScanMode =
+ if (standardSemantics) SparkStandard else PreserveNative
+
+ /** Parses a mode from its `toString` name; the inverse of
[[CharVarcharScanMode.toString]]. */
+ def fromName(name: String): CharVarcharScanMode = name match {
+ case "PreserveNative" => PreserveNative
+ case "SparkStandard" => SparkStandard
+ case other => throw new IllegalArgumentException(s"Unknown
CharVarcharScanMode: $other")
Review Comment:
Fixed in 7c4b2a11cce. The Scaladoc now names the registered key exactly:
spark.sql.charVarchar.standardSemantics.enabled.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileFormat.scala:
##########
@@ -165,6 +166,34 @@ trait FileFormat {
}
}
+ /**
+ * Same as [[buildReaderWithPartitionValues]] but also carries the analyzed
CHAR/VARCHAR scan
+ * mode. [[FileSourceScanExec]] calls this overload whenever the relation
has a bound mode,
+ * regardless of the concrete file format.
+ *
+ * The default implementation bridges the mode across the legacy
seven-argument signature: it
+ * clones the per-call Hadoop configuration, writes an engine-private entry
with the explicit
+ * mode, then invokes the seven-argument method virtually. A format that
honors first-class
+ * CHAR/VARCHAR types (e.g.
[[org.apache.spark.sql.execution.datasources.orc.OrcFileFormat]])
+ * reads that entry in its seven-argument override, so existing subclasses
keep their override
+ * and a call to `super` retains the bound mode. The Hadoop entry is only a
transport across the
+ * legacy signature; the authoritative state remains the typed plan field
and this parameter.
+ */
+ def buildReaderWithPartitionValues(
Review Comment:
Fixed in 7c4b2a11cce. CharVarcharScanMode, the typed reader overload, and
CHAR_VARCHAR_SCAN_MODE are now private[sql], while the existing seven-argument
FileFormat hook remains public for custom formats.
##########
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:
Fixed in 7c4b2a11cce. Added an InsertSuite overwrite regression under both
PreserveNative and SparkStandard. It materializes the cache, overwrites the
source, verifies the replacement query remains cached, and checks only
replacement rows are returned.
##########
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:
Fixed in 7c4b2a11cce. Removed all three discarded bindStandardSemantics
calls; the initial binding traversal and retained cleaned-relation thunks
provide the required bound nodes.
##########
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:
Fixed in 7c4b2a11cce. V2 rename discovery now uses a mutation-only cache
lookup that clears only charVarcharScanMode on both relations. Rename carries
the matched mode and prior StorageLevel onto the new relation before recaching.
The time-travel rename regression now runs under both bound modes and verifies
cache retention and storage level.
##########
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:
Fixed in 7c4b2a11cce. Added OrcSourceV1Suite and OrcSourceV2Suite coverage
with FORCE_POSITIONAL_EVOLUTION. The in-range CHAR is padded and the
over-length physical STRING raises EXCEED_LIMIT_LENGTH under standard semantics.
--
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]
