Skip to content

Commit d4063b0

Browse files
KKouldKKould
committed
refactor(execution): unify executor node API and trim test helpers
1 parent 7446641 commit d4063b0

40 files changed

Lines changed: 364 additions & 312 deletions

‎src/binder/expr.rs‎

Lines changed: 62 additions & 49 deletions
Original file line numberDiff line numberDiff line change
@@ -35,6 +35,7 @@ use crate::expression::{AliasType, ScalarExpression};
3535
use crate::planner::operator::scalar_subquery::ScalarSubqueryOperator;
3636
use crate::planner::{LogicalPlan, SchemaOutput};
3737
use crate::storage::Transaction;
38+
use crate::types::tuple::SchemaRef;
3839
use crate::types::value::{DataValue, Utf8Type};
3940
use crate::types::{ColumnId, LogicalType};
4041

@@ -81,6 +82,51 @@ impl<'a, T: Transaction, A: AsRef<[(&'static str, DataValue)]>> Binder<'a, '_, T
8182
}
8283
}
8384

85+
fn find_column_in_schema<'b>(
86+
schema_ref: &'b SchemaRef,
87+
column_name: &str,
88+
) -> Option<(usize, &'b ColumnRef)> {
89+
schema_ref
90+
.iter()
91+
.enumerate()
92+
.find(|(_, column)| column.name() == column_name)
93+
}
94+
95+
fn find_column_in_scope(
96+
context: &BinderContext<'a, T>,
97+
table_schema_buf: &mut HashMap<TableName, Option<SchemaOutput>>,
98+
column_name: &str,
99+
) -> Option<ScalarExpression> {
100+
let mut position_offset = 0;
101+
102+
for bound_source in &context.bind_table {
103+
let schema_buf = table_schema_buf
104+
.entry(bound_source.table_name.clone())
105+
.or_default();
106+
let schema_ref = bound_source.source.schema_ref(schema_buf);
107+
108+
if let Some((position, column)) = Self::find_column_in_schema(&schema_ref, column_name)
109+
{
110+
return Some(ScalarExpression::column_expr(
111+
column.clone(),
112+
position_offset + position,
113+
));
114+
}
115+
116+
position_offset += schema_ref.len();
117+
}
118+
119+
None
120+
}
121+
122+
fn column_not_found_with_span(idents: &[Ident], column_name: &str) -> DatabaseError {
123+
let err = DatabaseError::column_not_found(column_name.to_string());
124+
match idents.last() {
125+
Some(ident) => attach_span_from_sqlparser_span_if_absent(err, ident.span),
126+
None => err,
127+
}
128+
}
129+
84130
pub(crate) fn bind_expr(&mut self, expr: &Expr) -> Result<ScalarExpression, DatabaseError> {
85131
match expr {
86132
Expr::Identifier(ident) => {
@@ -497,69 +543,36 @@ impl<'a, T: Transaction, A: AsRef<[(&'static str, DataValue)]>> Binder<'a, '_, T
497543
}
498544
}
499545
};
500-
let (position, column) = schema_ref
501-
.iter()
502-
.enumerate()
503-
.find(|(_, column)| column.name() == full_name.1)
504-
.ok_or_else(|| {
505-
let err = DatabaseError::column_not_found(full_name.1.to_string());
506-
match idents.last() {
507-
Some(ident) => attach_span_from_sqlparser_span_if_absent(err, ident.span),
508-
None => err,
509-
}
510-
})?;
546+
let (position, column) = Self::find_column_in_schema(&schema_ref, full_name.1.as_str())
547+
.ok_or_else(|| Self::column_not_found_with_span(idents, full_name.1.as_str()))?;
511548

