You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by mx...@apache.org on 2026/09/28 12:10:13 UTC
(flink) branch master updated: [FLINK-40650][table] Enhance CatalogManager to support object identifiers in table resolution
This is an automated email from the ASF dual-hosted git repository.
mxm pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 710bb463094 [FLINK-40650][table] Enhance CatalogManager to support object identifiers in table resolution
710bb463094 is described below
commit 710bb4630946caadbfcf1839aa38853a31b29a35
Author: Talat Uyarer <ta...@apache.org>
AuthorDate: Sat Sep 12 07:58:42 2026 -0700
[FLINK-40650][table] Enhance CatalogManager to support object identifiers in table resolution
---
.../apache/flink/table/catalog/CatalogManager.java | 65 ++++++++++++++--
.../planner/catalog/HermeticViewExpansionTest.java | 89 ++++++++++++++++++++++
2 files changed, 149 insertions(+), 5 deletions(-)
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 9a306d4e0bc..ad0f9fb32ae 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 final class CatalogManager implements CatalogRegistry, AutoCloseable {
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 final class CatalogManager implements CatalogRegistry, AutoCloseable {
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 @@ public final class CatalogManager implements CatalogRegistry, AutoCloseable {
} else {
table = currentCatalog.getTable(objectPath);
}
- final ResolvedCatalogBaseTable<?> resolvedTable = resolveCatalogBaseTable(table);
+ final ResolvedCatalogBaseTable<?> resolvedTable =
+ resolveCatalogBaseTable(table, objectIdentifier);
return Optional.of(
ContextResolvedTable.permanent(
objectIdentifier, currentCatalog, resolvedTable));
@@ -2273,13 +2274,27 @@ public final class CatalogManager implements CatalogRegistry, AutoCloseable {
/** 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());
@@ -2394,6 +2409,21 @@ public final class CatalogManager implements CatalogRegistry, AutoCloseable {
/** 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;
@@ -2408,7 +2438,7 @@ public final class CatalogManager implements CatalogRegistry, AutoCloseable {
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);
@@ -2449,6 +2479,31 @@ public final class CatalogManager implements CatalogRegistry, AutoCloseable {
}
}
+ /**
+ * 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.
*
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 00000000000..3cc75163f76
--- /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");
+ }
+}