From beff34252eafe6a50001bad51afcffa4ce65e0c5 Mon Sep 17 00:00:00 2001 From: Talat Uyarer Date: Fri, 28 Aug 2026 01:06:00 -0700 Subject: [PATCH 1/5] Enhance FlinkCatalog to support view operations such as list, load --- .../apache/iceberg/flink/FlinkCatalog.java | 91 +++++- .../iceberg/flink/TestFlinkCatalogView.java | 301 ++++++++++++++++++ 2 files changed, 383 insertions(+), 9 deletions(-) create mode 100644 flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogView.java diff --git a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java index 1d1505a28c05..c57e875a1974 100644 --- a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java +++ b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java @@ -36,6 +36,7 @@ import org.apache.flink.table.catalog.CatalogPartition; import org.apache.flink.table.catalog.CatalogPartitionSpec; import org.apache.flink.table.catalog.CatalogTable; +import org.apache.flink.table.catalog.CatalogView; import org.apache.flink.table.catalog.ObjectPath; import org.apache.flink.table.catalog.ResolvedCatalogTable; import org.apache.flink.table.catalog.ResolvedSchema; @@ -67,9 +68,11 @@ import org.apache.iceberg.catalog.Namespace; import org.apache.iceberg.catalog.SupportsNamespaces; import org.apache.iceberg.catalog.TableIdentifier; +import org.apache.iceberg.catalog.ViewCatalog; import org.apache.iceberg.exceptions.AlreadyExistsException; import org.apache.iceberg.exceptions.NamespaceNotEmptyException; import org.apache.iceberg.exceptions.NoSuchNamespaceException; +import org.apache.iceberg.exceptions.NoSuchViewException; import org.apache.iceberg.flink.util.FlinkAlterTableUtil; import org.apache.iceberg.flink.util.FlinkCompatibilityUtil; import org.apache.iceberg.io.CloseableIterable; @@ -80,6 +83,10 @@ import org.apache.iceberg.relocated.com.google.common.collect.Lists; import org.apache.iceberg.relocated.com.google.common.collect.Maps; import org.apache.iceberg.relocated.com.google.common.collect.Sets; +import org.apache.iceberg.view.SQLViewRepresentation; +import org.apache.iceberg.view.View; +import org.apache.iceberg.view.ViewProperties; +import org.apache.iceberg.view.ViewVersion; /** * A Flink Catalog implementation that wraps an Iceberg {@link Catalog}. @@ -94,10 +101,15 @@ */ @Internal public class FlinkCatalog extends AbstractCatalog { + private static final String FLINK_DIALECT = "flink"; + private static final String DEFAULT_CATALOG_OPTION = "default-catalog"; + private static final String DEFAULT_NAMESPACE_OPTION = "default-namespace"; + private final CatalogLoader catalogLoader; private final Catalog icebergCatalog; private final Namespace baseNamespace; private final SupportsNamespaces asNamespaceCatalog; + private final ViewCatalog asViewCatalog; private final Closeable closeable; private final boolean cacheEnabled; @@ -120,6 +132,8 @@ public FlinkCatalog( : originalCatalog; asNamespaceCatalog = originalCatalog instanceof SupportsNamespaces ? (SupportsNamespaces) originalCatalog : null; + // the caching wrapper only caches tables, so view operations always go to the original catalog + asViewCatalog = originalCatalog instanceof ViewCatalog ? (ViewCatalog) originalCatalog : null; closeable = originalCatalog instanceof Closeable ? (Closeable) originalCatalog : null; FlinkEnvironmentContext.init(); @@ -321,8 +335,30 @@ public void alterDatabase(String name, CatalogDatabase newDatabase, boolean igno @Override public List listTables(String databaseName) throws DatabaseNotExistException, CatalogException { + List results; + try { + results = + icebergCatalog.listTables(appendLevel(baseNamespace, databaseName)).stream() + .map(TableIdentifier::name) + .collect(Collectors.toList()); + } catch (NoSuchNamespaceException e) { + throw new DatabaseNotExistException(getName(), databaseName, e); + } + + // Flink's Catalog#listTables contract requires this to return both tables and views + results.addAll(listViews(databaseName)); + return results; + } + + @Override + public List listViews(String databaseName) + throws DatabaseNotExistException, CatalogException { + if (asViewCatalog == null) { + return Collections.emptyList(); + } + try { - return icebergCatalog.listTables(appendLevel(baseNamespace, databaseName)).stream() + return asViewCatalog.listViews(appendLevel(baseNamespace, databaseName)).stream() .map(TableIdentifier::name) .collect(Collectors.toList()); } catch (NoSuchNamespaceException e) { @@ -331,9 +367,26 @@ public List listTables(String databaseName) } @Override - public CatalogTable getTable(ObjectPath tablePath) + public CatalogBaseTable getTable(ObjectPath tablePath) throws TableNotExistException, CatalogException { - Table table = loadIcebergTable(tablePath); + Table table; + try { + table = loadIcebergTable(tablePath); + } catch (TableNotExistException e) { + // metadata tables ("name$type") can never be views, and without a view catalog there is + // nothing else to look up + if (asViewCatalog == null || tablePath.getObjectName().contains("$")) { + throw e; + } + + try { + View view = asViewCatalog.loadView(toIdentifier(tablePath)); + return toCatalogView(view); + } catch (NoSuchViewException viewException) { + e.addSuppressed(viewException); + throw e; + } + } // Flink's CREATE TABLE LIKE clause relies on properties sent back here to create new table. // As Flink API accepts only Map for props, here we are serializing catalog @@ -378,7 +431,9 @@ private Table loadIcebergTable(ObjectPath tablePath) throws TableNotExistExcepti @Override public boolean tableExists(ObjectPath tablePath) throws CatalogException { - return icebergCatalog.tableExists(toIdentifier(tablePath)); + TableIdentifier identifier = toIdentifier(tablePath); + return icebergCatalog.tableExists(identifier) + || (asViewCatalog != null && asViewCatalog.viewExists(identifier)); } @Override @@ -683,6 +738,29 @@ static CatalogTable toCatalogTable(Table table) { return toCatalogTableWithProps(table, table.properties()); } + private CatalogView toCatalogView(View view) { + SQLViewRepresentation sqlRepresentation = view.sqlFor(FLINK_DIALECT); + Preconditions.checkState(sqlRepresentation != null, "Cannot load SQL for view %s", view.name()); + + ResolvedSchema resolvedSchema = FlinkSchemaUtil.toResolvedSchema(view.schema()); + org.apache.flink.table.api.Schema schema = + org.apache.flink.table.api.Schema.newBuilder().fromResolvedSchema(resolvedSchema).build(); + + Map options = Maps.newHashMap(view.properties()); + String comment = options.remove(ViewProperties.COMMENT); + + ViewVersion currentVersion = view.currentVersion(); + String defaultCatalog = + currentVersion.defaultCatalog() != null ? currentVersion.defaultCatalog() : getName(); + options.put(DEFAULT_CATALOG_OPTION, defaultCatalog); + options.put(DEFAULT_NAMESPACE_OPTION, currentVersion.defaultNamespace().toString()); + + // both original and expanded query hold the stored SQL: Flink expands the query itself, + // resolving unqualified references against the view's own catalog and database + return CatalogView.of( + schema, comment, sqlRepresentation.sql(), sqlRepresentation.sql(), options); + } + @Override public Optional getFactory() { return Optional.of(new FlinkDynamicTableFactory(this)); @@ -695,11 +773,6 @@ CatalogLoader getCatalogLoader() { // ------------------------------ Unsupported methods // --------------------------------------------- - @Override - public List listViews(String databaseName) throws CatalogException { - return Collections.emptyList(); - } - @Override public CatalogPartition getPartition(ObjectPath tablePath, CatalogPartitionSpec partitionSpec) throws CatalogException { diff --git a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogView.java b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogView.java new file mode 100644 index 000000000000..b87a305d55ce --- /dev/null +++ b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogView.java @@ -0,0 +1,301 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.iceberg.flink; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.assertj.core.api.Assumptions.assumeThat; + +import java.util.List; +import org.apache.flink.table.catalog.CatalogBaseTable; +import org.apache.flink.table.catalog.CatalogView; +import org.apache.flink.table.catalog.ObjectPath; +import org.apache.flink.table.catalog.exceptions.TableNotExistException; +import org.apache.flink.types.Row; +import org.apache.iceberg.Schema; +import org.apache.iceberg.catalog.TableIdentifier; +import org.apache.iceberg.catalog.ViewCatalog; +import org.apache.iceberg.relocated.com.google.common.collect.Lists; +import org.apache.iceberg.types.Types; +import org.apache.iceberg.view.View; +import org.apache.iceberg.view.ViewProperties; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.TestTemplate; + +public class TestFlinkCatalogView extends CatalogTestBase { + + private static final String TABLE_NAME = "test_table"; + private static final String VIEW_NAME = "test_view"; + + private static final Schema VIEW_SCHEMA = + new Schema( + Types.NestedField.optional(1, "id", Types.LongType.get()), + Types.NestedField.optional(2, "data", Types.StringType.get())); + + private static final Schema PROJECTED_VIEW_SCHEMA = + new Schema(Types.NestedField.optional(1, "id", Types.LongType.get())); + + @Override + @BeforeEach + public void before() { + super.before(); + assumeThat(isHadoopCatalog).as("HadoopCatalog does not implement ViewCatalog").isFalse(); + sql("CREATE DATABASE %s", flinkDatabase); + sql("USE CATALOG %s", catalogName); + sql("USE %s", DATABASE); + sql("CREATE TABLE %s (id BIGINT, data STRING)", TABLE_NAME); + sql("INSERT INTO %s VALUES (1, 'a'), (2, 'b'), (3, 'c')", TABLE_NAME); + } + + @AfterEach + public void cleanNamespaces() { + if (validationCatalog instanceof ViewCatalog) { + ViewCatalog viewCatalog = (ViewCatalog) validationCatalog; + viewCatalog.listViews(icebergNamespace).forEach(viewCatalog::dropView); + } + + sql("DROP TABLE IF EXISTS %s.%s", flinkDatabase, TABLE_NAME); + dropDatabase(flinkDatabase, true); + super.clean(); + } + + private ViewCatalog viewCatalog() { + assertThat(validationCatalog).isInstanceOf(ViewCatalog.class); + return (ViewCatalog) validationCatalog; + } + + private View createView(String dialect, String query) { + return viewCatalog() + .buildView(TableIdentifier.of(icebergNamespace, VIEW_NAME)) + .withSchema(VIEW_SCHEMA) + .withDefaultNamespace(icebergNamespace) + .withQuery(dialect, query) + .create(); + } + + private List expectedRows() { + return Lists.newArrayList(Row.of(1L, "a"), Row.of(2L, "b"), Row.of(3L, "c")); + } + + @TestTemplate + public void testSelectFromView() { + createView("flink", "SELECT id, data FROM test_table"); + assertSameElements(expectedRows(), sql("SELECT * FROM %s", VIEW_NAME)); + } + + @TestTemplate + public void testSelectFromProjectedView() { + viewCatalog() + .buildView(TableIdentifier.of(icebergNamespace, VIEW_NAME)) + .withSchema(PROJECTED_VIEW_SCHEMA) + .withDefaultNamespace(icebergNamespace) + .withQuery("flink", "SELECT id FROM test_table") + .create(); + assertSameElements( + Lists.newArrayList(Row.of(1L), Row.of(2L), Row.of(3L)), sql("SELECT * FROM %s", VIEW_NAME)); + } + + @TestTemplate + public void testSelectFromViewWithQualifiedQuery() { + createView( + "flink", String.format("SELECT id, data FROM %s.%s.test_table", catalogName, DATABASE)); + assertSameElements(expectedRows(), sql("SELECT * FROM %s", VIEW_NAME)); + } + + @TestTemplate + public void testSelectViewFromDifferentDatabase() { + // unqualified references in the stored SQL must resolve against the view's own database, + // not the session's current database (Flink expands views with the view's schema path) + createView("flink", "SELECT id, data FROM test_table"); + sql("CREATE DATABASE %s.db2", catalogName); + sql("USE db2"); + try { + assertSameElements( + expectedRows(), sql("SELECT * FROM %s.%s.%s", catalogName, DATABASE, VIEW_NAME)); + } finally { + sql("USE %s", DATABASE); + dropDatabase(catalogName + ".db2", true); + } + } + + @TestTemplate + public void testViewReferencingAnotherView() { + createView("flink", "SELECT id, data FROM test_table"); + viewCatalog() + .buildView(TableIdentifier.of(icebergNamespace, "second_view")) + .withSchema(PROJECTED_VIEW_SCHEMA) + .withDefaultNamespace(icebergNamespace) + .withQuery("flink", "SELECT id FROM " + VIEW_NAME) + .create(); + assertSameElements( + Lists.newArrayList(Row.of(1L), Row.of(2L), Row.of(3L)), sql("SELECT * FROM second_view")); + } + + @TestTemplate + public void testSqlForFallsBackToAnotherDialect() throws Exception { + // BaseView#sqlFor returns the first SQL representation when no "flink" one exists, + // e.g. a view created by Spark + createView("spark", "SELECT id, data FROM test_table"); + + CatalogView catalogView = + (CatalogView) + getTableEnv() + .getCatalog(catalogName) + .get() + .getTable(new ObjectPath(DATABASE, VIEW_NAME)); + assertThat(catalogView.getOriginalQuery()).isEqualTo("SELECT id, data FROM test_table"); + + // ANSI SQL that both engines understand is directly usable + assertSameElements(expectedRows(), sql("SELECT * FROM %s", VIEW_NAME)); + } + + @TestTemplate + public void testSqlForPrefersExactDialectMatch() throws Exception { + viewCatalog() + .buildView(TableIdentifier.of(icebergNamespace, VIEW_NAME)) + .withSchema(VIEW_SCHEMA) + .withDefaultNamespace(icebergNamespace) + .withQuery("spark", "SELECT id, data FROM spark_only_table") + .withQuery("flink", "SELECT id, data FROM test_table") + .create(); + + CatalogView catalogView = + (CatalogView) + getTableEnv() + .getCatalog(catalogName) + .get() + .getTable(new ObjectPath(DATABASE, VIEW_NAME)); + assertThat(catalogView.getOriginalQuery()).isEqualTo("SELECT id, data FROM test_table"); + assertSameElements(expectedRows(), sql("SELECT * FROM %s", VIEW_NAME)); + } + + @TestTemplate + public void testViewCommentAndProperties() throws Exception { + viewCatalog() + .buildView(TableIdentifier.of(icebergNamespace, VIEW_NAME)) + .withSchema(VIEW_SCHEMA) + .withDefaultNamespace(icebergNamespace) + .withQuery("flink", "SELECT id, data FROM test_table") + .withProperty(ViewProperties.COMMENT, "view comment") + .withProperty("key1", "value1") + .create(); + + CatalogBaseTable catalogView = + getTableEnv().getCatalog(catalogName).get().getTable(new ObjectPath(DATABASE, VIEW_NAME)); + assertThat(catalogView.getComment()).isEqualTo("view comment"); + assertThat(catalogView.getOptions()) + .containsEntry("key1", "value1") + .doesNotContainKey(ViewProperties.COMMENT); + } + + @TestTemplate + public void testDefaultCatalogFallsBackToCatalogName() throws Exception { + // no withDefaultCatalog() -> defaultCatalog() is null -> falls back to the Flink catalog name + createView("flink", "SELECT id, data FROM test_table"); + + CatalogBaseTable catalogView = + getTableEnv().getCatalog(catalogName).get().getTable(new ObjectPath(DATABASE, VIEW_NAME)); + assertThat(catalogView.getOptions()) + .containsEntry("default-catalog", catalogName) + .containsEntry("default-namespace", icebergNamespace.toString()); + } + + @TestTemplate + public void testListViews() throws Exception { + assertThat(sql("SHOW VIEWS")).isEmpty(); + createView("flink", "SELECT id, data FROM test_table"); + assertThat(sql("SHOW VIEWS")).containsExactly(Row.of(VIEW_NAME)); + assertThat(getTableEnv().getCatalog(catalogName).get().listViews(DATABASE)) + .containsExactly(VIEW_NAME); + } + + @TestTemplate + public void testListTablesIncludesViews() { + createView("flink", "SELECT id, data FROM test_table"); + // Flink's Catalog#listTables contract covers both tables and views + assertThat(sql("SHOW TABLES")).containsExactlyInAnyOrder(Row.of(TABLE_NAME), Row.of(VIEW_NAME)); + assertThat(sql("SHOW VIEWS")).containsExactly(Row.of(VIEW_NAME)); + } + + @TestTemplate + public void testTableExistsForView() { + createView("flink", "SELECT id, data FROM test_table"); + org.apache.flink.table.catalog.Catalog flinkCatalog = + getTableEnv().getCatalog(catalogName).get(); + assertThat(flinkCatalog.tableExists(new ObjectPath(DATABASE, VIEW_NAME))).isTrue(); + assertThat(flinkCatalog.tableExists(new ObjectPath(DATABASE, TABLE_NAME))).isTrue(); + assertThat(flinkCatalog.tableExists(new ObjectPath(DATABASE, "nonexistent"))).isFalse(); + } + + @TestTemplate + public void testViewNotExist() { + assertThatThrownBy( + () -> + getTableEnv() + .getCatalog(catalogName) + .get() + .getTable(new ObjectPath(DATABASE, "nonexistent"))) + .isInstanceOf(TableNotExistException.class) + .hasMessageContaining("Table (or view) db.nonexistent does not exist"); + assertThatThrownBy(() -> sql("SELECT * FROM nonexistent")) + .isInstanceOf(org.apache.flink.table.api.ValidationException.class) + .hasMessageContaining("Object 'nonexistent' not found"); + } + + @TestTemplate + public void testMetadataTableNotRoutedToViewPath() { + createView("flink", "SELECT id, data FROM test_table"); + // metadata table access must keep working and never hit the view branch + assertThat(sql("SELECT * FROM %s$snapshots", TABLE_NAME)).isNotEmpty(); + assertThatThrownBy( + () -> + getTableEnv() + .getCatalog(catalogName) + .get() + .getTable(new ObjectPath(DATABASE, "nonexistent$snapshots"))) + .isInstanceOf(TableNotExistException.class) + .hasMessageContaining("Table (or view) db.nonexistent$snapshots does not exist"); + } + + @TestTemplate + public void testDescribeView() { + createView("flink", "SELECT id, data FROM test_table"); + assertThat(sql("DESCRIBE %s", VIEW_NAME)) + .extracting(row -> row.getField(0)) + .containsExactly("id", "data"); + } + + @TestTemplate + public void testGetViewViaCatalogApi() throws Exception { + createView("flink", "SELECT id, data FROM test_table"); + + CatalogBaseTable catalogBaseTable = + getTableEnv().getCatalog(catalogName).get().getTable(new ObjectPath(DATABASE, VIEW_NAME)); + + assertThat(catalogBaseTable.getTableKind()).isEqualTo(CatalogBaseTable.TableKind.VIEW); + assertThat(catalogBaseTable).isInstanceOf(CatalogView.class); + CatalogView catalogView = (CatalogView) catalogBaseTable; + assertThat(catalogView.getOriginalQuery()).isEqualTo("SELECT id, data FROM test_table"); + assertThat(catalogView.getExpandedQuery()).isEqualTo("SELECT id, data FROM test_table"); + assertThat(catalogView.getUnresolvedSchema().getColumns()) + .extracting(org.apache.flink.table.api.Schema.UnresolvedColumn::getName) + .containsExactly("id", "data"); + } +} From a68e7c7e03f2e042f1633472c65f06f43c8df43d Mon Sep 17 00:00:00 2001 From: Talat Uyarer Date: Mon, 31 Aug 2026 17:35:48 -0700 Subject: [PATCH 2/5] Flink: Address review comments on view read path --- .../apache/iceberg/flink/FlinkCatalog.java | 21 +++++++------------ .../iceberg/flink/TestFlinkCatalogView.java | 18 ++++++---------- 2 files changed, 14 insertions(+), 25 deletions(-) diff --git a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java index c57e875a1974..3cf3efd7386a 100644 --- a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java +++ b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java @@ -86,7 +86,6 @@ import org.apache.iceberg.view.SQLViewRepresentation; import org.apache.iceberg.view.View; import org.apache.iceberg.view.ViewProperties; -import org.apache.iceberg.view.ViewVersion; /** * A Flink Catalog implementation that wraps an Iceberg {@link Catalog}. @@ -102,8 +101,6 @@ @Internal public class FlinkCatalog extends AbstractCatalog { private static final String FLINK_DIALECT = "flink"; - private static final String DEFAULT_CATALOG_OPTION = "default-catalog"; - private static final String DEFAULT_NAMESPACE_OPTION = "default-namespace"; private final CatalogLoader catalogLoader; private final Catalog icebergCatalog; @@ -181,6 +178,12 @@ TableIdentifier toIdentifier(ObjectPath path) { } } + private boolean canBeView(ObjectPath tablePath) { + // the view catalog is only consulted for names that can denote a view: metadata-table + // syntax ("name$type") never does + return asViewCatalog != null && !tablePath.getObjectName().contains("$"); + } + @Override public List listDatabases() throws CatalogException { if (asNamespaceCatalog == null) { @@ -373,9 +376,7 @@ public CatalogBaseTable getTable(ObjectPath tablePath) try { table = loadIcebergTable(tablePath); } catch (TableNotExistException e) { - // metadata tables ("name$type") can never be views, and without a view catalog there is - // nothing else to look up - if (asViewCatalog == null || tablePath.getObjectName().contains("$")) { + if (!canBeView(tablePath)) { throw e; } @@ -433,7 +434,7 @@ private Table loadIcebergTable(ObjectPath tablePath) throws TableNotExistExcepti public boolean tableExists(ObjectPath tablePath) throws CatalogException { TableIdentifier identifier = toIdentifier(tablePath); return icebergCatalog.tableExists(identifier) - || (asViewCatalog != null && asViewCatalog.viewExists(identifier)); + || (canBeView(tablePath) && asViewCatalog.viewExists(identifier)); } @Override @@ -749,12 +750,6 @@ private CatalogView toCatalogView(View view) { Map options = Maps.newHashMap(view.properties()); String comment = options.remove(ViewProperties.COMMENT); - ViewVersion currentVersion = view.currentVersion(); - String defaultCatalog = - currentVersion.defaultCatalog() != null ? currentVersion.defaultCatalog() : getName(); - options.put(DEFAULT_CATALOG_OPTION, defaultCatalog); - options.put(DEFAULT_NAMESPACE_OPTION, currentVersion.defaultNamespace().toString()); - // both original and expanded query hold the stored SQL: Flink expands the query itself, // resolving unqualified references against the view's own catalog and database return CatalogView.of( diff --git a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogView.java b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogView.java index b87a305d55ce..f58f4887c965 100644 --- a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogView.java +++ b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogView.java @@ -205,18 +205,6 @@ public void testViewCommentAndProperties() throws Exception { .doesNotContainKey(ViewProperties.COMMENT); } - @TestTemplate - public void testDefaultCatalogFallsBackToCatalogName() throws Exception { - // no withDefaultCatalog() -> defaultCatalog() is null -> falls back to the Flink catalog name - createView("flink", "SELECT id, data FROM test_table"); - - CatalogBaseTable catalogView = - getTableEnv().getCatalog(catalogName).get().getTable(new ObjectPath(DATABASE, VIEW_NAME)); - assertThat(catalogView.getOptions()) - .containsEntry("default-catalog", catalogName) - .containsEntry("default-namespace", icebergNamespace.toString()); - } - @TestTemplate public void testListViews() throws Exception { assertThat(sql("SHOW VIEWS")).isEmpty(); @@ -272,6 +260,12 @@ public void testMetadataTableNotRoutedToViewPath() { .getTable(new ObjectPath(DATABASE, "nonexistent$snapshots"))) .isInstanceOf(TableNotExistException.class) .hasMessageContaining("Table (or view) db.nonexistent$snapshots does not exist"); + assertThat( + getTableEnv() + .getCatalog(catalogName) + .get() + .tableExists(new ObjectPath(DATABASE, "nonexistent$snapshots"))) + .isFalse(); } @TestTemplate From 8830487d975ba6ae2d1cfd593250d1065c65b972 Mon Sep 17 00:00:00 2001 From: Talat Uyarer Date: Fri, 28 Aug 2026 17:20:44 -0700 Subject: [PATCH 3/5] Enhance FlinkCatalog to support creating, dropping, and renaming views, along with related test cases for view operations. --- .../apache/iceberg/flink/FlinkCatalog.java | 82 +++++++++++++++++- .../iceberg/flink/TestFlinkCatalogView.java | 85 +++++++++++++++++++ 2 files changed, 163 insertions(+), 4 deletions(-) diff --git a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java index 3cf3efd7386a..00703583749d 100644 --- a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java +++ b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java @@ -39,6 +39,7 @@ import org.apache.flink.table.catalog.CatalogView; import org.apache.flink.table.catalog.ObjectPath; import org.apache.flink.table.catalog.ResolvedCatalogTable; +import org.apache.flink.table.catalog.ResolvedCatalogView; import org.apache.flink.table.catalog.ResolvedSchema; import org.apache.flink.table.catalog.TableChange; import org.apache.flink.table.catalog.exceptions.CatalogException; @@ -440,12 +441,21 @@ public boolean tableExists(ObjectPath tablePath) throws CatalogException { @Override public void dropTable(ObjectPath tablePath, boolean ignoreIfNotExists) throws TableNotExistException, CatalogException { + TableIdentifier identifier = toIdentifier(tablePath); + + boolean dropped; try { - icebergCatalog.dropTable(toIdentifier(tablePath)); + dropped = icebergCatalog.dropTable(identifier); } catch (org.apache.iceberg.exceptions.NoSuchTableException e) { - if (!ignoreIfNotExists) { - throw new TableNotExistException(getName(), tablePath, e); - } + dropped = false; + } + + if (!dropped && asViewCatalog != null) { + dropped = asViewCatalog.dropView(identifier); + } + + if (!dropped && !ignoreIfNotExists) { + throw new TableNotExistException(getName(), tablePath); } } @@ -457,6 +467,17 @@ public void renameTable(ObjectPath tablePath, String newTableName, boolean ignor toIdentifier(tablePath), toIdentifier(new ObjectPath(tablePath.getDatabaseName(), newTableName))); } catch (org.apache.iceberg.exceptions.NoSuchTableException e) { + if (asViewCatalog != null) { + try { + asViewCatalog.renameView( + toIdentifier(tablePath), + toIdentifier(new ObjectPath(tablePath.getDatabaseName(), newTableName))); + return; + } catch (NoSuchViewException viewException) { + e.addSuppressed(viewException); + } + } + if (!ignoreIfNotExists) { throw new TableNotExistException(getName(), tablePath, e); } @@ -468,6 +489,20 @@ public void renameTable(ObjectPath tablePath, String newTableName, boolean ignor @Override public void createTable(ObjectPath tablePath, CatalogBaseTable table, boolean ignoreIfExists) throws CatalogException, TableAlreadyExistException { + if (table instanceof CatalogView) { + if (asViewCatalog == null) { + throw new UnsupportedOperationException( + "Creating a view is not supported by catalog: " + getName()); + } + + Preconditions.checkArgument( + table instanceof ResolvedCatalogView, + "Expected a ResolvedCatalogView but got: %s", + table.getClass().getName()); + createIcebergView(tablePath, (ResolvedCatalogView) table, ignoreIfExists); + return; + } + // Creating Iceberg table using connector is allowed only when table is created using LIKE if (Objects.equals( table.getOptions().get(FlinkCreateTableOptions.CONNECTOR_PROPS_KEY), @@ -529,6 +564,35 @@ private boolean isReservedProperty(String prop) { || FlinkCreateTableOptions.SRC_CATALOG_PROPS_KEY.equalsIgnoreCase(prop); } + private void createIcebergView( + ObjectPath tablePath, ResolvedCatalogView view, boolean ignoreIfExists) + throws CatalogException, TableAlreadyExistException { + Map properties = Maps.newHashMap(view.getOptions()); + properties.remove(DEFAULT_CATALOG_OPTION); + properties.remove(DEFAULT_NAMESPACE_OPTION); + if (!StringUtils.isNullOrWhitespaceOnly(view.getComment())) { + properties.put(ViewProperties.COMMENT, view.getComment()); + } + + try { + // the default catalog is intentionally left unset so that readers resolve it to the + // name this catalog is registered under (see toCatalogView) + asViewCatalog + .buildView(toIdentifier(tablePath)) + .withSchema(FlinkSchemaUtil.convert(view.getResolvedSchema())) + .withDefaultNamespace(appendLevel(baseNamespace, tablePath.getDatabaseName())) + .withQuery(FLINK_DIALECT, view.getOriginalQuery()) + .withProperties(properties) + .create(); + } catch (AlreadyExistsException e) { + if (!ignoreIfExists) { + throw new TableAlreadyExistException(getName(), tablePath, e); + } + } catch (NoSuchNamespaceException e) { + throw new CatalogException("Database does not exist: " + tablePath.getDatabaseName(), e); + } + } + private static void validateTableSchemaAndPartition(CatalogTable ct1, CatalogTable ct2) { if (!Objects.equals(ct1.getUnresolvedSchema(), ct2.getUnresolvedSchema())) { throw new UnsupportedOperationException( @@ -563,6 +627,11 @@ private static void validateTablePartition(CatalogTable ct1, CatalogTable ct2) { @Override public void alterTable(ObjectPath tablePath, CatalogBaseTable newTable, boolean ignoreIfNotExists) throws CatalogException, TableNotExistException { + if (newTable instanceof CatalogView) { + throw new UnsupportedOperationException( + "Altering a view is not supported yet for catalog: " + getName()); + } + validateFlinkTable(newTable); Table icebergTable; @@ -625,6 +694,11 @@ public void alterTable( List tableChanges, boolean ignoreIfNotExists) throws TableNotExistException, CatalogException { + if (newTable instanceof CatalogView) { + throw new UnsupportedOperationException( + "Altering a view is not supported yet for catalog: " + getName()); + } + validateFlinkTable(newTable); Table icebergTable; diff --git a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogView.java b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogView.java index f58f4887c965..765be7fb3058 100644 --- a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogView.java +++ b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogView.java @@ -276,6 +276,91 @@ public void testDescribeView() { .containsExactly("id", "data"); } + @TestTemplate + public void testCreateViewViaSql() { + sql("CREATE VIEW %s AS SELECT id, data FROM %s", VIEW_NAME, TABLE_NAME); + + assertSameElements(expectedRows(), sql("SELECT * FROM %s", VIEW_NAME)); + + View view = viewCatalog().loadView(TableIdentifier.of(icebergNamespace, VIEW_NAME)); + // Flink hands the catalog its normalized (unparsed) SQL; references must stay unexpanded + assertThat(view.sqlFor("flink").sql()) + .containsIgnoringCase(String.format("FROM `%s`", TABLE_NAME)) + .doesNotContain(catalogName); + assertThat(view.currentVersion().defaultNamespace()).isEqualTo(icebergNamespace); + assertThat(view.currentVersion().defaultCatalog()).isNull(); + assertThat(view.schema().columns()) + .extracting(Types.NestedField::name) + .containsExactly("id", "data"); + assertThat(view.properties()) + .doesNotContainKey("default-catalog") + .doesNotContainKey("default-namespace"); + } + + @TestTemplate + public void testCreateViewWithCommentAndColumnList() { + sql( + "CREATE VIEW %s (view_id, view_data) COMMENT 'a view comment' AS SELECT id, data FROM %s", + VIEW_NAME, TABLE_NAME); + + View view = viewCatalog().loadView(TableIdentifier.of(icebergNamespace, VIEW_NAME)); + assertThat(view.properties()).containsEntry(ViewProperties.COMMENT, "a view comment"); + assertThat(view.schema().columns()) + .extracting(Types.NestedField::name) + .containsExactly("view_id", "view_data"); + } + + @TestTemplate + public void testCreateViewIfNotExists() { + sql("CREATE VIEW %s AS SELECT id, data FROM %s", VIEW_NAME, TABLE_NAME); + // IF NOT EXISTS is silent + sql("CREATE VIEW IF NOT EXISTS %s AS SELECT id FROM %s", VIEW_NAME, TABLE_NAME); + // without it, creation fails + assertThatThrownBy(() -> sql("CREATE VIEW %s AS SELECT id FROM %s", VIEW_NAME, TABLE_NAME)) + .hasMessageContaining(VIEW_NAME); + } + + @TestTemplate + public void testCreateViewOverExistingTableFails() { + assertThatThrownBy(() -> sql("CREATE VIEW %s AS SELECT id FROM %s", TABLE_NAME, TABLE_NAME)) + .hasMessageContaining(TABLE_NAME); + } + + @TestTemplate + public void testDropView() { + sql("CREATE VIEW %s AS SELECT id, data FROM %s", VIEW_NAME, TABLE_NAME); + assertThat(sql("SHOW VIEWS")).containsExactly(Row.of(VIEW_NAME)); + + sql("DROP VIEW %s", VIEW_NAME); + assertThat(sql("SHOW VIEWS")).isEmpty(); + assertThat(viewCatalog().viewExists(TableIdentifier.of(icebergNamespace, VIEW_NAME))).isFalse(); + } + + @TestTemplate + public void testDropViewIfExists() { + sql("DROP VIEW IF EXISTS nonexistent_view"); + assertThatThrownBy(() -> sql("DROP VIEW nonexistent_view")) + .hasMessageContaining("nonexistent_view"); + } + + @TestTemplate + public void testRenameView() { + sql("CREATE VIEW %s AS SELECT id, data FROM %s", VIEW_NAME, TABLE_NAME); + sql("ALTER VIEW %s RENAME TO renamed_view", VIEW_NAME); + + assertThat(sql("SHOW VIEWS")).containsExactly(Row.of("renamed_view")); + assertSameElements(expectedRows(), sql("SELECT * FROM renamed_view")); + } + + @TestTemplate + public void testAlterViewAsIsRejected() { + sql("CREATE VIEW %s AS SELECT id, data FROM %s", VIEW_NAME, TABLE_NAME); + assertThatThrownBy(() -> sql("ALTER VIEW %s AS SELECT id FROM %s", VIEW_NAME, TABLE_NAME)) + .hasMessageContaining("Could not execute AlterTable") + .rootCause() + .hasMessageContaining("Altering a view is not supported"); + } + @TestTemplate public void testGetViewViaCatalogApi() throws Exception { createView("flink", "SELECT id, data FROM test_table"); From 5a6bcecdd7ede4852213c966d2107942fdbf9ecc Mon Sep 17 00:00:00 2001 From: Talat Uyarer Date: Mon, 31 Aug 2026 17:45:49 -0700 Subject: [PATCH 4/5] Flink: Use view lookup applicability check in dropTable and renameTable --- .../main/java/org/apache/iceberg/flink/FlinkCatalog.java | 6 ++---- .../java/org/apache/iceberg/flink/TestFlinkCatalogView.java | 3 --- 2 files changed, 2 insertions(+), 7 deletions(-) diff --git a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java index 00703583749d..408f866a3a07 100644 --- a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java +++ b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java @@ -450,7 +450,7 @@ public void dropTable(ObjectPath tablePath, boolean ignoreIfNotExists) dropped = false; } - if (!dropped && asViewCatalog != null) { + if (!dropped && canBeView(tablePath)) { dropped = asViewCatalog.dropView(identifier); } @@ -467,7 +467,7 @@ public void renameTable(ObjectPath tablePath, String newTableName, boolean ignor toIdentifier(tablePath), toIdentifier(new ObjectPath(tablePath.getDatabaseName(), newTableName))); } catch (org.apache.iceberg.exceptions.NoSuchTableException e) { - if (asViewCatalog != null) { + if (canBeView(tablePath)) { try { asViewCatalog.renameView( toIdentifier(tablePath), @@ -568,8 +568,6 @@ private void createIcebergView( ObjectPath tablePath, ResolvedCatalogView view, boolean ignoreIfExists) throws CatalogException, TableAlreadyExistException { Map properties = Maps.newHashMap(view.getOptions()); - properties.remove(DEFAULT_CATALOG_OPTION); - properties.remove(DEFAULT_NAMESPACE_OPTION); if (!StringUtils.isNullOrWhitespaceOnly(view.getComment())) { properties.put(ViewProperties.COMMENT, view.getComment()); } diff --git a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogView.java b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogView.java index 765be7fb3058..aa6c5624fe8f 100644 --- a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogView.java +++ b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogView.java @@ -292,9 +292,6 @@ public void testCreateViewViaSql() { assertThat(view.schema().columns()) .extracting(Types.NestedField::name) .containsExactly("id", "data"); - assertThat(view.properties()) - .doesNotContainKey("default-catalog") - .doesNotContainKey("default-namespace"); } @TestTemplate From 76a9d4ac8b631e0c3e337dddf6b3faf4df613352 Mon Sep 17 00:00:00 2001 From: Talat Uyarer Date: Wed, 2 Sep 2026 08:43:29 -0700 Subject: [PATCH 5/5] Map AlreadyExistsException in renameTable view fallback --- .../org/apache/iceberg/flink/FlinkCatalog.java | 8 +++++++- .../iceberg/flink/TestFlinkCatalogView.java | 18 ++++++++++++++++++ 2 files changed, 25 insertions(+), 1 deletion(-) diff --git a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java index 408f866a3a07..74616968cc0d 100644 --- a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java +++ b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java @@ -475,6 +475,11 @@ public void renameTable(ObjectPath tablePath, String newTableName, boolean ignor return; } catch (NoSuchViewException viewException) { e.addSuppressed(viewException); + } catch (AlreadyExistsException alreadyExistsException) { + throw new TableAlreadyExistException( + getName(), + new ObjectPath(tablePath.getDatabaseName(), newTableName), + alreadyExistsException); } } @@ -482,7 +487,8 @@ public void renameTable(ObjectPath tablePath, String newTableName, boolean ignor throw new TableNotExistException(getName(), tablePath, e); } } catch (AlreadyExistsException e) { - throw new TableAlreadyExistException(getName(), tablePath, e); + throw new TableAlreadyExistException( + getName(), new ObjectPath(tablePath.getDatabaseName(), newTableName), e); } } diff --git a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogView.java b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogView.java index aa6c5624fe8f..5fbc648fa8d6 100644 --- a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogView.java +++ b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogView.java @@ -26,6 +26,7 @@ import org.apache.flink.table.catalog.CatalogBaseTable; import org.apache.flink.table.catalog.CatalogView; import org.apache.flink.table.catalog.ObjectPath; +import org.apache.flink.table.catalog.exceptions.TableAlreadyExistException; import org.apache.flink.table.catalog.exceptions.TableNotExistException; import org.apache.flink.types.Row; import org.apache.iceberg.Schema; @@ -349,6 +350,23 @@ public void testRenameView() { assertSameElements(expectedRows(), sql("SELECT * FROM renamed_view")); } + @TestTemplate + public void testRenameViewToExistingObjectFails() { + sql("CREATE VIEW %s AS SELECT id, data FROM %s", VIEW_NAME, TABLE_NAME); + viewCatalog() + .buildView(TableIdentifier.of(icebergNamespace, "second_view")) + .withSchema(VIEW_SCHEMA) + .withDefaultNamespace(icebergNamespace) + .withQuery("flink", "SELECT id FROM test_table") + .create(); + + assertThatThrownBy(() -> sql("ALTER VIEW %s RENAME TO second_view", VIEW_NAME)) + .hasMessageContaining("Could not execute ALTER VIEW") + .cause() + .isInstanceOf(TableAlreadyExistException.class) + .hasMessageContaining("second_view"); + } + @TestTemplate public void testAlterViewAsIsRejected() { sql("CREATE VIEW %s AS SELECT id, data FROM %s", VIEW_NAME, TABLE_NAME);