feat: Implement EXISTS/NOT EXISTS subquery support

src/binder/expr.rs

@@ -142,6 +142,17 @@ impl<'a, T: Transaction, A: AsRef<[(&'static str, DataValue)]>> Binder<'a, '_, T
                     trim_where: *trim_where,
                 })
             }
+            Expr::Exists { subquery, negated } => {
+                let (sub_query, column) = self.bind_subquery(subquery)?;
+                let (_, sub_query) = if !self.context.is_step(&QueryBindStep::Where) {
+                    self.bind_temp_table(column, sub_query)?
+                } else {
+                    (ScalarExpression::ColumnRef(column), sub_query)
+                };
+                self.context
+                    .sub_query(SubQueryType::ExistsSubQuery(*negated, sub_query));
+                Ok(ScalarExpression::Constant(DataValue::Boolean(true)))
+            }
             Expr::Subquery(subquery) => {
                 let (sub_query, column) = self.bind_subquery(subquery)?;
                 let (expr, sub_query) = if !self.context.is_step(&QueryBindStep::Where) {

src/binder/mod.rs

@@ -85,6 +85,7 @@ pub enum QueryBindStep {
 #[derive(Debug, Clone, Hash, Eq, PartialEq)]
 pub enum SubQueryType {
     SubQuery(LogicalPlan),
+    ExistsSubQuery(bool, LogicalPlan),
     InSubQuery(bool, LogicalPlan),
 }
 

src/binder/select.rs

@@ -21,7 +21,9 @@ use super::{
 use crate::catalog::{ColumnCatalog, ColumnRef, ColumnSummary, TableName};
 use crate::errors::DatabaseError;
 use crate::execution::dql::join::joins_nullable;
+use crate::expression::agg::AggKind;
 use crate::expression::{AliasType, BinaryOperator};
+use crate::planner::operator::aggregate::AggregateOperator;
 use crate::planner::operator::function_scan::FunctionScanOperator;
 use crate::planner::operator::insert::InsertOperator;
 use crate::planner::operator::join::JoinCondition;
@@ -30,12 +32,14 @@ use crate::planner::operator::union::UnionOperator;
 use crate::planner::{Childrens, LogicalPlan, SchemaOutput};
 use crate::storage::Transaction;
 use crate::types::tuple::{Schema, SchemaRef};
+use crate::types::value::Utf8Type;
+use crate::types::LogicalType::Char;
 use crate::types::{ColumnId, LogicalType};
 use itertools::Itertools;
 use sqlparser::ast::{
-    Distinct, Expr, Ident, Join, JoinConstraint, JoinOperator, Offset, OrderByExpr, Query, Select,
-    SelectInto, SelectItem, SetExpr, SetOperator, SetQuantifier, TableAlias, TableFactor,
-    TableWithJoins,
+    CharLengthUnits, Distinct, Expr, Ident, Join, JoinConstraint, JoinOperator, Offset,
+    OrderByExpr, Query, Select, SelectInto, SelectItem, SetExpr, SetOperator, SetQuantifier,
+    TableAlias, TableFactor, TableWithJoins,
 };
 
 impl<'a: 'b, 'b, T: Transaction, A: AsRef<[(&'static str, DataValue)]>> Binder<'a, 'b, T, A> {
@@ -599,6 +603,57 @@ impl<'a: 'b, 'b, T: Transaction, A: AsRef<[(&'static str, DataValue)]>> Binder<'
 
                 let (mut plan, join_ty) = match sub_query {
                     SubQueryType::SubQuery(plan) => (plan, JoinType::Inner),
+                    SubQueryType::ExistsSubQuery(is_not, plan) => {
+                        let limit = LimitOperator::build(None, Some(1), plan);
+                        let mut agg = AggregateOperator::build(
+                            limit,
+                            vec![ScalarExpression::AggCall {
+                                distinct: false,
+                                kind: AggKind::Count,
+                                args: vec![ScalarExpression::Constant(DataValue::Utf8 {
+                                    value: "*".to_string(),
+                                    ty: Utf8Type::Fixed(1),
+                                    unit: CharLengthUnits::Characters,
+                                })],
+                                ty: LogicalType::Integer,
+                            }],
+                            vec![],
+                            false,
+                        );
+                        let filter = FilterOperator::build(
+                            ScalarExpression::Binary {
+                                op: if is_not {
+                                    BinaryOperator::NotEq
+                                } else {
+                                    BinaryOperator::Eq
+                                },
+                                left_expr: Box::new(ScalarExpression::ColumnRef(
+                                    agg.output_schema()[0].clone(),
+                                )),
+                                right_expr: Box::new(ScalarExpression::Constant(DataValue::Int32(
+                                    1,
+                                ))),
+                                evaluator: None,
+                                ty: LogicalType::Boolean,
+                            },
+                            agg,
+                            false,
+                        );
+                        let projection = ProjectOperator {
+                            exprs: vec![ScalarExpression::Constant(DataValue::Int32(1))],
+                        };
+                        let plan = LogicalPlan::new(
+                            Operator::Project(projection),
+                            Childrens::Only(filter),
+                        );
+                        children = LJoinOperator::build(
+                            children,
+                            plan,
+                            JoinCondition::None,
+                            JoinType::Cross,
+                        );
+                        continue;
+                    }
                     SubQueryType::InSubQuery(is_not, plan) => {
                         let join_ty = if is_not {
                             JoinType::LeftAnti

src/expression/mod.rs

@@ -479,8 +479,15 @@ impl ScalarExpression {
                 right_expr,
                 ..
             } => left_expr.has_count_star() || right_expr.has_count_star(),
-            ScalarExpression::AggCall { args, .. }
-            | ScalarExpression::ScalaFunction(ScalarFunction { args, .. })
+            ScalarExpression::AggCall { args, .. } => {
+                if args.len() == 1 {
+                    if let ScalarExpression::Constant(value) = &args[0] {
+                        return matches!(value.utf8(), Some("*"));
+                    }
+                }
+                args.iter().any(Self::has_count_star)
+            }
+            ScalarExpression::ScalaFunction(ScalarFunction { args, .. })
             | ScalarExpression::Coalesce { exprs: args, .. } => {
                 args.iter().any(Self::has_count_star)
             }

tests/slt/exists.slt

@@ -0,0 +1,34 @@
+statement ok
+CREATE TABLE t1 (id INT PRIMARY KEY, val INT);
+
+statement ok
+CREATE TABLE t2 (id INT PRIMARY KEY);
+
+statement ok
+INSERT INTO t1 VALUES (1, 10), (2, 20), (3, 30);
+
+statement ok
+INSERT INTO t2 VALUES (1), (3);
+
+query II
+SELECT id, val FROM t1 WHERE EXISTS (SELECT 1 FROM t2)
+----
+1 10
+2 20
+3 30
+
+query II
+SELECT id, val FROM t1 WHERE EXISTS (SELECT 1 FROM t2 WHERE 1 = 0)
+----
+
+query II
+SELECT id, val FROM t1 WHERE NOT EXISTS (SELECT 1 FROM t2)
+----
+
+
+query II
+SELECT id, val FROM t1 WHERE NOT EXISTS (SELECT 1 FROM t2 WHERE 1 = 0)
+----
+1 10
+2 20
+3 30

tests/slt/sql_2016/E061_08.slt

@@ -1,8 +1,7 @@
 # E061-08: EXISTS predicate
 
-# TODO: Support `EXISTS` on `WHERE`
+statement ok
+CREATE TABLE TABLE_E061_08_01_01 ( ID INT PRIMARY KEY, A INT );
 
-# statement ok
-# CREATE TABLE TABLE_E061_08_01_01 ( ID INT PRIMARY KEY, A INT );
-
-# SELECT A FROM TABLE_E061_08_01_01 WHERE EXISTS ( SELECT 1 )
+query I
+SELECT A FROM TABLE_E061_08_01_01 WHERE EXISTS ( SELECT 1 )

tests/slt/sql_2016/E071_06.slt

@@ -1,30 +1,31 @@
 # E071-06: Table operators in subqueries
 
-# TODO: Table operators in subqueries
+statement ok
+CREATE TABLE TABLE_E071_06_01_011 ( ID INT PRIMARY KEY, A INT );
 
-# statement ok
-# CREATE TABLE TABLE_E071_06_01_011 ( ID INT PRIMARY KEY, A INT );
+statement ok
+CREATE TABLE TABLE_E071_06_01_012 ( ID INT PRIMARY KEY, B INT );
 
-# statement ok
-# CREATE TABLE TABLE_E071_06_01_012 ( ID INT PRIMARY KEY, B FLOAT );
+query I
+SELECT A FROM TABLE_E071_06_01_011 WHERE EXISTS ( SELECT A FROM TABLE_E071_06_01_011 UNION ALL SELECT B FROM TABLE_E071_06_01_012 )
 
-# SELECT A FROM TABLE_E071_06_01_011 WHERE EXISTS ( SELECT A FROM TABLE_E071_06_01_011 UNION ALL SELECT B FROM TABLE_E071_06_01_012 )
+statement ok
+CREATE TABLE TABLE_E071_06_01_021 ( ID INT PRIMARY KEY, A INT );
 
-# statement ok
-# CREATE TABLE TABLE_E071_06_01_021 ( ID INT PRIMARY KEY, A INT );
+statement ok
+CREATE TABLE TABLE_E071_06_01_022 ( ID INT PRIMARY KEY, B INT );
 
-# statement ok
-# CREATE TABLE TABLE_E071_06_01_022 ( ID INT PRIMARY KEY, B FLOAT );
+query I
+SELECT A FROM TABLE_E071_06_01_021 WHERE EXISTS ( SELECT A FROM TABLE_E071_06_01_021 UNION DISTINCT SELECT B FROM TABLE_E071_06_01_022 )
 
-# SELECT A FROM TABLE_E071_06_01_021 WHERE EXISTS ( SELECT A FROM TABLE_E071_06_01_021 UNION DISTINCT SELECT B FROM TABLE_E071_06_01_022 )
+statement ok
+CREATE TABLE TABLE_E071_06_01_031 ( ID INT PRIMARY KEY, A INT );
 
-# statement ok
-# CREATE TABLE TABLE_E071_06_01_031 ( ID INT PRIMARY KEY, A INT );
-
-# statement ok
-# CREATE TABLE TABLE_E071_06_01_032 ( ID INT PRIMARY KEY, B FLOAT );
+statement ok
+CREATE TABLE TABLE_E071_06_01_032 ( ID INT PRIMARY KEY, B INT );
 
-# SELECT A FROM TABLE_E071_06_01_031 WHERE EXISTS ( SELECT A FROM TABLE_E071_06_01_031 UNION SELECT B FROM TABLE_E071_06_01_032 )
+query I
+SELECT A FROM TABLE_E071_06_01_031 WHERE EXISTS ( SELECT A FROM TABLE_E071_06_01_031 UNION SELECT B FROM TABLE_E071_06_01_032 )
 
 # statement ok
 # CREATE TABLE TABLE_E071_06_02_011 ( ID INT PRIMARY KEY, A INT );