You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@fluss.apache.org by ab...@apache.org on 2026/10/15 07:39:44 UTC
(fluss) branch main updated: [flink] Support SHOW MATERIALIZED TABLES on Flink 2.2+ (#4634)
This is an automated email from the ASF dual-hosted git repository.
fresh-borzoni pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/fluss.git
The following commit(s) were added to refs/heads/main by this push:
new 431bc17129 [flink] Support SHOW MATERIALIZED TABLES on Flink 2.2+ (#4634)
431bc17129 is described below
commit 431bc17129832ec166c6d7c7a0f17e11a82ae1cb
Author: Ramin Gharib <ra...@gmail.com>
AuthorDate: Thu Oct 15 09:39:38 2026 +0200
[flink] Support SHOW MATERIALIZED TABLES on Flink 2.2+ (#4634)
Flink 2.2 routes SHOW MATERIALIZED TABLES to Catalog#listMaterializedTables, whose default implementation throws. FlinkCatalog now implements it by filtering listTables by the materialized-table custom properties, fetching the table infos in parallel.
The method is declared without @Override so that fluss-flink-common still compiles against Flink 1.20, which doesn't have it.
---
.../fluss/flink/catalog/Flink22CatalogITCase.java | 20 +++++++++++
.../fluss/flink/catalog/Flink23CatalogITCase.java | 20 +++++++++++
.../apache/fluss/flink/catalog/FlinkCatalog.java | 39 ++++++++++++++++++++++
.../apache/fluss/flink/utils/FlinkConversions.java | 5 +++
.../fluss/flink/catalog/FlinkCatalogTest.java | 25 ++++++++++++++
5 files changed, 109 insertions(+)
diff --git a/fluss-flink/fluss-flink-2.2/src/test/java/org/apache/fluss/flink/catalog/Flink22CatalogITCase.java b/fluss-flink/fluss-flink-2.2/src/test/java/org/apache/fluss/flink/catalog/Flink22CatalogITCase.java
index 3a5bea7a1a..4c0a342e38 100644
--- a/fluss-flink/fluss-flink-2.2/src/test/java/org/apache/fluss/flink/catalog/Flink22CatalogITCase.java
+++ b/fluss-flink/fluss-flink-2.2/src/test/java/org/apache/fluss/flink/catalog/Flink22CatalogITCase.java
@@ -28,6 +28,8 @@ import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.catalog.CatalogTable;
import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.CollectionUtil;
import org.junit.jupiter.api.Test;
import static org.assertj.core.api.Assertions.assertThat;
@@ -173,6 +175,24 @@ public class Flink22CatalogITCase extends FlinkCatalogITCase {
assertThat(table.getUnresolvedSchema()).isEqualTo(expectedSchema);
}
+ @Test
+ @MultiVersionTest
+ void testShowMaterializedTables() throws Exception {
+ tEnv.executeSql("create table show_mt_source (a int, b string)").await();
+ tEnv.executeSql(
+ "create materialized table show_mt freshness = interval '30' second"
+ + " as select a, b from show_mt_source")
+ .await();
+
+ assertThat(
+ CollectionUtil.iteratorToList(
+ tEnv.executeSql("show materialized tables").collect()))
+ .containsExactly(Row.of("show_mt"));
+
+ catalog.dropTable(new ObjectPath(DEFAULT_DB, "show_mt"), false);
+ catalog.dropTable(new ObjectPath(DEFAULT_DB, "show_mt_source"), false);
+ }
+
@Override
protected void addDefaultIndexKey(Schema.Builder schemaBuilder) {
super.addDefaultIndexKey(schemaBuilder);
diff --git a/fluss-flink/fluss-flink-2.3/src/test/java/org/apache/fluss/flink/catalog/Flink23CatalogITCase.java b/fluss-flink/fluss-flink-2.3/src/test/java/org/apache/fluss/flink/catalog/Flink23CatalogITCase.java
index dd255c3af5..a217f6153e 100644
--- a/fluss-flink/fluss-flink-2.3/src/test/java/org/apache/fluss/flink/catalog/Flink23CatalogITCase.java
+++ b/fluss-flink/fluss-flink-2.3/src/test/java/org/apache/fluss/flink/catalog/Flink23CatalogITCase.java
@@ -28,6 +28,8 @@ import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.catalog.CatalogTable;
import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.CollectionUtil;
import org.junit.jupiter.api.Test;
import static org.assertj.core.api.Assertions.assertThat;
@@ -173,6 +175,24 @@ public class Flink23CatalogITCase extends FlinkCatalogITCase {
assertThat(table.getUnresolvedSchema()).isEqualTo(expectedSchema);
}
+ @Test
+ @MultiVersionTest
+ void testShowMaterializedTables() throws Exception {
+ tEnv.executeSql("create table show_mt_source (a int, b string)").await();
+ tEnv.executeSql(
+ "create materialized table show_mt freshness = interval '30' second"
+ + " as select a, b from show_mt_source")
+ .await();
+
+ assertThat(
+ CollectionUtil.iteratorToList(
+ tEnv.executeSql("show materialized tables").collect()))
+ .containsExactly(Row.of("show_mt"));
+
+ catalog.dropTable(new ObjectPath(DEFAULT_DB, "show_mt"), false);
+ catalog.dropTable(new ObjectPath(DEFAULT_DB, "show_mt_source"), false);
+ }
+
@Override
protected void addDefaultIndexKey(Schema.Builder schemaBuilder) {
super.addDefaultIndexKey(schemaBuilder);
diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/catalog/FlinkCatalog.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/catalog/FlinkCatalog.java
index 25ceb4917a..5a3dfc85e2 100644
--- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/catalog/FlinkCatalog.java
+++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/catalog/FlinkCatalog.java
@@ -103,6 +103,7 @@ import java.util.Locale;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
import java.util.function.Supplier;
import java.util.stream.Collectors;
@@ -391,6 +392,44 @@ public class FlinkCatalog extends AbstractCatalog {
return Collections.emptyList();
}
+ /**
+ * Lists the names of all materialized tables in the given database.
+ *
+ * <p>Flink 2.2 added this method to {@code Catalog} for {@code SHOW MATERIALIZED TABLES}. It is
+ * declared without {@code @Override} so that the class still compiles against older Flink
+ * versions, which never call it.
+ *
+ * @throws DatabaseNotExistException if the database does not exist
+ */
+ public List<String> listMaterializedTables(String databaseName)
+ throws DatabaseNotExistException, CatalogException {
+ List<String> tableNames = listTables(databaseName);
+ List<CompletableFuture<TableInfo>> tableInfoFutures = new ArrayList<>(tableNames.size());
+ for (String tableName : tableNames) {
+ tableInfoFutures.add(admin.getTableInfo(TablePath.of(databaseName, tableName)));
+ }
+
+ List<String> materializedTables = new ArrayList<>();
+ for (int i = 0; i < tableNames.size(); i++) {
+ try {
+ if (FlinkConversions.isMaterializedTable(tableInfoFutures.get(i).get())) {
+ materializedTables.add(tableNames.get(i));
+ }
+ } catch (Exception e) {
+ Throwable t = ExceptionUtils.stripExecutionException(e);
+ // the table may have been dropped after it was listed
+ if (!isTableNotExist(t)) {
+ throw new CatalogException(
+ String.format(
+ "Failed to list materialized tables in database %s in %s",
+ databaseName, getName()),
+ t);
+ }
+ }
+ }
+ return materializedTables;
+ }
+
@Override
public CatalogBaseTable getTable(ObjectPath objectPath)
throws TableNotExistException, CatalogException {
diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/utils/FlinkConversions.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/utils/FlinkConversions.java
index ec949ab531..5df19d29f8 100644
--- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/utils/FlinkConversions.java
+++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/utils/FlinkConversions.java
@@ -589,6 +589,11 @@ public class FlinkConversions {
encodeBytesToBase64(newHandlerBytes)));
}
+ /** Returns whether the given Fluss table stores a Flink materialized table. */
+ public static boolean isMaterializedTable(TableInfo tableInfo) {
+ return isMaterializedTable(tableInfo.getCustomProperties().toMap());
+ }
+
/**
* Determines if the given options represent a materialized table.
*
diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/catalog/FlinkCatalogTest.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/catalog/FlinkCatalogTest.java
index cdcee44d37..6d1862616b 100644
--- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/catalog/FlinkCatalogTest.java
+++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/catalog/FlinkCatalogTest.java
@@ -560,6 +560,31 @@ class FlinkCatalogTest {
nonExistDbPath.getDatabaseName(), CATALOG_NAME);
}
+ @Test
+ @MultiVersionTest
+ void testListMaterializedTables() throws Exception {
+ FlinkCatalog flinkCatalog = (FlinkCatalog) catalog;
+ assertThat(flinkCatalog.listMaterializedTables(DEFAULT_DB)).isEmpty();
+
+ catalog.createTable(tableInDefaultDb, newCatalogTable(Collections.emptyMap()), false);
+ CatalogMaterializedTable materializedTable =
+ newCatalogMaterializedTable(
+ this.createSchema(),
+ CatalogMaterializedTable.RefreshMode.CONTINUOUS,
+ Collections.emptyMap());
+ catalog.createTable(new ObjectPath(DEFAULT_DB, "mt1"), materializedTable, false);
+ catalog.createTable(new ObjectPath(DEFAULT_DB, "mt2"), materializedTable, false);
+
+ assertThat(flinkCatalog.listMaterializedTables(DEFAULT_DB))
+ .containsExactlyInAnyOrder("mt1", "mt2");
+ assertThat(catalog.listTables(DEFAULT_DB)).containsExactlyInAnyOrder("t1", "mt1", "mt2");
+
+ assertThatThrownBy(() -> flinkCatalog.listMaterializedTables("non_exist_db"))
+ .isInstanceOf(DatabaseNotExistException.class)
+ .hasMessage(
+ "Database %s does not exist in Catalog %s.", "non_exist_db", CATALOG_NAME);
+ }
+
@Test
void testAlterMaterializedTable() throws Exception {
ObjectPath mt2 = new ObjectPath(DEFAULT_DB, "mt2");