common_meta/ddl/alter_table/
partition_compat.rs

1// Copyright 2023 Greptime Team
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15//! Rejects an ALTER TABLE whose column type changes would break the persisted
16//! partition expressions of a region, before any region is altered.
17//!
18//! Bounds keep the type they were written with. Row routing compares them with
19//! row values as `Value`s, which only stays valid across a timestamp unit
20//! widening; scans coerce the bound to the common comparison type, which can
21//! overflow (a millisecond bound in year 3000 cast to nanoseconds).
22//!
23//! `partition` depends on this crate, so the `Persisted*` types mirror the JSON of
24//! `partition::partition::PartitionBound` and `partition::expr::PartitionExpr`.
25//! A test in `partition` runs real serialized expressions through this validator.
26
27use std::collections::HashMap;
28
29use common_telemetry::warn;
30use datafusion_common::ScalarValue;
31use datafusion_expr::type_coercion::binary::comparison_coercion;
32use datatypes::data_type::{ConcreteDataType, DataType};
33use datatypes::schema::Schema;
34use datatypes::value::{
35    Value, duration_to_scalar_value, time_to_scalar_value, timestamp_to_scalar_value,
36};
37use serde::Deserialize;
38use store_api::storage::RegionId;
39
40use crate::error::{PartitionExprIncompatibleSnafu, Result};
41use crate::rpc::router::RegionRoute;
42
43#[derive(Debug, Deserialize)]
44enum PersistedPartitionBound {
45    /// Deprecated since 0.9.0.
46    #[allow(dead_code)]
47    Value(Value),
48    /// Deprecated since 0.15.0.
49    MaxValue,
50    Expr(PersistedPartitionExpr),
51}
52
53#[derive(Debug, Deserialize)]
54struct PersistedPartitionExpr {
55    lhs: Box<PersistedOperand>,
56    op: PersistedOp,
57    rhs: Box<PersistedOperand>,
58}
59
60#[derive(Debug, Deserialize)]
61enum PersistedOperand {
62    Column(String),
63    Value(Value),
64    Expr(PersistedPartitionExpr),
65    /// `iceberg_hash(column)`.
66    IcebergHash(String),
67}
68
69#[derive(Debug, Deserialize)]
70enum PersistedOp {
71    Eq,
72    NotEq,
73    Lt,
74    LtEq,
75    Gt,
76    GtEq,
77    And,
78    Or,
79}
80
81/// Returns the columns present in both schemas whose type differs, as (old, new).
82pub(crate) fn changed_column_types(
83    old_schema: &Schema,
84    new_schema: &Schema,
85) -> HashMap<String, (ConcreteDataType, ConcreteDataType)> {
86    let mut changed = HashMap::new();
87    for new_column in new_schema.column_schemas() {
88        let Some(old_column) = old_schema.column_schema_by_name(&new_column.name) else {
89            continue;
90        };
91        if old_column.data_type != new_column.data_type {
92            changed.insert(
93                new_column.name.clone(),
94                (old_column.data_type.clone(), new_column.data_type.clone()),
95            );
96        }
97    }
98
99    changed
100}
101
102/// Returns an error naming the first region whose partition expression can't take
103/// the type changes in `changed_columns`.
104///
105/// Public so that tests in `partition` can run real serialized expressions through it.
106pub fn validate_region_partition_bounds(
107    changed_columns: &HashMap<String, (ConcreteDataType, ConcreteDataType)>,
108    region_routes: &[RegionRoute],
109) -> Result<()> {
110    if changed_columns.is_empty() {
111        return Ok(());
112    }
113
114    for route in region_routes {
115        let expr_json = route.region.partition_expr();
116        if expr_json.is_empty() {
117            continue;
118        }
119
120        let bound: PersistedPartitionBound = match serde_json::from_str(&expr_json) {
121            Ok(bound) => bound,
122            Err(err) => {
123                // Such a region already fails every scan, regardless of this ALTER.
124                warn!(
125                    "Skipping partition expression validation for region {}, \
126                     the persisted expression is not parseable: {}, expr: {}",
127                    route.region.id, err, expr_json
128                );
129                continue;
130            }
131        };
132        // Deprecated bound formats are not used to build partition predicates.
133        let PersistedPartitionBound::Expr(expr) = bound else {
134            continue;
135        };
136
137        validate_expr(&expr, changed_columns, route.region.id)?;
138    }
139
140    Ok(())
141}
142
143fn validate_expr(
144    expr: &PersistedPartitionExpr,
145    changed_columns: &HashMap<String, (ConcreteDataType, ConcreteDataType)>,
146    region_id: RegionId,
147) -> Result<()> {
148    match expr.op {
149        PersistedOp::And | PersistedOp::Or => {
150            validate_operand(&expr.lhs, changed_columns, region_id)?;
151            validate_operand(&expr.rhs, changed_columns, region_id)?;
152        }
153        PersistedOp::Eq
154        | PersistedOp::NotEq
155        | PersistedOp::Lt
156        | PersistedOp::LtEq
157        | PersistedOp::Gt
158        | PersistedOp::GtEq => {
159            validate_comparison(&expr.lhs, &expr.rhs, changed_columns, region_id)?;
160            validate_comparison(&expr.rhs, &expr.lhs, changed_columns, region_id)?;
161        }
162    }
163
164    Ok(())
165}
166
167fn validate_operand(
168    operand: &PersistedOperand,
169    changed_columns: &HashMap<String, (ConcreteDataType, ConcreteDataType)>,
170    region_id: RegionId,
171) -> Result<()> {
172    match operand {
173        PersistedOperand::Expr(expr) => validate_expr(expr, changed_columns, region_id),
174        PersistedOperand::Column(_)
175        | PersistedOperand::Value(_)
176        | PersistedOperand::IcebergHash(_) => Ok(()),
177    }
178}
179
180fn validate_comparison(
181    operand: &PersistedOperand,
182    other: &PersistedOperand,
183    changed_columns: &HashMap<String, (ConcreteDataType, ConcreteDataType)>,
184    region_id: RegionId,
185) -> Result<()> {
186    let (column, value) = match (operand, other) {
187        (PersistedOperand::Column(column), PersistedOperand::Value(value)) => (column, value),
188        // The hash depends on the column type, so any change reroutes rows. Table
189        // metadata rejects such changes first; this covers stale partition columns.
190        (PersistedOperand::IcebergHash(column), PersistedOperand::Value(_)) => {
191            return match changed_columns.get(column) {
192                Some((_, target_type)) => incompatible(region_id, column, target_type),
193                None => Ok(()),
194            };
195        }
196        _ => return Ok(()),
197    };
198    let Some((old_type, target_type)) = changed_columns.get(column) else {
199        return Ok(());
200    };
201    let Some(scalar) = value_to_scalar_value(value) else {
202        return Ok(());
203    };
204
205    if !old_type.is_timestamp_unit_widening_to(target_type) {
206        return incompatible(region_id, column, target_type);
207    }
208
209    // Mirrors the coercion `PredicateGroup::new` applies on scan: the bound must
210    // cast to the common comparison type.
211    let Some(common_type) = comparison_coercion(&target_type.as_arrow_type(), &scalar.data_type())
212    else {
213        return incompatible(region_id, column, target_type);
214    };
215    if scalar.data_type() != common_type
216        && let Err(error) = scalar.cast_to(&common_type)
217    {
218        warn!(error; "Partition bound of region {} can't cast to {}", region_id, common_type);
219        return incompatible(region_id, column, target_type);
220    }
221
222    Ok(())
223}
224
225fn incompatible(region_id: RegionId, column: &str, target_type: &ConcreteDataType) -> Result<()> {
226    PartitionExprIncompatibleSnafu {
227        region_id,
228        column,
229        target_type: target_type.to_string(),
230    }
231    .fail()
232}
233
234/// Converts a bound the same way as `partition::expr::Operand::try_as_logical_expr`.
235/// Returns `None` for values that can't be partition bounds.
236fn value_to_scalar_value(value: &Value) -> Option<ScalarValue> {
237    Some(match value {
238        Value::Boolean(v) => ScalarValue::Boolean(Some(*v)),
239        Value::UInt8(v) => ScalarValue::UInt8(Some(*v)),
240        Value::UInt16(v) => ScalarValue::UInt16(Some(*v)),
241        Value::UInt32(v) => ScalarValue::UInt32(Some(*v)),
242        Value::UInt64(v) => ScalarValue::UInt64(Some(*v)),
243        Value::Int8(v) => ScalarValue::Int8(Some(*v)),
244        Value::Int16(v) => ScalarValue::Int16(Some(*v)),
245        Value::Int32(v) => ScalarValue::Int32(Some(*v)),
246        Value::Int64(v) => ScalarValue::Int64(Some(*v)),
247        Value::Float32(v) => ScalarValue::Float32(Some(v.0)),
248        Value::Float64(v) => ScalarValue::Float64(Some(v.0)),
249        Value::String(v) => ScalarValue::Utf8(Some(v.as_utf8().to_string())),
250        Value::Binary(v) => ScalarValue::Binary(Some(v.to_vec())),
251        Value::Date(v) => ScalarValue::Date32(Some(v.val())),
252        Value::Null => ScalarValue::Null,
253        Value::Timestamp(t) => timestamp_to_scalar_value(t.unit(), Some(t.value())),
254        Value::Time(t) => time_to_scalar_value(*t.unit(), Some(t.value())).ok()?,
255        Value::IntervalYearMonth(v) => ScalarValue::IntervalYearMonth(Some(v.to_i32())),
256        Value::IntervalDayTime(v) => ScalarValue::IntervalDayTime(Some((*v).into())),
257        Value::IntervalMonthDayNano(v) => ScalarValue::IntervalMonthDayNano(Some((*v).into())),
258        Value::Duration(d) => duration_to_scalar_value(d.unit(), Some(d.value())),
259        Value::Decimal128(d) => {
260            let (v, p, s) = d.to_scalar_value();
261            ScalarValue::Decimal128(v, p, s)
262        }
263        Value::List(_) | Value::Struct(_) | Value::Json(_) => return None,
264    })
265}
266
267#[cfg(test)]
268mod tests {
269    use datatypes::schema::ColumnSchema;
270    use store_api::storage::RegionId;
271
272    use super::*;
273    use crate::rpc::router::Region;
274
275    fn schema(ts_type: ConcreteDataType) -> Schema {
276        Schema::new(vec![
277            ColumnSchema::new("ts", ts_type, false),
278            ColumnSchema::new("host", ConcreteDataType::string_datatype(), true),
279        ])
280    }
281
282    fn ts_bound(millis: i64) -> String {
283        format!(
284            r#"{{"Expr":{{"lhs":{{"Column":"ts"}},"op":"Lt","rhs":{{"Value":{{"Timestamp":{{"value":{millis},"unit":"Millisecond"}}}}}}}}}}"#
285        )
286    }
287
288    fn routes(exprs: &[&str]) -> Vec<RegionRoute> {
289        exprs
290            .iter()
291            .enumerate()
292            .map(|(index, expr)| RegionRoute {
293                region: Region {
294                    id: RegionId::new(1024, index as u32 + 1),
295                    partition_expr: expr.to_string(),
296                    ..Default::default()
297                },
298                ..Default::default()
299            })
300            .collect()
301    }
302
303    fn changed(
304        column: &str,
305        old_type: ConcreteDataType,
306        new_type: ConcreteDataType,
307    ) -> HashMap<String, (ConcreteDataType, ConcreteDataType)> {
308        HashMap::from([(column.to_string(), (old_type, new_type))])
309    }
310
311    #[test]
312    fn test_changed_column_types() {
313        let old = schema(ConcreteDataType::timestamp_millisecond_datatype());
314        let new = schema(ConcreteDataType::timestamp_nanosecond_datatype());
315        let changed = changed_column_types(&old, &new);
316        assert_eq!(changed.len(), 1);
317        assert_eq!(
318            changed["ts"],
319            (
320                ConcreteDataType::timestamp_millisecond_datatype(),
321                ConcreteDataType::timestamp_nanosecond_datatype()
322            )
323        );
324
325        // Added and dropped columns are not type changes.
326        let mut new_columns = old.column_schemas().to_vec();
327        new_columns.push(ColumnSchema::new(
328            "extra",
329            ConcreteDataType::int64_datatype(),
330            true,
331        ));
332        let new = Schema::new(new_columns);
333        assert!(changed_column_types(&old, &new).is_empty());
334    }
335
336    #[test]
337    fn test_overflowing_bound_is_rejected() {
338        // Year 3000 fits milliseconds but overflows nanoseconds.
339        let routes = routes(&[&ts_bound(32_503_680_000_000)]);
340        let err = validate_region_partition_bounds(
341            &changed(
342                "ts",
343                ConcreteDataType::timestamp_millisecond_datatype(),
344                ConcreteDataType::timestamp_nanosecond_datatype(),
345            ),
346            &routes,
347        )
348        .unwrap_err();
349        assert!(
350            err.to_string().contains("region 4398046511105"),
351            "unexpected error: {err}"
352        );
353        assert!(err.to_string().contains("'ts'"), "unexpected error: {err}");
354    }
355
356    #[test]
357    fn test_representable_bound_is_accepted() {
358        // 2024-01-01T00:00:00Z fits both microseconds and nanoseconds.
359        let routes = routes(&[&ts_bound(1_704_067_200_000)]);
360        validate_region_partition_bounds(
361            &changed(
362                "ts",
363                ConcreteDataType::timestamp_millisecond_datatype(),
364                ConcreteDataType::timestamp_nanosecond_datatype(),
365            ),
366            &routes,
367        )
368        .unwrap();
369        validate_region_partition_bounds(
370            &changed(
371                "ts",
372                ConcreteDataType::timestamp_millisecond_datatype(),
373                ConcreteDataType::timestamp_microsecond_datatype(),
374            ),
375            &routes,
376        )
377        .unwrap();
378    }
379
380    #[test]
381    fn test_later_region_rejection_names_the_region() {
382        let routes = routes(&[&ts_bound(1_704_067_200_000), &ts_bound(32_503_680_000_000)]);
383        let err = validate_region_partition_bounds(
384            &changed(
385                "ts",
386                ConcreteDataType::timestamp_millisecond_datatype(),
387                ConcreteDataType::timestamp_nanosecond_datatype(),
388            ),
389            &routes,
390        )
391        .unwrap_err();
392        assert!(
393            err.to_string().contains("region 4398046511106"),
394            "unexpected error: {err}"
395        );
396    }
397
398    #[test]
399    fn test_unrelated_columns_are_skipped() {
400        let host_bound =
401            r#"{"Expr":{"lhs":{"Column":"host"},"op":"Eq","rhs":{"Value":{"String":"a"}}}}"#;
402        // The changed column is not referenced by this bound.
403        validate_region_partition_bounds(
404            &changed(
405                "ts",
406                ConcreteDataType::timestamp_millisecond_datatype(),
407                ConcreteDataType::timestamp_nanosecond_datatype(),
408            ),
409            &routes(&[host_bound]),
410        )
411        .unwrap();
412        // The referenced column did not change.
413        validate_region_partition_bounds(
414            &changed(
415                "host",
416                ConcreteDataType::int32_datatype(),
417                ConcreteDataType::string_datatype(),
418            ),
419            &routes(&[&ts_bound(32_503_680_000_000)]),
420        )
421        .unwrap();
422    }
423
424    #[test]
425    fn test_nested_expression_is_validated() {
426        let expr = 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"}}}}}}}"#.to_string();
427        let err = validate_region_partition_bounds(
428            &changed(
429                "ts",
430                ConcreteDataType::timestamp_millisecond_datatype(),
431                ConcreteDataType::timestamp_nanosecond_datatype(),
432            ),
433            &routes(&[&expr]),
434        )
435        .unwrap_err();
436        assert!(err.to_string().contains("'ts'"), "unexpected error: {err}");
437    }
438
439    #[test]
440    fn test_deprecated_and_empty_bounds_are_skipped() {
441        validate_region_partition_bounds(
442            &changed(
443                "ts",
444                ConcreteDataType::timestamp_millisecond_datatype(),
445                ConcreteDataType::timestamp_nanosecond_datatype(),
446            ),
447            &routes(&["", r#""MaxValue""#, r#"{"Value":{"UInt32":1}}"#, "not-json"]),
448        )
449        .unwrap();
450    }
451
452    #[test]
453    fn test_field_narrowing_is_rejected_for_persisted_bounds() {
454        // Scans would coerce both sides, but row routing compares Int64 bounds with
455        // Int32 rows.
456        let bound =
457            r#"{"Expr":{"lhs":{"Column":"v"},"op":"Lt","rhs":{"Value":{"Int64":3000000000}}}}"#;
458        let err = validate_region_partition_bounds(
459            &changed(
460                "v",
461                ConcreteDataType::int64_datatype(),
462                ConcreteDataType::int32_datatype(),
463            ),
464            &routes(&[bound]),
465        )
466        .unwrap_err();
467        assert!(err.to_string().contains("'v'"), "unexpected error: {err}");
468    }
469
470    #[test]
471    fn test_non_timestamp_widening_is_rejected_for_persisted_bounds() {
472        // Row routing would compare Int32 bounds with Int64 rows.
473        let bound = r#"{"Expr":{"lhs":{"Column":"v"},"op":"Lt","rhs":{"Value":{"Int32":10}}}}"#;
474        let err = validate_region_partition_bounds(
475            &changed(
476                "v",
477                ConcreteDataType::int32_datatype(),
478                ConcreteDataType::int64_datatype(),
479            ),
480            &routes(&[bound]),
481        )
482        .unwrap_err();
483        assert!(err.to_string().contains("'v'"), "unexpected error: {err}");
484    }
485
486    #[test]
487    fn test_bound_without_common_comparison_type_is_rejected() {
488        // No common comparison type between a boolean and a timestamp.
489        let bound =
490            r#"{"Expr":{"lhs":{"Column":"ts"},"op":"Lt","rhs":{"Value":{"Boolean":true}}}}"#;
491        let err = validate_region_partition_bounds(
492            &changed(
493                "ts",
494                ConcreteDataType::timestamp_millisecond_datatype(),
495                ConcreteDataType::timestamp_nanosecond_datatype(),
496            ),
497            &routes(&[bound]),
498        )
499        .unwrap_err();
500        assert!(err.to_string().contains("'ts'"), "unexpected error: {err}");
501    }
502}
common_meta/ddl/alter_table/partition_compat.rs

common_meta/ddl/alter_table/
partition_compat.rs
