diff --git a/docs/content/flink/procedures.md b/docs/content/flink/procedures.md
index c3018600c99f..430b42c3bd7f 100644
--- a/docs/content/flink/procedures.md
+++ b/docs/content/flink/procedures.md
@@ -59,6 +59,16 @@ All available procedures are listed below.
    <tr>
       <td>compact</td>
       <td>
+         -- Use named argument<br/>
+         CALL [catalog.]sys.compact(
+            `table` => 'table', 
+            partitions => 'partitions', 
+            order_strategy => 'order_strategy', 
+            order_by => 'order_by', 
+            options => 'options', 
+            `where` => 'where', 
+            partition_idle_time => 'partition_idle_time') <br/><br/>
+         -- Use indexed argument<br/>
          CALL [catalog.]sys.compact('table') <br/><br/>
          CALL [catalog.]sys.compact('table', 'partitions') <br/><br/> 
          CALL [catalog.]sys.compact('table', 'partitions', 'order_strategy', 'order_by') <br/><br/>
@@ -86,12 +96,21 @@ All available procedures are listed below.
    <tr>
       <td>compact_database</td>
       <td>
+         -- Use named argument<br/>
+         CALL [catalog.]sys.compact_database(
+            including_databases => 'includingDatabases', 
+            mode => 'mode', 
+            including_tables => 'includingTables', 
+            excluding_tables => 'excludingTables', 
+            table_options => 'tableOptions', 
+            partition_idle_time => 'partitionIdleTime') <br/><br/>
+         -- Use indexed argument<br/>
          CALL [catalog.]sys.compact_database() <br/><br/>
          CALL [catalog.]sys.compact_database('includingDatabases') <br/><br/> 
          CALL [catalog.]sys.compact_database('includingDatabases', 'mode') <br/><br/> 
          CALL [catalog.]sys.compact_database('includingDatabases', 'mode', 'includingTables') <br/><br/> 
          CALL [catalog.]sys.compact_database('includingDatabases', 'mode', 'includingTables', 'excludingTables') <br/><br/>
-         CALL [catalog.]sys.compact_database('includingDatabases', 'mode', 'includingTables', 'excludingTables', 'tableOptions')
+         CALL [catalog.]sys.compact_database('includingDatabases', 'mode', 'includingTables', 'excludingTables', 'tableOptions') <br/><br/>
          CALL [catalog.]sys.compact_database('includingDatabases', 'mode', 'includingTables', 'excludingTables', 'tableOptions', 'partitionIdleTime')
       </td>
       <td>
@@ -106,12 +125,23 @@ All available procedures are listed below.
             <li>partition_idle_time: this is used to do a full compaction for partition which had not received any new data for 'partition_idle_time'. And only these partitions will be compacted.</li>
       </td>
       <td>
-         CALL sys.compact_database('db1|db2', 'combined', 'table_.*', 'ignore', 'sink.parallelism=4')
+         CALL sys.compact_database(
+            including_databases => 'db1|db2', 
+            mode => 'combined', 
+            including_tables => 'table_.*', 
+            excluding_tables => 'ignore', 
+            table_options => 'sink.parallelism=4')
       </td>
    </tr>
    <tr>
       <td>create_tag</td>
       <td>
+         -- Use named argument<br/>
+         -- based on the specified snapshot <br/>
+         CALL [catalog.]sys.create_tag(`table` => 'identifier', tag => 'tagName', snapshot_id => snapshotId) <br/>
+         -- based on the latest snapshot <br/>
+         CALL [catalog.]sys.create_tag(`table` => 'identifier', snapshot_id => 'tagName') <br/><br/>
+         -- Use indexed argument<br/>
          -- based on the specified snapshot <br/>
          CALL [catalog.]sys.create_tag('identifier', 'tagName', snapshotId) <br/>
          -- based on the latest snapshot <br/>
@@ -125,13 +155,16 @@ All available procedures are listed below.
             <li>time_retained: The maximum time retained for newly created tags.</li>
       </td>
       <td>
-         CALL sys.create_tag('default.T', 'my_tag', 10, '1 d')
+         CALL sys.create_tag(`table` => 'default.T', tag => 'my_tag', snapshot_id => cast(10 as bigint), time_retained => '1 d')
       </td>
    </tr>
     <tr>
       <td>create_tag_from_timestamp</td>
       <td>
          -- Create a tag from the first snapshot whose commit-time greater than the specified timestamp. <br/>
