[flink] Support Flink Materialized Table

docs/layouts/shortcodes/generated/core_configuration.html

@@ -886,7 +886,7 @@
             <td><h5>type</h5></td>
             <td style="word-wrap: break-word;">table</td>
             <td><p>Enum</p></td>
-            <td>Type of the table.<br /><br />Possible values:<ul><li>"table": Normal Paimon table.</li><li>"format-table": A file format table refers to a directory that contains multiple files of the same format.</li></ul></td>
+            <td>Type of the table.<br /><br />Possible values:<ul><li>"table": Normal Paimon table.</li><li>"format-table": A file format table refers to a directory that contains multiple files of the same format.</li><li>"materialized-table": A materialized table.</li></ul></td>
         </tr>
         <tr>
             <td><h5>write-buffer-for-append</h5></td>

paimon-common/src/main/java/org/apache/paimon/CoreOptions.java

@@ -1373,6 +1373,75 @@ public class CoreOptions implements Serializable {
                     .withDescription(
                             "Whether to enable asynchronous IO writing when writing files.");
 
+    @ExcludeFromDocumentation("Only used internally to support materialized table")
+    public static final ConfigOption<Long> MATERIALIZED_TABLE_SNAPSHOT =
+            key("materialized-table.snapshot")
+                    .longType()
+                    .noDefaultValue()
+                    .withDescription("The snapshot specified for the materialized table");
+
+    @ExcludeFromDocumentation("Only used internally to support materialized table")
+    public static final ConfigOption<String> MATERIALIZED_TABLE_DEFINITION_QUERY =
+            key("materialized-table.definition-query")
+                    .stringType()
+                    .noDefaultValue()
+                    .withDescription(
+                            "The definition query text of materialized table, text is expanded in contrast to the original SQL.");
+
+    @ExcludeFromDocumentation("Only used internally to support materialized table")
+    public static final ConfigOption<String> MATERIALIZED_TABLE_INTERVAL_FRESHNESS =
+            key("materialized-table.interval-freshness")
+                    .stringType()
+                    .noDefaultValue()
+                    .withDescription(
+                            "the freshness interval of materialized table which is used to determine the physical refresh mode.");
+
+    @ExcludeFromDocumentation("Only used internally to support materialized table")
+    public static final ConfigOption<MaterializedTableIntervalFreshnessTimeUnit>
+            MATERIALIZED_TABLE_INTERVAL_FRESHNESS_TIME_UNIT =
+                    key("materialized-table.interval-freshness.time-unit")
+                            .enumType(MaterializedTableIntervalFreshnessTimeUnit.class)
+                            .noDefaultValue()
+                            .withDescription("The time unit of freshness interval.");
+
+    @ExcludeFromDocumentation("Only used internally to support materialized table")
+    public static final ConfigOption<MaterializedTableRefreshMode>
+            MATERIALIZED_TABLE_LOGICAL_REFRESH_MODE =
+                    key("materialized-table.logical-refresh-mode")
+                            .enumType(MaterializedTableRefreshMode.class)
+                            .noDefaultValue()
+                            .withDescription("the logical refresh mode of materialized table.");
+
+    @ExcludeFromDocumentation("Only used internally to support materialized table")
+    public static final ConfigOption<MaterializedTableRefreshMode> MATERIALIZED_TABLE_REFRESH_MODE =
+            key("materialized-table.refresh-mode")
+                    .enumType(MaterializedTableRefreshMode.class)
+                    .noDefaultValue()
+                    .withDescription("the physical refresh mode of materialized table.");
+
+    @ExcludeFromDocumentation("Only used internally to support materialized table")
+    public static final ConfigOption<MaterializedTableRefreshStatus>
+            MATERIALIZED_TABLE_REFRESH_STATUS =
+                    key("materialized-table.refresh-status")
+                            .enumType(MaterializedTableRefreshStatus.class)
+                            .noDefaultValue()
+                            .withDescription("the refresh status of materialized table.");
+
+    @ExcludeFromDocumentation("Only used internally to support materialized table")
+    public static final ConfigOption<String> MATERIALIZED_TABLE_REFRESH_HANDLER_DESCRIPTION =
+            key("materialized-table.refresh-handler-description")
+                    .stringType()
+                    .noDefaultValue()
+                    .withDescription(
+                            "The summary description of materialized table's refresh handler");
+
+    @ExcludeFromDocumentation("Only used internally to support materialized table")
+    public static final ConfigOption<String> MATERIALIZED_TABLE_REFRESH_HANDLER_BYTES =
+            key("materialized-table.refresh-handler-bytes")
+                    .stringType()
+                    .noDefaultValue()
+                    .withDescription("The serialized refresh handler of materialized table.");
+
     private final Options options;
 
     public CoreOptions(Map<String, String> options) {
@@ -2801,4 +2870,34 @@ public InlineElement getDescription() {
             return text(description);
         }
     }
+
+    /** The time unit of materialized table freshness. */
+    public enum MaterializedTableIntervalFreshnessTimeUnit {
+        SECOND,
+        MINUTE,
+        HOUR,
+        DAY
+    }
+
+    /** The refresh mode of materialized table. */
+    public enum MaterializedTableRefreshMode {
+        /** The refresh pipeline will be executed in continuous mode. */
+        CONTINUOUS,
+
+        /** The refresh pipeline will be executed in full mode. */
+        FULL,
+
+        /**
+         * The refresh pipeline mode is determined by freshness of materialized table, either {@link
+         * #FULL} or {@link #CONTINUOUS}.
+         */
+        AUTOMATIC
+    }
+
+    /** The refresh status of materialized table. */
+    public enum MaterializedTableRefreshStatus {
+        INITIALIZING,
+        ACTIVATED,
+        SUSPENDED
+    }
 }