512549
Ok(ScalarExpression::column_expr(
513550
column.clone(),
514551
position_offset + position,
515552
))
516553
} else {
517-
let op =
518-
|got_column: &mut Option<ScalarExpression>,
519-
context: &BinderContext<'a, T>,
520-
table_schema_buf: &mut HashMap<TableName, Option<SchemaOutput>>| {
521-
let mut position_offset = 0;
522-
523-
for bound_source in &context.bind_table {
524-
if got_column.is_some() {
525-
break;
526-
}
527-
let schema_buf = table_schema_buf
528-
.entry(bound_source.table_name.clone())
529-
.or_default();
530-
let schema_ref = bound_source.source.schema_ref(schema_buf);
531-
if let Some((position, column)) = schema_ref
532-
.iter()
533-
.enumerate()
534-
.find(|(_, column)| column.name() == full_name.1)
535-
{
536-
*got_column = Some(ScalarExpression::column_expr(
537-
column.clone(),
538-
position_offset + position,
539-
));
540-
}
541-
position_offset += schema_ref.len();
542-
}
543-
};
544554
// handle col syntax
545-
let mut got_column = None;
546-
547-
op(&mut got_column, &self.context, &mut self.table_schema_buf);
555+
let mut got_column = Self::find_column_in_scope(
556+
&self.context,
557+
&mut self.table_schema_buf,
558+
full_name.1.as_str(),
559+
);
548560
if got_column.is_none() {
549561
if let Some(parent) = self.parent {
550562
self.context.mark_outer_ref();
551-
op(&mut got_column, &parent.context, &mut self.table_schema_buf);
563+
got_column = Self::find_column_in_scope(
564+
&parent.context,
565+
&mut self.table_schema_buf,
566+
full_name.1.as_str(),
567+
);
552568
}
553569
}
554570
match got_column {
555571
Some(column) => Ok(column),
556-
None => {
557-
let err = DatabaseError::column_not_found(full_name.1.clone());
558-
Err(match idents.last() {
559-
Some(ident) => attach_span_from_sqlparser_span_if_absent(err, ident.span),
560-
None => err,
561-
})
562-
}
572+
None => Err(Self::column_not_found_with_span(
573+
idents,
574+
full_name.1.as_str(),
575+
)),
563576
}
564577
}
565578
}

‎src/binder/insert.rs‎

Lines changed: 26 additions & 28 deletions
Original file line numberDiff line numberDiff line change
@@ -29,6 +29,7 @@ use crate::storage::Transaction;
2929
use crate::types::tuple::SchemaRef;
3030
use crate::types::value::DataValue;
3131
use sqlparser::ast::{Expr, Ident, ObjectName, Query};
32+
use std::borrow::Cow;
3233
use std::slice;
3334
use std::sync::Arc;
3435

@@ -147,17 +148,34 @@ impl<T: Transaction, A: AsRef<[(&'static str, DataValue)]>> Binder<'_, '_, T, A>
147148
is_overwrite: bool,
148149
) -> Result<LogicalPlan, DatabaseError> {
149150
let table_name: Arc<str> = lower_case_name(name)?.into();
150-
let source = self
151-
.context
152-
.source(&table_name)?
153-
.ok_or(DatabaseError::TableNotFound)?;
154-
let mut schema_buf = None;
155-
let table_schema = source.schema_ref(&mut schema_buf);
151+
let table_schema = {
152+
let source = self
153+
.context
154+
.source(&table_name)?
155+
.ok_or(DatabaseError::TableNotFound)?;
156+
let mut schema_buf = None;
157+
source.schema_ref(&mut schema_buf)
158+
};
159+
160+
let mut input_plan = self.bind_query(query)?;
161+
let input_schema = input_plan.output_schema().clone();
162+
let input_len = input_schema.len();
156163

157164
let target_columns = if idents.is_empty() {
158-
None
165+
if input_len > table_schema.len() {
166+
return Err(DatabaseError::ValuesLenMismatch(
167+
table_schema.len(),
168+
input_len,
169+
));
170+
}
171+
Cow::Borrowed(&table_schema[..input_len])
159172
} else {
160173
let mut columns = Vec::with_capacity(idents.len());
174+
let source = self
175+
.context
176+
.source(&table_name)?
177+
.ok_or(DatabaseError::TableNotFound)?;
178+
let mut schema_buf = None;
161179
for ident in idents {
162180
let column_name = lower_ident(ident);
163181
let column = source
@@ -170,30 +188,10 @@ impl<T: Transaction, A: AsRef<[(&'static str, DataValue)]>> Binder<'_, '_, T, A>
170188
})?;
171189
columns.push(column);
172190
}
173-
Some(columns)
174-
};
175-
176-
let mut input_plan = self.bind_query(query)?;
177-
let input_schema = input_plan.output_schema().clone();
178-
let input_len = input_schema.len();
179-
180-
let target_columns = if let Some(columns) = target_columns {
181191
if input_len != columns.len() {
182192
return Err(DatabaseError::ValuesLenMismatch(columns.len(), input_len));
183193
}
184-
columns
185-
} else {
186-
if input_len > table_schema.len() {
187-
return Err(DatabaseError::ValuesLenMismatch(
188-
table_schema.len(),
189-
input_len,
190-
));
191-
}
192-
table_schema
193-
.iter()
194-
.take(input_len)
195-
.cloned()
196-
.collect::<Vec<_>>()
194+
Cow::Owned(columns)
197195
};
198196

199197
let projection = input_schema

‎src/binder/select.rs‎

Lines changed: 24 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -263,7 +263,7 @@ impl<'a: 'b, 'b, T: Transaction, A: AsRef<[(&'static str, DataValue)]>> Binder<'
263263
mut children: LogicalPlan,
264264
mut plan: LogicalPlan,
265265
join_ty: JoinType,
266-
predicates: Vec<ScalarExpression>,
266+
predicates: impl IntoIterator<Item = ScalarExpression>,
267267
rebind_positions: bool,
268268
) -> Result<LogicalPlan, DatabaseError> {
269269
let left_schema = children.output_schema().clone();
@@ -1002,8 +1002,7 @@ impl<'a: 'b, 'b, T: Transaction, A: AsRef<[(&'static str, DataValue)]>> Binder<'
10021002
}
10031003
};
10041004

