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..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 @@ -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,9 @@ 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; /** * A Flink Catalog implementation that wraps an Iceberg {@link Catalog}. @@ -94,10 +100,13 @@ */ @Internal public class FlinkCatalog extends AbstractCatalog { + private static final String FLINK_DIALECT = "flink"; + 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 +129,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(); @@ -167,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) { @@ -321,8 +338,30 @@ public void alterDatabase(String name, CatalogDatabase newDatabase, boolean igno @Override public List listTables(String databaseName) throws DatabaseNotExistException, CatalogException { + List results; try { - return icebergCatalog.listTables(appendLevel(baseNamespace, databaseName)).stream() + 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 asViewCatalog.listViews(appendLevel(baseNamespace, databaseName)).stream() .map(TableIdentifier::name) .collect(Collectors.toList()); } catch (NoSuchNamespaceException e) { @@ -331,9 +370,24 @@ 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) { + if (!canBeView(tablePath)) { + 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 +432,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) + || (canBeView(tablePath) && asViewCatalog.viewExists(identifier)); } @Override @@ -683,6 +739,23 @@ 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); + + // 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 +768,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..f58f4887c965 --- /dev/null +++ b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogView.java @@ -0,0 +1,295 @@ +/* + * 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 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"); + assertThat( + getTableEnv() + .getCatalog(catalogName) + .get() + .tableExists(new ObjectPath(DATABASE, "nonexistent$snapshots"))) + .isFalse(); + } + + @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"); + } +}