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

query/optimizer/
type_conversion.rs
