[feat](iceberg) Support publish_changes for iceberg table

Author: xylaaaaa

docs/lakehouse/catalogs/iceberg-catalog.mdx

@@ -2199,6 +2199,51 @@ EXECUTE set_current_snapshot ("ref" = "v1.0");
 3. If the specified snapshot ID or reference does not exist, the operation will fail
 4. If the current snapshot is already the target snapshot, the operation returns directly without creating a new snapshot
 
+### publish_changes
+
+The `publish_changes` operation is used in the WAP (Write-Audit-Publish) mode to publish a snapshot with the specified `wap.id` as the current table state.  
+It locates the snapshot whose `wap.id` matches the given `wap_id` and cherry-picks it onto the current table, making the staged data visible to all read operations. 
+
+**Syntax:**
+
+```sql
+ALTER TABLE [catalog.][database.]table_name
+EXECUTE publish_changes("wap_id" = "<wap_id>")
+```
+
+**Parameters:**
+
+**Parameters:**
+
+| Parameter Name | Type | Required | Description |
+| -------------- | ---- | -------- | ----------- |
+| `wap_id` | STRING | Yes | The WAP snapshot ID to be published |
+
+**Return Value:**
+
+Executing `publish_changes` returns a result set with the following 2 columns:
+
+| Column Name | Type | Description |
+| ----------- | ---- | ----------- |
+| `previous_snapshot_id` | STRING | The ID of the current snapshot before the publish operation (NULL if none) |
+| `current_snapshot_id` | STRING | The ID of the new snapshot created and set as current after publishing |
+
+**Examples:**
+
+```sql
+-- Publish the snapshot whose WAP ID is test_wap_001
+ALTER TABLE iceberg_db.iceberg_table 
+EXECUTE publish_changes("wap_id" = "test_wap_001");
+```
+
+**Notes:**
+
+1. This operation does not support a WHERE clause, nor PARTITION/PARTITIONS clauses
+2. It is only meaningful for Iceberg tables with write.wap.enabled = true and WAP snapshots generated via wap.id
+3. If no snapshot is found for the specified wap_id, the operation fails and throws an error
+4. After publishing, the new snapshot becomes the current snapshot
+5. If there is no snapshot before publishing, previous_snapshot_id may be NULL
+
 ## Iceberg Table Optimization
 
 ### View Data File Distribution

i18n/zh-CN/docusaurus-plugin-content-docs/version-4.x/lakehouse/catalogs/iceberg-catalog.mdx

@@ -2210,6 +2210,49 @@ EXECUTE set_current_snapshot ("ref" = "v1.0");
 3. 如果指定的快照 ID 或引用不存在，操作会失败
 4. 如果当前快照已经是目标快照，操作会直接返回，不会创建新快照
 
+### publish_changes
+
+`publish_changes` 操作用于在 WAP（Write-Audit-Publish）模式下，将带有指定 `wap.id` 的快照发布为当前表状态。  
+它会找到与给定 `wap_id` 匹配的快照，并通过 *cherrypick* 的方式将其合并到当前表中，使这批暂存数据对所有读取操作可见。 
+
+**语法：**
+
+```sql
+ALTER TABLE [catalog.][database.]table_name
+EXECUTE publish_changes("wap_id" = "<wap_id>")
+```
+
+**参数说明：**
+
+| 参数名称 | 类型 | 必填 | 描述 |
+| -------- | ---- | ---- | ---- |
+| `wap_id` | STRING | 是 | 要发布的 WAP 快照 ID |
+
+**返回值：**
+
+执行 `publish_changes` 操作会返回一个结果集，包含以下 2 列：
+
+| 列名 | 类型 | 描述 |
+| ---- | ---- | ---- |
+| `previous_snapshot_id` | STRING | 发布操作前当前快照的 ID（如果不存在则为 NULL） |
+| `current_snapshot_id` | STRING | 发布操作后新创建并设置为当前快照的 ID |
+
+**示例：**
+
+```sql
+-- 发布 WAP ID 为 test_wap_001 的快照
+ALTER TABLE iceberg_db.iceberg_table 
+EXECUTE publish_changes("wap_id" = "test_wap_001");
+```
+
+**注意事项：**
+
+1. 该操作不支持WHERE条件，也不支持分区（PARTITION/PARTITIONS）子句
+2. 仅对启用了write.wap.enabled = true并通过wap.id产生 WAP 快照的 Iceberg 表有意义
+3. 如果找不到与指定wap_id对应的快照，操作会失败并抛出错误
+4. 发布后，新快照将成为当前快照
+5. 如果在发布前表还没有任何快照，则previous_snapshot_id可能为 NULL
+
 ## Iceberg 表优化
 
 ### 查看数据文件分布

versioned_docs/version-4.x/lakehouse/catalogs/iceberg-catalog.mdx

@@ -2199,6 +2199,51 @@ EXECUTE set_current_snapshot ("ref" = "v1.0");
 3. If the specified snapshot ID or reference does not exist, the operation will fail
 4. If the current snapshot is already the target snapshot, the operation returns directly without creating a new snapshot
 
+### publish_changes
+
+The `publish_changes` operation is used in the WAP (Write-Audit-Publish) mode to publish a snapshot with the specified `wap.id` as the current table state.  
+It locates the snapshot whose `wap.id` matches the given `wap_id` and cherry-picks it onto the current table, making the staged data visible to all read operations. 
+
+**Syntax:**
+
+```sql
+ALTER TABLE [catalog.][database.]table_name
+EXECUTE publish_changes("wap_id" = "<wap_id>")
+```
+
+**Parameters:**
+
+**Parameters:**
+
+| Parameter Name | Type | Required | Description |
+| -------------- | ---- | -------- | ----------- |
+| `wap_id` | STRING | Yes | The WAP snapshot ID to be published |
+
+**Return Value:**
+
+Executing `publish_changes` returns a result set with the following 2 columns:
+
+| Column Name | Type | Description |
+| ----------- | ---- | ----------- |
+| `previous_snapshot_id` | STRING | The ID of the current snapshot before the publish operation (NULL if none) |
+| `current_snapshot_id` | STRING | The ID of the new snapshot created and set as current after publishing |
+
+**Examples:**
+
+```sql
+-- Publish the snapshot whose WAP ID is test_wap_001
+ALTER TABLE iceberg_db.iceberg_table 
+EXECUTE publish_changes("wap_id" = "test_wap_001");
+```
+
+**Notes:**
+
+1. This operation does not support a WHERE clause, nor PARTITION/PARTITIONS clauses
+2. It is only meaningful for Iceberg tables with write.wap.enabled = true and WAP snapshots generated via wap.id
+3. If no snapshot is found for the specified wap_id, the operation fails and throws an error
+4. After publishing, the new snapshot becomes the current snapshot
+5. If there is no snapshot before publishing, previous_snapshot_id may be NULL
+
 ## Iceberg Table Optimization
 
 ### View Data File Distribution