fix(meta): compare partition bounds with the scan comparison coercion

The preflight validator cast the persisted bound directly to the altered
column type, which rejected ALTERs the scan itself accepts: the region
coerces both the column and the bound to the common comparison type
(ExprSimplifier::coerce in PredicateGroup::new), so narrowing a field
column with an out-of-range bound still scans. Compare through
comparison_coercion(target_type, bound_type) instead and only require the
bound to be castable to the common type.

Also add a round-trip test in the partition crate that serializes real
PartitionExprs and runs them through the validator, so drift between the
Persisted* mirror and the partition types fails CI, and fix the error
message to match the comparison semantics.

Signed-off-by: fengyue-xve <1759519137@qq.com>
This commit is contained in:
风月
2026-10-10 10:16:59 +08:00
committed by fengyue-xve
parent a6ed555322
commit ff9c3f4b57
5 changed files with 174 additions and 17 deletions
+1
View File
@@ -46,6 +46,7 @@ common-time.workspace = true
common-wal.workspace = true common-wal.workspace = true
common-workload.workspace = true common-workload.workspace = true
datafusion-common.workspace = true datafusion-common.workspace = true
datafusion-expr.workspace = true
datatypes.workspace = true datatypes.workspace = true
deadpool = { workspace = true, optional = true } deadpool = { workspace = true, optional = true }
deadpool-postgres = { workspace = true, optional = true } deadpool-postgres = { workspace = true, optional = true }
+2
View File
@@ -17,6 +17,8 @@ mod metadata;
mod partition_compat; mod partition_compat;
mod region_request; mod region_request;
pub use partition_compat::validate_region_partition_bounds;
use std::collections::HashSet; use std::collections::HashSet;
use std::vec; use std::vec;
@@ -15,27 +15,32 @@
//! Preflight validation of persisted partition expressions against an ALTER TABLE schema change. //! 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 //! 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 //! referenced by a partition expression changes its type, the region coerces the bound
//! literal to the new column type while building the partition predicate on scan. If a //! literal and the new column type to a common comparison type while building the
//! bound cannot be represented with the new type (a classic example is a millisecond //! partition predicate on scan. If no common type exists, or the bound cannot be cast
//! timestamp bound that overflows when the time index column is widened to nanoseconds), //! to it (a classic example is a millisecond timestamp bound that overflows when the
//! every scan of that region fails at that point. Such an ALTER must be rejected before //! time index column is widened to nanoseconds), every scan of that region fails at
//! it is submitted to any region, otherwise already-altered regions keep failing while //! that point. Such an ALTER must be rejected before it is submitted to any region,
//! the rest of the table is untouched. //! 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 //! The `partition` crate depends on `common-meta`, so this crate cannot depend on
//! `partition` back and cannot reuse [`partition::expr::PartitionExpr`]. The `Persisted*` //! `partition` back and cannot reuse [`partition::expr::PartitionExpr`]. The `Persisted*`
//! types below mirror the JSON representation of `partition::partition::PartitionBound` and //! 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` //! `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 //! and `partition/src/expr.rs`; round-trip tests in `partition` run real serialized
//! same conversion (value -> [`ScalarValue`]) and the same checked cast //! expressions through this validator so drift fails in CI. The compatibility check itself
//! ([`ScalarValue::cast_to`]) that the region performs when it coerces the persisted bound //! deliberately repeats the same conversion (value -> [`ScalarValue`]) and the same
//! to the new column type, so a bound that would fail on scan is rejected here as well. //! 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 std::collections::HashMap;
use common_telemetry::warn; use common_telemetry::warn;
use datafusion_common::ScalarValue; use datafusion_common::ScalarValue;
use datafusion_expr::type_coercion::binary::comparison_coercion;
use datatypes::data_type::{ConcreteDataType, DataType}; use datatypes::data_type::{ConcreteDataType, DataType};
use datatypes::schema::Schema; use datatypes::schema::Schema;
use datatypes::value::{ 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 /// 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 /// with the types of `changed_columns`. Returns an error naming the first offending
/// region without modifying anything. /// 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<String, ConcreteDataType>, changed_columns: &HashMap<String, ConcreteDataType>,
region_routes: &[RegionRoute], region_routes: &[RegionRoute],
) -> Result<()> { ) -> Result<()> {
@@ -205,11 +213,29 @@ fn validate_comparison(
return Ok(()); 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(); let target_arrow_type = target_type.as_arrow_type();
if scalar.data_type() == target_arrow_type { let Some(common_type) = comparison_coercion(&target_arrow_type, &scalar.data_type()) else {
return Ok(()); warn!(
} "Persisted partition bound has no common comparison type with the altered \
if let Err(error) = scalar.cast_to(&target_arrow_type) { 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!( warn!(
error; error;
"Persisted partition bound is not representable with the altered column type, \ "Persisted partition bound is not representable with the altered column type, \
@@ -408,4 +434,31 @@ mod tests {
) )
.unwrap(); .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}");
}
} }
+1 -1
View File
@@ -1143,7 +1143,7 @@ pub enum Error {
}, },
#[snafu(display( #[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, region_id,
column, column,
target_type target_type
+101
View File
@@ -463,6 +463,14 @@ impl Display for PartitionExpr {
#[cfg(test)] #[cfg(test)]
mod tests { 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::*; use super::*;
#[test] #[test]
@@ -863,4 +871,97 @@ mod tests {
assert!(columns.contains("b")); assert!(columns.contains("b"));
assert!(columns.contains("c")); 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::<Value>(
r#"{"Timestamp":{"value":32503680000000,"unit":"Millisecond"}}"#,
)
.unwrap();
let start_ms =
serde_json::from_str::<Value>(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();
}
} }