Spark: Make the view stored-schema coercion configurable

spark/v3.5/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveViews.scala

@@ -18,10 +18,12 @@
  */
 package org.apache.spark.sql.catalyst.analysis
 
+import org.apache.iceberg.spark.SparkSQLProperties
 import org.apache.spark.sql.SparkSession
 import org.apache.spark.sql.catalyst.FunctionIdentifier
 import org.apache.spark.sql.catalyst.analysis.ViewUtil.IcebergViewHelper
 import org.apache.spark.sql.catalyst.expressions.Alias
+import org.apache.spark.sql.catalyst.expressions.Cast
 import org.apache.spark.sql.catalyst.expressions.SubqueryExpression
 import org.apache.spark.sql.catalyst.expressions.UpCast
 import org.apache.spark.sql.catalyst.parser.ParseException
@@ -45,6 +47,12 @@ case class ResolveViews(spark: SparkSession) extends Rule[LogicalPlan] with Look
 
   protected lazy val catalogManager: CatalogManager = spark.sessionState.catalogManager
 
+  // Spark's own view schema binding confs, SQLConf.VIEW_SCHEMA_BINDING_ENABLED and
+  // VIEW_SCHEMA_COMPENSATION. Referenced by name because they were added in Spark 4.0 and this
+  // rule is also compiled against Spark 3.5. Both default to true.
+  private val sparkViewSchemaBindingMode = "spark.sql.legacy.viewSchemaBindingMode"
+  private val sparkViewSchemaCompensation = "spark.sql.legacy.viewSchemaCompensation"
+
   override def apply(plan: LogicalPlan): LogicalPlan = plan resolveOperators {
     case u @ UnresolvedRelation(nameParts, _, _)
         if catalogManager.v1SessionCatalog.isTempView(nameParts) =>
@@ -111,16 +119,72 @@ case class ResolveViews(spark: SparkSession) extends Rule[LogicalPlan] with Look
 
     // Apply the field aliases and column comments
     // This logic differs from how Spark handles views in SessionCatalog.fromCatalogTable.
-    // This is more strict because it doesn't allow resolution by field name.
+    // BINDING is more strict because it doesn't allow resolution by field name. COMPENSATION and
+    // TYPE_EVOLUTION coerce as SessionCatalog.castColToType does for those modes. Every mode keeps
+    // the stored name and metadata; only the coercion differs.
+    val mode = viewSchemaMode
     val aliases = view.schema.fields.zipWithIndex.map { case (expected, pos) =>
       val attr = GetColumnByOrdinal(pos, expected.dataType)
-      Alias(UpCast(attr, expected.dataType), expected.name)(explicitMetadata =
-        Some(expected.metadata))
+      val coerced =
+        if (mode == SparkSQLProperties.VIEW_SCHEMA_MODE_COMPENSATION) {
+          Cast(attr, expected.dataType, ansiEnabled = true)
+        } else if (mode == SparkSQLProperties.VIEW_SCHEMA_MODE_TYPE_EVOLUTION) {
+          attr
+        } else {
+          UpCast(attr, expected.dataType)
+        }
+      Alias(coerced, expected.name)(explicitMetadata = Some(expected.metadata))
     }.toIndexedSeq
 
     SubqueryAlias(nameParts, Project(aliases, rewritten))
   }
 
