diff --git a/src/common/meta/Cargo.toml b/src/common/meta/Cargo.toml index bd974deb4cf..64303b070c8 100644 --- a/src/common/meta/Cargo.toml +++ b/src/common/meta/Cargo.toml @@ -46,6 +46,7 @@ common-time.workspace = true common-wal.workspace = true common-workload.workspace = true datafusion-common.workspace = true +datafusion-expr.workspace = true datatypes.workspace = true deadpool = { workspace = true, optional = true } deadpool-postgres = { workspace = true, optional = true } diff --git a/src/common/meta/src/ddl/alter_table.rs b/src/common/meta/src/ddl/alter_table.rs index 7a10bb9c2af..232191def88 100644 --- a/src/common/meta/src/ddl/alter_table.rs +++ b/src/common/meta/src/ddl/alter_table.rs @@ -17,6 +17,8 @@ mod metadata; mod partition_compat; mod region_request; +pub use partition_compat::validate_region_partition_bounds; + use std::collections::HashSet; use std::vec; diff --git a/src/common/meta/src/ddl/alter_table/partition_compat.rs b/src/common/meta/src/ddl/alter_table/partition_compat.rs index d9390f905ee..e798755dc4f 100644 --- a/src/common/meta/src/ddl/alter_table/partition_compat.rs +++ b/src/common/meta/src/ddl/alter_table/partition_compat.rs @@ -15,27 +15,32 @@ //! 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. +//! referenced by a partition expression changes its type, the region coerces the bound +//! literal and the new column type to a common comparison type while building the +//! partition predicate on scan. If no common type exists, or the bound cannot be cast +//! to it (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. +//! and `partition/src/expr.rs`; round-trip tests in `partition` run real serialized +//! expressions through this validator so drift fails in CI. The compatibility check itself +//! deliberately repeats the same conversion (value -> [`ScalarValue`]) and the same +//! comparison coercion the region performs when it builds the partition predicate +//! (`ExprSimplifier::coerce` / [`comparison_coercion`]): both sides are cast to the common +//! comparison type, and the persisted bound must be castable to that type with +//! [`ScalarValue::cast_to`]. 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 datafusion_expr::type_coercion::binary::comparison_coercion; use datatypes::data_type::{ConcreteDataType, DataType}; use datatypes::schema::Schema; use datatypes::value::{ @@ -111,7 +116,10 @@ pub(crate) fn changed_column_types( /// 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( +/// +/// Public so that round-trip tests in the `partition` crate can run real serialized +/// partition expressions through the same mirror used here. +pub fn validate_region_partition_bounds( changed_columns: &HashMap, region_routes: &[RegionRoute], ) -> Result<()> { @@ -205,11 +213,29 @@ fn validate_comparison( return Ok(()); }; + // Mirror `ExprSimplifier::coerce` in `PredicateGroup::new`: the region casts + // both the column and the persisted bound to the common comparison type. The + // bound is not cast to the column type itself, so narrowing a column (e.g. + // BIGINT to INT with an out-of-range bound) is fine as long as the bound + // survives the common type, while a bound that cannot be cast to it — or a + // pair of types without a common comparison type — fails every scan. 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) { + let Some(common_type) = comparison_coercion(&target_arrow_type, &scalar.data_type()) else { + warn!( + "Persisted partition bound has no common comparison type with the altered \ + column type, region_id: {}, column: '{}', target_type: {}", + region_id, column, target_type + ); + return PartitionExprIncompatibleSnafu { + region_id, + column: column.clone(), + target_type: target_type.to_string(), + } + .fail(); + }; + if scalar.data_type() != common_type + && let Err(error) = scalar.cast_to(&common_type) + { warn!( error; "Persisted partition bound is not representable with the altered column type, \ @@ -408,4 +434,31 @@ mod tests { ) .unwrap(); } + + #[test] + fn test_field_narrowing_keeps_bounds_the_scan_coerces() { + // BIGINT bound outside the INT range with the column narrowed to INT: the + // region casts the column up to the common Int64 comparison type, so the + // bound still scans and the ALTER must not be rejected. + let bound = + r#"{"Expr":{"lhs":{"Column":"v"},"op":"Lt","rhs":{"Value":{"Int64":3000000000}}}}"#; + validate_region_partition_bounds( + &changed("v", ConcreteDataType::int32_datatype()), + &routes(&[bound]), + ) + .unwrap(); + } + + #[test] + fn test_bound_without_common_comparison_type_is_rejected() { + // No comparison coercion exists between a boolean bound and an int column, + // so the region's predicate build fails and the ALTER must be rejected. + let bound = r#"{"Expr":{"lhs":{"Column":"v"},"op":"Lt","rhs":{"Value":{"Boolean":true}}}}"#; + let err = validate_region_partition_bounds( + &changed("v", ConcreteDataType::int32_datatype()), + &routes(&[bound]), + ) + .unwrap_err(); + assert!(err.to_string().contains("'v'"), "unexpected error: {err}"); + } } diff --git a/src/common/meta/src/error.rs b/src/common/meta/src/error.rs index 1b5d8e83898..9637af30a19 100644 --- a/src/common/meta/src/error.rs +++ b/src/common/meta/src/error.rs @@ -1143,7 +1143,7 @@ pub enum Error { }, #[snafu(display( - "Partition expression of region {} is incompatible with the altered column '{}': the persisted bound cannot be converted to type {}", + "Partition expression of region {} is incompatible with the altered column '{}': the persisted bound can no longer be compared with the new column type {}", region_id, column, target_type diff --git a/src/partition/src/expr.rs b/src/partition/src/expr.rs index cb15f5ae5d7..c49e0a2b203 100644 --- a/src/partition/src/expr.rs +++ b/src/partition/src/expr.rs @@ -463,6 +463,14 @@ impl Display for PartitionExpr { #[cfg(test)] mod tests { + use std::collections::HashMap; + + use common_meta::ddl::alter_table::validate_region_partition_bounds; + use common_meta::error::Error; + use common_meta::rpc::router::{Region, RegionRoute}; + use datatypes::data_type::ConcreteDataType; + use store_api::storage::RegionId; + use super::*; #[test] @@ -863,4 +871,97 @@ mod tests { assert!(columns.contains("b")); assert!(columns.contains("c")); } + + /// Real serialized expressions must survive the round trip through + /// `common-meta`'s persisted-expression mirror: whenever that mirror drifts + /// from these types, deserialization fails, validation is skipped and the + /// overflowing cases below stop producing `PartitionExprIncompatible`. + #[test] + fn test_serialized_expressions_validate_against_type_changes() { + let overflow_ms = serde_json::from_str::( + r#"{"Timestamp":{"value":32503680000000,"unit":"Millisecond"}}"#, + ) + .unwrap(); + let start_ms = + serde_json::from_str::(r#"{"Timestamp":{"value":0,"unit":"Millisecond"}}"#) + .unwrap(); + let ts_ns = ConcreteDataType::timestamp_nanosecond_datatype(); + + let routes = |expr: &PartitionExpr| { + vec![RegionRoute { + region: Region { + id: RegionId::new(1024, 1), + partition_expr: expr.as_json_str().unwrap(), + ..Default::default() + }, + ..Default::default() + }] + }; + let changed = |column: &str, data_type: ConcreteDataType| { + HashMap::from([(column.to_string(), data_type)]) + }; + + // Compile-time tripwire: exhaustively matching the operand and operator + // variants makes a new variant break this test at compile time, so the + // `partition_compat::Persisted*` mirror and these cases must be extended + // together instead of silently skipping validation. + fn walk(expr: &PartitionExpr) { + walk_operand(&expr.lhs); + walk_operand(&expr.rhs); + match expr.op { + RestrictedOp::Eq + | RestrictedOp::NotEq + | RestrictedOp::Lt + | RestrictedOp::LtEq + | RestrictedOp::Gt + | RestrictedOp::GtEq + | RestrictedOp::And + | RestrictedOp::Or => {} + } + } + fn walk_operand(operand: &Operand) { + match operand { + Operand::Column(_) | Operand::Value(_) => {} + Operand::Expr(expr) => walk(expr), + } + } + + // Every expression shape must be rejected while the bound overflows + // nanoseconds, with the error naming the offending column. + let simple = col("ts").lt(overflow_ms.clone()); + let nested_and = col("ts") + .gt_eq(start_ms.clone()) + .and(col("ts").lt(overflow_ms.clone())); + let nested_or = PartitionExpr::new( + Operand::Expr(col("host").eq(Value::String("a".into()))), + RestrictedOp::Or, + Operand::Expr(col("ts").lt(overflow_ms.clone())), + ); + for expr in [&simple, &nested_and, &nested_or] { + walk(expr); + let err = + validate_region_partition_bounds(&changed("ts", ts_ns.clone()), &routes(expr)) + .unwrap_err(); + assert!( + matches!( + err, + Error::PartitionExprIncompatible { ref column, .. } if column == "ts" + ), + "unexpected error: {err}" + ); + } + + // A representable bound passes. + let compatible = col("ts").lt(start_ms); + validate_region_partition_bounds(&changed("ts", ts_ns), &routes(&compatible)).unwrap(); + + // Narrowing a field column must not be over-rejected: the scan casts the + // column up to the common Int64 type instead of the bound down to Int32. + let narrowing = col("v").lt(Value::Int64(3_000_000_000)); + validate_region_partition_bounds( + &changed("v", ConcreteDataType::int32_datatype()), + &routes(&narrowing), + ) + .unwrap(); + } }