fix: address CR

integration_tests/cases/env/cluster/ddl/partition_table.result

@@ -108,7 +108,7 @@ CREATE TABLE `random_partition_table_t`(
                                     `value` double NOT NULL,
                                     `t` timestamp NOT NULL,
                                     TIMESTAMP KEY(t)
-) PARTITIONS 4 ENGINE = Analytic with (enable_ttl='false', update_mode="APPEND");
+) PARTITION BY RANDOM PARTITIONS 4 ENGINE = Analytic with (enable_ttl='false', update_mode="APPEND");
 
 affected_rows: 0
 
@@ -163,7 +163,7 @@ CREATE TABLE `random_partition_table_t_overwrite`(
                                     `value` double NOT NULL,
                                     `t` timestamp NOT NULL,
                                     TIMESTAMP KEY(t)
-) PARTITIONS 4 ENGINE = Analytic with (enable_ttl='false', update_mode="OVERWRITE");
+) PARTITION BY RANDOM PARTITIONS 4 ENGINE = Analytic with (enable_ttl='false', update_mode="OVERWRITE");
 
-Failed to execute query, err: Server(ServerError { code: 500, msg: "Failed to execute plan, sql: CREATE TABLE `random_partition_table_t_overwrite`( \n                                     `name`string TAG, \n                                     `id` int TAG, \n                                     `value` double NOT NULL, \n                                     `t` timestamp NOT NULL, \n                                     TIMESTAMP KEY(t) \n ) PARTITIONS 4 ENGINE = Analytic with (enable_ttl='false', update_mode=\"OVERWRITE\");. Caused by: Internal error, msg:Failed to execute interpreter, err:Failed to execute create table, err:Failed to create table by table manipulator, err:Failed to create table, msg:invalid parameters to create table, plan:CreateTablePlan { engine: \"Analytic\", if_not_exists: false, table: \"random_partition_table_t_overwrite\", table_schema: Schema { timestamp_index: 1, tsid_index: Some(0), column_schemas: ColumnSchemas { columns: [ColumnSchema { id: 1, name: \"tsid\", data_type: UInt64, is_nullable: false, is_tag: false, is_dictionary: false, comment: \"\", escaped_name: \"tsid\", default_value: None }, ColumnSchema { id: 2, name: \"t\", data_type: Timestamp, is_nullable: false, is_tag: false, is_dictionary: false, comment: \"\", escaped_name: \"t\", default_value: None }, ColumnSchema { id: 3, name: \"name\", data_type: String, is_nullable: true, is_tag: true, is_dictionary: false, comment: \"\", escaped_name: \"name\", default_value: None }, ColumnSchema { id: 4, name: \"id\", data_type: Int32, is_nullable: true, is_tag: true, is_dictionary: false, comment: \"\", escaped_name: \"id\", default_value: None }, ColumnSchema { id: 5, name: \"value\", data_type: Double, is_nullable: false, is_tag: false, is_dictionary: false, comment: \"\", escaped_name: \"value\", default_value: None }] }, version: 1, primary_key_indexes: [0, 1] }, options: {\"enable_ttl\": \"false\", \"update_mode\": \"OVERWRITE\"} }, err:Invalid arguments, table:random_partition_table_t_overwrite, err:Try to create a random partition table in overwrite mode, table:random_partition_table_t_overwrite." })
+Failed to execute query, err: Server(ServerError { code: 500, msg: "Failed to execute plan, sql: CREATE TABLE `random_partition_table_t_overwrite`( \n                                     `name`string TAG, \n                                     `id` int TAG, \n                                     `value` double NOT NULL, \n                                     `t` timestamp NOT NULL, \n                                     TIMESTAMP KEY(t) \n ) PARTITION BY RANDOM PARTITIONS 4 ENGINE = Analytic with (enable_ttl='false', update_mode=\"OVERWRITE\");. Caused by: Internal error, msg:Failed to execute interpreter, err:Failed to execute create table, err:Failed to create table by table manipulator, err:Failed to create table, msg:invalid parameters to create table, plan:CreateTablePlan { engine: \"Analytic\", if_not_exists: false, table: \"random_partition_table_t_overwrite\", table_schema: Schema { timestamp_index: 1, tsid_index: Some(0), column_schemas: ColumnSchemas { columns: [ColumnSchema { id: 1, name: \"tsid\", data_type: UInt64, is_nullable: false, is_tag: false, is_dictionary: false, comment: \"\", escaped_name: \"tsid\", default_value: None }, ColumnSchema { id: 2, name: \"t\", data_type: Timestamp, is_nullable: false, is_tag: false, is_dictionary: false, comment: \"\", escaped_name: \"t\", default_value: None }, ColumnSchema { id: 3, name: \"name\", data_type: String, is_nullable: true, is_tag: true, is_dictionary: false, comment: \"\", escaped_name: \"name\", default_value: None }, ColumnSchema { id: 4, name: \"id\", data_type: Int32, is_nullable: true, is_tag: true, is_dictionary: false, comment: \"\", escaped_name: \"id\", default_value: None }, ColumnSchema { id: 5, name: \"value\", data_type: Double, is_nullable: false, is_tag: false, is_dictionary: false, comment: \"\", escaped_name: \"value\", default_value: None }] }, version: 1, primary_key_indexes: [0, 1] }, options: {\"enable_ttl\": \"false\", \"update_mode\": \"OVERWRITE\"} }, err:Invalid arguments, table:random_partition_table_t_overwrite, err:Try to create a random partition table in overwrite mode, table:random_partition_table_t_overwrite." })
 

