[spark] Support LEGACY store assignment for format tables

paimon-spark/paimon-spark-3.2/src/main/scala/org/apache/paimon/spark/catalyst/Compatibility.scala

@@ -19,7 +19,8 @@
 package org.apache.paimon.spark.catalyst
 
 import org.apache.spark.sql.SparkSession
-import org.apache.spark.sql.catalyst.expressions.{Cast, Expression}
+import org.apache.spark.sql.catalyst.analysis.TableOutputResolver
+import org.apache.spark.sql.catalyst.expressions.{Attribute, Cast, Expression}
 import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, V2WriteCommand}
 import org.apache.spark.sql.catalyst.trees.TreeNodeTag
 import org.apache.spark.sql.execution.ui.SQLPlanMetric
@@ -28,6 +29,26 @@ import org.apache.spark.sql.types.DataType
 
 object Compatibility {
 
+  def resolveTableOutputColumns(
+      tableName: String,
+      expected: Seq[Attribute],
+      query: LogicalPlan,
+      byName: Boolean,
+      conf: SQLConf): LogicalPlan = {
+    // SPARK-38228 fixed this separation in 3.3: LEGACY assignment must use non-ANSI casts even
+    // when ANSI expression evaluation is enabled. Scope the override to this resolution only.
+    val assignmentConf = if (conf.storeAssignmentPolicy == SQLConf.StoreAssignmentPolicy.LEGACY) {
+      val legacyConf = conf.clone()
+      legacyConf.setConf(SQLConf.ANSI_ENABLED, false)
+      legacyConf
+    } else {
+      conf
+    }
+    SQLConf.withExistingConf(assignmentConf) {
+      TableOutputResolver.resolveOutputColumns(tableName, expected, query, byName, assignmentConf)
+    }
+  }
+
   def withNewQuery(o: V2WriteCommand, query: LogicalPlan): V2WriteCommand = {
     o.withNewQuery(query)
   }

paimon-spark/paimon-spark-3.3/src/main/scala/org/apache/paimon/spark/catalyst/Compatibility.scala

@@ -19,7 +19,8 @@
 package org.apache.paimon.spark.catalyst
 
 import org.apache.spark.sql.SparkSession
-import org.apache.spark.sql.catalyst.expressions.{Cast, Expression}
+import org.apache.spark.sql.catalyst.analysis.TableOutputResolver
+import org.apache.spark.sql.catalyst.expressions.{Attribute, Cast, Expression}
 import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, V2WriteCommand}
 import org.apache.spark.sql.catalyst.trees.TreeNodeTag
 import org.apache.spark.sql.execution.ui.SQLPlanMetric
