feat: integrate repartition procedure into DdlManager (#7548)

* feat: add repartition procedure factory support to DdlManager

- Introduce RepartitionProcedureFactory trait for creating and registering
  repartition procedures
- Implement DefaultRepartitionProcedureFactory for metasrv with full support
- Implement StandaloneRepartitionProcedureFactory for standalone (unsupported)
- Add procedure loader registration for RepartitionProcedure and
  RepartitionGroupProcedure
- Add helper methods to TableMetadataAllocator for allocator access
- Add error types for repartition procedure operations
- Update DdlManager to accept and use RepartitionProcedureFactoryRef

Signed-off-by: WenyXu <wenymedia@gmail.com>

* feat: integrate repartition procedure into DdlManager

- Add submit_repartition_task() to handle repartition from alter table
- Route Repartition operations in submit_alter_table_task() to repartition factory
- Refactor: rename submit_procedure() to execute_procedure_and_wait()
- Make all DDL operations wait for completion by default
- Add submit_procedure() for fire-and-forget submissions
- Add CreateRepartitionProcedure error type
- Add placeholder Repartition handling in grpc-expr (unsupported)
- Update greptime-proto dependency

Signed-off-by: WenyXu <wenymedia@gmail.com>

* feat: implement ALTER TABLE REPARTITION procedure submission

Signed-off-by: WenyXu <wenymedia@gmail.com>

* refactor(repartition): handle central region in apply staging manifest

- Introduce ApplyStagingManifestInstructions struct to organize instructions
- Add special handling for central region when applying staging manifests
- Transition state from UpdateMetadata to RepartitionEnd after applying staging manifests
- Remove next_state() method in RepartitionStart and inline state transitions
- Improve logging and expression serialization in DDL statement executor
- Move repartition tests from standalone to distributed test suite

Signed-off-by: WenyXu <wenymedia@gmail.com>

* chore: apply suggestions from CR

Signed-off-by: WenyXu <wenymedia@gmail.com>

* chore: update proto

Signed-off-by: WenyXu <wenymedia@gmail.com>

---------

Signed-off-by: WenyXu <wenymedia@gmail.com>

Author: WenyXu

Cargo.lock

@@ -151,7 +151,7 @@ etcd-client = { version = "0.16.1", features = [
 fst = "0.4.7"
 futures = "0.3"
 futures-util = "0.3"
-greptime-proto = { git = "https://github.com/GreptimeTeam/greptime-proto.git", rev = "0e316b86d765e4718d6f0ca77b1ad179f222b822" }
+greptime-proto = { git = "https://github.com/GreptimeTeam/greptime-proto.git", rev = "58aeee49267fb1eafa6f9123f9d0c47dd0f62722" }
 hex = "0.4"
 http = "1"
 humantime = "2.1"

Cargo.toml

@@ -64,8 +64,8 @@ use plugins::frontend::context::{
 use plugins::standalone::context::DdlManagerConfigureContext;
 use servers::tls::{TlsMode, TlsOption, merge_tls_option};
 use snafu::ResultExt;
-use standalone::StandaloneInformationExtension;
 use standalone::options::StandaloneOptions;
+use standalone::{StandaloneInformationExtension, StandaloneRepartitionProcedureFactory};
 use tracing_appender::non_blocking::WorkerGuard;
 
 use crate::error::{OtherSnafu, Result, StartFlownodeSnafu};
@@ -509,8 +509,13 @@ impl StartCommand {
             region_failure_detector_controller: Arc::new(NoopRegionFailureDetectorControl),
         };
 
-        let ddl_manager = DdlManager::try_new(ddl_context, procedure_manager.clone(), true)
-            .context(error::InitDdlManagerSnafu)?;
+        let ddl_manager = DdlManager::try_new(
+            ddl_context,
+            procedure_manager.clone(),
+            Arc::new(StandaloneRepartitionProcedureFactory),
+            true,
+        )
+        .context(error::InitDdlManagerSnafu)?;
 
         let ddl_manager = if let Some(configurator) =
             plugins.get::<DdlManagerConfiguratorRef<DdlManagerConfigureContext>>()

src/cmd/src/standalone.rs

@@ -34,7 +34,7 @@ use table::requests::{
 };
 
 use crate::error::{
-    ColumnNotFoundSnafu, InvalidColumnDefSnafu, InvalidIndexOptionSnafu,
+    self, ColumnNotFoundSnafu, InvalidColumnDefSnafu, InvalidIndexOptionSnafu,
     InvalidSetFulltextOptionRequestSnafu, InvalidSetSkippingIndexOptionRequestSnafu,
     InvalidSetTableOptionRequestSnafu, InvalidUnsetTableOptionRequestSnafu,
     MissingAlterIndexOptionSnafu, MissingFieldSnafu, MissingTableMetaSnafu,
@@ -251,6 +251,10 @@ pub fn alter_expr_to_request(
                 .collect::<Result<Vec<_>>>()?;
             AlterKind::SetDefaults { defaults }
         }
+        Kind::Repartition(_) => error::UnexpectedSnafu {
+            err_msg: "Repartition operation should be handled through DdlManager and not converted to AlterTableRequest",
+        }
+        .fail()?,
     };
 
     let request = AlterTableRequest {

src/common/grpc-expr/src/alter.rs

@@ -161,6 +161,13 @@ pub enum Error {
         #[snafu(implicit)]
         location: Location,
     },
+
+    #[snafu(display("Unexpected: {err_msg}"))]
+    Unexpected {
+        err_msg: String,
+        #[snafu(implicit)]
+        location: Location,
+    },
 }
 
 pub type Result<T> = std::result::Result<T, Error>;
@@ -188,6 +195,7 @@ impl ErrorExt for Error {
             Error::ColumnNotFound { .. } => StatusCode::TableColumnNotFound,
             Error::SqlCommon { source, .. } => source.status_code(),
             Error::MissingTableMeta { .. } => StatusCode::Unexpected,
+            Error::Unexpected { .. } => StatusCode::Unexpected,
         }
     }
 

src/common/grpc-expr/src/error.rs

@@ -22,7 +22,7 @@ use snafu::OptionExt;
 use table::metadata::RawTableInfo;
 
 use crate::ddl::alter_table::AlterTableProcedure;
-use crate::error::{InvalidProtoMsgSnafu, Result};
+use crate::error::{self, InvalidProtoMsgSnafu, Result};
 
 impl AlterTableProcedure {
     /// Makes alter kind proto that all regions can reuse.
@@ -112,6 +112,10 @@ fn create_proto_alter_kind(
         Kind::UnsetIndexes(v) => Ok(Some(alter_request::Kind::UnsetIndexes(v.clone()))),
         Kind::DropDefaults(v) => Ok(Some(alter_request::Kind::DropDefaults(v.clone()))),
         Kind::SetDefaults(v) => Ok(Some(alter_request::Kind::SetDefaults(v.clone()))),
+        Kind::Repartition(_) => error::UnexpectedSnafu {
+            err_msg: "Repartition operation should be handled through DdlManager and not converted to AlterTableRequest",
+        }
+        .fail()?,
     }
 }
 

src/common/meta/src/ddl/alter_table/region_request.rs

@@ -161,7 +161,18 @@ impl TableMetadataAllocator {
         })
     }
 
+    /// Returns the table id allocator.
     pub fn table_id_allocator(&self) -> ResourceIdAllocatorRef {
         self.table_id_allocator.clone()
     }
+
+    /// Returns the wal options allocator.
+    pub fn wal_options_allocator(&self) -> WalOptionsAllocatorRef {
+        self.wal_options_allocator.clone()
+    }
+
+    /// Returns the region routes allocator.
+    pub fn region_routes_allocator(&self) -> RegionRoutesAllocatorRef {
+        self.region_routes_allocator.clone()
+    }
 }