Fix #1659: Support for IN clause in streaming SQL

dozer-sql/src/expression/builder.rs

@@ -0,0 +1,21 @@
+impl ExpressionBuilder {
+    pub fn parse_expr(&self, expr: &Expr, schema: &Schema) -> Result<Expression, Error> {
+        match expr {
+            Expr::InList { expr, list, negated } => {
+                let parsed_expr = self.parse_expr(expr, schema)?;
+                let parsed_list = list
+                    .iter()
+                    .map(|item| self.parse_expr(item, schema))
+                    .collect::<Result<Vec<_>, Error>>()?;
+                Ok(Expression::InList {
+                    expr: Box::new(parsed_expr),
+                    list: parsed_list,
+                    negated: *negated,
+                })
+            }
+            Expr::InSubquery { expr, subquery, negated } => {
+                Err(Error::SubqueryInWhereRequiresJoinRewrite)
+            }
+            _ => Err(Error::UnsupportedExpression(expr.to_string())),
+        }
+    }

dozer-sql/src/expression/execution.rs

@@ -0,0 +1,20 @@
+impl Expression {
+    pub fn evaluate(&self, record: &Record, schema: &Schema) -> Result<Field, Error> {
+        match self {
+            Expression::InList { expr, list, negated } => {
+                let target = expr.evaluate(record, schema)?;
+                if target.is_null() {
+                    return Ok(Field::Null);
+                }
+                let mut matched = false;
+                for item in list {
+                    let item_val = item.evaluate(record, schema)?;
+                    if target == item_val {
+                        matched = true;
+                        break;
+                    }
+                }
+                Ok(Field::Boolean(if *negated { !matched } else { matched }))
+            }
+            _ => Err(Error::UnsupportedEvaluation),
+        }

dozer-sql/src/expression/expression.rs

@@ -0,0 +1,16 @@
+#[derive(Debug, Clone, PartialEq)]
+pub enum Expression {
+    InList {
+        expr: Box<Expression>,
+        list: Vec<Expression>,
+        negated: bool,
+    },
+}
+
+impl Expression {
+    pub fn return_type(&self) -> DataType {
+        match self {
+            Expression::InList { .. } => DataType::Boolean,
+            _ => DataType::Boolean,
+        }
+    }

dozer-sql/src/expression/tests.rs

@@ -0,0 +1,29 @@
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_in_list_eval() {
+        let record = Record::new(vec![Field::Int(10)]);
+        let schema = Schema::default();
+        let expr = Expression::InList {
+            expr: Box::new(Expression::Column(0)),
+            list: vec![Expression::Literal(Field::Int(5)), Expression::Literal(Field::Int(10))],
+            negated: false,
+        };
+        let res = expr.evaluate(&record, &schema).unwrap();
+        assert_eq!(res, Field::Boolean(true));
+    }
+
+    #[test]
+    fn test_not_in_list_eval() {
+        let record = Record::new(vec![Field::Int(20)]);
+        let schema = Schema::default();
+        let expr = Expression::InList {
+            expr: Box::new(Expression::Column(0)),
+            list: vec![Expression::Literal(Field::Int(5)), Expression::Literal(Field::Int(10))],
+            negated: true,
+        };
+        let res = expr.evaluate(&record, &schema).unwrap();
+        assert_eq!(res, Field::Boolean(true));
+    }

dozer-sql/src/pipeline/builder.rs

@@ -0,0 +1,20 @@
+impl PipelineBuilder {
+    pub fn build_in_subquery_join(
+        &mut self,
+        left_expr: &Expr,
+        subquery: &TableSubquery,
+        negated: bool,
+    ) -> Result<ProcessorNode, PipelineError> {
+        let right_stream = self.build_subquery_pipeline(subquery)?;
+        let join_type = if negated {
+            JoinType::LeftAnti
+        } else {
+            JoinType::Inner
+        };
+        let join_processor = JoinProcessor::new(
+            join_type,
+            vec![(left_expr.clone(), Expr::Identifier(Ident::new("subquery_col")))],
+        );
+        self.add_processor_node(join_processor, vec![self.current_stream(), right_stream])
+    }
+}