paimon-common/src/main/java/org/apache/paimon/TableType.java

@@ -28,8 +28,8 @@ public enum TableType implements DescribedEnum {
     TABLE("table", "Normal Paimon table."),
     FORMAT_TABLE(
             "format-table",
-            "A file format table refers to a directory that contains multiple files of the same format.");
-
+            "A file format table refers to a directory that contains multiple files of the same format."),
+    MATERIALIZED_TABLE("materialized-table", "A materialized table.");
     private final String value;
     private final String description;
 

paimon-flink/paimon-flink-1.15/src/main/java/org/apache/flink/table/catalog/CatalogMaterializedTable.java

@@ -0,0 +1,34 @@
+/*
+ * 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.catalog;
+
+/**
+ * Dummy placeholder to resolve compatibility issue of CatalogMaterializedTable(introduced in flink
+ * 1.20).
+ */
+public interface CatalogMaterializedTable extends CatalogBaseTable {
+    /** Dummy LogicalRefreshMode placeholder. */
+    enum LogicalRefreshMode {}
+
+    /** Dummy RefreshMode placeholder. */
+    enum RefreshMode {}
+
+    /** Dummy RefreshStatus placeholder. */
+    enum RefreshStatus {}
+}

paimon-flink/paimon-flink-1.16/src/main/java/org/apache/flink/table/catalog/CatalogMaterializedTable.java

@@ -0,0 +1,34 @@
+/*
+ * 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.catalog;
+
+/**
+ * Dummy placeholder to resolve compatibility issue of CatalogMaterializedTable(introduced in flink
+ * 1.20).
+ */
+public interface CatalogMaterializedTable extends CatalogBaseTable {
+    /** Dummy LogicalRefreshMode placeholder. */
+    enum LogicalRefreshMode {}
+
+    /** Dummy RefreshMode placeholder. */
+    enum RefreshMode {}
+
+    /** Dummy RefreshStatus placeholder. */
+    enum RefreshStatus {}
+}

paimon-flink/paimon-flink-1.17/src/main/java/org/apache/flink/table/catalog/CatalogMaterializedTable.java

@@ -0,0 +1,34 @@
+/*
+ * 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.catalog;
+
+/**
+ * Dummy placeholder to resolve compatibility issue of CatalogMaterializedTable(introduced in flink
+ * 1.20).
+ */
+public interface CatalogMaterializedTable extends CatalogBaseTable {
+    /** Dummy LogicalRefreshMode placeholder. */
+    enum LogicalRefreshMode {}
+
+    /** Dummy RefreshMode placeholder. */
+    enum RefreshMode {}
+
+    /** Dummy RefreshStatus placeholder. */
+    enum RefreshStatus {}
+}

paimon-flink/paimon-flink-1.18/src/main/java/org/apache/flink/table/catalog/CatalogMaterializedTable.java

@@ -0,0 +1,34 @@
+/*
+ * 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.catalog;
+
+/**
+ * Dummy placeholder to resolve compatibility issue of CatalogMaterializedTable(introduced in flink
+ * 1.20).
+ */
+public interface CatalogMaterializedTable extends CatalogBaseTable {
+    /** Dummy LogicalRefreshMode placeholder. */
+    enum LogicalRefreshMode {}
+
+    /** Dummy RefreshMode placeholder. */
+    enum RefreshMode {}
+
+    /** Dummy RefreshStatus placeholder. */
+    enum RefreshStatus {}
+}

paimon-flink/paimon-flink-1.19/src/main/java/org/apache/flink/table/catalog/CatalogMaterializedTable.java

@@ -0,0 +1,34 @@
+/*
+ * 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.catalog;
+
+/**
+ * Dummy placeholder to resolve compatibility issue of CatalogMaterializedTable(introduced in flink
+ * 1.20).
+ */
+public interface CatalogMaterializedTable extends CatalogBaseTable {
+    /** Dummy LogicalRefreshMode placeholder. */
+    enum LogicalRefreshMode {}
+
+    /** Dummy RefreshMode placeholder. */
+    enum RefreshMode {}
+
+    /** Dummy RefreshStatus placeholder. */
+    enum RefreshStatus {}
+}

paimon-flink/paimon-flink-cdc/src/main/java/org/apache/flink/table/catalog/CatalogMaterializedTable.java

@@ -0,0 +1,34 @@
+/*
+ * 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.catalog;
+
+/**
+ * Dummy placeholder to resolve compatibility issue of CatalogMaterializedTable(introduced in flink
+ * 1.20).
+ */
+public interface CatalogMaterializedTable extends CatalogBaseTable {
+    /** Dummy LogicalRefreshMode placeholder. */
+    enum LogicalRefreshMode {}
+
+    /** Dummy RefreshMode placeholder. */
+    enum RefreshMode {}
+
+    /** Dummy RefreshStatus placeholder. */
+    enum RefreshStatus {}
+}

paimon-flink/paimon-flink-common/pom.xml

@@ -59,6 +59,28 @@ under the License.
             <scope>provided</scope>
         </dependency>
 
+        <dependency>
+          <groupId>org.apache.flink</groupId>
+          <artifactId>flink-clients</artifactId>
+          <version>${flink.version}</version>
+          <scope>test</scope>
+        </dependency>
+
+        <dependency>
+          <groupId>org.apache.flink</groupId>
+          <artifactId>flink-sql-gateway</artifactId>
+          <version>${flink.version}</version>
+          <scope>test</scope>
+        </dependency>
+
+        <dependency>
+          <groupId>org.apache.flink</groupId>
+          <artifactId>flink-sql-gateway</artifactId>
+          <version>${flink.version}</version>
+          <type>test-jar</type>
+          <scope>test</scope>
+        </dependency>
+
         <!-- lookup service -->
 
         <dependency>