+  /**
+   * How a view's stored schema is applied to the columns its SQL produces.
+   *
+   * Read on every resolution rather than cached, so that SET takes effect within a session.
+   */
+  private def viewSchemaMode: String = {
+    spark.conf.getOption(SparkSQLProperties.VIEW_SCHEMA_BINDING_MODE) match {
+      case Some(mode) =>
+        parseSchemaBindingMode(mode)
+      case None =>
+        // Mirror SessionCatalog.castColToType: turning binding mode off selects SchemaUnsupported,
+        // which compensates with an ANSI cast unless compensation is turned off as well. Neither conf
+        // can select TYPE_EVOLUTION: in Spark that mode is requested per view, with
+        // CREATE or ALTER VIEW ... WITH SCHEMA TYPE EVOLUTION, and stored on the view itself.
+        if (isExplicitlyFalse(sparkViewSchemaBindingMode) &&
+          !isExplicitlyFalse(sparkViewSchemaCompensation)) {
+          SparkSQLProperties.VIEW_SCHEMA_MODE_COMPENSATION
+        } else {
+          SparkSQLProperties.VIEW_SCHEMA_MODE_BINDING
+        }
+    }
+  }
+
+  // Spark spells this mode "TYPE EVOLUTION" in its WITH SCHEMA clause, so accept a space as well as
+  // an underscore. Preconditions.checkArgument is avoided: with this many message arguments the call
+  // is an ambiguous overload under Scala 2.12, which spark/v3.5 is cross-built against.
+  private def parseSchemaBindingMode(mode: String): String = {
+    val normalized = mode.trim.replace(' ', '_')
+    if (normalized.equalsIgnoreCase(SparkSQLProperties.VIEW_SCHEMA_MODE_BINDING)) {
+      SparkSQLProperties.VIEW_SCHEMA_MODE_BINDING
+    } else if (normalized.equalsIgnoreCase(SparkSQLProperties.VIEW_SCHEMA_MODE_COMPENSATION)) {
+      SparkSQLProperties.VIEW_SCHEMA_MODE_COMPENSATION
+    } else if (normalized.equalsIgnoreCase(SparkSQLProperties.VIEW_SCHEMA_MODE_TYPE_EVOLUTION)) {
+      SparkSQLProperties.VIEW_SCHEMA_MODE_TYPE_EVOLUTION
+    } else {
+      throw new IllegalArgumentException(
+        s"Invalid value for ${SparkSQLProperties.VIEW_SCHEMA_BINDING_MODE}: $mode, expected " +
+          s"${SparkSQLProperties.VIEW_SCHEMA_MODE_BINDING}, " +
+          s"${SparkSQLProperties.VIEW_SCHEMA_MODE_COMPENSATION} or " +
+          s"${SparkSQLProperties.VIEW_SCHEMA_MODE_TYPE_EVOLUTION}")
+    }
+  }
+
+  private def isExplicitlyFalse(key: String): Boolean =
+    spark.conf.getOption(key).exists(_.trim.equalsIgnoreCase("false"))
+
   private def parseViewText(name: String, viewText: String): LogicalPlan = {
     val origin = Origin(objectType = Some("VIEW"), objectName = Some(name))
 

spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestViews.java

@@ -25,6 +25,7 @@
 
 import java.nio.file.Paths;
 import java.util.List;
+import java.util.Map;
 import java.util.Random;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
@@ -42,6 +43,7 @@
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
 import org.apache.iceberg.spark.Spark3Util;
 import org.apache.iceberg.spark.SparkCatalogConfig;
+import org.apache.iceberg.spark.SparkSQLProperties;
 import org.apache.iceberg.spark.SparkSchemaUtil;
 import org.apache.iceberg.spark.source.SimpleRecord;
 import org.apache.iceberg.types.Types;
@@ -57,6 +59,7 @@
 import org.apache.spark.sql.Row;
 import org.apache.spark.sql.catalyst.analysis.NoSuchTableException;
 import org.apache.spark.sql.catalyst.catalog.SessionCatalog;
+import org.apache.spark.sql.types.DataTypes;
 import org.assertj.core.api.InstanceOfAssertFactories;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
@@ -69,6 +72,9 @@ public class TestViews extends ExtensionsTestBase {
   private static final String SPARK_CATALOG = "spark_catalog";
   private final String tableName = "table";
 
+  private static final String SPARK_BINDING_MODE = "spark.sql.legacy.viewSchemaBindingMode";
+  private static final String SPARK_COMPENSATION = "spark.sql.legacy.viewSchemaCompensation";
+
   @BeforeEach
   @Override
   public void before() {
@@ -2122,6 +2128,153 @@ public void createViewWithCustomMetadataLocationWithLocation() {
     assertThat(location).isEqualTo(customMetadataLocation);
   }
 
+  @TestTemplate
+  public void readFromViewWithNarrowedSchemaFailsUnderBinding() throws NoSuchTableException {
+    insertRows(3);
+    String viewName = viewName("narrowedSchemaView");
+    createViewWithNarrowedSchema(viewName);
+
+    assertThatThrownBy(() -> sql("SELECT * FROM %s", viewName))
+        .isInstanceOf(AnalysisException.class)
+        .hasMessageContaining("Cannot up cast id from \"DOUBLE\" to \"BIGINT\"");
+  }
+
+  @TestTemplate
+  public void readFromViewWithNarrowedSchemaUnderCompensation() throws NoSuchTableException {
+    insertRows(3);
+    String viewName = viewName("compensatedSchemaView");
+    createViewWithNarrowedSchema(viewName);
+
+    withSQLConf(
+        ImmutableMap.of(
+            SparkSQLProperties.VIEW_SCHEMA_BINDING_MODE,
+            SparkSQLProperties.VIEW_SCHEMA_MODE_COMPENSATION),
+        () ->
+            assertThat(sql("SELECT * FROM %s ORDER BY id", viewName))
+                .containsExactly(row(1L), row(2L), row(3L)));
+  }
+
+  @TestTemplate
+  public void readFromViewWithNarrowedSchemaUnderTypeEvolution() throws NoSuchTableException {
+    insertRows(3);
+    String viewName = viewName("typeEvolutionSchemaView");
+    createViewWithNarrowedSchema(viewName);
+
+    withSQLConf(
+        ImmutableMap.of(
+            SparkSQLProperties.VIEW_SCHEMA_BINDING_MODE,
+            SparkSQLProperties.VIEW_SCHEMA_MODE_TYPE_EVOLUTION),
+        () -> {
+          assertThat(spark.table(viewName).schema().fields()[0].dataType())
+              .isEqualTo(DataTypes.DoubleType);
+          assertThat(spark.table(viewName).schema().fields()[0].name()).isEqualTo("id");
+          assertThat(sql("SELECT * FROM %s ORDER BY id", viewName))
+              .containsExactly(row(1.0d), row(2.0d), row(3.0d));
+        });
+  }
+
+  @TestTemplate
+  public void readFromViewWithNarrowedSchemaHonorsSparkLegacyConf() throws NoSuchTableException {
+    insertRows(3);
+    String viewName = viewName("legacyConfSchemaView");
+    createViewWithNarrowedSchema(viewName);
+
+    assertViewSchemaIsStrict(viewName, ImmutableMap.of());
+    assertViewSchemaIsStrict(viewName, ImmutableMap.of(SPARK_BINDING_MODE, "true"));
+    assertViewSchemaIsStrict(viewName, ImmutableMap.of(SPARK_COMPENSATION, "true"));
+    assertViewSchemaIsStrict(viewName, ImmutableMap.of(SPARK_COMPENSATION, "false"));
+    assertViewSchemaIsStrict(
+        viewName, ImmutableMap.of(SPARK_BINDING_MODE, "true", SPARK_COMPENSATION, "true"));
+    assertViewSchemaIsStrict(
+        viewName, ImmutableMap.of(SPARK_BINDING_MODE, "true", SPARK_COMPENSATION, "false"));
+    assertViewSchemaIsStrict(
+        viewName, ImmutableMap.of(SPARK_BINDING_MODE, "false", SPARK_COMPENSATION, "false"));
+
+    assertViewSchemaCompensates(viewName, ImmutableMap.of(SPARK_BINDING_MODE, "false"));
+    assertViewSchemaCompensates(
+        viewName, ImmutableMap.of(SPARK_BINDING_MODE, "false", SPARK_COMPENSATION, "true"));
+
+    assertViewSchemaCompensates(viewName, ImmutableMap.of(SPARK_BINDING_MODE, "FALSE"));
+  }
+
+  @TestTemplate
+  public void icebergViewSchemaBindingModeOverridesSparkLegacyConf() throws NoSuchTableException {
+    insertRows(3);
+    String viewName = viewName("precedenceSchemaView");
+    createViewWithNarrowedSchema(viewName);
+
+    withSQLConf(
+        ImmutableMap.of(
+            SparkSQLProperties.VIEW_SCHEMA_BINDING_MODE,
+            SparkSQLProperties.VIEW_SCHEMA_MODE_BINDING,
+            SPARK_BINDING_MODE,
+            "false"),
+        () ->
+            assertThatThrownBy(() -> sql("SELECT * FROM %s", viewName))
+                .isInstanceOf(AnalysisException.class)
+                .hasMessageContaining("Cannot up cast id"));
+  }
+
+  @TestTemplate
+  public void schemaBindingModeAcceptsSparkSpelling() throws NoSuchTableException {
+    insertRows(3);
+    String viewName = viewName("sparkSpellingSchemaView");
+    createViewWithNarrowedSchema(viewName);
+
+    // Spark writes this mode as "TYPE EVOLUTION" in its WITH SCHEMA clause
+    withSQLConf(
+        ImmutableMap.of(SparkSQLProperties.VIEW_SCHEMA_BINDING_MODE, "type evolution"),
+        () ->
+            assertThat(spark.table(viewName).schema().fields()[0].dataType())
+                .isEqualTo(DataTypes.DoubleType));
+  }
+
+  @TestTemplate
+  public void readFromViewWithInvalidSchemaBindingMode() throws NoSuchTableException {
+    insertRows(3);
+    String viewName = viewName("invalidModeSchemaView");
+    createViewWithNarrowedSchema(viewName);
+
+    withSQLConf(
+        ImmutableMap.of(SparkSQLProperties.VIEW_SCHEMA_BINDING_MODE, "evolution"),
+        () ->
+            assertThatThrownBy(() -> sql("SELECT * FROM %s", viewName))
+                .hasMessageContaining(
+                    "Invalid value for spark.sql.iceberg.view.schema-binding-mode: evolution"));
+  }
+
+  private void createViewWithNarrowedSchema(String viewName) {
+    String storedSchemaSQL = String.format("SELECT CAST(id AS bigint) AS id FROM %s", tableName);
+    String viewSQL = String.format("SELECT CAST(id AS double) AS id FROM %s", tableName);
+
+    viewCatalog()
+        .buildView(TableIdentifier.of(NAMESPACE, viewName))
+        .withQuery("spark", viewSQL)
+        .withDefaultNamespace(NAMESPACE)
+        .withDefaultCatalog(catalogName)
+        .withSchema(schema(storedSchemaSQL))
+        .create();
+  }
+
+  private void assertViewSchemaIsStrict(String viewName, Map<String, String> conf) {
+    withSQLConf(
+        conf,
+        () ->
+            assertThatThrownBy(() -> sql("SELECT * FROM %s", viewName))
+                .as("expected strict binding with %s", conf)
+                .isInstanceOf(AnalysisException.class)
+                .hasMessageContaining("Cannot up cast id"));
+  }
+
+  private void assertViewSchemaCompensates(String viewName, Map<String, String> conf) {
+    withSQLConf(
+        conf,
+        () ->
+            assertThat(sql("SELECT * FROM %s ORDER BY id", viewName))
+                .as("expected an ANSI cast with %s", conf)
+                .containsExactly(row(1L), row(2L), row(3L)));
+  }
+
   private void insertRows(int numRows) throws NoSuchTableException {
     List<SimpleRecord> records = Lists.newArrayListWithCapacity(numRows);
     for (int i = 1; i <= numRows; i++) {

spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/SparkSQLProperties.java

@@ -120,4 +120,17 @@ private SparkSQLProperties() {}
   // defaults to max(spark.default.parallelism, spark.sql.shuffle.partitions).
   public static final String READ_ADAPTIVE_SPLIT_SIZE_PARALLELISM =
       "spark.sql.iceberg.read.adaptive-split-size.parallelism";
+
+  // Controls how a view's stored schema is applied to the columns its SQL produces. The modes and
+  // their coercions match Spark's ViewSchemaMode, which only applies to Spark's own views.
+  // BINDING permits only widening casts, so stored-vs-query type drift fails resolution.
+  // COMPENSATION permits any ANSI cast, which can truncate values or fail at runtime.
+  // TYPE_EVOLUTION applies no cast, so the view reports the types its SQL produces.
+  // When unset, Spark's spark.sql.legacy.viewSchemaBindingMode and
+  // spark.sql.legacy.viewSchemaCompensation are honored instead; neither can select TYPE_EVOLUTION.
+  public static final String VIEW_SCHEMA_BINDING_MODE =
+      "spark.sql.iceberg.view.schema-binding-mode";
+  public static final String VIEW_SCHEMA_MODE_BINDING = "BINDING";
+  public static final String VIEW_SCHEMA_MODE_COMPENSATION = "COMPENSATION";
+  public static final String VIEW_SCHEMA_MODE_TYPE_EVOLUTION = "TYPE_EVOLUTION";
 }