[docs] Add document for LIST/ADD/DROP PARTITION (#517)

Author: swuferhong

fluss-connectors/fluss-connector-flink/src/main/java/com/alibaba/fluss/connector/flink/catalog/FlinkCatalog.java

@@ -427,6 +427,8 @@ public List<CatalogPartitionSpec> listPartitions(
             ObjectPath objectPath, CatalogPartitionSpec catalogPartitionSpec)
             throws TableNotExistException, TableNotPartitionedException,
                     PartitionSpecInvalidException, CatalogException {
+        // TODO, list partitions by catalogPartitionSpec. Trace by
+        // https://github.com/alibaba/fluss/issues/514
         throw new UnsupportedOperationException();
     }
 

fluss-connectors/fluss-connector-flink/src/test/java/com/alibaba/fluss/connector/flink/catalog/FlinkCatalogITCase.java

@@ -213,6 +213,15 @@ void testPartitionedTable() throws Exception {
         expectedShowPartitionsResult = Arrays.asList("+I[b=2]", "+I[b=3]");
         showPartitionIterator = tEnv.executeSql("show partitions test_partitioned_table").collect();
         assertResultsIgnoreOrder(showPartitionIterator, expectedShowPartitionsResult, true);
+
+        // 4. show partitions with spec.
+        assertThatThrownBy(
+                        () ->
+                                tEnv.executeSql(
+                                                "show partitions test_partitioned_table partition (b=2)")
+                                        .collect())
+                .rootCause()
+                .isInstanceOf(UnsupportedOperationException.class);
     }
 
     @Test

website/docs/engine-flink/ddl.md

@@ -88,13 +88,13 @@ CREATE TABLE my_log_table (
 
 ### Partitioned (PrimaryKey/Log) Table
 
-The following SQL statement creates a Partitioned PrimaryKey Table in Fluss. Note that the partitioned field (`dt` in this case) must be a subset of the primary key (`dt, shop_id, user_id` in this case).
-Currently, Fluss only supports one partitioned field with `STRING` type.
-
 :::note
-Currently, partitioned table must enable auto partition and set auto partition time unit.
+1. Currently, Fluss only supports one partitioned field with `STRING` type
+2. For the Partitioned PrimaryKey Table, the partitioned field (`dt` in this case) must be a subset of the primary key (`dt, shop_id, user_id` in this case)
 :::
 
+The following SQL statement creates a Partitioned PrimaryKey Table in Fluss.
+
 ```sql title="Flink SQL"
 CREATE TABLE my_part_pk_table (
   dt STRING,
@@ -103,17 +103,48 @@ CREATE TABLE my_part_pk_table (
   num_orders INT,
   total_amount INT,
   PRIMARY KEY (dt, shop_id, user_id) NOT ENFORCED
+) PARTITIONED BY (dt);
+```
+
+The following SQL statement creates a Partitioned Log Table in Fluss.
+
+```sql title="Flink SQL"
+CREATE TABLE my_part_log_table (
+  order_id BIGINT,
+  item_id BIGINT,
+  amount INT,
+  address STRING,
+  dt STRING
+) PARTITIONED BY (dt);
+```
+:::note
+After the Partitioned (PrimaryKey/Log) Table is created, you need first manually create the corresponding partition using the [Add Partition](/docs/engine-flink/ddl.md#add-partition) statement
+before you write/read data into this partition.
+:::
+
+#### Auto partitioned (PrimaryKey/Log) table
+
+Fluss also support creat Auto Partitioned (PrimaryKey/Log) Table. The following SQL statement creates an Auto Partitioned PrimaryKey Table in Fluss.
+
+```sql title="Flink SQL"
+CREATE TABLE my_auto_part_pk_table (
+  dt STRING,
+  shop_id BIGINT,
+  user_id BIGINT,
+  num_orders INT,
+  total_amount INT,
+  PRIMARY KEY (dt, shop_id, user_id) NOT ENFORCED
 ) PARTITIONED BY (dt) WITH (
   'bucket.num' = '4',
   'table.auto-partition.enabled' = 'true',
   'table.auto-partition.time-unit' = 'day'
 );
 ```
 
-The following SQL statement creates a Partitioned Log Table in Fluss.
+The following SQL statement creates an Auto Partitioned Log Table in Fluss.
 
 ```sql title="Flink SQL"
-CREATE TABLE my_part_log_table (
+CREATE TABLE my_auto_part_log_table (
   order_id BIGINT,
   item_id BIGINT,
   amount INT,
@@ -126,6 +157,9 @@ CREATE TABLE my_part_log_table (
 );
 ```
 
+For more details about Auto Partitioned (PrimaryKey/Log) Table, refer to [Auto Partitioning Options](/docs/table-design/data-distribution/partitioning/#auto-partitioning-options).
+
+### Options
 
 The supported option in "with" parameters when creating a table are as follows:
 
@@ -172,4 +206,43 @@ DROP TABLE my_table;
 
 This will entirely remove all the data of the table in the Fluss cluster.
 
+## Show Partitions
+
+To show all the partitions of a partitioned table, run:
+```sql title="Flink SQL"
+SHOW PARTITIONS my_part_pk_table;
+```
+
+For more details, refer to the [Flink SHOW PARTITIONS](https://nightlies.apache.org/flink/flink-docs-release-1.20/docs/dev/table/sql/show/#show-partitions) documentation.
+
+:::note
+Currently, we only support show all partitions of a partitioned table, but not support show partitions with the given partition spec.
+:::
+
+## Add Partition
+
+Fluss support manually add partitions to an exists partitioned table by Fluss Catalog. If the specified partition 
+not exists, Fluss will create the partition. If the specified partition already exists, Fluss will ignore the request 
+or throw an exception.
+
+To add partitions, run:
+```sql title="Flink SQL"
+ALTER TABLE my_part_pk_table ADD PARTITION (dt = '2025-03-05');
+```
+
+For more details, refer to the [Flink ALTER TABLE(ADD)](https://nightlies.apache.org/flink/flink-docs-release-1.20/docs/dev/table/sql/alter/#add) documentation.
+
+## Drop Partition
+
+Fluss also support manually drop partitions from an exists partitioned table by Fluss Catalog. If the specified partition 
+not exists, Fluss will ignore the request or throw an exception.
+
+
+To drop partitions, run:
+```sql title="Flink SQL"
+ALTER TABLE my_part_pk_table DROP PARTITION (dt = '2025-03-05');
+```
+
+For more details, refer to the [Flink ALTER TABLE(DROP)](https://nightlies.apache.org/flink/flink-docs-release-1.20/docs/dev/table/sql/alter/#drop) documentation.
+
 

website/docs/table-design/data-distribution/partitioning.md

@@ -7,7 +7,10 @@ sidebar_position: 2
 ## Partitioned Tables
 In Fluss, a **Partitioned Table** organizes data based on one or more partition keys, providing a way to improve query performance and manageability for large datasets. Partitions allow the system to divide data into distinct segments, each corresponding to specific values of the partition keys.
 
-For partitioned tables, Fluss supports auto partitioning creation. Partitions can be automatically created based on the auto partitioning rules configured at the time of table creation, and expired partitions are automatically removed, ensuring data not expanding unlimited.
+For partitioned tables, Fluss not only supports manage partitions by users, like create/drop partitions, but also supports automatic manage partitions.
+   - For manually managing partitions, user can create new partitions or drop exists partitions. Learn how to create or drop partitions please refer to [Add Partition](/docs/engine-flink/ddl.md#add-partition) and [Drop Partition](/docs/engine-flink/ddl.md#drop-partition).
+   - For automatically managing partitions, the partitions will be created based on the auto partitioning rules configured at the time of table creation, and expired partitions are automatically removed, ensuring data not expanding unlimited. See [Auto Partitioning Options](/docs/table-design/data-distribution/partitioning.md#auto-partitioning-options).
+   - Manual management and automated management are orthogonal and can coexist on the same table
 
 ### Key Benefits of Partitioned Tables
 - **Improved Query Performance:** By narrowing down the query scope to specific partitions, the system reads fewer data, reducing query execution time.