[FLINK-40637][table] Expand catalog views hermetically against the view's own database

flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/CatalogManager.java

@@ -749,7 +749,7 @@ public Optional<ContextResolvedTable> 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<ContextResolvedTable> 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<ContextResolvedTable> 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<Operation> 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<Operation> 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.
      *

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");
+    }
+}