use super::super::result::QueryRow;
use super::super::select_optimizer::SSTablePredicate;
use super::value_ops::{compare_values_ordering_predicate, values_equal};
use crate::{types::Value, Error, Result};
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum LeafOutcome {
True,
False,
Unknown,
}
pub fn evaluate_leaf(row: &QueryRow, predicate: &SSTablePredicate) -> LeafOutcome {
use super::super::select_optimizer::SSTableFilterOp;
if predicate.is_token() {
if row.key.0.is_empty() {
return LeafOutcome::Unknown;
}
let row_token = crate::util::cassandra_murmur3::cassandra_murmur3_token(&row.key.0);
let Some(Value::BigInt(bound)) = predicate.values.first() else {
return LeafOutcome::Unknown;
};
let matches = match &predicate.operation {
SSTableFilterOp::Gt => row_token > *bound,
SSTableFilterOp::Gte => row_token >= *bound,
SSTableFilterOp::Lt => row_token < *bound,
SSTableFilterOp::Lte => row_token <= *bound,
SSTableFilterOp::Equal => row_token == *bound,
_ => return LeafOutcome::Unknown,
};
return if matches {
LeafOutcome::True
} else {
LeafOutcome::False
};
}
let column_value = match row.values.get(predicate.column.as_str()) {
None | Some(Value::Null) => return LeafOutcome::Unknown,
Some(v) => v,
};
let matches = match &predicate.operation {
SSTableFilterOp::Equal => predicate
.values
.first()
.is_some_and(|v| values_equal(column_value, v)),
SSTableFilterOp::In => predicate
.values
.iter()
.any(|v| values_equal(column_value, v)),
SSTableFilterOp::Range => {
if predicate.values.len() < 2 {
false
} else {
let lo = &predicate.values[0];
let hi = &predicate.values[1];
compare_values_ordering_predicate(column_value, lo).is_some_and(|o| o.is_ge())
&& compare_values_ordering_predicate(column_value, hi)
.is_some_and(|o| o.is_le())
}
}
SSTableFilterOp::Gt => predicate.values.first().is_some_and(|b| {
compare_values_ordering_predicate(column_value, b).is_some_and(|o| o.is_gt())
}),
SSTableFilterOp::Gte => predicate.values.first().is_some_and(|b| {
compare_values_ordering_predicate(column_value, b).is_some_and(|o| o.is_ge())
}),
SSTableFilterOp::Lt => predicate.values.first().is_some_and(|b| {
compare_values_ordering_predicate(column_value, b).is_some_and(|o| o.is_lt())
}),
SSTableFilterOp::Lte => predicate.values.first().is_some_and(|b| {
compare_values_ordering_predicate(column_value, b).is_some_and(|o| o.is_le())
}),
SSTableFilterOp::Prefix => matches!(
(column_value, predicate.values.first()),
(Value::Text(s), Some(Value::Text(p))) if s.starts_with(p)
),
SSTableFilterOp::BloomFilter => true, };
if matches {
LeafOutcome::True
} else {
LeafOutcome::False
}
}
pub fn evaluate_predicates(row: &QueryRow, predicates: &[SSTablePredicate]) -> Result<bool> {
for predicate in predicates {
if evaluate_leaf(row, predicate) != LeafOutcome::True {
return Ok(false);
}
}
Ok(true)
}
pub(super) fn validate_token_predicates(
predicates: &[SSTablePredicate],
schema: Option<&crate::schema::TableSchema>,
) -> Result<()> {
let token_predicates: Vec<&SSTablePredicate> =
predicates.iter().filter(|p| p.is_token()).collect();
if token_predicates.is_empty() {
return Ok(());
}
let Some(schema) = schema else {
return Err(Error::query_execution(
"token() restriction requires a known table schema to validate its argument \
against the partition key"
.to_string(),
));
};
let mut pk_cols: Vec<&crate::schema::KeyColumn> = schema.partition_keys.iter().collect();
pk_cols.sort_by_key(|c| c.position);
let expected: Vec<&str> = pk_cols.iter().map(|c| c.name.as_str()).collect();
for predicate in token_predicates {
let cols = predicate.token_columns.as_deref().unwrap_or(&[]);
let matches = cols.len() == expected.len()
&& cols
.iter()
.zip(expected.iter())
.all(|(got, want)| got == want);
if !matches {
return Err(Error::query_execution(format!(
"token() must be applied to the entire partition key in declared order \
({}); got token({})",
expected.join(", "),
cols.join(", "),
)));
}
}
Ok(())
}
#[cfg(test)]
mod tests {
use super::super::super::select_optimizer::{SSTableFilterOp, SSTablePredicate};
use super::super::test_support::{
composite_pk_schema, row_with_int, row_with_key, row_with_value, single_pk_schema,
};
use super::*;
use crate::types::RowKey;
#[test]
fn evaluate_leaf_token_predicate_filters_by_token() {
let key = b"some-partition-key".to_vec();
let token = crate::util::cassandra_murmur3::cassandra_murmur3_token(&key);
let row = row_with_key(&key);
let gte = SSTablePredicate::token(
vec!["id".to_string()],
SSTableFilterOp::Gte,
vec![Value::BigInt(token)],
);
assert_eq!(evaluate_leaf(&row, >e), LeafOutcome::True);
let gt = SSTablePredicate::token(
vec!["id".to_string()],
SSTableFilterOp::Gt,
vec![Value::BigInt(token)],
);
assert_eq!(evaluate_leaf(&row, >), LeafOutcome::False);
let gte_above = SSTablePredicate::token(
vec!["id".to_string()],
SSTableFilterOp::Gte,
vec![Value::BigInt(token.saturating_add(1))],
);
assert_eq!(evaluate_leaf(&row, >e_above), LeafOutcome::False);
let empty = row_with_key(&[]);
assert_eq!(evaluate_leaf(&empty, >e), LeafOutcome::Unknown);
}
#[test]
fn inequality_predicates_apply_single_bound() {
let bound =
|op: SSTableFilterOp| SSTablePredicate::column("ck", op, vec![Value::Integer(200)]);
let eval = |op: SSTableFilterOp, ck: i64| {
evaluate_predicates(&row_with_int("ck", ck), std::slice::from_ref(&bound(op))).unwrap()
};
assert!(!eval(SSTableFilterOp::Gt, 200));
assert!(eval(SSTableFilterOp::Gt, 201));
assert!(eval(SSTableFilterOp::Gte, 200));
assert!(!eval(SSTableFilterOp::Gte, 199));
assert!(eval(SSTableFilterOp::Lt, 199));
assert!(!eval(SSTableFilterOp::Lt, 200));
assert!(eval(SSTableFilterOp::Lte, 200));
assert!(!eval(SSTableFilterOp::Lte, 201));
}
#[test]
fn evaluate_leaf_is_three_valued() {
let gt200 = SSTablePredicate::column("ck", SSTableFilterOp::Gt, vec![Value::Integer(200)]);
assert_eq!(
evaluate_leaf(&row_with_int("ck", 201), >200),
LeafOutcome::True
);
assert_eq!(
evaluate_leaf(&row_with_int("ck", 200), >200),
LeafOutcome::False
);
assert_eq!(
evaluate_leaf(&row_with_int("other", 999), >200),
LeafOutcome::Unknown
);
let mut values: std::collections::HashMap<std::sync::Arc<str>, Value> =
std::collections::HashMap::new();
values.insert("ck".into(), Value::Null);
let null_row = QueryRow {
values,
key: RowKey::new(Vec::new()),
metadata: Default::default(),
cell_metadata: None,
};
assert_eq!(evaluate_leaf(&null_row, >200), LeafOutcome::Unknown);
}
#[test]
fn evaluate_leaf_in_coerces_narrow_numeric_columns() {
let in_pred = SSTablePredicate::column(
"v",
SSTableFilterOp::In,
vec![Value::Integer(7), Value::Integer(9)],
);
let row_of = |val: Value| {
let mut values: std::collections::HashMap<std::sync::Arc<str>, Value> =
std::collections::HashMap::new();
values.insert("v".into(), val);
QueryRow {
values,
key: RowKey::new(Vec::new()),
metadata: Default::default(),
cell_metadata: None,
}
};
assert_eq!(
evaluate_leaf(&row_of(Value::TinyInt(7)), &in_pred),
LeafOutcome::True
);
assert_eq!(
evaluate_leaf(&row_of(Value::SmallInt(9)), &in_pred),
LeafOutcome::True
);
assert_eq!(
evaluate_leaf(&row_of(Value::Float32(7.0)), &in_pred),
LeafOutcome::True
);
assert_eq!(
evaluate_leaf(&row_of(Value::TinyInt(8)), &in_pred),
LeafOutcome::False
);
}
#[test]
fn two_bound_inequality_slice_selects_half_open_range() {
let predicates = vec![
SSTablePredicate::column("ck", SSTableFilterOp::Gte, vec![Value::Integer(0)]),
SSTablePredicate::column("ck", SSTableFilterOp::Lt, vec![Value::Integer(200)]),
];
let in_slice = |ck: i64| evaluate_predicates(&row_with_int("ck", ck), &predicates).unwrap();
assert!(in_slice(0), "lower bound is inclusive");
assert!(in_slice(199), "last row in [0, 200) is included");
assert!(!in_slice(200), "upper bound is exclusive");
assert!(!in_slice(1000), "rows past the slice are excluded");
assert!(!in_slice(-1), "rows below the slice are excluded");
}
#[test]
fn nan_double_column_dropped_by_inequality_predicates() {
let nan_row = row_with_value("d", Value::Float(f64::NAN));
let gt = |op| SSTablePredicate::column("d", op, vec![Value::Float(1.5)]);
assert_eq!(
evaluate_leaf(&nan_row, >(SSTableFilterOp::Gt)),
LeafOutcome::False
);
assert_eq!(
evaluate_leaf(&nan_row, >(SSTableFilterOp::Gte)),
LeafOutcome::False
);
assert_eq!(
evaluate_leaf(&nan_row, >(SSTableFilterOp::Lt)),
LeafOutcome::False
);
assert_eq!(
evaluate_leaf(&nan_row, >(SSTableFilterOp::Lte)),
LeafOutcome::False
);
let range = SSTablePredicate::column(
"d",
SSTableFilterOp::Range,
vec![Value::Float(0.0), Value::Float(9.0)],
);
assert_eq!(evaluate_leaf(&nan_row, &range), LeafOutcome::False);
let ok_row = row_with_value("d", Value::Float(2.0));
assert_eq!(
evaluate_leaf(&ok_row, >(SSTableFilterOp::Gt)),
LeafOutcome::True
);
}
#[test]
fn large_bigint_equality_no_f64_collapse() {
let eq = |target: i64| {
SSTablePredicate::column(
"bigcol",
SSTableFilterOp::Equal,
vec![Value::BigInt(target)],
)
};
let row_992 = row_with_value("bigcol", Value::BigInt(9_007_199_254_740_992));
assert_eq!(
evaluate_leaf(&row_992, &eq(9_007_199_254_740_993)),
LeafOutcome::False
);
assert_eq!(
evaluate_leaf(&row_992, &eq(9_007_199_254_740_992)),
LeafOutcome::True
);
let in_wrong = SSTablePredicate::column(
"bigcol",
SSTableFilterOp::In,
vec![Value::BigInt(9_007_199_254_740_993)],
);
assert_eq!(evaluate_leaf(&row_992, &in_wrong), LeafOutcome::False);
}
#[test]
fn large_bigint_gt_and_range_no_f64_collapse() {
let row_993 = row_with_value("bigcol", Value::BigInt(9_007_199_254_740_993));
let gt_992 = SSTablePredicate::column(
"bigcol",
SSTableFilterOp::Gt,
vec![Value::BigInt(9_007_199_254_740_992)],
);
assert_eq!(
evaluate_leaf(&row_993, >_992),
LeafOutcome::True,
"9007199254740993 > 9007199254740992 must hold exactly"
);
let gt_993 = SSTablePredicate::column(
"bigcol",
SSTableFilterOp::Gt,
vec![Value::BigInt(9_007_199_254_740_993)],
);
assert_eq!(evaluate_leaf(&row_993, >_993), LeafOutcome::False);
let range_at_992 = SSTablePredicate::column(
"bigcol",
SSTableFilterOp::Range,
vec![
Value::BigInt(9_007_199_254_740_992),
Value::BigInt(9_007_199_254_740_992),
],
);
assert_eq!(evaluate_leaf(&row_993, &range_at_992), LeafOutcome::False);
let range_at_993 = SSTablePredicate::column(
"bigcol",
SSTableFilterOp::Range,
vec![
Value::BigInt(9_007_199_254_740_993),
Value::BigInt(9_007_199_254_740_993),
],
);
assert_eq!(evaluate_leaf(&row_993, &range_at_993), LeafOutcome::True);
}
#[test]
fn validate_token_predicates_accepts_full_key_in_order() {
let schema = composite_pk_schema(("a", "int"), ("b", "int"));
let pred = SSTablePredicate::token(
vec!["a".to_string(), "b".to_string()],
SSTableFilterOp::Gt,
vec![Value::BigInt(0)],
);
assert!(
validate_token_predicates(std::slice::from_ref(&pred), Some(&schema)).is_ok(),
"token(a, b) over the full partition key in declared order must be accepted",
);
let single = single_pk_schema("id", "int");
let pred_single = SSTablePredicate::token(
vec!["id".to_string()],
SSTableFilterOp::Gte,
vec![Value::BigInt(0)],
);
assert!(
validate_token_predicates(std::slice::from_ref(&pred_single), Some(&single)).is_ok(),
"token(id) over a single-column partition key must be accepted",
);
}
#[test]
fn validate_token_predicates_rejects_non_pk_column() {
let schema = single_pk_schema("id", "int");
let pred = SSTablePredicate::token(
vec!["not_the_pk".to_string()],
SSTableFilterOp::Gt,
vec![Value::BigInt(0)],
);
assert!(
validate_token_predicates(std::slice::from_ref(&pred), Some(&schema)).is_err(),
"token(non_pk_col) must be rejected, not evaluated against the real pk token",
);
}
#[test]
fn validate_token_predicates_rejects_reordered_composite() {
let schema = composite_pk_schema(("a", "int"), ("b", "int"));
let reordered = SSTablePredicate::token(
vec!["b".to_string(), "a".to_string()],
SSTableFilterOp::Lt,
vec![Value::BigInt(0)],
);
assert!(
validate_token_predicates(std::slice::from_ref(&reordered), Some(&schema)).is_err(),
"token(b, a) on a (a, b) key must be rejected (wrong order)",
);
let subset = SSTablePredicate::token(
vec!["a".to_string()],
SSTableFilterOp::Lt,
vec![Value::BigInt(0)],
);
assert!(
validate_token_predicates(std::slice::from_ref(&subset), Some(&schema)).is_err(),
"token(a) on a (a, b) key must be rejected (partial key)",
);
}
#[test]
fn validate_token_predicates_rejects_without_schema() {
let pred = SSTablePredicate::token(
vec!["id".to_string()],
SSTableFilterOp::Gt,
vec![Value::BigInt(0)],
);
assert!(
validate_token_predicates(std::slice::from_ref(&pred), None).is_err(),
"a token predicate with no schema must be rejected",
);
let col = SSTablePredicate::column("id", SSTableFilterOp::Equal, vec![Value::Integer(1)]);
assert!(
validate_token_predicates(std::slice::from_ref(&col), None).is_ok(),
"non-token predicates must pass token validation even without a schema",
);
}
}