Support for IN clause in streaming SQL

dozer-sql/expression/src/execution.rs

@@ -436,7 +436,7 @@ impl Expression {
                 negated: _,
             } => Ok(ExpressionType::new(
                 FieldType::Boolean,
-                false,
+                true,
                 SourceDefinition::Dynamic,
                 false,
             )),

dozer-sql/expression/src/in_list.rs

@@ -12,17 +12,25 @@ pub(crate) fn evaluate_in_list(
     record: &Record,
 ) -> Result<Field, Error> {
     let field = expr.evaluate(record, schema)?;
-    let mut result = false;
+    if field == Field::Null {
+        return Ok(Field::Null);
+    }
+
+    let mut contains_null = false;
     for item in list {
         let item = item.evaluate(record, schema)?;
+        if item == Field::Null {
+            contains_null = true;
+            continue;
+        }
         if field == item {
-            result = true;
-            break;
+            return Ok(Field::Boolean(!negated));
         }
     }
-    // Negate the result if the IN list was negated.
-    if negated {
-        result = !result;
+
+    if contains_null {
+        Ok(Field::Null)
+    } else {
+        Ok(Field::Boolean(negated))
     }
-    Ok(Field::Boolean(result))
 }

dozer-sql/src/builder/mod.rs

@@ -2,11 +2,14 @@ use crate::aggregation::factory::AggregationProcessorFactory;
 use crate::builder::PipelineError::InvalidQuery;
 use crate::errors::PipelineError;
 use crate::selection::factory::SelectionProcessorFactory;
+use crate::selection::in_subquery::{
+    InSubqueryProcessorFactory, IN_SUBQUERY_LEFT_PORT, IN_SUBQUERY_RIGHT_PORT,
+};
 use dozer_core::app::AppPipeline;
 use dozer_core::node::PortHandle;
 use dozer_core::DEFAULT_PORT_HANDLE;
 use dozer_sql_expression::builder::{ExpressionBuilder, NameOrAlias};
-use dozer_sql_expression::sqlparser::ast::{SetOperator, SetQuantifier, TableFactor};
+use dozer_sql_expression::sqlparser::ast::{Expr, SetOperator, SetQuantifier, TableFactor};
 use dozer_types::models::udf_config::UdfConfig;
 
 use dozer_sql_expression::sqlparser::{
@@ -294,21 +297,64 @@ fn select_to_pipeline(
 
     // Where clause
     if let Some(selection) = select.selection {
-        let selection = SelectionProcessorFactory::new(
-            gen_selection_name.clone(),
-            selection,
-            query_ctx.udfs.clone(),
-            query_ctx.runtime.clone(),
-        );
-
-        pipeline.add_processor(Box::new(selection), gen_selection_name.clone());
-
-        pipeline.connect_nodes(
-            gen_product_name,
-            product_output_port,
-            gen_selection_name.clone(),
-            DEFAULT_PORT_HANDLE,
-        );
+        if let Expr::InSubquery {
+            expr,
+            subquery,
+            negated,
+        } = selection
+        {
+            let subquery_name = format!("in_subquery--{}", query_ctx.get_next_processor_id());
+            query_to_pipeline(
+                TableInfo {
+                    name: NameOrAlias(subquery_name.clone(), None),
+                    override_name: None,
+                },
+                *subquery,
+                pipeline,
+                query_ctx,
+                pipeline_idx,
+                false,
+            )?;
+            let subquery_output = query_ctx
+                .pipeline_map
+                .get(&(pipeline_idx, subquery_name))
+                .cloned()
+                .ok_or_else(|| InvalidQuery("IN subquery did not produce output".to_string()))?;
+            let selection = InSubqueryProcessorFactory::new(
+                gen_selection_name.clone(),
+                *expr,
+                negated,
+                query_ctx.udfs.clone(),
+                query_ctx.runtime.clone(),
+            );
+            pipeline.add_processor(Box::new(selection), gen_selection_name.clone());
+            pipeline.connect_nodes(
+                gen_product_name,
+                product_output_port,
+                gen_selection_name.clone(),
+                IN_SUBQUERY_LEFT_PORT,
+            );
+            pipeline.connect_nodes(
+                subquery_output.node,
+                subquery_output.port,
+                gen_selection_name.clone(),
+                IN_SUBQUERY_RIGHT_PORT,
+            );
+        } else {
+            let selection = SelectionProcessorFactory::new(
+                gen_selection_name.clone(),
+                selection,
+                query_ctx.udfs.clone(),
+                query_ctx.runtime.clone(),
+            );
+            pipeline.add_processor(Box::new(selection), gen_selection_name.clone());
+            pipeline.connect_nodes(
+                gen_product_name,
+                product_output_port,
+                gen_selection_name.clone(),
+                DEFAULT_PORT_HANDLE,
+            );
+        }
 
         pipeline.connect_nodes(
             gen_selection_name,

dozer-sql/src/expression/tests/in_list.rs

@@ -84,3 +84,18 @@ fn test_not_in_list() {
     );
     assert_eq!(f, Field::Boolean(false));
 }
+
+#[test]
+fn test_in_list_null_semantics() {
+    let f = run_fct("SELECT NULL IN (1, 2)", Schema::default(), vec![]);
+    assert_eq!(f, Field::Null);
+
+    let f = run_fct("SELECT 3 IN (1, NULL)", Schema::default(), vec![]);
+    assert_eq!(f, Field::Null);
+
+    let f = run_fct("SELECT 3 NOT IN (1, NULL)", Schema::default(), vec![]);
+    assert_eq!(f, Field::Null);
+
+    let f = run_fct("SELECT 1 NOT IN (1, NULL)", Schema::default(), vec![]);
+    assert_eq!(f, Field::Boolean(false));
+}