fix(meta): validate persisted partition bounds before altering regions

A partition bound is persisted with the type of the column at the time
it was written. When an ALTER TABLE changes the type of a column
referenced by a partition expression (widening the time index unit is
the common case), each region rewrites the bound literal to the new
type while building its scan predicate. A millisecond bound like year
3000 fits milliseconds but overflows nanoseconds, so every scan of the
region fails once the table schema is updated -- and some regions may
already be altered when the failure surfaces.

Validate every region's persisted expression on prepare, before any
region is modified: parse the persisted JSON (mirrored types, since
common-meta cannot depend on the partition crate which depends on it),
replay the same value -> ScalarValue conversion and checked cast the
region performs, and reject the ALTER naming the offending region when
a bound cannot be represented with the new column type. Incompatible
changes are rejected atomically; compatible widening proceeds unchanged.

Adds meta-level tests (multi-region rejection naming the second region,
compatible widening completing) and an end-to-end sqlness case.

Fixes #9454

Signed-off-by: fengyue-xve <1759519137@qq.com>

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

@@ -14,6 +14,7 @@
 
 mod executor;
 mod metadata;
+mod partition_compat;
 mod region_request;
 
 use std::collections::HashSet;
@@ -147,6 +148,10 @@ impl AlterTableProcedure {
         )?;
         self.new_table_info = Some(new_table_info);
 
