[SPARK-49110][SQL] Fix reading metadata columns for tables with CHAR columns

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

@@ -65,6 +65,13 @@ object ApplyCharTypePaddingHelper {
     }
   }
 
+  private[sql] def isReadSidePadding(p: Project): Boolean = {
+    p.projectList.exists {
+      case Alias(e, _) => CharVarcharUtils.containsPaddingForScan(e)
+      case _ => false
+    }
+  }
+
   private[sql] def paddingForStringComparison(
       plan: LogicalPlan,
       padCharCol: Boolean): LogicalPlan = {

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicLogicalOperators.scala

@@ -18,7 +18,7 @@
 package org.apache.spark.sql.catalyst.plans.logical
 
 import org.apache.spark.sql.catalyst.{AliasIdentifier, InternalRow, SQLConfHelper}
-import org.apache.spark.sql.catalyst.analysis.{Analyzer, AnsiTypeCoercion, MultiInstanceRelation, Resolver, TypeCoercion, TypeCoercionBase, UnresolvedUnaryNode}
+import org.apache.spark.sql.catalyst.analysis.{Analyzer, AnsiTypeCoercion, ApplyCharTypePaddingHelper, MultiInstanceRelation, Resolver, TypeCoercion, TypeCoercionBase, UnresolvedUnaryNode}
 import org.apache.spark.sql.catalyst.catalog.{CatalogStorageFormat, CatalogTable}
 import org.apache.spark.sql.catalyst.catalog.CatalogTable.VIEW_STORING_ANALYZED_PLAN
 import org.apache.spark.sql.catalyst.expressions._
@@ -1701,7 +1701,13 @@ case class SubqueryAlias(
 
   override def metadataOutput: Seq[Attribute] = {
     // Propagate metadata columns from leaf nodes through a chain of `SubqueryAlias`.
-    if (child.isInstanceOf[LeafNode] || child.isInstanceOf[SubqueryAlias]) {
+    // Also propagate metadata columns through `Project`s inserted by `ApplyCharTypePadding`.
+    val shouldPropagate = child match {
+      case _: LeafNode | _: SubqueryAlias => true
+      case p: Project if ApplyCharTypePaddingHelper.isReadSidePadding(p) => true
+      case _ => false
+    }
+    if (shouldPropagate) {
       val qualifierList = identifier.qualifier :+ alias
       val nonHiddenMetadataOutput = child.metadataOutput.filter(!_.qualifiedAccessOnly)
       nonHiddenMetadataOutput.map(_.withQualifier(qualifierList))

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

@@ -243,6 +243,14 @@ object CharVarcharUtils extends Logging with SparkCharVarcharUtils {
     }.getOrElse(attr)
   }
 
+  private[sql] def containsPaddingForScan(e: Expression): Boolean = e.exists {
+    case s: StaticInvoke =>
+      s.staticObject == classOf[CharVarcharCodegenUtils] &&
+      s.functionName == "readSidePadding"
+    case _ =>
+      false
+  }
+
   /**
    * Return expressions to apply char type padding for the string comparison between the given
    * attributes. When comparing two char type columns/fields, we need to pad the shorter one to

sql/core/src/test/scala/org/apache/spark/sql/connector/MetadataColumnSuite.scala

@@ -22,6 +22,7 @@ import org.apache.spark.sql.connector.catalog.CatalogV2Implicits._
 import org.apache.spark.sql.connector.catalog.Identifier
 import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
 import org.apache.spark.sql.functions.{col, struct}
+import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.types.IntegerType
 
 class MetadataColumnSuite extends DatasourceV2SQLBase {
@@ -356,4 +357,20 @@ class MetadataColumnSuite extends DatasourceV2SQLBase {
       assert(cols.head.metadataInJSON() == null)
     }
   }
+
+  test("SPARK-49110: Project a metadata column while reading a padded char column") {
+    withSQLConf(SQLConf.READ_SIDE_CHAR_PADDING.key -> "true") {
+      withTable(tbl) {
+        sql(s"CREATE TABLE $tbl (id bigint, data char(1)) PARTITIONED BY (bucket(4, id), id)")
+        sql(s"INSERT INTO $tbl VALUES (1, 'a'), (2, 'b'), (3, 'c')")
+        val sqlQuery = sql(s"SELECT id, data, index, _partition FROM $tbl")
+        val dfQuery = spark.table(tbl).select("id", "data", "index", "_partition")
+
+        Seq(sqlQuery, dfQuery).foreach { query =>
+          checkAnswer(
+            query, Seq(Row(1, "a", 0, "3/1"), Row(2, "b", 0, "0/2"), Row(3, "c", 0, "1/3")))
+        }
+      }
+    }
+  }
 }