You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by br...@apache.org on 2026/08/27 23:46:57 UTC
(iceberg) branch main updated: Spark: Backport view schema binding mode to Spark 3.5 and 4.0 (#17838)
This is an automated email from the ASF dual-hosted git repository.
bryanck pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg.git
The following commit(s) were added to refs/heads/main by this push:
new a7d17f8e5d Spark: Backport view schema binding mode to Spark 3.5 and 4.0 (#17838)
a7d17f8e5d is described below
commit a7d17f8e5d7d1e098aa84e220fd5c5d71aadcc36
Author: Bobby Morck <32...@users.noreply.github.com>
AuthorDate: Thu Aug 27 19:46:50 2026 -0400
Spark: Backport view schema binding mode to Spark 3.5 and 4.0 (#17838)
---
.../spark/sql/catalyst/analysis/ResolveViews.scala | 34 +++++++++++--
.../apache/iceberg/spark/extensions/TestViews.java | 58 ++++++++++++++++++++++
.../apache/iceberg/spark/SparkSQLProperties.java | 10 ++++
.../spark/sql/catalyst/analysis/ResolveViews.scala | 34 +++++++++++--
.../apache/iceberg/spark/extensions/TestViews.java | 58 ++++++++++++++++++++++
.../apache/iceberg/spark/SparkSQLProperties.java | 10 ++++
6 files changed, 198 insertions(+), 6 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 4f7e2b4d0f..b1ebd6cb12 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
@@ -111,16 +113,42 @@ 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.
+ val compensate = viewSchemaMode == SparkSQLProperties.VIEW_SCHEMA_MODE_COMPENSATION
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 cast = if (compensate) {
+ Cast(attr, expected.dataType, ansiEnabled = true)
+ } else {
+ UpCast(attr, expected.dataType)
+ }
+ Alias(cast, expected.name)(explicitMetadata = Some(expected.metadata))
}.toIndexedSeq
SubqueryAlias(nameParts, Project(aliases, rewritten))
}
+ // Read on every resolution rather than cached, so that SET takes effect within a session.
+ private def viewSchemaMode: String =
+ parseSchemaBindingMode(
+ conf.getConfString(
+ SparkSQLProperties.VIEW_SCHEMA_BINDING_MODE,
+ SparkSQLProperties.VIEW_SCHEMA_MODE_BINDING))
+
+ private def parseSchemaBindingMode(mode: String): String = {
+ val normalized = mode.trim
+ 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 {
+ throw new IllegalArgumentException(
+ s"Invalid value for ${SparkSQLProperties.VIEW_SCHEMA_BINDING_MODE}: $mode, expected " +
+ s"${SparkSQLProperties.VIEW_SCHEMA_MODE_BINDING} or " +
+ s"${SparkSQLProperties.VIEW_SCHEMA_MODE_COMPENSATION}")
+ }
+ }
+
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 153f580e5b..9ea90bed1f 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
@@ -42,6 +42,7 @@ import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
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 +58,7 @@ import org.apache.spark.sql.Dataset;
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;
@@ -2122,6 +2124,62 @@ public class TestViews extends ExtensionsTestBase {
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(spark.table(viewName).schema().fields()[0].dataType())
+ .isEqualTo(DataTypes.LongType);
+ assertThat(sql("SELECT * FROM %s ORDER BY id", viewName))
+ .containsExactly(row(1L), row(2L), row(3L));
+ });
+ }
+
+ @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 insertRows(int numRows) throws NoSuchTableException {
List<SimpleRecord> 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 4b9509fc8b..7c403abf30 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,14 @@ public class 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
+ public static final String VIEW_SCHEMA_BINDING_MODE =
+ "spark.sql.iceberg.view.schema-binding-mode";
+
+ // Permits only widening casts
+ public static final String VIEW_SCHEMA_MODE_BINDING = "BINDING";
+
+ // Permits any ANSI cast, which can truncate values or fail at runtime
+ public static final String VIEW_SCHEMA_MODE_COMPENSATION = "COMPENSATION";
}
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 ff7d20241b..99397bea54 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
@@ -111,16 +113,42 @@ 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.
+ val compensate = viewSchemaMode == SparkSQLProperties.VIEW_SCHEMA_MODE_COMPENSATION
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 cast = if (compensate) {
+ Cast(attr, expected.dataType, ansiEnabled = true)
+ } else {
+ UpCast(attr, expected.dataType)
+ }
+ Alias(cast, expected.name)(explicitMetadata = Some(expected.metadata))
}.toIndexedSeq
SubqueryAlias(nameParts, Project(aliases, rewritten))
}
+ // Read on every resolution rather than cached, so that SET takes effect within a session.
+ private def viewSchemaMode: String =
+ parseSchemaBindingMode(
+ conf.getConfString(
+ SparkSQLProperties.VIEW_SCHEMA_BINDING_MODE,
+ SparkSQLProperties.VIEW_SCHEMA_MODE_BINDING))
+
+ private def parseSchemaBindingMode(mode: String): String = {
+ val normalized = mode.trim
+ 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 {
+ throw new IllegalArgumentException(
+ s"Invalid value for ${SparkSQLProperties.VIEW_SCHEMA_BINDING_MODE}: $mode, expected " +
+ s"${SparkSQLProperties.VIEW_SCHEMA_MODE_BINDING} or " +
+ s"${SparkSQLProperties.VIEW_SCHEMA_MODE_COMPENSATION}")
+ }
+ }
+
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 2375638603..640ea424b3 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
@@ -42,6 +42,7 @@ import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
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 +58,7 @@ import org.apache.spark.sql.Dataset;
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;
@@ -2120,6 +2122,62 @@ public class TestViews extends ExtensionsTestBase {
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(spark.table(viewName).schema().fields()[0].dataType())
+ .isEqualTo(DataTypes.LongType);
+ assertThat(sql("SELECT * FROM %s ORDER BY id", viewName))
+ .containsExactly(row(1L), row(2L), row(3L));
+ });
+ }
+
+ @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 insertRows(int numRows) throws NoSuchTableException {
List<SimpleRecord> 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 eb4d90292b..d69423b761 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,14 @@ public class 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
+ public static final String VIEW_SCHEMA_BINDING_MODE =
+ "spark.sql.iceberg.view.schema-binding-mode";
+
+ // Permits only widening casts
+ public static final String VIEW_SCHEMA_MODE_BINDING = "BINDING";
+
+ // Permits any ANSI cast, which can truncate values or fail at runtime
+ public static final String VIEW_SCHEMA_MODE_COMPENSATION = "COMPENSATION";
}