diff --git a/src/common/recordbatch/src/filter.rs b/src/common/recordbatch/src/filter.rs index 26c28aa3fa0..60ff399601f 100644 --- a/src/common/recordbatch/src/filter.rs +++ b/src/common/recordbatch/src/filter.rs @@ -12,215 +12,174 @@ // See the License for the specific language governing permissions and // limitations under the License. -//! Util record batch stream wrapper that can perform precise filter. +//! Precise single-column predicates and record batch filtering. + +mod predicate; +#[cfg(test)] +mod tests; +mod timestamp; use std::sync::Arc; -use common_time::timestamp::div_mod_units; use datafusion::error::Result as DfResult; use datafusion::logical_expr::{Expr, Literal, Operator}; use datafusion::physical_plan::PhysicalExpr; -use datafusion_common::arrow::array::{ArrayRef, Datum, Scalar}; +use datafusion_common::arrow::array::{ArrayRef, Datum}; use datafusion_common::arrow::buffer::BooleanBuffer; -use datafusion_common::arrow::compute::kernels::cmp; -use datafusion_common::cast::{as_boolean_array, as_null_array, as_string_array}; +use datafusion_common::cast::{as_boolean_array, as_null_array}; use datafusion_common::{DataFusionError, ScalarValue, internal_err}; -use datatypes::arrow::array::{ - Array, ArrayAccessor, ArrayData, BooleanArray, BooleanBufferBuilder, DictionaryArray, - RecordBatch, StringArrayType, -}; +use datatypes::arrow::array::{Array, BooleanArray, BooleanBufferBuilder, RecordBatch}; use datatypes::arrow::compute::filter_record_batch; -use datatypes::arrow::datatypes::{DataType, TimeUnit, UInt32Type}; -use datatypes::arrow::error::ArrowError; -use datatypes::compute::or_kleene; +use datatypes::arrow::datatypes::DataType; use datatypes::data_type::{ConcreteDataType, DataType as _}; use datatypes::value::Value; use datatypes::vectors::VectorRef; -use regex::Regex; use snafu::ResultExt; -use crate::error::{ArrowComputeSnafu, Result, ToArrowScalarSnafu, UnsupportedOperationSnafu}; +pub use self::predicate::{regexp_is_match_dictionary, regexp_is_match_scalar}; +pub use self::timestamp::timestamp_scalar_value; +use crate::error::{Result, ToArrowScalarSnafu}; +use crate::filter::predicate::SimplePredicate; -/// An inplace expr evaluator for simple filter. Only support -/// - `col` `op` `literal` -/// - `literal` `op` `col` -/// -/// And the `op` is one of `=`, `!=`, `>`, `>=`, `<`, `<=`, -/// or regex operators: `~`, `~*`, `!~`, `!~*`. -/// -/// This struct contains normalized predicate expr. In the form of -/// `col` `op` `literal` where the `col` is provided from input. +/// Evaluates supported single-column predicates with precompiled regular expressions. +/// Compound predicates preserve SQL null semantics until conversion to a scan mask. #[derive(Debug, Clone)] pub struct SimpleFilterEvaluator { - /// Name of the referenced column. column_name: String, - /// The literal value. - literal: Scalar, - /// The operator. - op: Operator, - /// Only used when the operator is `Or`-chain. - literal_list: Vec>, - /// Pre-compiled regex. - /// Only used when the operator is regex operators. - /// If the regex is empty, it is also `None`. - regex: Option, - /// Whether the regex is negative. - regex_negative: bool, + predicate: SimplePredicate, } impl SimpleFilterEvaluator { pub fn new(column_name: String, lit: T, op: Operator) -> Option { - match op { + if !matches!( + op, Operator::Eq - | Operator::NotEq - | Operator::Lt - | Operator::LtEq - | Operator::Gt - | Operator::GtEq => {} - _ => return None, + | Operator::NotEq + | Operator::Lt + | Operator::LtEq + | Operator::Gt + | Operator::GtEq + ) { + return None; } - - let Expr::Literal(val, _) = lit.lit() else { + let Expr::Literal(value, _) = lit.lit() else { return None; }; - Some(Self { column_name, - literal: val.to_scalar().ok()?, - op, - literal_list: vec![], - regex: None, - regex_negative: false, + predicate: SimplePredicate::Comparison { + literal: value.to_scalar().ok()?, + op, + regex: None, + regex_negative: false, + }, }) } pub fn try_new(predicate: &Expr) -> Option { - match predicate { - Expr::BinaryExpr(binary) => { - // check if the expr is in the supported form - match binary.op { - Operator::Eq - | Operator::NotEq - | Operator::Lt - | Operator::LtEq - | Operator::Gt - | Operator::GtEq - | Operator::RegexMatch - | Operator::RegexIMatch - | Operator::RegexNotMatch - | Operator::RegexNotIMatch => {} - Operator::Or => { - let lhs = Self::try_new(&binary.left)?; - let rhs = Self::try_new(&binary.right)?; - if lhs.column_name != rhs.column_name - || !matches!(lhs.op, Operator::Eq | Operator::Or) - || !matches!(rhs.op, Operator::Eq | Operator::Or) - { - return None; - } - let mut list = vec![]; - let placeholder_literal = lhs.literal.clone(); - // above check guarantees the op is either `Eq` or `Or` - if matches!(lhs.op, Operator::Or) { - list.extend(lhs.literal_list); - } else { - list.push(lhs.literal); - } - if matches!(rhs.op, Operator::Or) { - list.extend(rhs.literal_list); - } else { - list.push(rhs.literal); - } - return Some(Self { - column_name: lhs.column_name, - literal: placeholder_literal, - op: Operator::Or, - literal_list: list, - regex: None, - regex_negative: false, - }); - } - _ => return None, - } - - // swap the expr if it is in the form of `literal` `op` `col` - let mut op = binary.op; - let (lhs, rhs) = match (&*binary.left, &*binary.right) { - (Expr::Column(col), Expr::Literal(lit, _)) => (col, lit), - (Expr::Literal(lit, _), Expr::Column(col)) => { - // safety: The previous check ensures the operator is able to swap. - op = op.swap().unwrap(); - (col, lit) - } - _ => return None, - }; - - let (regex, regex_negative) = Self::maybe_build_regex(op, rhs).ok()?; - let literal = rhs.to_scalar().ok()?; - Some(Self { - column_name: lhs.name.clone(), - literal, - op, - literal_list: vec![], - regex, - regex_negative, - }) - } - _ => None, - } + Self::try_new_with_column_type(predicate, &|_| None) } - /// Get the name of the referenced column. + /// Recognizes single-column filters, including lossless string dictionary casts + /// when the caller can resolve the column's type. + pub fn try_new_with_column_type( + predicate: &Expr, + column_type: &impl Fn(&str) -> Option, + ) -> Option { + let columns = predicate.column_refs(); + if columns.len() != 1 { + return None; + } + let column_name = columns.into_iter().next()?.name.clone(); + let column_type = column_type(&column_name); + Some(Self { + column_name, + predicate: SimplePredicate::try_new(predicate, column_type.as_ref())?, + }) + } + + /// Returns the referenced column name. pub fn column_name(&self) -> &str { &self.column_name } pub fn is_eq(&self) -> bool { - matches!(self.op, Operator::Eq) + matches!( + self.predicate, + SimplePredicate::Comparison { + op: Operator::Eq, + .. + } + ) } - pub fn is_not_eq(&self) -> bool { - matches!(self.op, Operator::NotEq) + matches!( + self.predicate, + SimplePredicate::Comparison { + op: Operator::NotEq, + .. + } + ) } - pub fn is_lt(&self) -> bool { - matches!(self.op, Operator::Lt) + matches!( + self.predicate, + SimplePredicate::Comparison { + op: Operator::Lt, + .. + } + ) } - pub fn is_lt_eq(&self) -> bool { - matches!(self.op, Operator::LtEq) + matches!( + self.predicate, + SimplePredicate::Comparison { + op: Operator::LtEq, + .. + } + ) } - pub fn is_gt(&self) -> bool { - matches!(self.op, Operator::Gt) + matches!( + self.predicate, + SimplePredicate::Comparison { + op: Operator::Gt, + .. + } + ) } - pub fn is_gt_eq(&self) -> bool { - matches!(self.op, Operator::GtEq) + matches!( + self.predicate, + SimplePredicate::Comparison { + op: Operator::GtEq, + .. + } + ) } - /// Returns true if this filter represents an `OR` chain of equality comparisons, e.g. - /// `col = lit1 OR col = lit2 ...`. + /// Returns true for an OR chain of equality comparisons. pub fn is_or_eq_chain(&self) -> bool { - matches!(self.op, Operator::Or) + matches!(self.predicate, SimplePredicate::OrEqChain { .. }) } - /// Returns the literal as a [`Value`]. It returns `None` if the literal can't be converted. + /// Returns the literal of a bare comparison, without discarding compound conditions. pub fn literal_value(&self) -> Option { - let array = self.literal.get().0; - let scalar = ScalarValue::try_from_array(array, 0).ok()?; - Value::try_from(scalar).ok() + let SimplePredicate::Comparison { literal, .. } = &self.predicate else { + return None; + }; + Value::try_from(ScalarValue::try_from_array(literal.get().0, 0).ok()?).ok() } - /// Returns the literal list as a list of [`Value`]s. It returns `None` if any literal can't be - /// converted. + /// Returns equality-OR literals, or None for other predicates or unsupported values. pub fn literal_list_values(&self) -> Option> { - self.literal_list + let SimplePredicate::OrEqChain { literals } = &self.predicate else { + return None; + }; + literals .iter() - .map(|scalar| { - let array = scalar.get().0; - let scalar = ScalarValue::try_from_array(array, 0).ok()?; - Value::try_from(scalar).ok() + .map(|literal| { + Value::try_from(ScalarValue::try_from_array(literal.get().0, 0).ok()?).ok() }) .collect() } @@ -229,8 +188,7 @@ impl SimpleFilterEvaluator { let input = input .to_scalar() .with_context(|_| ToArrowScalarSnafu { v: input.clone() })?; - let result = self.evaluate_datum(&input, 1)?; - Ok(result.value(0)) + Ok(self.evaluate_datum(&input, 1)?.value(0)) } pub fn evaluate_array(&self, input: &ArrayRef) -> Result { @@ -238,200 +196,27 @@ impl SimpleFilterEvaluator { } pub fn evaluate_vector(&self, input: &VectorRef) -> Result { - self.evaluate_datum(&input.to_arrow_array(), input.len()) + self.evaluate_array(&input.to_arrow_array()) } - fn evaluate_datum(&self, input: &impl Datum, input_len: usize) -> Result { - let result = match self.op { - Operator::Eq => cmp::eq(input, &self.literal), - Operator::NotEq => cmp::neq(input, &self.literal), - Operator::Lt => cmp::lt(input, &self.literal), - Operator::LtEq => cmp::lt_eq(input, &self.literal), - Operator::Gt => cmp::gt(input, &self.literal), - Operator::GtEq => cmp::gt_eq(input, &self.literal), - Operator::RegexMatch => self.regex_match(input), - Operator::RegexIMatch => self.regex_match(input), - Operator::RegexNotMatch => self.regex_match(input), - Operator::RegexNotIMatch => self.regex_match(input), - Operator::Or => { - // OR operator stands for OR-chained EQs (or INLIST in other words) - let mut result: BooleanArray = vec![false; input_len].into(); - for literal in &self.literal_list { - let rhs = cmp::eq(input, literal).context(ArrowComputeSnafu)?; - result = or_kleene(&result, &rhs).context(ArrowComputeSnafu)?; - } - Ok(result) - } - _ => { - return UnsupportedOperationSnafu { - reason: format!("{:?}", self.op), - } - .fail(); - } - }; - result - .context(ArrowComputeSnafu) - .map(|array| boolean_array_to_scan_mask(&array).values().clone()) + fn evaluate_datum(&self, input: &impl Datum, len: usize) -> Result { + let result = self.predicate.evaluate(input, len)?; + Ok(boolean_array_to_scan_mask(&result).values().clone()) } - /// Builds a regex pattern from a scalar value and operator. - /// Returns the `(regex, negative)` and if successful. - /// - /// Returns `Err` if - /// - the value is not a string - /// - the regex pattern is invalid - /// - /// The regex is `None` if - /// - the operator is not a regex operator - /// - the pattern is empty - fn maybe_build_regex( - operator: Operator, - value: &ScalarValue, - ) -> Result<(Option, bool), ArrowError> { - let (ignore_case, negative) = match operator { - Operator::RegexMatch => (false, false), - Operator::RegexIMatch => (true, false), - Operator::RegexNotMatch => (false, true), - Operator::RegexNotIMatch => (true, true), - _ => return Ok((None, false)), - }; - let flag = if ignore_case { Some("i") } else { None }; - let regex = value - .try_as_str() - .ok_or_else(|| ArrowError::CastError(format!("Cannot cast {:?} to str", value)))? - .ok_or_else(|| ArrowError::CastError("Regex should not be null".to_string()))?; - let pattern = match flag { - Some(flag) => format!("(?{flag}){regex}"), - None => regex.to_string(), - }; - if pattern.is_empty() { - Ok((None, negative)) - } else { - Regex::new(pattern.as_str()) - .map_err(|e| { - ArrowError::ComputeError(format!("Regular expression did not compile: {e:?}")) - }) - .map(|regex| (Some(regex), negative)) - } - } - - fn regex_match(&self, input: &impl Datum) -> std::result::Result { - let array = input.get().0; - - // Try to cast to StringArray first - if let Ok(string_array) = as_string_array(array) { - let mut result = regexp_is_match_scalar(string_array, self.regex.as_ref())?; - if self.regex_negative { - result = datatypes::compute::not(&result)?; - } - return Ok(result); - } - - // Try to cast to StringDictionaryArray - if let Some(dict_array) = array.as_any().downcast_ref::>() { - let mut result = regexp_is_match_dictionary(dict_array, self.regex.as_ref())?; - if self.regex_negative { - result = datatypes::compute::not(&result)?; - } - return Ok(result); - } - - Err(ArrowError::CastError(format!( - "Cannot cast {:?} to StringArray or StringDictionaryArray", - array.data_type() - ))) - } - - /// Casts the filter's timestamp literal into the unit of the `target` - /// timestamp type, so it can be evaluated against a column stored in that - /// unit (e.g. an old-unit SST after the time index unit was widened). - /// - /// When the literal is not representable in `target`'s unit (e.g. - /// `ts = 7_000_500us` against a millisecond column), the outcome keeps the - /// row set instead of rounding the literal: `=` prunes / `!=` matches, and - /// inequalities strengthen the operator (e.g. `>= 2_500_500us` becomes - /// `> 2500ms`) so the excluded boundary row stays excluded. - /// - /// Returns `None` when the filter doesn't compare against a tz-naive - /// timestamp literal, or `target` is not a timestamp type. + /// Adapts timestamp predicates to an older SST's timestamp unit without rounding + /// away boundary rows. Nonrepresentable equality literals cannot match a row. pub fn cast_timestamp_unit(&self, target: &ConcreteDataType) -> Option { - let target_unit = match target.as_arrow_type() { - DataType::Timestamp(unit, _) => unit, - _ => return None, + let DataType::Timestamp(target_unit, _) = target.as_arrow_type() else { + return None; }; - - // An `OR` chain of equalities (an IN list): convert each literal - // with `=` semantics; literals not representable in the target unit - // cannot match any row and simply drop out of the chain. - if self.op == Operator::Or { - let mut literal_list = Vec::with_capacity(self.literal_list.len()); - for literal in &self.literal_list { - let scalar = ScalarValue::try_from_array(literal.get().0, 0).ok()?; - let Some((value, unit)) = timestamp_scalar_parts(&scalar) else { - continue; - }; - let Some(value) = value else { - continue; - }; - let cast = div_mod_units(value, unit.into(), target_unit.into())?; - if cast.remainder == 0 { - literal_list.push(timestamp_scalar(cast.quotient, target_unit)?); - } - } - if literal_list.is_empty() { - return Some(TimestampUnitCast::Pruned); - } - let literal = literal_list[0].clone(); - return Some(TimestampUnitCast::Filter(Self { + Some(match self.predicate.cast_timestamp_unit(target_unit)? { + SimplePredicate::Constant(Some(true)) => TimestampUnitCast::Matched, + SimplePredicate::Constant(_) => TimestampUnitCast::Pruned, + predicate => TimestampUnitCast::Filter(Self { column_name: self.column_name.clone(), - literal, - op: Operator::Or, - literal_list, - regex: None, - regex_negative: false, - })); - } - - let scalar = ScalarValue::try_from_array(self.literal.get().0, 0).ok()?; - let (value, unit) = timestamp_scalar_parts(&scalar)?; - let Some(value) = value else { - // A NULL literal never compares true (arrow comparison - // semantics: null results are filtered out). - return Some(TimestampUnitCast::Pruned); - }; - if unit == target_unit { - return Some(TimestampUnitCast::Filter(self.clone())); - } - let cast = div_mod_units(value, unit.into(), target_unit.into())?; - let divisible = cast.remainder == 0; - let literal = timestamp_scalar(cast.quotient, target_unit)?; - let filter = |op: Operator| Self { - column_name: self.column_name.clone(), - literal: literal.clone(), - op, - literal_list: vec![], - regex: None, - regex_negative: false, - }; - - Some(match self.op { - Operator::Eq if divisible => TimestampUnitCast::Filter(filter(Operator::Eq)), - Operator::Eq => TimestampUnitCast::Pruned, - Operator::NotEq if divisible => TimestampUnitCast::Filter(filter(Operator::NotEq)), - Operator::NotEq => TimestampUnitCast::Matched, - // v > L holds exactly for v > quotient, whether or not L is - // representable in the target unit. - Operator::Gt => TimestampUnitCast::Filter(filter(Operator::Gt)), - Operator::GtEq if divisible => TimestampUnitCast::Filter(filter(Operator::GtEq)), - // L strictly between two target values: v >= L is v > quotient. - Operator::GtEq => TimestampUnitCast::Filter(filter(Operator::Gt)), - Operator::Lt if divisible => TimestampUnitCast::Filter(filter(Operator::Lt)), - // L strictly between two target values: v < L is v <= quotient. - Operator::Lt => TimestampUnitCast::Filter(filter(Operator::LtEq)), - // v <= L holds exactly for v <= quotient. - Operator::LtEq => TimestampUnitCast::Filter(filter(Operator::LtEq)), - // Regex predicates don't apply to timestamps. - _ => return None, + predicate, + }), }) } } @@ -449,35 +234,6 @@ pub enum TimestampUnitCast { Matched, } -/// Extracts the value and unit from a tz-naive timestamp scalar. -fn timestamp_scalar_parts(scalar: &ScalarValue) -> Option<(Option, TimeUnit)> { - let (value, unit, timezone) = match scalar { - ScalarValue::TimestampSecond(v, tz) => (*v, TimeUnit::Second, tz), - ScalarValue::TimestampMillisecond(v, tz) => (*v, TimeUnit::Millisecond, tz), - ScalarValue::TimestampMicrosecond(v, tz) => (*v, TimeUnit::Microsecond, tz), - ScalarValue::TimestampNanosecond(v, tz) => (*v, TimeUnit::Nanosecond, tz), - _ => return None, - }; - // A timezone-aware literal doesn't compare against a tz-naive column; - // leave it to the caller instead of guessing the intended semantics. - (timezone.is_none()).then_some((value, unit)) -} - -/// Builds a tz-naive timestamp literal for `value` in `unit`. -pub fn timestamp_scalar_value(value: i64, unit: TimeUnit) -> ScalarValue { - match unit { - TimeUnit::Second => ScalarValue::TimestampSecond(Some(value), None), - TimeUnit::Millisecond => ScalarValue::TimestampMillisecond(Some(value), None), - TimeUnit::Microsecond => ScalarValue::TimestampMicrosecond(Some(value), None), - TimeUnit::Nanosecond => ScalarValue::TimestampNanosecond(Some(value), None), - } -} - -/// Builds a one-element scalar array holding `value` in `unit`. -fn timestamp_scalar(value: i64, unit: TimeUnit) -> Option> { - timestamp_scalar_value(value, unit).to_scalar().ok() -} - /// Evaluate the predicate on the input [RecordBatch], and return a new [RecordBatch]. /// Copy from datafusion::physical_plan::src::filter.rs pub fn batch_filter( @@ -520,726 +276,3 @@ fn boolean_array_to_scan_mask(array: &BooleanArray) -> BooleanArray { } BooleanArray::new(values.into(), None) } - -/// The same as arrow [regexp_is_match_scalar()](datatypes::compute::kernels::regexp::regexp_is_match_scalar()) -/// with pre-compiled regex. -/// See for the implementation details. -pub fn regexp_is_match_scalar<'a, S>( - array: &'a S, - regex: Option<&Regex>, -) -> Result -where - &'a S: StringArrayType<'a>, -{ - let null_bit_buffer = array.nulls().map(|x| x.inner().sliced()); - let mut result = BooleanBufferBuilder::new(array.len()); - - if let Some(re) = regex { - for i in 0..array.len() { - let value = array.value(i); - result.append(re.is_match(value)); - } - } else { - result.append_n(array.len(), true); - } - - let buffer = result.into(); - let data = unsafe { - ArrayData::new_unchecked( - DataType::Boolean, - array.len(), - None, - null_bit_buffer, - 0, - vec![buffer], - vec![], - ) - }; - - Ok(BooleanArray::from(data)) -} - -/// Similar to [regexp_is_match_scalar] but for StringDictionaryArray. -/// Iterates through dictionary keys to get string values and applies regex matching. -pub fn regexp_is_match_dictionary( - dict_array: &DictionaryArray, - regex: Option<&Regex>, -) -> Result { - // Get the string values from the dictionary - let string_values = dict_array - .values() - .as_any() - .downcast_ref::() - .ok_or_else(|| { - ArrowError::CastError("Dictionary values must be StringArray".to_string()) - })?; - - // Dictionary logical nulls include both null keys and keys whose dictionary value is null. - let logical_nulls = dict_array.logical_nulls(); - let null_bit_buffer = logical_nulls.as_ref().map(|x| x.inner().sliced()); - let mut result = BooleanBufferBuilder::new(dict_array.len()); - - if let Some(re) = regex { - let keys = dict_array.keys().values(); - for i in 0..dict_array.len() { - if logical_nulls.as_ref().is_some_and(|nulls| nulls.is_null(i)) { - result.append(false); - } else { - let key = keys[i] as usize; - let string_value = string_values.value(key); - result.append(re.is_match(string_value)); - } - } - } else { - result.append_n(dict_array.len(), true); - } - - let buffer = result.into(); - let data = unsafe { - ArrayData::new_unchecked( - DataType::Boolean, - dict_array.len(), - None, - null_bit_buffer, - 0, - vec![buffer], - vec![], - ) - }; - - Ok(BooleanArray::from(data)) -} - -#[cfg(test)] -mod test { - - use std::sync::Arc; - - use datafusion::execution::context::ExecutionProps; - use datafusion::logical_expr::physical_planning_context::PhysicalPlanningContext; - use datafusion::logical_expr::{BinaryExpr, col, lit}; - use datafusion::physical_expr::create_physical_expr; - use datafusion_common::{Column, DFSchema}; - use datatypes::arrow::array::{TimestampMillisecondArray, TimestampNanosecondArray}; - use datatypes::arrow::datatypes::{DataType, Field, Schema}; - - use super::*; - - #[test] - fn unsupported_filter_op() { - // `+` is not supported - let expr = Expr::BinaryExpr(BinaryExpr { - left: Box::new(Expr::Column(Column::from_name("foo"))), - op: Operator::Plus, - right: Box::new(1.lit()), - }); - assert!(SimpleFilterEvaluator::try_new(&expr).is_none()); - - // two literal is not supported - let expr = Expr::BinaryExpr(BinaryExpr { - left: Box::new(1.lit()), - op: Operator::Eq, - right: Box::new(1.lit()), - }); - assert!(SimpleFilterEvaluator::try_new(&expr).is_none()); - - // two column is not supported - let expr = Expr::BinaryExpr(BinaryExpr { - left: Box::new(Expr::Column(Column::from_name("foo"))), - op: Operator::Eq, - right: Box::new(Expr::Column(Column::from_name("bar"))), - }); - assert!(SimpleFilterEvaluator::try_new(&expr).is_none()); - - // compound expr is not supported - let expr = Expr::BinaryExpr(BinaryExpr { - left: Box::new(Expr::BinaryExpr(BinaryExpr { - left: Box::new(Expr::Column(Column::from_name("foo"))), - op: Operator::Eq, - right: Box::new(1.lit()), - })), - op: Operator::Eq, - right: Box::new(1.lit()), - }); - assert!(SimpleFilterEvaluator::try_new(&expr).is_none()); - } - - #[test] - fn supported_filter_op() { - // equal - let expr = Expr::BinaryExpr(BinaryExpr { - left: Box::new(Expr::Column(Column::from_name("foo"))), - op: Operator::Eq, - right: Box::new(1.lit()), - }); - let _ = SimpleFilterEvaluator::try_new(&expr).unwrap(); - - // swap operands - let expr = Expr::BinaryExpr(BinaryExpr { - left: Box::new(1.lit()), - op: Operator::Lt, - right: Box::new(Expr::Column(Column::from_name("foo"))), - }); - let evaluator = SimpleFilterEvaluator::try_new(&expr).unwrap(); - assert_eq!(evaluator.op, Operator::Gt); - assert_eq!(evaluator.column_name, "foo".to_string()); - } - - #[test] - fn run_on_array() { - let expr = Expr::BinaryExpr(BinaryExpr { - left: Box::new(Expr::Column(Column::from_name("foo"))), - op: Operator::Eq, - right: Box::new(1i64.lit()), - }); - let evaluator = SimpleFilterEvaluator::try_new(&expr).unwrap(); - - let input_1 = Arc::new(datatypes::arrow::array::Int64Array::from(vec![1, 2, 3])) as _; - let result = evaluator.evaluate_array(&input_1).unwrap(); - assert_eq!(result, BooleanBuffer::from(vec![true, false, false])); - - let input_2 = Arc::new(datatypes::arrow::array::Int64Array::from(vec![1, 1, 1])) as _; - let result = evaluator.evaluate_array(&input_2).unwrap(); - assert_eq!(result, BooleanBuffer::from(vec![true, true, true])); - - let input_3 = Arc::new(datatypes::arrow::array::Int64Array::new_null(0)) as _; - let result = evaluator.evaluate_array(&input_3).unwrap(); - assert_eq!(result, BooleanBuffer::from(vec![])); - } - - #[test] - fn run_on_scalar() { - let expr = Expr::BinaryExpr(BinaryExpr { - left: Box::new(Expr::Column(Column::from_name("foo"))), - op: Operator::Lt, - right: Box::new(1i64.lit()), - }); - let evaluator = SimpleFilterEvaluator::try_new(&expr).unwrap(); - - let input_1 = ScalarValue::Int64(Some(1)); - let result = evaluator.evaluate_scalar(&input_1).unwrap(); - assert!(!result); - - let input_2 = ScalarValue::Int64(Some(0)); - let result = evaluator.evaluate_scalar(&input_2).unwrap(); - assert!(result); - - let input_3 = ScalarValue::Int64(None); - let result = evaluator.evaluate_scalar(&input_3).unwrap(); - assert!(!result); - } - - #[test] - fn batch_filter_test() { - let expr = col("ts").gt(lit(123456u64)); - let schema = Schema::new(vec![ - Field::new("a", DataType::Int32, true), - Field::new("ts", DataType::UInt64, false), - ]); - let df_schema = DFSchema::try_from(schema.clone()).unwrap(); - let props = ExecutionProps::new(); - let physical_expr = create_physical_expr( - &expr, - &df_schema, - &props, - &PhysicalPlanningContext::default(), - ) - .unwrap(); - let batch = RecordBatch::try_new( - Arc::new(schema), - vec![ - Arc::new(datatypes::arrow::array::Int32Array::from(vec![4, 5, 6])), - Arc::new(datatypes::arrow::array::UInt64Array::from(vec![ - 123456, 123457, 123458, - ])), - ], - ) - .unwrap(); - let new_batch = batch_filter(&batch, &physical_expr).unwrap(); - assert_eq!(new_batch.num_rows(), 2); - let first_column_values = new_batch - .column(0) - .as_any() - .downcast_ref::() - .unwrap(); - let expected = datatypes::arrow::array::Int32Array::from(vec![5, 6]); - assert_eq!(first_column_values, &expected); - } - - #[test] - fn test_complex_filter_expression() { - // Create an expression tree for: col = 'B' OR col = 'C' OR col = 'D' - let col_eq_b = col("col").eq(lit("B")); - let col_eq_c = col("col").eq(lit("C")); - let col_eq_d = col("col").eq(lit("D")); - - // Build the OR chain - let col_or_expr = col_eq_b.or(col_eq_c).or(col_eq_d); - - // Check that SimpleFilterEvaluator can handle OR chain - let or_evaluator = SimpleFilterEvaluator::try_new(&col_or_expr).unwrap(); - assert_eq!(or_evaluator.column_name, "col"); - assert_eq!(or_evaluator.op, Operator::Or); - assert_eq!(or_evaluator.literal_list.len(), 3); - assert_eq!( - format!("{:?}", or_evaluator.literal_list), - "[Scalar(StringArray\n[\n \"B\",\n]), Scalar(StringArray\n[\n \"C\",\n]), Scalar(StringArray\n[\n \"D\",\n])]" - ); - - // Create a schema and batch for testing - let schema = Schema::new(vec![Field::new("col", DataType::Utf8, false)]); - let df_schema = DFSchema::try_from(schema.clone()).unwrap(); - let props = ExecutionProps::new(); - let physical_expr = create_physical_expr( - &col_or_expr, - &df_schema, - &props, - &PhysicalPlanningContext::default(), - ) - .unwrap(); - - // Create test data - let col_data = Arc::new(datatypes::arrow::array::StringArray::from(vec![ - "B", "C", "E", "B", "C", "D", "F", - ])); - let batch = RecordBatch::try_new(Arc::new(schema), vec![col_data]).unwrap(); - let expected = datatypes::arrow::array::StringArray::from(vec!["B", "C", "B", "C", "D"]); - - // Filter the batch - let filtered_batch = batch_filter(&batch, &physical_expr).unwrap(); - - // Expected: rows with col in ("B", "C", "D") - // That would be rows 0, 1, 3, 4, 5 - assert_eq!(filtered_batch.num_rows(), 5); - - let col_filtered = filtered_batch - .column(0) - .as_any() - .downcast_ref::() - .unwrap(); - assert_eq!(col_filtered, &expected); - } - - #[test] - fn test_maybe_build_regex() { - // Test case for RegexMatch (case sensitive, non-negative) - let (regex, negative) = SimpleFilterEvaluator::maybe_build_regex( - Operator::RegexMatch, - &ScalarValue::Utf8(Some("a.*b".to_string())), - ) - .unwrap(); - assert!(regex.is_some()); - assert!(!negative); - assert!(regex.unwrap().is_match("axxb")); - - // Test case for RegexIMatch (case insensitive, non-negative) - let (regex, negative) = SimpleFilterEvaluator::maybe_build_regex( - Operator::RegexIMatch, - &ScalarValue::Utf8(Some("a.*b".to_string())), - ) - .unwrap(); - assert!(regex.is_some()); - assert!(!negative); - assert!(regex.unwrap().is_match("AxxB")); - - // Test case for RegexNotMatch (case sensitive, negative) - let (regex, negative) = SimpleFilterEvaluator::maybe_build_regex( - Operator::RegexNotMatch, - &ScalarValue::Utf8(Some("a.*b".to_string())), - ) - .unwrap(); - assert!(regex.is_some()); - assert!(negative); - - // Test case for RegexNotIMatch (case insensitive, negative) - let (regex, negative) = SimpleFilterEvaluator::maybe_build_regex( - Operator::RegexNotIMatch, - &ScalarValue::Utf8(Some("a.*b".to_string())), - ) - .unwrap(); - assert!(regex.is_some()); - assert!(negative); - - // Test with empty regex pattern - let (regex, negative) = SimpleFilterEvaluator::maybe_build_regex( - Operator::RegexMatch, - &ScalarValue::Utf8(Some("".to_string())), - ) - .unwrap(); - assert!(regex.is_none()); - assert!(!negative); - - // Test with non-regex operator - let (regex, negative) = SimpleFilterEvaluator::maybe_build_regex( - Operator::Eq, - &ScalarValue::Utf8(Some("a.*b".to_string())), - ) - .unwrap(); - assert!(regex.is_none()); - assert!(!negative); - - // Test with invalid regex pattern - let result = SimpleFilterEvaluator::maybe_build_regex( - Operator::RegexMatch, - &ScalarValue::Utf8(Some("a(b".to_string())), - ); - assert!(result.is_err()); - - // Test with non-string value - let result = SimpleFilterEvaluator::maybe_build_regex( - Operator::RegexMatch, - &ScalarValue::Int64(Some(123)), - ); - assert!(result.is_err()); - - // Test with null value - let result = SimpleFilterEvaluator::maybe_build_regex( - Operator::RegexMatch, - &ScalarValue::Utf8(None), - ); - assert!(result.is_err()); - } - - #[test] - fn test_regex_match_dictionary_array() { - use datatypes::arrow::array::StringDictionaryBuilder; - - // Create a StringDictionaryArray - let mut builder = StringDictionaryBuilder::::new(); - builder.append("apple").unwrap(); - builder.append("banana").unwrap(); - builder.append("apple").unwrap(); - builder.append("cherry").unwrap(); - let dict_array = builder.finish(); - - // Test regex that matches "apple" - let regex = regex::Regex::new(r"app.*").unwrap(); - let result = regexp_is_match_dictionary(&dict_array, Some(®ex)).unwrap(); - - // Should match indices 0 and 2 (both "apple") - assert_eq!(result.len(), 4); - assert!(result.value(0)); // "apple" - assert!(!result.value(1)); // "banana" - assert!(result.value(2)); // "apple" - assert!(!result.value(3)); // "cherry" - - // Test regex that matches "banana" - let regex2 = regex::Regex::new(r"ban.*").unwrap(); - let result2 = regexp_is_match_dictionary(&dict_array, Some(®ex2)).unwrap(); - - assert!(!result2.value(0)); // "apple" - assert!(result2.value(1)); // "banana" - assert!(!result2.value(2)); // "apple" - assert!(!result2.value(3)); // "cherry" - - // Test with no regex (should match all) - let result3 = regexp_is_match_dictionary(&dict_array, None).unwrap(); - assert!(result3.value(0)); - assert!(result3.value(1)); - assert!(result3.value(2)); - assert!(result3.value(3)); - } - - #[test] - fn test_regex_scan_masks_preserve_sql_null_semantics() { - let plain = Arc::new(datatypes::arrow::array::StringArray::from(vec![ - Some("api"), - Some("API"), - Some("db"), - None, - ])) as ArrayRef; - assert_regex_scan_masks( - &plain, - [ - vec![true, false, false, false], - vec![true, true, false, false], - vec![false, true, true, false], - vec![false, false, true, false], - ], - ); - - let dictionary = DictionaryArray::new( - datatypes::arrow::array::UInt32Array::from(vec![ - Some(0), - Some(1), - Some(2), - None, - Some(3), - ]), - Arc::new(datatypes::arrow::array::StringArray::from(vec![ - Some("api"), - Some("API"), - Some("db"), - None, - ])), - ); - let raw = - regexp_is_match_dictionary(&dictionary, Some(&Regex::new("^api$").unwrap())).unwrap(); - assert!(raw.is_null(3)); // null dictionary key - assert!(raw.is_null(4)); // non-null key referencing a null dictionary value - - let dictionary = Arc::new(dictionary) as ArrayRef; - assert_regex_scan_masks( - &dictionary, - [ - vec![true, false, false, false, false], - vec![true, true, false, false, false], - vec![false, true, true, false, false], - vec![false, false, true, false, false], - ], - ); - - let negative = regex_evaluator(Operator::RegexNotMatch); - assert!(!negative.evaluate_scalar(&ScalarValue::Utf8(None)).unwrap()); - } - - #[test] - fn test_nullable_boolean_predicate_becomes_scan_mask() { - let predicate = BooleanArray::from(vec![Some(true), None, Some(false)]); - assert_eq!( - BooleanArray::from(vec![true, false, false]), - boolean_array_to_scan_mask(&predicate) - ); - } - - fn assert_regex_scan_masks(input: &ArrayRef, expected: [Vec; 4]) { - for (op, expected) in [ - Operator::RegexMatch, - Operator::RegexIMatch, - Operator::RegexNotMatch, - Operator::RegexNotIMatch, - ] - .into_iter() - .zip(expected) - { - assert_eq!( - BooleanBuffer::from(expected), - regex_evaluator(op).evaluate_array(input).unwrap(), - "{op:?}" - ); - } - } - - fn regex_evaluator(op: Operator) -> SimpleFilterEvaluator { - let expr = Expr::BinaryExpr(BinaryExpr { - left: Box::new(Expr::Column(Column::from_name("host"))), - op, - right: Box::new("^api$".lit()), - }); - SimpleFilterEvaluator::try_new(&expr).unwrap() - } - - fn ts_us(v: i64) -> ScalarValue { - ScalarValue::TimestampMicrosecond(Some(v), None) - } - - fn cast_to_ms(expr: &Expr) -> Option { - SimpleFilterEvaluator::try_new(expr) - .unwrap() - .cast_timestamp_unit(&ConcreteDataType::timestamp_millisecond_datatype()) - } - - /// Evaluates a cast `Filter` outcome against millisecond values and - /// returns the mask. - fn eval_ms_mask(cast: Option, values: &[i64]) -> Vec { - let filter = match cast.expect("cast must apply") { - TimestampUnitCast::Filter(filter) => filter, - other => panic!("expected Filter outcome, got {other:?}"), - }; - let array = Arc::new(TimestampMillisecondArray::from(values.to_vec())) as ArrayRef; - filter.evaluate_array(&array).unwrap().iter().collect() - } - - #[test] - fn cast_timestamp_unit_converts_representable_literal() { - // ts = 7_000_000us evaluated against a ms column becomes ts = 7000ms. - assert_eq!( - vec![false, true, false], - eval_ms_mask( - cast_to_ms(&col("ts").eq(lit(ts_us(7_000_000)))), - &[6_999, 7_000, 7_001] - ) - ); - // != complements =. - assert_eq!( - vec![true, false, true], - eval_ms_mask( - cast_to_ms(&col("ts").not_eq(lit(ts_us(7_000_000)))), - &[6_999, 7_000, 7_001] - ) - ); - } - - #[test] - fn cast_timestamp_unit_prunes_non_representable_equality() { - // 7_000_500us is not a whole millisecond: no ms row can equal it, - // and every ms row satisfies `!=`. - assert!(matches!( - cast_to_ms(&col("ts").eq(lit(ts_us(7_000_500)))), - Some(TimestampUnitCast::Pruned) - )); - assert!(matches!( - cast_to_ms(&col("ts").not_eq(lit(ts_us(7_000_500)))), - Some(TimestampUnitCast::Matched) - )); - } - - #[test] - fn cast_timestamp_unit_strengthens_inequalities() { - // 2_500_500us is strictly between 2500ms and 2501ms: the cast must - // not round the literal to a boundary the original predicate - // excludes (>= 2500ms would wrongly match 2500ms). - let values = vec![2_499, 2_500, 2_501]; - assert_eq!( - vec![false, false, true], - eval_ms_mask(cast_to_ms(&col("ts").gt(lit(ts_us(2_500_500)))), &values) - ); - assert_eq!( - vec![false, false, true], - eval_ms_mask(cast_to_ms(&col("ts").gt_eq(lit(ts_us(2_500_500)))), &values) - ); - assert_eq!( - vec![true, true, false], - eval_ms_mask(cast_to_ms(&col("ts").lt(lit(ts_us(2_500_500)))), &values) - ); - assert_eq!( - vec![true, true, false], - eval_ms_mask(cast_to_ms(&col("ts").lt_eq(lit(ts_us(2_500_500)))), &values) - ); - - // Representable boundary keeps the original operator. - let values = vec![2_499, 2_500, 2_501]; - assert_eq!( - vec![false, false, true], - eval_ms_mask(cast_to_ms(&col("ts").gt(lit(ts_us(2_500_000)))), &values) - ); - assert_eq!( - vec![false, true, true], - eval_ms_mask(cast_to_ms(&col("ts").gt_eq(lit(ts_us(2_500_000)))), &values) - ); - assert_eq!( - vec![true, false, false], - eval_ms_mask(cast_to_ms(&col("ts").lt(lit(ts_us(2_500_000)))), &values) - ); - assert_eq!( - vec![true, true, false], - eval_ms_mask(cast_to_ms(&col("ts").lt_eq(lit(ts_us(2_500_000)))), &values) - ); - } - - #[test] - fn cast_timestamp_unit_handles_negative_instants() { - // -2_500_500us floors to -2501ms with remainder 500us: only rows - // with instant >= -2.5005ms may match >= / >. - let values = vec![-2_502, -2_501, -2_500]; - assert_eq!( - vec![false, false, true], - eval_ms_mask( - cast_to_ms(&col("ts").gt_eq(lit(ts_us(-2_500_500)))), - &values - ) - ); - assert_eq!( - vec![false, false, true], - eval_ms_mask(cast_to_ms(&col("ts").gt(lit(ts_us(-2_500_500)))), &values) - ); - assert_eq!( - vec![true, true, false], - eval_ms_mask( - cast_to_ms(&col("ts").lt_eq(lit(ts_us(-2_500_500)))), - &values - ) - ); - // Exactly representable negative literal: -2_500_000us == -2500ms. - assert_eq!( - vec![false, false, true], - eval_ms_mask(cast_to_ms(&col("ts").eq(lit(ts_us(-2_500_000)))), &values) - ); - } - - #[test] - fn cast_timestamp_unit_or_chain_drops_unrepresentable_literals() { - // Only 7_000_000us is a whole millisecond; 7_000_500us cannot match. - let expr = col("ts") - .eq(lit(ts_us(7_000_500))) - .or(col("ts").eq(lit(ts_us(7_000_000)))); - assert_eq!( - vec![false, true], - eval_ms_mask(cast_to_ms(&expr), &[6_999, 7_000]) - ); - - // A chain where no literal is representable matches nothing. - let expr = col("ts") - .eq(lit(ts_us(7_000_500))) - .or(col("ts").eq(lit(ts_us(6_000_500)))); - assert!(matches!(cast_to_ms(&expr), Some(TimestampUnitCast::Pruned))); - } - - #[test] - fn cast_timestamp_unit_null_literal_prunes() { - // A NULL literal never compares true. - assert!(matches!( - cast_to_ms(&col("ts").eq(lit(ScalarValue::TimestampMicrosecond(None, None)))), - Some(TimestampUnitCast::Pruned) - )); - } - - #[test] - fn cast_timestamp_unit_same_unit_returns_filter_directly() { - // A literal already in the target unit is returned unchanged. - let expr = col("ts").eq(lit(ts_us(7_000_000))); - let filter = SimpleFilterEvaluator::try_new(&expr).unwrap(); - let cast = filter - .cast_timestamp_unit(&ConcreteDataType::timestamp_microsecond_datatype()) - .unwrap(); - let TimestampUnitCast::Filter(f) = cast else { - panic!("expected Filter, got {cast:?}") - }; - assert_eq!(filter.op, f.op); - assert_eq!(filter.column_name(), f.column_name()); - assert_eq!( - ts_us(7_000_000), - ScalarValue::try_from_array(f.literal.get().0, 0).unwrap() - ); - } - - #[test] - fn cast_timestamp_unit_rejects_non_timestamps() { - // Non-timestamp target type. - assert!( - SimpleFilterEvaluator::try_new(&col("ts").gt(lit(ts_us(1)))) - .unwrap() - .cast_timestamp_unit(&ConcreteDataType::int64_datatype()) - .is_none() - ); - // Non-timestamp literal against a timestamp target. - assert!(cast_to_ms(&col("ts").gt(lit(42_i64))).is_none()); - } - - #[test] - fn cast_timestamp_unit_to_finer_unit() { - // 5 seconds against a nanosecond column becomes 5_000_000_000ns. - let filter = SimpleFilterEvaluator::try_new( - &col("ts").eq(lit(ScalarValue::TimestampSecond(Some(5), None))), - ) - .unwrap() - .cast_timestamp_unit(&ConcreteDataType::timestamp_nanosecond_datatype()) - .and_then(|cast| match cast { - TimestampUnitCast::Filter(filter) => Some(filter), - _ => None, - }) - .unwrap(); - let array = Arc::new(TimestampNanosecondArray::from(vec![ - 4_999_999_999, - 5_000_000_000, - ])) as ArrayRef; - assert_eq!( - vec![false, true], - filter - .evaluate_array(&array) - .unwrap() - .iter() - .collect::>() - ); - } -} diff --git a/src/common/recordbatch/src/filter/predicate.rs b/src/common/recordbatch/src/filter/predicate.rs new file mode 100644 index 00000000000..e283633dc02 --- /dev/null +++ b/src/common/recordbatch/src/filter/predicate.rs @@ -0,0 +1,944 @@ +// Copyright 2023 Greptime Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! Predicate recognition, execution, and precompiled regular expressions. + +use datafusion::logical_expr::{Expr, Operator}; +use datafusion_common::ScalarValue; +use datafusion_common::arrow::array::{ArrayRef, Datum, Scalar}; +use datafusion_common::arrow::buffer::BooleanBuffer; +use datafusion_common::arrow::compute::kernels::cmp; +use datafusion_common::cast::as_string_array; +use datatypes::arrow::array::{ + Array, ArrayAccessor, ArrayData, BooleanArray, BooleanBufferBuilder, DictionaryArray, + StringArrayType, +}; +use datatypes::arrow::datatypes::{DataType, UInt32Type}; +use datatypes::arrow::error::ArrowError; +use datatypes::compute::or_kleene; +use regex::Regex; +use snafu::ResultExt; + +use crate::error::{ArrowComputeSnafu, Result, UnsupportedOperationSnafu}; + +#[derive(Debug, Clone)] +pub(crate) enum SimplePredicate { + Comparison { + literal: Scalar, + op: Operator, + regex: Option, + regex_negative: bool, + }, + OrEqChain { + literals: Vec>, + }, + IsNull { + negated: bool, + }, + Constant(Option), + And(Box, Box), + Or(Box, Box), +} + +impl SimplePredicate { + pub(crate) fn try_new(expr: &Expr, column_type: Option<&DataType>) -> Option { + match expr { + Expr::BinaryExpr(binary) if matches!(binary.op, Operator::And | Operator::Or) => { + let left = Self::try_new(&binary.left, column_type)?; + let right = Self::try_new(&binary.right, column_type)?; + if binary.op == Operator::And { + return Some(Self::And(Box::new(left), Box::new(right))); + } + // Preserve the encoded-key IN-list fast path for pure equality disjunctions. + if let (Some(mut lhs), Some(rhs)) = + (left.equality_literals(), right.equality_literals()) + { + lhs.extend(rhs); + return Some(Self::OrEqChain { literals: lhs }); + } + Some(Self::Or(Box::new(left), Box::new(right))) + } + Expr::BinaryExpr(binary) => { + let mut op = binary.op; + if !matches!( + op, + Operator::Eq + | Operator::NotEq + | Operator::Lt + | Operator::LtEq + | Operator::Gt + | Operator::GtEq + | Operator::RegexMatch + | Operator::RegexIMatch + | Operator::RegexNotMatch + | Operator::RegexNotIMatch + ) { + return None; + } + let literal = if let Expr::Literal(literal, _) = &*binary.right { + Self::check_column(&binary.left, column_type)?; + literal + } else if let Expr::Literal(literal, _) = &*binary.left { + Self::check_column(&binary.right, column_type)?; + op = op.swap()?; + literal + } else { + return None; + }; + let (regex, regex_negative) = Self::maybe_build_regex(op, literal).ok()?; + Some(Self::Comparison { + literal: literal.to_scalar().ok()?, + op, + regex, + regex_negative, + }) + } + Expr::IsNull(column) | Expr::IsNotNull(column) => { + Self::check_column(column, column_type)?; + Some(Self::IsNull { + negated: matches!(expr, Expr::IsNotNull(_)), + }) + } + Expr::Literal(ScalarValue::Boolean(value), _) => Some(Self::Constant(*value)), + _ => None, + } + } + + fn check_column(expr: &Expr, column_type: Option<&DataType>) -> Option<()> { + match expr { + Expr::Column(_) => Some(()), + Expr::Cast(cast) if cast.field.data_type() == &DataType::Utf8 => { + let Expr::Column(_) = &*cast.expr else { + return None; + }; + match column_type? { + DataType::Utf8 => Some(()), + DataType::Dictionary(key, value) + if **key == DataType::UInt32 && **value == DataType::Utf8 => + { + Some(()) + } + _ => None, + } + } + _ => None, + } + } + + fn equality_literals(&self) -> Option>> { + match self { + Self::Comparison { + literal, + op: Operator::Eq, + .. + } => Some(vec![literal.clone()]), + Self::OrEqChain { literals } => Some(literals.clone()), + _ => None, + } + } + + pub(crate) fn evaluate(&self, input: &impl Datum, len: usize) -> Result { + match self { + Self::Comparison { + literal, + op, + regex, + regex_negative, + } => match op { + Operator::Eq => cmp::eq(input, literal), + Operator::NotEq => cmp::neq(input, literal), + Operator::Lt => cmp::lt(input, literal), + Operator::LtEq => cmp::lt_eq(input, literal), + Operator::Gt => cmp::gt(input, literal), + Operator::GtEq => cmp::gt_eq(input, literal), + Operator::RegexMatch + | Operator::RegexIMatch + | Operator::RegexNotMatch + | Operator::RegexNotIMatch => { + Self::regex_match(input, regex.as_ref(), *regex_negative) + } + _ => { + return UnsupportedOperationSnafu { + reason: format!("{op:?}"), + } + .fail(); + } + } + .context(ArrowComputeSnafu), + Self::OrEqChain { literals } => { + let mut result = BooleanArray::from(vec![false; len]); + for literal in literals { + let rhs = cmp::eq(input, literal).context(ArrowComputeSnafu)?; + result = or_kleene(&result, &rhs).context(ArrowComputeSnafu)?; + } + Ok(result) + } + Self::IsNull { negated } => { + let values = match input.get().0.logical_nulls() { + Some(nulls) if *negated => nulls.inner().clone(), + Some(nulls) => !nulls.inner(), + None if *negated => BooleanBuffer::new_set(len), + None => BooleanBuffer::new_unset(len), + }; + Ok(BooleanArray::new(values, None)) + } + Self::Constant(value) => Ok(BooleanArray::from(vec![*value; len])), + Self::And(left, right) => datatypes::compute::and_kleene( + &left.evaluate(input, len)?, + &right.evaluate(input, len)?, + ) + .context(ArrowComputeSnafu), + Self::Or(left, right) => { + or_kleene(&left.evaluate(input, len)?, &right.evaluate(input, len)?) + .context(ArrowComputeSnafu) + } + } + } + + /// Builds a regex pattern from a scalar value and operator. + /// Returns the `(regex, negative)` and if successful. + /// + /// Returns `Err` if + /// - the value is not a string + /// - the regex pattern is invalid + /// + /// The regex is `None` if + /// - the operator is not a regex operator + /// - the pattern is empty + fn maybe_build_regex( + operator: Operator, + value: &ScalarValue, + ) -> Result<(Option, bool), ArrowError> { + let (ignore_case, negative) = match operator { + Operator::RegexMatch => (false, false), + Operator::RegexIMatch => (true, false), + Operator::RegexNotMatch => (false, true), + Operator::RegexNotIMatch => (true, true), + _ => return Ok((None, false)), + }; + let flag = if ignore_case { Some("i") } else { None }; + let regex = value + .try_as_str() + .ok_or_else(|| ArrowError::CastError(format!("Cannot cast {:?} to str", value)))? + .ok_or_else(|| ArrowError::CastError("Regex should not be null".to_string()))?; + let pattern = match flag { + Some(flag) => format!("(?{flag}){regex}"), + None => regex.to_string(), + }; + if pattern.is_empty() { + Ok((None, negative)) + } else { + Regex::new(pattern.as_str()) + .map_err(|e| { + ArrowError::ComputeError(format!("Regular expression did not compile: {e:?}")) + }) + .map(|regex| (Some(regex), negative)) + } + } + + fn regex_match( + input: &impl Datum, + regex: Option<&Regex>, + negative: bool, + ) -> std::result::Result { + let array = input.get().0; + + // Try to cast to StringArray first + if let Ok(string_array) = as_string_array(array) { + let mut result = regexp_is_match_scalar(string_array, regex)?; + if negative { + result = datatypes::compute::not(&result)?; + } + return Ok(result); + } + + // Try to cast to StringDictionaryArray + if let Some(dict_array) = array.as_any().downcast_ref::>() { + let mut result = regexp_is_match_dictionary(dict_array, regex)?; + if negative { + result = datatypes::compute::not(&result)?; + } + return Ok(result); + } + + Err(ArrowError::CastError(format!( + "Cannot cast {:?} to StringArray or StringDictionaryArray", + array.data_type() + ))) + } +} + +/// The same as arrow [regexp_is_match_scalar()](datatypes::compute::kernels::regexp::regexp_is_match_scalar()) +/// with pre-compiled regex. +/// See for the implementation details. +pub fn regexp_is_match_scalar<'a, S>( + array: &'a S, + regex: Option<&Regex>, +) -> Result +where + &'a S: StringArrayType<'a>, +{ + let null_bit_buffer = array.nulls().map(|x| x.inner().sliced()); + let mut result = BooleanBufferBuilder::new(array.len()); + + if let Some(re) = regex { + for i in 0..array.len() { + let value = array.value(i); + result.append(re.is_match(value)); + } + } else { + result.append_n(array.len(), true); + } + + let buffer = result.into(); + let data = unsafe { + ArrayData::new_unchecked( + DataType::Boolean, + array.len(), + None, + null_bit_buffer, + 0, + vec![buffer], + vec![], + ) + }; + + Ok(BooleanArray::from(data)) +} + +/// Similar to [regexp_is_match_scalar] but for StringDictionaryArray. +/// Iterates through dictionary keys to get string values and applies regex matching. +pub fn regexp_is_match_dictionary( + dict_array: &DictionaryArray, + regex: Option<&Regex>, +) -> Result { + // Get the string values from the dictionary + let string_values = dict_array + .values() + .as_any() + .downcast_ref::() + .ok_or_else(|| { + ArrowError::CastError("Dictionary values must be StringArray".to_string()) + })?; + + // Dictionary logical nulls include both null keys and keys whose dictionary value is null. + let logical_nulls = dict_array.logical_nulls(); + let null_bit_buffer = logical_nulls.as_ref().map(|x| x.inner().sliced()); + let mut result = BooleanBufferBuilder::new(dict_array.len()); + + if let Some(re) = regex { + let keys = dict_array.keys().values(); + for i in 0..dict_array.len() { + if logical_nulls.as_ref().is_some_and(|nulls| nulls.is_null(i)) { + result.append(false); + } else { + let key = keys[i] as usize; + let string_value = string_values.value(key); + result.append(re.is_match(string_value)); + } + } + } else { + result.append_n(dict_array.len(), true); + } + + let buffer = result.into(); + let data = unsafe { + ArrayData::new_unchecked( + DataType::Boolean, + dict_array.len(), + None, + null_bit_buffer, + 0, + vec![buffer], + vec![], + ) + }; + + Ok(BooleanArray::from(data)) +} + +#[cfg(test)] +mod tests { + use std::sync::Arc; + + use datafusion::execution::context::ExecutionProps; + use datafusion::logical_expr::physical_planning_context::PhysicalPlanningContext; + use datafusion::logical_expr::{BinaryExpr, Expr, Literal, Operator, col, lit}; + use datafusion::physical_expr::create_physical_expr; + use datafusion_common::arrow::array::ArrayRef; + use datafusion_common::arrow::buffer::BooleanBuffer; + use datafusion_common::{Column, DFSchema, ScalarValue}; + use datatypes::arrow::array::{Array, DictionaryArray, RecordBatch}; + use datatypes::arrow::datatypes::{DataType, Field, Schema, TimeUnit, UInt32Type}; + use datatypes::value::Value; + use regex::Regex; + + use crate::filter::predicate::{SimplePredicate, regexp_is_match_dictionary}; + use crate::filter::{SimpleFilterEvaluator, batch_filter}; + + #[test] + fn unsupported_filter_op() { + // `+` is not supported + let expr = Expr::BinaryExpr(BinaryExpr { + left: Box::new(Expr::Column(Column::from_name("foo"))), + op: Operator::Plus, + right: Box::new(1.lit()), + }); + assert!(SimpleFilterEvaluator::try_new(&expr).is_none()); + + // two literal is not supported + let expr = Expr::BinaryExpr(BinaryExpr { + left: Box::new(1.lit()), + op: Operator::Eq, + right: Box::new(1.lit()), + }); + assert!(SimpleFilterEvaluator::try_new(&expr).is_none()); + + // two column is not supported + let expr = Expr::BinaryExpr(BinaryExpr { + left: Box::new(Expr::Column(Column::from_name("foo"))), + op: Operator::Eq, + right: Box::new(Expr::Column(Column::from_name("bar"))), + }); + assert!(SimpleFilterEvaluator::try_new(&expr).is_none()); + + // compound expr is not supported + let expr = Expr::BinaryExpr(BinaryExpr { + left: Box::new(Expr::BinaryExpr(BinaryExpr { + left: Box::new(Expr::Column(Column::from_name("foo"))), + op: Operator::Eq, + right: Box::new(1.lit()), + })), + op: Operator::Eq, + right: Box::new(1.lit()), + }); + assert!(SimpleFilterEvaluator::try_new(&expr).is_none()); + for negated in [false, true] { + let membership = col("label").in_list(vec![lit(""), lit("lo")], negated); + assert!(SimpleFilterEvaluator::try_new(&membership).is_none()); + assert!( + SimpleFilterEvaluator::try_new(&col("label").is_null().or(membership)).is_none() + ); + } + for expr in [ + col("other").is_null().or(col("label").not_eq(lit("lo"))), + col("a.label") + .is_null() + .or(col("b.label").not_eq(lit("lo"))), + col("label").is_null().or(col("label").eq(col("other"))), + col("label") + .is_null() + .or(Expr::Cast(datafusion::logical_expr::Cast::new( + Box::new(col("label")), + DataType::Int64, + )) + .eq(lit("lo"))), + ] { + assert!(SimpleFilterEvaluator::try_new(&expr).is_none(), "{expr}"); + } + } + + #[test] + fn string_cast_requires_matching_column_type() { + let cast = Expr::Cast(datafusion::logical_expr::Cast::new( + Box::new(col("label")), + DataType::Utf8, + )); + let expr = col("label").is_null().or(cast.not_eq(lit("lo"))); + for data_type in [ + DataType::Int64, + DataType::Timestamp(TimeUnit::Millisecond, None), + DataType::Binary, + ] { + assert!( + SimpleFilterEvaluator::try_new_with_column_type(&expr, &|_| Some( + data_type.clone() + )) + .is_none() + ); + } + assert!(SimpleFilterEvaluator::try_new(&expr).is_none()); + } + + #[tokio::test] + async fn single_column_predicates_match_datafusion() { + use datafusion::prelude::SessionContext; + use datafusion_common::tree_node::{TreeNode, TreeNodeRecursion}; + use datatypes::arrow::array::{Int32Array, StringArray, UInt32Array}; + + let values = vec![ + None, + Some(""), + Some("lo"), + Some("eth0"), + Some("a\nb"), + Some("服务"), + None, + ]; + let strings: ArrayRef = Arc::new(StringArray::from(values)); + let dictionary: ArrayRef = Arc::new(DictionaryArray::::new( + UInt32Array::from(vec![ + None, + Some(0), + Some(1), + Some(2), + Some(3), + Some(4), + Some(5), + ]), + Arc::new(StringArray::from(vec![ + Some(""), + Some("lo"), + Some("eth0"), + Some("a\nb"), + Some("服务"), + None, + ])), + )); + let null = || lit(ScalarValue::Utf8(None)); + let mut predicates = vec![ + col("label").is_null(), + col("label").is_not_null(), + col("label") + .is_null() + .or(col("label").eq(lit(""))) + .and(col("label").is_not_null()), + col("label") + .not_eq(lit("lo")) + .and(col("label").not_eq(lit("eth0"))), + col("label") + .eq(lit("lo")) + .or(col("label").eq(lit("eth0"))) + .or(col("label").eq(lit(""))), + col("label").eq(null()).or(col("label").eq(lit("eth0"))), + col("label").gt(lit("a")).and(col("label").lt(lit("z"))), + lit("lo").not_eq(col("label")), + ]; + for (op, literal) in [ + (Operator::Eq, ""), + (Operator::NotEq, "lo"), + (Operator::RegexNotMatch, "^(?:lo|vfat)$"), + (Operator::RegexMatch, "^(?:|eth0)$"), + ] { + let comparison = Expr::BinaryExpr(BinaryExpr::new( + Box::new(col("label")), + op, + Box::new(lit(literal)), + )); + predicates.extend([ + col("label").is_null().or(comparison.clone()), + comparison.or(col("label").is_null()), + ]); + } + for input in [strings, dictionary] { + let ctx = SessionContext::new(); + let schema = Arc::new(Schema::new(vec![ + Field::new("label", input.data_type().clone(), true), + Field::new("id", DataType::Int32, false), + ])); + ctx.register_batch( + "labels", + RecordBatch::try_new( + schema, + vec![ + input.clone(), + Arc::new(Int32Array::from_iter_values(0..input.len() as i32)), + ], + ) + .unwrap(), + ) + .unwrap(); + for expr in &predicates { + let df = ctx + .table("labels") + .await + .unwrap() + .filter(expr.clone()) + .unwrap(); + let plan = df.clone().into_optimized_plan().unwrap(); + let expected = df + .collect() + .await + .unwrap() + .iter() + .flat_map(|batch| { + batch + .column_by_name("id") + .unwrap() + .as_any() + .downcast_ref::() + .unwrap() + .values() + .to_vec() + }) + .collect::>(); + let evaluate = |expr: &Expr| { + let evaluator = + SimpleFilterEvaluator::try_new_with_column_type(expr, &|name| { + (name == "label").then(|| input.data_type().clone()) + }) + .unwrap_or_else(|| panic!("unsupported: {expr}")); + let mask = evaluator.evaluate_array(&input).unwrap(); + for (i, matched) in mask.iter().enumerate() { + let scalar = ScalarValue::try_from_array(&input, i).unwrap(); + assert_eq!( + evaluator.evaluate_scalar(&scalar).unwrap(), + matched, + "{expr}, row={i}" + ); + } + mask + }; + let selected = |mask: BooleanBuffer| { + mask.iter() + .enumerate() + .filter_map(|(i, matched)| matched.then_some(i as i32)) + .collect::>() + }; + assert_eq!(selected(evaluate(expr)), expected, "{expr}"); + let mut filters = Vec::new(); + plan.apply(|plan| { + if let datafusion::logical_expr::LogicalPlan::Filter(filter) = plan { + filters.push(filter.predicate.clone()); + } + Ok(TreeNodeRecursion::Continue) + }) + .unwrap(); + if !filters.is_empty() { + let mut mask = BooleanBuffer::new_set(input.len()); + for filter in filters { + mask = &mask & &evaluate(&filter); + } + assert_eq!(selected(mask), expected, "optimized: {plan}"); + } + } + } + } + + #[test] + fn supported_filter_op() { + // equal + let expr = Expr::BinaryExpr(BinaryExpr { + left: Box::new(Expr::Column(Column::from_name("foo"))), + op: Operator::Eq, + right: Box::new(1.lit()), + }); + let _ = SimpleFilterEvaluator::try_new(&expr).unwrap(); + + // swap operands + let expr = Expr::BinaryExpr(BinaryExpr { + left: Box::new(1.lit()), + op: Operator::Lt, + right: Box::new(Expr::Column(Column::from_name("foo"))), + }); + let evaluator = SimpleFilterEvaluator::try_new(&expr).unwrap(); + assert!(evaluator.is_gt()); + assert_eq!(evaluator.column_name, "foo".to_string()); + } + + #[test] + fn run_on_array() { + let expr = Expr::BinaryExpr(BinaryExpr { + left: Box::new(Expr::Column(Column::from_name("foo"))), + op: Operator::Eq, + right: Box::new(1i64.lit()), + }); + let evaluator = SimpleFilterEvaluator::try_new(&expr).unwrap(); + + let input_1 = Arc::new(datatypes::arrow::array::Int64Array::from(vec![1, 2, 3])) as _; + let result = evaluator.evaluate_array(&input_1).unwrap(); + assert_eq!(result, BooleanBuffer::from(vec![true, false, false])); + + let input_2 = Arc::new(datatypes::arrow::array::Int64Array::from(vec![1, 1, 1])) as _; + let result = evaluator.evaluate_array(&input_2).unwrap(); + assert_eq!(result, BooleanBuffer::from(vec![true, true, true])); + + let input_3 = Arc::new(datatypes::arrow::array::Int64Array::new_null(0)) as _; + let result = evaluator.evaluate_array(&input_3).unwrap(); + assert_eq!(result, BooleanBuffer::from(vec![])); + } + + #[test] + fn run_on_scalar() { + let expr = Expr::BinaryExpr(BinaryExpr { + left: Box::new(Expr::Column(Column::from_name("foo"))), + op: Operator::Lt, + right: Box::new(1i64.lit()), + }); + let evaluator = SimpleFilterEvaluator::try_new(&expr).unwrap(); + + let input_1 = ScalarValue::Int64(Some(1)); + let result = evaluator.evaluate_scalar(&input_1).unwrap(); + assert!(!result); + + let input_2 = ScalarValue::Int64(Some(0)); + let result = evaluator.evaluate_scalar(&input_2).unwrap(); + assert!(result); + + let input_3 = ScalarValue::Int64(None); + let result = evaluator.evaluate_scalar(&input_3).unwrap(); + assert!(!result); + } + + #[test] + fn test_complex_filter_expression() { + // Create an expression tree for: col = 'B' OR col = 'C' OR col = 'D' + let col_eq_b = col("col").eq(lit("B")); + let col_eq_c = col("col").eq(lit("C")); + let col_eq_d = col("col").eq(lit("D")); + + // Build the OR chain + let col_or_expr = col_eq_b.or(col_eq_c).or(col_eq_d); + + // Check that SimpleFilterEvaluator can handle OR chain + let or_evaluator = SimpleFilterEvaluator::try_new(&col_or_expr).unwrap(); + assert_eq!(or_evaluator.column_name, "col"); + assert!(or_evaluator.is_or_eq_chain()); + assert_eq!( + or_evaluator.literal_list_values().unwrap(), + vec![Value::from("B"), Value::from("C"), Value::from("D")] + ); + + // Create a schema and batch for testing + let schema = Schema::new(vec![Field::new("col", DataType::Utf8, false)]); + let df_schema = DFSchema::try_from(schema.clone()).unwrap(); + let props = ExecutionProps::new(); + let physical_expr = create_physical_expr( + &col_or_expr, + &df_schema, + &props, + &PhysicalPlanningContext::default(), + ) + .unwrap(); + + // Create test data + let col_data = Arc::new(datatypes::arrow::array::StringArray::from(vec![ + "B", "C", "E", "B", "C", "D", "F", + ])); + let batch = RecordBatch::try_new(Arc::new(schema), vec![col_data]).unwrap(); + let expected = datatypes::arrow::array::StringArray::from(vec!["B", "C", "B", "C", "D"]); + + // Filter the batch + let filtered_batch = batch_filter(&batch, &physical_expr).unwrap(); + + // Expected: rows with col in ("B", "C", "D") + // That would be rows 0, 1, 3, 4, 5 + assert_eq!(filtered_batch.num_rows(), 5); + + let col_filtered = filtered_batch + .column(0) + .as_any() + .downcast_ref::() + .unwrap(); + assert_eq!(col_filtered, &expected); + } + + #[test] + fn test_maybe_build_regex() { + // Test case for RegexMatch (case sensitive, non-negative) + let (regex, negative) = SimplePredicate::maybe_build_regex( + Operator::RegexMatch, + &ScalarValue::Utf8(Some("a.*b".to_string())), + ) + .unwrap(); + assert!(regex.is_some()); + assert!(!negative); + assert!(regex.unwrap().is_match("axxb")); + + // Test case for RegexIMatch (case insensitive, non-negative) + let (regex, negative) = SimplePredicate::maybe_build_regex( + Operator::RegexIMatch, + &ScalarValue::Utf8(Some("a.*b".to_string())), + ) + .unwrap(); + assert!(regex.is_some()); + assert!(!negative); + assert!(regex.unwrap().is_match("AxxB")); + + // Test case for RegexNotMatch (case sensitive, negative) + let (regex, negative) = SimplePredicate::maybe_build_regex( + Operator::RegexNotMatch, + &ScalarValue::Utf8(Some("a.*b".to_string())), + ) + .unwrap(); + assert!(regex.is_some()); + assert!(negative); + + // Test case for RegexNotIMatch (case insensitive, negative) + let (regex, negative) = SimplePredicate::maybe_build_regex( + Operator::RegexNotIMatch, + &ScalarValue::Utf8(Some("a.*b".to_string())), + ) + .unwrap(); + assert!(regex.is_some()); + assert!(negative); + + // Test with empty regex pattern + let (regex, negative) = SimplePredicate::maybe_build_regex( + Operator::RegexMatch, + &ScalarValue::Utf8(Some("".to_string())), + ) + .unwrap(); + assert!(regex.is_none()); + assert!(!negative); + + // Test with non-regex operator + let (regex, negative) = SimplePredicate::maybe_build_regex( + Operator::Eq, + &ScalarValue::Utf8(Some("a.*b".to_string())), + ) + .unwrap(); + assert!(regex.is_none()); + assert!(!negative); + + // Test with invalid regex pattern + let result = SimplePredicate::maybe_build_regex( + Operator::RegexMatch, + &ScalarValue::Utf8(Some("a(b".to_string())), + ); + assert!(result.is_err()); + + // Test with non-string value + let result = SimplePredicate::maybe_build_regex( + Operator::RegexMatch, + &ScalarValue::Int64(Some(123)), + ); + assert!(result.is_err()); + + // Test with null value + let result = + SimplePredicate::maybe_build_regex(Operator::RegexMatch, &ScalarValue::Utf8(None)); + assert!(result.is_err()); + } + + #[test] + fn test_regex_match_dictionary_array() { + use datatypes::arrow::array::StringDictionaryBuilder; + + // Create a StringDictionaryArray + let mut builder = StringDictionaryBuilder::::new(); + builder.append("apple").unwrap(); + builder.append("banana").unwrap(); + builder.append("apple").unwrap(); + builder.append("cherry").unwrap(); + let dict_array = builder.finish(); + + // Test regex that matches "apple" + let regex = regex::Regex::new(r"app.*").unwrap(); + let result = regexp_is_match_dictionary(&dict_array, Some(®ex)).unwrap(); + + // Should match indices 0 and 2 (both "apple") + assert_eq!(result.len(), 4); + assert!(result.value(0)); // "apple" + assert!(!result.value(1)); // "banana" + assert!(result.value(2)); // "apple" + assert!(!result.value(3)); // "cherry" + + // Test regex that matches "banana" + let regex2 = regex::Regex::new(r"ban.*").unwrap(); + let result2 = regexp_is_match_dictionary(&dict_array, Some(®ex2)).unwrap(); + + assert!(!result2.value(0)); // "apple" + assert!(result2.value(1)); // "banana" + assert!(!result2.value(2)); // "apple" + assert!(!result2.value(3)); // "cherry" + + // Test with no regex (should match all) + let result3 = regexp_is_match_dictionary(&dict_array, None).unwrap(); + assert!(result3.value(0)); + assert!(result3.value(1)); + assert!(result3.value(2)); + assert!(result3.value(3)); + } + + #[test] + fn test_regex_scan_masks_preserve_sql_null_semantics() { + let plain = Arc::new(datatypes::arrow::array::StringArray::from(vec![ + Some("api"), + Some("API"), + Some("db"), + None, + ])) as ArrayRef; + assert_regex_scan_masks( + &plain, + [ + vec![true, false, false, false], + vec![true, true, false, false], + vec![false, true, true, false], + vec![false, false, true, false], + ], + ); + + let dictionary = DictionaryArray::new( + datatypes::arrow::array::UInt32Array::from(vec![ + Some(0), + Some(1), + Some(2), + None, + Some(3), + ]), + Arc::new(datatypes::arrow::array::StringArray::from(vec![ + Some("api"), + Some("API"), + Some("db"), + None, + ])), + ); + let raw = + regexp_is_match_dictionary(&dictionary, Some(&Regex::new("^api$").unwrap())).unwrap(); + assert!(raw.is_null(3)); // null dictionary key + assert!(raw.is_null(4)); // non-null key referencing a null dictionary value + + let dictionary = Arc::new(dictionary) as ArrayRef; + assert_regex_scan_masks( + &dictionary, + [ + vec![true, false, false, false, false], + vec![true, true, false, false, false], + vec![false, true, true, false, false], + vec![false, false, true, false, false], + ], + ); + + let negative = regex_evaluator(Operator::RegexNotMatch); + assert!(!negative.evaluate_scalar(&ScalarValue::Utf8(None)).unwrap()); + } + + fn assert_regex_scan_masks(input: &ArrayRef, expected: [Vec; 4]) { + for (op, expected) in [ + Operator::RegexMatch, + Operator::RegexIMatch, + Operator::RegexNotMatch, + Operator::RegexNotIMatch, + ] + .into_iter() + .zip(expected) + { + assert_eq!( + BooleanBuffer::from(expected), + regex_evaluator(op).evaluate_array(input).unwrap(), + "{op:?}" + ); + } + } + + fn regex_evaluator(op: Operator) -> SimpleFilterEvaluator { + let expr = Expr::BinaryExpr(BinaryExpr { + left: Box::new(Expr::Column(Column::from_name("host"))), + op, + right: Box::new("^api$".lit()), + }); + SimpleFilterEvaluator::try_new(&expr).unwrap() + } +} diff --git a/src/common/recordbatch/src/filter/tests.rs b/src/common/recordbatch/src/filter/tests.rs new file mode 100644 index 00000000000..77abf59b404 --- /dev/null +++ b/src/common/recordbatch/src/filter/tests.rs @@ -0,0 +1,72 @@ +// Copyright 2023 Greptime Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use std::sync::Arc; + +use datafusion::execution::context::ExecutionProps; +use datafusion::logical_expr::physical_planning_context::PhysicalPlanningContext; +use datafusion::logical_expr::{col, lit}; +use datafusion::physical_expr::create_physical_expr; +use datafusion_common::DFSchema; +use datatypes::arrow::datatypes::{DataType, Field, Schema}; + +use datatypes::arrow::array::{BooleanArray, RecordBatch}; + +use crate::filter::{batch_filter, boolean_array_to_scan_mask}; + +#[test] +fn batch_filter_test() { + let expr = col("ts").gt(lit(123456u64)); + let schema = Schema::new(vec![ + Field::new("a", DataType::Int32, true), + Field::new("ts", DataType::UInt64, false), + ]); + let df_schema = DFSchema::try_from(schema.clone()).unwrap(); + let props = ExecutionProps::new(); + let physical_expr = create_physical_expr( + &expr, + &df_schema, + &props, + &PhysicalPlanningContext::default(), + ) + .unwrap(); + let batch = RecordBatch::try_new( + Arc::new(schema), + vec![ + Arc::new(datatypes::arrow::array::Int32Array::from(vec![4, 5, 6])), + Arc::new(datatypes::arrow::array::UInt64Array::from(vec![ + 123456, 123457, 123458, + ])), + ], + ) + .unwrap(); + let new_batch = batch_filter(&batch, &physical_expr).unwrap(); + assert_eq!(new_batch.num_rows(), 2); + let first_column_values = new_batch + .column(0) + .as_any() + .downcast_ref::() + .unwrap(); + let expected = datatypes::arrow::array::Int32Array::from(vec![5, 6]); + assert_eq!(first_column_values, &expected); +} + +#[test] +fn test_nullable_boolean_predicate_becomes_scan_mask() { + let predicate = BooleanArray::from(vec![Some(true), None, Some(false)]); + assert_eq!( + BooleanArray::from(vec![true, false, false]), + boolean_array_to_scan_mask(&predicate) + ); +} diff --git a/src/common/recordbatch/src/filter/timestamp.rs b/src/common/recordbatch/src/filter/timestamp.rs new file mode 100644 index 00000000000..4c19aad546b --- /dev/null +++ b/src/common/recordbatch/src/filter/timestamp.rs @@ -0,0 +1,439 @@ +// Copyright 2023 Greptime Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! Timestamp-unit adaptation for predicates evaluated against historical SSTs. + +use common_time::timestamp::div_mod_units; +use datafusion::logical_expr::Operator; +use datafusion_common::ScalarValue; +use datafusion_common::arrow::array::{ArrayRef, Datum, Scalar}; +use datatypes::arrow::datatypes::TimeUnit; + +use crate::filter::predicate::SimplePredicate; + +/// Builds a tz-naive timestamp literal for `value` in `unit`. +pub fn timestamp_scalar_value(value: i64, unit: TimeUnit) -> ScalarValue { + match unit { + TimeUnit::Second => ScalarValue::TimestampSecond(Some(value), None), + TimeUnit::Millisecond => ScalarValue::TimestampMillisecond(Some(value), None), + TimeUnit::Microsecond => ScalarValue::TimestampMicrosecond(Some(value), None), + TimeUnit::Nanosecond => ScalarValue::TimestampNanosecond(Some(value), None), + } +} + +impl SimplePredicate { + pub(crate) fn cast_timestamp_unit(&self, target_unit: TimeUnit) -> Option { + // Constants represent scan outcomes here. As in the public Matched/Pruned + // result, UNKNOWN can be collapsed to false under these AND/OR-only trees. + match self { + Self::And(left, right) | Self::Or(left, right) => { + let left = left.cast_timestamp_unit(target_unit)?; + let right = right.cast_timestamp_unit(target_unit)?; + let and = matches!(self, Self::And(..)); + Some(match (left, right) { + (Self::Constant(Some(false)), _) | (_, Self::Constant(Some(false))) if and => { + Self::Constant(Some(false)) + } + (Self::Constant(Some(true)), _) | (_, Self::Constant(Some(true))) if !and => { + Self::Constant(Some(true)) + } + (Self::Constant(Some(true)), other) | (other, Self::Constant(Some(true))) + if and => + { + other + } + (Self::Constant(Some(false)), other) | (other, Self::Constant(Some(false))) + if !and => + { + other + } + (left, right) => { + if and { + Self::And(Box::new(left), Box::new(right)) + } else { + Self::Or(Box::new(left), Box::new(right)) + } + } + }) + } + Self::IsNull { .. } => Some(self.clone()), + Self::Constant(value) => Some(Self::Constant(Some(value.unwrap_or(false)))), + Self::OrEqChain { literals } => { + let mut converted = Vec::with_capacity(literals.len()); + for literal in literals { + let scalar = ScalarValue::try_from_array(literal.get().0, 0).ok()?; + let (value, unit) = timestamp_scalar_parts(&scalar)?; + let Some(value) = value else { + continue; + }; + let cast = div_mod_units(value, unit.into(), target_unit.into())?; + if cast.remainder == 0 { + converted.push(timestamp_scalar(cast.quotient, target_unit)?); + } + } + if converted.is_empty() { + return Some(Self::Constant(Some(false))); + } + Some(Self::OrEqChain { + literals: converted, + }) + } + Self::Comparison { literal, op, .. } => { + let scalar = ScalarValue::try_from_array(literal.get().0, 0).ok()?; + let (value, unit) = timestamp_scalar_parts(&scalar)?; + let Some(value) = value else { + return Some(Self::Constant(Some(false))); + }; + if unit == target_unit { + return Some(self.clone()); + } + let cast = div_mod_units(value, unit.into(), target_unit.into())?; + let divisible = cast.remainder == 0; + let literal = timestamp_scalar(cast.quotient, target_unit)?; + let filter = |op| Self::Comparison { + literal: literal.clone(), + op, + regex: None, + regex_negative: false, + }; + Some(match op { + Operator::Eq if divisible => filter(Operator::Eq), + Operator::Eq => Self::Constant(Some(false)), + Operator::NotEq if divisible => filter(Operator::NotEq), + Operator::NotEq => Self::Constant(Some(true)), + // The quotient is floor(L), including negative timestamps. + // v > L is v > quotient whether or not L is representable. + Operator::Gt => filter(Operator::Gt), + Operator::GtEq if divisible => filter(Operator::GtEq), + // Between target values, v >= L is v > quotient. + Operator::GtEq => filter(Operator::Gt), + Operator::Lt if divisible => filter(Operator::Lt), + // Between target values, v < L is v <= quotient. + Operator::Lt => filter(Operator::LtEq), + Operator::LtEq => filter(Operator::LtEq), + _ => return None, + }) + } + } + } +} + +/// Extracts the value and unit from a tz-naive timestamp scalar. +fn timestamp_scalar_parts(scalar: &ScalarValue) -> Option<(Option, TimeUnit)> { + let (value, unit, timezone) = match scalar { + ScalarValue::TimestampSecond(v, tz) => (*v, TimeUnit::Second, tz), + ScalarValue::TimestampMillisecond(v, tz) => (*v, TimeUnit::Millisecond, tz), + ScalarValue::TimestampMicrosecond(v, tz) => (*v, TimeUnit::Microsecond, tz), + ScalarValue::TimestampNanosecond(v, tz) => (*v, TimeUnit::Nanosecond, tz), + _ => return None, + }; + // A timezone-aware literal doesn't compare against a tz-naive column; + // leave it to the caller instead of guessing the intended semantics. + (timezone.is_none()).then_some((value, unit)) +} + +/// Builds a one-element scalar array holding `value` in `unit`. +fn timestamp_scalar(value: i64, unit: TimeUnit) -> Option> { + timestamp_scalar_value(value, unit).to_scalar().ok() +} + +#[cfg(test)] +mod tests { + use std::sync::Arc; + + use datafusion::execution::context::ExecutionProps; + use datafusion::logical_expr::physical_planning_context::PhysicalPlanningContext; + use datafusion::logical_expr::{Expr, col, lit}; + use datafusion::physical_expr::create_physical_expr; + use datafusion_common::arrow::array::ArrayRef; + use datafusion_common::arrow::buffer::BooleanBuffer; + use datafusion_common::cast::as_boolean_array; + use datafusion_common::{DFSchema, ScalarValue}; + use datatypes::arrow::array::{ + RecordBatch, TimestampMillisecondArray, TimestampNanosecondArray, + }; + use datatypes::arrow::datatypes::{Field, Schema}; + use datatypes::data_type::ConcreteDataType; + + use crate::filter::{SimpleFilterEvaluator, TimestampUnitCast, boolean_array_to_scan_mask}; + + #[test] + fn compound_timestamp_filters_preserve_unit_conversion() { + use datatypes::arrow::array::TimestampMicrosecondArray; + + let null = || lit(ScalarValue::TimestampMicrosecond(None, None)); + let predicates = [ + col("ts") + .gt(lit(ts_us(-1500))) + .and(col("ts").lt(lit(ts_us(500)))), + col("ts") + .eq(lit(ts_us(500))) + .or(col("ts").gt_eq(lit(ts_us(1500)))), + col("ts") + .is_not_null() + .and(col("ts").not_eq(lit(ts_us(500)))), + col("ts").eq(null()).or(col("ts").eq(lit(ts_us(-1000)))), + col("ts") + .eq(lit(ts_us(500))) + .and(col("ts").gt(lit(ts_us(0)))), + col("ts") + .not_eq(lit(ts_us(500))) + .or(col("ts").lt(lit(ts_us(0)))), + ]; + let native: ArrayRef = Arc::new(TimestampMicrosecondArray::from(vec![ + -2000, -1000, 0, 1000, 2000, + ])); + let old: ArrayRef = Arc::new(TimestampMillisecondArray::from(vec![-2, -1, 0, 1, 2])); + let schema = Arc::new(Schema::new(vec![Field::new( + "ts", + native.data_type().clone(), + false, + )])); + let batch = RecordBatch::try_new(schema.clone(), vec![native]).unwrap(); + for expr in predicates { + let physical = create_physical_expr( + &expr, + &DFSchema::try_from(schema.as_ref().clone()).unwrap(), + &ExecutionProps::new(), + &PhysicalPlanningContext::default(), + ) + .unwrap(); + let result = physical + .evaluate(&batch) + .unwrap() + .into_array(batch.num_rows()) + .unwrap(); + let expected = boolean_array_to_scan_mask(as_boolean_array(&result).unwrap()) + .values() + .clone(); + let actual = match cast_to_ms(&expr).unwrap() { + TimestampUnitCast::Matched => BooleanBuffer::new_set(old.len()), + TimestampUnitCast::Pruned => BooleanBuffer::new_unset(old.len()), + TimestampUnitCast::Filter(filter) => filter.evaluate_array(&old).unwrap(), + }; + assert_eq!(actual, expected, "{expr}"); + } + } + + fn ts_us(v: i64) -> ScalarValue { + ScalarValue::TimestampMicrosecond(Some(v), None) + } + + fn cast_to_ms(expr: &Expr) -> Option { + SimpleFilterEvaluator::try_new(expr) + .unwrap() + .cast_timestamp_unit(&ConcreteDataType::timestamp_millisecond_datatype()) + } + + /// Evaluates a cast `Filter` outcome against millisecond values and + /// returns the mask. + fn eval_ms_mask(cast: Option, values: &[i64]) -> Vec { + let filter = match cast.expect("cast must apply") { + TimestampUnitCast::Filter(filter) => filter, + other => panic!("expected Filter outcome, got {other:?}"), + }; + let array = Arc::new(TimestampMillisecondArray::from(values.to_vec())) as ArrayRef; + filter.evaluate_array(&array).unwrap().iter().collect() + } + + #[test] + fn cast_timestamp_unit_converts_representable_literal() { + // ts = 7_000_000us evaluated against a ms column becomes ts = 7000ms. + assert_eq!( + vec![false, true, false], + eval_ms_mask( + cast_to_ms(&col("ts").eq(lit(ts_us(7_000_000)))), + &[6_999, 7_000, 7_001] + ) + ); + // != complements =. + assert_eq!( + vec![true, false, true], + eval_ms_mask( + cast_to_ms(&col("ts").not_eq(lit(ts_us(7_000_000)))), + &[6_999, 7_000, 7_001] + ) + ); + } + + #[test] + fn cast_timestamp_unit_prunes_non_representable_equality() { + // 7_000_500us is not a whole millisecond: no ms row can equal it, + // and every ms row satisfies `!=`. + assert!(matches!( + cast_to_ms(&col("ts").eq(lit(ts_us(7_000_500)))), + Some(TimestampUnitCast::Pruned) + )); + assert!(matches!( + cast_to_ms(&col("ts").not_eq(lit(ts_us(7_000_500)))), + Some(TimestampUnitCast::Matched) + )); + } + + #[test] + fn cast_timestamp_unit_strengthens_inequalities() { + // 2_500_500us is strictly between 2500ms and 2501ms: the cast must + // not round the literal to a boundary the original predicate + // excludes (>= 2500ms would wrongly match 2500ms). + let values = vec![2_499, 2_500, 2_501]; + assert_eq!( + vec![false, false, true], + eval_ms_mask(cast_to_ms(&col("ts").gt(lit(ts_us(2_500_500)))), &values) + ); + assert_eq!( + vec![false, false, true], + eval_ms_mask(cast_to_ms(&col("ts").gt_eq(lit(ts_us(2_500_500)))), &values) + ); + assert_eq!( + vec![true, true, false], + eval_ms_mask(cast_to_ms(&col("ts").lt(lit(ts_us(2_500_500)))), &values) + ); + assert_eq!( + vec![true, true, false], + eval_ms_mask(cast_to_ms(&col("ts").lt_eq(lit(ts_us(2_500_500)))), &values) + ); + + // Representable boundary keeps the original operator. + let values = vec![2_499, 2_500, 2_501]; + assert_eq!( + vec![false, false, true], + eval_ms_mask(cast_to_ms(&col("ts").gt(lit(ts_us(2_500_000)))), &values) + ); + assert_eq!( + vec![false, true, true], + eval_ms_mask(cast_to_ms(&col("ts").gt_eq(lit(ts_us(2_500_000)))), &values) + ); + assert_eq!( + vec![true, false, false], + eval_ms_mask(cast_to_ms(&col("ts").lt(lit(ts_us(2_500_000)))), &values) + ); + assert_eq!( + vec![true, true, false], + eval_ms_mask(cast_to_ms(&col("ts").lt_eq(lit(ts_us(2_500_000)))), &values) + ); + } + + #[test] + fn cast_timestamp_unit_handles_negative_instants() { + // -2_500_500us floors to -2501ms with remainder 500us: only rows + // with instant >= -2.5005ms may match >= / >. + let values = vec![-2_502, -2_501, -2_500]; + assert_eq!( + vec![false, false, true], + eval_ms_mask( + cast_to_ms(&col("ts").gt_eq(lit(ts_us(-2_500_500)))), + &values + ) + ); + assert_eq!( + vec![false, false, true], + eval_ms_mask(cast_to_ms(&col("ts").gt(lit(ts_us(-2_500_500)))), &values) + ); + assert_eq!( + vec![true, true, false], + eval_ms_mask( + cast_to_ms(&col("ts").lt_eq(lit(ts_us(-2_500_500)))), + &values + ) + ); + // Exactly representable negative literal: -2_500_000us == -2500ms. + assert_eq!( + vec![false, false, true], + eval_ms_mask(cast_to_ms(&col("ts").eq(lit(ts_us(-2_500_000)))), &values) + ); + } + + #[test] + fn cast_timestamp_unit_or_chain_drops_unrepresentable_literals() { + // Only 7_000_000us is a whole millisecond; 7_000_500us cannot match. + let expr = col("ts") + .eq(lit(ts_us(7_000_500))) + .or(col("ts").eq(lit(ts_us(7_000_000)))); + assert_eq!( + vec![false, true], + eval_ms_mask(cast_to_ms(&expr), &[6_999, 7_000]) + ); + + // A chain where no literal is representable matches nothing. + let expr = col("ts") + .eq(lit(ts_us(7_000_500))) + .or(col("ts").eq(lit(ts_us(6_000_500)))); + assert!(matches!(cast_to_ms(&expr), Some(TimestampUnitCast::Pruned))); + } + + #[test] + fn cast_timestamp_unit_null_literal_prunes() { + // A NULL literal never compares true. + assert!(matches!( + cast_to_ms(&col("ts").eq(lit(ScalarValue::TimestampMicrosecond(None, None)))), + Some(TimestampUnitCast::Pruned) + )); + } + + #[test] + fn cast_timestamp_unit_same_unit_returns_filter_directly() { + // A literal already in the target unit is returned unchanged. + let expr = col("ts").eq(lit(ts_us(7_000_000))); + let filter = SimpleFilterEvaluator::try_new(&expr).unwrap(); + let cast = filter + .cast_timestamp_unit(&ConcreteDataType::timestamp_microsecond_datatype()) + .unwrap(); + let TimestampUnitCast::Filter(f) = cast else { + panic!("expected Filter, got {cast:?}") + }; + assert!(filter.is_eq() && f.is_eq()); + assert_eq!(filter.column_name(), f.column_name()); + assert_eq!(filter.literal_value(), f.literal_value()); + } + + #[test] + fn cast_timestamp_unit_rejects_non_timestamps() { + // Non-timestamp target type. + assert!( + SimpleFilterEvaluator::try_new(&col("ts").gt(lit(ts_us(1)))) + .unwrap() + .cast_timestamp_unit(&ConcreteDataType::int64_datatype()) + .is_none() + ); + // Non-timestamp literal against a timestamp target. + assert!(cast_to_ms(&col("ts").gt(lit(42_i64))).is_none()); + } + + #[test] + fn cast_timestamp_unit_to_finer_unit() { + // 5 seconds against a nanosecond column becomes 5_000_000_000ns. + let filter = SimpleFilterEvaluator::try_new( + &col("ts").eq(lit(ScalarValue::TimestampSecond(Some(5), None))), + ) + .unwrap() + .cast_timestamp_unit(&ConcreteDataType::timestamp_nanosecond_datatype()) + .and_then(|cast| match cast { + TimestampUnitCast::Filter(filter) => Some(filter), + _ => None, + }) + .unwrap(); + let array = Arc::new(TimestampNanosecondArray::from(vec![ + 4_999_999_999, + 5_000_000_000, + ])) as ArrayRef; + assert_eq!( + vec![false, true], + filter + .evaluate_array(&array) + .unwrap() + .iter() + .collect::>() + ); + } +} diff --git a/src/mito2/src/engine/prune_test.rs b/src/mito2/src/engine/prune_test.rs index f2171510345..71e6845e96f 100644 --- a/src/mito2/src/engine/prune_test.rs +++ b/src/mito2/src/engine/prune_test.rs @@ -288,7 +288,7 @@ async fn test_prune_memtable_complex_expr_with_format(flat_format: bool) { ) .await; - // ts filter will be ignored when pruning time series in memtable. + // Both the compound time range and the tag predicate are evaluated during the scan. let filters = vec![time_range_expr(4, 7), col("tag_0").lt(lit("6"))]; let stream = engine @@ -306,10 +306,6 @@ async fn test_prune_memtable_complex_expr_with_format(flat_format: bool) { +-------+---------+---------------------+ | tag_0 | field_0 | ts | +-------+---------+---------------------+ -| 0 | 0.0 | 1970-01-01T00:00:00 | -| 1 | 1.0 | 1970-01-01T00:00:01 | -| 2 | 2.0 | 1970-01-01T00:00:02 | -| 3 | 3.0 | 1970-01-01T00:00:03 | | 4 | 4.0 | 1970-01-01T00:00:04 | | 5 | 5.0 | 1970-01-01T00:00:05 | +-------+---------+---------------------+"; diff --git a/src/mito2/src/engine/scan_test.rs b/src/mito2/src/engine/scan_test.rs index e10968d3a32..39a1e66f6d8 100644 --- a/src/mito2/src/engine/scan_test.rs +++ b/src/mito2/src/engine/scan_test.rs @@ -73,6 +73,215 @@ use crate::test_util; use crate::test_util::sst_util::{new_sparse_primary_key, sst_region_metadata_with_encoding}; use crate::test_util::{CreateRequestBuilder, TestEnv, reopen_region}; +#[tokio::test] +async fn test_nullable_label_filters_in_memtables_and_ssts() { + use datafusion_expr::{BinaryExpr as LogicalBinaryExpr, Cast, Expr}; + + for (flat, memtable) in [(false, "time_series"), (true, "bulk")] { + let mut env = TestEnv::new().await; + let engine = env + .create_engine(MitoConfig { + default_flat_format: flat, + ..Default::default() + }) + .await; + let region_id = RegionId::new(1, 1); + let request = CreateRequestBuilder::new() + .insert_option("memtable.type", memtable) + .build(); + let schema = test_util::rows_schema(&request); + engine + .handle_request(region_id, RegionRequest::Create(request)) + .await + .unwrap(); + let rows = [None, Some(""), Some("tmpfs"), Some("ext4")] + .into_iter() + .enumerate() + .map(|(i, label)| { + let mut row = row(vec![ + ValueData::StringValue(String::new()), + ValueData::F64Value(i as f64), + ValueData::TimestampMillisecondValue(i as i64), + ]); + row.values[0].value_data = label.map(|v| ValueData::StringValue(v.into())); + row + }) + .collect(); + test_util::put_rows(&engine, region_id, Rows { schema, rows }).await; + for flushed in [false, true] { + if flushed { + test_util::flush_region(&engine, region_id, None).await; + } + let mut cases = Vec::new(); + for (op, literal, include_null, expected) in [ + (Operator::NotEq, "tmpfs", true, vec![0, 1, 3]), + ( + Operator::RegexNotMatch, + "^(?:tmpfs|vfat)$", + true, + vec![0, 1, 3], + ), + (Operator::Eq, "", true, vec![0, 1]), + (Operator::NotEq, "", false, vec![2, 3]), + (Operator::RegexMatch, "^(?:ext4)$", false, vec![3]), + ] { + for cast in [false, true] { + let column = if cast { + Expr::Cast(Cast::new(Box::new(col("tag_0")), DataType::Utf8)) + } else { + col("tag_0") + }; + let comparison = Expr::BinaryExpr(LogicalBinaryExpr::new( + Box::new(column), + op, + Box::new(lit(literal)), + )); + let filter = if include_null { + col("tag_0").is_null().or(comparison) + } else { + comparison + }; + cases.push((filter, expected.clone())); + } + } + let column = col("tag_0"); + cases.extend([ + (column.clone().is_null(), vec![0]), + (column.clone().is_not_null(), vec![1, 2, 3]), + ( + column + .clone() + .is_null() + .or(column.clone().not_eq(lit("tmpfs"))) + .and(column.clone().is_null().or(column.clone().not_eq(lit("")))), + vec![0, 3], + ), + ( + column.clone().eq(lit("ext4")).or(column.eq(lit("tmpfs"))), + vec![2, 3], + ), + ]); + for (filter, expected) in cases { + let stream = engine + .scan_to_stream( + region_id, + ScanRequest { + filters: vec![filter.clone()], + distribution: flat.then_some(TimeSeriesDistribution::PerSeries), + ..Default::default() + }, + ) + .await + .unwrap(); + let batches = RecordBatches::try_collect(stream).await.unwrap(); + let mut actual = Vec::new(); + for batch in batches.iter() { + let ts = batch + .column_by_name("ts") + .unwrap() + .as_primitive::(); + actual.extend(ts.values().iter().copied()); + } + actual.sort_unstable(); + assert_eq!(actual, expected, "flat={flat}, flushed={flushed}, {filter}"); + } + } + } +} + +#[tokio::test] +async fn test_compound_time_filters_before_last_row() { + for (flat, memtable) in [ + (false, "time_series"), + (true, "time_series"), + (true, "bulk"), + ] { + let mut env = TestEnv::new().await; + let engine = env + .create_engine(MitoConfig { + default_flat_format: flat, + ..Default::default() + }) + .await; + let region_id = RegionId::new(1, 1); + let request = CreateRequestBuilder::new() + .insert_option("memtable.type", memtable) + .build(); + let schema = test_util::rows_schema(&request); + engine + .handle_request(region_id, RegionRequest::Create(request)) + .await + .unwrap(); + let rows = (0..5) + .map(|i| { + row(vec![ + ValueData::StringValue("series".into()), + ValueData::F64Value(i as f64), + ValueData::TimestampMillisecondValue(i), + ]) + }) + .collect(); + test_util::put_rows(&engine, region_id, Rows { schema, rows }).await; + let ts = |value| lit(ScalarValue::TimestampMillisecond(Some(value), None)); + let cases = [ + (col("ts").eq(ts(1)).or(col("ts").eq(ts(3))), vec![1, 3]), + (col("ts").gt_eq(ts(1)).and(col("ts").lt(ts(3))), vec![1, 2]), + (col("ts").is_null().or(col("ts").lt(ts(2))), vec![0, 1]), + ]; + for flushed in [false, true] { + if flushed { + test_util::flush_region(&engine, region_id, None).await; + } + for (filter, expected) in &cases { + for selector in [ + None, + Some(TimeSeriesRowSelector::LastRow { after_merge: false }), + Some(TimeSeriesRowSelector::LastRow { after_merge: true }), + ] { + let stream = engine + .scan_to_stream( + region_id, + ScanRequest { + filters: vec![ + col("tag_0").is_null().or(col("tag_0").eq(lit("series"))), + filter.clone(), + ], + distribution: flat.then_some(TimeSeriesDistribution::PerSeries), + series_row_selector: selector, + ..Default::default() + }, + ) + .await + .unwrap(); + let batches = RecordBatches::try_collect(stream).await.unwrap(); + let mut actual = Vec::new(); + for batch in batches.iter() { + actual.extend( + batch + .column_by_name("ts") + .unwrap() + .as_primitive::() + .values() + .iter() + .copied(), + ); + } + actual.sort_unstable(); + let expected = if selector.is_some() { + expected.last().copied().into_iter().collect::>() + } else { + expected.clone() + }; + assert_eq!( + actual, expected, + "flat={flat}, memtable={memtable}, flushed={flushed}, selector={selector:?}, {filter}" + ); + } + } + } + } +} + #[tokio::test] async fn test_json_type_hint_pushdown_scanner_returns_batches() -> WhateverResult<()> { // Create a region with a JSON2 field whose physical Parquet representation is a nested struct. diff --git a/src/mito2/src/memtable/time_series.rs b/src/mito2/src/memtable/time_series.rs index 4e720df88df..fad746927d4 100644 --- a/src/mito2/src/memtable/time_series.rs +++ b/src/mito2/src/memtable/time_series.rs @@ -641,7 +641,13 @@ impl Iter { predicate .exprs() .iter() - .filter_map(SimpleFilterEvaluator::try_new) + .filter_map(|expr| { + SimpleFilterEvaluator::try_new_with_column_type(expr, &|name| { + metadata + .column_by_name(name) + .map(|column| column.column_schema.data_type.as_arrow_type()) + }) + }) .collect::>() }) .unwrap_or_default(); diff --git a/src/mito2/src/read/scan_region.rs b/src/mito2/src/read/scan_region.rs index dc68835a11f..5cfd6a29844 100644 --- a/src/mito2/src/read/scan_region.rs +++ b/src/mito2/src/read/scan_region.rs @@ -29,10 +29,12 @@ use common_telemetry::tracing::Instrument; use common_telemetry::{debug, error, tracing, warn}; use common_time::range::TimestampRange; use datafusion::execution::memory_pool::{MemoryPool, UnboundedMemoryPool}; +use datafusion::optimizer::simplify_expressions::ExprSimplifier; use datafusion::physical_plan::expressions::DynamicFilterPhysicalExpr; use datafusion_common::pruning::PruningStatistics; -use datafusion_common::{Column, ScalarValue}; +use datafusion_common::{Column, ScalarValue, ToDFSchema}; use datafusion_expr::Expr; +use datafusion_expr::simplify::SimplifyContext; use datafusion_expr::utils::expr_to_columns; use datatypes::arrow::array::{ArrayRef, BooleanArray, UInt64Array}; use datatypes::extension::json::is_json2_extension_type; @@ -57,8 +59,8 @@ use crate::access_layer::AccessLayerRef; use crate::cache::CacheStrategy; use crate::config::DEFAULT_MAX_CONCURRENT_SCAN_FILES; use crate::error::{ - InvalidPartitionExprSnafu, InvalidRequestSnafu, RegionSequenceDomainBrokenSnafu, Result, - SequenceRangeUnsupportedSnafu, + EvalPartitionFilterSnafu, InvalidPartitionExprSnafu, InvalidRequestSnafu, + RegionSequenceDomainBrokenSnafu, Result, SequenceRangeUnsupportedSnafu, }; #[cfg(feature = "enterprise")] use crate::extension::{BoxedExtensionRange, BoxedExtensionRangeProvider}; @@ -2197,6 +2199,22 @@ impl PredicateGroup { expr: expr_json.clone(), })?; + // Persisted partition bounds retain their original types after schema changes. + let schema = metadata + .schema + .arrow_schema() + .clone() + .to_dfschema_ref() + .context(EvalPartitionFilterSnafu)?; + let simplifier = ExprSimplifier::new( + SimplifyContext::builder() + .with_schema(schema.clone()) + .build(), + ); + let logical_expr = simplifier + .coerce(logical_expr, &schema) + .and_then(|expr| simplifier.simplify(expr)) + .context(EvalPartitionFilterSnafu)?; combined_exprs.push(logical_expr); region_partition_expr = Some(expr); } @@ -2339,6 +2357,75 @@ mod tests { .with_files(vec![file]) } + #[test] + fn test_partition_time_filter_after_unit_widening() { + use datatypes::arrow::array::TimestampMicrosecondArray; + + let bound = Value::Timestamp(Timestamp::new_millisecond(1)); + for (expr, expected) in [ + ( + partition_col("ts").lt(bound.clone()), + vec![true, false, false], + ), + (partition_col("ts").gt_eq(bound), vec![false, true, true]), + ] { + let mut builder = RegionMetadataBuilder::new(RegionId::new(1, 1)); + builder.push_column_metadata(ColumnMetadata { + column_schema: ColumnSchema::new( + "ts", + ConcreteDataType::timestamp_microsecond_datatype(), + false, + ), + semantic_type: SemanticType::Timestamp, + column_id: 0, + }); + builder.primary_key(vec![]); + builder.partition_expr_json(Some(expr.as_json_str().unwrap())); + let metadata = builder.build().unwrap(); + let group = PredicateGroup::new(&metadata, &[]).unwrap(); + let filters = group.time_filters().unwrap(); + let input: ArrayRef = Arc::new(TimestampMicrosecondArray::from(vec![999, 1000, 1001])); + assert_eq!(filters.len(), 1); + assert_eq!( + filters[0].evaluate_array(&input).unwrap(), + datafusion_common::arrow::buffer::BooleanBuffer::from(expected), + ); + } + } + + #[test] + fn test_partition_time_filter_overflow_after_unit_widening() { + // Year 3000 fits milliseconds but overflows nanoseconds. + let bound = Value::Timestamp(Timestamp::new_millisecond(32_503_680_000_000)); + let expr = partition_col("ts").lt(bound); + let mut builder = RegionMetadataBuilder::new(RegionId::new(1, 1)); + builder.push_column_metadata(ColumnMetadata { + column_schema: ColumnSchema::new( + "ts", + ConcreteDataType::timestamp_nanosecond_datatype(), + false, + ), + semantic_type: SemanticType::Timestamp, + column_id: 0, + }); + builder.primary_key(vec![]); + builder.partition_expr_json(Some(expr.as_json_str().unwrap())); + let metadata = builder.build().unwrap(); + + let error = PredicateGroup::new(&metadata, &[]) + .err() + .expect("an overflowing partition bound must fail scan construction"); + let crate::error::Error::EvalPartitionFilter { error, .. } = error else { + panic!("unexpected error: {error:?}"); + }; + assert!( + error + .to_string() + .contains("converted value exceeds the representable i64 range"), + "{error}" + ); + } + #[tokio::test] async fn test_total_rows_is_exact_after_partition_filter() { let expr = partition_col("k0").gt_eq(Value::String("foo".into())); diff --git a/src/mito2/src/sst/parquet/prefilter.rs b/src/mito2/src/sst/parquet/prefilter.rs index 4362440e415..d5401f4a6de 100644 --- a/src/mito2/src/sst/parquet/prefilter.rs +++ b/src/mito2/src/sst/parquet/prefilter.rs @@ -341,8 +341,7 @@ pub(crate) fn build_bulk_filter_plan( predicate: Option<&Predicate>, ) -> BulkFilterPlan { let metadata = read_format.metadata(); - // Bulk memtable only needs simple binary filters here. Any filter that - // cannot be reduced to a SimpleFilterContext stays out of this fast path. + // Filters not recognized by SimpleFilterContext stay out of this fast path. let simple_filters: Vec = predicate .into_iter() .flat_map(|predicate| { diff --git a/src/mito2/src/sst/parquet/reader.rs b/src/mito2/src/sst/parquet/reader.rs index 90c2728e35d..a3cbdef90f1 100644 --- a/src/mito2/src/sst/parquet/reader.rs +++ b/src/mito2/src/sst/parquet/reader.rs @@ -2014,7 +2014,12 @@ impl SimpleFilterContext { expected_meta: Option<&RegionMetadata>, expr: &Expr, ) -> Option { - let filter = SimpleFilterEvaluator::try_new(expr)?; + let filter = SimpleFilterEvaluator::try_new_with_column_type(expr, &|name| { + expected_meta + .unwrap_or(sst_meta) + .column_by_name(name) + .map(|column| column.column_schema.data_type.as_arrow_type()) + })?; let expr_str = format!("{expr:?}"); let (column_metadata, maybe_filter) = match expected_meta { Some(meta) => { @@ -3159,6 +3164,15 @@ mod tests { )); } + #[test] + fn test_nullable_label_filter_uses_missing_sst_column_default() { + let metadata = Arc::new(sst_region_metadata()); + let expected = expected_metadata_with_reused_tag_name(&metadata); + let expr = col("tag_0").is_null().or(col("tag_0").not_eq(lit("lo"))); + let ctx = SimpleFilterContext::new_opt(&metadata, Some(&expected), &expr).unwrap(); + assert!(matches!(ctx.filter(), MaybeFilter::Matched)); + } + #[test] fn test_simple_filter_context_drops_mismatched_field_filter() { let (sst_metadata, latest_metadata) = mock_metadata(); diff --git a/src/query/src/dummy_catalog.rs b/src/query/src/dummy_catalog.rs index 5644194f1f9..8f0a2c7ac60 100644 --- a/src/query/src/dummy_catalog.rs +++ b/src/query/src/dummy_catalog.rs @@ -31,6 +31,7 @@ use datafusion::physical_plan::ExecutionPlan; use datafusion_common::DataFusionError; use datafusion_expr::{Expr, TableProviderFilterPushDown, TableType}; use datatypes::arrow::datatypes::SchemaRef; +use datatypes::data_type::DataType; use datatypes::types::json_type::JsonNativeType; use futures::stream::BoxStream; use session::context::{QueryContext, QueryContextRef}; @@ -209,7 +210,13 @@ impl TableProvider for DummyTableProvider { .iter() .map(|e| { // Simple filter on primary key columns are precisely evaluated. - if let Some(simple_filter) = SimpleFilterEvaluator::try_new(e) { + if let Some(simple_filter) = + SimpleFilterEvaluator::try_new_with_column_type(e, &|name| { + self.metadata + .column_by_name(name) + .map(|column| column.column_schema.data_type.as_arrow_type()) + }) + { if self .metadata .column_by_name(simple_filter.column_name()) @@ -646,6 +653,37 @@ mod tests { FLOW_RETURN_REGION_SEQ, FLOW_SINK_TABLE_ID, }; + #[test] + fn test_nullable_label_filter_pushdown() { + use datafusion_expr::{Cast, col, lit}; + use datatypes::arrow::datatypes::DataType as ArrowDataType; + + let provider = crate::optimizer::test_util::mock_table_provider(RegionId::new(1, 1)); + let cast_label = Expr::Cast(Cast::new(Box::new(col("k0")), ArrowDataType::Utf8)); + let cast_field = Expr::Cast(Cast::new(Box::new(col("v0")), ArrowDataType::Utf8)); + let filters = [ + col("k0").is_null().or(col("k0").not_eq(lit("lo"))), + col("k0").is_null().or(cast_label.not_eq(lit("lo"))), + col("k0").is_null().or(col("v0").not_eq(lit(1.0))), + cast_field.eq(lit("1")), + col("k0").in_list(vec![lit(""), lit("lo")], false), + col("k0").in_list(vec![lit(""), lit("lo")], true), + ]; + assert_eq!( + provider + .supports_filters_pushdown(&filters.iter().collect::>()) + .unwrap(), + vec![ + TableProviderFilterPushDown::Exact, + TableProviderFilterPushDown::Exact, + TableProviderFilterPushDown::Inexact, + TableProviderFilterPushDown::Inexact, + TableProviderFilterPushDown::Inexact, + TableProviderFilterPushDown::Inexact, + ] + ); + } + fn test_region_id() -> RegionId { RegionId::new(1024, 1) } diff --git a/src/query/src/promql/planner.rs b/src/query/src/promql/planner.rs index 88464bc9ee8..07faf3b53dd 100644 --- a/src/query/src/promql/planner.rs +++ b/src/query/src/promql/planner.rs @@ -2557,28 +2557,21 @@ impl PromPlanner { // for it, or the column exists but is NULL on that row — the latter // is the norm for logical metrics sharing a physical table, which // holds the union of their label columns. + let mut null_label = None; let col = if let Some(column_name) = column_name { let column = DfExpr::Column(Column::from_name(&column_name)); let field = table_schema .index_of_column_by_name(None, &column_name) .map(|index| table_schema.field(index)); if accepts_empty - && let Some(data_type) = field - .filter(|field| { - field.is_nullable() - && Self::string_value_data_type(field.data_type()).is_some() - }) - .map(|field| field.data_type()) - { - let empty = Self::string_scalar_value(data_type, Some(String::new())) - .expect("nullable label has a string type"); - DfExpr::ScalarFunction(ScalarFunction { - func: coalesce(), - args: vec![column, DfExpr::Literal(empty, None)], + && field.is_some_and(|field| { + field.is_nullable() + && Self::string_value_data_type(field.data_type()).is_some() }) - } else { - column + { + null_label = Some(column.clone().is_null()); } + column } else { DfExpr::Literal(ScalarValue::Utf8(Some(String::new())), None) }; @@ -2631,6 +2624,10 @@ impl PromPlanner { } } }; + let expr = match null_label { + Some(null_label) => null_label.or(expr), + None => expr, + }; exprs.push(expr); } diff --git a/src/query/src/promql/planner/test/delta.rs b/src/query/src/promql/planner/test/delta.rs index f0c918351d1..9ca8a8b646c 100644 --- a/src/query/src/promql/planner/test/delta.rs +++ b/src/query/src/promql/planner/test/delta.rs @@ -535,6 +535,7 @@ async fn matchers_read_absent_labels_as_empty() { schema.clone(), vec![Arc::new(StringArray::from(vec![ None, + Some(""), Some(GREPTIME_TEMPORALITY_DELTA), ]))], ) @@ -542,10 +543,19 @@ async fn matchers_read_absent_labels_as_empty() { let table = Arc::new(MemTable::try_new(schema, vec![vec![batch]]).unwrap()); for (matcher, expected) in [ - (r#"="""#, vec![None]), - (r#"!="delta""#, vec![None]), - (r#"=~".*""#, vec![None, Some(GREPTIME_TEMPORALITY_DELTA)]), - (r#"!~"delta""#, vec![None]), + (r#"="""#, vec![None, Some("")]), + (r#"!="delta""#, vec![None, Some("")]), + ( + r#"=~".*""#, + vec![None, Some(""), Some(GREPTIME_TEMPORALITY_DELTA)], + ), + (r#"!~"delta""#, vec![None, Some("")]), + (r#"!="""#, vec![Some(GREPTIME_TEMPORALITY_DELTA)]), + ( + r#"=~"delta|""#, + vec![None, Some(""), Some(GREPTIME_TEMPORALITY_DELTA)], + ), + (r#"!~"delta|""#, vec![]), (r#"="delta""#, vec![Some(GREPTIME_TEMPORALITY_DELTA)]), ] { let query = format!(r#"metric{{{marker}{matcher}}}"#); @@ -558,10 +568,10 @@ async fn matchers_read_absent_labels_as_empty() { }; let expressions = PromPlanner::matchers_to_expr(selector.matchers, scan.schema()).unwrap(); let display = expressions.iter().map(ToString::to_string).join(" AND "); - if matcher == r#"="delta""# { - assert!(!display.contains("coalesce"), "{display}"); + if [r#"="delta""#, r#"!="""#, r#"!~"delta|""#].contains(&matcher) { + assert!(!display.contains("IS NULL"), "{display}"); } else if !expressions.is_empty() { - assert!(display.contains("coalesce"), "{display}"); + assert!(display.contains("IS NULL"), "{display}"); } let plan = if let Some(filter) = conjunction(expressions) { LogicalPlanBuilder::from(scan) @@ -599,7 +609,7 @@ async fn matchers_read_absent_labels_as_empty() { // The rule is about NULL, not about the marker: any nullable label column // can be NULL where the series does not carry the label. A non-nullable one // has nothing to normalize. - for (nullable, wants_coalesce) in [(true, true), (false, false)] { + for (nullable, wants_null_filter) in [(true, true), (false, false)] { let ordinary_schema = Arc::new(ArrowSchema::new(vec![Field::new( "label", ArrowDataType::Utf8, @@ -626,8 +636,8 @@ async fn matchers_read_absent_labels_as_empty() { .map(ToString::to_string) .join(" AND "); assert_eq!( - wants_coalesce, - expressions.contains("coalesce"), + wants_null_filter, + expressions.contains("IS NULL"), "nullable={nullable}: {expressions}" ); } diff --git a/tests/cases/standalone/common/promql/regex.result b/tests/cases/standalone/common/promql/regex.result index c76b743eafa..c2a10626ff8 100644 --- a/tests/cases/standalone/common/promql/regex.result +++ b/tests/cases/standalone/common/promql/regex.result @@ -146,13 +146,93 @@ TQL ANALYZE VERBOSE (0, 0, '1s') test{host!~".+"}; |_|_|_| | 1_| 0_|_PromInstantManipulateExec: range=[0..0], lookback=[300000], interval=[1000], time index=[ts] REDACTED |_|_|_PromSeriesDivideExec: tags=["host"] REDACTED -|_|_|_FilterExec: CASE WHEN host@1 IS NOT NULL THEN host@1 ELSE_END = REDACTED |_|_|_CooperativeExec REDACTED -|_|_|_SeriesScan: region=REDACTED, {"partition_count":{"count":1, "mem_ranges":1, "files":0, "file_ranges":0}, "distribution":"PerSeries", "projection": ["ts", "host", "val"], "filters": ["CASE WHEN host IS NOT NULL THEN host ELSE Dictionary(UInt32, Utf8(\"\")) END = Dictionary(UInt32, Utf8(\"\"))", "ts >= TimestampMillisecond(-299999, None)", "ts <= TimestampMillisecond(0, None)"], "REDACTED +|_|_|_SeriesScan: region=REDACTED, {"partition_count":{"count":1, "mem_ranges":1, "files":0, "file_ranges":0}, "selector":"LastRow { after_merge: true }", "distribution":"PerSeries", "projection": ["ts", "host", "val"], "filters": ["host IS NULL OR host = Dictionary(UInt32, Utf8(\"\"))", "ts >= TimestampMillisecond(-299999, None)", "ts <= TimestampMillisecond(0, None)"], "REDACTED |_|_|_| |_|_| Total rows: 0_| +-+-+-+ +INSERT INTO test VALUES (0, NULL, 2), (0, '', 3), (0, 'ext4', 4), (0, 'tmpfs', 5); + +Affected Rows: 4 + +TQL EVAL (0, 0, '1s') test{host!="tmpfs"}; + ++---------------------+-------------------+-----+ +| ts | host | val | ++---------------------+-------------------+-----+ +| 1970-01-01T00:00:00 | | 2 | +| 1970-01-01T00:00:00 | | 3 | +| 1970-01-01T00:00:00 | 10.0.160.237:8080 | 1 | +| 1970-01-01T00:00:00 | 10.0.160.237:8081 | 1 | +| 1970-01-01T00:00:00 | ext4 | 4 | ++---------------------+-------------------+-----+ + +TQL EVAL (0, 0, '1s') test{host!~"tmpfs|vfat"}; + ++---------------------+-------------------+-----+ +| ts | host | val | ++---------------------+-------------------+-----+ +| 1970-01-01T00:00:00 | | 2 | +| 1970-01-01T00:00:00 | | 3 | +| 1970-01-01T00:00:00 | 10.0.160.237:8080 | 1 | +| 1970-01-01T00:00:00 | 10.0.160.237:8081 | 1 | +| 1970-01-01T00:00:00 | ext4 | 4 | ++---------------------+-------------------+-----+ + +TQL EVAL (0, 0, '1s') test{host=""}; + ++---------------------+------+-----+ +| ts | host | val | ++---------------------+------+-----+ +| 1970-01-01T00:00:00 | | 2 | +| 1970-01-01T00:00:00 | | 3 | ++---------------------+------+-----+ + +TQL EVAL (0, 0, '1s') test{host!=""}; + ++---------------------+-------------------+-----+ +| ts | host | val | ++---------------------+-------------------+-----+ +| 1970-01-01T00:00:00 | 10.0.160.237:8080 | 1 | +| 1970-01-01T00:00:00 | 10.0.160.237:8081 | 1 | +| 1970-01-01T00:00:00 | ext4 | 4 | +| 1970-01-01T00:00:00 | tmpfs | 5 | ++---------------------+-------------------+-----+ + +TQL EVAL (0, 0, '1s') test{host=~"ext4|"}; + ++---------------------+------+-----+ +| ts | host | val | ++---------------------+------+-----+ +| 1970-01-01T00:00:00 | | 2 | +| 1970-01-01T00:00:00 | | 3 | +| 1970-01-01T00:00:00 | ext4 | 4 | ++---------------------+------+-----+ + +TQL EVAL (0, 0, '1s') test{host!~"ext4|"}; + ++---------------------+-------------------+-----+ +| ts | host | val | ++---------------------+-------------------+-----+ +| 1970-01-01T00:00:00 | 10.0.160.237:8080 | 1 | +| 1970-01-01T00:00:00 | 10.0.160.237:8081 | 1 | +| 1970-01-01T00:00:00 | tmpfs | 5 | ++---------------------+-------------------+-----+ + +TQL EVAL (0, 0, '1s') test{missing!="lo"}; + ++---------------------+-------------------+-----+ +| ts | host | val | ++---------------------+-------------------+-----+ +| 1970-01-01T00:00:00 | | 2 | +| 1970-01-01T00:00:00 | | 3 | +| 1970-01-01T00:00:00 | 10.0.160.237:8080 | 1 | +| 1970-01-01T00:00:00 | 10.0.160.237:8081 | 1 | +| 1970-01-01T00:00:00 | ext4 | 4 | +| 1970-01-01T00:00:00 | tmpfs | 5 | ++---------------------+-------------------+-----+ + DROP TABLE test; Affected Rows: 0 diff --git a/tests/cases/standalone/common/promql/regex.sql b/tests/cases/standalone/common/promql/regex.sql index 9804e098e64..80a8f3ed381 100644 --- a/tests/cases/standalone/common/promql/regex.sql +++ b/tests/cases/standalone/common/promql/regex.sql @@ -59,4 +59,14 @@ TQL ANALYZE VERBOSE (0, 0, '1s') test{host!~".*"}; -- SQLNESS REPLACE (flat_format.*) REDACTED TQL ANALYZE VERBOSE (0, 0, '1s') test{host!~".+"}; +INSERT INTO test VALUES (0, NULL, 2), (0, '', 3), (0, 'ext4', 4), (0, 'tmpfs', 5); + +TQL EVAL (0, 0, '1s') test{host!="tmpfs"}; +TQL EVAL (0, 0, '1s') test{host!~"tmpfs|vfat"}; +TQL EVAL (0, 0, '1s') test{host=""}; +TQL EVAL (0, 0, '1s') test{host!=""}; +TQL EVAL (0, 0, '1s') test{host=~"ext4|"}; +TQL EVAL (0, 0, '1s') test{host!~"ext4|"}; +TQL EVAL (0, 0, '1s') test{missing!="lo"}; + DROP TABLE test;