+         -- Use named argument<br/>
+         CALL [catalog.]sys.create_tag_from_timestamp(`table` => 'identifier', tag => 'tagName', timestamp => timestamp, time_retained => time_retained) <br/><br/>
+         -- Use indexed argument<br/>
          CALL [catalog.]sys.create_tag_from_timestamp('identifier', 'tagName', timestamp, time_retained)
       </td>
       <td>
@@ -152,6 +185,9 @@ All available procedures are listed below.
       <td>create_tag_from_watermark</td>
       <td>
          -- Create a tag from the first snapshot whose watermark greater than the specified timestamp.<br/>
+         -- Use named argument<br/>
+         CALL [catalog.]sys.create_tag_from_watermark(`table` => 'identifier', tag => 'tagName', watermark => watermark, time_retained => time_retained) <br/><br/>
+         -- Use indexed argument<br/>
          CALL [catalog.]sys.create_tag_from_watermark('identifier', 'tagName', watermark, time_retained)
       </td>
       <td>
@@ -171,6 +207,9 @@ All available procedures are listed below.
    <tr>
       <td>delete_tag</td>
       <td>
+         -- Use named argument<br/>
+         CALL [catalog.]sys.delete_tag(`table` => 'identifier', tag => 'tagName') <br/><br/>
+         -- Use indexed argument<br/>
          CALL [catalog.]sys.delete_tag('identifier', 'tagName')
       </td>
       <td>
@@ -179,17 +218,30 @@ All available procedures are listed below.
             <li>tagName: name of the tag to be deleted. If you specify multiple tags, delimiter is ','.</li>
       </td>
       <td>
-         CALL sys.delete_tag('default.T', 'my_tag')
+         CALL sys.delete_tag(`table` => 'default.T', tag => 'my_tag')
       </td>
    </tr>
    <tr>
       <td>merge_into</td>
       <td>
+         -- for Flink 1.18<br/>
          CALL [catalog].sys.merge_into('identifier','targetAlias',<br/>
             'sourceSqls','sourceTable','mergeCondition',<br/>
             'matchedUpsertCondition','matchedUpsertSetting',<br/>
             'notMatchedInsertCondition','notMatchedInsertValues',<br/>
             'matchedDeleteCondition')<br/><br/>
+         -- for Flink 1.19 and later <br/>
+         CALL [catalog].sys.merge_into(<br/>
+            target_table => 'identifier',<br/>
+            target_alias => 'targetAlias',<br/>
+            source_sqls => 'sourceSqls',<br/>
+            source_table => 'sourceTable',<br/>
+            merge_condition => 'mergeCondition',<br/>
+            matched_upsert_condition => 'matchedUpsertCondition',<br/>
+            matched_upsert_setting => 'matchedUpsertSetting',<br/>
+            not_matched_insert_condition => 'notMatchedInsertCondition',<br/>
+            not_matched_insert_values => 'notMatchedInsertValues',<br/>
+            matched_delete_condition => 'matchedDeleteCondition') <br/><br/>
       </td>
       <td>
          To perform "MERGE INTO" syntax. See <a href="/how-to/writing-tables#merging-into-table">merge_into action</a> for
@@ -201,6 +253,9 @@ All available procedures are listed below.
          -- and if there is no match,<br/> 
          -- insert the order from<br/>
          -- the source table<br/>
