query/optimizer/
type_conversion.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 common_time::Timezone;
18use common_time::timestamp::{TimeUnit, Timestamp};
19use datafusion::config::ConfigOptions;
20use datafusion_common::tree_node::{Transformed, TreeNode, TreeNodeRewriter};
21use datafusion_common::{DFSchemaRef, DataFusionError, Result, ScalarValue};
22use datafusion_expr::expr::InList;
23use datafusion_expr::{
24    Between, BinaryExpr, Expr, ExprSchemable, Filter, LogicalPlan, Operator, TableScan, WriteOp,
25};
26use datatypes::arrow::compute;
27use datatypes::arrow::datatypes::DataType;
28use session::context::QueryContextRef;
29
30use crate::QueryEngineContext;
31use crate::optimizer::ExtensionAnalyzerRule;
32use crate::optimizer::insert_assignment::rewrite_insert_assignments;
33use crate::plan::ExtractExpr;
34
35/// TypeConversionRule converts some literal values in logical plan to other types according
36/// to data type of corresponding columns.
37/// Specifically:
38/// - string literal of timestamp is converted to the target timestamp type
39/// - string literal of boolean is converted to `Expr::Literal(ScalarValue::Boolean)`
40pub struct TypeConversionRule;
41
42impl ExtensionAnalyzerRule for TypeConversionRule {
43    fn analyze(
44        &self,
45        plan: LogicalPlan,
46        ctx: &QueryEngineContext,
47        config: &ConfigOptions,
48    ) -> Result<LogicalPlan> {
49        plan.transform_up_with_subqueries(|plan| match plan {
50            LogicalPlan::Filter(filter) => {
51                let mut converter =
52                    TypeConverter::new(filter.input.schema().clone(), ctx.query_ctx());
53                let rewritten = filter.predicate.clone().rewrite(&mut converter)?.data;
54                Ok(Transformed::yes(LogicalPlan::Filter(Filter::try_new(
55                    rewritten,
56                    filter.input,
57                )?)))
58            }
59            LogicalPlan::TableScan(TableScan {
60                table_name,
61                source,
62                projection,
63                projected_schema,
64                filters,
65                fetch,
66                statistics_requests,
67            }) => {
68                let mut converter = TypeConverter::new(projected_schema.clone(), ctx.query_ctx());
69                let rewrite_filters = filters
70                    .into_iter()
71                    .map(|e| e.rewrite(&mut converter).map(|x| x.data))
72                    .collect::<Result<Vec<_>>>()?;
73                Ok(Transformed::yes(LogicalPlan::TableScan(TableScan {
74                    table_name: table_name.clone(),
75                    source: source.clone(),
76                    projection,
77                    projected_schema,
78                    filters: rewrite_filters,
79                    fetch,
80                    statistics_requests,
81                })))
82            }
83            LogicalPlan::Projection { .. } => {
84                let mut converter = TypeConverter::new(plan.schema().clone(), ctx.query_ctx());
85                let inputs = plan.inputs().into_iter().cloned().collect::<Vec<_>>();
86                let expr = plan
87                    .expressions_consider_join()
88                    .into_iter()
89                    .map(|e| e.rewrite(&mut converter).map(|x| x.data))
90                    .collect::<Result<Vec<_>>>()?;
91
92                plan.with_new_exprs(expr, inputs).map(Transformed::yes)
93            }
94            LogicalPlan::Window { .. }
95            | LogicalPlan::Aggregate { .. }
96            | LogicalPlan::Repartition { .. }
97            | LogicalPlan::Extension { .. }
98            | LogicalPlan::Sort { .. }
99            | LogicalPlan::Union { .. }
100            | LogicalPlan::Values { .. }
101            | LogicalPlan::Analyze { .. } => {
102                let mut converter = TypeConverter::new(plan.schema().clone(), ctx.query_ctx());
103                let inputs = plan.inputs().into_iter().cloned().collect::<Vec<_>>();
104                let expr = plan
105                    .expressions_consider_join()
106                    .into_iter()
107                    .map(|e| e.rewrite(&mut converter).map(|x| x.data))
108                    .collect::<Result<Vec<_>>>()?;
109
110                plan.with_new_exprs(expr, inputs).map(Transformed::yes)
111            }
112
113            LogicalPlan::Join(join) => {
114                let Ok(schema) = join.left.schema().join(join.right.schema()) else {
115                    return Ok(Transformed::no(LogicalPlan::Join(join)));
116                };
117                let mut converter = TypeConverter::new(Arc::new(schema), ctx.query_ctx());
118                let plan = LogicalPlan::Join(join);
119                let inputs = plan.inputs().into_iter().cloned().collect::<Vec<_>>();
120                let expr = plan
121                    .expressions_consider_join()
122                    .into_iter()
123                    .map(|e| e.rewrite(&mut converter).map(|x| x.data))
124                    .collect::<Result<Vec<_>>>()?;
125
126                plan.with_new_exprs(expr, inputs).map(Transformed::yes)
127            }
128
129            LogicalPlan::Dml(mut dml) if matches!(dml.op, WriteOp::Insert(_)) => {
130                dml.input = Arc::new(rewrite_insert_assignments(
131                    dml.input.as_ref().clone(),
132                    &ctx.query_ctx(),
133                    config,
134                )?);
135                Ok(Transformed::yes(LogicalPlan::Dml(dml)))
136            }
137
138            LogicalPlan::Distinct { .. }
139            | LogicalPlan::Limit { .. }
140            | LogicalPlan::Subquery { .. }
141            | LogicalPlan::Explain { .. }
142            | LogicalPlan::SubqueryAlias { .. }
143            | LogicalPlan::EmptyRelation(_)
144            | LogicalPlan::Dml(_)
145            | LogicalPlan::DescribeTable(_)
146            | LogicalPlan::Unnest(_)
147            | LogicalPlan::Statement(_)
148            | LogicalPlan::Ddl(_)
149            | LogicalPlan::Copy(_)
150            | LogicalPlan::RecursiveQuery(_) => Ok(Transformed::no(plan)),
151        })
152        .map(|x| x.data)
153    }
154}
155
156struct TypeConverter {
157    query_ctx: QueryContextRef,
158    schema: DFSchemaRef,
159}
160
161impl TypeConverter {
162    fn new(schema: DFSchemaRef, query_ctx: QueryContextRef) -> Self {
163        Self { query_ctx, schema }
164    }
165
166    fn column_type(&self, expr: &Expr) -> Option<DataType> {
167        if let Expr::Column(_) = expr
168            && let Ok(v) = expr.get_type(&self.schema)
169        {
170            return Some(v);
171        }
172        None
173    }
174
175    fn cast_scalar_value(
176        &self,
177        value: &ScalarValue,
178        target_type: &DataType,
179    ) -> Result<ScalarValue> {
180        match (target_type, value) {
181            (DataType::Timestamp(_, _), ScalarValue::Utf8(Some(v))) => {
182                parse_string_to_timestamp(v, Some(&self.query_ctx.timezone()))
183            }
184            (DataType::Boolean, ScalarValue::Utf8(Some(v))) => match v.to_lowercase().as_str() {
185                "true" => Ok(ScalarValue::Boolean(Some(true))),
186                "false" => Ok(ScalarValue::Boolean(Some(false))),
187                _ => Ok(ScalarValue::Boolean(None)),
188            },
189            (target_type, value) => {
190                let value_arr = value.to_array()?;
191                let arr = compute::cast(&value_arr, target_type)
192                    .map_err(|e| DataFusionError::ArrowError(Box::new(e), None))?;
193
194                ScalarValue::try_from_array(
195                    &arr,
196                    0, // index: Converts a value in `array` at `index` into a ScalarValue
197                )
198            }
199        }
200    }
201
202    fn convert_type<'b>(&self, left: &'b Expr, right: &'b Expr) -> Result<(Expr, Expr)> {
203        let left_type = self.column_type(left);
204        let right_type = self.column_type(right);
205
206        let target_type = match (&left_type, &right_type) {
207            (Some(v), None) => v,
208            (None, Some(v)) => v,
209            _ => return Ok((left.clone(), right.clone())),
210        };
211
212        // only try to convert timestamp or boolean types
213        if !matches!(target_type, DataType::Timestamp(_, _) | DataType::Boolean) {
214            return Ok((left.clone(), right.clone()));
215        }
216
217        match (left, right) {
218            (Expr::Column(col), Expr::Literal(value, _)) => {
219                let casted_right = self.cast_scalar_value(value, target_type)?;
220                if casted_right.is_null() {
221                    return Err(DataFusionError::Plan(format!(
222                        "column:{col:?}. Casting value:{value:?} to {target_type:?} is invalid",
223                    )));
224                }
225                Ok((left.clone(), Expr::Literal(casted_right, None)))
226            }
227            (Expr::Literal(value, _), Expr::Column(col)) => {
228                let casted_left = self.cast_scalar_value(value, target_type)?;
229                if casted_left.is_null() {
230                    return Err(DataFusionError::Plan(format!(
231                        "column:{col:?}. Casting value:{value:?} to {target_type:?} is invalid",
232                    )));
233                }
234                Ok((Expr::Literal(casted_left, None), right.clone()))
235            }
236            _ => Ok((left.clone(), right.clone())),
237        }
238    }
239}
240
241impl TreeNodeRewriter for TypeConverter {
242    type Node = Expr;
243
244    fn f_up(&mut self, expr: Expr) -> Result<Transformed<Expr>> {
245        let new_expr = match expr {
246            Expr::BinaryExpr(BinaryExpr { left, op, right }) => match op {
247                Operator::Eq
248                | Operator::NotEq
249                | Operator::Lt
250                | Operator::LtEq
251                | Operator::Gt
252                | Operator::GtEq => {
253                    let (left, right) = self.convert_type(&left, &right)?;
254                    Expr::BinaryExpr(BinaryExpr {
255                        left: Box::new(left),
256                        op,
257                        right: Box::new(right),
258                    })
259                }
260                _ => Expr::BinaryExpr(BinaryExpr { left, op, right }),
261            },
262            Expr::Between(Between {
263                expr,
264                negated,
265                low,
266                high,
267            }) => {
268                let (expr, low) = self.convert_type(&expr, &low)?;
269                let (expr, high) = self.convert_type(&expr, &high)?;
270                Expr::Between(Between {
271                    expr: Box::new(expr),
272                    negated,
273                    low: Box::new(low),
274                    high: Box::new(high),
275                })
276            }
277            Expr::InList(InList {
278                expr,
279                list,
280                negated,
281            }) => {
282                let mut list_expr = Vec::with_capacity(list.len());
283                for e in list {
284                    let (_, expr_conversion) = self.convert_type(&expr, &e)?;
285                    list_expr.push(expr_conversion);
286                }
287                Expr::InList(InList {
288                    expr,
289                    list: list_expr,
290                    negated,
291                })
292            }
293            Expr::Literal(value, _) => match value {
294                ScalarValue::TimestampSecond(Some(i), _) => {
295                    timestamp_to_timestamp_ms_expr(i, TimeUnit::Second)
296                }
297                ScalarValue::TimestampMillisecond(Some(i), _) => {
298                    timestamp_to_timestamp_ms_expr(i, TimeUnit::Millisecond)
299                }
300                ScalarValue::TimestampMicrosecond(Some(i), _) => {
301                    timestamp_to_timestamp_ms_expr(i, TimeUnit::Microsecond)
302                }
303                ScalarValue::TimestampNanosecond(Some(i), _) => {
304                    timestamp_to_timestamp_ms_expr(i, TimeUnit::Nanosecond)
305                }
306                _ => Expr::Literal(value, None),
307            },
308            expr => expr,
309        };
310        Ok(Transformed::yes(new_expr))
311    }
312}
313
314fn timestamp_to_timestamp_ms_expr(val: i64, unit: TimeUnit) -> Expr {
315    let timestamp = match unit {
316        TimeUnit::Second => val * 1_000,
317        TimeUnit::Millisecond => val,
318        TimeUnit::Microsecond => val / 1_000,
319        TimeUnit::Nanosecond => val / 1_000 / 1_000,
320    };
321
322    Expr::Literal(
323        ScalarValue::TimestampMillisecond(Some(timestamp), None),
324        None,
325    )
326}
327
328pub(crate) fn cast_string_to_timestamp(
329    string: &str,
330    target_type: &DataType,
331    timezone: Option<&Timezone>,
332) -> Result<ScalarValue> {
333    let parsed = parse_string_to_timestamp(string, timezone)?;
334    cast_timestamp(parsed, target_type)
335}
336
337fn parse_string_to_timestamp(string: &str, timezone: Option<&Timezone>) -> Result<ScalarValue> {
338    let ts = Timestamp::from_str(string, timezone)
339        .map_err(|e| DataFusionError::External(Box::new(e)))?;
340
341    let value = Some(ts.value());
342    Ok(match ts.unit() {
343        TimeUnit::Second => ScalarValue::TimestampSecond(value, None),
344        TimeUnit::Millisecond => ScalarValue::TimestampMillisecond(value, None),
345        TimeUnit::Microsecond => ScalarValue::TimestampMicrosecond(value, None),
346        TimeUnit::Nanosecond => ScalarValue::TimestampNanosecond(value, None),
347    })
348}
349
350fn cast_timestamp(parsed: ScalarValue, target_type: &DataType) -> Result<ScalarValue> {
351    let parsed = parsed.to_array()?;
352    let casted = compute::cast(&parsed, target_type)
353        .map_err(|e| DataFusionError::ArrowError(Box::new(e), None))?;
354
355    ScalarValue::try_from_array(&casted, 0)
356}
357
358#[cfg(test)]
359mod tests {
360    use std::collections::HashMap;
361    use std::sync::Arc;
362
363    use datafusion_common::arrow::datatypes::Field;
364    use datafusion_common::{Column, DFSchema, NullEquality, TableReference};
365    use datafusion_expr::expr::{Cast, Exists};
366    use datafusion_expr::{Join, JoinConstraint, JoinType, Literal, LogicalPlanBuilder, Subquery};
367    use session::context::QueryContext;
368
369    use super::*;
370
371    #[test]
372    fn test_cast_string_to_timestamp() {
373        use datafusion_common::arrow::datatypes::TimeUnit as ArrowTimeUnit;
374
375        let target_type = DataType::Timestamp(ArrowTimeUnit::Second, None);
376        assert_eq!(
377            cast_string_to_timestamp("2022-02-02 19:00:00+08:00", &target_type, None).unwrap(),
378            ScalarValue::TimestampSecond(Some(1643799600), None)
379        );
380        assert_eq!(
381            cast_string_to_timestamp("2009-02-13 23:31:30Z", &target_type, None).unwrap(),
382            ScalarValue::TimestampSecond(Some(1234567890), None)
383        );
384
385        assert_eq!(
386            cast_string_to_timestamp(
387                "2009-02-13 23:31:30",
388                &target_type,
389                Some(&Timezone::from_tz_string("Asia/Shanghai").unwrap())
390            )
391            .unwrap(),
392            ScalarValue::TimestampSecond(Some(1234567890 - 8 * 3600), None)
393        );
394
395        assert_eq!(
396            cast_string_to_timestamp(
397                "2009-02-13 23:31:30",
398                &target_type,
399                Some(&Timezone::from_tz_string("-8:00").unwrap())
400            )
401            .unwrap(),
402            ScalarValue::TimestampSecond(Some(1234567890 + 8 * 3600), None)
403        );
404
405        assert_eq!(
406            cast_string_to_timestamp(
407                "2009-02-13 23:31:30.123456789Z",
408                &DataType::Timestamp(ArrowTimeUnit::Nanosecond, None),
409                None,
410            )
411            .unwrap(),
412            ScalarValue::TimestampNanosecond(Some(1_234_567_890_123_456_789), None)
413        );
414    }
415
416    #[test]
417    fn test_timestamp_to_timestamp_ms_expr() {
418        assert_eq!(
419            timestamp_to_timestamp_ms_expr(123, TimeUnit::Second),
420            ScalarValue::TimestampMillisecond(Some(123000), None).lit()
421        );
422
423        assert_eq!(
424            timestamp_to_timestamp_ms_expr(123, TimeUnit::Millisecond),
425            ScalarValue::TimestampMillisecond(Some(123), None).lit()
426        );
427
428        assert_eq!(
429            timestamp_to_timestamp_ms_expr(123, TimeUnit::Microsecond),
430            ScalarValue::TimestampMillisecond(Some(0), None).lit()
431        );
432
433        assert_eq!(
434            timestamp_to_timestamp_ms_expr(1230, TimeUnit::Microsecond),
435            ScalarValue::TimestampMillisecond(Some(1), None).lit()
436        );
437
438        assert_eq!(
439            timestamp_to_timestamp_ms_expr(123000, TimeUnit::Microsecond),
440            ScalarValue::TimestampMillisecond(Some(123), None).lit()
441        );
442
443        assert_eq!(
444            timestamp_to_timestamp_ms_expr(1230, TimeUnit::Nanosecond),
445            ScalarValue::TimestampMillisecond(Some(0), None).lit()
446        );
447        assert_eq!(
448            timestamp_to_timestamp_ms_expr(123_000_000, TimeUnit::Nanosecond),
449            ScalarValue::TimestampMillisecond(Some(123), None).lit()
450        );
451    }
452
453    #[test]
454    fn test_convert_timestamp_str() {
455        use datatypes::arrow::datatypes::TimeUnit as ArrowTimeUnit;
456
457        let schema = Arc::new(
458            DFSchema::new_with_metadata(
459                vec![(
460                    None::<TableReference>,
461                    Arc::new(Field::new(
462                        "ts",
463                        DataType::Timestamp(ArrowTimeUnit::Millisecond, None),
464                        true,
465                    )),
466                )],
467                HashMap::new(),
468            )
469            .unwrap(),
470        );
471        let mut converter = TypeConverter::new(schema, QueryContext::arc());
472
473        assert_eq!(
474            Expr::Column(Column::from_name("ts")).gt(ScalarValue::TimestampSecond(
475                Some(1_599_514_949),
476                None
477            )
478            .lit()),
479            converter
480                .f_up(Expr::Column(Column::from_name("ts")).gt("2020-09-08T05:42:29+08:00".lit()))
481                .unwrap()
482                .data
483        );
484    }
485
486    #[test]
487    fn test_type_converter_leaves_explicit_timestamp_cast_unchanged() {
488        use datafusion_common::arrow::datatypes::TimeUnit as ArrowTimeUnit;
489
490        let mut converter = TypeConverter::new(Arc::new(DFSchema::empty()), QueryContext::arc());
491        let expr = Expr::Cast(Cast::new(
492            Box::new("2009-02-13 23:31:30".lit()),
493            DataType::Timestamp(ArrowTimeUnit::Millisecond, None),
494        ));
495
496        assert_eq!(converter.f_up(expr.clone()).unwrap().data, expr);
497    }
498
499    #[test]
500    fn test_convert_bool() {
501        let col_name = "is_valid";
502        let schema = Arc::new(
503            DFSchema::new_with_metadata(
504                vec![(
505                    None::<TableReference>,
506                    Arc::new(Field::new(col_name, DataType::Boolean, false)),
507                )],
508                HashMap::new(),
509            )
510            .unwrap(),
511        );
512        let mut converter = TypeConverter::new(schema, QueryContext::arc());
513
514        assert_eq!(
515            Expr::Column(Column::from_name(col_name)).eq(true.lit()),
516            converter
517                .f_up(Expr::Column(Column::from_name(col_name)).eq("true".lit()))
518                .unwrap()
519                .data
520        );
521    }
522
523    #[test]
524    fn test_retrieve_type_from_aggr_plan() {
525        let plan = LogicalPlanBuilder::values(vec![vec![
526            ScalarValue::Int64(Some(1)).lit(),
527            ScalarValue::Float64(Some(1.0)).lit(),
528            ScalarValue::TimestampMillisecond(Some(1), None).lit(),
529        ]])
530        .unwrap()
531        .filter(Expr::Column(Column::from_name("column3")).gt("1970-01-01 00:00:00+08:00".lit()))
532        .unwrap()
533        .filter(
534            "1970-01-01 00:00:00+08:00"
535                .lit()
536                .lt_eq(Expr::Column(Column::from_name("column3"))),
537        )
538        .unwrap()
539        .aggregate(
540            Vec::<Expr>::new(),
541            vec![Expr::AggregateFunction(
542                datafusion_expr::expr::AggregateFunction::new_udf(
543                    datafusion::functions_aggregate::count::count_udaf(),
544                    vec![Expr::Column(Column::from_name("column1"))],
545                    false,
546                    None,
547                    vec![],
548                    None,
549                ),
550            )],
551        )
552        .unwrap()
553        .build()
554        .unwrap();
555        let context = QueryEngineContext::mock();
556
557        let transformed_plan = TypeConversionRule
558            .analyze(plan, &context, &ConfigOptions::default())
559            .unwrap();
560        let expected = String::from(
561            "Aggregate: groupBy=[[]], aggr=[[count(column1)]]\
562            \n  Filter: TimestampSecond(-28800, None) <= column3\
563            \n    Filter: column3 > TimestampSecond(-28800, None)\
564            \n      Values: (Int64(1), Float64(1), TimestampMillisecond(1, None))",
565        );
566        assert_eq!(format!("{}", transformed_plan.display_indent()), expected);
567    }
568
569    #[test]
570    fn test_reverse_non_ts_type() {
571        let context = QueryEngineContext::mock();
572
573        let plan = LogicalPlanBuilder::values(vec![vec![1.0f64.lit()]])
574            .unwrap()
575            .filter(Expr::Column(Column::from_name("column1")).gt_eq("1.2345".lit()))
576            .unwrap()
577            .filter(
578                "1.2345"
579                    .lit()
580                    .lt(Expr::Column(Column::from_name("column1"))),
581            )
582            .unwrap()
583            .build()
584            .unwrap();
585        let transformed_plan = TypeConversionRule
586            .analyze(plan, &context, &ConfigOptions::default())
587            .unwrap();
588        let expected = String::from(
589            "Filter: Utf8(\"1.2345\") < column1\
590            \n  Filter: column1 >= Utf8(\"1.2345\")\
591            \n    Values: (Float64(1))",
592        );
593        assert_eq!(format!("{}", transformed_plan.display_indent()), expected);
594    }
595
596    #[test]
597    fn test_convert_join_filter_uses_input_schemas() {
598        let left = LogicalPlanBuilder::values(vec![vec![
599            ScalarValue::Int64(Some(1)).lit(),
600            ScalarValue::TimestampMillisecond(Some(1), None).lit(),
601        ]])
602        .unwrap()
603        .alias("left")
604        .unwrap()
605        .build()
606        .unwrap();
607        let right = LogicalPlanBuilder::values(vec![vec![ScalarValue::Int64(Some(1)).lit()]])
608            .unwrap()
609            .alias("right")
610            .unwrap()
611            .build()
612            .unwrap();
613        let left_key = Column::new(Some("left"), "column1");
614        let right_key = Column::new(Some("right"), "column1");
615        let timestamp_column = Column::new(Some("left"), "column2");
616        let plan = LogicalPlanBuilder::from(left)
617            .join(
618                right,
619                JoinType::RightSemi,
620                (vec![left_key.clone()], vec![right_key.clone()]),
621                Some(Expr::Column(timestamp_column.clone()).gt("2009-02-13 23:31:30".lit())),
622            )
623            .unwrap()
624            .build()
625            .unwrap();
626        let context = QueryEngineContext::mock();
627        context
628            .query_ctx()
629            .set_timezone(Timezone::from_tz_string("Asia/Shanghai").unwrap());
630
631        let transformed = TypeConversionRule
632            .analyze(plan, &context, &ConfigOptions::default())
633            .unwrap();
634        let LogicalPlan::Join(join) = transformed else {
635            panic!("expected join plan");
636        };
637
638        assert_eq!(
639            join.on,
640            vec![(Expr::Column(left_key), Expr::Column(right_key))]
641        );
642        assert_eq!(
643            join.filter,
644            Some(
645                Expr::Column(timestamp_column).gt(ScalarValue::TimestampSecond(
646                    Some(1_234_539_090),
647                    None
648                )
649                .lit())
650            )
651        );
652    }
653
654    #[test]
655    fn test_semi_anti_join_with_duplicate_unqualified_fields() {
656        let left = Arc::new(
657            LogicalPlanBuilder::values(vec![vec![1_i64.lit()]])
658                .unwrap()
659                .build()
660                .unwrap(),
661        );
662        let right = Arc::new(
663            LogicalPlanBuilder::values(vec![vec![2_i64.lit()]])
664                .unwrap()
665                .build()
666                .unwrap(),
667        );
668        let context = QueryEngineContext::mock();
669
670        for join_type in [JoinType::LeftSemi, JoinType::LeftAnti] {
671            let join = Join::try_new(
672                Arc::clone(&left),
673                Arc::clone(&right),
674                vec![(
675                    Expr::Column(Column::from_name("column1")),
676                    Expr::Column(Column::from_name("column1")),
677                )],
678                None,
679                join_type,
680                JoinConstraint::On,
681                NullEquality::NullEqualsNothing,
682                false,
683            )
684            .unwrap();
685
686            let result = TypeConversionRule.analyze(
687                LogicalPlan::Join(join),
688                &context,
689                &ConfigOptions::default(),
690            );
691            assert!(result.is_ok(), "{join_type:?}: {result:?}");
692        }
693    }
694
695    #[test]
696    fn test_convert_exists_subquery_filter() {
697        let inner = LogicalPlanBuilder::values(vec![vec![
698            ScalarValue::TimestampMillisecond(Some(1), None).lit(),
699            false.lit(),
700        ]])
701        .unwrap()
702        .filter(
703            Expr::Column(Column::from_name("column1"))
704                .gt("2009-02-13 23:31:30+08:00".lit())
705                .and(Expr::Column(Column::from_name("column2")).eq("true".lit())),
706        )
707        .unwrap()
708        .build()
709        .unwrap();
710        let outer = LogicalPlanBuilder::values(vec![vec![1_i64.lit()]])
711            .unwrap()
712            .filter(Expr::Exists(Exists {
713                subquery: Subquery {
714                    subquery: Arc::new(inner),
715                    outer_ref_columns: Default::default(),
716                    spans: Default::default(),
717                },
718                negated: false,
719            }))
720            .unwrap()
721            .build()
722            .unwrap();
723        let context = QueryEngineContext::mock();
724
725        let transformed = TypeConversionRule
726            .analyze(outer, &context, &ConfigOptions::default())
727            .unwrap();
728        let LogicalPlan::Filter(outer_filter) = transformed else {
729            panic!("expected outer filter");
730        };
731        let Expr::Exists(exists) = outer_filter.predicate else {
732            panic!("expected exists predicate");
733        };
734        let LogicalPlan::Filter(inner_filter) = exists.subquery.subquery.as_ref() else {
735            panic!("expected inner filter");
736        };
737
738        assert_eq!(
739            inner_filter.predicate,
740            Expr::Column(Column::from_name("column1"))
741                .gt(ScalarValue::TimestampSecond(Some(1_234_539_090), None).lit())
742                .and(Expr::Column(Column::from_name("column2")).eq(true.lit()))
743        );
744    }
745}
query/optimizer/type_conversion.rs

query/optimizer/
type_conversion.rs
