Skip to content

Commit f85baee

Browse files
KKouldKKould
committed
refactor: introduce execution runtime abstraction
1 parent de2163c commit f85baee

56 files changed

Lines changed: 2396 additions & 2014 deletions

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

‎Cargo.toml‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -24,7 +24,8 @@ doctest = false
2424
crate-type = ["cdylib", "rlib"]
2525

2626
[features]
27-
default = ["macros", "parser", "rocksdb"]
27+
default = ["time", "macros", "parser", "rocksdb"]
28+
time = ["dep:chrono"]
2829
copy = ["dep:csv"]
2930
decimal = ["dep:rust_decimal"]
3031
macros = []
@@ -47,7 +48,6 @@ required-features = ["pprof"]
4748
ahash = { version = "0.8" }
4849
bumpalo = { version = "3", default-features = false, features = ["collections"] }
4950
byteorder = { version = "1" }
50-
chrono = { version = "0.4" }
5151
fixedbitset = { version = "0.4" }
5252
itertools = { version = "0.12" }
5353
ordered-float = { version = "4" }
@@ -59,6 +59,7 @@ ulid = { version = "1" }
5959

6060
# Optional dependencies for features
6161
comfy-table = { version = "7", default-features = false, optional = true }
62+
chrono = { version = "0.4", optional = true }
6263
csv = { version = "1", optional = true }
6364
pyo3 = { version = "0.23", features = ["auto-initialize"], optional = true }
6465
rust_decimal = { version = "1", default-features = false, features = ["std"], optional = true }

‎src/binder/aggregate.rs‎

