feat(repartition): implement validation logic for repartition_table

WenyXu · WenyXu · commit 516a5169ceec · 2026-01-08T04:16:55.000Z
Signed-off-by: WenyXu &lt;wenymedia@gmail.com&gt;
diff --git a/src/operator/src/error.rs b/src/operator/src/error.rs
@@ -736,6 +736,14 @@ pub enum Error {
         location: Location,
     },
 
+    #[snafu(display("Failed to deserialize partition expression: {}", source))]
+    DeserializePartitionExpr {
+        #[snafu(source)]
+        source: partition::error::Error,
+        #[snafu(implicit)]
+        location: Location,
+    },
+
     #[snafu(display("Invalid configuration value."))]
     InvalidConfigValue {
         source: session::session_config::Error,
@@ -973,7 +981,9 @@ impl ErrorExt for Error {
             Error::BuildDfLogicalPlan { .. }
             | Error::BuildTableMeta { .. }
             | Error::MissingInsertBody { .. } => StatusCode::Internal,
-            Error::ExecuteAdminFunction { .. } | Error::EncodeJson { .. } => StatusCode::Unexpected,
+            Error::ExecuteAdminFunction { .. }
+            | Error::EncodeJson { .. }
+            | Error::DeserializePartitionExpr { .. } => StatusCode::Unexpected,
             Error::ViewNotFound { .. }
             | Error::ViewInfoNotFound { .. }
             | Error::TableNotFound { .. } => StatusCode::TableNotFound,
diff --git a/src/operator/src/statement/ddl.rs b/src/operator/src/statement/ddl.rs
@@ -83,14 +83,14 @@ use table::table_name::TableName;
 use crate::error::{
     self, AlterExprToRequestSnafu, BuildDfLogicalPlanSnafu, CatalogSnafu, ColumnDataTypeSnafu,
     ColumnNotFoundSnafu, ConvertSchemaSnafu, CreateLogicalTablesSnafu, CreateTableInfoSnafu,
-    EmptyDdlExprSnafu, ExternalSnafu, ExtractTableNamesSnafu, FlowNotFoundSnafu,
-    InvalidPartitionRuleSnafu, InvalidPartitionSnafu, InvalidSqlSnafu, InvalidTableNameSnafu,
-    InvalidViewNameSnafu, InvalidViewStmtSnafu, NotSupportedSnafu, PartitionExprToPbSnafu, Result,
-    SchemaInUseSnafu, SchemaNotFoundSnafu, SchemaReadOnlySnafu, SubstraitCodecSnafu,
-    TableAlreadyExistsSnafu, TableMetadataManagerSnafu, TableNotFoundSnafu,
+    DeserializePartitionExprSnafu, EmptyDdlExprSnafu, ExternalSnafu, ExtractTableNamesSnafu,
+    FlowNotFoundSnafu, InvalidPartitionRuleSnafu, InvalidPartitionSnafu, InvalidSqlSnafu,
+    InvalidTableNameSnafu, InvalidViewNameSnafu, InvalidViewStmtSnafu, NotSupportedSnafu,
+    PartitionExprToPbSnafu, Result, SchemaInUseSnafu, SchemaNotFoundSnafu, SchemaReadOnlySnafu,
+    SubstraitCodecSnafu, TableAlreadyExistsSnafu, TableMetadataManagerSnafu, TableNotFoundSnafu,
     UnrecognizedTableOptionSnafu, ViewAlreadyExistsSnafu,
 };
-use crate::expr_helper;
+use crate::expr_helper::{self, RepartitionRequest};
 use crate::statement::StatementExecutor;
 use crate::statement::show::create_partitions_stmt;
 
@@ -1262,17 +1262,175 @@ impl StatementExecutor {
             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 request = expr_helper::to_repartition_request(alter_table, &query_context)?;
+            return self.repartition_table(request, &query_context).await;
         }
 
         let expr = expr_helper::to_alter_table_expr(alter_table, &query_context)?;
         self.alter_table_inner(expr, query_context).await
     }
 
+    #[tracing::instrument(skip_all)]
+    pub async fn repartition_table(
+        &self,
+        request: RepartitionRequest,
+        query_context: &QueryContextRef,
+    ) -> Result<Output> {
+        // Check if the schema is read-only.
+        ensure!(
+            !is_readonly_schema(&request.schema_name),
+            SchemaReadOnlySnafu {
+                name: request.schema_name.clone()
+            }
+        );
+
+        let catalog_name = request.catalog_name;
+        let schema_name = request.schema_name;
+        let table_name = request.table_name;
+        // Get the table from the catalog.
+        let table = self
+            .catalog_manager
+            .table(
+                &catalog_name,
+                &schema_name,
+                &table_name,
+                Some(query_context),
+            )
+            .await
+            .context(CatalogSnafu)?
+            .with_context(|| TableNotFoundSnafu {
+                table_name: format_full_table_name(&catalog_name, &schema_name, &table_name),
+            })?;
+        let table_id = table.table_info().ident.table_id;
+        // Get existing partition expressions from the table route.
+        let (physical_table_id, physical_table_route) = self
+            .table_metadata_manager
+            .table_route_manager()
+            .get_physical_table_route(table_id)
+            .await
+            .context(TableMetadataManagerSnafu)?;
+
+        ensure!(
+            physical_table_id == table_id,
+            NotSupportedSnafu {
+                feat: "REPARTITION on logical tables"
+            }
+        );
+
+        let table_info = table.table_info();
+        // Get partition column names from the table metadata.
+        let existing_partition_columns: Vec<String> =
+            table_info.meta.partition_column_names().cloned().collect();
+        // Repartition requires the table to have partition columns.
+        ensure!(
+            !existing_partition_columns.is_empty(),
+            InvalidPartitionRuleSnafu {
+                reason: format!(
+                    "table {}.{}.{} does not have partition columns, cannot repartition",
+                    catalog_name, schema_name, table_name
+                )
+            }
+        );
+
+        // Build column name to type mapping for partition columns only.
+        let table_schema = &table_info.meta.schema;
+        let column_schemas = table_schema.column_schemas();
+        // Repartition operations involving columns outside the existing partition columns are not supported.
+        // This restriction ensures repartition only applies to current partition columns.
+        let column_name_and_type: HashMap<&String, ConcreteDataType> = existing_partition_columns
+            .iter()
+            .map(|pc| {
+                let column = column_schemas
+                    .iter()
+                    .find(|c| &c.name == pc)
+                    // partition column must exist in schema
+                    .unwrap();
+                (&column.name, column.data_type.clone())
+            })
+            .collect();
+        let timezone = query_context.timezone();
+        // Convert SQL Exprs to PartitionExprs.
+        let from_partition_exprs = request
+            .from_exprs
+            .iter()
+            .map(|expr| convert_one_expr(expr, &column_name_and_type, &timezone))
+            .collect::<Result<Vec<_>>>()?;
+
+        let into_partition_exprs = request
+            .into_exprs
+            .iter()
+            .map(|expr| convert_one_expr(expr, &column_name_and_type, &timezone))
+            .collect::<Result<Vec<_>>>()?;
+
+        // Parse existing partition expressions from region routes.
+        let mut existing_partition_exprs =
+            Vec::with_capacity(physical_table_route.region_routes.len());
+        for route in &physical_table_route.region_routes {
+            let expr_json = route.region.partition_expr();
+            if !expr_json.is_empty() {
+                match PartitionExpr::from_json_str(&expr_json) {
+                    Ok(Some(expr)) => existing_partition_exprs.push(expr),
+                    Ok(None) => {
+                        // Empty
+                    }
+                    Err(e) => {
+                        return Err(e).context(DeserializePartitionExprSnafu);
+                    }
+                }
+            }
+        }
+
+        // Validate that from_partition_exprs are a subset of existing partition exprs.
+        // We compare PartitionExpr directly since it implements Eq.
+        for from_expr in &from_partition_exprs {
+            ensure!(
+                existing_partition_exprs.contains(from_expr),
+                InvalidPartitionRuleSnafu {
+                    reason: format!(
+                        "partition expression '{}' does not exist in table {}.{}.{}",
+                        from_expr, catalog_name, schema_name, table_name
+                    )
+                }
+            );
+        }
+
+        // Build the new partition expressions:
+        // new_exprs = existing_exprs - from_exprs + into_exprs
+        let new_partition_exprs: Vec<PartitionExpr> = existing_partition_exprs
+            .into_iter()
+            .filter(|expr| !from_partition_exprs.contains(expr))
+            .chain(into_partition_exprs.clone().into_iter())
+            .collect();
+        let new_partition_exprs_len = new_partition_exprs.len();
+
+        // Validate the new partition expressions using MultiDimPartitionRule and PartitionChecker.
+        let _ = MultiDimPartitionRule::try_new(
+            existing_partition_columns.clone(),
+            vec![],
+            new_partition_exprs,
+            true,
+        )
+        .context(InvalidPartitionSnafu)?;
+
+        info!(
+            "Repartition table {}.{}.{} (table_id={}) from {:?} to {:?}, new partition count: {}",
+            catalog_name,
+            schema_name,
+            table_name,
+            table_id,
+            from_partition_exprs,
+            into_partition_exprs,
+            new_partition_exprs_len
+        );
+
+        // TODO(weny): Implement the repartition procedure submission.
+        // The repartition procedure infrastructure is not yet fully integrated with the DDL task system.
+        NotSupportedSnafu {
+            feat: "ALTER TABLE REPARTITION",
+        }
+        .fail()
+    }
+
     #[tracing::instrument(skip_all)]
     pub async fn alter_table_inner(
         &self,
diff --git a/tests/cases/standalone/common/alter/repartition_error.result b/tests/cases/standalone/common/alter/repartition_error.result
@@ -0,0 +1,104 @@
+-- Test repartition error cases
+-- Setup: Create a physical table with partitions
+CREATE TABLE repartition_test_table(
+  device_id INT,
+  area STRING,
+  ts TIMESTAMP TIME INDEX,
+  PRIMARY KEY(device_id)
+) PARTITION ON COLUMNS (device_id) (
+  device_id < 100,
+  device_id >= 100 AND device_id < 200,
+  device_id >= 200
+);
+
+Affected Rows: 0
+
+-- Setup: Create a logical table (metric engine)
+CREATE TABLE physical_metric_table(
+  ts TIMESTAMP TIME INDEX,
+  val DOUBLE
+) ENGINE = metric WITH ("physical_metric_table" = "");
+
+Affected Rows: 0
+
+CREATE TABLE logical_metric_table(
+  ts TIMESTAMP TIME INDEX,
+  val DOUBLE,
+  device_id STRING PRIMARY KEY
+) ENGINE = metric WITH ("on_physical_table" = "physical_metric_table");
+
+Affected Rows: 0
+
+-- Test 0: Logical table cannot be repartitioned
+ALTER TABLE logical_metric_table REPARTITION (
+  device_id < '100'
+) INTO (
+  device_id < '50',
+  device_id >= '50' AND device_id < '100'
+);
+
+Error: 1001(Unsupported), Not supported: REPARTITION on logical tables
+
+-- Test 1: New partition rule contains non-partition column (ts is not a partition column)
+ALTER TABLE repartition_test_table REPARTITION (
+  device_id < 100
+) INTO (
+  device_id < 50,
+  device_id >= 50 AND device_id < 100 AND ts < 1000
+);
+
+Error: 1004(InvalidArguments), Cannot find column by name: ts
+
+-- Test 2: From partition expr does not exist in existing partition exprs
+-- device_id < 50 is not in the existing partitions (which are < 100, >= 100 AND < 200, >= 200)
+ALTER TABLE repartition_test_table REPARTITION (
+  device_id < 50
+) INTO (
+  device_id < 25,
+  device_id >= 25 AND device_id < 50
+);
+
+Error: 1004(InvalidArguments), Invalid partition rule: partition expression 'device_id < 50' does not exist in table greptime.public.repartition_test_table
+
+-- Test 3: New partition rule is incomplete (cannot pass checker)
+-- This creates a gap: device_id < 50 and device_id >= 100, missing [50, 100)
+-- The existing partitions are: device_id < 100, device_id >= 100 AND device_id < 200, device_id >= 200
+-- After removing device_id < 100 and adding device_id < 50 and device_id >= 100, we get:
+-- device_id < 50, device_id >= 100 AND device_id < 200, device_id >= 200
+-- This leaves a gap [50, 100)
+ALTER TABLE repartition_test_table REPARTITION (
+  device_id < 100
+) INTO (
+  device_id < 50,
+  device_id >= 100
+);
+
+Error: 1004(InvalidArguments), Checkpoint `device_id=50` is not covered
+
+-- Test 4: New partition rule has overlapping partitions
+-- This creates overlapping ranges: device_id < 100 and device_id >= 50 AND device_id < 150
+-- After removing device_id < 100, we have: device_id >= 100 AND device_id < 200, device_id >= 200
+-- Adding the new ones: device_id < 100, device_id >= 50 AND device_id < 150
+-- This overlaps: [0, 100) and [50, 150) overlap in [50, 100)
+ALTER TABLE repartition_test_table REPARTITION (
+  device_id < 100
+) INTO (
+  device_id < 100,
+  device_id >= 50 AND device_id < 150
+);
+
+Error: 1004(InvalidArguments), Checkpoint `device_id=50` is overlapped
+
+-- Cleanup
+DROP TABLE repartition_test_table;
+
+Affected Rows: 0
+
+DROP TABLE logical_metric_table;
+
+Affected Rows: 0
+
+DROP TABLE physical_metric_table;
+
+Affected Rows: 0
+
diff --git a/tests/cases/standalone/common/alter/repartition_error.sql b/tests/cases/standalone/common/alter/repartition_error.sql