+        // Reject schema changes that the persisted partition expressions of some
+        // region cannot survive, before any region is altered.
+        self.validate_region_partition_exprs().await?;
+
         // Safety: Checked in `AlterTableProcedure::new`.
         let alter_kind = self.data.task.alter_table.kind.as_ref().unwrap();
         if sets_skip_wal(alter_kind) {
@@ -241,6 +246,44 @@ impl AlterTableProcedure {
         Ok(Status::executing(true))
     }
 
+    /// Validates the persisted partition expressions of every region against the
+    /// schema proposed by this ALTER TABLE, so an ALTER that would break a region's
+    /// partition predicate is rejected before any region is modified.
+    async fn validate_region_partition_exprs(&self) -> Result<()> {
+        // Safety: filled in `fill_table_info`.
+        let old_schema = &self
+            .data
+            .table_info_value
+            .as_ref()
+            .unwrap()
+            .table_info
+            .meta
+            .schema;
+        // Safety: set right after the new table info is built in `on_prepare`.
+        let new_schema = &self.new_table_info.as_ref().unwrap().meta.schema;
+        let changed_columns = partition_compat::changed_column_types(old_schema, new_schema);
+        if changed_columns.is_empty() {
+            return Ok(());
+        }
+
+        let (physical_table_id, physical_table_route) = self
+            .context
+            .table_metadata_manager
+            .table_route_manager()
+            .get_physical_table_route(self.data.table_id())
+            .await?;
+        if physical_table_id != self.data.table_id() {
+            // Partition expressions are persisted on the regions of the physical
+            // table; the altered schema here belongs to a logical table.
+            return Ok(());
+        }
+
+        partition_compat::validate_region_partition_bounds(
+            &changed_columns,
+            &physical_table_route.region_routes,
+        )
+    }
+
     fn table_poison_key(&self) -> PoisonKey {
         table_poison_key(self.data.table_id())
     }

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

@@ -0,0 +1,415 @@
+// Copyright 2023 Greptime Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Preflight validation of persisted partition expressions against an ALTER TABLE schema change.
+//!
+//! Persisted partition bounds keep the types they were written with. When a column
+//! referenced by a partition expression changes its type, the region rewrites the bound
+//! literal to the new column type while building the partition predicate on scan. If a
+//! bound cannot be represented with the new type (a classic example is a millisecond
+//! timestamp bound that overflows when the time index column is widened to nanoseconds),
+//! every scan of that region fails at that point. Such an ALTER must be rejected before
+//! it is submitted to any region, otherwise already-altered regions keep failing while
+//! the rest of the table is untouched.
+//!
+//! The `partition` crate depends on `common-meta`, so this crate cannot depend on
+//! `partition` back and cannot reuse [`partition::expr::PartitionExpr`]. The `Persisted*`
+//! types below mirror the JSON representation of `partition::partition::PartitionBound` and
+//! `partition::expr::PartitionExpr` and must be kept in sync with `partition/src/partition.rs`
+//! and `partition/src/expr.rs`. The compatibility check itself deliberately repeats the
+//! same conversion (value -> [`ScalarValue`]) and the same checked cast
+//! ([`ScalarValue::cast_to`]) that the region performs when it coerces the persisted bound
+//! to the new column type, so a bound that would fail on scan is rejected here as well.
+
+use std::collections::HashMap;
+
+use common_telemetry::warn;
+use datafusion_common::ScalarValue;
+use datatypes::data_type::{ConcreteDataType, DataType};
+use datatypes::schema::Schema;
+use datatypes::value::{
+    Value, duration_to_scalar_value, time_to_scalar_value, timestamp_to_scalar_value,
+};
+use serde::Deserialize;
+use store_api::storage::RegionId;
+
+use crate::error::{PartitionExprIncompatibleSnafu, Result};
+use crate::rpc::router::RegionRoute;
+
+/// Mirrors `partition::partition::PartitionBound`.
+#[derive(Debug, Deserialize)]
+enum PersistedPartitionBound {
+    /// Deprecated since 0.9.0. Only deserialized to keep unknown old formats parseable.
+    #[allow(dead_code)]
+    Value(Value),
+    /// Deprecated since 0.15.0.
+    MaxValue,
+    Expr(PersistedPartitionExpr),
+}
+
+/// Mirrors `partition::expr::PartitionExpr`.
+#[derive(Debug, Deserialize)]
+struct PersistedPartitionExpr {
+    lhs: Box<PersistedOperand>,
+    op: PersistedOp,
+    rhs: Box<PersistedOperand>,
+}
+
+/// Mirrors `partition::expr::Operand`.
+#[derive(Debug, Deserialize)]
+enum PersistedOperand {
+    Column(String),
+    Value(Value),
+    Expr(PersistedPartitionExpr),
+}
+
+/// Mirrors `partition::expr::RestrictedOp`.
+#[derive(Debug, Deserialize)]
+enum PersistedOp {
+    Eq,
+    NotEq,
+    Lt,
+    LtEq,
+    Gt,
+    GtEq,
+    And,
+    Or,
+}
+
+/// Returns the columns whose data type differs between `old_schema` and `new_schema`,
+/// mapped to the new type. Columns only present in one of the schemas (added or dropped)
+/// are not included; dropping a column referenced by a partition expression is already
+/// rejected by the table metadata validation.
+pub(crate) fn changed_column_types(
+    old_schema: &Schema,
+    new_schema: &Schema,
+) -> HashMap<String, ConcreteDataType> {
+    let mut changed = HashMap::new();
+    for new_column in new_schema.column_schemas() {
+        let Some(old_column) = old_schema.column_schema_by_name(new_column.name()) else {
+            continue;
+        };
+        if old_column.data_type() != new_column.data_type() {
+            changed.insert(
+                new_column.name().to_string(),
+                new_column.data_type().clone(),
+            );
+        }
+    }
+
+    changed
+}
+
+/// Validates that the persisted partition bounds of every region are still representable
+/// with the types of `changed_columns`. Returns an error naming the first offending
+/// region without modifying anything.
+pub(crate) fn validate_region_partition_bounds(
+    changed_columns: &HashMap<String, ConcreteDataType>,
+    region_routes: &[RegionRoute],
+) -> Result<()> {
+    if changed_columns.is_empty() {
+        return Ok(());
+    }
+
+    for route in region_routes {
+        let expr_json = route.region.partition_expr();
+        if expr_json.is_empty() {
+            continue;
+        }
+
+        let bound: PersistedPartitionBound = match serde_json::from_str(&expr_json) {
+            Ok(bound) => bound,
+            Err(err) => {
+                // A malformed persisted expression already fails every scan of this
+                // region. It is unrelated to the schema change, so don't block the ALTER.
+                warn!(
+                    "Skipping partition expression validation for region {}, \
+                     the persisted expression is not parseable: {}, expr: {}",
+                    route.region.id, err, expr_json
+                );
+                continue;
+            }
+        };
+        let PersistedPartitionBound::Expr(expr) = bound else {
+            // Deprecated bound formats don't carry expressions and are not used
+            // when the region builds its partition predicate.
+            continue;
+        };
+
+        validate_expr(&expr, changed_columns, route.region.id)?;
+    }
+
+    Ok(())
+}
+
+fn validate_expr(
+    expr: &PersistedPartitionExpr,
+    changed_columns: &HashMap<String, ConcreteDataType>,
+    region_id: RegionId,
+) -> Result<()> {
+    match expr.op {
+        PersistedOp::And | PersistedOp::Or => {
+            validate_operand(&expr.lhs, changed_columns, region_id)?;
+            validate_operand(&expr.rhs, changed_columns, region_id)?;
+        }
+        PersistedOp::Eq
+        | PersistedOp::NotEq
+        | PersistedOp::Lt
+        | PersistedOp::LtEq
+        | PersistedOp::Gt
+        | PersistedOp::GtEq => {
+            // Comparison between a column and a bound literal in either order.
+            validate_comparison(&expr.lhs, &expr.rhs, changed_columns, region_id)?;
+            validate_comparison(&expr.rhs, &expr.lhs, changed_columns, region_id)?;
+        }
+    }
+
+    Ok(())
+}
+
+fn validate_operand(
+    operand: &PersistedOperand,
+    changed_columns: &HashMap<String, ConcreteDataType>,
+    region_id: RegionId,
+) -> Result<()> {
+    match operand {
+        PersistedOperand::Expr(expr) => validate_expr(expr, changed_columns, region_id),
+        PersistedOperand::Column(_) | PersistedOperand::Value(_) => Ok(()),
+    }
+}
+
+fn validate_comparison(
+    operand: &PersistedOperand,
+    other: &PersistedOperand,
+    changed_columns: &HashMap<String, ConcreteDataType>,
+    region_id: RegionId,
+) -> Result<()> {
+    let (PersistedOperand::Column(column), PersistedOperand::Value(value)) = (operand, other)
+    else {
+        return Ok(());
+    };
+    let Some(target_type) = changed_columns.get(column) else {
+        return Ok(());
+    };
+    // Values not representable as a partition bound (lists, structs, json) are
+    // rejected by the partition parser and cannot appear in a persisted expression.
+    let Some(scalar) = value_to_scalar_value(value) else {
+        return Ok(());
+    };
+
+    let target_arrow_type = target_type.as_arrow_type();
+    if scalar.data_type() == target_arrow_type {
+        return Ok(());
+    }
+    if let Err(error) = scalar.cast_to(&target_arrow_type) {
+        warn!(
+            error;
+            "Persisted partition bound is not representable with the altered column type";
+            "region_id" => %region_id,
+            "column" => column,
+            "target_type" => target_type.to_string(),
+        );
+        return PartitionExprIncompatibleSnafu {
+            region_id,
+            column: column.clone(),
+            target_type: target_type.to_string(),
+        }
+        .fail();
+    }
+
+    Ok(())
+}
+
+/// Converts a persisted bound value to a [`ScalarValue`], mirroring
+/// `partition::expr::Operand::try_as_logical_expr`. Returns `None` for values that
+/// cannot appear in a persisted partition expression.
+fn value_to_scalar_value(value: &Value) -> Option<ScalarValue> {
+    Some(match value {
+        Value::Boolean(v) => ScalarValue::Boolean(Some(*v)),
+        Value::UInt8(v) => ScalarValue::UInt8(Some(*v)),
+        Value::UInt16(v) => ScalarValue::UInt16(Some(*v)),
+        Value::UInt32(v) => ScalarValue::UInt32(Some(*v)),
+        Value::UInt64(v) => ScalarValue::UInt64(Some(*v)),
+        Value::Int8(v) => ScalarValue::Int8(Some(*v)),
+        Value::Int16(v) => ScalarValue::Int16(Some(*v)),
+        Value::Int32(v) => ScalarValue::Int32(Some(*v)),
+        Value::Int64(v) => ScalarValue::Int64(Some(*v)),
+        Value::Float32(v) => ScalarValue::Float32(Some(v.0)),
+        Value::Float64(v) => ScalarValue::Float64(Some(v.0)),
+        Value::String(v) => ScalarValue::Utf8(Some(v.as_utf8().to_string())),
+        Value::Binary(v) => ScalarValue::Binary(Some(v.to_vec())),
+        Value::Date(v) => ScalarValue::Date32(Some(v.val())),
+        Value::Null => ScalarValue::Null,
+        Value::Timestamp(t) => timestamp_to_scalar_value(t.unit(), Some(t.value())),
+        Value::Time(t) => time_to_scalar_value(*t.unit(), Some(t.value())).ok()?,
+        Value::IntervalYearMonth(v) => ScalarValue::IntervalYearMonth(Some(v.to_i32())),
+        Value::IntervalDayTime(v) => ScalarValue::IntervalDayTime(Some((*v).into())),
+        Value::IntervalMonthDayNano(v) => ScalarValue::IntervalMonthDayNano(Some((*v).into())),
+        Value::Duration(d) => duration_to_scalar_value(d.unit(), Some(d.value())),
+        Value::Decimal128(d) => {
+            let (v, p, s) = d.to_scalar_value();
+            ScalarValue::Decimal128(v, p, s)
+        }
+        Value::List(_) | Value::Struct(_) | Value::Json(_) => return None,
+    })
+}
+
+#[cfg(test)]
+mod tests {
+    use datatypes::schema::ColumnSchema;
+    use store_api::storage::RegionId;
+
+    use super::*;
+    use crate::rpc::router::Region;
+
+    fn schema(ts_type: ConcreteDataType) -> Schema {
+        Schema::new(vec![
+            ColumnSchema::new("ts", ts_type, false),
+            ColumnSchema::new("host", ConcreteDataType::string_datatype(), true),
+        ])
+    }
+
+    fn ts_bound(millis: i64) -> String {
+        format!(
+            r#"{{"Expr":{{"lhs":{{"Column":"ts"}},"op":"Lt","rhs":{{"Value":{{"Timestamp":{{"value":{millis},"unit":"Millisecond"}}}}}}}}}}"#
+        )
+    }
+
+    fn routes(exprs: &[&str]) -> Vec<RegionRoute> {
+        exprs
+            .iter()
+            .enumerate()
+            .map(|(index, expr)| RegionRoute {
+                region: Region {
+                    id: RegionId::new(1024, index as u32 + 1),
+                    partition_expr: expr.to_string(),
+                    ..Default::default()
+                },
+                ..Default::default()
+            })
+            .collect()
+    }
+
+    fn changed(column: &str, data_type: ConcreteDataType) -> HashMap<String, ConcreteDataType> {
+        HashMap::from([(column.to_string(), data_type)])
+    }
+
+    #[test]
+    fn test_changed_column_types() {
+        let old = schema(ConcreteDataType::timestamp_millisecond_datatype());
+        let new = schema(ConcreteDataType::timestamp_nanosecond_datatype());
+        let changed = changed_column_types(&old, &new);
+        assert_eq!(changed.len(), 1);
+        assert_eq!(
+            changed["ts"],
+            ConcreteDataType::timestamp_nanosecond_datatype()
+        );
+
+        // Added and dropped columns are not type changes.
+        let mut new_columns = old.column_schemas().to_vec();
+        new_columns.push(ColumnSchema::new(
+            "extra",
+            ConcreteDataType::int64_datatype(),
+            true,
+        ));
+        let new = Schema::new(new_columns);
+        assert!(changed_column_types(&old, &new).is_empty());
+    }
+
+    #[test]
+    fn test_overflowing_bound_is_rejected() {
+        // Year 3000 fits milliseconds but overflows nanoseconds.
+        let routes = routes(&[&ts_bound(32_503_680_000_000)]);
+        let err = validate_region_partition_bounds(
+            &changed("ts", ConcreteDataType::timestamp_nanosecond_datatype()),
+            &routes,
+        )
+        .unwrap_err();
+        assert!(
+            err.to_string().contains("region 4398046511105"),
+            "unexpected error: {err}"
+        );
+        assert!(err.to_string().contains("'ts'"), "unexpected error: {err}");
+    }
+
+    #[test]
+    fn test_representable_bound_is_accepted() {
+        // 2024-01-01T00:00:00Z fits both microseconds and nanoseconds.
+        let routes = routes(&[&ts_bound(1_704_067_200_000)]);
+        validate_region_partition_bounds(
+            &changed("ts", ConcreteDataType::timestamp_nanosecond_datatype()),
+            &routes,
+        )
+        .unwrap();
+        validate_region_partition_bounds(
+            &changed("ts", ConcreteDataType::timestamp_microsecond_datatype()),
+            &routes,
+        )
+        .unwrap();
+    }
+
+    #[test]
+    fn test_later_region_rejection_names_the_region() {
+        let routes = routes(&[&ts_bound(1_704_067_200_000), &ts_bound(32_503_680_000_000)]);
+        let err = validate_region_partition_bounds(
+            &changed("ts", ConcreteDataType::timestamp_nanosecond_datatype()),
+            &routes,
+        )
+        .unwrap_err();
+        assert!(
+            err.to_string().contains("region 4398046511106"),
+            "unexpected error: {err}"
+        );
+    }
+
+    #[test]
+    fn test_unrelated_and_unchanged_columns_are_skipped() {
+        let other_column =
+            r#"{"Expr":{"lhs":{"Column":"host"},"op":"Eq","rhs":{"Value":{"String":"a"}}}}"#;
+        // `host` changed, but this bound is a string that any host type accepts.
+        validate_region_partition_bounds(
+            &changed("host", ConcreteDataType::string_datatype()),
+            &routes(&[other_column]),
+        )
+        .unwrap();
+        // `ts` bound with an unrelated column changed.
+        validate_region_partition_bounds(
+            &changed("host", ConcreteDataType::string_datatype()),
+            &routes(&[&ts_bound(32_503_680_000_000)]),
+        )
+        .unwrap();
+    }
+
+    #[test]
+    fn test_nested_expression_is_validated() {
+        let expr = format!(
+            r#"{{"Expr":{{"lhs":{{"Expr":{{"lhs":{{"Column":"ts"}},"op":"GtEq","rhs":{{"Value":{{"Timestamp":{{"value":0,"unit":"Millisecond"}}}}}}}}}},"op":"And","rhs":{{"Expr":{{"lhs":{{"Column":"ts"}},"op":"Lt","rhs":{{"Value":{{"Timestamp":{{"value":32503680000000,"unit":"Millisecond"}}}}}}}}}}}}}}"#
+        );
+        let err = validate_region_partition_bounds(
+            &changed("ts", ConcreteDataType::timestamp_nanosecond_datatype()),
+            &routes(&[&expr]),
+        )
+        .unwrap_err();
+        assert!(err.to_string().contains("'ts'"), "unexpected error: {err}");
+    }
+
+    #[test]
+    fn test_deprecated_and_empty_bounds_are_skipped() {
+        validate_region_partition_bounds(
+            &changed("ts", ConcreteDataType::timestamp_nanosecond_datatype()),
+            &routes(&["", r#""MaxValue""#, r#"{"Value":{"UInt32":1}}"#, "not-json"]),
+        )
+        .unwrap();
+    }
+}

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

@@ -1478,3 +1478,176 @@ async fn test_annotation_alter_is_metadata_only() {
         }
     }
 }
+
+/// Builds the persisted partition expression `ts < <millis>` in the format the
+/// partition crate writes to region metadata.
+fn test_ts_partition_bound(millis: i64) -> String {
+    format!(
+        r#"{{"Expr":{{"lhs":{{"Column":"ts"}},"op":"Lt","rhs":{{"Value":{{"Timestamp":{{"value":{millis},"unit":"Millisecond"}}}}}}}}}}"#
+    )
+}
+
+fn test_partitioned_table_route(table_id: u32, region_bounds: &[i64]) -> TableRouteValue {
+    TableRouteValue::physical(
+        region_bounds
+            .iter()
+            .enumerate()
+            .map(|(index, bound)| RegionRoute {
+                region: Region {
+                    id: RegionId::new(table_id, index as u32 + 1),
+                    partition_expr: test_ts_partition_bound(*bound),
+                    ..Default::default()
+                },
+                leader_peer: Some(Peer::empty(index as u64 + 1)),
+                ..Default::default()
+            })
+            .collect(),
+    )
+}
+
+fn test_widen_ts_unit_task(table_name: &str, target_type: ColumnDataType) -> AlterTableTask {
+    AlterTableTask {
+        alter_table: AlterTableExpr {
+            catalog_name: DEFAULT_CATALOG_NAME.to_string(),
+            schema_name: DEFAULT_SCHEMA_NAME.to_string(),
+            table_name: table_name.to_string(),
+            kind: Some(Kind::ModifyColumnTypes(api::v1::ModifyColumnTypes {
+                modify_column_types: vec![api::v1::ModifyColumnType {
+                    column_name: "ts".to_string(),
+                    target_type: target_type as i32,
+                    target_type_extension: None,
+                }],
+            })),
+        },
+    }
+}
+
+#[tokio::test]
+async fn test_widen_ts_unit_rejects_incompatible_partition_bounds_before_altering() {
+    let node_manager = Arc::new(MockDatanodeManager::new(()));
+    let ddl_context = new_ddl_context(node_manager);
+    let table_name = "foo";
+    let table_id = 1024;
+    let task = test_create_table_task(table_name, table_id);
+
+    // Region 1 holds 2024-01-01, which fits nanoseconds. Region 2 holds 3000-01-01,
+    // which fits milliseconds but overflows nanoseconds.
+    let routes = test_partitioned_table_route(table_id, &[1_704_067_200_000, 32_503_680_000_000]);
+    ddl_context
+        .table_metadata_manager
+        .create_table_metadata(task.table_info.clone(), routes.clone(), HashMap::new())
+        .await
+        .unwrap();
+
+    let task = test_widen_ts_unit_task(table_name, ColumnDataType::TimestampNanosecond);
+    let mut procedure = AlterTableProcedure::new(table_id, task, ddl_context.clone()).unwrap();
+    let err = procedure.on_prepare().await.unwrap_err();
+
+    assert_matches!(err, Error::PartitionExprIncompatible { .. });
+    assert_eq!(err.status_code(), StatusCode::InvalidArguments);
+    // The rejection names the offending region and column.
+    assert!(
+        err.to_string().contains("4398046511106"),
+        "unexpected error: {err}"
+    );
+    assert!(err.to_string().contains("'ts'"), "unexpected error: {err}");
+
+    // The rejected ALTER left the table schema and the whole route unchanged.
+    let table_info = ddl_context
+        .table_metadata_manager
+        .table_info_manager()
+        .get(table_id)
+        .await
+        .unwrap()
+        .unwrap()
+        .into_inner()
+        .table_info;
+    assert_eq!(
+        table_info
+            .meta
+            .schema
+            .column_schema_by_name("ts")
+            .unwrap()
+            .data_type(),
+        &ConcreteDataType::timestamp_millisecond_datatype()
+    );
+    let (physical_table_id, physical_table_route) = ddl_context
+        .table_metadata_manager
+        .table_route_manager()
+        .get_physical_table_route(table_id)
+        .await
+        .unwrap();
+    assert_eq!(physical_table_id, table_id);
+    assert_eq!(
+        routes,
+        TableRouteValue::physical(physical_table_route.region_routes)
+    );
+}
+
+#[tokio::test]
+async fn test_widen_ts_unit_with_compatible_partition_bounds_completes() {
+    let node_manager = Arc::new(MockDatanodeManager::new(()));
+    let ddl_context = new_ddl_context(node_manager);
+    let table_name = "foo";
+    let table_id = 1024;
+    let task = test_create_table_task(table_name, table_id);
+
+    // Both persisted bounds fit nanoseconds.
+    let routes = test_partitioned_table_route(table_id, &[0, 1_704_067_200_000]);
+    ddl_context
+        .table_metadata_manager
+        .create_table_metadata(task.table_info.clone(), routes, HashMap::new())
+        .await
+        .unwrap();
+
+    let task = test_widen_ts_unit_task(table_name, ColumnDataType::TimestampNanosecond);
+    let procedure_id = ProcedureId::random();
+    let provider = Arc::new(MockContextProvider::default());
+    let mut procedure = AlterTableProcedure::new(table_id, task, ddl_context.clone()).unwrap();
+    procedure.on_prepare().await.unwrap();
+    procedure
+        .submit_alter_region_requests(procedure_id, provider.as_ref())
+        .await
+        .unwrap();
+    procedure.mut_data().set_column_metadatas(vec![
+        ColumnMetadata {
+            column_schema: ColumnSchema::new(
+                "ts",
+                ConcreteDataType::timestamp_nanosecond_datatype(),
+                false,
+            ),
+            semantic_type: SemanticType::Timestamp,
+            column_id: 0,
+        },
+        ColumnMetadata {
+            column_schema: ColumnSchema::new("host", ConcreteDataType::string_datatype(), false),
+            semantic_type: SemanticType::Tag,
+            column_id: 1,
+        },
+        ColumnMetadata {
+            column_schema: ColumnSchema::new("cpu", ConcreteDataType::float64_datatype(), false),
+            semantic_type: SemanticType::Field,
+            column_id: 2,
+        },
+    ]);
+    procedure.on_update_metadata().await.unwrap();
+
+    let table_info = ddl_context
+        .table_metadata_manager
+        .table_info_manager()
+        .get(table_id)
+        .await
+        .unwrap()
+        .unwrap()
+        .into_inner()
+        .table_info;
+    assert_eq!(
+        table_info
+            .meta
+            .schema
+            .column_schema_by_name("ts")
+            .unwrap()
+            .data_type(),
+        &ConcreteDataType::timestamp_nanosecond_datatype()
+    );
+}

src/common/meta/src/error.rs

@@ -1142,6 +1142,20 @@ pub enum Error {
         actual_column_id: u32,
     },
 
+    #[snafu(display(
+        "Partition expression of region {} is incompatible with the altered column '{}': the persisted bound cannot be converted to type {}",
+        region_id,
+        column,
+        target_type
+    ))]
+    PartitionExprIncompatible {
+        region_id: RegionId,
+        column: String,
+        target_type: String,
+        #[snafu(implicit)]
+        location: Location,
+    },
+
     #[cfg(feature = "enterprise")]
     #[snafu(display("Too large duration"))]
     TooLargeDuration {
@@ -1260,7 +1274,8 @@ impl ErrorExt for Error {
             | InvalidFileName { .. }
             | InvalidFlowRequestBody { .. }
             | InvalidFilePath { .. }
-            | ConflictingSchemaOptions { .. } => StatusCode::InvalidArguments,
+            | ConflictingSchemaOptions { .. }
+            | PartitionExprIncompatible { .. } => StatusCode::InvalidArguments,
 
             #[cfg(feature = "enterprise")]
             MissingInterval { .. } | NegativeDuration { .. } | TooLargeDuration { .. } => {

tests/cases/standalone/common/alter/change_col_type.result

@@ -386,3 +386,70 @@ DROP TABLE ts_overflow;
 
 Affected Rows: 0
 
+
+-- Issue 9454: persisted partition bounds keep their original unit. Widening the
+-- time index unit must be rejected before any region is altered when a persisted
+-- bound overflows the new unit.
+CREATE TABLE ts_bound_overflow (host STRING, ts TIMESTAMP TIME INDEX)
+PARTITION ON COLUMNS (ts) (
+  ts < '2024-01-01 00:00:00',
+  ts >= '2024-01-01 00:00:00' AND ts < '3000-01-01 00:00:00'
+);
+
+Affected Rows: 0
+
+INSERT INTO ts_bound_overflow VALUES ("a", "2024-06-01 12:00:00");
+
+Affected Rows: 1
+
+-- 3000-01-01 fits milliseconds but overflows nanoseconds, so the widening is
+-- rejected before any region is altered.
+-- SQLNESS REPLACE \d+\(\d+,\s+\d+\) REDACTED
+ALTER TABLE ts_bound_overflow MODIFY COLUMN ts TIMESTAMP_NS;
+
+Error: 1004(InvalidArguments), Partition expression of region REDACTED is incompatible with the altered column 'ts': the persisted bound cannot be converted to type TimestampNanosecond
+
+-- The rejected ALTER leaves the schema and the stored data unchanged.
+DESCRIBE ts_bound_overflow;
+
++--------+----------------------+-----+------+---------+---------------+
+| Column | Type                 | Key | Null | Default | Semantic Type |
++--------+----------------------+-----+------+---------+---------------+
+| host   | String               |     | YES  |         | FIELD         |
+| ts     | TimestampMillisecond | PRI | NO   |         | TIMESTAMP     |
++--------+----------------------+-----+------+---------+---------------+
+
+SELECT * FROM ts_bound_overflow ORDER BY ts;
+
++------+---------------------+
+| host | ts                  |
++------+---------------------+
+| a    | 2024-06-01T12:00:00 |
++------+---------------------+
+
+-- Widening to microseconds keeps every persisted bound representable.
+ALTER TABLE ts_bound_overflow MODIFY COLUMN ts TIMESTAMP_US;
+
+Affected Rows: 0
+
+DESCRIBE ts_bound_overflow;
+
++--------+----------------------+-----+------+---------+---------------+
+| Column | Type                 | Key | Null | Default | Semantic Type |
++--------+----------------------+-----+------+---------+---------------+
+| host   | String               |     | YES  |         | FIELD         |
+| ts     | TimestampMicrosecond | PRI | NO   |         | TIMESTAMP     |
++--------+----------------------+-----+------+---------+---------------+
+
+SELECT * FROM ts_bound_overflow ORDER BY ts;
+
++------+---------------------+
+| host | ts                  |
++------+---------------------+
+| a    | 2024-06-01T12:00:00 |
++------+---------------------+
+
+DROP TABLE ts_bound_overflow;
+
+Affected Rows: 0
+

tests/cases/standalone/common/alter/change_col_type.sql

@@ -136,3 +136,33 @@ ALTER TABLE ts_overflow MODIFY COLUMN ts TIMESTAMP_NS;
 DESCRIBE ts_overflow;
 
 DROP TABLE ts_overflow;
+
+-- Issue 9454: persisted partition bounds keep their original unit. Widening the
+-- time index unit must be rejected before any region is altered when a persisted
+-- bound overflows the new unit.
+CREATE TABLE ts_bound_overflow (host STRING, ts TIMESTAMP TIME INDEX)
+PARTITION ON COLUMNS (ts) (
+  ts < '2024-01-01 00:00:00',
+  ts >= '2024-01-01 00:00:00' AND ts < '3000-01-01 00:00:00'
+);
+
+INSERT INTO ts_bound_overflow VALUES ("a", "2024-06-01 12:00:00");
+
+-- 3000-01-01 fits milliseconds but overflows nanoseconds, so the widening is
+-- rejected before any region is altered.
+-- SQLNESS REPLACE \d+\(\d+,\s+\d+\) REDACTED
+ALTER TABLE ts_bound_overflow MODIFY COLUMN ts TIMESTAMP_NS;
+
+-- The rejected ALTER leaves the schema and the stored data unchanged.
+DESCRIBE ts_bound_overflow;
+
+SELECT * FROM ts_bound_overflow ORDER BY ts;
+
+-- Widening to microseconds keeps every persisted bound representable.
+ALTER TABLE ts_bound_overflow MODIFY COLUMN ts TIMESTAMP_US;
+
+DESCRIBE ts_bound_overflow;
+
+SELECT * FROM ts_bound_overflow ORDER BY ts;
+
+DROP TABLE ts_bound_overflow;