@@ -28,6 +29,15 @@ import org.apache.spark.sql.types.DataType
 
 object Compatibility {
 
+  def resolveTableOutputColumns(
+      tableName: String,
+      expected: Seq[Attribute],
+      query: LogicalPlan,
+      byName: Boolean,
+      conf: SQLConf): LogicalPlan = {
+    TableOutputResolver.resolveOutputColumns(tableName, expected, query, byName, conf)
+  }
+
   def withNewQuery(o: V2WriteCommand, query: LogicalPlan): V2WriteCommand = {
     o.withNewQuery(query)
   }

paimon-spark/paimon-spark-3.4/src/main/scala/org/apache/paimon/spark/catalyst/Compatibility.scala

@@ -0,0 +1,60 @@
+/*
+ * 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.paimon.spark.catalyst
+
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.analysis.TableOutputResolver
+import org.apache.spark.sql.catalyst.expressions.{Attribute, Cast, Expression}
+import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, V2WriteCommand}
+import org.apache.spark.sql.catalyst.trees.TreeNodeTag
+import org.apache.spark.sql.execution.ui.SQLPlanMetric
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types.DataType
+
+object Compatibility {
+
+  def resolveTableOutputColumns(
+      tableName: String,
+      expected: Seq[Attribute],
+      query: LogicalPlan,
+      byName: Boolean,
+      conf: SQLConf): LogicalPlan = {
+    TableOutputResolver.resolveOutputColumns(tableName, expected, query, byName, conf)
+  }
+
+  def withNewQuery(o: V2WriteCommand, query: LogicalPlan): V2WriteCommand = {
+    o.withNewQuery(query)
+  }
+
+  def castByTableInsertionTag: TreeNodeTag[Unit] = {
+    Cast.BY_TABLE_INSERTION
+  }
+
+  def cast(
+      child: Expression,
+      dataType: DataType,
+      timeZoneId: Option[String] = None,
+      ansiEnabled: Boolean = SQLConf.get.ansiEnabled): Cast = {
+    Cast(child, dataType, timeZoneId, ansiEnabled)
+  }
+
+  def getExecutionMetrics(spark: SparkSession, executionId: Long): Seq[SQLPlanMetric] = {
+    spark.sharedState.statusStore.execution(executionId).get.metrics.toSeq
+  }
+}

paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/Compatibility.scala

@@ -19,7 +19,8 @@
 package org.apache.paimon.spark.catalyst
 
 import org.apache.spark.sql.SparkSession
-import org.apache.spark.sql.catalyst.expressions.{Cast, Expression}
+import org.apache.spark.sql.catalyst.analysis.TableOutputResolver
+import org.apache.spark.sql.catalyst.expressions.{Attribute, Cast, Expression}
 import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, V2WriteCommand}
 import org.apache.spark.sql.catalyst.trees.TreeNodeTag
 import org.apache.spark.sql.execution.ui.SQLPlanMetric
@@ -28,6 +29,21 @@ import org.apache.spark.sql.types.DataType
 
 object Compatibility {
 
+  def resolveTableOutputColumns(
+      tableName: String,
+      expected: Seq[Attribute],
+      query: LogicalPlan,
+      byName: Boolean,
+      conf: SQLConf): LogicalPlan = {
+    TableOutputResolver.resolveOutputColumns(
+      tableName,
+      expected,
+      query,
+      byName,
+      conf,
+      supportColDefaultValue = false)
+  }
+
   def withNewQuery(o: V2WriteCommand, query: LogicalPlan): V2WriteCommand = {
     o.withNewQuery(query)
   }

paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/PaimonAnalysis.scala

@@ -52,6 +52,9 @@ class PaimonAnalysis(session: SparkSession) extends Rule[LogicalPlan] {
         case o @ PaimonDynamicPartitionOverwrite(r, d) if o.resolved =>
           PaimonDynamicPartitionOverwriteCommand(r, d, o.query, o.writeOptions, o.isByName)
 
+        case a: V2WriteCommand if PaimonFormatOutputResolver.isLegacyFormatWrite(a) =>
+          PaimonFormatOutputResolver.resolve(a)
+
         case a @ PaimonV2WriteCommand(table)
             if a.query.getTagValue(PAIMON_WRITE_RESOLVED).isEmpty =>
           val options = Options.fromMap(writeOptions(a).asJava)

paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/PaimonFormatOutputResolver.scala

@@ -0,0 +1,59 @@
+/*
+ * 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.paimon.spark.catalyst.analysis
+
+import org.apache.paimon.spark.catalyst.Compatibility
+import org.apache.paimon.spark.format.PaimonFormatTable
+
+import org.apache.spark.sql.catalyst.SQLConfHelper
+import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, V2WriteCommand}
+import org.apache.spark.sql.catalyst.trees.TreeNodeTag
+import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
+import org.apache.spark.sql.internal.SQLConf.StoreAssignmentPolicy
+
+/** Align LEGACY format writes without enabling missing-column filling or schema evolution. */
+object PaimonFormatOutputResolver extends SQLConfHelper {
+
+  private val resolvedQuery = TreeNodeTag[LogicalPlan]("paimon.format.write.resolved-query")
+
+  def isLegacyFormatWrite(write: V2WriteCommand): Boolean = write.table match {
+    case relation: DataSourceV2Relation if relation.table.isInstanceOf[PaimonFormatTable] =>
+      write.query.resolved && conf.storeAssignmentPolicy == StoreAssignmentPolicy.LEGACY
+    case _ => false
+  }
+
+  def resolve(write: V2WriteCommand): V2WriteCommand = {
+    if (write.getTagValue(resolvedQuery).contains(write.query)) {
+      write
+    } else {
+      val query = Compatibility.resolveTableOutputColumns(
+        write.table.name,
+        write.table.output,
+        write.query,
+        write.isByName,
+        conf)
+      val resolved = Compatibility.withNewQuery(write, query)
+      // LEGACY casts may remain nullable for NOT NULL targets. Keep the existing writer-side
+      // nullability checks; do not repeatedly align the query or falsify expression nullability.
+      // Tag the command rather than its source query, which can be reused for another table.
+      resolved.setTagValue(resolvedQuery, query)
+      resolved
+    }
+  }
+}

paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/format/PaimonFormatTable.scala

@@ -34,14 +34,15 @@ import org.apache.spark.sql.catalyst.InternalRow
 import org.apache.spark.sql.catalyst.analysis.{NoSuchPartitionException, NoSuchPartitionsException}
 import org.apache.spark.sql.catalyst.util.CharVarcharUtils
 import org.apache.spark.sql.connector.catalog.{SupportsRead, SupportsWrite, TableCapability, TableCatalog, TruncatableTable}
-import org.apache.spark.sql.connector.catalog.TableCapability.{BATCH_READ, BATCH_WRITE, OVERWRITE_BY_FILTER, OVERWRITE_DYNAMIC}
+import org.apache.spark.sql.connector.catalog.TableCapability.{ACCEPT_ANY_SCHEMA, BATCH_READ, BATCH_WRITE, OVERWRITE_BY_FILTER, OVERWRITE_DYNAMIC}
 import org.apache.spark.sql.connector.distributions.Distribution
 import org.apache.spark.sql.connector.expressions.SortOrder
 import org.apache.spark.sql.connector.read.ScanBuilder
 import org.apache.spark.sql.connector.write._
 import org.apache.spark.sql.connector.write.streaming.StreamingWrite
+import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.paimon.shims.SparkShimLoader
-import org.apache.spark.sql.types.{StringType, StructType}
+import org.apache.spark.sql.types.{DataType, StringType, StructType}
 import org.apache.spark.sql.util.CaseInsensitiveStringMap
 
 import java.util
@@ -71,7 +72,14 @@ case class PaimonFormatTable(table: FormatTable)
   def hasCatalogManagedPartitions: Boolean = partitionManager != null
 
   override def capabilities(): util.Set[TableCapability] = {
-    util.EnumSet.of(BATCH_READ, BATCH_WRITE, OVERWRITE_DYNAMIC, OVERWRITE_BY_FILTER)
+    val capabilities =
+      util.EnumSet.of(BATCH_READ, BATCH_WRITE, OVERWRITE_DYNAMIC, OVERWRITE_BY_FILTER)
+    // Only LEGACY needs our output resolver: Spark's generic V2 analyzer rejects it before
+    // aligning columns. Leave ANSI/STRICT on Spark's path, also when our extension is not loaded.
+    if (SQLConf.get.storeAssignmentPolicy == SQLConf.StoreAssignmentPolicy.LEGACY) {
+      capabilities.add(ACCEPT_ANY_SCHEMA)
+    }
+    capabilities
   }
 
   override def properties: util.Map[String, String] = {
@@ -97,6 +105,13 @@ case class PaimonFormatTable(table: FormatTable)
   }
 
   override def newWriteBuilder(info: LogicalWriteInfo): WriteBuilder = {
+    require(
+      DataType.equalsIgnoreNullability(schema, info.schema),
+      "Format table writes must match the table schema. " +
+        s"Expected ${schema.catalogString}, but found ${info.schema.catalogString}. " +
+        "Configure org.apache.paimon.spark.extensions.PaimonSparkSessionExtensions " +
+        "to resolve write columns and types."
+    )
     PaimonFormatTableWriterBuilder(table, info.schema)
   }