You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ro...@apache.org on 2024/07/05 09:55:47 UTC
(flink) 02/03: [FLINK-35643][doc] Add materialized table statement doc
This is an automated email from the ASF dual-hosted git repository.
ron pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 87be68a3dd6cac6e67d762a21d2e684c0573b958
Author: Feng Jin <ji...@gmail.com>
AuthorDate: Sun Jun 23 20:43:59 2024 +0800
[FLINK-35643][doc] Add materialized table statement doc
This closes #24975
---
docs/content.zh/docs/dev/table/functions/_index.md | 4 +-
.../{functions => materialized-table}/_index.md | 4 +-
.../docs/dev/table/materialized-table/overview.md | 66 ++++
.../dev/table/materialized-table/statements.md | 348 +++++++++++++++++++++
docs/content/docs/dev/table/functions/_index.md | 4 +-
.../docs/dev/table/materialized-table}/_index.md | 4 +-
.../docs/dev/table/materialized-table/overview.md | 65 ++++
.../dev/table/materialized-table/statements.md | 342 ++++++++++++++++++++
8 files changed, 829 insertions(+), 8 deletions(-)
diff --git a/docs/content.zh/docs/dev/table/functions/_index.md b/docs/content.zh/docs/dev/table/functions/_index.md
index 7e059a9a1c7..b01f9295fe5 100644
--- a/docs/content.zh/docs/dev/table/functions/_index.md
+++ b/docs/content.zh/docs/dev/table/functions/_index.md
@@ -1,7 +1,7 @@
---
title: 函数
bookCollapseSection: true
-weight: 33
+weight: 34
---
<!--
Licensed to the Apache Software Foundation (ASF) under one
@@ -20,4 +20,4 @@ software distributed under the License is distributed on an
KIND, either express or implied. See the License for the
specific language governing permissions and limitations
under the License.
--->
\ No newline at end of file
+-->
diff --git a/docs/content.zh/docs/dev/table/functions/_index.md b/docs/content.zh/docs/dev/table/materialized-table/_index.md
similarity index 96%
copy from docs/content.zh/docs/dev/table/functions/_index.md
copy to docs/content.zh/docs/dev/table/materialized-table/_index.md
index 7e059a9a1c7..9dbebca64bb 100644
--- a/docs/content.zh/docs/dev/table/functions/_index.md
+++ b/docs/content.zh/docs/dev/table/materialized-table/_index.md
@@ -1,5 +1,5 @@
---
-title: 函数
+title: Materialized Table
bookCollapseSection: true
weight: 33
---
@@ -20,4 +20,4 @@ software distributed under the License is distributed on an
KIND, either express or implied. See the License for the
specific language governing permissions and limitations
under the License.
--->
\ No newline at end of file
+-->
diff --git a/docs/content.zh/docs/dev/table/materialized-table/overview.md b/docs/content.zh/docs/dev/table/materialized-table/overview.md
new file mode 100644
index 00000000000..da82ca0ebc9
--- /dev/null
+++ b/docs/content.zh/docs/dev/table/materialized-table/overview.md
@@ -0,0 +1,66 @@
+---
+title: 概览
+weight: 1
+type: docs
+aliases:
+- /dev/table/materialized-table/
+---
+<!--
+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.
+-->
+
+# 介绍
+
+物化表是 Flink SQL 引入的一种新的表类型，旨在简化批处理和流处理数据管道，提供一致的开发体验。在创建物化表时，通过指定数据新鲜度和查询，Flink 引擎会自动推导出物化表的 Schema ，并创建相应的数据刷新管道，以达到指定的新鲜度。
+
+{{< hint warning >}}
+**注意**：该功能目前是一个 MVP（最小可行产品）功能，仅在 [SQL Gateway]({{< ref "docs/dev/table/sql-gateway/overview" >}})中可用，并且只支持部署作业到 Flink [Standalone]({{< ref "docs/deployment/resource-providers/standalone/overview" >}})集群。
+{{< /hint >}}
+
+# 核心概念
+
+物化表包含以下核心概念：数据新鲜度、刷新模式、查询定义和 `Schema` 。
+
+## 数据新鲜度
+
+数据新鲜度定义了物化表数据相对于基础表更新的最大滞后时间。它并非绝对保证，而是 Flink 尝试达到的目标。框架会尽力确保物化表中的数据在指定的新鲜度内刷新。
+
+数据新鲜度是物化表的一个关键属性，具有两个主要作用：
+- **确定刷新模式**：目前有连续模式和全量模式。关于如何确定刷新模式的详细信息，请参阅 [materialized-table.refresh-mode.freshness-threshold]({{< ref "docs/dev/table/config" >}}#materialized-table-refresh-mode-freshness-threshold) 配置项。
+ - 连续模式：启动 Flink 流作业，持续刷新物化表数据。
+ - 全量模式：工作流调度器定期触发 Flink 批处理作业，全量刷新物化表数据。
+- **确定刷新频率**：
+ - 连续模式下，数据新鲜度转换为 Flink 流作业的 `checkpoint` 间隔。
+ - 全量模式下，数据新鲜度转换为工作流的调度周期，例如 `cron` 表达式。
+
+## 刷新模式
+
+刷新模式有连续模式和全量模式两种。默认情况下，根据数据新鲜度推断刷新模式。用户可以为特定业务场景显式指定刷新模式，它的优先级高于根据数据新鲜度推导的刷新模式。
+
+- **连续模式**：Flink 流作业会增量更新物化表数据，下游数据会立即可见，或者等 checkpoint 完成时才可见，由对应的 Connector 行为决定。
+- **全量模式**：调度器会定期触发对物化表数据的全量覆盖，其数据刷新周期与工作流的调度周期相匹配。
+ - 默认的覆盖行为是表级别的。如果分区字段存在，并且通过 [partition.fields.#.date-formatter]({{< ref "docs/dev/table/config" >}}#partition-fields-date-formatter) 指定了时间分区字段格式，则按照分区粒度覆盖，即每次只刷新最新的分区。
+
+## 查询定义
+
+物化表的查询定义支持所有 Flink SQL [查询]({{< ref "docs/dev/table/sql/queries/overview" >}})。查询结果用于填充物化表。在连续模式下，查询结果会持续更新到物化表中，而在全量模式下，每次查询结果都会覆盖更新到物化表。
+
+## Schema
+
+物化表的 `Schema` 定义与普通表相同，可以声明主键和分区字段。其列名和类型会从相应的查询中推导，用户无法手动指定。
+
diff --git a/docs/content.zh/docs/dev/table/materialized-table/statements.md b/docs/content.zh/docs/dev/table/materialized-table/statements.md
new file mode 100644
index 00000000000..ceda0c92364
--- /dev/null
+++ b/docs/content.zh/docs/dev/table/materialized-table/statements.md
@@ -0,0 +1,348 @@
+---
+title: 语法说明
+weight: 2
+type: docs
+aliases:
+- /dev/table/materialized-table/statements.html
+---
+<!--
+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.
+-->
+
+# 物化表语法
+
+Flink SQL 目前支持以下物化表操作：
+- [CREATE MATERIALIZED TABLE](#create-materialized-table)
+- [ALTER MATERIALIZED TABLE](#alter-materialized-table)
+- [DROP MATERIALIZED TABLE](#drop-materialized-table)
+
+# CREATE MATERIALIZED TABLE
+
+```
+CREATE MATERIALIZED TABLE [catalog_name.][db_name.]table_name
+
+[ ([ <table_constraint> ]) ]
+
+[COMMENT table_comment]
+
+[PARTITIONED BY (partition_column_name1, partition_column_name2, ...)]
+
+[WITH (key1=val1, key2=val2, ...)]
+
+FRESHNESS = INTERVAL '<num>' { SECOND | MINUTE | HOUR | DAY }
+
+[REFRESH_MODE = { CONTINUOUS | FULL }]
+
+AS <select_statement>
+
+<table_constraint>:
+ [CONSTRAINT constraint_name] PRIMARY KEY (column_name, ...) NOT ENFORCED
+```
+
+## PRIMARY KEY
+
+`PRIMARY KEY` 定义了一组可选的列，用于唯一标识表中的每一行。主键列必须非空。
+
+## PARTITIONED BY
+
+`PARTITIONED BY` 定义了一组可选的列，用于对物化表进行分区。如果物化表使用了 `filesystem connector` ，每个分区将创建一个目录。
+
+**示例：**
+
+```sql
+-- 创建一个物化表并将分区字段设置为 `ds`。
+CREATE MATERIALIZED TABLE my_materialized_table
+ PARTITIONED BY (ds)
+ FRESHNESS = INTERVAL '1' HOUR
+ AS SELECT
+ ds
+ FROM
+ ...
+```
+
+<span class="label label-danger">注意</span>
+- 分区字段必须是物化表查询语句中的字段。
+
+## WITH Options
+
+`WITH Options` 可以定义创建物化表所需的属性，包括[连接器参数]({{< ref "docs/connectors/table/" >}})和分区字段的[时间格式参数]({{< ref "docs/dev/table/config" >}}#partition-fields-date-formatter)。
+
+```sql
+-- 创建一个物化表，指定分区字段为 'ds' 和对应的时间格式为 'yyyy-MM-dd'
+CREATE MATERIALIZED TABLE my_materialized_table
+ PARTITIONED BY (ds)
+ WITH (
+ 'format' = 'json',
+ 'partition.fields.ds.date-formatter' = 'yyyy-MM-dd'
+ )
+ ...
+```
+
+如上例所示，我们为 `ds` 分区列指定了 `date-formatter` 选项。每次调度时，调度时间将转换为相应的 `ds` 分区值。例如，在 `2024-01-01 00:00:00` 的调度时间下，只有分区 `ds = '2024-01-01'` 会被刷新。
+
+<span class="label label-danger">注意</span>
+- `partition.fields.#.date-formatter` 选项仅适用于全量模式。
+- [partition.fields.#.date-formatter]({{< ref "docs/dev/table/config" >}}#partition-fields-date-formatter) 中的字段必须是有效的字符串类型分区字段。
+
+## FRESHNESS
+
+`FRESHNESS` 用于指定物化表的数据新鲜度。
+
+**数据新鲜度与刷新模式关系**
+
+数据新鲜度定义了物化表内容滞后于基础表更新的最长时间。它有两个作用，首先通过[配置]({{< ref "docs/dev/table/config" >}}#materialized-table-refresh-mode-freshness-threshold)确定物化表的[刷新模式]({{< ref "docs/dev/table/materialized-table/overview" >}}#刷新模式)，然后确定数据刷新频率以满足实际数据新鲜度要求。
+
+**FRESHNESS 参数详解**
+
+FRESHNESS 参数的取值范围为 INTERVAL `'<num>'` { SECOND | MINUTE | HOUR | DAY }。`'<num>'` 必须为正整数，并且在全量模式下，`'<num>'` 应该是相应时间间隔单位的公约数。
+
+**示例：**
+（假定 `materialized-table.refresh-mode.freshness-threshold` 为 30 分钟）
+
+```sql
+-- 对应的刷新管道是一个 checkpoint 间隔为 1 秒的流处理作业
+FRESHNESS = INTERVAL '1' SECOND
+
+-- 对应的刷新管道是一个 checkpoint 间隔为 1 分钟的流处理作业
+FRESHNESS = INTERVAL '1' MINUTE
+
+-- 对应的刷新管道是一个调度周期为 1 小时的调度工作流
+FRESHNESS = INTERVAL '1' HOUR
+
+-- 对应的刷新管道是一个调度周期为 1 天的调度工作流
+FRESHNESS = INTERVAL '1' DAY
+```
+
+**不合法的 `FRESHNESS` 示例：**
+
+```sql
+-- 间隔为负数
+FRESHNESS = INTERVAL '-1' SECOND
+
+-- 间隔为0
+FRESHNESS = INTERVAL '0' SECOND
+
+-- 间隔为月或者年
+FRESHNESS = INTERVAL '1' MONTH
+FRESHNESS = INTERVAL '1' YEAR
+
+-- 全量模式下，间隔不是对应时间间隔单位的公约数
+FRESHNESS = INTERVAL '60' SECOND
+FRESHNESS = INTERVAL '5' HOUR
+```
+
+<span class="label label-danger">注意</span>
+- 尽管物化表数据将尽可能在定义的新鲜度内刷新，但不能保证完全满足新鲜度要求。
+- 在连续模式下，数据新鲜度和 `checkpoint` 间隔一致，设置过短的数据新鲜度可能会对作业性能产生影响。此外，为了优化 `checkpoint` 性能，建议[开启 Changelog]({{< ref "docs/ops/state/state_backends" >}}#开启-changelog)。
+- 在全量模式下，数据新鲜度会转换为 `cron` 表达式，因此目前仅支持在预定义时间间隔单位内的新鲜度间隔，这种设计确保了与 `cron` 表达式语义的一致性。具体支持以下新鲜度间隔：
+ - 秒：30、15、10、5、2 和 1 秒间隔。
+ - 分钟：30、15、10、5、2 和 1 分钟间隔。
+ - 小时：8、4、2 和 1 小时间隔。
+ - 天：1 天。
+
+## REFRESH_MODE
+
+`REFRESH_MODE` 用于显式指定物化表的刷新模式。指定的刷新模式比框架自动推导的模式具有更高的优先级，以满足特定场景的需求。
+
+**示例：**
+(假定 `materialized-table.refresh-mode.freshness-threshold` 为 30 分钟)
+
+```sql
+-- 创建的物化表的刷新模式为连续模式，作业的 checkpoint 间隔为 1 小时。
+CREATE MATERIALIZED TABLE my_materialized_table
+ REFRESH_MODE = CONTINUOUS
+ FRESHNESS = INTERVAL '1' HOUR
+ AS SELECT
+ ...
+
+-- 创建的物化表的刷新模式为全量模式，作业的调度周期为 10 分钟。
+CREATE MATERIALIZED TABLE my_materialized_table
+ REFRESH_MODE = FULL
+ FRESHNESS = INTERVAL '10' MINUTE
+ AS SELECT
+ ...
+```
+
+## AS <select_statement>
+
+该子句用于定义填充物化表数据的查询。上游表可以是物化表、表或视图。select 语句支持所有 Flink SQL [查询]({{< ref "docs/dev/table/sql/queries/overview" >}})。
+
+**示例：**
+
+```sql
+CREATE MATERIALIZED TABLE my_materialized_table
+ FRESHNESS = INTERVAL '10' SECOND
+ AS SELECT * FROM kafka_catalog.db1.kafka_table;
+```
+
+## 示例
+
+（假设 materialized-table.refresh-mode.freshness-threshold 为 30 分钟）
+
+创建一个数据新鲜度为 `10` 秒的物化表，推导出的刷新模式为连续模式:
+
+```sql
+CREATE MATERIALIZED TABLE my_materialized_table_continuous
+ PARTITIONED BY (ds)
+ WITH (
+ 'format' = 'json',
+ 'partition.fields.ds.date-formatter' = 'yyyy-MM-dd'
+ )
+ FRESHNESS = INTERVAL '10' SECOND
+ REFRESH_MODE = 'CONTINUOUS'
+ AS
+ SELECT
+ k.ds,
+ k.user_id,
+ COUNT(*) AS event_count,
+ SUM(k.amount) AS total_amount,
+ MAX(u.age) AS max_age
+ FROM
+ kafka_catalog.db1.kafka_table k
+ JOIN
+ user_catalog.db1.user_table u
+ ON
+ k.user_id = u.user_id
+ WHERE
+ k.event_type = 'purchase'
+ GROUP BY
+ k.ds, k.user_id
+```
+
+创建一个数据新鲜度为 `1` 小时的物化表，推导出的刷新模式为全量模式:
+
+```sql
+CREATE MATERIALIZED TABLE my_materialized_table_full
+ PARTITIONED BY (ds)
+ WITH (
+ 'format' = 'json',
+ 'partition.fields.ds.date-formatter' = 'yyyy-MM-dd'
+ )
+ FRESHNESS = INTERVAL '10' MINUTE
+ REFRESH_MODE = 'FULL'
+ AS
+ SELECT
+ p.ds,
+ p.product_id,
+ p.product_name,
+ AVG(s.sale_price) AS avg_sale_price,
+ SUM(s.quantity) AS total_quantity
+ FROM
+ paimon_catalog.db1.product_table p
+ LEFT JOIN
+ paimon_catalog.db1.sales_table s
+ ON
+ p.product_id = s.product_id
+ WHERE
+ p.category = 'electronics'
+ GROUP BY
+ p.ds, p.product_id, p.product_name
+```
+
+## 限制
+- 不支持显式指定列
+- 不支持修改查询语句
+- 不支持在 select 查询中使用临时表、临时视图或临时函数
+
+# ALTER MATERIALIZED TABLE
+
+```
+ALTER MATERIALIZED TABLE [catalog_name.][db_name.]table_name SUSPEND | RESUME [WITH (key1=val1, key2=val2, ...)] | REFRESH [PARTITION partition_spec]
+```
+
+`ALTER MATERIALIZED TABLE` 用于管理物化表。用户可以使用此命令暂停和恢复物化表的刷新管道，并手动触发数据刷新。
+
+
+## SUSPEND
+
+```
+ALTER MATERIALIZED TABLE [catalog_name.][db_name.]table_name SUSPEND
+```
+
+`SUSPEND` 用于暂停物化表的后台刷新管道。
+
+**示例:**
+
+```sql
+-- 暂停前指定 SAVEPOINT 路径
+SET 'execution.checkpointing.savepoint-dir' = 'hdfs://savepoint_path';
+
+-- 暂停指定的物化表
+ALTER MATERIALIZED TABLE my_materialized_table SUSPEND;
+```
+
+<span class="label label-danger">注意</span>
+- 暂停连续模式的表时，默认会使用 `STOP WITH SAVEPOINT` 暂停作业，你需要使用[参数]({{< ref "docs/deployment/config" >}}#execution-checkpointing-savepoint-dir)设置 `SAVEPOINT` 保存路径。
+
+## RESUME
+
+```
+ALTER MATERIALIZED TABLE [catalog_name.][db_name.]table_name RESUME [WITH (key1=val1, key2=val2, ...)]
+```
+
+`RESUME` 用于恢复物化表的刷新管道。在恢复时，可以通过 `WITH` 子句动态指定物化表的参数，该参数仅对当前恢复的刷新管道生效，并不会持久化到物化表中。
+
+**示例:**
+
+```sql
+-- 恢复指定的物化表
+ALTER MATERIALIZED TABLE my_materialized_table RESUME;
+
+-- 恢复指定的物化表并指定 sink 的并行度
+ALTER MATERIALIZED TABLE my_materialized_table RESUME WITH ('sink.parallelism'='10');
+```
+
+## REFRESH
+
+```
+ALTER MATERIALIZED TABLE [catalog_name.][db_name.]table_name REFRESH [PARTITION partition_spec]
+```
+
+`REFRESH` 用于主动触发物化表的刷新。
+
+**示例:**
+
+```sql
+-- 刷新全表数据
+ALTER MATERIALIZED TABLE my_materialized_table REFRESH;
+
+-- 刷新指定分区数据
+ALTER MATERIALIZED TABLE my_materialized_table REFRESH PARTITION (ds='2024-06-28');
+```
+
+<span class="label label-danger">注意</span>
+- REFRESH 操作会启动批作业来刷新表的数据。
+
+# DROP MATERIALIZED TABLE
+
+```text
+DROP MATERIALIZED TABLE [IF EXISTS] [catalog_name.][database_name.]table_name
+```
+
+删除物化表时，首先删除后台刷新管道，然后从 Catalog 中移除该物化表的元数据。
+
+**示例:**
+
+```sql
+-- 删除指定的物化表
+DROP MATERIALIZED TABLE IF EXISTS my_materialized_table;
+```
+
+
+
+
diff --git a/docs/content/docs/dev/table/functions/_index.md b/docs/content/docs/dev/table/functions/_index.md
index c1cd26a8fa3..b982b054845 100644
--- a/docs/content/docs/dev/table/functions/_index.md
+++ b/docs/content/docs/dev/table/functions/_index.md
@@ -1,7 +1,7 @@
---
title: Functions
bookCollapseSection: true
-weight: 33
+weight: 34
---
<!--
Licensed to the Apache Software Foundation (ASF) under one
@@ -20,4 +20,4 @@ software distributed under the License is distributed on an
KIND, either express or implied. See the License for the
specific language governing permissions and limitations
under the License.
--->
\ No newline at end of file
+-->
diff --git a/docs/content.zh/docs/dev/table/functions/_index.md b/docs/content/docs/dev/table/materialized-table/_index.md
similarity index 96%
copy from docs/content.zh/docs/dev/table/functions/_index.md
copy to docs/content/docs/dev/table/materialized-table/_index.md
index 7e059a9a1c7..9dbebca64bb 100644
--- a/docs/content.zh/docs/dev/table/functions/_index.md
+++ b/docs/content/docs/dev/table/materialized-table/_index.md
@@ -1,5 +1,5 @@
---
-title: 函数
+title: Materialized Table
bookCollapseSection: true
weight: 33
---
@@ -20,4 +20,4 @@ software distributed under the License is distributed on an
KIND, either express or implied. See the License for the
specific language governing permissions and limitations
under the License.
--->
\ No newline at end of file
+-->
diff --git a/docs/content/docs/dev/table/materialized-table/overview.md b/docs/content/docs/dev/table/materialized-table/overview.md
new file mode 100644
index 00000000000..a2e6c830773
--- /dev/null
+++ b/docs/content/docs/dev/table/materialized-table/overview.md
@@ -0,0 +1,65 @@
+---
+title: Overview
+weight: 1
+type: docs
+aliases:
+- /dev/table/materialized-table.html
+---
+<!--
+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.
+-->
+
+# Introduction
+
+Materialized Table is a new table type introduced in Flink SQL, aimed at simplifying both batch and stream data pipelines, providing a consistent development experience. By specifying data freshness and query when creating Materialized Table, the engine automatically derives the schema for the materialized table and creates corresponding data refresh pipeline to achieve the specified freshness.
+
+{{< hint warning >}}
+**Note**: This feature is currently an MVP (“minimum viable product”) feature and only available within [SQL Gateway]({{< ref "docs/dev/table/sql-gateway/overview" >}}) which connected to a [Standalone]({{< ref "docs/deployment/resource-providers/standalone/overview" >}}) deployed Flink cluster.
+{{< /hint >}}
+
+# Core Concepts
+
+Materialized Table encompass the following core concepts: Data Freshness, Refresh Mode, Query Definition and Schema.
+
+## Data Freshness
+
+Data freshness defines the maximum amount of time that the materialized table’s content should lag behind updates to the base tables. Freshness is not a guarantee. Instead, it is a target that Flink attempts to meet. Data in materialized table is refreshed as closely as possible within the freshness.
+
+Data freshness is a crucial attribute of a materialized table, serving two main purposes:
+- **Determining the Refresh Mode**. Currently, there are CONTINUOUS and FULL modes. For details on how to determine the refresh mode, refer to the [materialized-table.refresh-mode.freshness-threshold]({{< ref "docs/dev/table/config" >}}#materialized-table-refresh-mode-freshness-threshold) configuration item.
+ - CONTINUOUS mode: Launches a Flink streaming job that continuously refreshes the materialized table data.
+ - FULL mode: The workflow scheduler periodically triggers a Flink batch job to refresh the materialized table data.
+- **Determining the Refresh Frequency**.
+ - In CONTINUOUS mode, data freshness is converted into the `checkpoint` interval of the Flink streaming job currently.
+ - In FULL mode, data freshness is converted into the scheduling cycle of the workflow, e.g. cron expression.
+
+## Refresh Mode
+
+There are two refresh modes: FULL and CONTINUOUS. By default, the refresh mode is inferred based on data freshness. Users can explicitly specify the refresh mode for specific business scenarios, which will take precedence over the data freshness inference.
+
+- **CONTINUOUS Mode**: The Flink streaming job incrementally updates the materialized table data. The visibility of this data depends on the behavior of the corresponding Connector, either being immediate or after checkpoint completion.
+- **FULL Mode**: The scheduler periodically triggers a full overwrite of the materialized table data, with the data refresh cycle matching the workflow's scheduling cycle.
+ - The default overwrite behavior is table-level. If there are partition fields and the time partition field format is specified via [partition.fields.#.date-formatter]({{< ref "docs/dev/table/config" >}}#partition-fields-date-formatter), the overwrite is by partition. Only the latest partition is refreshed each time.
+
+## Query Definition
+
+The query definition of a materialized table supports all Flink SQL [Queries]({{< ref "docs/dev/table/sql/queries/overview" >}}). The query results are used to populate the materialized table. In CONTINUOUS mode, the query results are updated to the materialized table continuously, while in FULL mode, the query results overwrite the materialized table each time.
+
+## Schema
+
+The schema definition of a materialized table is the same as the regular table. It can declare primary keys and partition keys. The column names and types of materialized table are automatically inferred from the corresponding query, and users cannot specify them.
diff --git a/docs/content/docs/dev/table/materialized-table/statements.md b/docs/content/docs/dev/table/materialized-table/statements.md
new file mode 100644
index 00000000000..18d90675bb9
--- /dev/null
+++ b/docs/content/docs/dev/table/materialized-table/statements.md
@@ -0,0 +1,342 @@
+---
+title: Statements
+weight: 2
+type: docs
+aliases:
+- /dev/table/materialized-table/statements.html
+---
+<!--
+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.
+-->
+
+# Materialized Table Statements
+
+Flink SQL supports the following Materialized Table statements for now:
+- [CREATE MATERIALIZED TABLE](#create-materialized-table)
+- [Alter MATERIALIZED TABLE](#alter-materialized-table)
+- [DROP MATERIALIZED TABLE](#drop-materialized-table)
+
+# CREATE MATERIALIZED TABLE
+
+```
+CREATE MATERIALIZED TABLE [catalog_name.][db_name.]table_name
+
+[ ([ <table_constraint> ]) ]
+
+[COMMENT table_comment]
+
+[PARTITIONED BY (partition_column_name1, partition_column_name2, ...)]
+
+[WITH (key1=val1, key2=val2, ...)]
+
+FRESHNESS = INTERVAL '<num>' { SECOND | MINUTE | HOUR | DAY }
+
+[REFRESH_MODE = { CONTINUOUS | FULL }]
+
+AS <select_statement>
+
+<table_constraint>:
+ [CONSTRAINT constraint_name] PRIMARY KEY (column_name, ...) NOT ENFORCED
+```
+
+## PRIMARY KEY
+
+`PRIMARY KEY` defines an optional list of columns that uniquely identifies each row within the table. The column as the primary key must be non-null.
+
+## PARTITIONED BY
+
+`PARTITIONED BY` define an optional list of columns to partition the materialized table. A directory is created for each partition if this materialized table is used as a filesystem sink.
+
+**Example:**
+
+```sql
+-- Create a materialized table and specify the partition field as `ds`.
+CREATE MATERIALIZED TABLE my_materialized_table
+ PARTITIONED BY (ds)
+ FRESHNESS = INTERVAL '1' HOUR
+ AS SELECT
+ ds
+ FROM
+ ...
+```
+
+<span class="label label-danger">Note</span>
+- The partition column must be included in the query statement of the materialized table.
+
+## WITH Options
+
+`WITH Options` are used to specify the materialized table properties, including [connector options]({{< ref "docs/connectors/table/" >}}) and [time format option]({{< ref "docs/dev/table/config" >}}#partition-fields-date-formatter) for partition fields.
+
+```sql
+-- Create a materialized table, specify the partition field as 'ds', and the corresponding time format as 'yyyy-MM-dd'
+CREATE MATERIALIZED TABLE my_materialized_table
+ PARTITIONED BY (ds)
+ WITH (
+ 'format' = 'json',
+ 'partition.fields.ds.date-formatter' = 'yyyy-MM-dd'
+ )
+ ...
+```
+
+As shown in the above example, we specified the date-formatter option for the `ds` partition column. During each scheduling, the scheduling time will be converted to the ds partition value. For example, for a scheduling time of `2024-01-01 00:00:00`, only the partition `ds = '2024-01-01'` will be refreshed.
+
+<span class="label label-danger">Note</span>
+- The `partition.fields.#.date-formatter` option only works in full mode.
+- The field in the [partition.fields.#.date-formatter]({{< ref "docs/dev/table/config" >}}#partition-fields-date-formatter) must be a valid string type partition field.
+
+## FRESHNESS
+
+`FRESHNESS` define the data freshness of a materialized table.
+
+**FRESHNESS and Refresh Mode Relationship**
+
+FRESHNESS defines the maximum amount of time that the materialized table’s content should lag behind updates to the base tables. It does two things, firstly it determines the [refresh mode]({{< ref "docs/dev/table/materialized-table/overview" >}}#refresh-mode) of the materialized table through [configuration]({{< ref "docs/dev/table/config" >}}#materialized-table-refresh-mode-freshness-threshold), followed by determines the data refresh frequency to meet the actual data freshness requirements.
+
+**Explanation of FRESHNESS Parameter**
+
+The FRESHNESS parameter range is INTERVAL `'<num>'` { SECOND | MINUTE | HOUR | DAY }. `'<num>'` must be a positive integer, and in FULL mode, `'<num>'` should be a common divisor of the respective time interval unit.
+
+**Examples:**
+(Assuming `materialized-table.refresh-mode.freshness-threshold` is 30 minutes)
+
+```sql
+-- The corresponding refresh pipeline is a streaming job with a checkpoint interval of 1 second
+FRESHNESS = INTERVAL '1' SECOND
+
+-- The corresponding refresh pipeline is a real-time job with a checkpoint interval of 1 minute
+FRESHNESS = INTERVAL '1' MINUTE
+
+-- The corresponding refresh pipeline is a scheduled workflow with a schedule cycle of 1 hour
+FRESHNESS = INTERVAL '1' HOUR
+
+-- The corresponding refresh pipeline is a scheduled workflow with a schedule cycle of 1 day
+FRESHNESS = INTERVAL '1' DAY
+```
+
+**Invalid `FRESHNESS` Examples:**
+
+```sql
+-- Interval is a negative number
+FRESHNESS = INTERVAL '-1' SECOND
+
+-- Interval is 0
+FRESHNESS = INTERVAL '0' SECOND
+
+-- Interval is in months or years
+FRESHNESS = INTERVAL '1' MONTH
+FRESHNESS = INTERVAL '1' YEAR
+
+-- In FULL mode, the interval is not a common divisor of the respective time range
+FRESHNESS = INTERVAL '60' SECOND
+FRESHNESS = INTERVAL '5' HOUR
+```
+
+<span class="label label-danger">Note</span>
+- The materialized table data will be refreshed as closely as possible within the defined freshness but cannot guarantee complete satisfaction.
+- In CONTINUOUS mode, setting a data freshness interval that is too short can impact job performance as it aligns with the checkpoint interval. To optimize checkpoint performance, consider [enabling-changelog]({{< ref "docs/ops/state/state_backends" >}}#incremental-checkpoints).
+- In FULL mode, data freshness must be translated into a cron expression, consequently, only freshness intervals within predefined time spans are presently accommodated, this design ensures alignment with cron's capabilities. Specifically, support for the following freshness:
+ - Second: 30, 15, 10, 5, 2, and 1 second intervals.
+ - Minute: 30, 15, 10, 5, 2, and 1 minute intervals.
+ - Hour: 8, 4, 2, and 1 hour intervals.
+ - Day: 1 day.
+
+## REFRESH_MODE
+
+`REFRESH_MODE` is used to explicitly specify the refresh mode of the materialized table. The specified mode takes precedence over the framework's automatic inference to meet specific scenarios' needs.
+
+**Examples:**
+(Assuming `materialized-table.refresh-mode.freshness-threshold` is 30 minutes)
+
+```sql
+-- The refresh mode of the created materialized table is CONTINUOUS, and the job's checkpoint interval is 1 hour.
+CREATE MATERIALIZED TABLE my_materialized_table
+ REFRESH_MODE = CONTINUOUS
+ FRESHNESS = INTERVAL '1' HOUR
+ AS SELECT
+ ...
+
+-- The refresh mode of the created materialized table is FULL, and the job's schedule cycle is 10 minutes.
+CREATE MATERIALIZED TABLE my_materialized_table
+ REFRESH_MODE = FULL
+ FRESHNESS = INTERVAL '10' MINUTE
+ AS SELECT
+ ...
+```
+
+## AS <select_statement>
+
+This clause is used to define the query for populating materialized view data. The upstream table can be a materialized table, table, or view. The select statement supports all Flink SQL [Queries]({{< ref "docs/dev/table/sql/queries/overview" >}}).
+
+**Example:**
+
+```sql
+CREATE MATERIALIZED TABLE my_materialized_table
+ FRESHNESS = INTERVAL '10' SECOND
+ AS SELECT * FROM kafka_catalog.db1.kafka_table;
+```
+
+## Examples
+
+**(Assuming `materialized-table.refresh-mode.freshness-threshold` is 30 minutes)**
+
+Create a materialized table with a data freshness of 10 seconds and the derived refresh mode is CONTINUOUS:
+
+```sql
+CREATE MATERIALIZED TABLE my_materialized_table_continuous
+ PARTITIONED BY (ds)
+ WITH (
+ 'format' = 'debezium-json',
+ 'partition.fields.ds.date-formatter' = 'yyyy-MM-dd'
+ )
+ FRESHNESS = INTERVAL '10' SECOND
+ AS
+ SELECT
+ k.ds,
+ k.user_id,
+ COUNT(*) AS event_count,
+ SUM(k.amount) AS total_amount,
+ MAX(u.age) AS max_age
+ FROM
+ kafka_catalog.db1.kafka_table k
+ JOIN
+ user_catalog.db1.user_table u
+ ON
+ k.user_id = u.user_id
+ WHERE
+ k.event_type = 'purchase'
+ GROUP BY
+ k.ds, k.user_id
+```
+
+Create a materialized table with a data freshness of 1 hour and the derived refresh mode is FULL:
+
+```sql
+CREATE MATERIALIZED TABLE my_materialized_table_full
+ PARTITIONED BY (ds)
+ WITH (
+ 'format' = 'json',
+ 'partition.fields.ds.date-formatter' = 'yyyy-MM-dd'
+ )
+ FRESHNESS = INTERVAL '1' HOUR
+ AS
+ SELECT
+ p.ds,
+ p.product_id,
+ p.product_name,
+ AVG(s.sale_price) AS avg_sale_price,
+ SUM(s.quantity) AS total_quantity
+ FROM
+ paimon_catalog.db1.product_table p
+ LEFT JOIN
+ paimon_catalog.db1.sales_table s
+ ON
+ p.product_id = s.product_id
+ WHERE
+ p.category = 'electronics'
+ GROUP BY
+ p.ds, p.product_id, p.product_name
+```
+
+## Limitations
+
+- Does not support explicitly specifying columns
+- Does not support modified query statements
+- Does not support using temporary tables, temporary views, or temporary functions in the select query
+
+# ALTER MATERIALIZED TABLE
+
+```
+ALTER MATERIALIZED TABLE [catalog_name.][db_name.]table_name SUSPEND | RESUME [WITH (key1=val1, key2=val2, ...)] | REFRESH [PARTITION partition_spec]
+```
+
+`ALTER MATERIALIZED TABLE` is used to manage materialized tables. This command allows users to suspend and resume refresh pipeline of materialized tables and manually trigger data refreshes.
+
+## SUSPEND
+
+```
+ALTER MATERIALIZED TABLE [catalog_name.][db_name.]table_name SUSPEND
+```
+
+`SUSPEND` is used to pause the background refresh pipeline of the materialized table.
+
+**Example:**
+
+```sql
+-- Specify SAVEPOINT path before pausing
+SET 'execution.checkpointing.savepoint-dir' = 'hdfs://savepoint_path';
+
+-- Suspend the specified materialized table
+ALTER MATERIALIZED TABLE my_materialized_table SUSPEND;
+```
+
+<span class="label label-danger">Note</span>
+- When suspending a table in CONTINUOUS mode, the job will be paused using STOP WITH SAVEPOINT by default. You need to set the SAVEPOINT save path using [parameters]({{< ref "docs/deployment/config" >}}#execution-checkpointing-savepoint-dir).
+
+## RESUME
+
+```
+ALTER MATERIALIZED TABLE [catalog_name.][db_name.]table_name RESUME [WITH (key1=val1, key2=val2, ...)]
+```
+
+`RESUME` is used to resume the refresh pipeline of a materialized table. Materialized table dynamic options can be specified through `WITH options` clause, which only take effect on the current refreshed pipeline and are not persistent.
+
+**Example:**
+
+```sql
+-- Resume the specified materialized table
+ALTER MATERIALIZED TABLE my_materialized_table RESUME;
+
+-- Resume the specified materialized table and specify sink parallelism
+ALTER MATERIALIZED TABLE my_materialized_table RESUME WITH ('sink.parallelism'='10');
+```
+
+## REFRESH
+
+```
+ALTER MATERIALIZED TABLE [catalog_name.][db_name.]table_name REFRESH [PARTITION partition_spec]
+```
+
+`REFRESH` is used to proactively trigger the refresh of the materialized table.
+
+**Example:**
+
+```sql
+-- Refresh the entire table data
+ALTER MATERIALIZED TABLE my_materialized_table REFRESH;
+
+-- Refresh specified partition data
+ALTER MATERIALIZED TABLE my_materialized_table REFRESH PARTITION (ds='2024-06-28');
+```
+
+<span class="label label-danger">Note</span>
+- The REFRESH operation will start a Flink batch job to refresh the materialized table data.
+
+# DROP MATERIALIZED TABLE
+
+```text
+DROP MATERIALIZED TABLE [IF EXISTS] [catalog_name.][database_name.]table_name
+```
+
+When dropping a materialized table, the background refresh pipeline will be deleted first, and then the metadata corresponding to the materialized table will be removed from the Catalog.
+
+**Example:**
+
+```sql
+-- Delete the specified materialized table
+DROP MATERIALIZED TABLE IF EXISTS my_materialized_table;
+```