1005-
if let Some(idents) = alias_idents {
1006-
let alias_name = table_alias.unwrap();
1005+
if let (Some(idents), Some(alias_name)) = (alias_idents, table_alias) {
10071006
plan = self.bind_alias(plan, idents, alias_name.clone(), table_name.clone())?;
10081007
self.context.add_bound_source(
10091008
table_name,
@@ -1043,19 +1042,20 @@ impl<'a: 'b, 'b, T: Transaction, A: AsRef<[(&'static str, DataValue)]>> Binder<'
10431042
});
10441043
}
10451044
SelectItem::Wildcard(_) => {
1046-
let mut seen_tables = HashSet::new();
1047-
let bound_sources = self.context.bind_table.clone();
1048-
for bound_source in bound_sources {
1049-
let visible_name = bound_source.visible_name().clone();
1050-
if !seen_tables.insert(visible_name.clone()) {
1051-
continue;
1052-
}
1053-
if Self::is_joined_values_source(
1054-
bound_source.join_type,
1055-
&bound_source.source,
1056-
) {
1057-
continue;
1058-
}
1045+
for visible_name in self
1046+
.context
1047+
.bind_table
1048+
.iter()
1049+
.filter(|bound_source| {
1050+
!Self::is_joined_values_source(
1051+
bound_source.join_type,
1052+
&bound_source.source,
1053+
)
1054+
})
1055+
.map(|bound_source| bound_source.visible_name())
1056+
.unique()
1057+
.cloned()
1058+
{
10591059
Self::bind_table_column_refs(
10601060
&self.context,
10611061
&mut self.table_schema_buf,
@@ -1286,7 +1286,7 @@ impl<'a: 'b, 'b, T: Transaction, A: AsRef<[(&'static str, DataValue)]>> Binder<'
12861286
children,
12871287
plan,
12881288
join_ty,
1289-
vec![predicate.clone()],
1289+
std::iter::once(predicate.clone()),
12901290
true,
12911291
)?;
12921292
}
@@ -1382,11 +1382,13 @@ impl<'a: 'b, 'b, T: Transaction, A: AsRef<[(&'static str, DataValue)]>> Binder<'
13821382
let (plan, correlated_filters) =
13831383
Self::prepare_correlated_subquery_plan(plan, children.output_schema(), true)?;
13841384
let predicate = Self::rewrite_correlated_in_predicate(predicate);
1385-
let predicates = std::iter::once(predicate)
1386-
.chain(correlated_filters)
1387-
.collect();
1388-
1389-
Self::build_join_from_split_scope_predicates(children, plan, join_ty, predicates, false)
1385+
Self::build_join_from_split_scope_predicates(
1386+
children,
1387+
plan,
1388+
join_ty,
1389+
std::iter::once(predicate).chain(correlated_filters),
1390+
false,
1391+
)
13901392
}
13911393

