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 <[email protected]>
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");
+ }
+}