+         -- for Flink 1.18<br/>
+         CALL [catalog].sys.merge_into('default.T','','','default.S','T.id=S.order_id','','price=T.price+20','','*','')<br/><br/>
+         -- for Flink 1.19 and later <br/>
          CALL sys.merge_into(<br/>
             target_table => 'default.T',<br/>
             source_table => 'default.S',<br/>
@@ -212,6 +267,9 @@ All available procedures are listed below.
    <tr>
       <td>remove_orphan_files</td>
       <td>
+         -- Use named argument<br/>
+         CALL [catalog.]sys.remove_orphan_files(`table` => 'identifier', older_than => 'olderThan', dry_run => 'dryRun') <br/><br/>
+         -- Use indexed argument<br/>
          CALL [catalog.]sys.remove_orphan_files('identifier')<br/><br/>
          CALL [catalog.]sys.remove_orphan_files('identifier', 'olderThan')<br/><br/>
          CALL [catalog.]sys.remove_orphan_files('identifier', 'olderThan', 'dryRun')<br/><br/>
@@ -226,15 +284,18 @@ All available procedures are listed below.
             <li>dryRun: when true, view only orphan files, don't actually remove files. Default is false.</li>
             <li>parallelism: The maximum number of concurrent deleting files. By default is the number of processors available to the Java virtual machine.</li>
       </td>
-      <td>CALL remove_orphan_files('default.T', '2023-10-31 12:00:00')<br/><br/>
-          CALL remove_orphan_files('default.*', '2023-10-31 12:00:00')<br/><br/>
-          CALL remove_orphan_files('default.T', '2023-10-31 12:00:00', true)<br/><br/>
-          CALL remove_orphan_files('default.T', '2023-10-31 12:00:00', false, '5')
+      <td>CALL remove_orphan_files(`table` => 'default.T', older_than => '2023-10-31 12:00:00')<br/><br/>
+          CALL remove_orphan_files(`table` => 'default.*', older_than => '2023-10-31 12:00:00')<br/><br/>
+          CALL remove_orphan_files(`table` => 'default.T', older_than => '2023-10-31 12:00:00', dry_run => true)<br/><br/>
+          CALL remove_orphan_files(`table` => 'default.T', older_than => '2023-10-31 12:00:00', dry_run => false, parallelism => '5')
       </td>
    </tr>
    <tr>
       <td>reset_consumer</td>
       <td>
+         -- Use named argument<br/>
+         CALL [catalog.]sys.reset_consumer(`table` => 'identifier', consumer_id => 'consumerId', next_snapshot_id => 'nextSnapshotId') <br/><br/>
+         -- Use indexed argument<br/>
          -- reset the new next snapshot id in the consumer<br/>
          CALL [catalog.]sys.reset_consumer('identifier', 'consumerId', nextSnapshotId)<br/><br/>
          -- delete consumer<br/>
@@ -246,11 +307,17 @@ All available procedures are listed below.
             <li>consumerId: consumer to be reset or deleted.</li>
             <li>nextSnapshotId (Long): the new next snapshot id of the consumer.</li>
       </td>
-      <td>CALL sys.reset_consumer('default.T', 'myid', 10)</td>
+      <td>CALL sys.reset_consumer(`table` => 'default.T', consumer_id => 'myid', next_snapshot_id => cast(10 as bigint))</td>
    </tr>
    <tr>
       <td>rollback_to</td>
       <td>
+         -- for Flink 1.18<br/>
+         -- rollback to a snapshot<br/>
+         CALL sys.rollback_to('identifier', snapshotId)<br/><br/>
+         -- rollback to a tag<br/>
+         CALL sys.rollback_to('identifier', 'tagName')<br/><br/>
+         -- for Flink 1.19 and later<br/>
          -- rollback to a snapshot<br/>
          CALL sys.rollback_to(`table` => 'identifier', snapshot_id => snapshotId)<br/><br/>
          -- rollback to a tag<br/>
@@ -262,11 +329,24 @@ All available procedures are listed below.
             <li>snapshotId (Long): id of the snapshot that will roll back to.</li>
             <li>tagName: name of the tag that will roll back to.</li>
       </td>
-      <td>CALL sys.rollback_to(`table` => 'default.T', snapshot_id => 10)</td>
+      <td>
+         -- for Flink 1.18<br/>
+         CALL sys.rollback_to('default.T', 10)
+         -- for Flink 1.19 and later<br/>
+         CALL sys.rollback_to(`table` => 'default.T', snapshot_id => 10)
+      </td>
    </tr>
    <tr>
       <td>expire_snapshots</td>
       <td>
+         -- Use named argument<br/>
+         CALL [catalog.]sys.reset_consumer(<br/>
+            `table` => 'identifier', <br/>
+            retain_max => 'retain_max', <br/>
+            retain_min => 'retain_min', <br/>
+            older_than => 'older_than', <br/>
+            max_deletes => 'max_deletes') <br/><br/>
+         -- Use indexed argument<br/>
          -- for Flink 1.18<br/>
          CALL sys.expire_snapshots(table, retain_max)<br/><br/>
          -- for Flink 1.19 and later<br/>
@@ -329,11 +409,14 @@ All available procedures are listed below.
             <li>databaseName : the target database name.</li>
             <li>tableName: the target table identifier.</li>
       </td>
-      <td>CALL sys.repair('test_db.T')</td>
+      <td>CALL sys.repair(`table` => 'test_db.T')</td>
    </tr>
     <tr>
       <td>rewrite_file_index</td>
       <td>
+         -- Use named argument<br/>
+         CALL sys.rewrite_file_index(&lt`table` => identifier&gt [, &ltpartitions => partitions&gt])<br/><br/>
+         -- Use indexed argument<br/>
          CALL sys.rewrite_file_index(&ltidentifier&gt [, &ltpartitions&gt])<br/><br/>
       </td>
       <td>
@@ -343,13 +426,16 @@ All available procedures are listed below.
       </td>
       <td>
          -- rewrite the file index for the whole table<br/>
-         CALL sys.rewrite_file_index('test_db.T')<br/><br/>
+         CALL sys.rewrite_file_index(`table` => 'test_db.T')<br/><br/>
          -- repair all tables in a specific partition<br/>
-         CALL sys.rewrite_file_index('test_db.T', 'pt=a')<br/><br/>
+         CALL sys.rewrite_file_index(`table` => 'test_db.T', partitions => 'pt=a')<br/><br/>
      </td>
    <tr>
       <td>create_branch</td>
       <td>
+         -- Use named argument<br/>
+         CALL [catalog.]sys.create_branch(`table` => 'identifier', branch => 'branchName', tag => 'tagName')<br/><br/>
+         -- Use indexed argument<br/>
          -- based on the specified tag <br/>
          CALL [catalog.]sys.create_branch('identifier', 'branchName', 'tagName')
          -- create empty branch <br/>
@@ -362,13 +448,16 @@ All available procedures are listed below.
             <li>tagName: name of the tag which the new branch is based on.</li>
       </td>
       <td>
-         CALL sys.create_branch('default.T', 'branch1', 'tag1')<br/><br/>
-         CALL sys.create_branch('default.T', 'branch1')<br/><br/>
+         CALL sys.create_branch(`table` => 'default.T', branch => 'branch1', tag => 'tag1')<br/><br/>
+         CALL sys.create_branch(`table` => 'default.T', branch => 'branch1')<br/><br/>
       </td>
    </tr>
    <tr>
       <td>delete_branch</td>
       <td>
+         -- Use named argument<br/>
+         CALL [catalog.]sys.delete_branch(`table` => 'identifier', branch => 'branchName')<br/><br/>
+         -- Use indexed argument<br/>
          CALL [catalog.]sys.delete_branch('identifier', 'branchName')
       </td>
       <td>
@@ -377,12 +466,15 @@ All available procedures are listed below.
             <li>branchName: name of the branch to be deleted. If you specify multiple branches, delimiter is ','.</li>
       </td>
       <td>
-         CALL sys.delete_branch('default.T', 'branch1')
+         CALL sys.delete_branch(`table` => 'default.T', branch => 'branch1')
       </td>
    </tr>
    <tr>
       <td>fast_forward</td>
       <td>
+         -- Use named argument<br/>
+         CALL [catalog.]sys.fast_forward(`table` => 'identifier', branch => 'branchName')<br/><br/>
+         -- Use indexed argument<br/>
          CALL [catalog.]sys.fast_forward('identifier', 'branchName')
       </td>
       <td>
@@ -391,7 +483,7 @@ All available procedures are listed below.
             <li>branchName: name of the branch to be merged.</li>
       </td>
       <td>
-         CALL sys.fast_forward('default.T', 'branch1')
+         CALL sys.fast_forward(`table` => 'default.T', branch => 'branch1')
       </td>
    </tr>
    </tbody>
diff --git a/docs/content/flink/sql-query.md b/docs/content/flink/sql-query.md
index e58d98d62063..bb26d5d3c66a 100644
--- a/docs/content/flink/sql-query.md
+++ b/docs/content/flink/sql-query.md
@@ -204,6 +204,21 @@ First, you need to stop the streaming task using this consumer ID, and then exec
 
 Run the following command:
 
+{{< tabs "reset_consumer" >}}
+
+{{< tab "Flink SQL" >}}
+
+```sql
+CALL sys.reset_consumer(
+   `table` => 'database_name.table_name', 
+   consumer_id => 'consumer_id', 
+   next_snapshot_id -> <snapshot_id>
+);
+```
+{{< /tab >}}
+
+{{< tab "Flink Action" >}}
+
 ```bash
 <FLINK_HOME>/bin/flink run \
     /path/to/paimon-flink-action-{{< version >}}.jar \
@@ -215,6 +230,9 @@ Run the following command:
     [--next_snapshot <next-snapshot-id>] \
     [--catalog_conf <paimon-catalog-conf> [--catalog_conf <paimon-catalog-conf> ...]]
 ```
+{{< /tab >}}
+
+{{< /tabs >}}
 
 please don't specify --next_snapshot parameter if you want to delete the consumer.
 
diff --git a/docs/content/learn-paimon/understand-files.md b/docs/content/learn-paimon/understand-files.md
index d9e437b0d4e1..b18c259993a7 100644
--- a/docs/content/learn-paimon/understand-files.md
+++ b/docs/content/learn-paimon/understand-files.md
@@ -245,6 +245,22 @@ is needed in order to reduce the number of small files.
 Let's trigger the full-compaction now, and run a dedicated compaction job through `flink run`:
 
 {{< label Batch >}}
+
+{{< tabs "compact" >}}
+
+{{< tab "Flink SQL" >}}
+```sql  
+CALL sys.compact(
+   `table` => 'database_name.table_name', 
+   partitions => 'partition_name', 
+   order_strategy => 'order_strategy',
+   order_by => 'order_by',
+   options => 'paimon_table_dynamic_conf'
+);
+```
+{{< /tab >}}
+
+{{< tab "Flink Action" >}}
 ```bash  
 <FLINK_HOME>/bin/flink run \
     -D execution.runtime-mode=batch \
@@ -257,15 +273,32 @@ Let's trigger the full-compaction now, and run a dedicated compaction job throug
     [--catalog_conf <paimon-catalog-conf> [--catalog_conf <paimon-catalog-conf> ...]] \
     [--table_conf <paimon-table-dynamic-conf> [--table_conf <paimon-table-dynamic-conf>] ...]
 ```
+{{< /tab >}}
+
+{{< /tabs >}}
 
 an example would be (suppose you're already in Flink home)
 
+{{< tabs "compact example" >}}
+
+{{< tab "Flink SQL" >}}
+
+```sql
+CALL sys.compact('T');
+```
+{{< /tab >}}
+
+{{< tab "Flink Action" >}}
+
 ```bash
 ./bin/flink run \
     ./lib/paimon-flink-action-{{< version >}}.jar \
     compact \
     --path file:///tmp/paimon/default.db/T
 ```
+{{< /tab >}}
+
+{{< /tabs >}}
 
 All current table files will be compacted and a new snapshot, namely `snapshot-4`, is
 made and contains the following information:
diff --git a/docs/content/maintenance/dedicated-compaction.md b/docs/content/maintenance/dedicated-compaction.md
index 819ce448bb1d..471bdad22275 100644
--- a/docs/content/maintenance/dedicated-compaction.md
+++ b/docs/content/maintenance/dedicated-compaction.md
@@ -81,6 +81,20 @@ To run a dedicated job for compaction, follow these instructions.
 
 {{< tabs "dedicated-compaction-job" >}}
 
+{{< tab "Flink SQL" >}}
+
+Run the following sql:
+
+```sql
+CALL sys.compact(
+  `table` => 'default.T', 
+  partitions => 'p=0', 
+  options => 'sink.parallelism=4',
+  `where` => 'dt>10 and h<20'
+);
+```
+{{< /tab >}}
+
 {{< tab "Flink Action Jar" >}}
 
 Run the following command to submit a compaction job for the table.
@@ -128,25 +142,6 @@ For more usage of the compact action, see
 
 {{< /tab >}}
 
-{{< tab "Flink SQL" >}}
-
-Run the following sql:
-
-```sql
--- compact table
-CALL sys.compact(`table` => 'default.T');
-
--- compact table with options
-CALL sys.compact(`table` => 'default.T', `options` => 'sink.parallelism=4');
-
--- compact table partition
-CALL sys.compact(`table` => 'default.T', `partitions` => 'p=0');
-
--- compact table partition with filter
-CALL sys.compact(`table` => 'default.T', `where` => 'dt>10 and h<20');
-```
-{{< /tab >}}
-
 {{< /tabs >}}
 
 {{< hint info >}}
@@ -160,6 +155,30 @@ You can run the following command to submit a compaction job for multiple databa
 
 {{< tabs "database-compaction-job" >}}
 
+{{< tab "Flink SQL" >}}
+
+Run the following sql:
+
+```sql
+CALL sys.compact_database(
+  including_databases => 'includingDatabases', 
+  mode => 'mode', 
+  including_tables => 'includingTables', 
+  excluding_tables => 'excludingTables', 
+  table_options => 'tableOptions'
+)
+
+-- example
+CALL sys.compact_database(
+  including_databases => 'db1|db2', 
+  mode => 'combined', 
+  including_tables => 'table_.*', 
+  excluding_tables => 'ignore', 
+  table_options => 'sink.parallelism=4'
+)
+```
+{{< /tab >}}
+
 {{< tab "Flink Action Jar" >}}
 
 ```bash
@@ -243,26 +262,6 @@ For more usage of the compact_database action, see
 
 {{< /tab >}}
 
-{{< tab "Flink SQL" >}}
-
-Run the following sql:
-
-```sql
-CALL sys.compact_database('includingDatabases')
-
-CALL sys.compact_database('includingDatabases', 'mode')
-
-CALL sys.compact_database('includingDatabases', 'mode', 'includingTables')
-
-CALL sys.compact_database('includingDatabases', 'mode', 'includingTables', 'excludingTables')
-
-CALL sys.compact_database('includingDatabases', 'mode', 'includingTables', 'excludingTables', 'tableOptions')
-
--- example
-CALL sys.compact_database('db1|db2', 'combined', 'table_.*', 'ignore', 'sink.parallelism=4')
-```
-{{< /tab >}}
-
 {{< /tabs >}}
 
 ## Sort Compact
@@ -273,6 +272,16 @@ you can trigger a compact with specified column sort to speed up queries.
 
 {{< tabs "sort-compaction-job" >}}
 
+{{< tab "Flink SQL" >}}
+
+Run the following sql:
+
+```sql
+-- sort compact table
+CALL sys.compact(`table` => 'default.T', order_strategy => 'zorder', order_by => 'a,b')
+```
+{{< /tab >}}
+
 {{< tab "Flink Action Jar" >}}
 
 ```bash  
@@ -296,16 +305,6 @@ The sort parallelism is the same as the sink parallelism, you can dynamically sp
 
 {{< /tab >}}
 
-{{< tab "Flink SQL" >}}
-
-Run the following sql:
-
-```sql
--- sort compact table
-CALL sys.compact(`table` => 'default.T', order_strategy => 'zorder', order_by => 'a,b')
-```
-{{< /tab >}}
-
 {{< /tabs >}}
 
 ## Historical Partition Compact
@@ -324,6 +323,17 @@ This feature now is only used in batch mode.
 This is for one table.
 {{< tabs "history-partition-compaction-job for table" >}}
 
+{{< tab "Flink SQL" >}}
+
+Run the following sql:
+
+```sql
+-- history partition compact table
+CALL sys.compact(`table` => 'default.T', 'partition_idle_time' => '1 d')
+```
+
+{{< /tab >}}
+
 {{< tab "Flink Action Jar" >}}
 
 ```bash  
@@ -347,23 +357,41 @@ There are one new configuration in `Historical Partition Compact`
 
 {{< /tab >}}
 
+{{< /tabs >}}
+
+### For Databases
+
+This is for multiple tables in different databases.
+{{< tabs "history-partition-compaction-job for databases" >}}
+
 {{< tab "Flink SQL" >}}
 
 Run the following sql:
 
 ```sql
 -- history partition compact table
-CALL sys.compact(`table` => 'default.T', 'partition_idle_time' => '1 d')
+CALL sys.compact_database(
+  including_databases => 'includingDatabases', 
+  mode => 'mode', 
+  including_tables => 'includingTables',
+  excluding_tables => 'excludingTables',
+  table_options => 'tableOptions',
+  partition_idle_time => 'partition_idle_time'
+);
 ```
 
-{{< /tab >}}
-
-{{< /tabs >}}
+Example: compact historical partitions for tables in database
 
-### For Databases
+```sql
+-- history partition compact table
+CALL sys.compact_database(
+  includingDatabases => 'test_db', 
+  mode => 'combined', 
+  partition_idle_time => '1 d'
+);
+```
 
-This is for multiple tables in different databases.
-{{< tabs "history-partition-compaction-job for databases" >}}
+{{< /tab >}}
 
 {{< tab "Flink Action Jar" >}}
 
@@ -398,22 +426,4 @@ Example: compact historical partitions for tables in database
 
 {{< /tab >}}
 
-{{< tab "Flink SQL" >}}
-
-Run the following sql:
-
-```sql
--- history partition compact table
-CALL sys.compact_database('includingDatabases', 'mode', 'includingTables', 'excludingTables', 'tableOptions', 'partition_idle_time')
-```
-
-Example: compact historical partitions for tables in database
-
-```sql
--- history partition compact table
-CALL sys.compact_database('test_db', 'combined', '', '', '', '1 d')
-```
-
-{{< /tab >}}
-
 {{< /tabs >}}
\ No newline at end of file
diff --git a/docs/content/maintenance/manage-branches.md b/docs/content/maintenance/manage-branches.md
index 5b1fa7ba1d2b..b34a881b0f03 100644
--- a/docs/content/maintenance/manage-branches.md
+++ b/docs/content/maintenance/manage-branches.md
@@ -38,7 +38,7 @@ Paimon supports creating branch from a specific tag or snapshot, or just creatin
 
 {{< tabs "create-branches" >}}
 
-{{< tab "Flink" >}}
+{{< tab "Flink SQL" >}}
 
 Run the following sql:
 
@@ -89,7 +89,7 @@ You can delete branch by its name.
 
 {{< tabs "delete-branches" >}}
 
-{{< tab "Flink" >}}
+{{< tab "Flink SQL" >}}
 
 Run the following sql:
 
@@ -173,7 +173,7 @@ Fast-Forward the custom branch to main will delete all the snapshots, tags and s
 
 {{< tabs "fast_forward" >}}
 
-{{< tab "Flink" >}}
+{{< tab "Flink SQL" >}}
 
 ```sql
 CALL sys.fast_forward('default.T', 'branch1');
diff --git a/docs/content/maintenance/manage-snapshots.md b/docs/content/maintenance/manage-snapshots.md
index d25124008d35..00c0322de8c2 100644
--- a/docs/content/maintenance/manage-snapshots.md
+++ b/docs/content/maintenance/manage-snapshots.md
@@ -232,7 +232,17 @@ Rollback a table to a specific snapshot ID.
 
 {{< tabs "rollback-to" >}}
 
-{{< tab "Flink" >}}
+{{< tab "Flink SQL" >}}
+
+Run the following command:
+
+```sql
+CALL sys.rollback_to(`table` => 'database_name.table_name', snapshot_id => <snasphot-id>);
+```
+
+{{< /tab >}}
+
+{{< tab "Flink Action" >}}
 
 Run the following command:
 
@@ -298,9 +308,9 @@ submit a `remove_orphan_files` job to clean them:
 
 {{< tab "Spark SQL/Flink SQL" >}}
 ```sql
-CALL sys.remove_orphan_files(table => "my_db.my_table", [older_than => "2023-10-31 12:00:00"])
+CALL sys.remove_orphan_files(`table` => "my_db.my_table", [older_than => "2023-10-31 12:00:00"])
 
-CALL sys.remove_orphan_files(table => "my_db.*", [older_than => "2023-10-31 12:00:00"])
+CALL sys.remove_orphan_files(`table` => "my_db.*", [older_than => "2023-10-31 12:00:00"])
 ```
 {{< /tab >}}
 
diff --git a/docs/content/maintenance/manage-tags.md b/docs/content/maintenance/manage-tags.md
index 6c78c0973f9c..5a5ad765d51c 100644
--- a/docs/content/maintenance/manage-tags.md
+++ b/docs/content/maintenance/manage-tags.md
@@ -96,7 +96,19 @@ You can create a tag with given name and snapshot ID.
 
 {{< tabs "create-tag" >}}
 
-{{< tab "Flink" >}}
+{{< tab "Flink SQL" >}}
+
+Run the following command:
+
+```sql
+CALL sys.create_tag(`table` => 'database_name.table_name', tag => 'tag_name', [snapshot_id => <snapshot-id>]);
+```
+
+If `snapshot_id` unset, snapshot_id defaults to the latest.
+
+{{< /tab >}}
+
+{{< tab "Flink Action" >}}
 
  Run the following command:
 
@@ -160,7 +172,17 @@ You can delete a tag by its name.
 
 {{< tabs "delete-tag" >}}
 
-{{< tab "Flink" >}}
+{{< tab "Flink SQL" >}}
+
+Run the following command:
+
+```sql
+CALL sys.delete_tag(`table` => 'database_name.table_name', tag => 'tag_name');
+```
+
+{{< /tab >}}
+
+{{< tab "Flink Action" >}}
 
 Run the following command:
 
@@ -211,7 +233,17 @@ the data will be deleted too).
 
 {{< tabs "rollback-to" >}}
 
-{{< tab "Flink" >}}
+{{< tab "Flink SQL" >}}
+
+Run the following command:
+
+```sql
+CALL sys.rollback_to(`table` => 'database_name.table_name', tag => 'tag_name');
+```
+
+{{< /tab >}}
+
+{{< tab "Flink Action" >}}
 
 Run the following command:
 
diff --git a/docs/content/migration/clone-tables.md b/docs/content/migration/clone-tables.md
index 871baee6f2cb..36355dd85483 100644
--- a/docs/content/migration/clone-tables.md
+++ b/docs/content/migration/clone-tables.md
@@ -33,6 +33,46 @@ To clone a table, run the following command to submit a clone job.
 If the table you clone is not modified at the same time, it is recommended to submit a Flink batch job for better performance.
 However, if you want to clone the table while writing it at the same time, submit a Flink streaming job for automatic failure recovery.
 
+{{< tabs "clone" >}}
+
+{{< tab "Flink SQL" >}}
+
+```sql
+CALL sys.clone(
+    warehouse => 'source_warehouse_path`,
+    [`database` => 'source_database_name',]
+    [`table` => 'source_table_name',] 
+    target_warehouse => 'target_warehouse_path`,
+    [target_database => 'target_database_name',]
+    [target_table => 'target_table_name',]
+    [parallelism => <parallelism>]
+);
+```
+
+{{< hint info >}}
+1. If `database` is not specified, all tables in all databases of the specified warehouse will be cloned.
+2. If `table` is not specified, all tables of the specified database will be cloned.
+   {{< /hint >}}
+
+Example: Clone `test_db.test_table` from source warehouse to target warehouse.
+
+```sql
+CALL sys.clone(
+    warehouse => 's3:///path/to/warehouse_source`,
+    `database` => 'test_db',
+    `table` => 'test_table', 
+    catalog_conf => 's3.endpoint=https://****.com;s3.access-key=*****;s3.secret-key=*****',
+    target_warehouse => 's3:///path/to/warehouse_target',
+    target_database => 'test_db',
+    target_table => 'test_table',
+    target_catalog_conf => 's3.endpoint=https://****.com;s3.access-key=*****;s3.secret-key=*****'
+);
+```
+
+{{< /tab >}}
+
+{{< tab "Flink Action" >}}
+
 ```bash
 <FLINK_HOME>/bin/flink run \
     /path/to/paimon-flink-action-{{< version >}}.jar \
