From c435a9c7273cfc6ede528b1979f05f73ad5fb4e5 Mon Sep 17 00:00:00 2001 From: Bobby Morck Date: Fri, 31 Jul 2026 11:54:24 -0400 Subject: [PATCH 1/2] Spark: Make the view stored-schema coercion configurable ResolveViews rebuilds a view's output from the stored schema, by position, wrapping each column in an UpCast. UpCast only widens, so a view whose stored type is narrower than what its SQL produces cannot be read at all, and there is no way to relax it. Add spark.sql.iceberg.view.schema-binding-mode, taking its mode names and coercions from Spark's ViewSchemaMode: BINDING (UpCast, the default and current behaviour), COMPENSATION (an ANSI cast, allowing narrowing) and TYPE_EVOLUTION (no cast, so the view reports the types its SQL produces). All three keep the stored column name and metadata. When the conf is unset, Spark's spark.sql.legacy.viewSchemaBindingMode and viewSchemaCompensation are honored instead, reproducing how SessionCatalog.castColToType treats SchemaUnsupported. --- .../sql/catalyst/analysis/ResolveViews.scala | 72 +++++++- .../iceberg/spark/extensions/TestViews.java | 154 ++++++++++++++++++ .../iceberg/spark/SparkSQLProperties.java | 13 ++ .../sql/catalyst/analysis/ResolveViews.scala | 72 +++++++- .../iceberg/spark/extensions/TestViews.java | 154 ++++++++++++++++++ .../iceberg/spark/SparkSQLProperties.java | 13 ++ .../sql/catalyst/analysis/ResolveViews.scala | 72 +++++++- .../iceberg/spark/extensions/TestViews.java | 154 ++++++++++++++++++ .../iceberg/spark/SparkSQLProperties.java | 13 ++ 9 files changed, 708 insertions(+), 9 deletions(-) diff --git a/spark/v3.5/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveViews.scala b/spark/v3.5/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveViews.scala index 4f7e2b4d0f24..9698caec490a 100644 --- a/spark/v3.5/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveViews.scala +++ b/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,74 @@ 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) => + // The declared type here is discarded when the ordinal is resolved to the child attribute, + // so under TYPE_EVOLUTION the column keeps the type the view's SQL produces. 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)) diff --git a/spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestViews.java b/spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestViews.java index 153f580e5ba6..74a89516b0d1 100644 --- a/spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestViews.java +++ b/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,154 @@ 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")); + + // the value is matched without regard to case + 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 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 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 records = Lists.newArrayListWithCapacity(numRows); for (int i = 1; i <= numRows; i++) { diff --git a/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/SparkSQLProperties.java b/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/SparkSQLProperties.java index 4b9509fc8b9b..ef8eaea9c3ef 100644 --- a/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/SparkSQLProperties.java +++ b/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"; } diff --git a/spark/v4.0/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveViews.scala b/spark/v4.0/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveViews.scala index ff7d20241bed..decfd0956492 100644 --- a/spark/v4.0/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveViews.scala +++ b/spark/v4.0/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,74 @@ 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) => + // The declared type here is discarded when the ordinal is resolved to the child attribute, + // so under TYPE_EVOLUTION the column keeps the type the view's SQL produces. 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)) diff --git a/spark/v4.0/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestViews.java b/spark/v4.0/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestViews.java index 237563860366..dc5d67b3c018 100644 --- a/spark/v4.0/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestViews.java +++ b/spark/v4.0/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() { @@ -2120,6 +2126,154 @@ 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")); + + // the value is matched without regard to case + 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 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 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 records = Lists.newArrayListWithCapacity(numRows); for (int i = 1; i <= numRows; i++) { diff --git a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/SparkSQLProperties.java b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/SparkSQLProperties.java index eb4d90292b5d..ad9139ca9416 100644 --- a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/SparkSQLProperties.java +++ b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/SparkSQLProperties.java @@ -131,4 +131,17 @@ private SparkSQLProperties() {} // This determines how many rows are buffered before inferring shredded schema public static final String VARIANT_INFERENCE_BUFFER_SIZE = "spark.sql.iceberg.variant-inference-buffer-size"; + + // 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"; } diff --git a/spark/v4.1/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveViews.scala b/spark/v4.1/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveViews.scala index 76db30a5b619..71958201da1a 100644 --- a/spark/v4.1/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveViews.scala +++ b/spark/v4.1/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,74 @@ 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) => + // The declared type here is discarded when the ordinal is resolved to the child attribute, + // so under TYPE_EVOLUTION the column keeps the type the view's SQL produces. 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)) diff --git a/spark/v4.1/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestViews.java b/spark/v4.1/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestViews.java index 237563860366..dc5d67b3c018 100644 --- a/spark/v4.1/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestViews.java +++ b/spark/v4.1/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() { @@ -2120,6 +2126,154 @@ 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")); + + // the value is matched without regard to case + 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 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 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 records = Lists.newArrayListWithCapacity(numRows); for (int i = 1; i <= numRows; i++) { diff --git a/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/SparkSQLProperties.java b/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/SparkSQLProperties.java index 428644ecfc91..0f659373f7ff 100644 --- a/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/SparkSQLProperties.java +++ b/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/SparkSQLProperties.java @@ -134,4 +134,17 @@ private SparkSQLProperties() {} // This determines how many rows are buffered before inferring shredded schema public static final String VARIANT_INFERENCE_BUFFER_SIZE = "spark.sql.iceberg.variant-inference-buffer-size"; + + // 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"; } From e6871b1502f22650525cf08fd9272c6368aa5afb Mon Sep 17 00:00:00 2001 From: Bobby Morck Date: Fri, 31 Jul 2026 12:17:13 -0400 Subject: [PATCH 2/2] Spark: Drop two redundant comments in the view schema-mode tests and rule --- .../org/apache/spark/sql/catalyst/analysis/ResolveViews.scala | 2 -- .../java/org/apache/iceberg/spark/extensions/TestViews.java | 1 - .../org/apache/spark/sql/catalyst/analysis/ResolveViews.scala | 2 -- .../java/org/apache/iceberg/spark/extensions/TestViews.java | 1 - .../org/apache/spark/sql/catalyst/analysis/ResolveViews.scala | 2 -- .../java/org/apache/iceberg/spark/extensions/TestViews.java | 1 - 6 files changed, 9 deletions(-) diff --git a/spark/v3.5/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveViews.scala b/spark/v3.5/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveViews.scala index 9698caec490a..c5681aef93de 100644 --- a/spark/v3.5/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveViews.scala +++ b/spark/v3.5/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveViews.scala @@ -124,8 +124,6 @@ case class ResolveViews(spark: SparkSession) extends Rule[LogicalPlan] with Look // the stored name and metadata; only the coercion differs. val mode = viewSchemaMode val aliases = view.schema.fields.zipWithIndex.map { case (expected, pos) => - // The declared type here is discarded when the ordinal is resolved to the child attribute, - // so under TYPE_EVOLUTION the column keeps the type the view's SQL produces. val attr = GetColumnByOrdinal(pos, expected.dataType) val coerced = if (mode == SparkSQLProperties.VIEW_SCHEMA_MODE_COMPENSATION) { diff --git a/spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestViews.java b/spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestViews.java index 74a89516b0d1..881d83e5f120 100644 --- a/spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestViews.java +++ b/spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestViews.java @@ -2194,7 +2194,6 @@ public void readFromViewWithNarrowedSchemaHonorsSparkLegacyConf() throws NoSuchT assertViewSchemaCompensates( viewName, ImmutableMap.of(SPARK_BINDING_MODE, "false", SPARK_COMPENSATION, "true")); - // the value is matched without regard to case assertViewSchemaCompensates(viewName, ImmutableMap.of(SPARK_BINDING_MODE, "FALSE")); } diff --git a/spark/v4.0/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveViews.scala b/spark/v4.0/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveViews.scala index decfd0956492..0ffb2e98349a 100644 --- a/spark/v4.0/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveViews.scala +++ b/spark/v4.0/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveViews.scala @@ -124,8 +124,6 @@ case class ResolveViews(spark: SparkSession) extends Rule[LogicalPlan] with Look // the stored name and metadata; only the coercion differs. val mode = viewSchemaMode val aliases = view.schema.fields.zipWithIndex.map { case (expected, pos) => - // The declared type here is discarded when the ordinal is resolved to the child attribute, - // so under TYPE_EVOLUTION the column keeps the type the view's SQL produces. val attr = GetColumnByOrdinal(pos, expected.dataType) val coerced = if (mode == SparkSQLProperties.VIEW_SCHEMA_MODE_COMPENSATION) { diff --git a/spark/v4.0/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestViews.java b/spark/v4.0/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestViews.java index dc5d67b3c018..3ffea0ce1c67 100644 --- a/spark/v4.0/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestViews.java +++ b/spark/v4.0/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestViews.java @@ -2192,7 +2192,6 @@ public void readFromViewWithNarrowedSchemaHonorsSparkLegacyConf() throws NoSuchT assertViewSchemaCompensates( viewName, ImmutableMap.of(SPARK_BINDING_MODE, "false", SPARK_COMPENSATION, "true")); - // the value is matched without regard to case assertViewSchemaCompensates(viewName, ImmutableMap.of(SPARK_BINDING_MODE, "FALSE")); } diff --git a/spark/v4.1/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveViews.scala b/spark/v4.1/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveViews.scala index 71958201da1a..2e864c69e577 100644 --- a/spark/v4.1/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveViews.scala +++ b/spark/v4.1/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveViews.scala @@ -124,8 +124,6 @@ case class ResolveViews(spark: SparkSession) extends Rule[LogicalPlan] with Look // the stored name and metadata; only the coercion differs. val mode = viewSchemaMode val aliases = view.schema.fields.zipWithIndex.map { case (expected, pos) => - // The declared type here is discarded when the ordinal is resolved to the child attribute, - // so under TYPE_EVOLUTION the column keeps the type the view's SQL produces. val attr = GetColumnByOrdinal(pos, expected.dataType) val coerced = if (mode == SparkSQLProperties.VIEW_SCHEMA_MODE_COMPENSATION) { diff --git a/spark/v4.1/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestViews.java b/spark/v4.1/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestViews.java index dc5d67b3c018..3ffea0ce1c67 100644 --- a/spark/v4.1/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestViews.java +++ b/spark/v4.1/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestViews.java @@ -2192,7 +2192,6 @@ public void readFromViewWithNarrowedSchemaHonorsSparkLegacyConf() throws NoSuchT assertViewSchemaCompensates( viewName, ImmutableMap.of(SPARK_BINDING_MODE, "false", SPARK_COMPENSATION, "true")); - // the value is matched without regard to case assertViewSchemaCompensates(viewName, ImmutableMap.of(SPARK_BINDING_MODE, "FALSE")); }