fix(promql): restore exact filtering for nullable label matchers (#9418)

* fix(promql): restore exact filtering for nullable label matchers

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>

* test: verify compound filter semantics and last-row selection

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>

* test: consolidate label filter regression coverage

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>

* refactor: separate filter evaluation and timestamp adaptation

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>

* refactor: inline predicate and timestamp tests

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>

* refactor: keep direct IN predicates outside simple filters

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>

* fix: coerce partition bounds before simple filtering

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>

* test: cover partition bound overflow after timestamp widening

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>

---------

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>
(cherry picked from commit 526dd48bda)
Signed-off-by: WenyXu <wenymedia@gmail.com>

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<ArrayRef>,
+        op: Operator,
+        regex: Option<Regex>,
+        regex_negative: bool,
+    },
+    OrEqChain {
+        literals: Vec<Scalar<ArrayRef>>,
+    },
+    IsNull {
+        negated: bool,
+    },
+    Constant(Option<bool>),
+    And(Box<Self>, Box<Self>),
+    Or(Box<Self>, Box<Self>),
+}
+
+impl SimplePredicate {
+    pub(crate) fn try_new(expr: &Expr, column_type: Option<&DataType>) -> Option<Self> {
+        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<Vec<Scalar<ArrayRef>>> {
+        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<BooleanArray> {
+        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<Regex>, 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<BooleanArray, ArrowError> {
+        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::<DictionaryArray<UInt32Type>>() {
+            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 <https://github.com/apache/arrow-rs/blob/54.2.0/arrow-string/src/regexp.rs#L204-L246> for the implementation details.
+pub fn regexp_is_match_scalar<'a, S>(
+    array: &'a S,
+    regex: Option<&Regex>,
+) -> Result<BooleanArray, ArrowError>
+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<UInt32Type>,
+    regex: Option<&Regex>,
+) -> Result<BooleanArray, ArrowError> {
+    // Get the string values from the dictionary
+    let string_values = dict_array
+        .values()
+        .as_any()
+        .downcast_ref::<datatypes::arrow::array::StringArray>()
+        .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::<UInt32Type>::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::<Int32Array>()
+                            .unwrap()
+                            .values()
+                            .to_vec()
+                    })
+                    .collect::<Vec<_>>();
+                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::<Vec<_>>()
+                };
+                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::<datatypes::arrow::array::StringArray>()
+            .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::<UInt32Type>::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(&regex)).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(&regex2)).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<bool>; 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()
+    }
+}

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::<datatypes::arrow::array::Int32Array>()
+        .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)
+    );
+}

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<Self> {
+        // 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<i64>, 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<Scalar<ArrayRef>> {
+    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<TimestampUnitCast> {
+        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<TimestampUnitCast>, values: &[i64]) -> Vec<bool> {
+        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::<Vec<_>>()
+        );
+    }
+}

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 |
 +-------+---------+---------------------+";

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::<TimestampMillisecondType>();
+                    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::<TimestampMillisecondType>()
+                                .values()
+                                .iter()
+                                .copied(),
+                        );
+                    }
+                    actual.sort_unstable();
+                    let expected = if selector.is_some() {
+                        expected.last().copied().into_iter().collect::<Vec<_>>()
+                    } 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.

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::<Vec<_>>()
             })
             .unwrap_or_default();

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()));

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<SimpleFilterContext> = predicate
         .into_iter()
         .flat_map(|predicate| {

src/mito2/src/sst/parquet/reader.rs

@@ -2014,7 +2014,12 @@ impl SimpleFilterContext {
         expected_meta: Option<&RegionMetadata>,
         expr: &Expr,
     ) -> Option<Self> {
-        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();

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::<Vec<_>>())
+                .unwrap(),
+            vec![
+                TableProviderFilterPushDown::Exact,
+                TableProviderFilterPushDown::Exact,
+                TableProviderFilterPushDown::Inexact,
+                TableProviderFilterPushDown::Inexact,
+                TableProviderFilterPushDown::Inexact,
+                TableProviderFilterPushDown::Inexact,
+            ]
+        );
+    }
+
     fn test_region_id() -> RegionId {
         RegionId::new(1024, 1)
     }

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);
         }
 

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}"
         );
     }

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

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;