@@ -80,3 +120,7 @@ For more usage of the clone action, see
     /path/to/paimon-flink-action-{{< version >}}.jar \
     clone --help
 ```
+
+{{< /tab >}}
+
+{{< /tabs >}}
diff --git a/docs/content/migration/migration-from-hive.md b/docs/content/migration/migration-from-hive.md
index 8327e8a928ef..a40079fed84f 100644
--- a/docs/content/migration/migration-from-hive.md
+++ b/docs/content/migration/migration-from-hive.md
@@ -56,14 +56,18 @@ CREATE CATALOG PAIMON WITH ('type'='paimon', 'metastore' = 'hive', 'uri' = 'thri
 
 USE CATALOG PAIMON;
 
-CALL sys.migrate_table('hive', 'default.hivetable', 'file.format=orc');
+CALL sys.migrate_table(connector => 'hive', source_table => 'default.hivetable', options => 'file.format=orc');
 ```
 After invoke, "hivetable" will totally convert to paimon format. Writing and reading the table by old "hive way" will fail.
 We can add our table properties while importing by sys.migrate_table('<database>.<tablename>', '<tableproperties>').
 <tableproperties> here should be separated by ",".  For example:
 
 ```sql
-CALL sys.migrate_table('hive', 'my_db.wait_to_upgrate', 'file.format=orc,read.batch-size=2096,write-only=true')
+CALL sys.migrate_table(
+    connector => 'hive', 
+    source_table => 'my_db.wait_to_upgrate', 
+    options => 'file.format=orc,read.batch-size=2096,write-only=true'
+);
 ```
 
 If your flink version is below 1.17, you can use flink action to achieve this:
