diff --git a/src/binder/expr.rs b/src/binder/expr.rs index 9a9b6cfb..1febc2b4 100644 --- a/src/binder/expr.rs +++ b/src/binder/expr.rs @@ -134,25 +134,7 @@ impl<'a, T: Transaction, A: AsRef<[(&'static str, DataValue)]>> Binder<'a, '_, T &mut PlanArena<'arena>, ) -> Result, { - let BinderContext { - table_cache, - view_cache, - transaction, - scala_functions, - table_functions, - .. - } = &self.context; - let mut binder = Binder::new( - BinderContext::new( - table_cache, - view_cache, - *transaction, - scala_functions, - table_functions, - ), - self.args, - Some(&self.context), - ); + let mut binder = Binder::new(self.context.fork_empty(), self.args, Some(&self.context)); let sub_query = build(&mut binder, arena)?; let correlated = binder.context.has_outer_refs(); Ok((sub_query, correlated)) diff --git a/src/binder/mod.rs b/src/binder/mod.rs index 3901ba69..2a87623d 100644 --- a/src/binder/mod.rs +++ b/src/binder/mod.rs @@ -22,6 +22,7 @@ macro_rules! with_query_bind_step { }}; } +#[cfg(feature = "orm")] pub(crate) use with_query_bind_step; pub mod aggregate; @@ -65,7 +66,7 @@ use crate::errors::DatabaseError; use crate::expression::ScalarExpression; use crate::planner::operator::join::JoinType; use crate::planner::operator::mark_apply::MarkApplyQuantifier; -use crate::planner::{LogicalPlan, PlanArena}; +use crate::planner::{LogicalPlan, PlanArena, PlanRef}; use crate::storage::{TableCache, Transaction, ViewCache}; use crate::types::tuple::Schema; use crate::types::value::DataValue; @@ -134,6 +135,19 @@ pub struct BoundSource<'a> { pub(crate) source: Source<'a>, } +#[derive(Debug, Clone)] +pub(crate) struct CteBinding { + pub(crate) table_name: TableName, + pub(crate) depth: usize, + pub(crate) plan_ref: PlanRef, +} + +#[derive(Clone, Copy)] +pub(crate) struct CteCheckpoint { + len: usize, + depth: usize, +} + impl BoundSource<'_> { pub(crate) fn matches_name(&self, table_name: &str) -> bool { self.table_name.as_ref() == table_name @@ -231,6 +245,8 @@ pub struct BinderContext<'a, T: Transaction> { // Tips: retain binding order so wildcard expansion and position derivation // follow FROM/JOIN order directly. pub(crate) bind_table: Vec>, + ctes: Vec, + pub(crate) cte_depth: usize, // alias expr_aliases: BTreeMap<(Option, String), ScalarExpression>, table_aliases: HashMap, @@ -295,6 +311,8 @@ impl<'a, T: Transaction> BinderContext<'a, T> { view_cache, transaction, bind_table: Default::default(), + ctes: Default::default(), + cte_depth: 0, expr_aliases: Default::default(), table_aliases: Default::default(), group_by_exprs: vec![], @@ -319,6 +337,8 @@ impl<'a, T: Transaction> BinderContext<'a, T> { view_cache: self.view_cache, transaction: self.transaction, bind_table: self.bind_table.clone(), + ctes: self.ctes.clone(), + cte_depth: self.cte_depth, expr_aliases: self.expr_aliases.clone(), table_aliases: self.table_aliases.clone(), group_by_exprs: self.group_by_exprs.clone(), @@ -336,13 +356,52 @@ impl<'a, T: Transaction> BinderContext<'a, T> { /// This is used while binding an independent input, such as the right side /// of a join, before merging its newly bound sources into the parent scope. pub(crate) fn fork_empty(&self) -> Self { - BinderContext::new( + let mut context = BinderContext::new( self.table_cache, self.view_cache, self.transaction, self.scala_functions, self.table_functions, - ) + ); + context.ctes = self.ctes.clone(); + context.cte_depth = self.cte_depth; + context + } + + pub(crate) fn cte(&self, table_name: &TableName) -> Option<&CteBinding> { + self.ctes + .iter() + .rev() + .find(|cte| cte.table_name == *table_name) + } + + pub(crate) fn add_cte(&mut self, cte: CteBinding) -> Result<(), DatabaseError> { + if self + .ctes + .iter() + .rev() + .take_while(|binding| binding.depth == cte.depth) + .any(|binding| binding.table_name == cte.table_name) + { + return Err(DatabaseError::UnsupportedStmt(format!( + "duplicate CTE name: {}", + cte.table_name + ))); + } + self.ctes.push(cte); + Ok(()) + } + + pub(crate) fn cte_checkpoint(&self) -> CteCheckpoint { + CteCheckpoint { + len: self.ctes.len(), + depth: self.cte_depth, + } + } + + pub(crate) fn restore_ctes(&mut self, checkpoint: CteCheckpoint) { + self.ctes.truncate(checkpoint.len); + self.cte_depth = checkpoint.depth; } pub fn step(&mut self, bind_step: QueryBindStep) { diff --git a/src/binder/parser.rs b/src/binder/parser.rs index 91129eb2..42206ad0 100644 --- a/src/binder/parser.rs +++ b/src/binder/parser.rs @@ -16,7 +16,7 @@ use super::select::{ BindPlanAggregated, BindPlanComplete, BindPlanDistinct, BindPlanFiltered, BindPlanFrom, BindPlanProjected, BindPlanSelectList, BindPlanStart, JoinConstraintInput, TableAliasInput, }; -use super::{is_valid_identifier, with_query_bind_step, Binder, QueryBindStep, SetOperatorKind}; +use super::{is_valid_identifier, Binder, CteBinding, QueryBindStep, SetOperatorKind}; #[cfg(feature = "copy")] use crate::binder::copy::{ExtSource, FileFormat}; use crate::catalog::{ColumnCatalog, ColumnDesc, ColumnRef, TableName}; @@ -34,6 +34,7 @@ use crate::planner::operator::alter_table::change_column::{DefaultChange, NotNul use crate::planner::operator::join::{JoinCondition, JoinOperator as LJoinOperator, JoinType}; use crate::planner::operator::mark_apply::MarkApplyQuantifier; use crate::planner::operator::project::ProjectOperator; +use crate::planner::operator::recursive_cte::{RecursiveCteOperator, RecursiveScanOperator}; use crate::planner::operator::sort::SortField; use crate::planner::operator::Operator; use crate::planner::{Childrens, LogicalPlan, PlanArena}; @@ -2832,15 +2833,171 @@ impl<'a, 'parent, T: Transaction, A: AsRef<[(&'static str, DataValue)]>> Binder< Ok(select_items) } + fn bind_cte_plan( + &mut self, + cte: &sqlparser::ast::Cte, + alias: &TableAliasInput, + arena: &mut PlanArena, + ) -> Result { + let mut binder = Binder::new(self.context.fork(), self.args, Some(&self.context)); + let plan = binder.bind_query(&cte.query, arena)?; + let source_name = arena.temp_table(); + binder.bind_alias(plan, &alias.columns, alias.name.clone(), source_name, arena) + } + + fn add_cte_plan( + &mut self, + table_name: TableName, + plan: LogicalPlan, + arena: &mut PlanArena, + ) -> Result<(), DatabaseError> { + let plan_ref = arena.alloc_plan(plan); + self.context.add_cte(CteBinding { + table_name, + depth: self.context.cte_depth, + plan_ref, + }) + } + + fn bind_recursive_cte_plan( + &mut self, + cte: &sqlparser::ast::Cte, + alias: &TableAliasInput, + arena: &mut PlanArena, + ) -> Result { + if cte.query.with.is_some() + || cte.query.order_by.is_some() + || cte.query.limit_clause.is_some() + { + return Err(DatabaseError::UnsupportedStmt( + "recursive CTE query clauses are not supported".to_string(), + )); + } + + let SetExpr::SetOperation { + op: SetOperator::Union, + set_quantifier: SetQuantifier::All, + left, + right, + } = cte.query.body.as_ref() + else { + return Err(DatabaseError::UnsupportedStmt( + "recursive CTEs require a top-level UNION ALL".to_string(), + )); + }; + + let mut anchor_binder = Binder::new(self.context.fork(), self.args, Some(&self.context)); + let anchor = anchor_binder.bind_set_expr(left, arena)?; + if anchor + .referenced_table() + .iter() + .any(|table| table == &alias.name) + { + return Err(DatabaseError::UnsupportedStmt( + "the recursive CTE cannot be referenced by its anchor".to_string(), + )); + } + + let source_name = arena.temp_table(); + let mut anchor = anchor_binder.bind_alias( + anchor, + &alias.columns, + alias.name.clone(), + source_name, + arena, + )?; + let schema = anchor.output_schema(arena).clone(); + let scan = LogicalPlan::new( + Operator::RecursiveScan(RecursiveScanOperator { + schema_ref: schema.clone(), + }), + Childrens::None, + ); + + let cte_checkpoint = self.context.cte_checkpoint(); + self.add_cte_plan(alias.name.clone(), scan, arena)?; + let recursive = { + let mut binder = Binder::new(self.context.fork(), self.args, Some(&self.context)); + binder.bind_set_expr(right, arena) + }; + self.context.restore_ctes(cte_checkpoint); + let mut recursive = recursive?; + + fn recursive_scan_count(plan: &LogicalPlan) -> usize { + usize::from(matches!(&plan.operator, Operator::RecursiveScan(_))) + + plan + .childrens + .iter() + .map(recursive_scan_count) + .sum::() + } + + if recursive_scan_count(&recursive) != 1 { + return Err(DatabaseError::UnsupportedStmt( + "the recursive term must reference its CTE exactly once".to_string(), + )); + } + + let recursive_schema = recursive.output_schema(arena); + if schema.len() != recursive_schema.len() { + return Err(DatabaseError::MisMatch( + "the anchor column count", + "the recursive column count", + )); + } + if !schema + .iter() + .zip(recursive_schema) + .all(|(anchor, recursive)| { + arena.column(*anchor).datatype() == arena.column(*recursive).datatype() + }) + { + return Err(DatabaseError::UnsupportedStmt( + "recursive CTE column types must match the anchor".to_string(), + )); + } + + Ok(RecursiveCteOperator::build(schema, anchor, recursive)) + } + pub(crate) fn bind_query( &mut self, query: &Query, arena: &mut PlanArena, ) -> Result { let origin_step = self.context.step_now(); + let cte_checkpoint = self.context.cte_checkpoint(); + if let Some(with) = &query.with { + if with.recursive && with.cte_tables.len() != 1 { + return Err(DatabaseError::UnsupportedStmt( + "only one recursive CTE is supported".to_string(), + )); + } + + self.context.cte_depth += 1; + for cte in &with.cte_tables { + if cte.from.is_some() { + return Err(DatabaseError::UnsupportedStmt( + "CTE FROM clauses are not supported".to_string(), + )); + } + if matches!( + cte.materialized, + Some(sqlparser::ast::CteAsMaterialized::Materialized) + ) { + return Err(DatabaseError::UnsupportedStmt( + "materialized CTEs are not supported".to_string(), + )); + } - if let Some(_with) = &query.with { - // TODO support with clause. + let alias = sql_table_alias(cte.alias.clone()); + let plan = if with.recursive { + self.bind_recursive_cte_plan(cte, &alias, arena)? + } else { + self.bind_cte_plan(cte, &alias, arena)? + }; + self.add_cte_plan(alias.name, plan, arena)?; + } } let order_by_exprs = if let Some(order_by) = &query.order_by { @@ -2883,6 +3040,7 @@ impl<'a, 'parent, T: Transaction, A: AsRef<[(&'static str, DataValue)]>> Binder< plan = self.bind_limit(plan, limit_clause, arena)?; } + self.context.restore_ctes(cte_checkpoint); self.context.step(origin_step); Ok(plan) } @@ -3160,6 +3318,36 @@ mod tests { Ok(()) } + #[test] + fn test_bind_non_recursive_ctes() -> Result<(), DatabaseError> { + let tables = build_t1_table()?; + let mut arena = PlanArena::new(&tables.table_arena); + let mut plan = tables.plan_with_arena( + "with first(a) as (select c1 from t1), \ + second as (select a from first) select a from second", + &mut arena, + )?; + + let schema = plan.output_schema(&mut arena); + assert_eq!(schema.len(), 1); + assert_eq!(arena.column(schema[0]).name(), "a"); + + assert_unsupported( + tables + .plan("with recursive cte as (select 1) select * from cte") + .unwrap_err(), + "recursive CTEs", + ); + assert_unsupported( + tables + .plan("with cte as (select 1), cte as (select 2) select * from cte") + .unwrap_err(), + "duplicate CTE name", + ); + + Ok(()) + } + #[test] fn force_nest_loop_join_marks_join_operator() -> Result<(), DatabaseError> { let tables = build_t1_table()?; diff --git a/src/binder/select.rs b/src/binder/select.rs index 055625ba..1bad19eb 100644 --- a/src/binder/select.rs +++ b/src/binder/select.rs @@ -1192,6 +1192,27 @@ impl<'a: 'b, 'b, T: Transaction, A: AsRef<[(&'static str, DataValue)]>> Binder<' ) -> Result { let table_alias = alias.as_ref().map(|alias| alias.name.clone()); + if let Some(plan_ref) = self.context.cte(&table_name).map(|cte| cte.plan_ref) { + let mut plan = arena.plan(plan_ref).clone(); + if let Some(alias) = alias { + plan = self.bind_alias( + plan, + &alias.columns, + alias.name.clone(), + table_name.clone(), + arena, + )?; + } + let output_schema = plan.output_schema(arena).clone(); + self.context.add_bound_source( + table_name, + table_alias, + join_type, + Source::Schema(output_schema), + ); + return Ok(plan); + } + let with_pk = self.is_scan_with_pk(&table_name); let source = self .context diff --git a/src/execution/dql/mod.rs b/src/execution/dql/mod.rs index 9291ec57..03e8ac43 100644 --- a/src/execution/dql/mod.rs +++ b/src/execution/dql/mod.rs @@ -25,6 +25,7 @@ pub(crate) mod join; pub(crate) mod limit; pub(crate) mod mark_apply; pub(crate) mod projection; +pub(crate) mod recursive_cte; pub(crate) mod scalar_apply; pub(crate) mod scalar_subquery; pub(crate) mod seq_scan; diff --git a/src/execution/dql/recursive_cte.rs b/src/execution/dql/recursive_cte.rs new file mode 100644 index 00000000..5e5751b4 --- /dev/null +++ b/src/execution/dql/recursive_cte.rs @@ -0,0 +1,471 @@ +// Copyright 2024 KipData/KiteSQL +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use crate::errors::DatabaseError; +#[cfg(feature = "spill")] +use crate::execution::spill::{SpillReader, SpillVec}; +use crate::execution::{ + build_read, ExecArena, ExecId, ExecNode, ExecutionContext, ExecutorNode, ReadExecutor, +}; +use crate::planner::operator::recursive_cte::RecursiveScanOperator; +use crate::planner::{LogicalPlan, PlanArena}; +use crate::storage::Transaction; +use crate::types::tuple::Tuple; +use std::mem; + +pub(crate) enum RecursiveInput { + One(Option), + #[cfg(feature = "spill")] + Many(SpillReader), + #[cfg(not(feature = "spill"))] + Many(std::vec::IntoIter), +} + +impl Iterator for RecursiveInput { + type Item = Result; + + fn next(&mut self) -> Option { + match self { + Self::One(tuple) => tuple.take().map(Ok), + #[cfg(feature = "spill")] + Self::Many(rows) => rows.next(), + #[cfg(not(feature = "spill"))] + Self::Many(rows) => rows.next().map(Ok), + } + } +} + +#[derive(Default)] +enum RecursiveRows { + #[default] + Empty, + One { + tuple: Tuple, + output_done: bool, + }, + #[cfg(feature = "spill")] + Writing(SpillVec<'static, Tuple>), + #[cfg(feature = "spill")] + Reading(SpillReader), + #[cfg(not(feature = "spill"))] + Writing(Vec), + #[cfg(not(feature = "spill"))] + Reading { + rows: Vec, + output_index: usize, + }, +} + +impl RecursiveRows { + fn push(&mut self, tuple: Tuple) -> Result<(), DatabaseError> { + match self { + Self::Empty => { + *self = Self::One { + tuple, + output_done: false, + }; + } + Self::One { + tuple: first, + output_done: false, + } => { + let first = mem::take(first); + #[cfg(feature = "spill")] + { + let mut rows = SpillVec::new(); + let _ = rows.push(first)?; + let _ = rows.push(tuple)?; + *self = Self::Writing(rows); + } + #[cfg(not(feature = "spill"))] + { + *self = Self::Writing(vec![first, tuple]); + } + } + #[cfg(feature = "spill")] + Self::Writing(rows) => { + let _ = rows.push(tuple)?; + } + #[cfg(not(feature = "spill"))] + Self::Writing(rows) => rows.push(tuple), + #[cfg(feature = "spill")] + Self::One { .. } | Self::Reading(_) => { + unreachable!("cannot append to a finished recursive generation") + } + #[cfg(not(feature = "spill"))] + Self::One { .. } | Self::Reading { .. } => { + unreachable!("cannot append to a finished recursive generation") + } + } + Ok(()) + } + + fn finish(self) -> Result { + match self { + #[cfg(feature = "spill")] + Self::Writing(mut rows) => { + let _ = rows.flush()?; + Ok(Self::Reading(rows.into_iter())) + } + #[cfg(not(feature = "spill"))] + Self::Writing(rows) => Ok(Self::Reading { + rows, + output_index: 0, + }), + rows => Ok(rows), + } + } + + fn next_output(&mut self) -> Result, DatabaseError> { + match self { + Self::Empty => Ok(None), + Self::One { tuple, output_done } => { + if *output_done { + return Ok(None); + } + *output_done = true; + Ok(Some(tuple.clone())) + } + #[cfg(feature = "spill")] + Self::Reading(reader) => reader.next().transpose(), + #[cfg(not(feature = "spill"))] + Self::Reading { rows, output_index } => { + let tuple = rows.get(*output_index).cloned(); + *output_index += usize::from(tuple.is_some()); + Ok(tuple) + } + Self::Writing(_) => unreachable!("recursive generation must be finished first"), + } + } + + fn into_input(self) -> Result, DatabaseError> { + match self { + Self::Empty => Ok(None), + Self::One { tuple, .. } => Ok(Some(RecursiveInput::One(Some(tuple)))), + #[cfg(feature = "spill")] + Self::Reading(mut reader) => { + reader.reset()?; + Ok(Some(RecursiveInput::Many(reader))) + } + #[cfg(not(feature = "spill"))] + Self::Reading { rows, .. } => Ok(Some(RecursiveInput::Many(rows.into_iter()))), + Self::Writing(_) => unreachable!("recursive generation must be finished first"), + } + } +} + +enum RecursivePhase { + Anchor, + Output, + Recursive, +} + +pub struct RecursiveCte<'a, T: Transaction + 'a> { + recursive_plan: LogicalPlan, + anchor_input: ExecId, + recursive_arena: ExecArena<'a, T>, + recursive_root: ExecId, + working: RecursiveRows, + next: RecursiveRows, + phase: RecursivePhase, +} + +impl<'a, T: Transaction + 'a> RecursiveCte<'a, T> { + fn new( + anchor_input: ExecId, + recursive_plan: LogicalPlan, + recursive_arena: ExecArena<'a, T>, + ) -> Self { + Self { + recursive_plan, + anchor_input, + recursive_arena, + recursive_root: 0, + working: RecursiveRows::default(), + next: RecursiveRows::default(), + phase: RecursivePhase::Anchor, + } + } + + fn start_recursive(&mut self, plan_arena: &mut PlanArena<'a>) -> Result { + let Some(input) = mem::take(&mut self.working).into_input()? else { + return Ok(false); + }; + + self.recursive_arena.reset_for_rebuild(); + self.recursive_arena.set_recursive_input(input); + let cache = self.recursive_arena.context(); + let transaction = self.recursive_arena.transaction(); + self.recursive_root = build_read( + &mut self.recursive_arena, + plan_arena, + self.recursive_plan.clone(), + cache, + transaction, + ); + Ok(true) + } +} + +impl<'a, T: Transaction + 'a> ReadExecutor<'a, T> for RecursiveCte<'a, T> { + type Input = (LogicalPlan, LogicalPlan); + + fn into_executor( + (anchor_plan, recursive_plan): Self::Input, + arena: &mut ExecArena<'a, T>, + plan_arena: &mut PlanArena<'a>, + cache: ExecutionContext<'_>, + transaction: &T, + ) -> ExecId { + let mut recursive_arena = ExecArena::new(); + recursive_arena.init_context(arena.context(), arena.transaction()); + let anchor_input = build_read(arena, plan_arena, anchor_plan, cache, transaction); + arena.push(ExecNode::RecursiveCte(Self::new( + anchor_input, + recursive_plan, + recursive_arena, + ))) + } +} + +impl<'a, T: Transaction + 'a> ExecutorNode<'a, T> for RecursiveCte<'a, T> { + fn next_tuple( + &mut self, + arena: &mut ExecArena<'a, T>, + plan_arena: &mut PlanArena<'a>, + ) -> Result<(), DatabaseError> { + loop { + match self.phase { + RecursivePhase::Anchor => { + while arena.next_tuple(self.anchor_input, plan_arena)? { + self.next.push(mem::take(arena.result_tuple_mut()))?; + } + self.working = mem::take(&mut self.next).finish()?; + self.phase = RecursivePhase::Output; + } + RecursivePhase::Output => { + if let Some(tuple) = self.working.next_output()? { + arena.produce_tuple(tuple); + return Ok(()); + } + if !self.start_recursive(plan_arena)? { + arena.finish(); + return Ok(()); + } + self.phase = RecursivePhase::Recursive; + } + RecursivePhase::Recursive => { + while self + .recursive_arena + .next_tuple(self.recursive_root, plan_arena)? + { + self.next + .push(mem::take(self.recursive_arena.result_tuple_mut()))?; + } + self.recursive_arena.reset_for_rebuild(); + self.working = mem::take(&mut self.next).finish()?; + self.phase = RecursivePhase::Output; + } + } + } + } +} + +pub struct RecursiveScan { + input: RecursiveInput, +} + +impl<'a, T: Transaction + 'a> ReadExecutor<'a, T> for RecursiveScan { + type Input = RecursiveScanOperator; + + fn into_executor( + _input: Self::Input, + arena: &mut ExecArena<'a, T>, + _plan_arena: &mut PlanArena<'a>, + _cache: ExecutionContext<'_>, + _transaction: &T, + ) -> ExecId { + let input = arena.take_recursive_input(); + arena.push(ExecNode::RecursiveScan(Self { input })) + } +} + +impl<'a, T: Transaction + 'a> ExecutorNode<'a, T> for RecursiveScan { + fn next_tuple( + &mut self, + arena: &mut ExecArena<'a, T>, + _plan_arena: &mut PlanArena<'a>, + ) -> Result<(), DatabaseError> { + match self.input.next().transpose()? { + Some(tuple) => arena.produce_tuple(tuple), + None => arena.finish(), + } + Ok(()) + } +} + +#[cfg(all(test, not(target_arch = "wasm32")))] +mod tests { + use super::*; + use crate::catalog::{ColumnCatalog, ColumnDesc}; + use crate::execution::{empty_context, execute_input, try_collect}; + use crate::expression::{BinaryOperator, ScalarExpression}; + use crate::planner::operator::filter::FilterOperator; + use crate::planner::operator::project::ProjectOperator; + use crate::planner::operator::recursive_cte::RecursiveScanOperator; + use crate::planner::operator::values::ValuesOperator; + use crate::planner::operator::Operator; + use crate::planner::Childrens; + use crate::storage::rocksdb::RocksStorage; + use crate::storage::{StatisticsMetaCache, Storage, TableCache, ViewCache}; + use crate::types::evaluator::binary_create; + use crate::types::value::DataValue; + use crate::types::LogicalType; + use std::borrow::Cow; + use tempfile::TempDir; + + #[cfg(feature = "spill")] + #[test] + fn spilled_generation_is_written_once_and_replayed_for_scan() -> Result<(), DatabaseError> { + let expected = (0..1100) + .map(|value| Tuple::new(None, vec![DataValue::Int32(value)])) + .collect::>(); + let mut rows = RecursiveRows::default(); + for tuple in expected.iter().cloned() { + rows.push(tuple)?; + } + + let mut working = rows.finish()?; + assert!(matches!(&working, RecursiveRows::Reading(_))); + let mut output = Vec::new(); + while let Some(tuple) = working.next_output()? { + output.push(tuple); + } + assert_eq!(output, expected); + + let scan = working.into_input()?.unwrap(); + assert_eq!(scan.collect::, _>>()?, expected); + Ok(()) + } + + #[cfg(not(feature = "spill"))] + #[test] + fn memory_generation_is_replayed_for_scan() -> Result<(), DatabaseError> { + let expected = (0..3) + .map(|value| Tuple::new(None, vec![DataValue::Int32(value)])) + .collect::>(); + let mut rows = RecursiveRows::default(); + for tuple in expected.iter().cloned() { + rows.push(tuple)?; + } + + let mut working = rows.finish()?; + assert!(matches!(&working, RecursiveRows::Reading { .. })); + let mut output = Vec::new(); + while let Some(tuple) = working.next_output()? { + output.push(tuple); + } + assert_eq!(output, expected); + + let scan = working.into_input()?.unwrap(); + assert_eq!(scan.collect::, _>>()?, expected); + Ok(()) + } + + #[test] + fn empty_generation_has_no_recursive_input() -> Result<(), DatabaseError> { + let rows = RecursiveRows::default(); + let working = rows.finish()?; + assert!(working.into_input()?.is_none()); + Ok(()) + } + + #[test] + fn recursive_cte_replays_each_generation() -> Result<(), DatabaseError> { + let table_arena = crate::planner::TableArenaCell::default(); + let mut plan_arena = PlanArena::new(&table_arena); + let column = plan_arena.alloc_column(ColumnCatalog::new( + "value".to_string(), + true, + ColumnDesc::new(LogicalType::Integer, None, false, None).unwrap(), + )); + let schema_ref = vec![column]; + let anchor = LogicalPlan::new( + Operator::Values(ValuesOperator { + rows: vec![vec![DataValue::Int32(1)]], + schema_ref: schema_ref.clone(), + }), + Childrens::None, + ); + let scan = LogicalPlan::new( + Operator::RecursiveScan(RecursiveScanOperator { schema_ref }), + Childrens::None, + ); + let filter = FilterOperator::build( + ScalarExpression::Binary { + op: BinaryOperator::Lt, + left_expr: Box::new(ScalarExpression::column_expr(column, 0)), + right_expr: Box::new(DataValue::Int32(3).into()), + evaluator: Some(binary_create( + Cow::Owned(LogicalType::Integer), + BinaryOperator::Lt, + )?), + ty: LogicalType::Boolean, + }, + scan, + false, + ); + let recursive = LogicalPlan::new( + Operator::Project(ProjectOperator { + exprs: vec![ScalarExpression::Binary { + op: BinaryOperator::Plus, + left_expr: Box::new(ScalarExpression::column_expr(column, 0)), + right_expr: Box::new(DataValue::Int32(1).into()), + evaluator: Some(binary_create( + Cow::Owned(LogicalType::Integer), + BinaryOperator::Plus, + )?), + ty: LogicalType::Integer, + }], + }), + Childrens::Only(Box::new(filter)), + ); + + let temp_dir = TempDir::new().expect("unable to create temporary working directory"); + let storage = RocksStorage::new(temp_dir.path())?; + let transaction = storage.transaction()?; + let table_cache = TableCache::default(); + let view_cache = ViewCache::default(); + let meta_cache = StatisticsMetaCache::default(); + let tuples = try_collect(execute_input::<_, RecursiveCte<'_, _>>( + (anchor, recursive), + empty_context(&table_cache, &view_cache, &meta_cache), + plan_arena, + &transaction, + ))?; + + assert_eq!( + tuples + .into_iter() + .flat_map(|tuple| tuple.values) + .collect::>(), + vec![ + DataValue::Int32(1), + DataValue::Int32(2), + DataValue::Int32(3), + ] + ); + Ok(()) + } +} diff --git a/src/execution/mod.rs b/src/execution/mod.rs index 1e046616..71ae7bff 100644 --- a/src/execution/mod.rs +++ b/src/execution/mod.rs @@ -59,6 +59,7 @@ use crate::execution::dql::index_scan::IndexScan; use crate::execution::dql::join::hash_join::HashJoin; use crate::execution::dql::limit::Limit; use crate::execution::dql::projection::Projection; +use crate::execution::dql::recursive_cte::{RecursiveCte, RecursiveInput, RecursiveScan}; use crate::execution::dql::scalar_subquery::ScalarSubquery; use crate::execution::dql::seq_scan::SeqScan; use crate::execution::dql::set_membership::SetMembership; @@ -197,6 +198,8 @@ pub(crate) enum ExecNode<'a, T: Transaction + 'a> { MarkApply(MarkApply), NestedLoopJoin(NestedLoopJoin), Projection(Projection), + RecursiveCte(RecursiveCte<'a, T>), + RecursiveScan(RecursiveScan), ScalarApply(ScalarApply), ScalarSubquery(ScalarSubquery), SetMembership(SetMembership), @@ -315,6 +318,12 @@ impl<'a, T: Transaction + 'a> ExecNode<'a, T> { ExecNode::Projection(exec) => { >::next_tuple(exec, arena, plan_arena) } + ExecNode::RecursiveCte(exec) => { + as ExecutorNode<'a, T>>::next_tuple(exec, arena, plan_arena) + } + ExecNode::RecursiveScan(exec) => { + >::next_tuple(exec, arena, plan_arena) + } ExecNode::ScalarApply(exec) => { >::next_tuple(exec, arena, plan_arena) } @@ -377,6 +386,7 @@ pub(crate) struct ExecArena<'a, T: Transaction + 'a> { transaction: *mut T, runtime_probe_stack: Vec, ddl_apply: Vec, + recursive_input: Option, } pub(crate) struct ExecArenaLocalState<'b, 'a, T: Transaction + 'a> { @@ -425,6 +435,7 @@ impl<'a, T: Transaction + 'a> ExecArena<'a, T> { transaction: std::ptr::null_mut(), runtime_probe_stack: Vec::new(), ddl_apply: Vec::new(), + recursive_input: None, } } } @@ -538,6 +549,25 @@ impl<'a, T: Transaction + 'a> ExecArena<'a, T> { self.runtime_probe_stack.len() } + pub(crate) fn set_recursive_input(&mut self, input: RecursiveInput) { + debug_assert!(self.recursive_input.is_none()); + self.recursive_input = Some(input); + } + + pub(crate) fn take_recursive_input(&mut self) -> RecursiveInput { + self.recursive_input + .take() + .expect("recursive input initialized") + } + + pub(crate) fn reset_for_rebuild(&mut self) { + debug_assert!(self.runtime_probe_stack.is_empty()); + debug_assert!(self.ddl_apply.is_empty()); + self.nodes.clear(); + self.result.status = None; + self.recursive_input = None; + } + #[inline] pub(crate) fn result_tuple(&self) -> &Tuple { &self.result.tuple @@ -891,6 +921,20 @@ where cache, transaction, ), + Operator::RecursiveCte(_) => as ReadExecutor<'a, T>>::into_executor( + childrens.pop_twins(), + arena, + plan_arena, + cache, + transaction, + ), + Operator::RecursiveScan(op) => >::into_executor( + op, + arena, + plan_arena, + cache, + transaction, + ), Operator::SetMembership(op) => { let (left, right) = childrens.pop_twins(); >::into_executor( diff --git a/src/execution/spill/mod.rs b/src/execution/spill/mod.rs index e5eb82ad..b79870e3 100644 --- a/src/execution/spill/mod.rs +++ b/src/execution/spill/mod.rs @@ -174,10 +174,7 @@ impl Iterator for SpillReader { }; match result { Ok(Some(value)) => Some(Ok(value)), - Ok(None) => { - self.state = ReadState::Exhausted(None); - None - } + Ok(None) => None, Err(error) => { self.state = ReadState::Exhausted(None); Some(Err(error)) @@ -187,6 +184,21 @@ impl Iterator for SpillReader { } impl SpillReader { + /// Rewinds a fully flushed spill file. The reader must not contain an in-memory tail. + pub(crate) fn reset(&mut self) -> Result<(), DatabaseError> { + let ReadState::Spilled { reader, tail, .. } = &mut self.state else { + return Err(DatabaseError::InvalidValue( + "cannot reset an in-memory SpillReader".to_string(), + )); + }; + if tail.len() != 0 { + return Err(DatabaseError::InvalidValue( + "cannot reset a SpillReader with an in-memory tail".to_string(), + )); + } + reader.reset(0) + } + pub(crate) fn open_segment_reader<'source>( &'source self, ) -> Result, T>, DatabaseError> { @@ -439,6 +451,25 @@ mod tests { Ok(()) } + #[test] + fn reset_replays_a_fully_flushed_reader_after_eof() -> Result<(), DatabaseError> { + let mut values = SpillVec::new().limit(2, usize::MAX); + for value in 0..5 { + let _ = values.push(row(value))?; + } + let _ = values.flush()?; + + let mut reader = values.into_iter(); + let first = reader.by_ref().collect::, _>>()?; + assert_eq!(reader.next().transpose()?, None); + reader.reset()?; + let second = reader.collect::, _>>()?; + let expected = (0..5).map(row).collect::>(); + assert_eq!(first, expected); + assert_eq!(second, expected); + Ok(()) + } + #[test] fn spill_reader_stays_exhausted() -> Result<(), DatabaseError> { let mut reader = SpillVec::from(vec![row(1)]).into_iter(); diff --git a/src/optimizer/rule/implementation/mod.rs b/src/optimizer/rule/implementation/mod.rs index f2fcb9ff..97855b82 100644 --- a/src/optimizer/rule/implementation/mod.rs +++ b/src/optimizer/rule/implementation/mod.rs @@ -134,6 +134,7 @@ impl ImplementationRuleRootTag { | Operator::CreateView(_) | Operator::DropView(_) | Operator::DropIndex(_) => None, + Operator::RecursiveCte(_) | Operator::RecursiveScan(_) => None, } } } diff --git a/src/optimizer/rule/normalization/column_pruning.rs b/src/optimizer/rule/normalization/column_pruning.rs index 41882c92..3c6a8e24 100644 --- a/src/optimizer/rule/normalization/column_pruning.rs +++ b/src/optimizer/rule/normalization/column_pruning.rs @@ -662,10 +662,24 @@ impl ColumnPruning { } } } + Operator::RecursiveCte(_) => { + changed |= Self::apply_twins( + required_columns, + true, + childrens, + outcome, + output_start, + arena, + )?; + outcome.removed_positions.truncate(output_start); + } // Last Operator Operator::Dummy | Operator::Values(_) | Operator::FunctionScan(_) => { outcome.removed_positions.truncate(output_start); } + Operator::RecursiveScan(_) => { + outcome.removed_positions.truncate(output_start); + } Operator::Explain => { let child_start = outcome.removed_positions.len(); let child_changed = Self::apply_only_child( diff --git a/src/optimizer/rule/normalization/compilation_in_advance.rs b/src/optimizer/rule/normalization/compilation_in_advance.rs index 6b092fbc..d0aa1bd0 100644 --- a/src/optimizer/rule/normalization/compilation_in_advance.rs +++ b/src/optimizer/rule/normalization/compilation_in_advance.rs @@ -36,14 +36,16 @@ impl EvaluatorBind { Childrens::Only(child) => Self::_apply(child, arena)?, Childrens::Twins { left, right } => { Self::_apply(left, arena)?; - if matches!( + let bind_right = matches!( plan.operator, Operator::ScalarApply(_) | Operator::MarkApply(_) | Operator::Join(_) | Operator::Union(_) | Operator::SetMembership(_) - ) { + ); + let bind_right = bind_right || matches!(plan.operator, Operator::RecursiveCte(_)); + if bind_right { Self::_apply(right, arena)?; } } diff --git a/src/optimizer/rule/normalization/mod.rs b/src/optimizer/rule/normalization/mod.rs index f8b41f1f..08e8e820 100644 --- a/src/optimizer/rule/normalization/mod.rs +++ b/src/optimizer/rule/normalization/mod.rs @@ -135,6 +135,7 @@ impl NormalizationRuleRootTag { | Operator::Union(_) | Operator::SetMembership(_) | Operator::Window(_) => None, + Operator::RecursiveCte(_) | Operator::RecursiveScan(_) => None, #[cfg(feature = "copy")] Operator::CopyFromFile(_) | Operator::CopyToFile(_) => None, } diff --git a/src/planner/arena.rs b/src/planner/arena.rs index 5a1749db..c041281a 100644 --- a/src/planner/arena.rs +++ b/src/planner/arena.rs @@ -13,6 +13,7 @@ // limitations under the License. use crate::catalog::{ColumnCatalog, ColumnRef, TableName}; +use crate::planner::LogicalPlan; use crate::types::index::{IndexMeta, IndexMetaRef}; use crate::types::tuple::Schema; use std::cell::UnsafeCell; @@ -54,6 +55,12 @@ pub struct PlanArena<'a> { temp_table_id: usize, columns: Vec, indexes: Vec, + plans: Vec, +} + +#[derive(Debug, Clone, Copy, Hash, Eq, PartialEq)] +pub(crate) struct PlanRef { + pos: usize, } pub trait MetaArena { @@ -340,6 +347,7 @@ impl<'a> PlanArena<'a> { temp_table_id: 0, columns: Vec::new(), indexes: Vec::new(), + plans: Vec::new(), } } @@ -434,6 +442,18 @@ impl<'a> PlanArena<'a> { table_name.into() } + pub(crate) fn alloc_plan(&mut self, plan: LogicalPlan) -> PlanRef { + let plan_ref = PlanRef { + pos: self.plans.len(), + }; + self.plans.push(plan); + plan_ref + } + + pub(crate) fn plan(&self, plan_ref: PlanRef) -> &LogicalPlan { + &self.plans[plan_ref.pos] + } + pub fn column(&self, column: ColumnRef) -> &ColumnCatalog { ::column(self, column) } diff --git a/src/planner/mod.rs b/src/planner/mod.rs index 99e20338..99f811d4 100644 --- a/src/planner/mod.rs +++ b/src/planner/mod.rs @@ -17,6 +17,7 @@ pub mod operator; use crate::catalog::TableName; use crate::errors::DatabaseError; +use crate::planner::operator::recursive_cte::{RecursiveCteOperator, RecursiveScanOperator}; use crate::planner::operator::set_membership::SetMembershipOperator; use crate::planner::operator::union::UnionOperator; use crate::planner::operator::values::ValuesOperator; @@ -24,6 +25,7 @@ use crate::planner::operator::{Operator, PhysicalOption}; use kite_sql_serde_macros::ReferenceSerialization; use std::hash::{Hash, Hasher}; +pub(crate) use arena::PlanRef; pub use arena::{MetaArena, PlanArena, TableArena, TableArenaCell}; #[derive(Debug, PartialEq, Eq, Clone, Hash, ReferenceSerialization)] @@ -239,6 +241,8 @@ impl LogicalPlan { left_schema_ref: schema_ref, .. }) => schema_ref.clone(), + Operator::RecursiveCte(RecursiveCteOperator { schema_ref }) + | Operator::RecursiveScan(RecursiveScanOperator { schema_ref }) => schema_ref.clone(), Operator::Dummy => Vec::new(), Operator::ShowTable => Self::dummy_schema(arena, ["TABLE"]), Operator::ShowView => Self::dummy_schema(arena, ["VIEW"]), diff --git a/src/planner/operator/mod.rs b/src/planner/operator/mod.rs index 13e9473e..0cdaca43 100644 --- a/src/planner/operator/mod.rs +++ b/src/planner/operator/mod.rs @@ -34,6 +34,7 @@ pub mod join; pub mod limit; pub mod mark_apply; pub mod project; +pub mod recursive_cte; pub mod scalar_apply; pub mod scalar_subquery; pub mod set_membership; @@ -48,6 +49,7 @@ pub mod visitor; pub mod visitor_mut; pub mod window; +use self::recursive_cte::{RecursiveCteOperator, RecursiveScanOperator}; use self::{ aggregate::AggregateOperator, alter_table::add_column::AddColumnOperator, alter_table::change_column::ChangeColumnOperator, filter::FilterOperator, join::JoinOperator, @@ -114,6 +116,8 @@ pub enum Operator { Describe(DescribeOperator), SetMembership(SetMembershipOperator), Union(UnionOperator), + RecursiveCte(RecursiveCteOperator), + RecursiveScan(RecursiveScanOperator), // DML Insert(InsertOperator), Update(UpdateOperator), @@ -354,6 +358,26 @@ impl Operator { Ok(()) } + fn visit_recursive_cte( + &mut self, + op: &'operator RecursiveCteOperator, + ) -> Result<(), DatabaseError> { + for column in &op.schema_ref { + self.visit_column_ref(column)?; + } + Ok(()) + } + + fn visit_recursive_scan( + &mut self, + op: &'operator RecursiveScanOperator, + ) -> Result<(), DatabaseError> { + for column in &op.schema_ref { + self.visit_column_ref(column)?; + } + Ok(()) + } + fn visit_set_membership( &mut self, op: &'operator SetMembershipOperator, @@ -487,6 +511,8 @@ impl fmt::Display for Operator { #[cfg(feature = "copy")] Operator::CopyToFile(op) => write!(f, "{op}"), Operator::Union(op) => write!(f, "{op}"), + Operator::RecursiveCte(op) => write!(f, "{op}"), + Operator::RecursiveScan(op) => write!(f, "{op}"), Operator::SetMembership(op) => write!(f, "{op}"), Operator::Window(op) => write!(f, "{op}"), } @@ -887,6 +913,34 @@ mod tests { Ok(()) } + #[test] + fn recursive_operators_visit_display_and_build() -> Result<(), DatabaseError> { + let table_arena = TableArenaCell::default(); + let mut arena = PlanArena::new(&table_arena); + let value = column("value", &mut arena); + let depth = column("depth", &mut arena); + let schema = vec![value, depth]; + + let recursive_cte = Operator::RecursiveCte(RecursiveCteOperator { + schema_ref: schema.clone(), + }); + assert_eq!(referenced_columns(&recursive_cte, &mut arena)?, schema); + assert_eq!(recursive_cte.to_string(), "Recursive CTE: [#0, #1]"); + + let recursive_scan = Operator::RecursiveScan(RecursiveScanOperator { + schema_ref: schema.clone(), + }); + assert_eq!(referenced_columns(&recursive_scan, &mut arena)?, schema); + assert_eq!(recursive_scan.to_string(), "Recursive Scan: [#0, #1]"); + + let anchor = LogicalPlan::new(Operator::ShowTable, Childrens::None); + let recursive = LogicalPlan::new(Operator::ShowView, Childrens::None); + let plan = RecursiveCteOperator::build(schema, anchor, recursive); + assert!(matches!(plan.operator, Operator::RecursiveCte(_))); + assert!(matches!(*plan.childrens, Childrens::Twins { .. })); + Ok(()) + } + #[test] fn mark_apply_constructors_and_accessors_cover_quantified_paths() { let left = LogicalPlan::new(Operator::ShowTable, Childrens::None); diff --git a/src/planner/operator/recursive_cte.rs b/src/planner/operator/recursive_cte.rs new file mode 100644 index 00000000..fa0959e5 --- /dev/null +++ b/src/planner/operator/recursive_cte.rs @@ -0,0 +1,54 @@ +// Copyright 2024 KipData/KiteSQL +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use crate::iter_ext::Itertools; +use crate::planner::operator::Operator; +use crate::planner::{Childrens, LogicalPlan}; +use crate::types::tuple::Schema; +use kite_sql_serde_macros::ReferenceSerialization; +use std::fmt; + +#[derive(Debug, PartialEq, Eq, Clone, Hash, ReferenceSerialization)] +pub struct RecursiveCteOperator { + pub schema_ref: Schema, +} + +impl RecursiveCteOperator { + pub fn build(schema_ref: Schema, anchor: LogicalPlan, recursive: LogicalPlan) -> LogicalPlan { + LogicalPlan::new( + Operator::RecursiveCte(Self { schema_ref }), + Childrens::Twins { + left: Box::new(anchor), + right: Box::new(recursive), + }, + ) + } +} + +impl fmt::Display for RecursiveCteOperator { + fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { + write!(f, "Recursive CTE: [{}]", self.schema_ref.iter().join(", ")) + } +} + +#[derive(Debug, PartialEq, Eq, Clone, Hash, ReferenceSerialization)] +pub struct RecursiveScanOperator { + pub schema_ref: Schema, +} + +impl fmt::Display for RecursiveScanOperator { + fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { + write!(f, "Recursive Scan: [{}]", self.schema_ref.iter().join(", ")) + } +} diff --git a/src/planner/operator/visitor.rs b/src/planner/operator/visitor.rs index 09b1cabe..a1da1aa9 100644 --- a/src/planner/operator/visitor.rs +++ b/src/planner/operator/visitor.rs @@ -112,6 +112,17 @@ pub trait OperatorVisitor<'a>: Sized { Ok(()) } + fn visit_recursive_cte(&mut self, _op: &'a RecursiveCteOperator) -> Result<(), DatabaseError> { + Ok(()) + } + + fn visit_recursive_scan( + &mut self, + _op: &'a RecursiveScanOperator, + ) -> Result<(), DatabaseError> { + Ok(()) + } + fn visit_insert(&mut self, _op: &'a InsertOperator) -> Result<(), DatabaseError> { Ok(()) } @@ -334,6 +345,8 @@ pub fn walk_operator<'a, V: OperatorVisitor<'a>>( Operator::Describe(op) => visitor.visit_describe(op), Operator::SetMembership(op) => visitor.visit_set_membership(op), Operator::Union(op) => visitor.visit_union(op), + Operator::RecursiveCte(op) => visitor.visit_recursive_cte(op), + Operator::RecursiveScan(op) => visitor.visit_recursive_scan(op), Operator::Insert(op) => visitor.visit_insert(op), Operator::Update(op) => visitor.visit_update(op), Operator::Delete(op) => visitor.visit_delete(op), @@ -563,6 +576,16 @@ pub(crate) mod tests { table_name: "t1".into(), }), ]; + let operators = { + let mut with_recursive_operators = operators; + with_recursive_operators.push(Operator::RecursiveCte(RecursiveCteOperator { + schema_ref: vec![column_ref], + })); + with_recursive_operators.push(Operator::RecursiveScan(RecursiveScanOperator { + schema_ref: vec![column_ref], + })); + with_recursive_operators + }; #[cfg(feature = "copy")] let operators = { let mut with_copy_operators = operators; diff --git a/src/planner/operator/visitor_mut.rs b/src/planner/operator/visitor_mut.rs index b3e3d96e..864f36f6 100644 --- a/src/planner/operator/visitor_mut.rs +++ b/src/planner/operator/visitor_mut.rs @@ -118,6 +118,20 @@ pub trait OperatorVisitorMut<'a>: Sized { Ok(()) } + fn visit_recursive_cte( + &mut self, + _op: &'a mut RecursiveCteOperator, + ) -> Result<(), DatabaseError> { + Ok(()) + } + + fn visit_recursive_scan( + &mut self, + _op: &'a mut RecursiveScanOperator, + ) -> Result<(), DatabaseError> { + Ok(()) + } + fn visit_insert(&mut self, _op: &'a mut InsertOperator) -> Result<(), DatabaseError> { Ok(()) } @@ -362,6 +376,8 @@ pub fn walk_mut_operator<'a, V: OperatorVisitorMut<'a>>( Operator::Describe(op) => visitor.visit_describe(op), Operator::SetMembership(op) => visitor.visit_set_membership(op), Operator::Union(op) => visitor.visit_union(op), + Operator::RecursiveCte(op) => visitor.visit_recursive_cte(op), + Operator::RecursiveScan(op) => visitor.visit_recursive_scan(op), Operator::Insert(op) => visitor.visit_insert(op), Operator::Update(op) => visitor.visit_update(op), Operator::Delete(op) => visitor.visit_delete(op), diff --git a/tests/slt/cte.slt b/tests/slt/cte.slt new file mode 100644 index 00000000..f3a245d5 --- /dev/null +++ b/tests/slt/cte.slt @@ -0,0 +1,240 @@ +statement ok +CREATE TABLE cte_source (id INT PRIMARY KEY, value INT) + +statement ok +INSERT INTO cte_source VALUES (1, 10), (2, 20), (3, 30) + +query II +WITH selected AS ( + SELECT id, value FROM cte_source WHERE id > 1 +) +SELECT id, value FROM selected ORDER BY id +---- +2 20 +3 30 + +query II +WITH +first(id, amount) AS ( + SELECT id, value FROM cte_source +), +second AS ( + SELECT id, amount FROM first WHERE amount >= 20 +) +SELECT id, amount FROM second ORDER BY id +---- +2 20 +3 30 + +query II +WITH selected AS ( + SELECT id FROM cte_source WHERE id <= 2 +) +SELECT left_cte.id, right_cte.id +FROM selected AS left_cte +JOIN selected AS right_cte ON left_cte.id = right_cte.id +ORDER BY left_cte.id +---- +1 1 +2 2 + +query I +WITH cte_source AS ( + SELECT 42 AS id +) +SELECT id FROM cte_source +---- +42 + +query I +WITH selected AS ( + SELECT id FROM cte_source WHERE id = 2 +) +SELECT id FROM cte_source +WHERE EXISTS (SELECT 1 FROM selected) +ORDER BY id +---- +1 +2 +3 + +query II +WITH scoped(value) AS (SELECT 1) +SELECT outer_cte.value, nested.value +FROM scoped AS outer_cte +CROSS JOIN ( + WITH scoped(value) AS (SELECT 2) + SELECT value FROM scoped +) AS nested +---- +1 2 + +statement error +WITH first AS (SELECT * FROM second), second AS (SELECT 1) +SELECT * FROM first + +statement error +WITH duplicate AS (SELECT 1), duplicate AS (SELECT 2) +SELECT * FROM duplicate + +query I +WITH RECURSIVE recursive_cte(value) AS ( + SELECT 1 + UNION ALL + SELECT value + 1 FROM recursive_cte WHERE value < 5 +) +SELECT value FROM recursive_cte ORDER BY value +---- +1 +2 +3 +4 +5 + +query T +EXPLAIN WITH RECURSIVE recursive_cte(value) AS ( + SELECT 1 + UNION ALL + SELECT value + 1 FROM recursive_cte WHERE value < 3 +) +SELECT value FROM recursive_cte +---- +Projection [#4] [Project => (Sort Option: Follow)] Recursive CTE: [#4] Projection [(#3) as (#4)] [Project => (Sort Option: Follow)] Projection [1] [Project => (Sort Option: Follow)] Dummy [Dummy => (Sort Option: None)] Projection [(#4 + 1)] [Project => (Sort Option: Follow)] Filter (#4 < 3), Is Having: false [Filter => (Sort Option: Follow)] Recursive Scan: [#4] + +query I +WITH RECURSIVE recursive_cte(value) AS ( + SELECT 1 + UNION ALL + SELECT value FROM recursive_cte WHERE FALSE +) +SELECT value FROM recursive_cte +---- +1 + +query I +WITH RECURSIVE recursive_cte(value) AS ( + SELECT 1 + UNION ALL + SELECT value + 1 + FROM recursive_cte + CROSS JOIN table(numbers(2)) + WHERE value < 3 +) +SELECT count(*) FROM recursive_cte +---- +7 + +query I +WITH RECURSIVE recursive_cte(value) AS ( + SELECT 1 + UNION ALL + SELECT value + 1 FROM recursive_cte +) +SELECT value FROM recursive_cte LIMIT 5 +---- +1 +2 +3 +4 +5 + +query II +WITH RECURSIVE recursive_cte(value) AS ( + SELECT 1 + UNION ALL + SELECT value + 1 FROM recursive_cte WHERE value < 3 +) +SELECT left_cte.value, right_cte.value +FROM recursive_cte AS left_cte +JOIN recursive_cte AS right_cte ON left_cte.value = right_cte.value +ORDER BY left_cte.value +---- +1 1 +2 2 +3 3 + +query I +WITH RECURSIVE recursive_cte(value) AS ( + SELECT number FROM table(numbers(1100)) + UNION ALL + SELECT value FROM recursive_cte WHERE value < 0 +) +SELECT count(*) FROM recursive_cte +---- +1100 + +statement error +WITH RECURSIVE recursive_cte(value) AS ( + SELECT 1 + UNION ALL + SELECT left_cte.value + 1 + FROM recursive_cte AS left_cte + JOIN recursive_cte AS right_cte ON left_cte.value = right_cte.value +) +SELECT value FROM recursive_cte + +statement error +WITH RECURSIVE recursive_cte(value) AS ( + SELECT 1 + UNION + SELECT value + 1 FROM recursive_cte WHERE value < 3 +) +SELECT value FROM recursive_cte + +statement error +WITH selected AS MATERIALIZED ( + SELECT id FROM cte_source +) +SELECT * FROM selected + +statement error +WITH RECURSIVE first(x) AS ( + SELECT 1 UNION ALL SELECT x FROM first +), second(x) AS ( + SELECT 1 UNION ALL SELECT x FROM second +) +SELECT * FROM first + +statement error +WITH RECURSIVE recursive_cte(value) AS ( + SELECT 1 + UNION ALL + SELECT value FROM recursive_cte ORDER BY value +) +SELECT * FROM recursive_cte + +statement error +WITH RECURSIVE cte_source(id, value) AS ( + SELECT id, value FROM cte_source + UNION ALL + SELECT id, value FROM cte_source +) +SELECT * FROM cte_source + +statement error +WITH RECURSIVE recursive_cte(value) AS ( + SELECT 1 + UNION ALL + SELECT 2 +) +SELECT * FROM recursive_cte + +statement error +WITH RECURSIVE recursive_cte(value) AS ( + SELECT 1 + UNION ALL + SELECT value, value FROM recursive_cte +) +SELECT * FROM recursive_cte + +statement error +WITH RECURSIVE recursive_cte(value) AS ( + SELECT 1 + UNION ALL + SELECT 'value' FROM recursive_cte +) +SELECT * FROM recursive_cte + +statement error +WITH selected AS (SELECT 1) FROM cte_source +SELECT * FROM selected