mirror of
https://github.com/GreptimeTeam/greptimedb.git
synced 2026-10-21 11:22:31 +00:00
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:
@@ -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 }
|
||||||
|
|||||||
@@ -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}");
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -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
|
||||||
|
|||||||
@@ -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();
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user