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..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 @@ -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)) 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..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 @@ -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 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..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 @@ -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)) 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..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 @@ -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,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 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..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 @@ -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)) 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..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 @@ -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,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 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"; }