You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@arrow.apache.org by ks...@apache.org on 2019/03/04 11:43:51 UTC
[arrow] branch master updated: ARROW-4739: [Rust] LogicalPlan can
now be passed to threads
This is an automated email from the ASF dual-hosted git repository.
kszucs pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/arrow.git
The following commit(s) were added to refs/heads/master by this push:
new 72f5774 ARROW-4739: [Rust] LogicalPlan can now be passed to threads
72f5774 is described below
commit 72f577453036f0686229a3676fb99727b13540c5
Author: Andy Grove <an...@gmail.com>
AuthorDate: Mon Mar 4 12:43:30 2019 +0100
ARROW-4739: [Rust] LogicalPlan can now be passed to threads
I have been working on a PoC of parallel execution with partitioned data sources. This PR is for one small step towards that - the ability to pass logical plans to threads. For example, we often want to execute the same plan simultaneously across all partitions.
The change was mostly just replacing `Rc` with `Arc`.
Author: Andy Grove <an...@gmail.com>
Closes #3798 from andygrove/ARROW-4739 and squashes the following commits:
88f27d2e <Andy Grove> Add test
8e12fb8d <Andy Grove> Logical plan uses Arc instead of Rc
---
rust/datafusion/src/execution/context.rs | 18 +++---
rust/datafusion/src/logicalplan.rs | 72 ++++++++++++++--------
rust/datafusion/src/optimizer/optimizer.rs | 4 +-
.../src/optimizer/projection_push_down.rs | 48 +++++++--------
rust/datafusion/src/sqlplanner.rs | 59 +++++++++---------
5 files changed, 110 insertions(+), 91 deletions(-)
diff --git a/rust/datafusion/src/execution/context.rs b/rust/datafusion/src/execution/context.rs
index ae1f1cd..b26b310 100644
--- a/rust/datafusion/src/execution/context.rs
+++ b/rust/datafusion/src/execution/context.rs
@@ -55,12 +55,18 @@ impl ExecutionContext {
/// Execute a SQL query and produce a Relation (a schema-aware iterator over a series
/// of RecordBatch instances)
pub fn sql(&mut self, sql: &str, batch_size: usize) -> Result<Rc<RefCell<Relation>>> {
+ let plan = self.create_logical_plan(sql)?;
+ Ok(self.execute(&plan, batch_size)?)
+ }
+
+ /// Creates a logical plan
+ pub fn create_logical_plan(&mut self, sql: &str) -> Result<Arc<LogicalPlan>> {
let ast = DFParser::parse_sql(String::from(sql))?;
match ast {
DFASTNode::ANSI(ansi) => {
- let schema_provider: Rc<SchemaProvider> =
- Rc::new(ExecutionContextSchemaProvider {
+ let schema_provider: Arc<SchemaProvider> =
+ Arc::new(ExecutionContextSchemaProvider {
datasources: self.datasources.clone(),
});
@@ -70,11 +76,7 @@ impl ExecutionContext {
// plan the query (create a logical relational plan)
let plan = query_planner.sql_to_rel(&ansi)?;
- let optimized_plan = self.optimize(&plan)?;
-
- let relation = self.execute(&optimized_plan, batch_size)?;
-
- Ok(relation)
+ Ok(self.optimize(&plan)?)
}
_ => unimplemented!(),
}
@@ -99,7 +101,7 @@ impl ExecutionContext {
}
/// Optimize the logical plan by applying optimizer rules
- fn optimize(&self, plan: &LogicalPlan) -> Result<Rc<LogicalPlan>> {
+ fn optimize(&self, plan: &LogicalPlan) -> Result<Arc<LogicalPlan>> {
let mut rule = ProjectionPushDown::new();
Ok(rule.optimize(plan)?)
}
diff --git a/rust/datafusion/src/logicalplan.rs b/rust/datafusion/src/logicalplan.rs
index 5e3b17e..b4f4b27 100644
--- a/rust/datafusion/src/logicalplan.rs
+++ b/rust/datafusion/src/logicalplan.rs
@@ -19,7 +19,6 @@
use std::fmt;
use std::fmt::{Error, Formatter};
-use std::rc::Rc;
use std::sync::Arc;
use arrow::datatypes::*;
@@ -109,7 +108,7 @@ pub enum ScalarValue {
UInt16(u16),
UInt32(u32),
UInt64(u64),
- Utf8(Rc<String>),
+ Utf8(Arc<String>),
Struct(Vec<ScalarValue>),
}
@@ -143,18 +142,21 @@ pub enum Expr {
Literal(ScalarValue),
/// binary expression e.g. "age > 21"
BinaryExpr {
- left: Rc<Expr>,
+ left: Arc<Expr>,
op: Operator,
- right: Rc<Expr>,
+ right: Arc<Expr>,
},
/// unary IS NOT NULL
- IsNotNull(Rc<Expr>),
+ IsNotNull(Arc<Expr>),
/// unary IS NULL
- IsNull(Rc<Expr>),
+ IsNull(Arc<Expr>),
/// cast a value to a different type
- Cast { expr: Rc<Expr>, data_type: DataType },
+ Cast {
+ expr: Arc<Expr>,
+ data_type: DataType,
+ },
/// sort expression
- Sort { expr: Rc<Expr>, asc: bool },
+ Sort { expr: Arc<Expr>, asc: bool },
/// scalar function
ScalarFunction {
name: String,
@@ -210,7 +212,7 @@ impl Expr {
Ok(self.clone())
} else if can_coerce_from(cast_to_type, &this_type) {
Ok(Expr::Cast {
- expr: Rc::new(self.clone()),
+ expr: Arc::new(self.clone()),
data_type: cast_to_type.clone(),
})
} else {
@@ -223,49 +225,49 @@ impl Expr {
pub fn eq(&self, other: &Expr) -> Expr {
Expr::BinaryExpr {
- left: Rc::new(self.clone()),
+ left: Arc::new(self.clone()),
op: Operator::Eq,
- right: Rc::new(other.clone()),
+ right: Arc::new(other.clone()),
}
}
pub fn not_eq(&self, other: &Expr) -> Expr {
Expr::BinaryExpr {
- left: Rc::new(self.clone()),
+ left: Arc::new(self.clone()),
op: Operator::NotEq,
- right: Rc::new(other.clone()),
+ right: Arc::new(other.clone()),
}
}
pub fn gt(&self, other: &Expr) -> Expr {
Expr::BinaryExpr {
- left: Rc::new(self.clone()),
+ left: Arc::new(self.clone()),
op: Operator::Gt,
- right: Rc::new(other.clone()),
+ right: Arc::new(other.clone()),
}
}
pub fn gt_eq(&self, other: &Expr) -> Expr {
Expr::BinaryExpr {
- left: Rc::new(self.clone()),
+ left: Arc::new(self.clone()),
op: Operator::GtEq,
- right: Rc::new(other.clone()),
+ right: Arc::new(other.clone()),
}
}
pub fn lt(&self, other: &Expr) -> Expr {
Expr::BinaryExpr {
- left: Rc::new(self.clone()),
+ left: Arc::new(self.clone()),
op: Operator::Lt,
- right: Rc::new(other.clone()),
+ right: Arc::new(other.clone()),
}
}
pub fn lt_eq(&self, other: &Expr) -> Expr {
Expr::BinaryExpr {
- left: Rc::new(self.clone()),
+ left: Arc::new(self.clone()),
op: Operator::LtEq,
- right: Rc::new(other.clone()),
+ right: Arc::new(other.clone()),
}
}
}
@@ -323,14 +325,14 @@ pub enum LogicalPlan {
/// A Projection (essentially a SELECT with an expression list)
Projection {
expr: Vec<Expr>,
- input: Rc<LogicalPlan>,
+ input: Arc<LogicalPlan>,
schema: Arc<Schema>,
},
/// A Selection (essentially a WHERE clause with a predicate expression)
- Selection { expr: Expr, input: Rc<LogicalPlan> },
+ Selection { expr: Expr, input: Arc<LogicalPlan> },
/// Represents a list of aggregate expressions with optional grouping expressions
Aggregate {
- input: Rc<LogicalPlan>,
+ input: Arc<LogicalPlan>,
group_expr: Vec<Expr>,
aggr_expr: Vec<Expr>,
schema: Arc<Schema>,
@@ -338,7 +340,7 @@ pub enum LogicalPlan {
/// Represents a list of sort expressions to be applied to a relation
Sort {
expr: Vec<Expr>,
- input: Rc<LogicalPlan>,
+ input: Arc<LogicalPlan>,
schema: Arc<Schema>,
},
/// A table scan against a table that has been registered on a context
@@ -353,7 +355,7 @@ pub enum LogicalPlan {
// Represents the maximum number of records to return
Limit {
expr: Expr,
- input: Rc<LogicalPlan>,
+ input: Arc<LogicalPlan>,
schema: Arc<Schema>,
},
}
@@ -621,6 +623,24 @@ pub fn can_coerce_from(left: &DataType, other: &DataType) -> bool {
mod tests {
use super::*;
use serde_json;
+ use std::thread;
+
+ #[test]
+ fn logical_plan_can_be_shared_between_threads() {
+ let schema = Schema::new(vec![]);
+ let plan = Arc::new(LogicalPlan::TableScan {
+ schema_name: "".to_string(),
+ table_name: "people".to_string(),
+ schema: Arc::new(schema),
+ projection: Some(vec![0, 1, 4]),
+ });
+
+ // prove that a plan can be passed to a thread
+ let plan1 = plan.clone();
+ thread::spawn(move || {
+ println!("plan: {:?}", plan1);
+ });
+ }
#[test]
fn serialize_plan() {
diff --git a/rust/datafusion/src/optimizer/optimizer.rs b/rust/datafusion/src/optimizer/optimizer.rs
index 9ab8e8c..f4e368c 100644
--- a/rust/datafusion/src/optimizer/optimizer.rs
+++ b/rust/datafusion/src/optimizer/optimizer.rs
@@ -19,9 +19,9 @@
use crate::logicalplan::LogicalPlan;
use arrow::error::Result;
-use std::rc::Rc;
+use std::sync::Arc;
/// An optimizer rules performs a transformation on a logical plan to produce an optimized logical plan.
pub trait OptimizerRule {
- fn optimize(&mut self, plan: &LogicalPlan) -> Result<Rc<LogicalPlan>>;
+ fn optimize(&mut self, plan: &LogicalPlan) -> Result<Arc<LogicalPlan>>;
}
diff --git a/rust/datafusion/src/optimizer/projection_push_down.rs b/rust/datafusion/src/optimizer/projection_push_down.rs
index b8d98fe..a9d2787 100644
--- a/rust/datafusion/src/optimizer/projection_push_down.rs
+++ b/rust/datafusion/src/optimizer/projection_push_down.rs
@@ -24,7 +24,6 @@ use crate::optimizer::optimizer::OptimizerRule;
use arrow::datatypes::{Field, Schema};
use arrow::error::{ArrowError, Result};
use std::collections::{HashMap, HashSet};
-use std::rc::Rc;
use std::sync::Arc;
/// Projection Push Down optimizer rule ensures that only referenced columns are
@@ -32,7 +31,7 @@ use std::sync::Arc;
pub struct ProjectionPushDown {}
impl OptimizerRule for ProjectionPushDown {
- fn optimize(&mut self, plan: &LogicalPlan) -> Result<Rc<LogicalPlan>> {
+ fn optimize(&mut self, plan: &LogicalPlan) -> Result<Arc<LogicalPlan>> {
let mut accum: HashSet<usize> = HashSet::new();
let mut mapping: HashMap<usize, usize> = HashMap::new();
self.optimize_plan(plan, &mut accum, &mut mapping)
@@ -49,7 +48,7 @@ impl ProjectionPushDown {
plan: &LogicalPlan,
accum: &mut HashSet<usize>,
mapping: &mut HashMap<usize, usize>,
- ) -> Result<Rc<LogicalPlan>> {
+ ) -> Result<Arc<LogicalPlan>> {
match plan {
LogicalPlan::Projection {
expr,
@@ -65,7 +64,7 @@ impl ProjectionPushDown {
// rewrite projection expressions to use new column indexes
let new_expr = self.rewrite_exprs(expr, mapping)?;
- Ok(Rc::new(LogicalPlan::Projection {
+ Ok(Arc::new(LogicalPlan::Projection {
expr: new_expr,
input,
schema: schema.clone(),
@@ -81,7 +80,7 @@ impl ProjectionPushDown {
// rewrite filter expression to use new column indexes
let new_expr = self.rewrite_expr(expr, mapping)?;
- Ok(Rc::new(LogicalPlan::Selection {
+ Ok(Arc::new(LogicalPlan::Selection {
expr: new_expr,
input,
}))
@@ -103,7 +102,7 @@ impl ProjectionPushDown {
let new_group_expr = self.rewrite_exprs(group_expr, mapping)?;
let new_aggr_expr = self.rewrite_exprs(aggr_expr, mapping)?;
- Ok(Rc::new(LogicalPlan::Aggregate {
+ Ok(Arc::new(LogicalPlan::Aggregate {
input,
group_expr: new_group_expr,
aggr_expr: new_aggr_expr,
@@ -124,14 +123,14 @@ impl ProjectionPushDown {
// rewrite sort expressions to use new column indexes
let new_expr = self.rewrite_exprs(expr, mapping)?;
- Ok(Rc::new(LogicalPlan::Sort {
+ Ok(Arc::new(LogicalPlan::Sort {
expr: new_expr,
input,
schema: schema.clone(),
}))
}
LogicalPlan::EmptyRelation { schema } => {
- Ok(Rc::new(LogicalPlan::EmptyRelation {
+ Ok(Arc::new(LogicalPlan::EmptyRelation {
schema: schema.clone(),
}))
}
@@ -172,7 +171,7 @@ impl ProjectionPushDown {
}
// return the table scan with projection
- Ok(Rc::new(LogicalPlan::TableScan {
+ Ok(Arc::new(LogicalPlan::TableScan {
schema_name: schema_name.to_string(),
table_name: table_name.to_string(),
schema: Arc::new(projected_schema),
@@ -183,7 +182,7 @@ impl ProjectionPushDown {
expr,
input,
schema,
- } => Ok(Rc::new(LogicalPlan::Limit {
+ } => Ok(Arc::new(LogicalPlan::Limit {
expr: expr.clone(),
input: input.clone(),
schema: schema.clone(),
@@ -229,21 +228,21 @@ impl ProjectionPushDown {
match expr {
Expr::Column(i) => Ok(Expr::Column(self.new_index(mapping, i)?)),
Expr::Literal(_) => Ok(expr.clone()),
- Expr::IsNull(e) => Ok(Expr::IsNull(Rc::new(self.rewrite_expr(e, mapping)?))),
+ Expr::IsNull(e) => Ok(Expr::IsNull(Arc::new(self.rewrite_expr(e, mapping)?))),
Expr::IsNotNull(e) => {
- Ok(Expr::IsNotNull(Rc::new(self.rewrite_expr(e, mapping)?)))
+ Ok(Expr::IsNotNull(Arc::new(self.rewrite_expr(e, mapping)?)))
}
Expr::BinaryExpr { left, op, right } => Ok(Expr::BinaryExpr {
- left: Rc::new(self.rewrite_expr(left, mapping)?),
+ left: Arc::new(self.rewrite_expr(left, mapping)?),
op: op.clone(),
- right: Rc::new(self.rewrite_expr(right, mapping)?),
+ right: Arc::new(self.rewrite_expr(right, mapping)?),
}),
Expr::Cast { expr, data_type } => Ok(Expr::Cast {
- expr: Rc::new(self.rewrite_expr(expr, mapping)?),
+ expr: Arc::new(self.rewrite_expr(expr, mapping)?),
data_type: data_type.clone(),
}),
Expr::Sort { expr, asc } => Ok(Expr::Sort {
- expr: Rc::new(self.rewrite_expr(expr, mapping)?),
+ expr: Arc::new(self.rewrite_expr(expr, mapping)?),
asc: *asc,
}),
Expr::AggregateFunction {
@@ -285,7 +284,6 @@ mod tests {
use crate::logicalplan::LogicalPlan::*;
use arrow::datatypes::{DataType, Field, Schema};
use std::borrow::Borrow;
- use std::rc::Rc;
use std::sync::Arc;
#[test]
@@ -300,7 +298,7 @@ mod tests {
DataType::UInt32,
false,
)])),
- input: Rc::new(table_scan),
+ input: Arc::new(table_scan),
};
assert_optimized_plan_eq(&aggregate, "Aggregate: groupBy=[[]], aggr=[[#0]]\n TableScan: test projection=Some([1])");
@@ -317,7 +315,7 @@ mod tests {
Field::new("c", DataType::UInt32, false),
Field::new("MAX(b)", DataType::UInt32, false),
])),
- input: Rc::new(table_scan),
+ input: Arc::new(table_scan),
};
assert_optimized_plan_eq(&aggregate, "Aggregate: groupBy=[[#1]], aggr=[[#0]]\n TableScan: test projection=Some([1, 2])");
@@ -329,7 +327,7 @@ mod tests {
let selection = Selection {
expr: Column(2),
- input: Rc::new(table_scan),
+ input: Arc::new(table_scan),
};
let aggregate = Aggregate {
@@ -340,7 +338,7 @@ mod tests {
DataType::UInt32,
false,
)])),
- input: Rc::new(selection),
+ input: Arc::new(selection),
};
assert_optimized_plan_eq(&aggregate, "Aggregate: groupBy=[[]], aggr=[[#0]]\n Selection: #1\n TableScan: test projection=Some([1, 2])");
@@ -352,10 +350,10 @@ mod tests {
let projection = Projection {
expr: vec![Cast {
- expr: Rc::new(Column(2)),
+ expr: Arc::new(Column(2)),
data_type: DataType::Float64,
}],
- input: Rc::new(table_scan),
+ input: Arc::new(table_scan),
schema: Arc::new(Schema::new(vec![Field::new(
"CAST(c AS float)",
DataType::Float64,
@@ -376,7 +374,7 @@ mod tests {
let projection = Projection {
expr: vec![Column(0), Column(1)],
- input: Rc::new(table_scan),
+ input: Arc::new(table_scan),
schema: Arc::new(Schema::new(vec![
Field::new("a", DataType::UInt32, false),
Field::new("b", DataType::UInt32, false),
@@ -403,7 +401,7 @@ mod tests {
assert_eq!(formatted_plan, expected);
}
- fn optimize(plan: &LogicalPlan) -> Rc<LogicalPlan> {
+ fn optimize(plan: &LogicalPlan) -> Arc<LogicalPlan> {
let mut rule = ProjectionPushDown::new();
rule.optimize(plan).unwrap()
}
diff --git a/rust/datafusion/src/sqlplanner.rs b/rust/datafusion/src/sqlplanner.rs
index a45d5a4..cc98806 100644
--- a/rust/datafusion/src/sqlplanner.rs
+++ b/rust/datafusion/src/sqlplanner.rs
@@ -18,7 +18,6 @@
//! SQL Query Planner (produces logical plan from SQL AST)
use std::collections::HashSet;
-use std::rc::Rc;
use std::string::String;
use std::sync::Arc;
@@ -36,17 +35,17 @@ pub trait SchemaProvider {
/// SQL query planner
pub struct SqlToRel {
- schema_provider: Rc<SchemaProvider>,
+ schema_provider: Arc<SchemaProvider>,
}
impl SqlToRel {
/// Create a new query planner
- pub fn new(schema_provider: Rc<SchemaProvider>) -> Self {
+ pub fn new(schema_provider: Arc<SchemaProvider>) -> Self {
SqlToRel { schema_provider }
}
/// Generate a logic plan from a SQL AST node
- pub fn sql_to_rel(&self, sql: &ASTNode) -> Result<Rc<LogicalPlan>> {
+ pub fn sql_to_rel(&self, sql: &ASTNode) -> Result<Arc<LogicalPlan>> {
match sql {
&ASTNode::SQLSelect {
ref projection,
@@ -61,7 +60,7 @@ impl SqlToRel {
// parse the input relation so we have access to the row type
let input = match relation {
&Some(ref r) => self.sql_to_rel(r)?,
- &None => Rc::new(LogicalPlan::EmptyRelation {
+ &None => Arc::new(LogicalPlan::EmptyRelation {
schema: Arc::new(Schema::empty()),
}),
};
@@ -93,8 +92,8 @@ impl SqlToRel {
.collect();
if aggr_expr.len() > 0 {
- let aggregate_input: Rc<LogicalPlan> = match selection_plan {
- Some(s) => Rc::new(s),
+ let aggregate_input: Arc<LogicalPlan> = match selection_plan {
+ Some(s) => Arc::new(s),
_ => input.clone(),
};
@@ -114,15 +113,15 @@ impl SqlToRel {
Schema::new(exprlist_to_fields(&all_fields, input_schema));
//TODO: selection, projection, everything else
- Ok(Rc::new(LogicalPlan::Aggregate {
+ Ok(Arc::new(LogicalPlan::Aggregate {
input: aggregate_input,
group_expr,
aggr_expr,
schema: Arc::new(aggr_schema),
}))
} else {
- let projection_input: Rc<LogicalPlan> = match selection_plan {
- Some(s) => Rc::new(s),
+ let projection_input: Arc<LogicalPlan> = match selection_plan {
+ Some(s) => Arc::new(s),
_ => input.clone(),
};
@@ -150,7 +149,7 @@ impl SqlToRel {
.iter()
.map(|e| {
Ok(Expr::Sort {
- expr: Rc::new(
+ expr: Arc::new(
self.sql_to_rex(&e.expr, &input_schema)
.unwrap(),
),
@@ -161,7 +160,7 @@ impl SqlToRel {
LogicalPlan::Sort {
expr: order_by_rex?,
- input: Rc::new(projection.clone()),
+ input: Arc::new(projection.clone()),
schema: input_schema.clone(),
}
}
@@ -176,20 +175,20 @@ impl SqlToRel {
LogicalPlan::Limit {
expr: limit_rex,
- input: Rc::new(order_by_plan.clone()),
+ input: Arc::new(order_by_plan.clone()),
schema: input_schema.clone(),
}
}
_ => order_by_plan,
};
- Ok(Rc::new(limit_plan))
+ Ok(Arc::new(limit_plan))
}
}
&ASTNode::SQLIdentifier(ref id) => {
match self.schema_provider.get_table_meta(id.as_ref()) {
- Some(schema) => Ok(Rc::new(LogicalPlan::TableScan {
+ Some(schema) => Ok(Arc::new(LogicalPlan::TableScan {
schema_name: String::from("default"),
table_name: id.clone(),
schema: schema.clone(),
@@ -219,7 +218,7 @@ impl SqlToRel {
Ok(Expr::Literal(ScalarValue::Float64(n)))
}
&ASTNode::SQLValue(sqlparser::sqlast::Value::SingleQuotedString(ref s)) => {
- Ok(Expr::Literal(ScalarValue::Utf8(Rc::new(s.clone()))))
+ Ok(Expr::Literal(ScalarValue::Utf8(Arc::new(s.clone()))))
}
&ASTNode::SQLIdentifier(ref id) => {
@@ -243,16 +242,16 @@ impl SqlToRel {
ref expr,
ref data_type,
} => Ok(Expr::Cast {
- expr: Rc::new(self.sql_to_rex(&expr, schema)?),
+ expr: Arc::new(self.sql_to_rex(&expr, schema)?),
data_type: convert_data_type(data_type)?,
}),
&ASTNode::SQLIsNull(ref expr) => {
- Ok(Expr::IsNull(Rc::new(self.sql_to_rex(expr, schema)?)))
+ Ok(Expr::IsNull(Arc::new(self.sql_to_rex(expr, schema)?)))
}
&ASTNode::SQLIsNotNull(ref expr) => {
- Ok(Expr::IsNotNull(Rc::new(self.sql_to_rex(expr, schema)?)))
+ Ok(Expr::IsNotNull(Arc::new(self.sql_to_rex(expr, schema)?)))
}
&ASTNode::SQLBinaryExpr {
@@ -286,9 +285,9 @@ impl SqlToRel {
match get_supertype(&left_type, &right_type) {
Some(supertype) => Ok(Expr::BinaryExpr {
- left: Rc::new(left_expr.cast_to(&supertype, schema)?),
+ left: Arc::new(left_expr.cast_to(&supertype, schema)?),
op: operator,
- right: Rc::new(right_expr.cast_to(&supertype, schema)?),
+ right: Arc::new(right_expr.cast_to(&supertype, schema)?),
}),
None => {
return Err(ExecutionError::General(format!(
@@ -301,7 +300,7 @@ impl SqlToRel {
}
// &ASTNode::SQLOrderBy { ref expr, asc } => Ok(Expr::Sort {
- // expr: Rc::new(self.sql_to_rex(&expr, &schema)?),
+ // expr: Arc::new(self.sql_to_rex(&expr, &schema)?),
// asc,
// }),
&ASTNode::SQLFunction { ref id, ref args } => {
@@ -463,9 +462,9 @@ fn collect_expr(e: &Expr, accum: &mut HashSet<usize>) {
}
pub fn push_down_projection(
- plan: &Rc<LogicalPlan>,
+ plan: &Arc<LogicalPlan>,
projection: &HashSet<usize>,
-) -> Rc<LogicalPlan> {
+) -> Arc<LogicalPlan> {
//println!("push_down_projection() projection={:?}", projection);
match plan.as_ref() {
LogicalPlan::Aggregate {
@@ -478,7 +477,7 @@ pub fn push_down_projection(
let mut accum: HashSet<usize> = HashSet::new();
group_expr.iter().for_each(|e| collect_expr(e, &mut accum));
aggr_expr.iter().for_each(|e| collect_expr(e, &mut accum));
- Rc::new(LogicalPlan::Aggregate {
+ Arc::new(LogicalPlan::Aggregate {
input: push_down_projection(&input, &accum),
group_expr: group_expr.clone(),
aggr_expr: aggr_expr.clone(),
@@ -491,7 +490,7 @@ pub fn push_down_projection(
} => {
let mut accum: HashSet<usize> = projection.clone();
collect_expr(expr, &mut accum);
- Rc::new(LogicalPlan::Selection {
+ Arc::new(LogicalPlan::Selection {
expr: expr.clone(),
input: push_down_projection(&input, &accum),
})
@@ -501,7 +500,7 @@ pub fn push_down_projection(
ref table_name,
ref schema,
..
- } => Rc::new(LogicalPlan::TableScan {
+ } => Arc::new(LogicalPlan::TableScan {
schema_name: schema_name.to_string(),
table_name: table_name.to_string(),
schema: schema.clone(),
@@ -646,14 +645,14 @@ mod tests {
let mut accum: HashSet<usize> = HashSet::new();
collect_expr(
&Expr::Cast {
- expr: Rc::new(Expr::Column(3)),
+ expr: Arc::new(Expr::Column(3)),
data_type: DataType::Float64,
},
&mut accum,
);
collect_expr(
&Expr::Cast {
- expr: Rc::new(Expr::Column(3)),
+ expr: Arc::new(Expr::Column(3)),
data_type: DataType::Float64,
},
&mut accum,
@@ -667,7 +666,7 @@ mod tests {
fn quick_test(sql: &str, expected: &str) {
use sqlparser::dialect::*;
let dialect = GenericSqlDialect {};
- let planner = SqlToRel::new(Rc::new(MockSchemaProvider {}));
+ let planner = SqlToRel::new(Arc::new(MockSchemaProvider {}));
let ast = Parser::parse_sql(&dialect, sql.to_string()).unwrap();
let plan = planner.sql_to_rel(&ast).unwrap();
assert_eq!(expected, format!("{:?}", plan));