This is an automated email from the ASF dual-hosted git repository.
gustavodemorais 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 5b04b40b373 [FLINK-41211][table] Add listMaterializedTables for a
given catalog and database
5b04b40b373 is described below
commit 5b04b40b373d05f8a1e727135dcdd03007923990
Author: Timo Theusner <[email protected]>
AuthorDate: Fri Oct 30 10:13:34 2026 +0100
[FLINK-41211][table] Add listMaterializedTables for a given catalog and
database
Fixed in #29661
---
flink-python/pyflink/table/table_environment.py | 26 ++++++---
.../table/tests/test_table_environment_api.py | 13 +++++
.../apache/flink/table/api/TableEnvironment.java | 12 ++++-
.../table/api/internal/TableEnvironmentImpl.java | 7 +++
.../flink/table/api/TableEnvironmentTest.java | 62 ++++++++++++++++++++++
5 files changed, 113 insertions(+), 7 deletions(-)
diff --git a/flink-python/pyflink/table/table_environment.py
b/flink-python/pyflink/table/table_environment.py
index ebab2034a38..a8abb9bd546 100644
--- a/flink-python/pyflink/table/table_environment.py
+++ b/flink-python/pyflink/table/table_environment.py
@@ -678,17 +678,31 @@ class TableEnvironment(object):
j_view_name_array = self._j_tenv.listViews()
return [item for item in j_view_name_array]
- def list_materialized_tables(self) -> List[str]:
+ def list_materialized_tables(self, catalog_name: Optional[str] = None,
+ database_name: Optional[str] = None) ->
List[str]:
"""
- Gets the names of all materialized tables available
- in the current namespace (the current database of the current catalog).
+ Gets the names of all materialized tables available in the given
namespace (the given
+ database of the given catalog), or in the current namespace (the
current database of
+ the current catalog) if neither is given.
- :return: A list of the names of all registered materialized tables
- in the current database of the current catalog.
+ :param catalog_name: The name of the catalog. Must be given together
with
+ ``database_name``.
+ :param database_name: The name of the database. Must be given together
with
+ ``catalog_name``.
+ :return: A list of the names of all registered materialized tables in
the namespace.
.. versionadded:: 2.2.0
+
+ .. versionchanged:: 2.5.0
+ Added the ``catalog_name`` and ``database_name`` parameters.
"""
- j_materialized_table_name_array = self._j_tenv.listMaterializedTables()
+ if catalog_name is None and database_name is None:
+ j_materialized_table_name_array =
self._j_tenv.listMaterializedTables()
+ elif catalog_name is None or database_name is None:
+ raise ValueError("catalog_name and database_name must be given
together.")
+ else:
+ j_materialized_table_name_array =
self._j_tenv.listMaterializedTables(
+ catalog_name, database_name)
return [item for item in j_materialized_table_name_array]
def list_user_defined_functions(self) -> List[str]:
diff --git a/flink-python/pyflink/table/tests/test_table_environment_api.py
b/flink-python/pyflink/table/tests/test_table_environment_api.py
index 765b9329e84..0562337c775 100644
--- a/flink-python/pyflink/table/tests/test_table_environment_api.py
+++ b/flink-python/pyflink/table/tests/test_table_environment_api.py
@@ -460,6 +460,19 @@ class TableEnvironmentTest(PyFlinkUTTestCase):
Py4JJavaError, "No module with name 'dummy' exists",
self.t_env.use_modules, 'core', 'dummy')
+ def test_list_materialized_tables_in_given_namespace(self):
+ self.t_env.execute_sql(
+ "CREATE CATALOG other_catalog WITH ("
+ "'type' = 'generic_in_memory', 'default-database' =
'other_database')")
+
+ self.assertEqual(
+ [], self.t_env.list_materialized_tables('other_catalog',
'other_database'))
+ self.assertEqual([], self.t_env.list_materialized_tables())
+ self.assertRaisesRegex(
+ Py4JJavaError, "Catalog unknown_catalog does not exist",
+ self.t_env.list_materialized_tables, 'unknown_catalog',
'other_database')
+ self.assertRaises(ValueError, self.t_env.list_materialized_tables,
'other_catalog')
+
def assert_created_table_from_descriptor(self, schema: Schema):
catalog = self.t_env.get_current_catalog()
database = self.t_env.get_current_database()
diff --git
a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/TableEnvironment.java
b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/TableEnvironment.java
index 5f6c584cced..44be50b8acb 100644
---
a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/TableEnvironment.java
+++
b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/TableEnvironment.java
@@ -1285,10 +1285,20 @@ public interface TableEnvironment {
*
* @return A list of the names of all registered materialized tables in
the current database of
* the current catalog.
- * @see #listTemporaryViews()
+ * @see #listMaterializedTables(String, String)
*/
String[] listMaterializedTables();
+ /**
+ * Gets the names of all materialized tables available in the given
namespace (the given
+ * database of the given catalog).
+ *
+ * @return A list of the names of all registered materialized tables in
the given database of
+ * the given catalog.
+ * @see #listMaterializedTables()
+ */
+ String[] listMaterializedTables(String catalogName, String databaseName);
+
/**
* Gets the names of all temporary tables and views available in the
current namespace (the
* current database of the current catalog).
diff --git
a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java
b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java
index 496369e78f2..eb17b63791c 100644
---
a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java
+++
b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java
@@ -774,6 +774,13 @@ public class TableEnvironmentImpl implements
TableEnvironmentInternal {
return
catalogManager.listMaterializedTables().stream().sorted().toArray(String[]::new);
}
+ @Override
+ public String[] listMaterializedTables(String catalogName, String
databaseName) {
+ return catalogManager.listMaterializedTables(catalogName,
databaseName).stream()
+ .sorted()
+ .toArray(String[]::new);
+ }
+
@Override
public String[] listTemporaryTables() {
return
catalogManager.listTemporaryTables().stream().sorted().toArray(String[]::new);
diff --git
a/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/api/TableEnvironmentTest.java
b/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/api/TableEnvironmentTest.java
index 290a2823138..e2d7a6bf3f0 100644
---
a/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/api/TableEnvironmentTest.java
+++
b/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/api/TableEnvironmentTest.java
@@ -18,12 +18,17 @@
package org.apache.flink.table.api;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.table.catalog.Catalog;
import org.apache.flink.table.catalog.CatalogBaseTable;
+import org.apache.flink.table.catalog.CatalogDescriptor;
+import org.apache.flink.table.catalog.CatalogMaterializedTable;
import org.apache.flink.table.catalog.CatalogModel;
import org.apache.flink.table.catalog.CatalogTable;
import org.apache.flink.table.catalog.CatalogView;
import org.apache.flink.table.catalog.ContextResolvedModel;
import org.apache.flink.table.catalog.ContextResolvedTable;
+import org.apache.flink.table.catalog.IntervalFreshness;
import org.apache.flink.table.catalog.ObjectIdentifier;
import org.apache.flink.table.catalog.ObjectPath;
import org.apache.flink.table.catalog.exceptions.ModelNotExistException;
@@ -258,6 +263,50 @@ class TableEnvironmentTest {
assertThat(tEnv.listModels()).containsExactly("M1", "M2");
}
+ @Test
+ void testListMaterializedTablesInGivenNamespace() throws Exception {
+ final String currentCatalog = tEnv.getCurrentCatalog();
+ tEnv.createCatalog(
+ "other_catalog",
+ CatalogDescriptor.of(
+ "other_catalog",
+ Configuration.fromMap(
+ Map.of(
+ "type", "generic_in_memory",
+ "default-database",
"other_database"))));
+ final Catalog otherCatalog =
+
tEnv.getCatalog("other_catalog").orElseThrow(AssertionError::new);
+ otherCatalog.createTable(
+ new ObjectPath("other_database", "mt_b"),
createMaterializedTable(), false);
+ otherCatalog.createTable(
+ new ObjectPath("other_database", "mt_a"),
createMaterializedTable(), false);
+
+ tEnv.useCatalog("other_catalog");
+ tEnv.createTable("T", TEST_DESCRIPTOR);
+ tEnv.createView("V", tEnv.from("T"));
+ tEnv.createTemporaryTable("TMP", TEST_DESCRIPTOR);
+ tEnv.useCatalog(currentCatalog);
+
+ assertThat(tEnv.listMaterializedTables("other_catalog",
"other_database"))
+ .containsExactly("mt_a", "mt_b");
+ }
+
+ @Test
+ void testListMaterializedTablesThrowsForUnknownNamespace() {
+ assertThatThrownBy(
+ () ->
+ tEnv.listMaterializedTables(
+ "unknown_catalog",
tEnv.getCurrentDatabase()))
+ .isInstanceOf(ValidationException.class)
+ .hasMessage("Catalog unknown_catalog does not exist");
+ assertThatThrownBy(
+ () ->
+ tEnv.listMaterializedTables(
+ tEnv.getCurrentCatalog(),
"unknown_database"))
+ .isInstanceOf(ValidationException.class)
+ .hasMessage("Database unknown_database does not exist");
+ }
+
private static void assertCreateTableFromDescriptor(
TableEnvironmentMock tEnv, Schema schema, boolean ignoreIfExists)
throws
org.apache.flink.table.catalog.exceptions.TableNotExistException {
@@ -352,4 +401,17 @@ class TableEnvironmentTest {
assertThat(catalogModel.getOptions()).containsEntry(entry.getKey(),
entry.getValue());
}
}
+
+ private static CatalogMaterializedTable createMaterializedTable() {
+ return CatalogMaterializedTable.newBuilder()
+ .schema(TEST_SCHEMA)
+ .options(Map.of())
+ .originalQuery("SELECT 1 AS f0")
+ .expandedQuery("SELECT 1 AS f0")
+ .freshness(IntervalFreshness.ofSecond(30))
+
.logicalRefreshMode(CatalogMaterializedTable.LogicalRefreshMode.AUTOMATIC)
+ .refreshMode(CatalogMaterializedTable.RefreshMode.CONTINUOUS)
+
.refreshStatus(CatalogMaterializedTable.RefreshStatus.INITIALIZING)
+ .build();
+ }
}