13921394
fn rewrite_correlated_in_predicate(predicate: ScalarExpression) -> ScalarExpression {

‎src/execution/ddl/add_column.rs‎

Lines changed: 0 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -47,9 +47,7 @@ impl AddColumn {
4747
pub(crate) fn next_tuple<'a, T: Transaction>(
4848
&mut self,
4949
arena: &mut ExecArena<'a, T>,
50-
id: ExecId,
5150
) -> Result<(), DatabaseError> {
52-
let _ = id;
5351
let table_cache = arena.table_cache();
5452
let Some(AddColumnOperator {
5553
table_name,

‎src/execution/ddl/change_column.rs‎

Lines changed: 0 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -45,9 +45,7 @@ impl ChangeColumn {
4545
pub(crate) fn next_tuple<'a, T: Transaction>(
4646
&mut self,
4747
arena: &mut ExecArena<'a, T>,
48-
id: ExecId,
4948
) -> Result<(), DatabaseError> {
50-
let _ = id;
5149
let table_cache = arena.table_cache();
5250
let Some(ChangeColumnOperator {
5351
table_name,

‎src/execution/ddl/create_index.rs‎

Lines changed: 0 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -66,9 +66,7 @@ impl CreateIndex {
6666
pub(crate) fn next_tuple<'a, T: Transaction>(
6767
&mut self,
6868
arena: &mut ExecArena<'a, T>,
69-
id: ExecId,
7069
) -> Result<(), DatabaseError> {
71-
let _ = id;
7270
let table_cache = arena.table_cache();
7371

7472
let Some(CreateIndexOperator {

‎src/execution/ddl/create_table.rs‎

Lines changed: 0 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -43,9 +43,7 @@ impl CreateTable {
4343
pub(crate) fn next_tuple<'a, T: Transaction>(
4444
&mut self,
4545
arena: &mut ExecArena<'a, T>,
46-
id: ExecId,
4746
) -> Result<(), DatabaseError> {
48-
let _ = id;
4947
let Some(CreateTableOperator {
5048
table_name,
5149
columns,

‎src/execution/ddl/create_view.rs‎

Lines changed: 0 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -43,9 +43,7 @@ impl CreateView {
4343
pub(crate) fn next_tuple<'a, T: Transaction>(
4444
&mut self,
4545
arena: &mut ExecArena<'a, T>,
46-
id: ExecId,
4746
) -> Result<(), DatabaseError> {
48-
let _ = id;
4947
let Some(CreateViewOperator { view, or_replace }) = self.op.take() else {
5048
arena.finish();
5149
return Ok(());

‎src/execution/ddl/drop_column.rs‎

Lines changed: 0 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -45,9 +45,7 @@ impl DropColumn {
4545
pub(crate) fn next_tuple<'a, T: Transaction>(
4646
&mut self,
4747
arena: &mut ExecArena<'a, T>,
48-
id: ExecId,
4948
) -> Result<(), DatabaseError> {
50-
let _ = id;
5149
let table_cache = arena.table_cache();
5250
let meta_cache = arena.meta_cache();
5351
let Some(DropColumnOperator {

‎src/execution/ddl/drop_index.rs‎

Lines changed: 0 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -43,9 +43,7 @@ impl DropIndex {
4343
pub(crate) fn next_tuple<'a, T: Transaction>(
4444
&mut self,
4545
arena: &mut ExecArena<'a, T>,
46-
id: ExecId,
4746
) -> Result<(), DatabaseError> {
48-
let _ = id;
4947
let Some(DropIndexOperator {
5048
table_name,
5149
index_name,

0 commit comments

Comments
 (0)