feat(parser): ALTER TABLE ... REPARTITION ... (#7082)

* initial impl

Signed-off-by: Ruihang Xia <waynestxia@gmail.com>

* sqlness tests

Signed-off-by: Ruihang Xia <waynestxia@gmail.com>

* tidy up

Signed-off-by: Ruihang Xia <waynestxia@gmail.com>

---------

Signed-off-by: Ruihang Xia <waynestxia@gmail.com>

src/operator/src/expr_helper.rs

@@ -682,6 +682,40 @@ pub fn column_schemas_to_defs(
         .collect()
 }
 
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct RepartitionRequest {
+    pub catalog_name: String,
+    pub schema_name: String,
+    pub table_name: String,
+    pub from_exprs: Vec<Expr>,
+    pub into_exprs: Vec<Expr>,
+}
+
+pub(crate) fn to_repartition_request(
+    alter_table: AlterTable,
+    query_ctx: &QueryContextRef,
+) -> Result<RepartitionRequest> {
+    let (catalog_name, schema_name, table_name) =
+        table_idents_to_full_name(alter_table.table_name(), query_ctx)
+            .map_err(BoxedError::new)
+            .context(ExternalSnafu)?;
+
+    let AlterTableOperation::Repartition { operation } = alter_table.alter_operation else {
+        return InvalidSqlSnafu {
+            err_msg: "expected REPARTITION operation",
+        }
+        .fail();
+    };
+
+    Ok(RepartitionRequest {
+        catalog_name,
+        schema_name,
+        table_name,
+        from_exprs: operation.from_exprs,
+        into_exprs: operation.into_exprs,
+    })
+}
+
 /// Converts a SQL alter table statement into a gRPC alter table expression.
 pub(crate) fn to_alter_table_expr(
     alter_table: AlterTable,
@@ -764,6 +798,12 @@ pub(crate) fn to_alter_table_expr(
         AlterTableOperation::UnsetTableOptions { keys } => {
             AlterTableKind::UnsetTableOptions(UnsetTableOptions { keys })
         }
+        AlterTableOperation::Repartition { .. } => {
+            return NotSupportedSnafu {
+                feat: "ALTER TABLE ... REPARTITION",
+            }
+            .fail();
+        }
         AlterTableOperation::SetIndex { options } => {
             let option = match options {
                 sql::statements::alter::SetIndexOperation::Fulltext {
@@ -1391,6 +1431,50 @@ SELECT max(c1), min(c2) FROM schema_2.table_2;";
         assert!(modify_column_type.target_type_extension.is_none());
     }
 
+    #[test]
+    fn test_to_repartition_request() {
+        let sql = r#"
+ALTER TABLE metrics REPARTITION (
+  device_id < 100
+) INTO (
+  device_id < 100 AND area < 'South',
+  device_id < 100 AND area >= 'South'
+);"#;
+        let stmt =
+            ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default())
+                .unwrap()
+                .pop()
+                .unwrap();
+
+        let Statement::AlterTable(alter_table) = stmt else {
+            unreachable!()
+        };
+
+        let request = to_repartition_request(alter_table, &QueryContext::arc()).unwrap();
+        assert_eq!("greptime", request.catalog_name);
+        assert_eq!("public", request.schema_name);
+        assert_eq!("metrics", request.table_name);
+        assert_eq!(
+            request
+                .from_exprs
+                .into_iter()
+                .map(|x| x.to_string())
+                .collect::<Vec<_>>(),
+            vec!["device_id < 100".to_string()]
+        );
+        assert_eq!(
+            request
+                .into_exprs
+                .into_iter()
+                .map(|x| x.to_string())
+                .collect::<Vec<_>>(),
+            vec![
+                "device_id < 100 AND area < 'South'".to_string(),
+                "device_id < 100 AND area >= 'South'".to_string()
+            ]
+        );
+    }
+
     fn new_test_table_names() -> Vec<TableName> {
         vec![
             TableName {

src/operator/src/statement/ddl.rs

@@ -66,7 +66,7 @@ use snafu::{OptionExt, ResultExt, ensure};
 use sql::parser::{ParseOptions, ParserContext};
 #[cfg(feature = "enterprise")]
 use sql::statements::alter::trigger::AlterTrigger;
-use sql::statements::alter::{AlterDatabase, AlterTable};
+use sql::statements::alter::{AlterDatabase, AlterTable, AlterTableOperation};
 #[cfg(feature = "enterprise")]
 use sql::statements::create::trigger::CreateTrigger;
 use sql::statements::create::{
@@ -87,10 +87,10 @@ use crate::error::{
     ColumnNotFoundSnafu, ConvertSchemaSnafu, CreateLogicalTablesSnafu, CreateTableInfoSnafu,
     EmptyDdlExprSnafu, ExternalSnafu, ExtractTableNamesSnafu, FlowNotFoundSnafu,
     InvalidPartitionRuleSnafu, InvalidPartitionSnafu, InvalidSqlSnafu, InvalidTableNameSnafu,
-    InvalidViewNameSnafu, InvalidViewStmtSnafu, PartitionExprToPbSnafu, Result, SchemaInUseSnafu,
-    SchemaNotFoundSnafu, SchemaReadOnlySnafu, SubstraitCodecSnafu, TableAlreadyExistsSnafu,
-    TableMetadataManagerSnafu, TableNotFoundSnafu, UnrecognizedTableOptionSnafu,
-    ViewAlreadyExistsSnafu,
+    InvalidViewNameSnafu, InvalidViewStmtSnafu, NotSupportedSnafu, PartitionExprToPbSnafu, Result,
+    SchemaInUseSnafu, SchemaNotFoundSnafu, SchemaReadOnlySnafu, SubstraitCodecSnafu,
+    TableAlreadyExistsSnafu, TableMetadataManagerSnafu, TableNotFoundSnafu,
+    UnrecognizedTableOptionSnafu, ViewAlreadyExistsSnafu,
 };
 use crate::expr_helper;
 use crate::statement::StatementExecutor;
@@ -1194,6 +1194,17 @@ impl StatementExecutor {
         alter_table: AlterTable,
         query_context: QueryContextRef,
     ) -> Result<Output> {
+        if matches!(
+            alter_table.alter_operation(),
+            AlterTableOperation::Repartition { .. }
+        ) {
+            let _request = expr_helper::to_repartition_request(alter_table, &query_context)?;
+            return NotSupportedSnafu {
+                feat: "ALTER TABLE REPARTITION",
+            }
+            .fail();
+        }
+
         let expr = expr_helper::to_alter_table_expr(alter_table, &query_context)?;
         self.alter_table_inner(expr, query_context).await
     }