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..8cfd259b055a 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,8 +36,10 @@ 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.ResolvedCatalogView; import org.apache.flink.table.catalog.ResolvedSchema; import org.apache.flink.table.catalog.TableChange; import org.apache.flink.table.catalog.exceptions.CatalogException; @@ -67,9 +69,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 +84,11 @@ 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.UpdateViewProperties; +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,12 +103,18 @@ */ @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; + private final boolean viewDialectStrict; public FlinkCatalog( String catalogName, @@ -108,10 +123,29 @@ public FlinkCatalog( CatalogLoader catalogLoader, boolean cacheEnabled, long cacheExpirationIntervalMs) { + this( + catalogName, + defaultDatabase, + baseNamespace, + catalogLoader, + cacheEnabled, + cacheExpirationIntervalMs, + FlinkCatalogFactory.VIEW_DIALECT_STRICT_DEFAULT); + } + + public FlinkCatalog( + String catalogName, + String defaultDatabase, + Namespace baseNamespace, + CatalogLoader catalogLoader, + boolean cacheEnabled, + long cacheExpirationIntervalMs, + boolean viewDialectStrict) { super(catalogName, defaultDatabase); this.catalogLoader = catalogLoader; this.baseNamespace = baseNamespace; this.cacheEnabled = cacheEnabled; + this.viewDialectStrict = viewDialectStrict; Catalog originalCatalog = catalogLoader.loadCatalog(); icebergCatalog = @@ -120,6 +154,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 +357,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 +389,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,18 +453,29 @@ 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 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); } } @@ -401,6 +487,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); } @@ -412,6 +509,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), @@ -473,6 +584,85 @@ 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 void alterIcebergView( + ObjectPath tablePath, ResolvedCatalogView newView, boolean ignoreIfNotExists) + throws TableNotExistException, CatalogException { + View view; + try { + view = asViewCatalog.loadView(toIdentifier(tablePath)); + } catch (NoSuchViewException e) { + if (!ignoreIfNotExists) { + throw new TableNotExistException(getName(), tablePath, e); + } + + return; + } + + SQLViewRepresentation currentRepresentation = view.sqlFor(FLINK_DIALECT); + if (currentRepresentation == null + || !newView.getOriginalQuery().equals(currentRepresentation.sql())) { + // a new query becomes a new view version; the stored resolution context is preserved + view.replaceVersion() + .withQuery(FLINK_DIALECT, newView.getOriginalQuery()) + .withSchema(FlinkSchemaUtil.convert(newView.getResolvedSchema())) + .withDefaultNamespace(view.currentVersion().defaultNamespace()) + .commit(); + } + + // only set changed or added properties: Flink SQL cannot express a property removal + // (ALTER VIEW SET merges), and removing keys absent from the incoming options could + // strip properties that Flink did not carry over + Map newProperties = Maps.newHashMap(newView.getOptions()); + newProperties.remove(DEFAULT_CATALOG_OPTION); + newProperties.remove(DEFAULT_NAMESPACE_OPTION); + if (!StringUtils.isNullOrWhitespaceOnly(newView.getComment())) { + newProperties.put(ViewProperties.COMMENT, newView.getComment()); + } + + Map currentProperties = view.properties(); + UpdateViewProperties update = view.updateProperties(); + boolean changed = false; + for (Map.Entry entry : newProperties.entrySet()) { + if (!entry.getValue().equals(currentProperties.get(entry.getKey()))) { + update.set(entry.getKey(), entry.getValue()); + changed = true; + } + } + + if (changed) { + update.commit(); + } + } + private static void validateTableSchemaAndPartition(CatalogTable ct1, CatalogTable ct2) { if (!Objects.equals(ct1.getUnresolvedSchema(), ct2.getUnresolvedSchema())) { throw new UnsupportedOperationException( @@ -507,6 +697,20 @@ 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) { + if (asViewCatalog == null) { + throw new UnsupportedOperationException( + "Altering a view is not supported by catalog: " + getName()); + } + + Preconditions.checkArgument( + newTable instanceof ResolvedCatalogView, + "Expected a ResolvedCatalogView but got: %s", + newTable.getClass().getName()); + alterIcebergView(tablePath, (ResolvedCatalogView) newTable, ignoreIfNotExists); + return; + } + validateFlinkTable(newTable); Table icebergTable; @@ -569,6 +773,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; @@ -683,6 +892,35 @@ 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()); + if (viewDialectStrict && !FLINK_DIALECT.equalsIgnoreCase(sqlRepresentation.dialect())) { + throw new UnsupportedOperationException( + String.format( + "View %s does not have a flink dialect representation and %s is enabled", + view.name(), FlinkCatalogFactory.VIEW_DIALECT_STRICT)); + } + + 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 +933,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/main/java/org/apache/iceberg/flink/FlinkCatalogFactory.java b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalogFactory.java index 24e2bdbba37a..3cbc43cf2599 100644 --- a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalogFactory.java +++ b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalogFactory.java @@ -74,6 +74,8 @@ public class FlinkCatalogFactory implements CatalogFactory { public static final String DEFAULT_DATABASE_NAME = "default"; public static final String DEFAULT_CATALOG_NAME = "default_catalog"; public static final String BASE_NAMESPACE = "base-namespace"; + public static final String VIEW_DIALECT_STRICT = "view-dialect-strict"; + public static final boolean VIEW_DIALECT_STRICT_DEFAULT = false; /** * Create an Iceberg {@link org.apache.iceberg.catalog.Catalog} loader to be used by this Flink @@ -165,13 +167,18 @@ protected Catalog createCatalog( "%s is not allowed to be 0.", CatalogProperties.CACHE_EXPIRATION_INTERVAL_MS); + boolean viewDialectStrict = + PropertyUtil.propertyAsBoolean( + properties, VIEW_DIALECT_STRICT, VIEW_DIALECT_STRICT_DEFAULT); + return new FlinkCatalog( name, defaultDatabase, baseNamespace, catalogLoader, cacheEnabled, - cacheExpirationIntervalMs); + cacheExpirationIntervalMs, + viewDialectStrict); } private static Configuration mergeHiveConf( 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..5e07295c0e9c --- /dev/null +++ b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogView.java @@ -0,0 +1,473 @@ +/* + * 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 java.util.Map; +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.ResolvedCatalogView; +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.relocated.com.google.common.collect.Maps; +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 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 testAlterViewAs() { + sql("CREATE VIEW %s AS SELECT id, data FROM %s", VIEW_NAME, TABLE_NAME); + sql("ALTER VIEW %s AS SELECT id FROM %s", VIEW_NAME, TABLE_NAME); + + assertSameElements( + Lists.newArrayList(Row.of(1L), Row.of(2L), Row.of(3L)), sql("SELECT * FROM %s", VIEW_NAME)); + + View view = viewCatalog().loadView(TableIdentifier.of(icebergNamespace, VIEW_NAME)); + assertThat(view.versions()).hasSize(2); + assertThat(view.schema().columns()).extracting(Types.NestedField::name).containsExactly("id"); + assertThat(view.currentVersion().defaultNamespace()).isEqualTo(icebergNamespace); + assertThat(view.currentVersion().defaultCatalog()).isNull(); + } + + @TestTemplate + public void testAlterViewAsPreservesProperties() { + sql("CREATE VIEW %s COMMENT 'keep me' AS SELECT id, data FROM %s", VIEW_NAME, TABLE_NAME); + sql("ALTER VIEW %s AS SELECT id FROM %s", VIEW_NAME, TABLE_NAME); + + View view = viewCatalog().loadView(TableIdentifier.of(icebergNamespace, VIEW_NAME)); + assertThat(view.properties()).containsEntry(ViewProperties.COMMENT, "keep me"); + } + + @TestTemplate + public void testAlterViewPropertiesViaCatalogApi() throws Exception { + // Flink's default SQL dialect has no ALTER VIEW ... SET syntax (only RENAME and AS); + // property updates arrive through the catalog API, e.g. from the Hive dialect + sql("CREATE VIEW %s AS SELECT id, data FROM %s", VIEW_NAME, TABLE_NAME); + + org.apache.flink.table.catalog.Catalog flinkCatalog = + getTableEnv().getCatalog(catalogName).get(); + ObjectPath path = new ObjectPath(DATABASE, VIEW_NAME); + CatalogView current = (CatalogView) flinkCatalog.getTable(path); + + Map newOptions = Maps.newHashMap(current.getOptions()); + newOptions.put("key1", "value1"); + CatalogView newView = + CatalogView.of( + current.getUnresolvedSchema(), + current.getComment(), + current.getOriginalQuery(), + current.getExpandedQuery(), + newOptions); + flinkCatalog.alterTable( + path, + new ResolvedCatalogView(newView, FlinkSchemaUtil.toResolvedSchema(VIEW_SCHEMA)), + false); + + View view = viewCatalog().loadView(TableIdentifier.of(icebergNamespace, VIEW_NAME)); + assertThat(view.properties()).containsEntry("key1", "value1"); + // a property change must not create a new view version + assertThat(view.versions()).hasSize(1); + } + + @TestTemplate + public void testAlterViewAsDroppingOtherDialectFails() { + // core refuses a replace that loses another engine's dialect unless + // replace.drop-dialect.allowed=true (default false) + viewCatalog() + .buildView(TableIdentifier.of(icebergNamespace, VIEW_NAME)) + .withSchema(VIEW_SCHEMA) + .withDefaultNamespace(icebergNamespace) + .withQuery("spark", "SELECT id, data FROM test_table") + .withQuery("flink", "SELECT id, data FROM test_table") + .create(); + + assertThatThrownBy(() -> sql("ALTER VIEW %s AS SELECT id FROM %s", VIEW_NAME, TABLE_NAME)) + .hasMessageContaining("Could not execute AlterTable") + .rootCause() + .hasMessageContaining("dialect"); + } + + @TestTemplate + public void testStrictDialectRejectsForeignDialect() { + createView("spark", "SELECT id, data FROM test_table"); + + String strictCatalog = catalogName + "_strict"; + Map strictConfig = Maps.newHashMap(config); + strictConfig.put(FlinkCatalogFactory.VIEW_DIALECT_STRICT, "true"); + sql("CREATE CATALOG %s WITH %s", strictCatalog, toWithClause(strictConfig)); + try { + assertThatThrownBy(() -> sql("SELECT * FROM %s.%s.%s", strictCatalog, DATABASE, VIEW_NAME)) + .rootCause() + .hasMessageContaining("does not have a flink dialect"); + + // the default (lenient) catalog still reads it + assertSameElements(expectedRows(), sql("SELECT * FROM %s", VIEW_NAME)); + } finally { + dropCatalog(strictCatalog, true); + } + } + + @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"); + } +}