query/optimizer/
const_normalization.rs

1// Copyright 2023 Greptime Team
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use std::sync::Arc;
16
17use arrow_schema::{DataType, TimeUnit as ArrowTimeUnit};
18use datafusion::config::ConfigOptions;
19use datafusion_common::tree_node::{Transformed, TreeNode, TreeNodeRecursion, TreeNodeRewriter};
20use datafusion_common::{DFSchemaRef, Result, ScalarValue};
21use datafusion_expr::expr::{Cast, InList, Like, TryCast};
22use datafusion_expr::{Between, BinaryExpr, Expr, ExprSchemable, LogicalPlan, Operator, lit};
23use datafusion_expr_common::casts::try_cast_literal_to_type;
24use datafusion_optimizer::analyzer::AnalyzerRule;
25
26use crate::plan::ExtractExpr;
27
28/// ConstNormalizationRule rewrites castable constants against their
29/// non-constant comparison operand ahead of filter pushdown.
30#[derive(Debug)]
31pub struct ConstNormalizationRule;
32
33impl AnalyzerRule for ConstNormalizationRule {
34    fn analyze(&self, plan: LogicalPlan, _config: &ConfigOptions) -> Result<LogicalPlan> {
35        plan.transform(|plan| match plan {
36            LogicalPlan::Filter(filter) => {
37                let schema = filter.input.schema().clone();
38                rewrite_plan_exprs(LogicalPlan::Filter(filter), schema)
39            }
40            LogicalPlan::TableScan(scan) => {
41                let schema = scan.projected_schema.clone();
42                rewrite_plan_exprs(LogicalPlan::TableScan(scan), schema)
43            }
44            _ => Ok(Transformed::no(plan)),
45        })
46        .map(|x| x.data)
47    }
48
49    fn name(&self) -> &str {
50        "ConstNormalizationRule"
51    }
52}
53
54fn rewrite_plan_exprs(plan: LogicalPlan, schema: DFSchemaRef) -> Result<Transformed<LogicalPlan>> {
55    let mut rewriter = ConstNormalizationRewriter {
56        schema,
57        transformed: false,
58    };
59    let exprs = plan
60        .expressions_consider_join()
61        .into_iter()
62        .map(|expr| expr.rewrite(&mut rewriter).map(|rewritten| rewritten.data))
63        .collect::<Result<Vec<_>>>()?;
64    if !rewriter.transformed {
65        return Ok(Transformed::no(plan));
66    }
67
68    let inputs = plan.inputs().into_iter().cloned().collect::<Vec<_>>();
69    plan.with_new_exprs(exprs, inputs).map(Transformed::yes)
70}
71
72struct ConstNormalizationRewriter {
73    schema: DFSchemaRef,
74    transformed: bool,
75}
76
77impl TreeNodeRewriter for ConstNormalizationRewriter {
78    type Node = Expr;
79
80    fn f_down(&mut self, expr: Expr) -> Result<Transformed<Expr>> {
81        let recursion = if matches!(
82            expr,
83            Expr::Exists(_) | Expr::InSubquery(_) | Expr::ScalarSubquery(_)
84        ) {
85            TreeNodeRecursion::Jump
86        } else {
87            TreeNodeRecursion::Continue
88        };
89
90        Ok(Transformed::new(expr, false, recursion))
91    }
92
93    fn f_up(&mut self, expr: Expr) -> Result<Transformed<Expr>> {
94        let rewritten = rewrite_expr_node(expr, &self.schema)?;
95        self.transformed |= rewritten.transformed;
96        Ok(rewritten)
97    }
98}
99
100fn rewrite_expr_node(expr: Expr, schema: &DFSchemaRef) -> Result<Transformed<Expr>> {
101    match expr {
102        Expr::BinaryExpr(binary) => match rewrite_binary_expr(binary.clone(), schema)? {
103            Some(expr) => Ok(Transformed::yes(expr)),
104            None => Ok(Transformed::no(Expr::BinaryExpr(binary))),
105        },
106        Expr::Between(between) => match rewrite_between_expr(between.clone(), schema)? {
107            Some(expr) => Ok(Transformed::yes(expr)),
108            None => Ok(Transformed::no(Expr::Between(between))),
109        },
110        Expr::InList(in_list) => match rewrite_in_list_expr(in_list.clone(), schema)? {
111            Some(expr) => Ok(Transformed::yes(expr)),
112            None => Ok(Transformed::no(Expr::InList(in_list))),
113        },
114        Expr::Like(like) => rewrite_like_expr(like, PatternMatchKind::Like, schema),
115        Expr::SimilarTo(like) => rewrite_like_expr(like, PatternMatchKind::SimilarTo, schema),
116        expr => Ok(Transformed::no(expr)),
117    }
118}
119
120fn rewrite_between_expr(between: Between, schema: &DFSchemaRef) -> Result<Option<Expr>> {
121    let Between {
122        expr,
123        negated,
124        low,
125        high,
126    } = between;
127    let expr = *expr;
128    let low_expr = *low;
129    let high_expr = *high;
130    let Some((target, constants)) =
131        extract_rewrite_operands(&expr, &[low_expr.clone(), high_expr.clone()], schema)?
132    else {
133        return Ok(None);
134    };
135
136    if let Some(mut constants) = target.normalize_constants(&constants) {
137        let high = constants
138            .pop()
139            .expect("between normalization expects high constant");
140        let low = constants
141            .pop()
142            .expect("between normalization expects low constant");
143        return Ok(Some(Expr::Between(Between {
144            expr: Box::new(target.expr.clone()),
145            negated,
146            low: Box::new(lit(low)),
147            high: Box::new(lit(high)),
148        })));
149    }
150
151    Ok((!negated)
152        .then(|| target.normalize_timestamp_between(&constants[0], &constants[1]))
153        .flatten())
154}
155
156fn rewrite_in_list_expr(in_list: InList, schema: &DFSchemaRef) -> Result<Option<Expr>> {
157    let InList {
158        expr,
159        list,
160        negated,
161    } = in_list;
162    let expr = *expr;
163    let Some((target, constants)) = extract_rewrite_operands(&expr, &list, schema)? else {
164        return Ok(None);
165    };
166
167    Ok(target.normalize_constants(&constants).map(|constants| {
168        target
169            .expr
170            .clone()
171            .in_list(constants.into_iter().map(lit).collect(), negated)
172    }))
173}
174
175fn rewrite_like_expr(
176    like: Like,
177    kind: PatternMatchKind,
178    schema: &DFSchemaRef,
179) -> Result<Transformed<Expr>> {
180    let original = match kind {
181        PatternMatchKind::Like => Expr::Like(like.clone()),
182        PatternMatchKind::SimilarTo => Expr::SimilarTo(like.clone()),
183    };
184    let Like {
185        negated,
186        expr,
187        pattern,
188        escape_char,
189        case_insensitive,
190    } = like;
191    let expr = *expr;
192    let pattern = *pattern;
193    let Some((target, constants)) =
194        extract_rewrite_operands(&expr, std::slice::from_ref(&pattern), schema)?
195    else {
196        return Ok(Transformed::no(original));
197    };
198    let Some(mut constants) = target.normalize_constants(&constants) else {
199        return Ok(Transformed::no(original));
200    };
201
202    let pattern = lit(constants
203        .pop()
204        .expect("pattern normalization expects one constant"));
205    let like = Like::new(
206        negated,
207        Box::new(target.expr.clone()),
208        Box::new(pattern),
209        escape_char,
210        case_insensitive,
211    );
212    let rewritten = match kind {
213        PatternMatchKind::Like => Expr::Like(like),
214        PatternMatchKind::SimilarTo => Expr::SimilarTo(like),
215    };
216    Ok(Transformed::yes(rewritten))
217}
218
219fn rewrite_binary_expr(binary: BinaryExpr, schema: &DFSchemaRef) -> Result<Option<Expr>> {
220    if !binary.op.supports_propagation() {
221        return Ok(None);
222    }
223
224    let BinaryExpr { left, op, right } = binary;
225    let left = *left;
226    let right = *right;
227    if let Some(expr) = rewrite_binary_side(left.clone(), op, right.clone(), schema)? {
228        return Ok(Some(expr));
229    }
230
231    let Some(swapped_op) = op.swap() else {
232        return Ok(None);
233    };
234
235    rewrite_binary_side(right, swapped_op, left, schema)
236}
237
238fn rewrite_binary_side(
239    target_expr: Expr,
240    op: Operator,
241    constant_expr: Expr,
242    schema: &DFSchemaRef,
243) -> Result<Option<Expr>> {
244    let Some((target, constants)) =
245        extract_rewrite_operands(&target_expr, std::slice::from_ref(&constant_expr), schema)?
246    else {
247        return Ok(None);
248    };
249
250    if let Some(mut constants) = target.normalize_constants(&constants) {
251        let constant = constants
252            .pop()
253            .expect("binary normalization expects one constant");
254        return Ok(Some(Expr::BinaryExpr(BinaryExpr {
255            left: Box::new(target.expr.clone()),
256            op,
257            right: Box::new(lit(constant)),
258        })));
259    }
260
261    Ok(target.normalize_timestamp_binary(op, &constants[0]))
262}
263
264fn extract_rewrite_operands(
265    target_expr: &Expr,
266    constant_exprs: &[Expr],
267    schema: &DFSchemaRef,
268) -> Result<Option<(NormalizationTarget, Vec<ScalarValue>)>> {
269    let Some(target) = extract_normalization_target(target_expr, schema)? else {
270        return Ok(None);
271    };
272
273    extract_constant_scalars(constant_exprs)
274        .map(|constants| constants.map(|constants| (target, constants)))
275}
276
277#[derive(Clone)]
278struct NormalizationTarget {
279    expr: Expr,
280    data_type: DataType,
281    kind: NormalizationKind,
282}
283
284#[derive(Clone)]
285enum NormalizationKind {
286    /// The cast preserves every source value exactly, so literals can be cast directly.
287    Lossless,
288    /// The cast drops timestamp precision and must widen predicate bounds to preserve semantics.
289    TimestampDowncast {
290        source_unit: ArrowTimeUnit,
291        target_unit: ArrowTimeUnit,
292        timezone: Option<Arc<str>>,
293    },
294}
295
296impl NormalizationTarget {
297    /// Normalizes constants for rewrites that can preserve the original predicate with a direct
298    /// literal cast. Timestamp precision-changing casts are handled by timestamp-specific helpers.
299    fn normalize_constants(&self, constants: &[ScalarValue]) -> Option<Vec<ScalarValue>> {
300        constants
301            .iter()
302            .map(|constant| self.normalize_constant(constant))
303            .collect()
304    }
305
306    fn normalize_constant(&self, constant: &ScalarValue) -> Option<ScalarValue> {
307        match self.kind {
308            NormalizationKind::TimestampDowncast { .. } => None,
309            NormalizationKind::Lossless => try_cast_literal_to_type(constant, &self.data_type),
310        }
311    }
312
313    /// Rewrites predicates over timestamp downcasts into source-side half-open bounds.
314    fn normalize_timestamp_binary(&self, op: Operator, constant: &ScalarValue) -> Option<Expr> {
315        let NormalizationKind::TimestampDowncast {
316            source_unit,
317            target_unit,
318            timezone,
319        } = &self.kind
320        else {
321            return None;
322        };
323
324        let constant = constant
325            .cast_to(&DataType::Timestamp(*target_unit, timezone.clone()))
326            .ok()?;
327        let value = timestamp_scalar_value(&constant)?;
328        let bound = match op {
329            Operator::GtEq => lower_bound_for_ge(value, *source_unit, *target_unit)?,
330            Operator::Gt => lower_bound_for_ge(value.checked_add(1)?, *source_unit, *target_unit)?,
331            Operator::Lt => lower_bound_for_ge(value, *source_unit, *target_unit)?,
332            Operator::LtEq => {
333                lower_bound_for_ge(value.checked_add(1)?, *source_unit, *target_unit)?
334            }
335            _ => return None,
336        };
337
338        let normalized_op = match op {
339            Operator::GtEq | Operator::Gt => Operator::GtEq,
340            Operator::Lt | Operator::LtEq => Operator::Lt,
341            _ => return None,
342        };
343
344        Some(match normalized_op {
345            Operator::GtEq => self.expr.clone().gt_eq(lit(timestamp_scalar(
346                *source_unit,
347                timezone.clone(),
348                bound,
349            ))),
350            Operator::Lt => {
351                self.expr
352                    .clone()
353                    .lt(lit(timestamp_scalar(*source_unit, timezone.clone(), bound)))
354            }
355            _ => unreachable!("timestamp normalization only rewrites to >= or <"),
356        })
357    }
358
359    /// Rewrites `BETWEEN` over timestamp downcasts into an inclusive lower bound and exclusive
360    /// upper bound over the source timestamp unit.
361    fn normalize_timestamp_between(&self, low: &ScalarValue, high: &ScalarValue) -> Option<Expr> {
362        let NormalizationKind::TimestampDowncast {
363            source_unit,
364            target_unit,
365            timezone,
366        } = &self.kind
367        else {
368            return None;
369        };
370
371        let target_type = DataType::Timestamp(*target_unit, timezone.clone());
372        let low = low.cast_to(&target_type).ok()?;
373        let high = high.cast_to(&target_type).ok()?;
374        let low = timestamp_scalar_value(&low)?;
375        let high = timestamp_scalar_value(&high)?;
376
377        let lower = lower_bound_for_ge(low, *source_unit, *target_unit)?;
378        let upper = lower_bound_for_ge(high.checked_add(1)?, *source_unit, *target_unit)?;
379
380        Some(
381            self.expr
382                .clone()
383                .gt_eq(lit(timestamp_scalar(*source_unit, timezone.clone(), lower)))
384                .and(self.expr.clone().lt(lit(timestamp_scalar(
385                    *source_unit,
386                    timezone.clone(),
387                    upper,
388                )))),
389        )
390    }
391}
392
393/// Returns the non-constant side we should normalize against.
394///
395/// Plain expressions normalize literals to their own type. Cast expressions only participate when
396/// the cast is lossless or when timestamp downcasts can be rewritten as wider source-side bounds.
397fn extract_normalization_target(
398    expr: &Expr,
399    schema: &DFSchemaRef,
400) -> Result<Option<NormalizationTarget>> {
401    if extract_constant_scalar(expr)?.is_some() {
402        return Ok(None);
403    }
404
405    let Some((_, source_expr, target_type)) = extract_cast_input(expr) else {
406        return Ok(Some(NormalizationTarget {
407            expr: expr.clone(),
408            data_type: expr.get_type(schema)?,
409            kind: NormalizationKind::Lossless,
410        }));
411    };
412
413    let data_type = source_expr.get_type(schema)?;
414    let Some(kind) = classify_normalization_kind(&data_type, target_type) else {
415        return Ok(None);
416    };
417
418    Ok(Some(NormalizationTarget {
419        expr: source_expr.clone(),
420        data_type,
421        kind,
422    }))
423}
424
425fn classify_normalization_kind(
426    source_type: &DataType,
427    target_type: &DataType,
428) -> Option<NormalizationKind> {
429    // Timestamp casts that change precision need boundary-aware rewrites. A finer target literal
430    // may not map exactly back to the coarser source unit, so the generic lossless path is only
431    // safe for timestamp casts that keep the same unit.
432    if is_lossless_cast(source_type, target_type) {
433        return Some(NormalizationKind::Lossless);
434    }
435
436    match (source_type, target_type) {
437        (
438            DataType::Timestamp(source_unit, source_tz),
439            DataType::Timestamp(target_unit, target_tz),
440        ) if source_tz == target_tz
441            && time_unit_rank(*source_unit) > time_unit_rank(*target_unit) =>
442        {
443            Some(NormalizationKind::TimestampDowncast {
444                source_unit: *source_unit,
445                target_unit: *target_unit,
446                timezone: source_tz.clone(),
447            })
448        }
449        _ => None,
450    }
451}
452
453/// Returns whether every value of `source_type` is representable in `target_type`.
454fn is_lossless_cast(source_type: &DataType, target_type: &DataType) -> bool {
455    match (source_type, target_type) {
456        (DataType::Int8, DataType::Int16 | DataType::Int32 | DataType::Int64)
457        | (DataType::Int16, DataType::Int32 | DataType::Int64)
458        | (DataType::Int32, DataType::Int64)
459        | (DataType::UInt8, DataType::UInt16 | DataType::UInt32 | DataType::UInt64)
460        | (DataType::UInt8, DataType::Int16 | DataType::Int32 | DataType::Int64)
461        | (DataType::UInt16, DataType::UInt32 | DataType::UInt64)
462        | (DataType::UInt16, DataType::Int32 | DataType::Int64)
463        | (DataType::UInt32, DataType::UInt64 | DataType::Int64)
464        | (DataType::Utf8, DataType::Utf8View | DataType::LargeUtf8) => true,
465        (
466            DataType::Timestamp(source_unit, source_tz),
467            DataType::Timestamp(target_unit, target_tz),
468        ) => source_tz == target_tz && source_unit == target_unit,
469        _ => false,
470    }
471}
472
473#[derive(Clone, Copy)]
474enum PatternMatchKind {
475    Like,
476    SimilarTo,
477}
478
479fn extract_constant_scalars(exprs: &[Expr]) -> Result<Option<Vec<ScalarValue>>> {
480    let mut values = Vec::with_capacity(exprs.len());
481    for expr in exprs {
482        let Some(value) = extract_constant_scalar(expr)? else {
483            return Ok(None);
484        };
485        values.push(value);
486    }
487
488    Ok(Some(values))
489}
490
491/// Extracts a literal scalar from an expression, folding constant `CAST` and `TRY_CAST` nodes.
492fn extract_constant_scalar(expr: &Expr) -> Result<Option<ScalarValue>> {
493    if let Some(value) = expr.as_literal() {
494        return Ok(Some(value.clone()));
495    }
496
497    let Some((kind, expr, data_type)) = extract_cast_input(expr) else {
498        return Ok(None);
499    };
500
501    match kind {
502        CastInputKind::Cast => extract_constant_scalar(expr)?
503            .map(|value| value.cast_to(data_type))
504            .transpose(),
505        CastInputKind::TryCast => {
506            Ok(extract_constant_scalar(expr)?.and_then(|value| value.cast_to(data_type).ok()))
507        }
508    }
509}
510
511#[derive(Clone, Copy)]
512enum CastInputKind {
513    Cast,
514    TryCast,
515}
516
517/// Returns the input expression and target type for `CAST` and `TRY_CAST` expressions.
518fn extract_cast_input(expr: &Expr) -> Option<(CastInputKind, &Expr, &DataType)> {
519    match expr {
520        Expr::Cast(Cast { expr, data_type }) => {
521            Some((CastInputKind::Cast, expr.as_ref(), data_type))
522        }
523        Expr::TryCast(TryCast { expr, data_type }) => {
524            Some((CastInputKind::TryCast, expr.as_ref(), data_type))
525        }
526        _ => None,
527    }
528}
529
530fn time_unit_rank(unit: ArrowTimeUnit) -> usize {
531    match unit {
532        ArrowTimeUnit::Second => 0,
533        ArrowTimeUnit::Millisecond => 1,
534        ArrowTimeUnit::Microsecond => 2,
535        ArrowTimeUnit::Nanosecond => 3,
536    }
537}
538
539fn time_unit_scale(unit: ArrowTimeUnit) -> i64 {
540    match unit {
541        ArrowTimeUnit::Second => 1,
542        ArrowTimeUnit::Millisecond => 1_000,
543        ArrowTimeUnit::Microsecond => 1_000_000,
544        ArrowTimeUnit::Nanosecond => 1_000_000_000,
545    }
546}
547
548/// Returns the number of source-unit ticks in one target-unit tick for finer-to-coarser casts.
549fn finer_to_coarser_ratio(source_unit: ArrowTimeUnit, target_unit: ArrowTimeUnit) -> Option<i64> {
550    let source_scale = time_unit_scale(source_unit);
551    let target_scale = time_unit_scale(target_unit);
552    (source_scale >= target_scale).then_some(source_scale / target_scale)
553}
554
555/// Returns the smallest source-unit timestamp whose downcast is greater than or equal to
556/// `target_value`.
557///
558/// DataFusion timestamp downcasts truncate toward zero. For non-positive buckets that means the
559/// bucket starts before `target_value * ratio`, so `<= x` can be rewritten as `< lower_bound(x+1)`
560/// without dropping rows near zero or across negative boundaries.
561fn lower_bound_for_ge(
562    target_value: i64,
563    source_unit: ArrowTimeUnit,
564    target_unit: ArrowTimeUnit,
565) -> Option<i64> {
566    let ratio = finer_to_coarser_ratio(source_unit, target_unit)?;
567    let base = target_value.checked_mul(ratio)?;
568    if target_value <= 0 {
569        base.checked_sub(ratio - 1)
570    } else {
571        Some(base)
572    }
573}
574
575fn timestamp_scalar_value(value: &ScalarValue) -> Option<i64> {
576    match value {
577        ScalarValue::TimestampSecond(Some(value), _)
578        | ScalarValue::TimestampMillisecond(Some(value), _)
579        | ScalarValue::TimestampMicrosecond(Some(value), _)
580        | ScalarValue::TimestampNanosecond(Some(value), _) => Some(*value),
581        _ => None,
582    }
583}
584
585fn timestamp_scalar(unit: ArrowTimeUnit, timezone: Option<Arc<str>>, value: i64) -> ScalarValue {
586    match unit {
587        ArrowTimeUnit::Second => ScalarValue::TimestampSecond(Some(value), timezone),
588        ArrowTimeUnit::Millisecond => ScalarValue::TimestampMillisecond(Some(value), timezone),
589        ArrowTimeUnit::Microsecond => ScalarValue::TimestampMicrosecond(Some(value), timezone),
590        ArrowTimeUnit::Nanosecond => ScalarValue::TimestampNanosecond(Some(value), timezone),
591    }
592}
593
594#[cfg(test)]
595mod tests {
596    use std::sync::Arc;
597
598    use arrow_schema::{DataType, TimeUnit as ArrowTimeUnit};
599    use async_trait::async_trait;
600    use common_time::Timestamp;
601    use common_time::range::TimestampRange;
602    use common_time::timestamp::TimeUnit;
603    use datafusion::catalog::Session;
604    use datafusion::config::ConfigOptions;
605    use datafusion::datasource::{TableProvider, provider_as_source};
606    use datafusion::physical_plan::ExecutionPlan;
607    use datafusion_common::arrow::datatypes::Field;
608    use datafusion_common::{DFSchema, ScalarValue, ToDFSchema};
609    use datafusion_expr::expr::{Between, Like};
610    use datafusion_expr::expr_fn::{cast, col, try_cast};
611    use datafusion_expr::{
612        Expr, LogicalPlan, LogicalPlanBuilder, TableProviderFilterPushDown, TableScan, TableSource,
613        TableType, lit,
614    };
615    use datafusion_optimizer::analyzer::AnalyzerRule;
616    use datafusion_optimizer::optimizer::{Optimizer, OptimizerContext};
617    use datafusion_optimizer::push_down_filter::PushDownFilter;
618    use datafusion_optimizer::simplify_expressions::SimplifyExpressions;
619    use table::predicate::build_time_range_predicate;
620
621    use super::{
622        ConstNormalizationRule, PatternMatchKind, lower_bound_for_ge, try_cast_literal_to_type,
623    };
624
625    #[test]
626    fn test_normalize_direct_integer_cast_comparison() {
627        assert_filter_plan(
628            vec![Field::new("v", DataType::Int32, false)],
629            cast(col("v"), DataType::Int64).gt_eq(lit(42_i64)),
630            "Filter: t.v >= Int32(42)\n  TableScan: t",
631        );
632    }
633
634    #[test]
635    fn test_normalize_non_column_operand() {
636        assert_filter_plan(
637            vec![Field::new("v", DataType::Int32, false)],
638            cast(col("v") + lit(1_i32), DataType::Int64).gt_eq(lit(42_i64)),
639            "Filter: t.v + Int32(1) >= Int32(42)\n  TableScan: t",
640        );
641    }
642
643    #[test]
644    fn test_normalize_swapped_binary_comparison() {
645        assert_filter_plan(
646            vec![Field::new("v", DataType::Int16, false)],
647            lit(42_i64).lt_eq(cast(col("v"), DataType::Int64)),
648            "Filter: t.v >= Int16(42)\n  TableScan: t",
649        );
650    }
651
652    #[test]
653    fn test_normalize_try_cast_target() {
654        assert_filter_plan(
655            vec![Field::new("v", DataType::Int16, false)],
656            try_cast(col("v"), DataType::Int64).gt_eq(lit(42_i64)),
657            "Filter: t.v >= Int16(42)\n  TableScan: t",
658        );
659    }
660
661    #[test]
662    fn test_normalize_casted_constants() {
663        let fields = vec![Field::new("v", DataType::Int16, false)];
664        let cases = [
665            (
666                col("v").gt_eq(cast(lit(42_i8), DataType::Int64)),
667                "Filter: t.v >= Int16(42)\n  TableScan: t",
668            ),
669            (
670                col("v").in_list(
671                    vec![
672                        cast(lit(1_i8), DataType::Int64),
673                        try_cast(lit(2_i8), DataType::Int64),
674                    ],
675                    false,
676                ),
677                "Filter: t.v IN ([Int16(1), Int16(2)])\n  TableScan: t",
678            ),
679        ];
680
681        for (predicate, expected) in cases {
682            assert_filter_plan(fields.clone(), predicate, expected);
683        }
684    }
685
686    #[test]
687    fn test_normalize_plain_integer_literals() {
688        let fields = vec![Field::new("v", DataType::Int16, false)];
689        let cases = [
690            (
691                col("v").gt_eq(lit(42_i64)),
692                "Filter: t.v >= Int16(42)\n  TableScan: t",
693            ),
694            (
695                col("v").in_list(vec![lit(1_i64), lit(2_i64)], false),
696                "Filter: t.v IN ([Int16(1), Int16(2)])\n  TableScan: t",
697            ),
698            (
699                col("v").between(lit(3_i64), lit(5_i64)),
700                "Filter: t.v BETWEEN Int16(3) AND Int16(5)\n  TableScan: t",
701            ),
702        ];
703
704        for (predicate, expected) in cases {
705            assert_filter_plan(fields.clone(), predicate, expected);
706        }
707    }
708
709    #[test]
710    fn test_normalize_unsigned_to_signed_literals() {
711        let cases = [
712            (
713                vec![Field::new("v", DataType::UInt8, false)],
714                cast(col("v"), DataType::Int16).lt_eq(lit(255_i16)),
715                "Filter: t.v <= UInt8(255)\n  TableScan: t",
716            ),
717            (
718                vec![Field::new("v", DataType::UInt16, false)],
719                cast(col("v"), DataType::Int32).gt_eq(lit(42_i32)),
720                "Filter: t.v >= UInt16(42)\n  TableScan: t",
721            ),
722            (
723                vec![Field::new("v", DataType::UInt32, false)],
724                cast(col("v"), DataType::Int64).between(lit(3_i64), lit(5_i64)),
725                "Filter: t.v BETWEEN UInt32(3) AND UInt32(5)\n  TableScan: t",
726            ),
727        ];
728
729        for (fields, predicate, expected) in cases {
730            assert_filter_plan(fields, predicate, expected);
731        }
732    }
733
734    #[test]
735    fn test_normalize_in_list_and_between() {
736        let fields = vec![Field::new("v", DataType::Int16, false)];
737        let cases = [
738            (
739                cast(col("v"), DataType::Int64).in_list(vec![lit(1_i64), lit(2_i64)], false),
740                "Filter: t.v IN ([Int16(1), Int16(2)])\n  TableScan: t",
741            ),
742            (
743                cast(col("v"), DataType::Int64).between(lit(3_i64), lit(5_i64)),
744                "Filter: t.v BETWEEN Int16(3) AND Int16(5)\n  TableScan: t",
745            ),
746        ];
747
748        for (predicate, expected) in cases {
749            assert_filter_plan(fields.clone(), predicate, expected);
750        }
751    }
752
753    #[test]
754    fn test_keep_non_lossless_literal_unchanged() {
755        assert_filter_plan(
756            vec![Field::new("v", DataType::Int16, false)],
757            col("v").gt_eq(lit(100_000_i64)),
758            "Filter: t.v >= Int64(100000)\n  TableScan: t",
759        );
760    }
761
762    #[test]
763    fn test_normalize_scan_filters() {
764        let scan = build_scan_plan(test_schema(vec![Field::new("v", DataType::Int16, false)]));
765        let LogicalPlan::TableScan(scan) = scan else {
766            panic!("expected table scan");
767        };
768        let plan = LogicalPlan::TableScan(TableScan {
769            filters: vec![cast(col("v"), DataType::Int64).gt_eq(lit(42_i64))],
770            ..scan
771        });
772
773        let analyzed = analyze_plan(plan);
774
775        assert_eq!(
776            vec![col("v").gt_eq(lit(42_i16))],
777            extract_scan_filters(&analyzed)
778        );
779    }
780
781    #[test]
782    fn test_normalize_negated_between() {
783        assert_filter_plan(
784            vec![Field::new("v", DataType::Int16, false)],
785            Expr::Between(Between {
786                expr: Box::new(cast(col("v"), DataType::Int64)),
787                negated: true,
788                low: Box::new(lit(3_i64)),
789                high: Box::new(lit(5_i64)),
790            }),
791            "Filter: t.v NOT BETWEEN Int16(3) AND Int16(5)\n  TableScan: t",
792        );
793    }
794
795    #[test]
796    fn test_normalize_like_literal() {
797        assert_pattern_match_plan(
798            PatternMatchKind::Like,
799            ScalarValue::LargeUtf8(Some("api%".to_string())),
800            "Filter: t.s LIKE Utf8(\"api%\")\n  TableScan: t",
801        );
802    }
803
804    #[test]
805    fn test_normalize_similar_to_literal() {
806        assert_pattern_match_plan(
807            PatternMatchKind::SimilarTo,
808            ScalarValue::LargeUtf8(Some("api.*".to_string())),
809            "Filter: t.s SIMILAR TO Utf8(\"api.*\")\n  TableScan: t",
810        );
811    }
812
813    #[test]
814    fn test_normalize_direct_timestamp_filter() {
815        assert_timestamp_pushdown(
816            vec![
817                Field::new(
818                    "ts",
819                    DataType::Timestamp(ArrowTimeUnit::Nanosecond, None),
820                    false,
821                ),
822                Field::new("tag", DataType::Utf8, true),
823            ],
824            ts_cast_to_ms()
825                .gt_eq(ts_ms_literal(-299_999))
826                .and(ts_cast_to_ms().lt_eq(ts_ms_literal(10_000)))
827                .and(col("tag").eq(lit("api"))),
828            "Filter: t.ts >= TimestampNanosecond(-299999999999, None) AND t.ts < TimestampNanosecond(10001000000, None) AND t.tag = Utf8(\"api\")\n  TableScan: t",
829            "TableScan: t, full_filters=[t.ts >= TimestampNanosecond(-299999999999, None), t.ts < TimestampNanosecond(10001000000, None), t.tag = Utf8(\"api\")]",
830            TimestampRange::new_inclusive(
831                Some(Timestamp::new_nanosecond(-299_999_999_999)),
832                Some(Timestamp::new_nanosecond(10_000_999_999)),
833            ),
834        );
835    }
836
837    #[test]
838    fn test_normalize_timestamp_between_filter() {
839        assert_timestamp_pushdown(
840            vec![Field::new(
841                "ts",
842                DataType::Timestamp(ArrowTimeUnit::Nanosecond, None),
843                false,
844            )],
845            ts_cast_to_ms().between(ts_ms_literal(-299_999), ts_ms_literal(10_000)),
846            "Filter: t.ts >= TimestampNanosecond(-299999999999, None) AND t.ts < TimestampNanosecond(10001000000, None)\n  TableScan: t",
847            "TableScan: t, full_filters=[t.ts >= TimestampNanosecond(-299999999999, None), t.ts < TimestampNanosecond(10001000000, None)]",
848            TimestampRange::new_inclusive(
849                Some(Timestamp::new_nanosecond(-299_999_999_999)),
850                Some(Timestamp::new_nanosecond(10_000_999_999)),
851            ),
852        );
853    }
854
855    #[test]
856    fn test_normalize_strict_timestamp_filter() {
857        assert_timestamp_pushdown(
858            vec![Field::new(
859                "ts",
860                DataType::Timestamp(ArrowTimeUnit::Nanosecond, None),
861                false,
862            )],
863            ts_cast_to_ms()
864                .gt(ts_ms_literal(10_000))
865                .and(ts_cast_to_ms().lt(ts_ms_literal(20_000))),
866            "Filter: t.ts >= TimestampNanosecond(10001000000, None) AND t.ts < TimestampNanosecond(20000000000, None)\n  TableScan: t",
867            "TableScan: t, full_filters=[t.ts >= TimestampNanosecond(10001000000, None), t.ts < TimestampNanosecond(20000000000, None)]",
868            TimestampRange::new_inclusive(
869                Some(Timestamp::new_nanosecond(10_001_000_000)),
870                Some(Timestamp::new_nanosecond(19_999_999_999)),
871            ),
872        );
873    }
874
875    #[test]
876    fn test_normalize_zero_boundary_timestamp_filter() {
877        let fields = vec![Field::new(
878            "ts",
879            DataType::Timestamp(ArrowTimeUnit::Nanosecond, None),
880            false,
881        )];
882
883        assert_timestamp_pushdown(
884            fields.clone(),
885            ts_cast_to_ms().gt_eq(ts_ms_literal(0)),
886            "Filter: t.ts >= TimestampNanosecond(-999999, None)\n  TableScan: t",
887            "TableScan: t, full_filters=[t.ts >= TimestampNanosecond(-999999, None)]",
888            TimestampRange::from_start(Timestamp::new_nanosecond(-999_999)),
889        );
890
891        assert_timestamp_pushdown(
892            fields.clone(),
893            ts_cast_to_ms().lt(ts_ms_literal(0)),
894            "Filter: t.ts < TimestampNanosecond(-999999, None)\n  TableScan: t",
895            "TableScan: t, full_filters=[t.ts < TimestampNanosecond(-999999, None)]",
896            TimestampRange::until_end(Timestamp::new_nanosecond(-999_999), false),
897        );
898
899        assert_timestamp_pushdown(
900            fields,
901            ts_cast_to_ms().between(ts_ms_literal(0), ts_ms_literal(0)),
902            "Filter: t.ts >= TimestampNanosecond(-999999, None) AND t.ts < TimestampNanosecond(1000000, None)\n  TableScan: t",
903            "TableScan: t, full_filters=[t.ts >= TimestampNanosecond(-999999, None), t.ts < TimestampNanosecond(1000000, None)]",
904            TimestampRange::new_inclusive(
905                Some(Timestamp::new_nanosecond(-999_999)),
906                Some(Timestamp::new_nanosecond(999_999)),
907            ),
908        );
909    }
910
911    #[test]
912    fn test_timestamp_downcast_contract_matches_datafusion_casts() {
913        let cases = [
914            (-1_000_001, -1),
915            (-1_000_000, -1),
916            (-999_999, 0),
917            (-1, 0),
918            (0, 0),
919            (999_999, 0),
920            (1_000_000, 1),
921        ];
922
923        for (source, expected) in cases {
924            let casted = try_cast_literal_to_type(
925                &ScalarValue::TimestampNanosecond(Some(source), None),
926                &DataType::Timestamp(ArrowTimeUnit::Millisecond, None),
927            )
928            .unwrap();
929            assert_eq!(
930                ScalarValue::TimestampMillisecond(Some(expected), None),
931                casted
932            );
933        }
934
935        assert_eq!(
936            Some(-1_999_999),
937            lower_bound_for_ge(-1, ArrowTimeUnit::Nanosecond, ArrowTimeUnit::Millisecond)
938        );
939        assert_eq!(
940            Some(-999_999),
941            lower_bound_for_ge(0, ArrowTimeUnit::Nanosecond, ArrowTimeUnit::Millisecond)
942        );
943        assert_eq!(
944            Some(1_000_000),
945            lower_bound_for_ge(1, ArrowTimeUnit::Nanosecond, ArrowTimeUnit::Millisecond)
946        );
947    }
948
949    #[test]
950    fn test_normalize_plain_timestamp_literals() {
951        assert_timestamp_pushdown(
952            vec![Field::new(
953                "ts",
954                DataType::Timestamp(ArrowTimeUnit::Nanosecond, None),
955                false,
956            )],
957            col("ts")
958                .gt_eq(ts_ms_literal(-299_999))
959                .and(col("ts").lt_eq(ts_ms_literal(10_000))),
960            "Filter: t.ts >= TimestampNanosecond(-299999000000, None) AND t.ts <= TimestampNanosecond(10000000000, None)\n  TableScan: t",
961            "TableScan: t, full_filters=[t.ts >= TimestampNanosecond(-299999000000, None), t.ts <= TimestampNanosecond(10000000000, None)]",
962            TimestampRange::new_inclusive(
963                Some(Timestamp::new_nanosecond(-299_999_000_000)),
964                Some(Timestamp::new_nanosecond(10_000_000_000)),
965            ),
966        );
967    }
968
969    #[test]
970    fn test_keep_timestamp_upcast_filter_unchanged() {
971        assert_filter_plan(
972            vec![Field::new(
973                "ts",
974                DataType::Timestamp(ArrowTimeUnit::Millisecond, None),
975                false,
976            )],
977            cast(
978                col("ts"),
979                DataType::Timestamp(ArrowTimeUnit::Nanosecond, None),
980            )
981            .gt_eq(lit(ScalarValue::TimestampNanosecond(Some(1), None))),
982            "Filter: CAST(t.ts AS Timestamp(ns)) >= TimestampNanosecond(1, None)\n  TableScan: t",
983        );
984    }
985
986    #[test]
987    fn test_const_normalization_vs_datafusion_cast_preimage_overlap() {
988        struct Case {
989            name: &'static str,
990            fields: Vec<Field>,
991            predicate: Expr,
992            expected_greptime: &'static str,
993            expected_datafusion: &'static str,
994        }
995
996        let cases = [
997            Case {
998                name: "integer widening binary",
999                fields: vec![Field::new("v", DataType::Int16, false)],
1000                predicate: cast(col("v"), DataType::Int64).gt_eq(lit(42_i64)),
1001                expected_greptime: "Filter: t.v >= Int16(42)\n  TableScan: t",
1002                expected_datafusion: "Filter: t.v >= Int16(42)\n  TableScan: t",
1003            },
1004            Case {
1005                name: "swapped integer comparison",
1006                fields: vec![Field::new("v", DataType::Int16, false)],
1007                predicate: lit(42_i64).lt_eq(cast(col("v"), DataType::Int64)),
1008                expected_greptime: "Filter: t.v >= Int16(42)\n  TableScan: t",
1009                expected_datafusion: "Filter: t.v >= Int16(42)\n  TableScan: t",
1010            },
1011            Case {
1012                name: "try_cast integer widening binary",
1013                fields: vec![Field::new("v", DataType::Int16, false)],
1014                predicate: try_cast(col("v"), DataType::Int64).gt_eq(lit(42_i64)),
1015                expected_greptime: "Filter: t.v >= Int16(42)\n  TableScan: t",
1016                expected_datafusion: "Filter: t.v >= Int16(42)\n  TableScan: t",
1017            },
1018            Case {
1019                name: "exact in-list",
1020                fields: vec![Field::new("v", DataType::Int16, false)],
1021                predicate: cast(col("v"), DataType::Int64)
1022                    .in_list(vec![lit(1_i64), lit(2_i64)], false),
1023                expected_greptime: "Filter: t.v IN ([Int16(1), Int16(2)])\n  TableScan: t",
1024                expected_datafusion: "Filter: t.v = Int16(1) OR t.v = Int16(2)\n  TableScan: t",
1025            },
1026            Case {
1027                name: "integer between",
1028                fields: vec![Field::new("v", DataType::Int16, false)],
1029                predicate: cast(col("v"), DataType::Int64).between(lit(3_i64), lit(5_i64)),
1030                expected_greptime: "Filter: t.v BETWEEN Int16(3) AND Int16(5)\n  TableScan: t",
1031                expected_datafusion: "Filter: t.v >= Int16(3) AND t.v <= Int16(5)\n  TableScan: t",
1032            },
1033            Case {
1034                name: "not between",
1035                fields: vec![Field::new("v", DataType::Int16, false)],
1036                predicate: Expr::Between(Between {
1037                    expr: Box::new(cast(col("v"), DataType::Int64)),
1038                    negated: true,
1039                    low: Box::new(lit(3_i64)),
1040                    high: Box::new(lit(5_i64)),
1041                }),
1042                expected_greptime: "Filter: t.v NOT BETWEEN Int16(3) AND Int16(5)\n  TableScan: t",
1043                expected_datafusion: "Filter: t.v < Int16(3) OR t.v > Int16(5)\n  TableScan: t",
1044            },
1045            Case {
1046                name: "plain literal",
1047                fields: vec![Field::new("v", DataType::Int16, false)],
1048                predicate: col("v").gt_eq(lit(42_i64)),
1049                expected_greptime: "Filter: t.v >= Int16(42)\n  TableScan: t",
1050                expected_datafusion: "Filter: t.v >= Int64(42)\n  TableScan: t",
1051            },
1052            Case {
1053                name: "casted constant",
1054                fields: vec![Field::new("v", DataType::Int16, false)],
1055                predicate: col("v").gt_eq(cast(lit(42_i8), DataType::Int64)),
1056                expected_greptime: "Filter: t.v >= Int16(42)\n  TableScan: t",
1057                expected_datafusion: "Filter: t.v >= Int64(42)\n  TableScan: t",
1058            },
1059            Case {
1060                name: "timestamp downcast equality",
1061                fields: vec![Field::new(
1062                    "ts_ns",
1063                    DataType::Timestamp(ArrowTimeUnit::Nanosecond, None),
1064                    false,
1065                )],
1066                predicate: cast(
1067                    col("ts_ns"),
1068                    DataType::Timestamp(ArrowTimeUnit::Millisecond, None),
1069                )
1070                .eq(ts_ms_literal(5000)),
1071                expected_greptime: "Filter: CAST(t.ts_ns AS Timestamp(ms)) = TimestampMillisecond(5000, None)\n  TableScan: t",
1072                expected_datafusion: "Filter: t.ts_ns >= TimestampNanosecond(5000000000, None) AND t.ts_ns < TimestampNanosecond(5001000000, None)\n  TableScan: t",
1073            },
1074            Case {
1075                name: "timestamp widening exact",
1076                fields: vec![Field::new(
1077                    "ts_ms",
1078                    DataType::Timestamp(ArrowTimeUnit::Millisecond, None),
1079                    false,
1080                )],
1081                predicate: cast(
1082                    col("ts_ms"),
1083                    DataType::Timestamp(ArrowTimeUnit::Nanosecond, None),
1084                )
1085                .eq(lit(ScalarValue::TimestampNanosecond(
1086                    Some(5_000_000_000),
1087                    None,
1088                ))),
1089                expected_greptime: "Filter: CAST(t.ts_ms AS Timestamp(ns)) = TimestampNanosecond(5000000000, None)\n  TableScan: t",
1090                expected_datafusion: "Filter: t.ts_ms = TimestampMillisecond(5000, None)\n  TableScan: t",
1091            },
1092        ];
1093
1094        for case in cases {
1095            let greptime =
1096                greptime_const_normalized_filter(case.fields.clone(), case.predicate.clone());
1097            let datafusion = datafusion_simplified_filter(case.fields, case.predicate);
1098            assert_eq!(case.expected_greptime, greptime, "{} greptime", case.name);
1099            assert_eq!(
1100                case.expected_datafusion, datafusion,
1101                "{} datafusion",
1102                case.name
1103            );
1104        }
1105    }
1106
1107    fn assert_pattern_match_plan(kind: PatternMatchKind, pattern: ScalarValue, expected: &str) {
1108        let predicate = match kind {
1109            PatternMatchKind::Like => Expr::Like(Like::new(
1110                false,
1111                Box::new(cast(col("s"), DataType::LargeUtf8)),
1112                Box::new(lit(pattern)),
1113                None,
1114                false,
1115            )),
1116            PatternMatchKind::SimilarTo => Expr::SimilarTo(Like::new(
1117                false,
1118                Box::new(cast(col("s"), DataType::LargeUtf8)),
1119                Box::new(lit(pattern)),
1120                None,
1121                false,
1122            )),
1123        };
1124
1125        assert_filter_plan(
1126            vec![Field::new("s", DataType::Utf8, false)],
1127            predicate,
1128            expected,
1129        );
1130    }
1131
1132    fn assert_filter_plan(fields: Vec<Field>, predicate: Expr, expected: &str) {
1133        assert_eq!(expected, analyze_filter(fields, predicate).to_string());
1134    }
1135
1136    fn assert_timestamp_pushdown(
1137        fields: Vec<Field>,
1138        predicate: Expr,
1139        expected_analyzed: &str,
1140        expected_pushed: &str,
1141        expected_range: TimestampRange,
1142    ) {
1143        let analyzed = analyze_filter(fields, predicate);
1144        assert_eq!(expected_analyzed, analyzed.to_string());
1145
1146        let pushed = push_down_filters(analyzed);
1147        assert_eq!(expected_pushed, pushed.to_string());
1148
1149        let range =
1150            build_time_range_predicate("ts", TimeUnit::Nanosecond, &extract_scan_filters(&pushed));
1151        assert_eq!(expected_range, range);
1152    }
1153
1154    fn analyze_filter(fields: Vec<Field>, predicate: Expr) -> LogicalPlan {
1155        analyze_plan(build_filter_plan(test_schema(fields), predicate))
1156    }
1157
1158    fn greptime_const_normalized_filter(fields: Vec<Field>, predicate: Expr) -> String {
1159        analyze_filter(fields, predicate).to_string()
1160    }
1161
1162    fn datafusion_simplified_filter(fields: Vec<Field>, predicate: Expr) -> String {
1163        let plan = build_filter_plan(test_schema(fields), predicate);
1164        Optimizer::with_rules(vec![Arc::new(SimplifyExpressions::new())])
1165            .optimize(plan, &OptimizerContext::new(), |_, _| {})
1166            .unwrap()
1167            .to_string()
1168    }
1169
1170    fn analyze_plan(plan: LogicalPlan) -> LogicalPlan {
1171        ConstNormalizationRule
1172            .analyze(plan, &ConfigOptions::default())
1173            .unwrap()
1174    }
1175
1176    fn build_filter_plan(schema: Arc<DFSchema>, predicate: Expr) -> LogicalPlan {
1177        LogicalPlanBuilder::scan("t", test_source(schema), None)
1178            .unwrap()
1179            .filter(predicate)
1180            .unwrap()
1181            .build()
1182            .unwrap()
1183    }
1184
1185    fn build_scan_plan(schema: Arc<DFSchema>) -> LogicalPlan {
1186        LogicalPlanBuilder::scan("t", test_source(schema), None)
1187            .unwrap()
1188            .build()
1189            .unwrap()
1190    }
1191
1192    fn push_down_filters(plan: LogicalPlan) -> LogicalPlan {
1193        Optimizer::with_rules(vec![Arc::new(PushDownFilter::new())])
1194            .optimize(plan, &OptimizerContext::new(), |_, _| {})
1195            .unwrap()
1196    }
1197
1198    fn ts_cast_to_ms() -> Expr {
1199        cast(
1200            col("ts"),
1201            DataType::Timestamp(ArrowTimeUnit::Millisecond, None),
1202        )
1203    }
1204
1205    fn ts_ms_literal(value: i64) -> Expr {
1206        lit(ScalarValue::TimestampMillisecond(Some(value), None))
1207    }
1208
1209    fn extract_scan_filters(plan: &LogicalPlan) -> Vec<Expr> {
1210        match plan {
1211            LogicalPlan::TableScan(scan) => scan.filters.clone(),
1212            _ => plan
1213                .inputs()
1214                .into_iter()
1215                .flat_map(extract_scan_filters)
1216                .collect(),
1217        }
1218    }
1219
1220    fn test_schema(fields: Vec<Field>) -> Arc<DFSchema> {
1221        arrow_schema::Schema::new(fields).to_dfschema_ref().unwrap()
1222    }
1223
1224    fn test_source(schema: Arc<DFSchema>) -> Arc<dyn TableSource> {
1225        let table = ExactPushdownProvider {
1226            schema: Arc::new(schema.as_ref().as_arrow().clone()),
1227        };
1228        provider_as_source(Arc::new(table))
1229    }
1230
1231    #[derive(Debug)]
1232    struct ExactPushdownProvider {
1233        schema: arrow_schema::SchemaRef,
1234    }
1235
1236    #[async_trait]
1237    impl TableProvider for ExactPushdownProvider {
1238        fn as_any(&self) -> &dyn std::any::Any {
1239            self
1240        }
1241
1242        fn schema(&self) -> arrow_schema::SchemaRef {
1243            self.schema.clone()
1244        }
1245
1246        fn table_type(&self) -> TableType {
1247            TableType::Base
1248        }
1249
1250        async fn scan(
1251            &self,
1252            _state: &dyn Session,
1253            _projection: Option<&Vec<usize>>,
1254            _filters: &[Expr],
1255            _limit: Option<usize>,
1256        ) -> datafusion::error::Result<Arc<dyn ExecutionPlan>> {
1257            unreachable!("scan should not be called in const_normalization tests")
1258        }
1259
1260        fn supports_filters_pushdown(
1261            &self,
1262            filters: &[&Expr],
1263        ) -> datafusion::error::Result<Vec<TableProviderFilterPushDown>> {
1264            Ok(vec![TableProviderFilterPushDown::Exact; filters.len()])
1265        }
1266    }
1267}
query/optimizer/const_normalization.rs

query/optimizer/
const_normalization.rs
