[SPARK-58814][SQL] Cover CHAR/VARCHAR round-trips and stop ORC truncating scans

connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala

@@ -40,7 +40,7 @@ import org.apache.spark.sql.TestingUDT.IntervalData
 import org.apache.spark.sql.avro.AvroCompressionCodec._
 import org.apache.spark.sql.catalyst.expressions.AttributeReference
 import org.apache.spark.sql.catalyst.plans.logical.Filter
-import org.apache.spark.sql.catalyst.util.DateTimeTestUtils
+import org.apache.spark.sql.catalyst.util.{CharVarcharUtils, DateTimeTestUtils}
 import org.apache.spark.sql.catalyst.util.DateTimeTestUtils.{withDefaultTimeZone, LA, UTC}
 import org.apache.spark.sql.connector.catalog.TableCapability
 import org.apache.spark.sql.execution.{FileSourceScanExec, FormattedMode, SparkPlan}
@@ -3862,6 +3862,87 @@ abstract class AvroSuite
     }
   }
 
+  test("SPARK-58814: Avro infers nested CHAR/VARCHAR schema and values") {
+    withSQLConf(SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "true") {
+      withTempPath { dir =>
+        val path = dir.getCanonicalPath
+        val input = spark.range(1).selectExpr(
+          "cast('ab' AS CHAR(4)) AS c",
+          "cast('xy' AS VARCHAR(3)) AS v",
+          "named_struct('c', cast('z' AS CHAR(2))) AS s",
+          "array(cast('q' AS VARCHAR(2))) AS a",
+          "map(cast('k' AS CHAR(2)), cast('v' AS VARCHAR(2))) AS m")
+        input.write.mode("overwrite").format("avro").save(path)
+
+        val readBack = spark.read.format("avro").load(path)
+        assert(DataType.equalsIgnoreNullability(readBack.schema, input.schema))
+        checkAnswer(
+          readBack.selectExpr(
+            "concat('<', c, '>')",
+            "v",
+            "concat('<', s.c, '>')",
+            "a",
+            "m"),
+          Row("<ab  >", "xy", "<z >", Seq("q"), Map("k " -> "v")))
+
+        withSQLConf(SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "false") {
+          assert(DataType.equalsIgnoreNullability(
+            spark.read.format("avro").load(path).schema,
+            CharVarcharUtils.replaceCharVarcharWithString(input.schema)))
+        }
+        withSQLConf(
+            SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "false",
+            SQLConf.PRESERVE_CHAR_VARCHAR_TYPE_INFO.key -> "true") {
+          assert(DataType.equalsIgnoreNullability(
+            spark.read.format("avro").load(path).schema,
+            input.schema))
+        }
+      }
+
+      withTempPath { dir =>
+        Seq("ab").toDF("c").write.format("avro").save(dir.getCanonicalPath)
+        val charDf = spark.read.schema("c CHAR(4)").format("avro").load(dir.getCanonicalPath)
+        checkAnswer(
+          charDf.selectExpr("concat('<', c, '>')"),
+          Row("<ab  >"))
+      }
+      withTempPath { dir =>
+        Seq("abcdef").toDF("c").write.format("avro").save(dir.getCanonicalPath)
+        Seq("CHAR", "VARCHAR").foreach { typ =>
+          checkError(
+            exception = intercept[SparkRuntimeException] {
+              spark.read.schema(s"c $typ(4)").format("avro")
+                .load(dir.getCanonicalPath).collect()
+            },
+            condition = "EXCEED_LIMIT_LENGTH",
+            parameters = Map("limit" -> "4"))
+        }
+      }
+
+      withTable("avro_char_varchar_assignment") {
+        sql(
+          """CREATE TABLE avro_char_varchar_assignment
+            |(c CHAR(4), v VARCHAR(4)) USING avro""".stripMargin)
+        sql("INSERT INTO avro_char_varchar_assignment VALUES ('ab', 'xy')")
+        assert(spark.table("avro_char_varchar_assignment").schema.map(_.dataType) ===
+          Seq(CharType(4), VarcharType(4)))
+        checkAnswer(
+          sql(
+            """SELECT concat('<', c, '>'), v
+              |FROM avro_char_varchar_assignment""".stripMargin),
+          Row("<ab  >", "xy"))
+        checkError(
+          exception = intercept[SparkRuntimeException] {
+            sql(
+              """INSERT INTO avro_char_varchar_assignment
+                |VALUES ('abcde', 'xy')""".stripMargin).collect()
+          },
+          condition = "EXCEED_LIMIT_LENGTH",
+          parameters = Map("limit" -> "4"))
+      }
+    }
+  }
+
 }
 
 class AvroV1Suite extends AvroSuite {

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala

@@ -1171,14 +1171,34 @@ class Analyzer(
 
     // Resolve V2TableReference nodes inside temp view plans. These are created by
     // V2TableReference.createForTempView. We only need to resolve it when returning
-    // the plan of temp views (in resolveViews and unwrapRelationPlan).
+    // the plan of temp views (in resolveViews and unwrapRelationPlan). Discard a
+    // stored CHAR/VARCHAR policy Project so ApplyCharTypePadding can rebind both
+    // the scan mode and the generated expressions under the current SQLConf.
     private def resolveTableReferencesInTempView(plan: LogicalPlan): LogicalPlan = {
-      plan.resolveOperatorsUp {
-        case r: V2TableReference if r.context.isInstanceOf[V2TableReference.TemporaryViewContext] =>
+      plan.transformDown {
+        case project @ Project(_, ref: V2TableReference)
+            if isTempViewReadSidePaddingProject(project, ref) =>
+          // The policy Project's output retains the raw CHAR/VARCHAR metadata and is referenced
+          // by any parent operators in the stored view plan. Resolve the replacement relation
+          // with those attributes so ApplyCharTypePadding can generate the current policy while
+          // keeping parent references valid.
+          val reboundRef = ref.copy(
+            output = project.output.map(_.asInstanceOf[AttributeReference]))
+          reboundRef.copyTagsFrom(ref)
+          relationResolution.resolveReference(reboundRef)
+        case r: V2TableReference
+            if r.context.isInstanceOf[V2TableReference.TemporaryViewContext] =>
           relationResolution.resolveReference(r)
       }
     }
 
+    private def isTempViewReadSidePaddingProject(
+        project: Project,
+        ref: V2TableReference): Boolean = {
+      ref.context.isInstanceOf[V2TableReference.TemporaryViewContext] &&
+        ApplyCharTypePaddingHelper.isAnyReadSidePaddingProject(project, ref)
+    }
+
     def apply(plan: LogicalPlan)
         : LogicalPlan = plan.resolveOperatorsUpWithPruning(AlwaysProcess.fn, ruleId) {
       case i @ InsertIntoStatement(table, _, _, _, _, _, _, _, _) =>

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ApplyCharTypePaddingHelper.scala

@@ -27,9 +27,14 @@ import org.apache.spark.sql.catalyst.expressions.{
   NamedExpression,
   OuterReference
 }
+import org.apache.spark.sql.catalyst.expressions.objects.StaticInvoke
 import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, Project}
 import org.apache.spark.sql.catalyst.trees.TreePattern.{BINARY_COMPARISON, IN}
-import org.apache.spark.sql.catalyst.util.CharVarcharUtils
+import org.apache.spark.sql.catalyst.util.{
+  CharVarcharCodegenUtils,
+  CharVarcharScanMode,
+  CharVarcharUtils
+}
 import org.apache.spark.sql.catalyst.util.CharVarcharUtils.createStringRPad
 import org.apache.spark.sql.types.{CharType, Metadata, StringType}
 import org.apache.spark.unsafe.types.UTF8String
@@ -65,6 +70,56 @@ object ApplyCharTypePaddingHelper {
     }
   }
 
+  /**
+   * Returns whether `project` is a read-side CHAR/VARCHAR projection generated for `relation`
+   * under either scan mode. Callers use this to strip a stale policy Project before rebinding.
+   */
+  private[sql] def isAnyReadSidePaddingProject(
+      project: Project,
+      relation: LogicalPlan): Boolean = {
+    isReadSidePaddingProject(project, relation, CharVarcharScanMode.PreserveNative) ||
+      isReadSidePaddingProject(project, relation, CharVarcharScanMode.SparkStandard)
+  }
+
+  /**
+   * Returns whether `project` is exactly the read-side CHAR/VARCHAR projection generated by this
+   * helper for `relation`. This rejects user projections and dependent cached queries.
+   */
+  private[sql] def isReadSidePaddingProject(
+      project: Project,
+      relation: LogicalPlan,
+      mode: CharVarcharScanMode): Boolean = {
+    if (project.projectList.length != relation.output.length ||
+        project.output.length != relation.output.length) {
+      false
+    } else {
+      val allowedFunctions = mode match {
+        case CharVarcharScanMode.SparkStandard =>
+          Set("charTypeReadSideCheck", "varcharTypeReadSideCheck")
+        case CharVarcharScanMode.PreserveNative => Set("readSidePadding")
+      }
+      var foundGeneratedExpression = false
+      val expressionsMatch = project.projectList.zip(relation.output).forall {
+        case (actual, childAttr) if actual.semanticEquals(childAttr) => true
+        case (actual, childAttr) =>
+          val generated = actual.references == childAttr.references &&
+            actual.exists {
+              case invoke: StaticInvoke
+                  if invoke.staticObject == classOf[CharVarcharCodegenUtils] &&
+                    allowedFunctions.contains(invoke.functionName) =>
+                true
+              case _ => false
+            }
+          foundGeneratedExpression ||= generated
+          generated
+      }
+      expressionsMatch && foundGeneratedExpression &&
+        project.output.zip(relation.output).forall { case (outputAttr, childAttr) =>
+          outputAttr.name == childAttr.name
+        }
+    }
+  }
+
   private[sql] def paddingForStringComparison(
       plan: LogicalPlan,
       padCharCol: Boolean): LogicalPlan = {

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RelationResolution.scala

@@ -344,7 +344,10 @@ class RelationResolution(
                 writePrivileges == null && !u.isStreaming
               cached <- lookupSharedRelationCache(catalog, ident, t, tableKey.stateOptions)
             } yield {
-              val updatedRelation = cached.copy(options = finalOptions)
+              // A shared cache entry may have been analyzed under another session's CHAR/VARCHAR
+              // policy. Rebind it in this analysis instead of inheriting that session's scan mode.
+              val updatedRelation =
+                cached.copy(options = finalOptions, charVarcharScanMode = None)
               updatedRelation.copyTagsFrom(cached)
               val nameParts = ident.toQualifiedNameParts(catalog)
               val aliasedRelation = SubqueryAlias(nameParts, updatedRelation)
@@ -591,7 +594,10 @@ class RelationResolution(
     cached transform {
       case r: DataSourceV2Relation if matchesReference(r, ref) =>
         V2TableReferenceUtils.validateLoadedTable(r.table, ref)
-        r.copy(output = ref.output, options = ref.options)
+        r.copy(
+          output = ref.output,
+          options = ref.options,
+          charVarcharScanMode = None)
     }
   }
 

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveChangelogTable.scala

@@ -121,7 +121,7 @@ object ResolveChangelogTable extends Rule[LogicalPlan] {
   }
 
   override def apply(plan: LogicalPlan): LogicalPlan = plan.resolveOperatorsUp {
-    case rel @ DataSourceV2Relation(table: ChangelogTable, _, _, _, _, _) if !table.resolved =>
+    case rel @ DataSourceV2Relation(table: ChangelogTable, _, _, _, _, _, _) if !table.resolved =>
       val changelog = table.changelog
       val req = evaluateRequirements(changelog, table.changelogContext)
 

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/interface.scala

@@ -1184,7 +1184,8 @@ case class HiveTableRelation(
     dataCols: Seq[AttributeReference],
     partitionCols: Seq[AttributeReference],
     tableStats: Option[Statistics] = None,
-    @transient prunedPartitions: Option[Seq[CatalogTablePartition]] = None)
+    @transient prunedPartitions: Option[Seq[CatalogTablePartition]] = None,
+    charVarcharScanMode: Option[CharVarcharScanMode] = None)
   extends LeafNode with MultiInstanceRelation with NormalizeableRelation {
   assert(tableMeta.identifier.database.isDefined,
     "Table identifier " + tableMeta.identifier.quotedString + " is missing database name. " +
@@ -1197,6 +1198,11 @@ case class HiveTableRelation(
 
   def isPartitioned: Boolean = partitionCols.nonEmpty
 
+  def hasCharVarchar: Boolean = output.exists { attr =>
+    CharVarcharUtils.hasCharVarchar(attr.dataType) ||
+      CharVarcharUtils.getRawType(attr.metadata).exists(CharVarcharUtils.hasCharVarchar)
+  }
+
   override def doCanonicalize(): HiveTableRelation = copy(
     tableMeta = CatalogTable.normalize(tableMeta),
     dataCols = dataCols.zipWithIndex.map {

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/CharVarcharScanMode.scala

@@ -0,0 +1,79 @@
+/*
+ * 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
+
+import org.apache.spark.sql.internal.SQLConf
+
+/**
+ * The CHAR/VARCHAR scan mode bound to a relation (and its scan) during analysis.
+ *
+ * A relation carries `Option[CharVarcharScanMode]`: `None` means no mode was bound, including for
+ * a relation with no CHAR/VARCHAR columns. A `Some` value pins the mode so that `sameResult`
+ * comparisons and cache reuse keep the two variants distinct.
+ */
+private[sql] sealed trait CharVarcharScanMode
+
+/**
+ * A scan builder that accepts the CHAR/VARCHAR mode captured during relation analysis.
+ *
+ * For example, if `SELECT c FROM t` is analyzed with standard semantics enabled, the relation
+ * binds [[CharVarcharScanMode.SparkStandard]] before the builder creates the physical scan.
+ */
+private[sql] trait SupportsCharVarcharScanMode {
+  def bindCharVarcharScanMode(mode: CharVarcharScanMode): Unit
+}
+
+private[sql] 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 readers that honor this mode so Spark observes the original
+   * value and applies standard CHAR/VARCHAR length checks. Corresponds to standard semantics.
+   * Native Hive ORC with CONVERT_METASTORE_ORC=false does not implement this contract and still
+   * applies native CHAR/VARCHAR truncation.
+   */
+  case object SparkStandard extends CharVarcharScanMode
+
+  /**
+   * Maps the boolean `spark.sql.charVarchar.standardSemantics.enabled` value to the typed mode.
+   */
+  def apply(standardSemantics: Boolean): CharVarcharScanMode =
+    if (standardSemantics) SparkStandard else PreserveNative
+
+  /**
+   * Configures `conf` so analysis binds `mode` and generates the matching read-side Project.
+   */
+  def configure(conf: SQLConf, mode: CharVarcharScanMode): Unit = mode match {
+    case SparkStandard =>
+      conf.setConfString(SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key, "true")
+    case PreserveNative =>
+      conf.setConfString(SQLConf.PRESERVE_CHAR_VARCHAR_TYPE_INFO.key, "true")
+      conf.setConfString(SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key, "false")
+  }
+
+  /** 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")
+  }
+}