integration_tests/cases/env/cluster/ddl/partition_table.sql

@@ -52,7 +52,7 @@ CREATE TABLE `random_partition_table_t`(
                                     `value` double NOT NULL,
                                     `t` timestamp NOT NULL,
                                     TIMESTAMP KEY(t)
-) PARTITIONS 4 ENGINE = Analytic with (enable_ttl='false', update_mode="APPEND");
+) PARTITION BY RANDOM PARTITIONS 4 ENGINE = Analytic with (enable_ttl='false', update_mode="APPEND");
 
 SHOW CREATE TABLE random_partition_table_t;
 
@@ -85,4 +85,4 @@ CREATE TABLE `random_partition_table_t_overwrite`(
                                     `value` double NOT NULL,
                                     `t` timestamp NOT NULL,
                                     TIMESTAMP KEY(t)
-) PARTITIONS 4 ENGINE = Analytic with (enable_ttl='false', update_mode="OVERWRITE");
+) PARTITION BY RANDOM PARTITIONS 4 ENGINE = Analytic with (enable_ttl='false', update_mode="OVERWRITE");

query_frontend/src/parser.rs

@@ -332,14 +332,8 @@ impl<'a> Parser<'a> {
         let table_name = self.parser.parse_object_name()?.into();
         let (columns, constraints) = self.parse_columns()?;
 
-        // Parse the simple partition rule, starting with `Partitions $Num`.
-        let partition = match self.maybe_parse_simple_partition()? {
-            Some(p) => Some(p),
-            None => {
-                // Parse the complex partition rule, starting with `PARTITION BY`
-                self.maybe_parse_complex_partition(Keyword::PARTITION, &columns)?
-            }
-        };
+        // Parse the partition clause, starting with `PARTITION BY ...`
+        let partition = self.maybe_parse_partition(Keyword::PARTITION, &columns)?;
 
         // ENGINE = ...
         let engine = self.parse_table_engine()?;
@@ -568,15 +562,7 @@ impl<'a> Parser<'a> {
         }
     }
 