@@ -101,14 +105,18 @@ CREATE CATALOG PAIMON WITH ('type'='paimon', 'metastore' = 'hive', 'uri' = 'thri
 
 USE CATALOG PAIMON;
 
-CALL sys.migrate_database('hive', 'default', 'file.format=orc');
+CALL sys.migrate_database(connector => 'hive', source_database => 'default', options => 'file.format=orc');
 ```
 After invoke, all tables in "default" database will totally convert to paimon format. Writing and reading the table by old "hive way" will fail.
 We can add our table properties while importing by sys.migrate_database('<database>', '<tableproperties>').
 <tableproperties> here should be separated by ",".  For example:
 
 ```sql
-CALL sys.migrate_database('hive', 'my_db', 'file.format=orc,read.batch-size=2096,write-only=true')
+CALL sys.migrate_database(
+    connector => 'hive', 
+    source_database => 'my_db', 
+    options => 'file.format=orc,read.batch-size=2096,write-only=true'
+);
 ```
 
 If your flink version is below 1.17, you can use flink action to achieve this:
@@ -146,7 +154,7 @@ CREATE CATALOG PAIMON WITH ('type'='paimon', 'metastore' = 'hive', 'uri' = 'thri
 
 USE CATALOG PAIMON;
 
-CALL sys.migrate_file('hive', 'default.hivetable', 'default.paimontable');
+CALL sys.migrate_file(connector => 'hive', source_table => 'default.hivetable', target_table => 'default.paimontable');
 ```
 After invoke, "hivetable" will disappear. And all files will be moved and renamed to paimon directory. "paimontable" here must have the same
 partition keys with "hivetable", and "paimontable" should be in unaware-bucket mode.