Lines changed: 13 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -98,11 +98,22 @@ impl<T: Transaction, A: AsRef<[(&'static str, DataValue)]>> Binder<'_, '_, T, A>
9898
exprs: impl IntoIterator<Item = &'c mut ScalarExpression>,
9999
arena: &mut crate::planner::PlanArena,
100100
) -> Result<(), DatabaseError> {
101-
let mut binder = AggregateOutputBinder::new(
101+
self.bind_aggregate_output_exprs_with_outputs(
102102
&self.context.agg_calls,
103103
&self.context.group_by_exprs,
104+
exprs,
104105
arena,
105-
);
106+
)
107+
}
108+
109+
pub(crate) fn bind_aggregate_output_exprs_with_outputs<'c>(
110+
&self,
111+
agg_calls: &[ScalarExpression],
112+
group_by_exprs: &[ScalarExpression],
113+
exprs: impl IntoIterator<Item = &'c mut ScalarExpression>,
114+
arena: &mut crate::planner::PlanArena,
115+
) -> Result<(), DatabaseError> {
116+
let mut binder = AggregateOutputBinder::new(agg_calls, group_by_exprs, arena);
106117
for expr in exprs {
107118
binder.visit(expr)?;
108119
}

‎src/binder/parser.rs‎

Lines changed: 67 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -1672,45 +1672,86 @@ impl TryFrom<DataType> for LogicalType {
16721672
| DataType::UBigInt
16731673
| DataType::UInt64 => Ok(Self::UBigint),
16741674
DataType::Boolean => Ok(Self::Boolean),
1675-
DataType::Date => Ok(Self::Date),
1675+
DataType::Date => {
1676+
#[cfg(feature = "time")]
1677+
{
1678+
Ok(Self::Date)
1679+
}
1680+
#[cfg(not(feature = "time"))]
1681+
{
1682+
Err(DatabaseError::UnsupportedStmt(
1683+
"time types require the `time` feature".to_string(),
1684+
))
1685+
}
1686+
}
16761687
DataType::Datetime(precision) => {
1677-
if precision.is_some() {
1678-
return Err(DatabaseError::UnsupportedStmt(
1679-
"time's precision".to_string(),
1680-
));
1688+
#[cfg(feature = "time")]
1689+
{
1690+
if precision.is_some() {
1691+
return Err(DatabaseError::UnsupportedStmt(
1692+
"time's precision".to_string(),
1693+
));
1694+
}
1695+
Ok(Self::DateTime)
1696+
}
1697+
#[cfg(not(feature = "time"))]
1698+
{
1699+
let _ = precision;
1700+
Err(DatabaseError::UnsupportedStmt(
1701+
"time types require the `time` feature".to_string(),
1702+
))
16811703
}
1682-
Ok(Self::DateTime)
16831704
}
16841705
DataType::Time(precision, info) => {
1685-
match precision {
1686-
Some(0..5) | None => (),
1687-
_ => {
1706+
#[cfg(feature = "time")]
1707+
{
1708+
match precision {
1709+
Some(0..5) | None => (),
1710+
_ => {
1711+
return Err(DatabaseError::UnsupportedStmt(
1712+
"time's precision must be less than 5".to_string(),
1713+
))
1714+
}
1715+
}
1716+
if !matches!(info, sqlparser::ast::TimezoneInfo::None) {
16881717
return Err(DatabaseError::UnsupportedStmt(
1689-
"time's precision must be less than 5".to_string(),
1690-
))
1718+
"time's zone is not supported".to_string(),
1719+
));
16911720
}
1721+
Ok(Self::Time(precision))
16921722
}
1693-
if !matches!(info, sqlparser::ast::TimezoneInfo::None) {
1694-
return Err(DatabaseError::UnsupportedStmt(
1695-
"time's zone is not supported".to_string(),
1696-
));
1723+
#[cfg(not(feature = "time"))]
1724+
{
1725+
let _ = (precision, info);
1726+
Err(DatabaseError::UnsupportedStmt(
1727+
"time types require the `time` feature".to_string(),
1728+
))
16971729
}
1698-
Ok(Self::Time(precision))
16991730
}
17001731
DataType::Timestamp(precision, info) => {
1701-
let mut zone = false;
1702-
match precision {
1703-
Some(3 | 6 | 9) | None => (),
1704-
_ => {
1705-
return Err(DatabaseError::UnsupportedStmt(
1706-
"timestamp's precision must be 3,6,9".to_string(),
1707-
))
1732+
#[cfg(feature = "time")]
1733+
{
1734+
let mut zone = false;
1735+
match precision {
1736+
Some(3 | 6 | 9) | None => (),
1737+
_ => {
1738+
return Err(DatabaseError::UnsupportedStmt(
1739+
"timestamp's precision must be 3,6,9".to_string(),
1740+
))
1741+
}
17081742
}
1743+
if matches!(info, sqlparser::ast::TimezoneInfo::WithTimeZone) {
1744+
zone = true;
1745+
}
1746+
Ok(Self::TimeStamp(precision, zone))
17091747
}
1710-
if matches!(info, sqlparser::ast::TimezoneInfo::WithTimeZone) {
1711-
zone = true;
1748+
#[cfg(not(feature = "time"))]
1749+
{
1750+
let _ = (precision, info);
1751+
Err(DatabaseError::UnsupportedStmt(
1752+
"time types require the `time` feature".to_string(),
1753+
))
17121754
}
1713-
Ok(Self::TimeStamp(precision, zone))
17141755
}
17151756
DataType::Decimal(info)
17161757
| DataType::DecimalUnsigned(info)

‎src/binder/select.rs‎

Lines changed: 15 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -395,15 +395,13 @@ where
395395
}
396396

397397
pub fn distinct(mut self) -> Result<Self, DatabaseError> {
398-
self.plan = self
399-
.binder
400-
.bind_distinct(self.plan, self.select_list.clone())?;
401398
let distinct_outputs = self.select_list.clone();
402399
self.binder.bind_distinct_output_exprs(
403400
&distinct_outputs,
404401
self.select_list.iter_mut(),
405402
self.arena,
406403
)?;
404+
self.plan = self.binder.bind_distinct(self.plan, distinct_outputs)?;
407405
Ok(self)
408406
}
409407

@@ -510,19 +508,25 @@ where
510508
if !self.binder.context.agg_calls.is_empty()
511509
|| !self.binder.context.group_by_exprs.is_empty()
512510
{
513-
self.plan = self.binder.bind_aggregate(
514-
self.plan,
515-
self.binder.context.agg_calls.clone(),
516-
self.binder.context.group_by_exprs.clone(),
511+
let agg_calls = std::mem::take(&mut self.binder.context.agg_calls);
512+
let group_by_exprs = std::mem::take(&mut self.binder.context.group_by_exprs);
513+
self.binder.bind_aggregate_output_exprs_with_outputs(
514+
&agg_calls,
515+
&group_by_exprs,
516+
self.select_list.iter_mut(),
517+
self.arena,
517518
)?;
518-
self.binder
519-
.bind_aggregate_output_exprs(self.select_list.iter_mut(), self.arena)?;
520519
if let Some(orderby) = having_orderby.1.as_mut() {
521-
self.binder.bind_aggregate_output_exprs(
520+
self.binder.bind_aggregate_output_exprs_with_outputs(
521+
&agg_calls,
522+
&group_by_exprs,
522523
orderby.iter_mut().map(|field| &mut field.expr),
523524
self.arena,
524525
)?;
525526
}
527+
self.plan = self
528+
.binder
529+
.bind_aggregate(self.plan, agg_calls, group_by_exprs)?;
526530
}
527531

528532
Ok(BindPlanAggregated {
@@ -568,9 +572,6 @@ where
568572
distinct: bool,
569573
) -> Result<BindPlanDistinct<'s, 'a, 'b, 'arena, T, A>, DatabaseError> {
570574
if distinct {
571-
self.plan = self
572-
.binder
573-
.bind_distinct(self.plan, self.select_list.clone())?;
574575
let distinct_outputs = self.select_list.clone();
575576
self.binder.bind_distinct_output_exprs(
576577
&distinct_outputs,
@@ -581,6 +582,7 @@ where
581582
self.binder
582583
.bind_distinct_orderby_exprs(&distinct_outputs, orderby, self.arena)?;
583584
}
585+
self.plan = self.binder.bind_distinct(self.plan, distinct_outputs)?;
584586
}
585587

586588
Ok(BindPlanDistinct {

‎src/db.rs‎

Lines changed: 22 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -18,15 +18,18 @@ use crate::binder::{Binder, BinderContext};
1818
use crate::catalog::TableName;
1919
use crate::errors::DatabaseError;
2020
use crate::execution::{
21-
build_write, build_write_read_context, DDLApply, ExecArena, Executor, WriteExecutionContext,
21+
build_write, build_write_read_context, DDLApply, ExecArena, ExecuteRuntime, Executor,
22+
WriteExecutionContext,
2223
};
2324
use crate::expression::function::scala::ScalarFunctionImpl;
2425
use crate::expression::function::table::{
2526
ArcTableFunctionImpl, TableFunctionCatalog, TableFunctionImpl,
2627
};
2728
use crate::expression::function::FunctionSummary;
2829
use crate::function::char_length::CharLength;
30+
#[cfg(feature = "time")]
2931
use crate::function::current_date::CurrentDate;
32+
#[cfg(feature = "time")]
3033
use crate::function::current_timestamp::CurrentTimeStamp;
3134
use crate::function::lower::Lower;
3235
use crate::function::numbers::Numbers;
@@ -289,7 +292,9 @@ impl DataBaseBuilder {
289292

290293
state.load_scalar_function(CharLength::new("char_length".to_lowercase()));
291294
state.load_scalar_function(CharLength::new("character_length".to_lowercase()));
295+
#[cfg(feature = "time")]
292296
state.load_scalar_function(CurrentDate::new());
297+
#[cfg(feature = "time")]
293298
state.load_scalar_function(CurrentTimeStamp::new());
294299
state.load_scalar_function(Lower::new());
295300
state.load_scalar_function(OctetLength::new());
@@ -503,7 +508,7 @@ impl<S: Storage> State<S> {
503508
BinderContext::new(
504509
self.table_cache(),
505510
self.view_cache(),
506-
transaction,
511+
&*transaction,
507512
self.scala_functions(),
508513
self.table_functions(),
509514
),
@@ -552,20 +557,23 @@ impl<S: Storage> State<S> {
552557
let (mut plan, mut plan_arena) = self.build_plan(params, transaction, build)?;
553558
let schema = plan.take_schema(&mut plan_arena);
554559
let mut arena = ExecArena::new();
560+
let read_context = (
561+
&self.table_cache,
562+
&self.view_cache,
563+
&self.meta_cache,
564+
&self.scala_functions,
565+
&self.table_functions,
566+
);
555567
let root = build_write_read_context(
556568
&mut arena,
557569
&mut plan_arena,
558570
plan,
559-
(
560-
&self.table_cache,
561-
&self.view_cache,
562-
&self.meta_cache,
563-
&self.scala_functions,
564-
&self.table_functions,
565-
),
571+
read_context,
566572
transaction,
567573
);
568-
let executor = Executor::new(arena, root);
574+
let mut runtime = ExecuteRuntime::new(arena);
575+
runtime.init_context(read_context, transaction);
576+
let executor = Executor::new(runtime, root);
569577

570578
Ok((schema, plan_arena, executor))
571579
})() {
@@ -640,8 +648,10 @@ impl<S: Storage> State<S> {
640648
scala_functions,
641649
table_functions,
642650
);
643-
let root = build_write(&mut arena, &mut plan_arena, plan, cache, transaction);
644-
let executor = Executor::new(arena, root);
651+
let root = build_write(&mut arena, &mut plan_arena, plan, &cache, &*transaction);
652+
let mut runtime = ExecuteRuntime::new(arena);
653+
runtime.init_context_mut(cache, transaction);
654+
let executor = Executor::new(runtime, root);
645655

646656
Ok((schema, plan_arena, executor))
647657
}

‎src/errors.rs‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,7 @@
1515
use crate::expression::{BinaryOperator, UnaryOperator};
1616
use crate::types::tuple::TupleId;
1717
use crate::types::LogicalType;
18+
#[cfg(feature = "time")]
1819
use chrono::ParseError;
1920
#[cfg(feature = "parser")]
2021
use sqlparser::parser::ParserError;
@@ -114,6 +115,7 @@ pub enum DatabaseError {
114115
},
115116
OverFlow,
116117
ParseBool(ParseBoolError),
118+
#[cfg(feature = "time")]
117119
ParseDate(ParseError),
118120
ParseFloat(ParseFloatError),
119121
ParseInt(ParseIntError),
@@ -217,6 +219,7 @@ impl fmt::Display for DatabaseError {
217219
Self::NotNull { column, span } => f.write_str(&format_not_null_message(column, span)),
218220
Self::OverFlow => f.write_str("over flow"),
219221
Self::ParseBool(err) => write!(f, "parser bool: {err}"),
222+
#[cfg(feature = "time")]
220223
Self::ParseDate(err) => write!(f, "parser date: {err}"),
221224
Self::ParseFloat(err) => write!(f, "parser float: {err}"),
222225
Self::ParseInt(err) => write!(f, "parser int: {err}"),
@@ -279,6 +282,7 @@ impl Error for DatabaseError {
279282
Self::FromUtf8Error(err) => Some(err),
280283
Self::IO(err) => Some(err),
281284
Self::ParseBool(err) => Some(err),
285+
#[cfg(feature = "time")]
282286
Self::ParseDate(err) => Some(err),
283287
Self::ParseFloat(err) => Some(err),
284288
Self::ParseInt(err) => Some(err),
@@ -312,6 +316,7 @@ impl_from_database_error!(csv::Error, Csv);
312316
impl_from_database_error!(FromUtf8Error, FromUtf8Error);
313317
impl_from_database_error!(std::io::Error, IO);
314318
impl_from_database_error!(ParseBoolError, ParseBool);
319+
#[cfg(feature = "time")]
315320
impl_from_database_error!(ParseError, ParseDate);
316321
impl_from_database_error!(ParseFloatError, ParseFloat);
317322
impl_from_database_error!(ParseIntError, ParseInt);

0 commit comments

Comments
 (0)