-    fn maybe_parse_simple_partition(&mut self) -> Result<Option<Partition>> {
-        let partition = self
-            .parse_partition_num()?
-            .map(|partition_num| Partition::Random(RandomPartition { partition_num }));
-
-        Ok(partition)
-    }
-
-    fn maybe_parse_complex_partition(
+    fn maybe_parse_partition(
         &mut self,
         keyword: Keyword,
         columns: &[ColumnDef],
@@ -595,6 +581,10 @@ impl<'a> Parser<'a> {
         &mut self,
         columns: &[ColumnDef],
     ) -> Result<Option<Partition>> {
+        if let Some(key) = self.maybe_parse_and_check_random_partition()? {
+            return Ok(Some(Partition::Random(key)));
+        }
+
         if let Some(key) = self.maybe_parse_and_check_key_partition(columns)? {
             return Ok(Some(Partition::Key(key)));
         }
@@ -605,6 +595,16 @@ impl<'a> Parser<'a> {
         Ok(None)
     }
 
+    fn maybe_parse_and_check_random_partition(&mut self) -> Result<Option<RandomPartition>> {
+        if !self.consume_token("RANDOM") {
+            return Ok(None);
+        }
+
+        // Parse the clause `PARTITIONS ...`.
+        let partition_num = self.parse_partition_num()?.unwrap_or(1);
+        Ok(Some(RandomPartition { partition_num }))
+    }
+
     fn maybe_parse_and_check_hash_partition(
         &mut self,
         columns: &[ColumnDef],

table_engine/src/engine.rs

@@ -313,7 +313,10 @@ pub trait TableEngine: Send + Sync {
     /// Close the engine gracefully.
     async fn close(&self) -> Result<()>;
 
-    /// Validate the request of create table.
+    /// Validate the params used to create a table.
+    ///
+    /// This validation can be used before doing real table creation to avoid
+    /// unnecessary works if the params is invalid.
     async fn validate_create_table(&self, request: &CreateTableParams) -> Result<()>;
 
     /// Create table

table_engine/src/partition/rule/df_adapter/extractor.rs

@@ -36,6 +36,14 @@ pub trait FilterExtractor: Send + Sync + 'static {
     fn extract(&self, filters: &[Expr], columns: &[String]) -> Vec<PartitionFilter>;
 }
 
+pub struct NoopExtractor;
+
+impl FilterExtractor for NoopExtractor {
+    fn extract(&self, _filters: &[Expr], _columns: &[String]) -> Vec<PartitionFilter> {
+        vec![]
+    }
+}
+
 pub struct KeyExtractor;
 
 impl FilterExtractor for KeyExtractor {

table_engine/src/partition/rule/df_adapter/mod.rs

@@ -17,7 +17,7 @@
 use common_types::{row::RowGroup, schema::Schema};
 use datafusion::logical_expr::Expr;
 
-use self::extractor::KeyExtractor;
+use self::extractor::{KeyExtractor, NoopExtractor};
 use crate::partition::{
     rule::{
         df_adapter::extractor::FilterExtractorRef, factory::PartitionRuleFactory, PartitionRuleRef,
@@ -33,7 +33,7 @@ pub struct DfPartitionRuleAdapter {
     rule: PartitionRuleRef,
 
     /// `PartitionFilter` extractor for datafusion `Expr`
-    extractor: Option<FilterExtractorRef>,
+    extractor: FilterExtractorRef,
 }
 
 impl DfPartitionRuleAdapter {
@@ -55,23 +55,20 @@ impl DfPartitionRuleAdapter {
     pub fn locate_partitions_for_read(&self, filters: &[Expr]) -> Result<Vec<usize>> {
         // Extract partition filters from datafusion filters.
         let columns = self.columns();
-        let partition_filters = match &self.extractor {
-            Some(extractor) => extractor.extract(filters, &columns),
-            None => vec![],
-        };
+        let partition_filters = self.extractor.extract(filters, &columns);
 
         // Locate partitions from filters.
         self.rule.locate_partitions_for_read(&partition_filters)
     }
 
-    fn create_extractor(partition_info: &PartitionInfo) -> Result<Option<FilterExtractorRef>> {
+    fn create_extractor(partition_info: &PartitionInfo) -> Result<FilterExtractorRef> {
         match partition_info {
-            PartitionInfo::Key(_) => Ok(Some(Box::new(KeyExtractor))),
+            PartitionInfo::Key(_) => Ok(Box::new(KeyExtractor)),
             PartitionInfo::Hash(_) => BuildPartitionRule {
                 msg: format!("unsupported partition strategy, strategy:{partition_info:?}"),
             }
             .fail(),
-            PartitionInfo::Random(_) => Ok(None),
+            PartitionInfo::Random(_) => Ok(Box::new(NoopExtractor)),
         }
     }
 }