You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by dw...@apache.org on 2024/12/03 17:54:53 UTC
(iceberg) branch main updated: Spark: Add view support to SparkSessionCatalog (#11388)
This is an automated email from the ASF dual-hosted git repository.
dweeks 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 6501d29b2d Spark: Add view support to SparkSessionCatalog (#11388)
6501d29b2d is described below
commit 6501d29b2d46c8d57f46ad646e2daf7d8865f646
Author: Eduard Tudenhoefner <et...@gmail.com>
AuthorDate: Tue Dec 3 18:54:47 2024 +0100
Spark: Add view support to SparkSessionCatalog (#11388)
* Spark: Add view support to SparkSessionCatalog
* Don't replace view non-atomically
---
.../apache/iceberg/spark/extensions/TestViews.java | 149 ++++++++++++--------
.../java/org/apache/iceberg/spark/BaseCatalog.java | 5 +-
.../org/apache/iceberg/spark/SparkCatalog.java | 3 +-
.../apache/iceberg/spark/SparkSessionCatalog.java | 151 ++++++++++++++++++++-
.../apache/iceberg/spark/SparkCatalogConfig.java | 6 +-
5 files changed, 254 insertions(+), 60 deletions(-)
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 61eafc5f4e..a6149a0ae4 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
@@ -21,12 +21,15 @@ package org.apache.iceberg.spark.extensions;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatNoException;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.assertj.core.api.Assumptions.assumeThat;
import java.util.List;
import java.util.Locale;
import java.util.Random;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
+import org.apache.iceberg.CatalogProperties;
+import org.apache.iceberg.CatalogUtil;
import org.apache.iceberg.IcebergBuild;
import org.apache.iceberg.ParameterizedTestExtension;
import org.apache.iceberg.Parameters;
@@ -35,6 +38,7 @@ import org.apache.iceberg.catalog.Catalog;
import org.apache.iceberg.catalog.Namespace;
import org.apache.iceberg.catalog.TableIdentifier;
import org.apache.iceberg.catalog.ViewCatalog;
+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;
@@ -61,21 +65,30 @@ import org.junit.jupiter.api.extension.ExtendWith;
@ExtendWith(ParameterizedTestExtension.class)
public class TestViews extends ExtensionsTestBase {
private static final Namespace NAMESPACE = Namespace.of("default");
+ private static final String SPARK_CATALOG = "spark_catalog";
private final String tableName = "table";
@BeforeEach
+ @Override
public void before() {
super.before();
spark.conf().set("spark.sql.defaultCatalog", catalogName);
sql("USE %s", catalogName);
sql("CREATE NAMESPACE IF NOT EXISTS %s", NAMESPACE);
- sql("CREATE TABLE %s (id INT, data STRING)", tableName);
+ sql(
+ "CREATE TABLE IF NOT EXISTS %s.%s (id INT, data STRING)%s",
+ NAMESPACE, tableName, catalogName.equals(SPARK_CATALOG) ? " USING iceberg" : "");
+ sql("USE %s.%s", catalogName, NAMESPACE);
}
@AfterEach
public void removeTable() {
sql("USE %s", catalogName);
- sql("DROP TABLE IF EXISTS %s", tableName);
+ sql("DROP TABLE IF EXISTS %s.%s", NAMESPACE, tableName);
+
+ // reset spark session catalog
+ spark.sessionState().catalogManager().reset();
+ spark.conf().unset("spark.sql.catalog.spark_catalog");
}
@Parameters(name = "catalogName = {0}, implementation = {1}, config = {2}")
@@ -85,6 +98,14 @@ public class TestViews extends ExtensionsTestBase {
SparkCatalogConfig.SPARK_WITH_VIEWS.catalogName(),
SparkCatalogConfig.SPARK_WITH_VIEWS.implementation(),
SparkCatalogConfig.SPARK_WITH_VIEWS.properties()
+ },
+ {
+ SparkCatalogConfig.SPARK_SESSION_WITH_VIEWS.catalogName(),
+ SparkCatalogConfig.SPARK_SESSION_WITH_VIEWS.implementation(),
+ ImmutableMap.builder()
+ .putAll(SparkCatalogConfig.SPARK_SESSION_WITH_VIEWS.properties())
+ .put(CatalogProperties.URI, restCatalog.properties().get(CatalogProperties.URI))
+ .build()
}
};
}
@@ -502,11 +523,20 @@ public class TestViews extends ExtensionsTestBase {
assertThat(sql(sql)).hasSize(1).containsExactly(row(5.5));
+ String expectedErrorMsg =
+ String.format("Cannot load function: %s.%s.%s", catalogName, NAMESPACE, functionName);
+ if (SPARK_CATALOG.equals(catalogName)) {
+ // spark session catalog tries to load a V1 function and has a different error msg
+ expectedErrorMsg =
+ String.format(
+ "[ROUTINE_NOT_FOUND] The function `%s`.`%s` cannot be found",
+ NAMESPACE, functionName);
+ }
+
// reading from a view that references a TEMP FUNCTION shouldn't be possible
assertThatThrownBy(() -> sql("SELECT * FROM %s", viewName))
.isInstanceOf(AnalysisException.class)
- .hasMessageStartingWith(
- String.format("Cannot load function: %s.%s.%s", catalogName, NAMESPACE, functionName));
+ .hasMessageStartingWith(expectedErrorMsg);
}
@TestTemplate
@@ -534,6 +564,9 @@ public class TestViews extends ExtensionsTestBase {
@TestTemplate
public void rewriteFunctionIdentifier() {
+ assumeThat(catalogName)
+ .as("system namespace doesn't exist in SparkSessionCatalog")
+ .isNotEqualTo(SPARK_CATALOG);
String viewName = viewName("rewriteFunctionIdentifier");
String sql = "SELECT iceberg_version() AS version";
@@ -579,6 +612,9 @@ public class TestViews extends ExtensionsTestBase {
@TestTemplate
public void rewriteFunctionIdentifierWithNamespace() {
+ assumeThat(catalogName)
+ .as("system namespace doesn't exist in SparkSessionCatalog")
+ .isNotEqualTo(SPARK_CATALOG);
String viewName = viewName("rewriteFunctionIdentifierWithNamespace");
String sql = "SELECT system.bucket(100, 'a') AS bucket_result, 'a' AS value";
@@ -596,8 +632,7 @@ public class TestViews extends ExtensionsTestBase {
assertThatThrownBy(() -> sql(sql))
.isInstanceOf(AnalysisException.class)
- .hasMessageContaining("Cannot resolve function")
- .hasMessageContaining("`system`.`bucket`");
+ .hasMessageContaining("Cannot resolve function `system`.`bucket`");
assertThat(sql("SELECT * FROM %s.%s.%s", catalogName, NAMESPACE, viewName))
.hasSize(1)
@@ -606,6 +641,9 @@ public class TestViews extends ExtensionsTestBase {
@TestTemplate
public void fullFunctionIdentifier() {
+ assumeThat(catalogName)
+ .as("system namespace doesn't exist in SparkSessionCatalog")
+ .isNotEqualTo(SPARK_CATALOG);
String viewName = viewName("fullFunctionIdentifier");
String sql =
String.format(
@@ -754,10 +792,19 @@ public class TestViews extends ExtensionsTestBase {
.withSchema(schema(sql))
.create();
- assertThatThrownBy(() -> sql("ALTER VIEW %s RENAME TO spark_catalog.%s", viewName, renamedView))
+ String targetCatalog =
+ catalogName.equals(SPARK_CATALOG)
+ ? SparkCatalogConfig.SPARK_WITH_VIEWS.catalogName()
+ : SPARK_CATALOG;
+
+ assertThatThrownBy(
+ () ->
+ sql(
+ "ALTER VIEW %s RENAME TO %s.%s.%s",
+ viewName, targetCatalog, NAMESPACE, renamedView))
.isInstanceOf(AnalysisException.class)
.hasMessageContaining(
- "Cannot move view between catalogs: from=spark_with_views and to=spark_catalog");
+ "Cannot move view between catalogs: from=%s and to=%s", catalogName, targetCatalog);
}
@TestTemplate
@@ -813,7 +860,9 @@ public class TestViews extends ExtensionsTestBase {
.withSchema(schema(sql))
.create();
- sql("CREATE TABLE %s (id INT, data STRING)", target);
+ sql(
+ "CREATE TABLE %s.%s.%s (id INT, data STRING)%s",
+ catalogName, NAMESPACE, target, catalogName.equals(SPARK_CATALOG) ? " USING iceberg" : "");
assertThatThrownBy(() -> sql("ALTER VIEW %s RENAME TO %s", viewName, target))
.isInstanceOf(AnalysisException.class)
.hasMessageContaining(
@@ -895,30 +944,6 @@ public class TestViews extends ExtensionsTestBase {
assertThat(v1SessionCatalog().getTempView(tempView).isDefined()).isFalse();
}
- /** The purpose of this test is mainly to make sure that normal view deletion isn't messed up */
- @TestTemplate
- public void dropV1View() {
- String v1View = viewName("v1ViewToBeDropped");
- sql("USE spark_catalog");
- sql("CREATE NAMESPACE IF NOT EXISTS %s", NAMESPACE);
- sql("CREATE TABLE %s (id INT, data STRING)", tableName);
- sql("CREATE VIEW %s AS SELECT id FROM %s", v1View, tableName);
- sql("USE %s", catalogName);
- assertThat(
- v1SessionCatalog()
- .tableExists(new org.apache.spark.sql.catalyst.TableIdentifier(v1View)))
- .isTrue();
-
- sql("DROP VIEW spark_catalog.%s.%s", NAMESPACE, v1View);
- assertThat(
- v1SessionCatalog()
- .tableExists(new org.apache.spark.sql.catalyst.TableIdentifier(v1View)))
- .isFalse();
-
- sql("USE spark_catalog");
- sql("DROP TABLE IF EXISTS %s", tableName);
- }
-
private SessionCatalog v1SessionCatalog() {
return spark.sessionState().catalogManager().v1SessionCatalog();
}
@@ -1316,11 +1341,13 @@ public class TestViews extends ExtensionsTestBase {
assertThat(sql("SELECT * FROM %s", viewName)).hasSize(1).containsExactly(row(5));
- sql("USE spark_catalog");
+ if (!catalogName.equals(SPARK_CATALOG)) {
+ sql("USE spark_catalog");
- assertThatThrownBy(() -> sql(sql))
- .isInstanceOf(AnalysisException.class)
- .hasMessageContaining(String.format("The table or view `%s` cannot be found", tableName));
+ assertThatThrownBy(() -> sql(sql))
+ .isInstanceOf(AnalysisException.class)
+ .hasMessageContaining(String.format("The table or view `%s` cannot be found", tableName));
+ }
// the underlying SQL in the View should be rewritten to have catalog & namespace
assertThat(sql("SELECT * FROM %s.%s.%s", catalogName, NAMESPACE, viewName))
@@ -1341,11 +1368,13 @@ public class TestViews extends ExtensionsTestBase {
.hasSize(3)
.containsExactly(row(3), row(3), row(3));
- sql("USE spark_catalog");
+ if (!catalogName.equals(SPARK_CATALOG)) {
+ sql("USE spark_catalog");
- assertThatThrownBy(() -> sql(sql))
- .isInstanceOf(AnalysisException.class)
- .hasMessageContaining(String.format("The table or view `%s` cannot be found", tableName));
+ assertThatThrownBy(() -> sql(sql))
+ .isInstanceOf(AnalysisException.class)
+ .hasMessageContaining(String.format("The table or view `%s` cannot be found", tableName));
+ }
// the underlying SQL in the View should be rewritten to have catalog & namespace
assertThat(sql("SELECT * FROM %s.%s.%s", catalogName, NAMESPACE, viewName))
@@ -1370,6 +1399,7 @@ public class TestViews extends ExtensionsTestBase {
sql(
"CREATE VIEW %s (new_id COMMENT 'ID', new_data COMMENT 'DATA') COMMENT 'view comment' AS %s",
viewName, sql);
+ String location = viewCatalog().loadView(TableIdentifier.of(NAMESPACE, viewName)).location();
assertThat(sql("DESCRIBE EXTENDED %s", viewName))
.contains(
row("new_id", "int", "ID"),
@@ -1382,8 +1412,8 @@ public class TestViews extends ExtensionsTestBase {
row(
"View Properties",
String.format(
- "['format-version' = '1', 'location' = '/%s/%s', 'provider' = 'iceberg']",
- NAMESPACE, viewName),
+ "['format-version' = '1', 'location' = '%s', 'provider' = 'iceberg']",
+ location),
""));
}
@@ -1441,12 +1471,15 @@ public class TestViews extends ExtensionsTestBase {
row(NAMESPACE.toString(), v1, false),
tempView);
- assertThat(sql("SHOW VIEWS IN %s", catalogName))
- .contains(
- row(NAMESPACE.toString(), prefixV2, false),
- row(NAMESPACE.toString(), prefixV3, false),
- row(NAMESPACE.toString(), v1, false),
- tempView);
+ if (!"rest".equals(catalogConfig.get(CatalogUtil.ICEBERG_CATALOG_TYPE))) {
+ // REST catalog requires a namespace
+ assertThat(sql("SHOW VIEWS IN %s", catalogName))
+ .contains(
+ row(NAMESPACE.toString(), prefixV2, false),
+ row(NAMESPACE.toString(), prefixV3, false),
+ row(NAMESPACE.toString(), v1, false),
+ tempView);
+ }
assertThat(sql("SHOW VIEWS IN %s.%s", catalogName, NAMESPACE))
.contains(
@@ -1461,7 +1494,10 @@ public class TestViews extends ExtensionsTestBase {
assertThat(sql("SHOW VIEWS LIKE 'non-existing'")).isEmpty();
- assertThat(sql("SHOW VIEWS IN spark_catalog.default")).contains(tempView);
+ if (!catalogName.equals(SPARK_CATALOG)) {
+ sql("CREATE NAMESPACE IF NOT EXISTS spark_catalog.%s", NAMESPACE);
+ assertThat(sql("SHOW VIEWS IN spark_catalog.%s", NAMESPACE)).contains(tempView);
+ }
assertThat(sql("SHOW VIEWS IN global_temp"))
.contains(
@@ -1512,6 +1548,7 @@ public class TestViews extends ExtensionsTestBase {
sql("CREATE VIEW %s AS %s", viewName, sql);
+ String location = viewCatalog().loadView(TableIdentifier.of(NAMESPACE, viewName)).location();
String expected =
String.format(
"CREATE VIEW %s.%s.%s (\n"
@@ -1519,10 +1556,10 @@ public class TestViews extends ExtensionsTestBase {
+ " data)\n"
+ "TBLPROPERTIES (\n"
+ " 'format-version' = '1',\n"
- + " 'location' = '/%s/%s',\n"
+ + " 'location' = '%s',\n"
+ " 'provider' = 'iceberg')\n"
+ "AS\n%s\n",
- catalogName, NAMESPACE, viewName, NAMESPACE, viewName, sql);
+ catalogName, NAMESPACE, viewName, location, sql);
assertThat(sql("SHOW CREATE TABLE %s", viewName)).containsExactly(row(expected));
}
@@ -1536,6 +1573,7 @@ public class TestViews extends ExtensionsTestBase {
+ "COMMENT 'view comment' TBLPROPERTIES ('key1'='val1', 'key2'='val2') AS %s",
viewName, sql);
+ String location = viewCatalog().loadView(TableIdentifier.of(NAMESPACE, viewName)).location();
String expected =
String.format(
"CREATE VIEW %s.%s.%s (\n"
@@ -1546,10 +1584,10 @@ public class TestViews extends ExtensionsTestBase {
+ " 'format-version' = '1',\n"
+ " 'key1' = 'val1',\n"
+ " 'key2' = 'val2',\n"
- + " 'location' = '/%s/%s',\n"
+ + " 'location' = '%s',\n"
+ " 'provider' = 'iceberg')\n"
+ "AS\n%s\n",
- catalogName, NAMESPACE, viewName, NAMESPACE, viewName, sql);
+ catalogName, NAMESPACE, viewName, location, sql);
assertThat(sql("SHOW CREATE TABLE %s", viewName)).containsExactly(row(expected));
}
@@ -1897,6 +1935,7 @@ public class TestViews extends ExtensionsTestBase {
@TestTemplate
public void createViewWithRecursiveCycleToV1View() {
+ assumeThat(catalogName).isNotEqualTo(SPARK_CATALOG);
String viewOne = viewName("view_one");
String viewTwo = viewName("view_two");
@@ -1908,7 +1947,7 @@ public class TestViews extends ExtensionsTestBase {
sql("USE %s", catalogName);
// viewOne points to viewTwo points to viewOne, creating a recursive cycle
String view1 = String.format("%s.%s.%s", catalogName, NAMESPACE, viewOne);
- String view2 = String.format("%s.%s.%s", "spark_catalog", NAMESPACE, viewTwo);
+ String view2 = String.format("%s.%s.%s", SPARK_CATALOG, NAMESPACE, viewTwo);
String cycle = String.format("%s -> %s -> %s", view1, view2, view1);
assertThatThrownBy(() -> sql("CREATE OR REPLACE VIEW %s AS SELECT * FROM %s", viewOne, view2))
.isInstanceOf(AnalysisException.class)
diff --git a/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/BaseCatalog.java b/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/BaseCatalog.java
index 2082c05846..78533d163d 100644
--- a/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/BaseCatalog.java
+++ b/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/BaseCatalog.java
@@ -26,6 +26,7 @@ import org.apache.spark.sql.catalyst.analysis.NoSuchProcedureException;
import org.apache.spark.sql.connector.catalog.Identifier;
import org.apache.spark.sql.connector.catalog.StagingTableCatalog;
import org.apache.spark.sql.connector.catalog.SupportsNamespaces;
+import org.apache.spark.sql.connector.catalog.ViewCatalog;
import org.apache.spark.sql.connector.iceberg.catalog.Procedure;
import org.apache.spark.sql.connector.iceberg.catalog.ProcedureCatalog;
import org.apache.spark.sql.util.CaseInsensitiveStringMap;
@@ -35,7 +36,9 @@ abstract class BaseCatalog
ProcedureCatalog,
SupportsNamespaces,
HasIcebergCatalog,
- SupportsFunctions {
+ SupportsFunctions,
+ ViewCatalog,
+ SupportsReplaceView {
private static final String USE_NULLABLE_QUERY_SCHEMA_CTAS_RTAS = "use-nullable-query-schema";
private static final boolean USE_NULLABLE_QUERY_SCHEMA_CTAS_RTAS_DEFAULT = true;
diff --git a/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/SparkCatalog.java b/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/SparkCatalog.java
index 0c36159862..bb8926d34a 100644
--- a/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/SparkCatalog.java
+++ b/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/SparkCatalog.java
@@ -120,8 +120,7 @@ import org.apache.spark.sql.util.CaseInsensitiveStringMap;
*
* <p>
*/
-public class SparkCatalog extends BaseCatalog
- implements org.apache.spark.sql.connector.catalog.ViewCatalog, SupportsReplaceView {
+public class SparkCatalog extends BaseCatalog {
private static final Set<String> DEFAULT_NS_KEYS = ImmutableSet.of(TableCatalog.PROP_OWNER);
private static final Splitter COMMA = Splitter.on(",");
private static final Joiner COMMA_JOINER = Joiner.on(",");
diff --git a/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/SparkSessionCatalog.java b/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/SparkSessionCatalog.java
index fa3f1fbe4b..4137937ecf 100644
--- a/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/SparkSessionCatalog.java
+++ b/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/SparkSessionCatalog.java
@@ -31,8 +31,10 @@ import org.apache.spark.sql.catalyst.analysis.NamespaceAlreadyExistsException;
import org.apache.spark.sql.catalyst.analysis.NoSuchFunctionException;
import org.apache.spark.sql.catalyst.analysis.NoSuchNamespaceException;
import org.apache.spark.sql.catalyst.analysis.NoSuchTableException;
+import org.apache.spark.sql.catalyst.analysis.NoSuchViewException;
import org.apache.spark.sql.catalyst.analysis.NonEmptyNamespaceException;
import org.apache.spark.sql.catalyst.analysis.TableAlreadyExistsException;
+import org.apache.spark.sql.catalyst.analysis.ViewAlreadyExistsException;
import org.apache.spark.sql.connector.catalog.CatalogExtension;
import org.apache.spark.sql.connector.catalog.CatalogPlugin;
import org.apache.spark.sql.connector.catalog.FunctionCatalog;
@@ -44,6 +46,9 @@ import org.apache.spark.sql.connector.catalog.SupportsNamespaces;
import org.apache.spark.sql.connector.catalog.Table;
import org.apache.spark.sql.connector.catalog.TableCatalog;
import org.apache.spark.sql.connector.catalog.TableChange;
+import org.apache.spark.sql.connector.catalog.View;
+import org.apache.spark.sql.connector.catalog.ViewCatalog;
+import org.apache.spark.sql.connector.catalog.ViewChange;
import org.apache.spark.sql.connector.catalog.functions.UnboundFunction;
import org.apache.spark.sql.connector.expressions.Transform;
import org.apache.spark.sql.types.StructType;
@@ -55,13 +60,15 @@ import org.apache.spark.sql.util.CaseInsensitiveStringMap;
* @param <T> CatalogPlugin class to avoid casting to TableCatalog, FunctionCatalog and
* SupportsNamespaces.
*/
-public class SparkSessionCatalog<T extends TableCatalog & FunctionCatalog & SupportsNamespaces>
+public class SparkSessionCatalog<
+ T extends TableCatalog & FunctionCatalog & SupportsNamespaces & ViewCatalog>
extends BaseCatalog implements CatalogExtension {
private static final String[] DEFAULT_NAMESPACE = new String[] {"default"};
private String catalogName = null;
private TableCatalog icebergCatalog = null;
private StagingTableCatalog asStagingCatalog = null;
+ private ViewCatalog asViewCatalog = null;
private T sessionCatalog = null;
private boolean createParquetAsIceberg = false;
private boolean createAvroAsIceberg = false;
@@ -319,6 +326,10 @@ public class SparkSessionCatalog<T extends TableCatalog & FunctionCatalog & Supp
this.asStagingCatalog = (StagingTableCatalog) icebergCatalog;
}
+ if (icebergCatalog instanceof ViewCatalog) {
+ this.asViewCatalog = (ViewCatalog) icebergCatalog;
+ }
+
this.createParquetAsIceberg = options.getBoolean("parquet-enabled", createParquetAsIceberg);
this.createAvroAsIceberg = options.getBoolean("avro-enabled", createAvroAsIceberg);
this.createOrcAsIceberg = options.getBoolean("orc-enabled", createOrcAsIceberg);
@@ -389,6 +400,10 @@ public class SparkSessionCatalog<T extends TableCatalog & FunctionCatalog & Supp
return ((HasIcebergCatalog) icebergCatalog).icebergCatalog();
}
+ private boolean isViewCatalog() {
+ return getSessionCatalog() instanceof ViewCatalog;
+ }
+
@Override
public UnboundFunction loadFunction(Identifier ident) throws NoSuchFunctionException {
try {
@@ -397,4 +412,138 @@ public class SparkSessionCatalog<T extends TableCatalog & FunctionCatalog & Supp
return getSessionCatalog().loadFunction(ident);
}
}
+
+ @Override
+ public Identifier[] listViews(String... namespace) {
+ try {
+ if (null != asViewCatalog) {
+ return asViewCatalog.listViews(namespace);
+ } else if (isViewCatalog()) {
+ getSessionCatalog().listViews(namespace);
+ }
+ } catch (NoSuchNamespaceException e) {
+ throw new RuntimeException(e);
+ }
+
+ return new Identifier[0];
+ }
+
+ @Override
+ public View loadView(Identifier ident) throws NoSuchViewException {
+ if (null != asViewCatalog && asViewCatalog.viewExists(ident)) {
+ return asViewCatalog.loadView(ident);
+ } else if (isViewCatalog() && getSessionCatalog().viewExists(ident)) {
+ return getSessionCatalog().loadView(ident);
+ }
+
+ throw new NoSuchViewException(ident);
+ }
+
+ @Override
+ public View createView(
+ Identifier ident,
+ String sql,
+ String currentCatalog,
+ String[] currentNamespace,
+ StructType schema,
+ String[] queryColumnNames,
+ String[] columnAliases,
+ String[] columnComments,
+ Map<String, String> properties)
+ throws ViewAlreadyExistsException, NoSuchNamespaceException {
+ if (null != asViewCatalog) {
+ return asViewCatalog.createView(
+ ident,
+ sql,
+ currentCatalog,
+ currentNamespace,
+ schema,
+ queryColumnNames,
+ columnAliases,
+ columnComments,
+ properties);
+ } else if (isViewCatalog()) {
+ return getSessionCatalog()
+ .createView(
+ ident,
+ sql,
+ currentCatalog,
+ currentNamespace,
+ schema,
+ queryColumnNames,
+ columnAliases,
+ columnComments,
+ properties);
+ }
+
+ throw new UnsupportedOperationException(
+ "Creating a view is not supported by catalog: " + catalogName);
+ }
+
+ @Override
+ public View replaceView(
+ Identifier ident,
+ String sql,
+ String currentCatalog,
+ String[] currentNamespace,
+ StructType schema,
+ String[] queryColumnNames,
+ String[] columnAliases,
+ String[] columnComments,
+ Map<String, String> properties)
+ throws NoSuchNamespaceException, NoSuchViewException {
+ if (asViewCatalog instanceof SupportsReplaceView) {
+ return ((SupportsReplaceView) asViewCatalog)
+ .replaceView(
+ ident,
+ sql,
+ currentCatalog,
+ currentNamespace,
+ schema,
+ queryColumnNames,
+ columnAliases,
+ columnComments,
+ properties);
+ }
+
+ throw new UnsupportedOperationException(
+ "Replacing a view is not supported by catalog: " + catalogName);
+ }
+
+ @Override
+ public View alterView(Identifier ident, ViewChange... changes)
+ throws NoSuchViewException, IllegalArgumentException {
+ if (null != asViewCatalog && asViewCatalog.viewExists(ident)) {
+ return asViewCatalog.alterView(ident, changes);
+ } else if (isViewCatalog()) {
+ return getSessionCatalog().alterView(ident, changes);
+ }
+
+ throw new UnsupportedOperationException(
+ "Altering a view is not supported by catalog: " + catalogName);
+ }
+
+ @Override
+ public boolean dropView(Identifier ident) {
+ if (null != asViewCatalog && asViewCatalog.viewExists(ident)) {
+ return asViewCatalog.dropView(ident);
+ } else if (isViewCatalog()) {
+ return getSessionCatalog().dropView(ident);
+ }
+
+ return false;
+ }
+
+ @Override
+ public void renameView(Identifier fromIdentifier, Identifier toIdentifier)
+ throws NoSuchViewException, ViewAlreadyExistsException {
+ if (null != asViewCatalog && asViewCatalog.viewExists(fromIdentifier)) {
+ asViewCatalog.renameView(fromIdentifier, toIdentifier);
+ } else if (isViewCatalog()) {
+ getSessionCatalog().renameView(fromIdentifier, toIdentifier);
+ } else {
+ throw new UnsupportedOperationException(
+ "Renaming a view is not supported by catalog: " + catalogName);
+ }
+ }
}
diff --git a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/SparkCatalogConfig.java b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/SparkCatalogConfig.java
index e03f500dc4..ef6c49db57 100644
--- a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/SparkCatalogConfig.java
+++ b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/SparkCatalogConfig.java
@@ -57,7 +57,11 @@ public enum SparkCatalogConfig {
"default-namespace",
"default",
"cache-enabled",
- "false"));
+ "false")),
+ SPARK_SESSION_WITH_VIEWS(
+ "spark_catalog",
+ SparkSessionCatalog.class.getName(),
+ ImmutableMap.of("type", "rest", "default-namespace", "default", "cache-enabled", "false"));
private final String catalogName;
private final String implementation;