diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/CatalogManager.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/CatalogManager.java index fa9db8cb50dbd..8729c618ea771 100644 --- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/CatalogManager.java +++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/CatalogManager.java @@ -749,7 +749,7 @@ public Optional getTable(ObjectIdentifier objectIdentifier CatalogBaseTable temporaryTable = temporaryTables.get(objectIdentifier); if (temporaryTable != null) { final ResolvedCatalogBaseTable resolvedTable = - resolveCatalogBaseTable(temporaryTable); + resolveCatalogBaseTable(temporaryTable, objectIdentifier); return Optional.of(ContextResolvedTable.temporary(objectIdentifier, resolvedTable)); } else { return getPermanentTable(objectIdentifier, null); @@ -770,7 +770,7 @@ public Optional getTable( CatalogBaseTable temporaryTable = temporaryTables.get(objectIdentifier); if (temporaryTable != null) { final ResolvedCatalogBaseTable resolvedTable = - resolveCatalogBaseTable(temporaryTable); + resolveCatalogBaseTable(temporaryTable, objectIdentifier); return Optional.of(ContextResolvedTable.temporary(objectIdentifier, resolvedTable)); } else { return getPermanentTable(objectIdentifier, timestamp); @@ -860,7 +860,8 @@ private Optional getPermanentTable( } else { table = currentCatalog.getTable(objectPath); } - final ResolvedCatalogBaseTable resolvedTable = resolveCatalogBaseTable(table); + final ResolvedCatalogBaseTable resolvedTable = + resolveCatalogBaseTable(table, objectIdentifier); return Optional.of( ContextResolvedTable.permanent( objectIdentifier, currentCatalog, resolvedTable)); @@ -2261,13 +2262,27 @@ private String getErrorMessage(ObjectIdentifier objectIdentifier, String command /** Resolves a {@link CatalogBaseTable} to a validated {@link ResolvedCatalogBaseTable}. */ public ResolvedCatalogBaseTable resolveCatalogBaseTable(CatalogBaseTable baseTable) { + return resolveCatalogBaseTable(baseTable, null); + } + + /** + * Resolves a {@link CatalogBaseTable} to a validated {@link ResolvedCatalogBaseTable}. + * + * @param baseTable the table to resolve + * @param objectIdentifier the identifier the table is stored under, used to resolve unqualified + * references in a {@link CatalogView}'s expanded query against the view's own + * catalog/database; may be {@code null} when unknown, in which case the current session + * catalog/database is used + */ + public ResolvedCatalogBaseTable resolveCatalogBaseTable( + CatalogBaseTable baseTable, @Nullable ObjectIdentifier objectIdentifier) { Preconditions.checkNotNull(schemaResolver, "Schema resolver is not initialized."); if (baseTable instanceof CatalogTable) { return resolveCatalogTable((CatalogTable) baseTable); } else if (baseTable instanceof CatalogMaterializedTable) { return resolveCatalogMaterializedTable((CatalogMaterializedTable) baseTable); } else if (baseTable instanceof CatalogView) { - return resolveCatalogView((CatalogView) baseTable); + return resolveCatalogView((CatalogView) baseTable, objectIdentifier); } throw new IllegalArgumentException( "Unknown kind of catalog base table: " + baseTable.getClass()); @@ -2382,6 +2397,21 @@ public ResolvedCatalogMaterializedTable resolveCatalogMaterializedTable( /** Resolves a {@link CatalogView} to a validated {@link ResolvedCatalogView}. */ public ResolvedCatalogView resolveCatalogView(CatalogView view) { + return resolveCatalogView(view, null); + } + + /** + * Resolves a {@link CatalogView} to a validated {@link ResolvedCatalogView}. + * + * @param view the view to resolve + * @param viewIdentifier the identifier the view is stored under. When non-{@code null}, the + * view's expanded query is parsed against the view's own catalog/database so that + * unqualified references resolve hermetically (see FLIP-71) instead of falling back to the + * current session database. When {@code null}, the current session catalog/database is + * used. + */ + public ResolvedCatalogView resolveCatalogView( + CatalogView view, @Nullable ObjectIdentifier viewIdentifier) { Preconditions.checkNotNull(schemaResolver, "Schema resolver is not initialized."); if (view instanceof ResolvedCatalogView) { return (ResolvedCatalogView) view; @@ -2396,7 +2426,7 @@ public ResolvedCatalogView resolveCatalogView(CatalogView view) { final ResolvedSchema resolvedSchema = view.getUnresolvedSchema().resolve(schemaResolver); final List parse; try { - parse = parser.parse(view.getExpandedQuery()); + parse = parseViewQuery(view.getExpandedQuery(), viewIdentifier); } catch (Throwable e) { // in case of a failure during parsing, let the lower layers fail return new ResolvedCatalogView(view, resolvedSchema); @@ -2437,6 +2467,31 @@ public ResolvedCatalogView resolveCatalogView(CatalogView view) { } } + /** + * Parses a view's expanded query. When {@code viewIdentifier} is non-{@code null}, parsing + * happens with the current catalog/database temporarily switched to the view's own + * catalog/database, so that unqualified references in the query resolve hermetically against + * the view rather than falling back to the current session database (FLIP-71). External + * catalogs (e.g. Iceberg) may return portable, unqualified SQL from {@link + * CatalogView#getExpandedQuery()} for which this matters. + */ + private List parseViewQuery( + String expandedQuery, @Nullable ObjectIdentifier viewIdentifier) { + if (viewIdentifier == null) { + return parser.parse(expandedQuery); + } + final String savedCurrentCatalog = currentCatalogName; + final String savedCurrentDatabase = currentDatabaseName; + try { + currentCatalogName = viewIdentifier.getCatalogName(); + currentDatabaseName = viewIdentifier.getDatabaseName(); + return parser.parse(expandedQuery); + } finally { + currentCatalogName = savedCurrentCatalog; + currentDatabaseName = savedCurrentDatabase; + } + } + /** * Create a database. * diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/catalog/HermeticViewExpansionTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/catalog/HermeticViewExpansionTest.java new file mode 100644 index 0000000000000..3cc75163f76e4 --- /dev/null +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/catalog/HermeticViewExpansionTest.java @@ -0,0 +1,89 @@ +/* + * 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.flink.table.planner.catalog; + +import org.apache.flink.table.api.DataTypes; +import org.apache.flink.table.api.EnvironmentSettings; +import org.apache.flink.table.api.Schema; +import org.apache.flink.table.api.TableEnvironment; +import org.apache.flink.table.catalog.Catalog; +import org.apache.flink.table.catalog.CatalogDatabaseImpl; +import org.apache.flink.table.catalog.CatalogView; +import org.apache.flink.table.catalog.ObjectPath; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import java.util.Collections; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** + * Tests that a catalog view's expanded query resolves unqualified table references against the + * view's own catalog/database rather than falling back to the current session database + * (FLINK-40637, FLIP-71). This mirrors external catalogs (e.g. Iceberg) that return portable, + * unqualified SQL from {@link CatalogView#getExpandedQuery()}. + */ +class HermeticViewExpansionTest { + + private TableEnvironment tEnv; + private Catalog catalog; + + @BeforeEach + void setUp() throws Exception { + tEnv = TableEnvironment.create(EnvironmentSettings.inStreamingMode()); + catalog = tEnv.getCatalog(tEnv.getCurrentCatalog()).get(); + catalog.createDatabase("db2", new CatalogDatabaseImpl(Collections.emptyMap(), null), false); + } + + private CatalogView unqualifiedView() { + // Simulates an external catalog returning portable SQL with an unqualified reference. + return CatalogView.of( + Schema.newBuilder().column("a", DataTypes.INT()).build(), + null, + "SELECT * FROM t1", + "SELECT * FROM t1", + Collections.emptyMap()); + } + + @Test + void testViewExpansionDoesNotFallBackToSessionDatabase() throws Exception { + // t1 exists only in the session database (default_database), NOT in db2 where the view + // lives. Expanding the view must resolve `t1` against db2 and therefore fail, instead of + // silently picking up the session's default_database.t1. + tEnv.executeSql("CREATE TABLE t1 (a INT) WITH ('connector' = 'datagen')"); + catalog.createTable(new ObjectPath("db2", "v"), unqualifiedView(), false); + + assertThatThrownBy(() -> tEnv.explainSql("SELECT * FROM db2.v")) + .hasMessageContaining("Object 't1' not found"); + } + + @Test + void testViewExpansionResolvesAgainstViewDatabase() throws Exception { + // t1 exists in the view's own database (db2). Expansion must resolve `t1` to db2.t1, + // regardless of the session sitting on default_database. + tEnv.executeSql("CREATE TABLE db2.t1 (a INT) WITH ('connector' = 'datagen')"); + catalog.createTable(new ObjectPath("db2", "v"), unqualifiedView(), false); + + assertThat(tEnv.explainSql("SELECT * FROM db2.v")) + .contains("default_catalog, db2, t1") + .doesNotContain("default_database, t1"); + } +}