You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@arrow.apache.org by GitBox <gi...@apache.org> on 2021/06/14 08:01:35 UTC
[GitHub] [arrow-datafusion] houqp commented on a change in pull request #55: Support qualified columns in queries
houqp commented on a change in pull request #55:
URL: https://github.com/apache/arrow-datafusion/pull/55#discussion_r650435911
##########
File path: ballista/rust/core/src/serde/logical_plan/from_proto.rs
##########
@@ -1361,43 +1224,3 @@ impl TryFrom<protobuf::WindowFrame> for WindowFrame {
})
}
}
-
-impl From<protobuf::AggregateFunction> for AggregateFunction {
- fn from(aggr_function: protobuf::AggregateFunction) -> Self {
- match aggr_function {
- protobuf::AggregateFunction::Min => AggregateFunction::Min,
- protobuf::AggregateFunction::Max => AggregateFunction::Max,
- protobuf::AggregateFunction::Sum => AggregateFunction::Sum,
- protobuf::AggregateFunction::Avg => AggregateFunction::Avg,
- protobuf::AggregateFunction::Count => AggregateFunction::Count,
- }
- }
-}
-
-impl From<protobuf::BuiltInWindowFunction> for BuiltInWindowFunction {
- fn from(built_in_function: protobuf::BuiltInWindowFunction) -> Self {
- match built_in_function {
- protobuf::BuiltInWindowFunction::RowNumber => {
- BuiltInWindowFunction::RowNumber
- }
- protobuf::BuiltInWindowFunction::Rank => BuiltInWindowFunction::Rank,
- protobuf::BuiltInWindowFunction::PercentRank => {
- BuiltInWindowFunction::PercentRank
- }
- protobuf::BuiltInWindowFunction::DenseRank => {
- BuiltInWindowFunction::DenseRank
- }
- protobuf::BuiltInWindowFunction::Lag => BuiltInWindowFunction::Lag,
- protobuf::BuiltInWindowFunction::Lead => BuiltInWindowFunction::Lead,
- protobuf::BuiltInWindowFunction::FirstValue => {
- BuiltInWindowFunction::FirstValue
- }
- protobuf::BuiltInWindowFunction::CumeDist => BuiltInWindowFunction::CumeDist,
- protobuf::BuiltInWindowFunction::Ntile => BuiltInWindowFunction::Ntile,
- protobuf::BuiltInWindowFunction::NthValue => BuiltInWindowFunction::NthValue,
- protobuf::BuiltInWindowFunction::LastValue => {
- BuiltInWindowFunction::LastValue
- }
- }
- }
-}
Review comment:
all these shared code are moved into `ballista/rust/core/src/serde/mod.rs`.
##########
File path: datafusion/src/logical_plan/builder.rs
##########
@@ -210,69 +238,103 @@ impl LogicalPlanBuilder {
}
/// Apply a sort
- pub fn sort(&self, expr: impl IntoIterator<Item = Expr>) -> Result<Self> {
+ pub fn sort(&self, exprs: impl IntoIterator<Item = Expr>) -> Result<Self> {
+ let schemas = self.plan.all_schemas();
Ok(Self::from(&LogicalPlan::Sort {
- expr: expr.into_iter().collect(),
+ expr: normalize_cols(exprs, &schemas)?,
input: Arc::new(self.plan.clone()),
}))
}
/// Apply a union
pub fn union(&self, plan: LogicalPlan) -> Result<Self> {
- let schema = self.plan.schema();
Review comment:
refactored implementation into `union_with_alias` function so the same logic can be reused in sql planner.
##########
File path: datafusion/src/physical_plan/planner.rs
##########
@@ -56,6 +56,121 @@ use expressions::col;
use log::debug;
use std::sync::Arc;
+fn create_function_physical_name(
+ fun: &str,
+ distinct: bool,
+ args: &[Expr],
+ input_schema: &DFSchema,
+) -> Result<String> {
+ let names: Vec<String> = args
+ .iter()
+ .map(|e| physical_name(e, input_schema))
+ .collect::<Result<_>>()?;
+
+ let distinct_str = match distinct {
+ true => "DISTINCT ",
+ false => "",
+ };
+ Ok(format!("{}({}{})", fun, distinct_str, names.join(",")))
+}
+
+fn physical_name(e: &Expr, input_schema: &DFSchema) -> Result<String> {
Review comment:
This function is responsible for generating physical field name for a given expression based on https://github.com/apache/arrow-datafusion/blob/master/docs/specification/output-field-name-semantic.md.
I kept `Expr.name` as is for better readability in query explain and debug output.
##########
File path: datafusion/src/physical_plan/planner.rs
##########
@@ -501,17 +684,20 @@ impl DefaultPhysicalPlanner {
pub fn create_physical_expr(
&self,
e: &Expr,
+ input_dfschema: &DFSchema,
Review comment:
`create_physical_expr` now requires logical plan schema for column index lookup
##########
File path: datafusion/src/sql/planner.rs
##########
@@ -424,17 +408,24 @@ impl<'a, S: ContextProvider> SqlToRel<'a, S> {
ctes: &mut HashMap<String, LogicalPlan>,
) -> Result<LogicalPlan> {
match relation {
- TableFactor::Table { name, .. } => {
+ TableFactor::Table { name, alias, .. } => {
let table_name = name.to_string();
let cte = ctes.get(&table_name);
match (
cte,
self.schema_provider.get_table_provider(name.try_into()?),
) {
(Some(cte_plan), _) => Ok(cte_plan.clone()),
- (_, Some(provider)) => {
- LogicalPlanBuilder::scan(&table_name, provider, None)?.build()
- }
+ (_, Some(provider)) => LogicalPlanBuilder::scan(
+ // take alias into account to support `JOIN table1 as table2`
+ alias
+ .as_ref()
+ .map(|a| a.name.value.as_str())
+ .unwrap_or(&table_name),
+ provider,
+ None,
+ )?
+ .build(),
Review comment:
this change has been split out into https://github.com/apache/arrow-datafusion/pull/547 for easier review.
##########
File path: datafusion/src/logical_plan/expr.rs
##########
@@ -1240,7 +1394,7 @@ impl fmt::Debug for Expr {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
match self {
Expr::Alias(expr, alias) => write!(f, "{:?} AS {}", expr, alias),
- Expr::Column(name) => write!(f, "#{}", name),
+ Expr::Column(c) => write!(f, "{}", c),
Review comment:
`#` prefix is now provided by Column's `fmt` method.
##########
File path: datafusion/src/physical_plan/expressions/column.rs
##########
@@ -61,26 +68,21 @@ impl PhysicalExpr for Column {
/// Get the data type of this expression, given the schema of the input
fn data_type(&self, input_schema: &Schema) -> Result<DataType> {
- Ok(input_schema
- .field_with_name(&self.name)?
- .data_type()
- .clone())
+ Ok(input_schema.field(self.index).data_type().clone())
}
/// Decide whehter this expression is nullable, given the schema of the input
fn nullable(&self, input_schema: &Schema) -> Result<bool> {
- Ok(input_schema.field_with_name(&self.name)?.is_nullable())
+ Ok(input_schema.field(self.index).is_nullable())
}
/// Evaluate the expression
fn evaluate(&self, batch: &RecordBatch) -> Result<ColumnarValue> {
- Ok(ColumnarValue::Array(
- batch.column(batch.schema().index_of(&self.name)?).clone(),
- ))
+ Ok(ColumnarValue::Array(batch.column(self.index).clone()))
Review comment:
Physical column expression evaluation now indexes record batch column directly instead of doing string comparisons on all field names.
##########
File path: datafusion/src/physical_plan/functions.rs
##########
@@ -512,39 +512,35 @@ macro_rules! invoke_if_unicode_expressions_feature_flag {
};
}
-/// Create a physical (function) expression.
-/// This function errors when `args`' can't be coerced to a valid argument type of the function.
-pub fn create_physical_expr(
+/// Create a physical scalar function.
Review comment:
this change has been split out into https://github.com/apache/arrow-datafusion/pull/548 for easier review.
##########
File path: ballista/rust/core/src/serde/logical_plan/from_proto.rs
##########
@@ -1361,43 +1224,3 @@ impl TryFrom<protobuf::WindowFrame> for WindowFrame {
})
}
}
-
-impl From<protobuf::AggregateFunction> for AggregateFunction {
- fn from(aggr_function: protobuf::AggregateFunction) -> Self {
- match aggr_function {
- protobuf::AggregateFunction::Min => AggregateFunction::Min,
- protobuf::AggregateFunction::Max => AggregateFunction::Max,
- protobuf::AggregateFunction::Sum => AggregateFunction::Sum,
- protobuf::AggregateFunction::Avg => AggregateFunction::Avg,
- protobuf::AggregateFunction::Count => AggregateFunction::Count,
- }
- }
-}
-
-impl From<protobuf::BuiltInWindowFunction> for BuiltInWindowFunction {
- fn from(built_in_function: protobuf::BuiltInWindowFunction) -> Self {
- match built_in_function {
- protobuf::BuiltInWindowFunction::RowNumber => {
- BuiltInWindowFunction::RowNumber
- }
- protobuf::BuiltInWindowFunction::Rank => BuiltInWindowFunction::Rank,
- protobuf::BuiltInWindowFunction::PercentRank => {
- BuiltInWindowFunction::PercentRank
- }
- protobuf::BuiltInWindowFunction::DenseRank => {
- BuiltInWindowFunction::DenseRank
- }
- protobuf::BuiltInWindowFunction::Lag => BuiltInWindowFunction::Lag,
- protobuf::BuiltInWindowFunction::Lead => BuiltInWindowFunction::Lead,
- protobuf::BuiltInWindowFunction::FirstValue => {
- BuiltInWindowFunction::FirstValue
- }
- protobuf::BuiltInWindowFunction::CumeDist => BuiltInWindowFunction::CumeDist,
- protobuf::BuiltInWindowFunction::Ntile => BuiltInWindowFunction::Ntile,
- protobuf::BuiltInWindowFunction::NthValue => BuiltInWindowFunction::NthValue,
- protobuf::BuiltInWindowFunction::LastValue => {
- BuiltInWindowFunction::LastValue
- }
- }
- }
-}
Review comment:
all these shared code are moved into `ballista/rust/core/src/serde/mod.rs` so it can be shared with physical plan ser/de.
##########
File path: ballista/rust/core/proto/ballista.proto
##########
@@ -408,6 +421,119 @@ message PhysicalPlanNode {
}
}
+// physical expressions
Review comment:
Physical expressions are now serialized and deserialized without involving logical expressions. It is needed because logical columns and physical columns are represented differently, one with relation qualifier and the other one with column index. This should also speed up the deserialization for physical plane because we won't need to run physical expressions through physical planner anymore.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org