From 69bb78bcde0877590d127c664b8020876c882941 Mon Sep 17 00:00:00 2001 From: Sushant kumar Date: Wed, 9 Sep 2026 13:28:38 +0530 Subject: [PATCH] feat(sql): support streaming IN subqueries with mark joins --- Cargo.lock | 13 +- dozer-sql/expression/Cargo.toml | 2 +- dozer-sql/expression/src/builder.rs | 4 +- dozer-sql/expression/src/execution.rs | 6 +- dozer-sql/expression/src/logical.rs | 181 ++--- dozer-sql/src/builder/in_subquery.rs | 117 ++++ dozer-sql/src/builder/mod.rs | 97 ++- dozer-sql/src/product/join/mark.rs | 448 +++++++++++++ dozer-sql/src/product/join/mod.rs | 1 + dozer-sql/src/selection/factory.rs | 26 +- dozer-sql/src/selection/processor.rs | 62 +- dozer-sql/src/tests/in_subquery.rs | 930 ++++++++++++++++++++++++++ dozer-sql/src/tests/mod.rs | 1 + dozer-tests/in_subquery/README.md | 24 + 14 files changed, 1733 insertions(+), 179 deletions(-) create mode 100644 dozer-sql/src/builder/in_subquery.rs create mode 100644 dozer-sql/src/product/join/mark.rs create mode 100644 dozer-sql/src/tests/in_subquery.rs create mode 100644 dozer-tests/in_subquery/README.md diff --git a/Cargo.lock b/Cargo.lock index 55a96a2c33..478b362183 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -8177,6 +8177,7 @@ source = "git+https://github.com/getdozer/sqlparser-rs.git#3dd4e9f14a9631c9707c4 dependencies = [ "bigdecimal 0.3.1", "log", + "sqlparser_derive 0.1.1", ] [[package]] @@ -8186,7 +8187,17 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5cc2c25a6c66789625ef164b4c7d2e548d627902280c13710d33da8222169964" dependencies = [ "log", - "sqlparser_derive", + "sqlparser_derive 0.2.2", +] + +[[package]] +name = "sqlparser_derive" +version = "0.1.1" +source = "git+https://github.com/getdozer/sqlparser-rs.git#3dd4e9f14a9631c9707c40d7e497ffe0558a88cd" +dependencies = [ + "proc-macro2", + "quote", + "syn 1.0.109", ] [[package]] diff --git a/dozer-sql/expression/Cargo.toml b/dozer-sql/expression/Cargo.toml index 3423d7904a..04219f699a 100644 --- a/dozer-sql/expression/Cargo.toml +++ b/dozer-sql/expression/Cargo.toml @@ -9,7 +9,7 @@ license = "AGPL-3.0-or-later" dozer-types = { path = "../../dozer-types" } dozer-core = { path = "../../dozer-core" } num-traits = "0.2.16" -sqlparser = { git = "https://github.com/getdozer/sqlparser-rs.git" } +sqlparser = { git = "https://github.com/getdozer/sqlparser-rs.git", features = ["visitor"] } bigdecimal = { version = "0.3", features = ["serde"], optional = true } ort = { version = "1.15.2", optional = true } ndarray = { version = "0.15", optional = true } diff --git a/dozer-sql/expression/src/builder.rs b/dozer-sql/expression/src/builder.rs index cefcda0039..15ce584161 100644 --- a/dozer-sql/expression/src/builder.rs +++ b/dozer-sql/expression/src/builder.rs @@ -219,7 +219,7 @@ impl ExpressionBuilder { .collect(); match matching_by_field.len() { - 1 => Ok(Expression::Column { + 1 if src_table_or_alias.is_none() => Ok(Expression::Column { index: matching_by_field[0].0, }), _ => match src_table_or_alias { @@ -239,7 +239,7 @@ impl ExpressionBuilder { .collect(); match matching_by_table_or_alias.len() { - 1 => Ok(Expression::Column { + 1 if src_connection.is_none() => Ok(Expression::Column { index: matching_by_table_or_alias[0].0, }), _ => match src_connection { diff --git a/dozer-sql/expression/src/execution.rs b/dozer-sql/expression/src/execution.rs index 1c2891b234..ee35beac2a 100644 --- a/dozer-sql/expression/src/execution.rs +++ b/dozer-sql/expression/src/execution.rs @@ -544,7 +544,7 @@ fn get_binary_operator_type( match (left_field_type.return_type, right_field_type.return_type) { (FieldType::Boolean, FieldType::Boolean) => Ok(ExpressionType::new( FieldType::Boolean, - false, + left_field_type.nullable || right_field_type.nullable, SourceDefinition::Dynamic, false, )), @@ -558,7 +558,7 @@ fn get_binary_operator_type( | FieldType::Text, ) => Ok(ExpressionType::new( FieldType::Boolean, - false, + left_field_type.nullable || right_field_type.nullable, SourceDefinition::Dynamic, false, )), @@ -572,7 +572,7 @@ fn get_binary_operator_type( FieldType::Boolean, ) => Ok(ExpressionType::new( FieldType::Boolean, - false, + left_field_type.nullable || right_field_type.nullable, SourceDefinition::Dynamic, false, )), diff --git a/dozer-sql/expression/src/logical.rs b/dozer-sql/expression/src/logical.rs index 63a9017840..74e1823d9d 100644 --- a/dozer-sql/expression/src/logical.rs +++ b/dozer-sql/expression/src/logical.rs @@ -4,72 +4,27 @@ use dozer_types::types::{Field, Schema}; use crate::error::Error; use crate::execution::Expression; +fn boolean_operand(field: Field, operator: &str) -> Result, Error> { + match field { + Field::Boolean(value) => Ok(Some(value)), + Field::Null => Ok(None), + other => Err(Error::InvalidType(other, operator.to_string())), + } +} + pub fn evaluate_and( schema: &Schema, left: &mut Expression, right: &mut Expression, record: &Record, ) -> Result { - let l_field = left.evaluate(record, schema)?; - let r_field = right.evaluate(record, schema)?; - match l_field { - Field::Boolean(true) => match r_field { - Field::Boolean(true) => Ok(Field::Boolean(true)), - Field::Boolean(false) => Ok(Field::Boolean(false)), - Field::Null => Ok(Field::Boolean(false)), - Field::UInt(_) - | Field::U128(_) - | Field::Int(_) - | Field::Int8(_) - | Field::I128(_) - | Field::Float(_) - | Field::String(_) - | Field::Text(_) - | Field::Binary(_) - | Field::Decimal(_) - | Field::Timestamp(_) - | Field::Date(_) - | Field::Json(_) - | Field::Point(_) - | Field::Duration(_) => Err(Error::InvalidType(r_field, "AND".to_string())), - }, - Field::Boolean(false) => match r_field { - Field::Boolean(true) => Ok(Field::Boolean(false)), - Field::Boolean(false) => Ok(Field::Boolean(false)), - Field::Null => Ok(Field::Boolean(false)), - Field::UInt(_) - | Field::U128(_) - | Field::Int(_) - | Field::Int8(_) - | Field::I128(_) - | Field::Float(_) - | Field::String(_) - | Field::Text(_) - | Field::Binary(_) - | Field::Decimal(_) - | Field::Timestamp(_) - | Field::Date(_) - | Field::Json(_) - | Field::Point(_) - | Field::Duration(_) => Err(Error::InvalidType(r_field, "AND".to_string())), - }, - Field::Null => Ok(Field::Boolean(false)), - Field::UInt(_) - | Field::U128(_) - | Field::Int(_) - | Field::Int8(_) - | Field::I128(_) - | Field::Float(_) - | Field::String(_) - | Field::Text(_) - | Field::Binary(_) - | Field::Decimal(_) - | Field::Timestamp(_) - | Field::Date(_) - | Field::Json(_) - | Field::Point(_) - | Field::Duration(_) => Err(Error::InvalidType(l_field, "AND".to_string())), - } + let left = boolean_operand(left.evaluate(record, schema)?, "AND")?; + let right = boolean_operand(right.evaluate(record, schema)?, "AND")?; + Ok(match (left, right) { + (Some(false), _) | (_, Some(false)) => Field::Boolean(false), + (Some(true), Some(true)) => Field::Boolean(true), + _ => Field::Null, + }) } pub fn evaluate_or( @@ -78,65 +33,13 @@ pub fn evaluate_or( right: &mut Expression, record: &Record, ) -> Result { - let l_field = left.evaluate(record, schema)?; - let r_field = right.evaluate(record, schema)?; - match l_field { - Field::Boolean(true) => match r_field { - Field::Boolean(false) => Ok(Field::Boolean(true)), - Field::Boolean(true) => Ok(Field::Boolean(true)), - Field::Null => Ok(Field::Boolean(true)), - Field::UInt(_) - | Field::U128(_) - | Field::Int(_) - | Field::Int8(_) - | Field::I128(_) - | Field::Float(_) - | Field::String(_) - | Field::Text(_) - | Field::Binary(_) - | Field::Decimal(_) - | Field::Timestamp(_) - | Field::Date(_) - | Field::Json(_) - | Field::Point(_) - | Field::Duration(_) => Err(Error::InvalidType(r_field, "OR".to_string())), - }, - Field::Boolean(false) | Field::Null => match right.evaluate(record, schema)? { - Field::Boolean(false) => Ok(Field::Boolean(false)), - Field::Boolean(true) => Ok(Field::Boolean(true)), - Field::Null => Ok(Field::Boolean(false)), - Field::UInt(_) - | Field::U128(_) - | Field::Int(_) - | Field::Int8(_) - | Field::I128(_) - | Field::Float(_) - | Field::String(_) - | Field::Text(_) - | Field::Binary(_) - | Field::Decimal(_) - | Field::Timestamp(_) - | Field::Date(_) - | Field::Json(_) - | Field::Point(_) - | Field::Duration(_) => Err(Error::InvalidType(r_field, "OR".to_string())), - }, - Field::UInt(_) - | Field::U128(_) - | Field::Int(_) - | Field::Int8(_) - | Field::I128(_) - | Field::Float(_) - | Field::String(_) - | Field::Text(_) - | Field::Binary(_) - | Field::Decimal(_) - | Field::Timestamp(_) - | Field::Date(_) - | Field::Json(_) - | Field::Point(_) - | Field::Duration(_) => Err(Error::InvalidType(l_field, "OR".to_string())), - } + let left = boolean_operand(left.evaluate(record, schema)?, "OR")?; + let right = boolean_operand(right.evaluate(record, schema)?, "OR")?; + Ok(match (left, right) { + (Some(true), _) | (_, Some(true)) => Field::Boolean(true), + (Some(false), Some(false)) => Field::Boolean(false), + _ => Field::Null, + }) } pub fn evaluate_not( @@ -235,13 +138,17 @@ mod tests { fn _test_bool_null_and(f1: Field, f2: Field) { let row = Record::new(vec![]); + let expected = if f1 == Field::Boolean(false) || f2 == Field::Boolean(false) { + Field::Boolean(false) + } else { + Field::Null + }; let mut l = Box::new(Literal(f1)); let mut r = Box::new(Literal(f2)); - assert!(matches!( - evaluate_and(&Schema::default(), &mut l, &mut r, &row) - .unwrap_or_else(|e| panic!("{}", e.to_string())), - Field::Boolean(false) - )); + assert_eq!( + evaluate_and(&Schema::default(), &mut l, &mut r, &row).unwrap(), + expected + ); } fn _test_bool_bool_or(bool1: bool, bool2: bool) { @@ -259,22 +166,28 @@ mod tests { let row = Record::new(vec![]); let mut l = Box::new(Literal(Field::Boolean(_bool))); let mut r = Box::new(Literal(Field::Null)); - assert!(matches!( - evaluate_or(&Schema::default(), &mut l, &mut r, &row) - .unwrap_or_else(|e| panic!("{}", e.to_string())), - Field::Boolean(_bool) - )); + assert_eq!( + evaluate_or(&Schema::default(), &mut l, &mut r, &row).unwrap(), + if _bool { + Field::Boolean(true) + } else { + Field::Null + } + ); } fn _test_null_bool_or(_bool: bool) { let row = Record::new(vec![]); let mut l = Box::new(Literal(Field::Null)); let mut r = Box::new(Literal(Field::Boolean(_bool))); - assert!(matches!( - evaluate_or(&Schema::default(), &mut l, &mut r, &row) - .unwrap_or_else(|e| panic!("{}", e.to_string())), - Field::Boolean(_bool) - )); + assert_eq!( + evaluate_or(&Schema::default(), &mut l, &mut r, &row).unwrap(), + if _bool { + Field::Boolean(true) + } else { + Field::Null + } + ); } fn _test_bool_not(bool: bool) { diff --git a/dozer-sql/src/builder/in_subquery.rs b/dozer-sql/src/builder/in_subquery.rs new file mode 100644 index 0000000000..48b0362776 --- /dev/null +++ b/dozer-sql/src/builder/in_subquery.rs @@ -0,0 +1,117 @@ +use std::ops::ControlFlow; + +use dozer_core::{app::AppPipeline, node::PortHandle, DEFAULT_PORT_HANDLE}; +use dozer_sql_expression::{ + builder::NameOrAlias, + sqlparser::ast::{Expr, Ident, Value, VisitMut, VisitorMut}, +}; + +use crate::{errors::PipelineError, product::join::mark::MarkJoinProcessorFactory}; + +use super::{query_to_pipeline, QueryContext, TableInfo}; + +pub(super) fn insert_mark_joins( + predicate: &mut Expr, + input: (String, PortHandle), + pipeline: &mut AppPipeline, + context: &mut QueryContext, + pipeline_idx: usize, +) -> Result<((String, PortHandle), usize), PipelineError> { + let mut visitor = InSubqueryVisitor { + input, + pipeline, + context, + pipeline_idx, + count: 0, + }; + if let ControlFlow::Break(error) = predicate.visit(&mut visitor) { + return Err(error); + } + Ok((visitor.input, visitor.count)) +} + +struct InSubqueryVisitor<'a> { + input: (String, PortHandle), + pipeline: &'a mut AppPipeline, + context: &'a mut QueryContext, + pipeline_idx: usize, + count: usize, +} + +impl InSubqueryVisitor<'_> { + fn insert(&mut self, expression: &mut Expr) -> Result<(), PipelineError> { + if !matches!(expression, Expr::InSubquery { .. }) { + return Ok(()); + } + let Expr::InSubquery { + mut expr, + subquery, + negated, + } = std::mem::replace(expression, Expr::Value(Value::Null)) + else { + unreachable!() + }; + // Visit the left operand in the current scope; build the inner SELECT + // independently so its own IN expressions never bind to outer fields. + if let ControlFlow::Break(error) = expr.visit(self) { + return Err(error); + } + let id = self.context.get_next_processor_id(); + let query_name = format!("__dozer_in_query_{id}"); + let mark_name = format!("__dozer_in_mark_{id}"); + let join_name = format!("mark_join--{id}"); + // Inner WITH names are lexical: they can shadow, but cannot overwrite, + // outer CTE names used by later subqueries in the same predicate. + let outer_tables = self.context.pipeline_map.clone(); + let result = query_to_pipeline( + TableInfo { + name: NameOrAlias(query_name.clone(), None), + override_name: None, + }, + *subquery, + self.pipeline, + self.context, + self.pipeline_idx, + false, + ); + let output = self + .context + .pipeline_map + .get(&(self.pipeline_idx, query_name)) + .cloned(); + self.context.pipeline_map = outer_tables; + result?; + let output = output + .ok_or_else(|| PipelineError::InvalidQuery("IN subquery has no output".into()))?; + self.pipeline.add_processor( + Box::new(MarkJoinProcessorFactory::new( + join_name.clone(), + *expr, + mark_name.clone(), + negated, + self.context.udfs.clone(), + self.context.runtime.clone(), + )), + join_name.clone(), + ); + self.pipeline + .connect_nodes(self.input.0.clone(), self.input.1, join_name.clone(), 0); + self.pipeline + .connect_nodes(output.node, output.port, join_name.clone(), 1); + self.input = (join_name, DEFAULT_PORT_HANDLE); + self.count += 1; + *expression = Expr::Identifier(Ident::new(mark_name)); + Ok(()) + } +} + +impl VisitorMut for InSubqueryVisitor<'_> { + type Break = PipelineError; + + fn pre_visit_expr(&mut self, expression: &mut Expr) -> ControlFlow { + match self.insert(expression) { + Ok(()) => ControlFlow::Continue(()), + Err(error) => ControlFlow::Break(error), + } + } +} diff --git a/dozer-sql/src/builder/mod.rs b/dozer-sql/src/builder/mod.rs index 3ccf3ed356..d25dfc22d2 100644 --- a/dozer-sql/src/builder/mod.rs +++ b/dozer-sql/src/builder/mod.rs @@ -128,6 +128,35 @@ fn query_to_pipeline( query_ctx: &mut QueryContext, pipeline_idx: usize, is_top_select: bool, +) -> Result<(), PipelineError> { + // WITH names belong to this query, including parenthesized UNION operands. + // Only its output relation is visible to the caller after planning. + let outer_tables = query_ctx.pipeline_map.clone(); + let output_key = (pipeline_idx, table_info.name.0.clone()); + let result = query_body_to_pipeline( + table_info, + query, + pipeline, + query_ctx, + pipeline_idx, + is_top_select, + ); + let output = query_ctx.pipeline_map.get(&output_key).cloned(); + query_ctx.pipeline_map = outer_tables; + result?; + if let Some(output) = output { + query_ctx.pipeline_map.insert(output_key, output); + } + Ok(()) +} + +fn query_body_to_pipeline( + table_info: TableInfo, + query: Query, + pipeline: &mut AppPipeline, + query_ctx: &mut QueryContext, + pipeline_idx: usize, + is_top_select: bool, ) -> Result<(), PipelineError> { // return error if there is unsupported syntax if !query.order_by.is_empty() { @@ -150,6 +179,7 @@ fn query_to_pipeline( )); } + let mut cte_names = HashSet::new(); for table in with.cte_tables { if table.from.is_some() { return Err(PipelineError::UnsupportedSqlError( @@ -157,10 +187,7 @@ fn query_to_pipeline( )); } let table_name = table.alias.name.to_string(); - if query_ctx - .pipeline_map - .contains_key(&(pipeline_idx, table_name.clone())) - { + if !cte_names.insert(table_name.clone()) { return Err(InvalidQuery(format!( "WITH query name {table_name:?} specified more than once" ))); @@ -190,21 +217,14 @@ fn query_to_pipeline( is_top_select, )?; } - SetExpr::Query(query) => { - let query_name = format!("subquery_{}", query_ctx.get_next_processor_id()); - let mut ctx = QueryContext::new(query_ctx.udfs.clone(), query_ctx.runtime.clone()); - query_to_pipeline( - TableInfo { - name: NameOrAlias(query_name, None), - override_name: None, - }, - *query, - pipeline, - &mut ctx, - pipeline_idx, - false, //Inside a subquery, so not top select - )? - } + SetExpr::Query(query) => query_to_pipeline( + table_info, + *query, + pipeline, + query_ctx, + pipeline_idx, + is_top_select, + )?, SetExpr::SetOperation { op, set_quantifier, @@ -293,19 +313,27 @@ fn select_to_pipeline( pipeline.add_processor(Box::new(aggregation), gen_agg_name.clone()); // Where clause - if let Some(selection) = select.selection { + if let Some(mut selection) = select.selection { + let (selection_input, internal_fields) = in_subquery::insert_mark_joins( + &mut selection, + (gen_product_name, product_output_port), + pipeline, + query_ctx, + pipeline_idx, + )?; let selection = SelectionProcessorFactory::new( gen_selection_name.clone(), selection, query_ctx.udfs.clone(), query_ctx.runtime.clone(), - ); + ) + .with_internal_fields(internal_fields); pipeline.add_processor(Box::new(selection), gen_selection_name.clone()); pipeline.connect_nodes( - gen_product_name, - product_output_port, + selection_input.0, + selection_input.1, gen_selection_name.clone(), DEFAULT_PORT_HANDLE, ); @@ -383,6 +411,17 @@ fn set_to_pipeline( }; let _left_pipeline_name = match *left_select { + SetExpr::Query(query) => { + query_to_pipeline( + left_table_info, + *query, + pipeline, + query_ctx, + pipeline_idx, + false, + )?; + gen_left_set_name.clone() + } SetExpr::Select(select) => select_to_pipeline( left_table_info, *select, @@ -414,6 +453,17 @@ fn set_to_pipeline( }; let _right_pipeline_name = match *right_select { + SetExpr::Query(query) => { + query_to_pipeline( + right_table_info, + *query, + pipeline, + query_ctx, + pipeline_idx, + false, + )?; + gen_right_set_name.clone() + } SetExpr::Select(select) => select_to_pipeline( right_table_info, *select, @@ -560,6 +610,7 @@ struct ConnectionInfo { mod common; mod from; +mod in_subquery; mod join; mod table_operator; diff --git a/dozer-sql/src/product/join/mark.rs b/dozer-sql/src/product/join/mark.rs new file mode 100644 index 0000000000..b122456250 --- /dev/null +++ b/dozer-sql/src/product/join/mark.rs @@ -0,0 +1,448 @@ +//! A null-aware mark join for a scalar IN subquery. Unlike an inner join, an +//! inner duplicate does not multiply outer rows. The appended mark lets the +//! normal selection processor evaluate arbitrary compound WHERE expressions. + +use std::{collections::HashMap, sync::Arc}; + +use dozer_core::{ + channels::ProcessorChannelForwarder, + epoch::Epoch, + event::EventHub, + node::{PortHandle, Processor, ProcessorFactory}, + DEFAULT_PORT_HANDLE, +}; +use dozer_sql_expression::{ + builder::ExpressionBuilder, execution::Expression, operator::BinaryOperatorType, + sqlparser::ast::Expr, +}; +use dozer_types::{ + errors::internal::BoxedError, + models::udf_config::UdfConfig, + tonic::async_trait, + types::{Field, FieldDefinition, FieldType, Operation, Record, Schema, TableOperation}, +}; +use tokio::runtime::Runtime; + +use crate::errors::PipelineError; + +use super::factory::{LEFT_JOIN_PORT, RIGHT_JOIN_PORT}; + +#[derive(Debug)] +pub(crate) struct MarkJoinProcessorFactory { + id: String, + expression: Expr, + mark_name: String, + negated: bool, + udfs: Vec, + runtime: Arc, +} + +impl MarkJoinProcessorFactory { + pub(crate) fn new( + id: String, + expression: Expr, + mark_name: String, + negated: bool, + udfs: Vec, + runtime: Arc, + ) -> Self { + Self { + id, + expression, + mark_name, + negated, + udfs, + runtime, + } + } + + fn schemas<'a>( + &self, + schemas: &'a HashMap, + ) -> Result<(&'a Schema, &'a Schema), PipelineError> { + let left = schemas + .get(&LEFT_JOIN_PORT) + .ok_or(PipelineError::InvalidPortHandle(LEFT_JOIN_PORT))?; + let right = schemas + .get(&RIGHT_JOIN_PORT) + .ok_or(PipelineError::InvalidPortHandle(RIGHT_JOIN_PORT))?; + if right.fields.len() != 1 { + return Err(PipelineError::InvalidQuery( + "IN subquery must return exactly one column".into(), + )); + } + if left.fields.iter().any(|field| field.name == self.mark_name) { + return Err(PipelineError::InvalidQuery(format!( + "Reserved IN subquery column: {}", + self.mark_name + ))); + } + Ok((left, right)) + } +} + +#[async_trait] +impl ProcessorFactory for MarkJoinProcessorFactory { + fn id(&self) -> String { + self.id.clone() + } + fn type_name(&self) -> String { + "MarkJoin".into() + } + fn get_input_ports(&self) -> Vec { + vec![LEFT_JOIN_PORT, RIGHT_JOIN_PORT] + } + fn get_output_ports(&self) -> Vec { + vec![DEFAULT_PORT_HANDLE] + } + + async fn get_output_schema( + &self, + _port: &PortHandle, + schemas: &HashMap, + ) -> Result { + let (left, _) = self.schemas(schemas)?; + let mut output = left.clone(); + output.fields.push(FieldDefinition::new( + self.mark_name.clone(), + FieldType::Boolean, + true, + Default::default(), + )); + Ok(output) + } + + async fn build( + &self, + schemas: HashMap, + _outputs: HashMap, + _hub: EventHub, + ) -> Result, BoxedError> { + let (left, right) = self.schemas(&schemas)?; + let expression = ExpressionBuilder::new(left.fields.len(), self.runtime.clone()) + .build(false, &self.expression, left, &self.udfs) + .await?; + let same_type = matches!(&expression, Expression::Literal(Field::Null)) + || expression.get_type(left)?.return_type == right.fields[0].typ; + Ok(Box::new(MarkJoinProcessor { + expression, + schema: left.clone(), + same_type, + negated: self.negated, + left: HashMap::new(), + right: HashMap::new(), + right_len: 0, + })) + } +} + +#[derive(Debug, Default)] +struct OuterGroup { + records: HashMap, + matches: usize, +} + +#[derive(Debug)] +struct MarkJoinProcessor { + expression: Expression, + schema: Schema, + same_type: bool, + negated: bool, + left: HashMap, + right: HashMap, + right_len: usize, +} + +// A batch/update is applied as a single change. In particular replacing one +// matching inner row with another must not produce a transient retraction. +#[derive(Default)] +struct Delta { + removed: usize, + added: usize, +} + +impl Delta { + fn apply(&self, count: usize) -> Result { + count + .checked_sub(self.removed) + .and_then(|count| count.checked_add(self.added)) + .ok_or_else(|| PipelineError::InvalidValue("Unbalanced IN subquery change".into())) + } +} + +impl MarkJoinProcessor { + fn equals(&self, left: &Field, right: &Field) -> Result { + if left == &Field::Null || right == &Field::Null { + return Ok(false); + } + if self.same_type { + return Ok(left == right); + } + match BinaryOperatorType::Eq.evaluate( + &self.schema, + &mut Expression::Literal(left.clone()), + &mut Expression::Literal(right.clone()), + &Record::default(), + )? { + Field::Boolean(equal) => Ok(equal), + _ => Err(PipelineError::InvalidQuery(format!( + "IN subquery operands cannot be compared: {left:?} and {right:?}" + ))), + } + } + + fn match_count(&self, key: &Field) -> Result { + if let Some(group) = self.left.get(key) { + return Ok(group.matches); + } + if key == &Field::Null { + return Ok(0); + } + if self.same_type { + return Ok(self.right.get(key).copied().unwrap_or(0)); + } + self.right.iter().try_fold(0, |total, (value, count)| { + Ok(total + if self.equals(key, value)? { *count } else { 0 }) + }) + } + + fn mark(&self, key: &Field, matches: usize, nulls: usize, total: usize) -> Field { + let included = if total == 0 { + false + } else if matches > 0 { + true + } else if key == &Field::Null || nulls > 0 { + return Field::Null; + } else { + false + }; + Field::Boolean(included != self.negated) + } + + fn marked(&self, record: &Record, key: &Field, matches: usize) -> Record { + let mut record = record.clone(); + record.values.push(self.mark( + key, + matches, + self.right.get(&Field::Null).copied().unwrap_or(0), + self.right_len, + )); + record + } + + fn insert_left(&mut self, record: Record, key: Field, matches: usize) { + let group = self.left.entry(key).or_insert_with(|| OuterGroup { + records: HashMap::new(), + matches, + }); + *group.records.entry(record).or_default() += 1; + } + + fn delete_left(&mut self, record: &Record, key: &Field) -> Result<(), PipelineError> { + let group = self.left.get_mut(key).ok_or_else(|| { + PipelineError::InvalidValue("Unknown outer row in IN subquery delete".into()) + })?; + let count = group.records.get_mut(record).ok_or_else(|| { + PipelineError::InvalidValue("Unknown outer row in IN subquery delete".into()) + })?; + *count -= 1; + if *count == 0 { + group.records.remove(record); + } + if group.records.is_empty() { + self.left.remove(key); + } + Ok(()) + } + + fn process_left(&mut self, op: Operation) -> Result { + Ok(match op { + Operation::Insert { new } => { + let key = self.expression.evaluate(&new, &self.schema)?; + let matches = self.match_count(&key)?; + let marked = self.marked(&new, &key, matches); + self.insert_left(new, key, matches); + Operation::Insert { new: marked } + } + Operation::Delete { old } => { + let key = self.expression.evaluate(&old, &self.schema)?; + let matches = self.match_count(&key)?; + let marked = self.marked(&old, &key, matches); + self.delete_left(&old, &key)?; + Operation::Delete { old: marked } + } + Operation::Update { old, new } => { + let old_key = self.expression.evaluate(&old, &self.schema)?; + let new_key = self.expression.evaluate(&new, &self.schema)?; + let old_matches = self.match_count(&old_key)?; + let new_matches = self.match_count(&new_key)?; + let marked_old = self.marked(&old, &old_key, old_matches); + let marked_new = self.marked(&new, &new_key, new_matches); + self.delete_left(&old, &old_key)?; + self.insert_left(new, new_key, new_matches); + Operation::Update { + old: marked_old, + new: marked_new, + } + } + Operation::BatchInsert { new } => { + let prepared = new + .into_iter() + .map(|record| { + let key = self.expression.evaluate(&record, &self.schema)?; + let matches = self.match_count(&key)?; + Ok((record, key, matches)) + }) + .collect::, PipelineError>>()?; + let mut new = Vec::with_capacity(prepared.len()); + for (record, key, matches) in prepared { + new.push(self.marked(&record, &key, matches)); + self.insert_left(record, key, matches); + } + Operation::BatchInsert { new } + } + }) + } + + fn process_right(&mut self, op: Operation) -> Result, PipelineError> { + let mut deltas: HashMap = HashMap::new(); + let (removed, added) = match op { + Operation::Insert { new } => (vec![], vec![new]), + Operation::Delete { old } => (vec![old], vec![]), + Operation::Update { old, new } => (vec![old], vec![new]), + Operation::BatchInsert { new } => (vec![], new), + }; + let total = Delta { + removed: removed.len(), + added: added.len(), + } + .apply(self.right_len)?; + for record in removed { + deltas + .entry(record.get_value(0)?.clone()) + .or_default() + .removed += 1; + } + for record in added { + deltas + .entry(record.get_value(0)?.clone()) + .or_default() + .added += 1; + } + let mut next_counts = HashMap::new(); + for (key, delta) in &deltas { + next_counts.insert( + key.clone(), + delta.apply(self.right.get(key).copied().unwrap_or(0))?, + ); + } + let old_nulls = self.right.get(&Field::Null).copied().unwrap_or(0); + let nulls = next_counts.get(&Field::Null).copied().unwrap_or(old_nulls); + let global_change = + (old_nulls == 0) != (nulls == 0) || (self.right_len == 0) != (total == 0); + // Same-type keys use a hash lookup. Cross-type keys use the same SQL + // equality/coercion implementation as WHERE, over distinct outer keys. + let affected: Vec<_> = if global_change || !self.same_type { + self.left.keys().cloned().collect() + } else { + deltas + .keys() + .filter(|key| **key != Field::Null && self.left.contains_key(*key)) + .cloned() + .collect() + }; + let mut changes = Vec::new(); + for key in affected { + let group = &self.left[&key]; + let mut matching_delta = Delta::default(); + if self.same_type { + if key != Field::Null { + if let Some(delta) = deltas.get(&key) { + matching_delta.added = delta.added; + matching_delta.removed = delta.removed; + } + } + } else { + for (value, delta) in &deltas { + if self.equals(&key, value)? { + matching_delta.added += delta.added; + matching_delta.removed += delta.removed; + } + } + } + let matches = matching_delta.apply(group.matches)?; + let old_mark = self.mark(&key, group.matches, old_nulls, self.right_len); + let new_mark = self.mark(&key, matches, nulls, total); + changes.push((key, matches, old_mark, new_mark)); + } + let mut output = Vec::new(); + for (key, matches, old_mark, new_mark) in changes { + let group = self.left.get_mut(&key).unwrap(); + group.matches = matches; + if old_mark != new_mark { + for (record, count) in &group.records { + let mut old = record.clone(); + old.values.push(old_mark.clone()); + let mut new = record.clone(); + new.values.push(new_mark.clone()); + for _ in 0..*count { + output.push(Operation::Update { + old: old.clone(), + new: new.clone(), + }); + } + } + } + } + for (key, count) in next_counts { + if count == 0 { + self.right.remove(&key); + } else { + self.right.insert(key, count); + } + } + self.right_len = total; + Ok(output) + } +} + +impl Processor for MarkJoinProcessor { + fn commit(&self, _epoch: &Epoch) -> Result<(), BoxedError> { + Ok(()) + } + + fn process( + &mut self, + op: TableOperation, + fw: &mut dyn ProcessorChannelForwarder, + ) -> Result<(), BoxedError> { + // TTL expiration currently has no explicit delete event in Dozer. + // Silently expiring a join key would leave downstream marks stale, so + // reject this combination until expiration can propagate atomically. + let has_lifetime = match &op.op { + Operation::Insert { new } => new.lifetime.is_some(), + Operation::Delete { old } => old.lifetime.is_some(), + Operation::Update { old, new } => old.lifetime.is_some() || new.lifetime.is_some(), + Operation::BatchInsert { new } => new.iter().any(|record| record.lifetime.is_some()), + }; + if has_lifetime { + return Err(PipelineError::InvalidQuery( + "IN subqueries do not support TTL inputs".into(), + ) + .into()); + } + let output = match op.port { + LEFT_JOIN_PORT => vec![self.process_left(op.op)?], + RIGHT_JOIN_PORT => self.process_right(op.op)?, + port => return Err(PipelineError::InvalidPortHandle(port).into()), + }; + for change in output { + fw.send(TableOperation { + id: op.id, + op: change, + port: DEFAULT_PORT_HANDLE, + }); + } + Ok(()) + } +} diff --git a/dozer-sql/src/product/join/mod.rs b/dozer-sql/src/product/join/mod.rs index e3e84169aa..2d3fc78820 100644 --- a/dozer-sql/src/product/join/mod.rs +++ b/dozer-sql/src/product/join/mod.rs @@ -1,6 +1,7 @@ use crate::errors::JoinError; pub mod factory; +pub(crate) mod mark; pub(crate) mod operator; mod processor; diff --git a/dozer-sql/src/selection/factory.rs b/dozer-sql/src/selection/factory.rs index ab5567584c..2f9d00676e 100644 --- a/dozer-sql/src/selection/factory.rs +++ b/dozer-sql/src/selection/factory.rs @@ -20,6 +20,7 @@ pub struct SelectionProcessorFactory { id: String, udfs: Vec, runtime: Arc, + internal_fields: usize, } impl SelectionProcessorFactory { @@ -35,8 +36,15 @@ impl SelectionProcessorFactory { id, udfs: udf_config, runtime, + internal_fields: 0, } } + + /// Remove the trailing mark-join fields after evaluating WHERE. + pub(crate) fn with_internal_fields(mut self, count: usize) -> Self { + self.internal_fields = count; + self + } } #[async_trait] @@ -63,7 +71,15 @@ impl ProcessorFactory for SelectionProcessorFactory { let schema = input_schemas .get(&DEFAULT_PORT_HANDLE) .ok_or(PipelineError::InvalidPortHandle(DEFAULT_PORT_HANDLE))?; - Ok(schema.clone()) + let mut output = schema.clone(); + let width = output + .fields + .len() + .checked_sub(self.internal_fields) + .ok_or_else(|| PipelineError::InvalidQuery("Invalid internal WHERE fields".into()))?; + output.fields.truncate(width); + output.primary_index.retain(|index| *index < width); + Ok(output) } async fn build( @@ -80,10 +96,10 @@ impl ProcessorFactory for SelectionProcessorFactory { .build(false, &self.statement, schema, &self.udfs) .await { - Ok(expression) => Ok(Box::new(SelectionProcessor::new( - schema.clone(), - expression, - )?)), + Ok(expression) => Ok(Box::new( + SelectionProcessor::new(schema.clone(), expression)? + .with_internal_fields(self.internal_fields), + )), Err(e) => Err(e.into()), } } diff --git a/dozer-sql/src/selection/processor.rs b/dozer-sql/src/selection/processor.rs index 8881286e86..16d3d6e576 100644 --- a/dozer-sql/src/selection/processor.rs +++ b/dozer-sql/src/selection/processor.rs @@ -12,6 +12,7 @@ use crate::errors::PipelineError; pub struct SelectionProcessor { expression: Expression, input_schema: Schema, + internal_fields: usize, } impl SelectionProcessor { @@ -19,9 +20,22 @@ impl SelectionProcessor { Ok(Self { input_schema, expression, + internal_fields: 0, }) } + pub(crate) fn with_internal_fields(mut self, count: usize) -> Self { + self.internal_fields = count; + self + } + + fn project(&self, mut record: Record) -> Record { + record + .values + .truncate(record.values.len() - self.internal_fields); + record + } + fn filter(&mut self, record: &Record) -> Result { Ok(self.expression.evaluate(record, &self.input_schema)? == Field::Boolean(true)) } @@ -34,25 +48,47 @@ impl Processor for SelectionProcessor { fn process( &mut self, - mut op: TableOperation, + op: TableOperation, fw: &mut dyn ProcessorChannelForwarder, ) -> Result<(), BoxedError> { match op.op { - Operation::Delete { ref old } => { - if self.filter(old)? { - op.port = DEFAULT_PORT_HANDLE; - fw.send(op); + Operation::Delete { old } => { + if self.filter(&old)? { + fw.send(TableOperation { + id: op.id, + op: Operation::Delete { + old: self.project(old), + }, + port: DEFAULT_PORT_HANDLE, + }); } } - Operation::Insert { ref new } => { - if self.filter(new)? { - op.port = DEFAULT_PORT_HANDLE; - fw.send(op); + Operation::Insert { new } => { + if self.filter(&new)? { + fw.send(TableOperation { + id: op.id, + op: Operation::Insert { + new: self.project(new), + }, + port: DEFAULT_PORT_HANDLE, + }); } } Operation::Update { old, new } => { let old_fulfilled = self.filter(&old)?; let new_fulfilled = self.filter(&new)?; + let changed = old != new; + let old = self.project(old); + let new = self.project(new); + // A mark changed, but the projected row and compound predicate + // may be unchanged (for example, the other arm of an OR is true). + if self.internal_fields > 0 + && changed + && old_fulfilled == new_fulfilled + && old == new + { + return Ok(()); + } match (old_fulfilled, new_fulfilled) { (true, true) => { // both records fulfills the WHERE condition, forward the operation @@ -88,7 +124,13 @@ impl Processor for SelectionProcessor { .into_iter() .filter_map(|record| { self.filter(&record) - .map(|fulfilled| if fulfilled { Some(record) } else { None }) + .map(|fulfilled| { + if fulfilled { + Some(self.project(record)) + } else { + None + } + }) .transpose() }) .collect::, _>>()?; diff --git a/dozer-sql/src/tests/in_subquery.rs b/dozer-sql/src/tests/in_subquery.rs new file mode 100644 index 0000000000..eab1ec9402 --- /dev/null +++ b/dozer-sql/src/tests/in_subquery.rs @@ -0,0 +1,930 @@ +//! Exercise actual parser, DAG wiring, schema propagation and processors. The +//! deterministic driver controls the order of changes on both source ports. +use std::collections::{HashMap, VecDeque}; + +use dozer_core::{ + app::{App, AppPipeline}, + appsource::{AppSourceManager, AppSourceMappings}, + channels::ProcessorChannelForwarder, + dag_schemas::{DagHaveSchemas, DagSchemas}, + event::EventHub, + node::{OutputPortDef, OutputPortType, PortHandle, Processor, Source, SourceFactory}, + petgraph::visit::EdgeRef, + NodeKind, DEFAULT_PORT_HANDLE, +}; +use dozer_types::{ + errors::internal::BoxedError, + types::{Field, FieldDefinition, FieldType, Operation, Record, Schema, TableOperation}, +}; + +use super::{ + builder_test::{TestSinkFactory, TestSource}, + utils::create_test_runtime, +}; +use crate::builder::statement_to_pipeline; + +#[derive(Debug)] +struct Sources(Vec); +impl SourceFactory for Sources { + fn get_output_ports(&self) -> Vec { + (0..self.0.len()) + .map(|port| OutputPortDef::new(port as u16, OutputPortType::Stateless)) + .collect() + } + fn get_output_schema(&self, port: &PortHandle) -> Result { + Ok(self.0[*port as usize].clone()) + } + fn get_output_port_name(&self, port: &PortHandle) -> String { + format!("table_{port}") + } + fn build( + &self, + _: HashMap, + _: EventHub, + _: Option>, + ) -> Result, BoxedError> { + Ok(Box::new(TestSource {})) + } +} + +#[derive(Default)] +struct Output(Vec); +impl ProcessorChannelForwarder for Output { + fn send(&mut self, op: TableOperation) { + self.0.push(op); + } +} + +struct Pipeline { + processors: HashMap>, + edges: HashMap<(usize, PortHandle), Vec<(usize, PortHandle)>>, + source: usize, + sink: usize, + schema: Schema, + rows: Vec, +} + +fn schema(value_type: FieldType) -> Schema { + Schema { + fields: vec![ + FieldDefinition::new("id".into(), FieldType::Int, false, Default::default()), + FieldDefinition::new("value".into(), value_type, true, Default::default()), + ], + primary_index: vec![], + } +} + +impl Pipeline { + fn new(sql: &str) -> Self { + Self::build(sql, vec![schema(FieldType::Int); 3]).unwrap() + } + + fn build(sql: &str, schemas: Vec) -> Result { + let runtime = create_test_runtime(); + let mut pipeline = AppPipeline::new_with_default_flags(); + let context = statement_to_pipeline( + sql, + &mut pipeline, + Some("results".into()), + vec![], + runtime.clone(), + )?; + let output = &context.output_tables_map["results"]; + pipeline.add_sink( + Box::new(TestSinkFactory::new(vec![DEFAULT_PORT_HANDLE])), + "sink".into(), + ); + pipeline.connect_nodes( + output.node.clone(), + output.port, + "sink".into(), + DEFAULT_PORT_HANDLE, + ); + let mut sources = AppSourceManager::new(); + let names = ["outer_rows", "inner_rows", "third_rows"]; + let mappings = names + .iter() + .enumerate() + .take(schemas.len()) + .map(|(port, name)| (name.to_string(), port as u16)) + .collect(); + sources.add( + Box::new(Sources(schemas)), + AppSourceMappings::new("memory".into(), mappings), + )?; + let mut app = App::new(sources); + app.add_pipeline(pipeline); + let dag = runtime.block_on(DagSchemas::new(app.into_dag()?))?; + let graph = dag.graph(); + let mut result = Self { + processors: HashMap::new(), + edges: HashMap::new(), + source: 0, + sink: 0, + schema: Schema::default(), + rows: vec![], + }; + for node in graph.graph().node_indices() { + match &graph[node].kind { + NodeKind::Source(_) => result.source = node.index(), + NodeKind::Sink(_) => { + result.sink = node.index(); + result.schema = dag.get_node_input_schemas(node)[&DEFAULT_PORT_HANDLE].clone(); + } + NodeKind::Processor(factory) => { + let processor = runtime.block_on(factory.build( + dag.get_node_input_schemas(node), + dag.get_node_output_schemas(node), + EventHub::new(16), + ))?; + result.processors.insert(node.index(), processor); + } + } + } + for edge in graph.graph().edge_references() { + result + .edges + .entry((edge.source().index(), edge.weight().output_port)) + .or_default() + .push((edge.target().index(), edge.weight().input_port)); + } + Ok(result) + } + + fn send(&mut self, source_port: PortHandle, op: Operation) -> Vec { + let mut pending = + VecDeque::from([(self.source, TableOperation::without_id(op, source_port))]); + let mut emitted = Vec::new(); + while let Some((from, op)) = pending.pop_front() { + for &(target, port) in self.edges.get(&(from, op.port)).into_iter().flatten() { + let mut input = op.clone(); + input.port = port; + if target == self.sink { + emitted.push(input.op); + } else { + let mut output = Output::default(); + self.processors + .get_mut(&target) + .unwrap() + .process(input, &mut output) + .unwrap(); + pending.extend(output.0.into_iter().map(|op| (target, op))); + } + } + } + for op in &emitted { + match op { + Operation::Insert { new } => self.rows.push(new.clone()), + Operation::Delete { old } => { + let index = self + .rows + .iter() + .position(|row| row == old) + .expect("retracted row must exist"); + self.rows.remove(index); + } + Operation::Update { old, new } => { + let index = self + .rows + .iter() + .position(|row| row == old) + .expect("updated row must exist"); + self.rows[index] = new.clone(); + } + Operation::BatchInsert { new } => self.rows.extend(new.clone()), + } + } + emitted + } + + fn expect(&self, rows: &[Record]) { + let bag = |rows: &[Record]| { + let mut counts = HashMap::new(); + for row in rows { + *counts.entry(row.clone()).or_insert(0) += 1; + } + counts + }; + assert_eq!(bag(&self.rows), bag(rows)); + } +} + +fn row(id: i64, value: i64) -> Record { + Record::new(vec![Field::Int(id), Field::Int(value)]) +} +fn null_row(id: i64) -> Record { + Record::new(vec![Field::Int(id), Field::Null]) +} +fn insert(new: Record) -> Operation { + Operation::Insert { new } +} +fn delete(old: Record) -> Operation { + Operation::Delete { old } +} +fn update(old: Record, new: Record) -> Operation { + Operation::Update { old, new } +} +const IN_SQL: &str = "SELECT * FROM outer_rows WHERE value IN (SELECT value FROM inner_rows)"; + +#[test] +fn in_subquery_late_inner_changes_and_duplicates() { + let mut p = Pipeline::new(IN_SQL); + assert_eq!( + p.schema.fields.len(), + 2, + "SELECT * must hide internal marks" + ); + assert!(p.send(0, insert(row(1, 10))).is_empty()); + p.send(0, insert(row(1, 10))); + p.send(0, insert(row(2, 20))); + p.send(1, insert(row(101, 10))); + p.expect(&[row(1, 10), row(1, 10)]); + assert!(p.send(1, insert(row(102, 10))).is_empty()); + assert!(p.send(1, delete(row(101, 10))).is_empty()); + p.send(0, delete(row(1, 10))); + p.expect(&[row(1, 10)]); + p.send(1, delete(row(102, 10))); + p.expect(&[]); + p.send(1, insert(row(103, 20))); + p.expect(&[row(2, 20)]); +} + +#[test] +fn in_subquery_outer_updates_cover_where_transitions() { + let mut p = Pipeline::new(IN_SQL); + p.send( + 1, + Operation::BatchInsert { + new: vec![row(101, 10), row(102, 20)], + }, + ); + p.send(0, insert(row(1, 30))); + assert!(p.send(0, update(row(1, 30), row(1, 40))).is_empty()); + p.send(0, update(row(1, 40), row(1, 10))); + p.expect(&[row(1, 10)]); + assert!(matches!( + &p.send(0, update(row(1, 10), row(1, 10)))[..], + [Operation::Update { .. }] + )); + assert!(matches!( + &p.send(0, update(row(1, 10), row(1, 20)))[..], + [Operation::Update { .. }] + )); + p.expect(&[row(1, 20)]); + p.send(0, update(row(1, 20), row(1, 30))); + p.expect(&[]); + p.send(0, delete(row(1, 30))); + p.send(1, insert(row(103, 30))); + p.expect(&[]); +} + +#[test] +fn in_subquery_inner_updates_are_atomic() { + let mut p = Pipeline::new(IN_SQL); + p.send( + 0, + Operation::BatchInsert { + new: vec![row(1, 10), row(2, 20)], + }, + ); + p.send(1, insert(row(101, 10))); + assert!(p.send(1, update(row(101, 10), row(102, 10))).is_empty()); + p.send(1, update(row(102, 10), row(102, 20))); + p.expect(&[row(2, 20)]); +} + +#[test] +fn in_subquery_null_and_empty_set_truth_tables() { + for negated in [false, true] { + let sql = format!( + "SELECT * FROM outer_rows WHERE value {}IN (SELECT value FROM inner_rows)", + if negated { "NOT " } else { "" } + ); + let mut p = Pipeline::new(&sql); + p.send( + 0, + Operation::BatchInsert { + new: vec![row(1, 10), null_row(2)], + }, + ); + let expected = if negated { + vec![row(1, 10), null_row(2)] + } else { + vec![] + }; + p.expect(&expected); + p.send(1, insert(null_row(101))); + p.expect(&[]); + p.send(1, insert(row(102, 10))); + let expected = if negated { vec![] } else { vec![row(1, 10)] }; + p.expect(&expected); + p.send(1, delete(null_row(101))); + let expected = if negated { vec![] } else { vec![row(1, 10)] }; + p.expect(&expected); + p.send(1, delete(row(102, 10))); + let expected = if negated { + vec![row(1, 10), null_row(2)] + } else { + vec![] + }; + p.expect(&expected); + } +} + +#[test] +fn in_subquery_compound_predicates_preserve_three_valued_logic() { + let mut p = Pipeline::new("SELECT * FROM outer_rows WHERE (value IN (SELECT value FROM inner_rows) OR id = 1) AND value NOT IN (SELECT value FROM third_rows)"); + p.send( + 0, + Operation::BatchInsert { + new: vec![row(1, 10), row(2, 20)], + }, + ); + p.expect(&[row(1, 10)]); + assert!( + p.send(1, insert(row(101, 10))).is_empty(), + "unchanged OR arm must not emit a spurious update" + ); + p.send(1, insert(row(102, 20))); + p.expect(&[row(1, 10), row(2, 20)]); + p.send(2, insert(null_row(201))); + p.expect(&[]); + p.send(2, delete(null_row(201))); + p.expect(&[row(1, 10), row(2, 20)]); + p.send(2, insert(row(202, 20))); + p.expect(&[row(1, 10)]); +} + +#[test] +fn in_subquery_nested_select_and_cte() { + let mut p = Pipeline::new("WITH allowed AS (SELECT value FROM third_rows) SELECT * FROM outer_rows WHERE value IN (SELECT value FROM inner_rows WHERE value IN (SELECT value FROM allowed))"); + p.send(0, insert(row(1, 10))); + p.send(1, insert(row(101, 10))); + p.expect(&[]); + p.send(2, insert(row(201, 10))); + p.expect(&[row(1, 10)]); + p.send(2, delete(row(201, 10))); + p.expect(&[]); +} + +#[test] +fn in_subquery_mixed_numeric_and_string_values_use_sql_equality() { + for (typ, value) in [ + (FieldType::UInt, Field::UInt(10)), + (FieldType::String, Field::String("10".into())), + ] { + let mut p = Pipeline::build(IN_SQL, vec![schema(FieldType::Int), schema(typ)]).unwrap(); + p.send(0, insert(row(1, 10))); + let inner = Record::new(vec![Field::Int(101), value]); + p.send(1, insert(inner.clone())); + p.expect(&[row(1, 10)]); + p.send(1, delete(inner)); + p.expect(&[]); + } +} + +#[test] +fn in_subquery_validates_width_and_correlated_references() { + let error = Pipeline::build( + "SELECT * FROM outer_rows WHERE value IN (SELECT * FROM inner_rows)", + vec![schema(FieldType::Int); 2], + ) + .err() + .unwrap(); + assert!(error.to_string().contains("exactly one column"), "{error}"); + assert!(Pipeline::build("SELECT * FROM outer_rows o WHERE value IN (SELECT value FROM inner_rows i WHERE i.id = o.id)", vec![schema(FieldType::Int); 2]).is_err()); +} + +#[test] +fn in_subquery_left_expression_and_filtered_inner() { + let mut p = Pipeline::new("SELECT id FROM outer_rows WHERE value + 1 IN (SELECT value FROM inner_rows WHERE id > 100)"); + p.send(0, insert(row(1, 10))); + p.send(1, insert(row(99, 11))); + p.expect(&[]); + p.send(1, update(row(99, 11), row(101, 11))); + p.expect(&[Record::new(vec![Field::Int(1)])]); + p.send(1, update(row(101, 11), row(99, 11))); + p.expect(&[]); +} + +#[test] +fn in_subquery_deterministic_stream_matches_snapshot_oracle() { + for negated in [false, true] { + let sql = format!( + "SELECT * FROM outer_rows WHERE value {}IN (SELECT value FROM inner_rows)", + if negated { "NOT " } else { "" } + ); + let mut p = Pipeline::new(&sql); + let mut tables: [Vec; 2] = Default::default(); + let mut seed = 7_u64; + for step in 0..180 { + seed = seed.wrapping_mul(6364136223846793005).wrapping_add(1); + let side = ((seed >> 32) & 1) as usize; + let value = ((seed >> 36) % 5) as i64; + let new = if value == 4 { + null_row(step) + } else { + row(step, value) + }; + let change = match (seed >> 40) % 3 { + 0 if !tables[side].is_empty() => delete(tables[side].remove(0)), + 1 if !tables[side].is_empty() => { + update(std::mem::replace(&mut tables[side][0], new.clone()), new) + } + _ => { + tables[side].push(new.clone()); + insert(new) + } + }; + p.send(side as u16, change); + let expected: Vec<_> = tables[0] + .iter() + .filter(|outer| { + let matched = tables[1].iter().any(|inner| { + outer.values[1] != Field::Null && outer.values[1] == inner.values[1] + }); + if negated { + tables[1].is_empty() + || (!matched + && outer.values[1] != Field::Null + && tables[1].iter().all(|inner| inner.values[1] != Field::Null)) + } else { + matched + } + }) + .cloned() + .collect(); + p.expect(&expected); + } + } +} + +#[test] +fn in_subquery_negation_keeps_unknown_compounds_unknown() { + for predicate in [ + "NOT (value IN (SELECT value FROM inner_rows) AND 1 = 1)", + "NOT (value IN (SELECT value FROM inner_rows) OR 1 = 2)", + "NOT (1 = 2 OR value IN (SELECT value FROM inner_rows))", + ] { + let mut p = Pipeline::new(&format!("SELECT * FROM outer_rows WHERE {predicate}")); + p.send(1, insert(null_row(101))); + p.send(0, insert(row(1, 10))); + p.expect(&[]); + } +} + +#[test] +fn in_subquery_inner_cte_shadows_without_leaking() { + let mut p = Pipeline::new("WITH ids AS (SELECT value FROM third_rows) SELECT * FROM outer_rows WHERE value IN (WITH ids AS (SELECT value FROM inner_rows) SELECT value FROM ids) AND value IN (SELECT value FROM ids)"); + p.send(0, insert(row(1, 10))); + p.send(1, insert(row(101, 10))); + p.expect(&[]); + p.send(2, insert(row(201, 10))); + p.expect(&[row(1, 10)]); +} + +// Supplemental cases for dozer-sql/src/tests/in_subquery.rs. +// Uses its real SQL-to-DAG Pipeline fixture. Kept outside the root checkout +// while the implementation is being edited by the parent agent. + +#[test] +fn in_subquery_review_duplicate_null_batch_is_one_membership_change() { + let mut p = Pipeline::new(IN_SQL); + p.send( + 0, + Operation::BatchInsert { + new: vec![row(1, 10), row(1, 10), row(2, 20), null_row(3)], + }, + ); + let emitted = p.send( + 1, + Operation::BatchInsert { + new: vec![row(101, 10), null_row(102), row(103, 10)], + }, + ); + assert_eq!(emitted.len(), 2, "each outer duplicate enters once"); + assert!(emitted + .iter() + .all(|op| matches!(op, Operation::Insert { .. }))); + p.expect(&[row(1, 10), row(1, 10)]); + assert!(p.send(1, delete(row(101, 10))).is_empty()); + assert_eq!(p.send(1, delete(row(103, 10))).len(), 2); + p.expect(&[]); + assert!(p.send(1, delete(null_row(102))).is_empty()); +} + +#[test] +fn in_subquery_review_not_in_batch_null_then_last_match_removal() { + let mut p = + Pipeline::new("SELECT * FROM outer_rows WHERE value NOT IN (SELECT value FROM inner_rows)"); + p.send( + 0, + Operation::BatchInsert { + new: vec![row(1, 10), row(2, 20), null_row(3)], + }, + ); + let emitted = p.send( + 1, + Operation::BatchInsert { + new: vec![row(101, 10), null_row(102), row(103, 10)], + }, + ); + assert_eq!(emitted.len(), 3); + assert!(emitted + .iter() + .all(|op| matches!(op, Operation::Delete { .. }))); + p.expect(&[]); + assert!(p.send(1, delete(row(101, 10))).is_empty()); + assert_eq!(p.send(1, delete(null_row(102))).len(), 1); + p.expect(&[row(2, 20)]); + assert_eq!(p.send(1, delete(row(103, 10))).len(), 2); + p.expect(&[row(1, 10), row(2, 20), null_row(3)]); +} + +#[test] +fn in_subquery_review_distinct_cross_type_keys_share_membership_count() { + let mut p = Pipeline::build( + IN_SQL, + vec![schema(FieldType::Int), schema(FieldType::String)], + ) + .unwrap(); + p.send( + 0, + Operation::BatchInsert { + new: vec![row(1, 10), row(2, 11)], + }, + ); + let string_row = + |id, value: &str| Record::new(vec![Field::Int(id), Field::String(value.into())]); + p.send( + 1, + Operation::BatchInsert { + new: vec![string_row(101, "10"), string_row(102, "010")], + }, + ); + p.expect(&[row(1, 10)]); + assert!(p.send(1, delete(string_row(101, "10"))).is_empty()); + p.send(1, update(string_row(102, "010"), string_row(102, "11"))); + p.expect(&[row(2, 11)]); +} + +#[test] +fn in_subquery_review_outer_and_inner_share_source() { + let mut p = + Pipeline::new("SELECT * FROM outer_rows WHERE value IN (SELECT value FROM outer_rows)"); + p.send( + 0, + Operation::BatchInsert { + new: vec![row(1, 10), row(1, 10), null_row(2)], + }, + ); + p.expect(&[row(1, 10), row(1, 10)]); + p.send(0, update(row(1, 10), row(1, 20))); + p.expect(&[row(1, 10), row(1, 20)]); + p.send(0, delete(row(1, 10))); + p.expect(&[row(1, 20)]); + p.send(0, update(null_row(2), row(2, 20))); + p.expect(&[row(1, 20), row(2, 20)]); + p.send(0, delete(row(1, 20))); + p.send(0, delete(row(2, 20))); + p.expect(&[]); +} + +#[test] +fn in_subquery_review_ttl_rejection_preserves_both_ports_state() { + use crate::product::join::mark::MarkJoinProcessorFactory; + use dozer_core::node::ProcessorFactory; + use dozer_sql_expression::sqlparser::ast::{Expr, Ident}; + use dozer_types::types::Lifetime; + + let runtime = create_test_runtime(); + let factory = MarkJoinProcessorFactory::new( + "mark".into(), + Expr::Identifier(Ident::new("value")), + "membership".into(), + false, + vec![], + runtime.clone(), + ); + let inputs = HashMap::from([ + (0, schema(FieldType::Int)), + ( + 1, + Schema { + fields: vec![FieldDefinition::new( + "value".into(), + FieldType::Int, + true, + Default::default(), + )], + primary_index: vec![], + }, + ), + ]); + let outer = row(1, 10); + let inner = Record::new(vec![Field::Int(10)]); + for port in [0, 1] { + let normal = if port == 0 { + outer.clone() + } else { + inner.clone() + }; + let mut expiring = normal.clone(); + expiring.lifetime = Some(Lifetime { + reference: "2024-01-01T00:00:00Z".parse().unwrap(), + duration: std::time::Duration::from_secs(10), + }); + for rejected in [ + insert(expiring.clone()), + delete(expiring.clone()), + update(expiring.clone(), normal.clone()), + update(normal.clone(), expiring.clone()), + Operation::BatchInsert { + new: vec![normal.clone(), expiring.clone()], + }, + ] { + let mut processor = runtime + .block_on(factory.build(inputs.clone(), HashMap::new(), EventHub::new(16))) + .unwrap(); + let mut output = Output::default(); + processor + .process( + TableOperation::without_id(insert(inner.clone()), 1), + &mut output, + ) + .unwrap(); + processor + .process( + TableOperation::without_id(insert(outer.clone()), 0), + &mut output, + ) + .unwrap(); + output.0.clear(); + let error = processor + .process(TableOperation::without_id(rejected, port), &mut output) + .unwrap_err(); + assert!( + error.to_string().contains("do not support TTL inputs"), + "{error}" + ); + assert!(output.0.is_empty()); + + processor + .process( + TableOperation::without_id(delete(inner.clone()), 1), + &mut output, + ) + .unwrap(); + let mut old = outer.clone(); + old.values.push(Field::Boolean(true)); + let mut new = outer.clone(); + new.values.push(Field::Boolean(false)); + assert_eq!( + output.0.iter().map(|op| op.op.clone()).collect::>(), + vec![update(old, new.clone())] + ); + output.0.clear(); + processor + .process( + TableOperation::without_id(delete(outer.clone()), 0), + &mut output, + ) + .unwrap(); + assert_eq!( + output.0.iter().map(|op| op.op.clone()).collect::>(), + vec![delete(new)] + ); + } + } +} + +#[test] +fn in_subquery_review_incompatible_comparison_rejects_without_state_change() { + use crate::product::join::mark::MarkJoinProcessorFactory; + use dozer_core::node::ProcessorFactory; + use dozer_sql_expression::sqlparser::ast::{Expr, Ident}; + + let runtime = create_test_runtime(); + let factory = MarkJoinProcessorFactory::new( + "mark".into(), + Expr::Identifier(Ident::new("value")), + "membership".into(), + true, + vec![], + runtime.clone(), + ); + let inputs = HashMap::from([ + (0, schema(FieldType::Boolean)), + ( + 1, + Schema { + fields: vec![FieldDefinition::new( + "value".into(), + FieldType::Binary, + true, + Default::default(), + )], + primary_index: vec![], + }, + ), + ]); + let outer = Record::new(vec![Field::Int(1), Field::Boolean(true)]); + let inner = Record::new(vec![Field::Binary(vec![1])]); + let mut included = outer.clone(); + included.values.push(Field::Boolean(true)); + for first_port in [0, 1] { + let mut processor = runtime + .block_on(factory.build(inputs.clone(), HashMap::new(), EventHub::new(16))) + .unwrap(); + let mut output = Output::default(); + let first = if first_port == 0 { + outer.clone() + } else { + inner.clone() + }; + let second = if first_port == 0 { + inner.clone() + } else { + outer.clone() + }; + processor + .process( + TableOperation::without_id(insert(first.clone()), first_port), + &mut output, + ) + .unwrap(); + output.0.clear(); + let error = processor + .process( + TableOperation::without_id(insert(second), 1 - first_port), + &mut output, + ) + .unwrap_err(); + assert!( + error.to_string().to_lowercase().contains("compar"), + "{error}" + ); + assert!(output.0.is_empty()); + processor + .process( + TableOperation::without_id(delete(first), first_port), + &mut output, + ) + .unwrap(); + let expected = if first_port == 0 { + vec![delete(included.clone())] + } else { + vec![] + }; + assert_eq!( + output.0.iter().map(|op| op.op.clone()).collect::>(), + expected + ); + output.0.clear(); + processor + .process( + TableOperation::without_id(insert(outer.clone()), 0), + &mut output, + ) + .unwrap(); + assert_eq!( + output.0.iter().map(|op| op.op.clone()).collect::>(), + vec![insert(included.clone())] + ); + } +} + +#[test] +fn in_subquery_review_logical_output_schema_remains_nullable() { + for expression in [ + "value AND (1 = 1)", + "value OR (1 = 2)", + "NOT (value AND (1 = 1))", + ] { + let sql = format!("SELECT {expression} AS result FROM outer_rows"); + let mut p = Pipeline::build(&sql, vec![schema(FieldType::Boolean)]).unwrap(); + assert!(p.schema.fields[0].nullable, "{expression} can produce NULL"); + p.send(0, insert(null_row(1))); + p.expect(&[Record::new(vec![Field::Null])]); + } +} + +#[test] +fn in_subquery_review_literal_null_membership_empty_and_nonempty() { + for negated in [false, true] { + let sql = format!( + "SELECT * FROM outer_rows WHERE NULL {}IN (SELECT value FROM inner_rows)", + if negated { "NOT " } else { "" } + ); + let mut p = Pipeline::new(&sql); + let expected_empty = if negated { vec![row(1, 10)] } else { vec![] }; + p.send(0, insert(row(1, 10))); + p.expect(&expected_empty); + p.send(1, insert(null_row(101))); + p.expect(&[]); + p.send(1, delete(null_row(101))); + p.expect(&expected_empty); + p.send(1, insert(row(102, 10))); + p.expect(&[]); + p.send(1, delete(row(102, 10))); + p.expect(&expected_empty); + } +} + +#[test] +fn in_subquery_review_parenthesized_union_branch() { + let mut p = Pipeline::new("SELECT * FROM outer_rows WHERE value IN (SELECT value FROM inner_rows UNION (SELECT value FROM third_rows))"); + p.send(0, insert(row(1, 10))); + p.send(1, insert(row(101, 10))); + p.expect(&[row(1, 10)]); + p.send(2, insert(row(201, 10))); + p.send(1, delete(row(101, 10))); + p.expect(&[row(1, 10)]); + p.send(2, delete(row(201, 10))); + p.expect(&[]); +} + +#[test] +fn in_subquery_review_parenthesized_cte_query_body() { + let mut p = Pipeline::new("SELECT * FROM outer_rows WHERE value IN (WITH allowed AS (SELECT value FROM inner_rows) (SELECT value FROM allowed))"); + p.send(0, insert(row(1, 10))); + p.send(1, insert(row(101, 10))); + p.expect(&[row(1, 10)]); + p.send(1, delete(row(101, 10))); + p.expect(&[]); +} + +#[test] +fn in_subquery_review_parenthesized_union_cte_does_not_leak_to_sibling() { + let mut p = Pipeline::new("SELECT * FROM outer_rows WHERE value IN (WITH allowed AS (SELECT value FROM third_rows) (WITH allowed AS (SELECT value FROM inner_rows) SELECT value FROM allowed) UNION SELECT value FROM allowed)"); + p.send( + 0, + Operation::BatchInsert { + new: vec![row(1, 10), row(2, 20)], + }, + ); + p.send(1, insert(row(101, 10))); + p.expect(&[row(1, 10)]); + p.send(2, insert(row(201, 20))); + p.expect(&[row(1, 10), row(2, 20)]); + p.send(1, delete(row(101, 10))); + p.expect(&[row(2, 20)]); + p.send(2, delete(row(201, 20))); + p.expect(&[]); +} + +#[test] +fn in_subquery_demo() { + let mut p = Pipeline::new(IN_SQL); + println!("SQL: {IN_SQL}"); + let step = |p: &mut Pipeline, label: &str, port, input: Operation, expected: &[Record]| { + println!("\n{label}\nSource {port}: {input:?}"); + let emitted = p.send(port, input); + println!("Emitted: {emitted:?}\nResult rows: {:?}", p.rows); + p.expect(expected); + }; + step( + &mut p, + "Outer batch waits for membership", + 0, + Operation::BatchInsert { + new: vec![row(1, 10), row(2, 20)], + }, + &[], + ); + step( + &mut p, + "Inner insert admits the matching outer row", + 1, + insert(row(101, 10)), + &[row(1, 10)], + ); + step( + &mut p, + "Inner duplicate does not multiply the result", + 1, + insert(row(102, 10)), + &[row(1, 10)], + ); + step( + &mut p, + "Removing one duplicate preserves membership", + 1, + delete(row(102, 10)), + &[row(1, 10)], + ); + step( + &mut p, + "Inner key update retracts old membership and admits the new one", + 1, + update(row(101, 10), row(101, 20)), + &[row(2, 20)], + ); + step( + &mut p, + "Last matching inner delete retracts the outer row", + 1, + delete(row(101, 20)), + &[], + ); +} diff --git a/dozer-sql/src/tests/mod.rs b/dozer-sql/src/tests/mod.rs index 0ed1750729..1585649c13 100644 --- a/dozer-sql/src/tests/mod.rs +++ b/dozer-sql/src/tests/mod.rs @@ -1,2 +1,3 @@ mod builder_test; +mod in_subquery; pub mod utils; diff --git a/dozer-tests/in_subquery/README.md b/dozer-tests/in_subquery/README.md new file mode 100644 index 0000000000..18725decea --- /dev/null +++ b/dozer-tests/in_subquery/README.md @@ -0,0 +1,24 @@ +# Streaming IN subqueries + +`WHERE value IN (SELECT value FROM allowed)` is a live membership test. Inner inserts can admit already-seen outer rows; deleting the final matching inner value retracts them. `NOT IN` uses SQL three-valued logic: a nonmatching value is unknown while the inner result contains NULL, and an empty inner result makes `NOT IN` true, including for a NULL outer value. + +```sql +SELECT * FROM orders +WHERE customer_id IN ( + SELECT customer_id FROM customers WHERE active = 1 +); +``` + +The planner builds the inner SELECT and a two-input mark join. Reference-counted inner values preserve outer multiplicity without multiplying rows by inner duplicates. Same-type membership updates use a hash index; comparisons between different field types use Dozer's SQL equality conversion rules over distinct outer values. Incomparable non-null operands are rejected explicitly. Updates and batches apply their complete inner delta before emitting membership changes. + +Marks remain internal to WHERE and are removed before SELECT projection or aggregation, including SELECT *. Multiple and nested subqueries, CTEs, scalar left operands, and compound AND/OR/NOT predicates are supported. Inner CTE names have their own scope. Correlated subqueries are not supported: a reference to an outer table in the inner SELECT is rejected instead of accidentally binding to a same-named inner column. The inner query must return exactly one column. Existing restrictions on ORDER BY and LIMIT also apply. + +TTL-bearing inputs are explicitly rejected before mutating either join input. Dozer currently expires join indexes without emitting delete events; supporting TTL membership requires explicit expiration propagation so downstream rows do not become stale. Ordinary insert/update/delete streams and non-TTL window results are supported. Like the existing join processors, state is reconstructed by replaying source events. + +Run the SQL-to-DAG regressions with: + +```sh +cargo test -p dozer-sql --no-default-features in_subquery +``` + +These exercise the actual parser, schema validation, generated graph, and processors with deterministic source-port ordering. They cover both arrival orders, duplicate and NULL values, atomic updates and batches, independent compound predicates, nested queries, CTE scope, cross-type equality, shared sources, TTL rejection, and 360 successive stream changes checked against a snapshot oracle.