You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@arrow.apache.org by al...@apache.org on 2023/11/09 14:18:11 UTC
(arrow-datafusion) branch main updated: Projection Pushdown rule and test changes (#8073)
This is an automated email from the ASF dual-hosted git repository.
alamb pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/arrow-datafusion.git
The following commit(s) were added to refs/heads/main by this push:
new 1c17c47b4e Projection Pushdown rule and test changes (#8073)
1c17c47b4e is described below
commit 1c17c47b4e08754c8c19b4cf578a34d2c9249e30
Author: Berkay Şahin <12...@users.noreply.github.com>
AuthorDate: Thu Nov 9 17:18:05 2023 +0300
Projection Pushdown rule and test changes (#8073)
---
datafusion/core/src/physical_optimizer/mod.rs | 1 +
.../core/src/physical_optimizer/optimizer.rs | 8 +
.../src/physical_optimizer/output_requirements.rs | 6 +-
.../src/physical_optimizer/projection_pushdown.rs | 2181 ++++++++++++++++++++
datafusion/physical-expr/src/expressions/cast.rs | 5 +
datafusion/physical-expr/src/scalar_function.rs | 5 +
.../physical-plan/src/joins/symmetric_hash_join.rs | 5 +
datafusion/physical-plan/src/joins/utils.rs | 2 +-
datafusion/physical-plan/src/memory.rs | 12 +
datafusion/physical-plan/src/projection.rs | 2 +-
datafusion/sqllogictest/test_files/explain.slt | 6 +-
datafusion/sqllogictest/test_files/groupby.slt | 27 +-
datafusion/sqllogictest/test_files/insert.slt | 5 +-
.../sqllogictest/test_files/insert_to_external.slt | 5 +-
datafusion/sqllogictest/test_files/joins.slt | 191 +-
datafusion/sqllogictest/test_files/subquery.slt | 77 +-
datafusion/sqllogictest/test_files/union.slt | 60 +-
17 files changed, 2395 insertions(+), 203 deletions(-)
diff --git a/datafusion/core/src/physical_optimizer/mod.rs b/datafusion/core/src/physical_optimizer/mod.rs
index 9e22bff340..d2a0c6fefd 100644
--- a/datafusion/core/src/physical_optimizer/mod.rs
+++ b/datafusion/core/src/physical_optimizer/mod.rs
@@ -30,6 +30,7 @@ pub mod join_selection;
pub mod optimizer;
pub mod output_requirements;
pub mod pipeline_checker;
+mod projection_pushdown;
pub mod pruning;
pub mod replace_with_order_preserving_variants;
mod sort_pushdown;
diff --git a/datafusion/core/src/physical_optimizer/optimizer.rs b/datafusion/core/src/physical_optimizer/optimizer.rs
index 95035e5f81..20a59b58ea 100644
--- a/datafusion/core/src/physical_optimizer/optimizer.rs
+++ b/datafusion/core/src/physical_optimizer/optimizer.rs
@@ -19,6 +19,7 @@
use std::sync::Arc;
+use super::projection_pushdown::ProjectionPushdown;
use crate::config::ConfigOptions;
use crate::physical_optimizer::aggregate_statistics::AggregateStatistics;
use crate::physical_optimizer::coalesce_batches::CoalesceBatches;
@@ -107,6 +108,13 @@ impl PhysicalOptimizer {
// into an `order by max(x) limit y`. In this case it will copy the limit value down
// to the aggregation, allowing it to use only y number of accumulators.
Arc::new(TopKAggregation::new()),
+ // The ProjectionPushdown rule tries to push projections towards
+ // the sources in the execution plan. As a result of this process,
+ // a projection can disappear if it reaches the source providers, and
+ // sequential projections can merge into one. Even if these two cases
+ // are not present, the load of executors such as join or union will be
+ // reduced by narrowing their input tables.
+ Arc::new(ProjectionPushdown::new()),
];
Self::with_rules(rules)
diff --git a/datafusion/core/src/physical_optimizer/output_requirements.rs b/datafusion/core/src/physical_optimizer/output_requirements.rs
index d9cdc292dd..f8bf3bb965 100644
--- a/datafusion/core/src/physical_optimizer/output_requirements.rs
+++ b/datafusion/core/src/physical_optimizer/output_requirements.rs
@@ -88,14 +88,14 @@ enum RuleMode {
///
/// See [`OutputRequirements`] for more details
#[derive(Debug)]
-struct OutputRequirementExec {
+pub(crate) struct OutputRequirementExec {
input: Arc<dyn ExecutionPlan>,
order_requirement: Option<LexRequirement>,
dist_requirement: Distribution,
}
impl OutputRequirementExec {
- fn new(
+ pub(crate) fn new(
input: Arc<dyn ExecutionPlan>,
requirements: Option<LexRequirement>,
dist_requirement: Distribution,
@@ -107,7 +107,7 @@ impl OutputRequirementExec {
}
}
- fn input(&self) -> Arc<dyn ExecutionPlan> {
+ pub(crate) fn input(&self) -> Arc<dyn ExecutionPlan> {
self.input.clone()
}
}
diff --git a/datafusion/core/src/physical_optimizer/projection_pushdown.rs b/datafusion/core/src/physical_optimizer/projection_pushdown.rs
new file mode 100644
index 0000000000..1849595561
--- /dev/null
+++ b/datafusion/core/src/physical_optimizer/projection_pushdown.rs
@@ -0,0 +1,2181 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you 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.
+
+//! This file implements the `ProjectionPushdown` physical optimization rule.
+//! The function [`remove_unnecessary_projections`] tries to push down all
+//! projections one by one if the operator below is amenable to this. If a
+//! projection reaches a source, it can even dissappear from the plan entirely.
+
+use super::output_requirements::OutputRequirementExec;
+use super::PhysicalOptimizerRule;
+use crate::datasource::physical_plan::CsvExec;
+use crate::error::Result;
+use crate::physical_plan::coalesce_partitions::CoalescePartitionsExec;
+use crate::physical_plan::filter::FilterExec;
+use crate::physical_plan::joins::utils::{ColumnIndex, JoinFilter};
+use crate::physical_plan::joins::{
+ CrossJoinExec, HashJoinExec, NestedLoopJoinExec, SortMergeJoinExec,
+ SymmetricHashJoinExec,
+};
+use crate::physical_plan::memory::MemoryExec;
+use crate::physical_plan::projection::ProjectionExec;
+use crate::physical_plan::repartition::RepartitionExec;
+use crate::physical_plan::sorts::sort::SortExec;
+use crate::physical_plan::sorts::sort_preserving_merge::SortPreservingMergeExec;
+use crate::physical_plan::{Distribution, ExecutionPlan};
+
+use arrow_schema::SchemaRef;
+
+use datafusion_common::config::ConfigOptions;
+use datafusion_common::tree_node::{Transformed, TreeNode};
+use datafusion_common::JoinSide;
+use datafusion_physical_expr::expressions::{
+ BinaryExpr, CaseExpr, CastExpr, Column, Literal, NegativeExpr,
+};
+use datafusion_physical_expr::{
+ Partitioning, PhysicalExpr, PhysicalSortExpr, PhysicalSortRequirement,
+ ScalarFunctionExpr,
+};
+use datafusion_physical_plan::union::UnionExec;
+
+use itertools::Itertools;
+use std::sync::Arc;
+
+/// This rule inspects [`ProjectionExec`]'s in the given physical plan and tries to
+/// remove or swap with its child.
+#[derive(Default)]
+pub struct ProjectionPushdown {}
+
+impl ProjectionPushdown {
+ #[allow(missing_docs)]
+ pub fn new() -> Self {
+ Self {}
+ }
+}
+
+impl PhysicalOptimizerRule for ProjectionPushdown {
+ fn optimize(
+ &self,
+ plan: Arc<dyn ExecutionPlan>,
+ _config: &ConfigOptions,
+ ) -> Result<Arc<dyn ExecutionPlan>> {
+ plan.transform_down(&remove_unnecessary_projections)
+ }
+
+ fn name(&self) -> &str {
+ "ProjectionPushdown"
+ }
+
+ fn schema_check(&self) -> bool {
+ true
+ }
+}
+
+/// This function checks if `plan` is a [`ProjectionExec`], and inspects its
+/// input(s) to test whether it can push `plan` under its input(s). This function
+/// will operate on the entire tree and may ultimately remove `plan` entirely
+/// by leveraging source providers with built-in projection capabilities.
+pub fn remove_unnecessary_projections(
+ plan: Arc<dyn ExecutionPlan>,
+) -> Result<Transformed<Arc<dyn ExecutionPlan>>> {
+ let maybe_modified = if let Some(projection) =
+ plan.as_any().downcast_ref::<ProjectionExec>()
+ {
+ // If the projection does not cause any change on the input, we can
+ // safely remove it:
+ if is_projection_removable(projection) {
+ return Ok(Transformed::Yes(projection.input().clone()));
+ }
+ // If it does, check if we can push it under its child(ren):
+ let input = projection.input().as_any();
+ if let Some(csv) = input.downcast_ref::<CsvExec>() {
+ try_swapping_with_csv(projection, csv)
+ } else if let Some(memory) = input.downcast_ref::<MemoryExec>() {
+ try_swapping_with_memory(projection, memory)?
+ } else if let Some(child_projection) = input.downcast_ref::<ProjectionExec>() {
+ let maybe_unified = try_unifying_projections(projection, child_projection)?;
+ return if let Some(new_plan) = maybe_unified {
+ // To unify 3 or more sequential projections:
+ remove_unnecessary_projections(new_plan)
+ } else {
+ Ok(Transformed::No(plan))
+ };
+ } else if let Some(output_req) = input.downcast_ref::<OutputRequirementExec>() {
+ try_swapping_with_output_req(projection, output_req)?
+ } else if input.is::<CoalescePartitionsExec>() {
+ try_swapping_with_coalesce_partitions(projection)?
+ } else if let Some(filter) = input.downcast_ref::<FilterExec>() {
+ try_swapping_with_filter(projection, filter)?
+ } else if let Some(repartition) = input.downcast_ref::<RepartitionExec>() {
+ try_swapping_with_repartition(projection, repartition)?
+ } else if let Some(sort) = input.downcast_ref::<SortExec>() {
+ try_swapping_with_sort(projection, sort)?
+ } else if let Some(spm) = input.downcast_ref::<SortPreservingMergeExec>() {
+ try_swapping_with_sort_preserving_merge(projection, spm)?
+ } else if let Some(union) = input.downcast_ref::<UnionExec>() {
+ try_pushdown_through_union(projection, union)?
+ } else if let Some(hash_join) = input.downcast_ref::<HashJoinExec>() {
+ try_pushdown_through_hash_join(projection, hash_join)?
+ } else if let Some(cross_join) = input.downcast_ref::<CrossJoinExec>() {
+ try_swapping_with_cross_join(projection, cross_join)?
+ } else if let Some(nl_join) = input.downcast_ref::<NestedLoopJoinExec>() {
+ try_swapping_with_nested_loop_join(projection, nl_join)?
+ } else if let Some(sm_join) = input.downcast_ref::<SortMergeJoinExec>() {
+ try_swapping_with_sort_merge_join(projection, sm_join)?
+ } else if let Some(sym_join) = input.downcast_ref::<SymmetricHashJoinExec>() {
+ try_swapping_with_sym_hash_join(projection, sym_join)?
+ } else {
+ // If the input plan of the projection is not one of the above, we
+ // conservatively assume that pushing the projection down may hurt.
+ // When adding new operators, consider adding them here if you
+ // think pushing projections under them is beneficial.
+ None
+ }
+ } else {
+ return Ok(Transformed::No(plan));
+ };
+
+ Ok(maybe_modified.map_or(Transformed::No(plan), Transformed::Yes))
+}
+
+/// Tries to swap `projection` with its input (`csv`). If possible, performs
+/// the swap and returns [`CsvExec`] as the top plan. Otherwise, returns `None`.
+fn try_swapping_with_csv(
+ projection: &ProjectionExec,
+ csv: &CsvExec,
+) -> Option<Arc<dyn ExecutionPlan>> {
+ // If there is any non-column or alias-carrier expression, Projection should not be removed.
+ // This process can be moved into CsvExec, but it would be an overlap of their responsibility.
+ all_alias_free_columns(projection.expr()).then(|| {
+ let mut file_scan = csv.base_config().clone();
+ let new_projections =
+ new_projections_for_columns(projection, &file_scan.projection);
+ file_scan.projection = Some(new_projections);
+
+ Arc::new(CsvExec::new(
+ file_scan,
+ csv.has_header(),
+ csv.delimiter(),
+ csv.quote(),
+ csv.escape(),
+ csv.file_compression_type,
+ )) as _
+ })
+}
+
+/// Tries to swap `projection` with its input (`memory`). If possible, performs
+/// the swap and returns [`MemoryExec`] as the top plan. Otherwise, returns `None`.
+fn try_swapping_with_memory(
+ projection: &ProjectionExec,
+ memory: &MemoryExec,
+) -> Result<Option<Arc<dyn ExecutionPlan>>> {
+ // If there is any non-column or alias-carrier expression, Projection should not be removed.
+ // This process can be moved into MemoryExec, but it would be an overlap of their responsibility.
+ all_alias_free_columns(projection.expr())
+ .then(|| {
+ let new_projections =
+ new_projections_for_columns(projection, memory.projection());
+
+ MemoryExec::try_new(
+ memory.partitions(),
+ memory.original_schema(),
+ Some(new_projections),
+ )
+ .map(|e| Arc::new(e) as _)
+ })
+ .transpose()
+}
+
+/// Unifies `projection` with its input (which is also a [`ProjectionExec`]).
+/// Two consecutive projections can always merge into a single projection unless
+/// the [`update_expr`] function does not support one of the expression
+/// types involved in the projection.
+fn try_unifying_projections(
+ projection: &ProjectionExec,
+ child: &ProjectionExec,
+) -> Result<Option<Arc<dyn ExecutionPlan>>> {
+ let mut projected_exprs = vec![];
+ for (expr, alias) in projection.expr() {
+ // If there is no match in the input projection, we cannot unify these
+ // projections. This case will arise if the projection expression contains
+ // a `PhysicalExpr` variant `update_expr` doesn't support.
+ let Some(expr) = update_expr(expr, child.expr(), true)? else {
+ return Ok(None);
+ };
+ projected_exprs.push((expr, alias.clone()));
+ }
+
+ ProjectionExec::try_new(projected_exprs, child.input().clone())
+ .map(|e| Some(Arc::new(e) as _))
+}
+
+/// Tries to swap `projection` with its input (`output_req`). If possible,
+/// performs the swap and returns [`OutputRequirementExec`] as the top plan.
+/// Otherwise, returns `None`.
+fn try_swapping_with_output_req(
+ projection: &ProjectionExec,
+ output_req: &OutputRequirementExec,
+) -> Result<Option<Arc<dyn ExecutionPlan>>> {
+ // If the projection does not narrow the the schema, we should not try to push it down:
+ if projection.expr().len() >= projection.input().schema().fields().len() {
+ return Ok(None);
+ }
+
+ let mut updated_sort_reqs = vec![];
+ // None or empty_vec can be treated in the same way.
+ if let Some(reqs) = &output_req.required_input_ordering()[0] {
+ for req in reqs {
+ let Some(new_expr) = update_expr(&req.expr, projection.expr(), false)? else {
+ return Ok(None);
+ };
+ updated_sort_reqs.push(PhysicalSortRequirement {
+ expr: new_expr,
+ options: req.options,
+ });
+ }
+ }
+
+ let dist_req = match &output_req.required_input_distribution()[0] {
+ Distribution::HashPartitioned(exprs) => {
+ let mut updated_exprs = vec![];
+ for expr in exprs {
+ let Some(new_expr) = update_expr(expr, projection.expr(), false)? else {
+ return Ok(None);
+ };
+ updated_exprs.push(new_expr);
+ }
+ Distribution::HashPartitioned(updated_exprs)
+ }
+ dist => dist.clone(),
+ };
+
+ make_with_child(projection, &output_req.input())
+ .map(|input| {
+ OutputRequirementExec::new(
+ input,
+ (!updated_sort_reqs.is_empty()).then_some(updated_sort_reqs),
+ dist_req,
+ )
+ })
+ .map(|e| Some(Arc::new(e) as _))
+}
+
+/// Tries to swap `projection` with its input, which is known to be a
+/// [`CoalescePartitionsExec`]. If possible, performs the swap and returns
+/// [`CoalescePartitionsExec`] as the top plan. Otherwise, returns `None`.
+fn try_swapping_with_coalesce_partitions(
+ projection: &ProjectionExec,
+) -> Result<Option<Arc<dyn ExecutionPlan>>> {
+ // If the projection does not narrow the the schema, we should not try to push it down:
+ if projection.expr().len() >= projection.input().schema().fields().len() {
+ return Ok(None);
+ }
+ // CoalescePartitionsExec always has a single child, so zero indexing is safe.
+ make_with_child(projection, &projection.input().children()[0])
+ .map(|e| Some(Arc::new(CoalescePartitionsExec::new(e)) as _))
+}
+
+/// Tries to swap `projection` with its input (`filter`). If possible, performs
+/// the swap and returns [`FilterExec`] as the top plan. Otherwise, returns `None`.
+fn try_swapping_with_filter(
+ projection: &ProjectionExec,
+ filter: &FilterExec,
+) -> Result<Option<Arc<dyn ExecutionPlan>>> {
+ // If the projection does not narrow the the schema, we should not try to push it down:
+ if projection.expr().len() >= projection.input().schema().fields().len() {
+ return Ok(None);
+ }
+ // Each column in the predicate expression must exist after the projection.
+ let Some(new_predicate) = update_expr(filter.predicate(), projection.expr(), false)?
+ else {
+ return Ok(None);
+ };
+
+ FilterExec::try_new(new_predicate, make_with_child(projection, filter.input())?)
+ .map(|e| Some(Arc::new(e) as _))
+}
+
+/// Tries to swap the projection with its input [`RepartitionExec`]. If it can be done,
+/// it returns the new swapped version having the [`RepartitionExec`] as the top plan.
+/// Otherwise, it returns None.
+fn try_swapping_with_repartition(
+ projection: &ProjectionExec,
+ repartition: &RepartitionExec,
+) -> Result<Option<Arc<dyn ExecutionPlan>>> {
+ // If the projection does not narrow the the schema, we should not try to push it down.
+ if projection.expr().len() >= projection.input().schema().fields().len() {
+ return Ok(None);
+ }
+
+ // If pushdown is not beneficial or applicable, break it.
+ if projection.benefits_from_input_partitioning()[0] || !all_columns(projection.expr())
+ {
+ return Ok(None);
+ }
+
+ let new_projection = make_with_child(projection, repartition.input())?;
+
+ let new_partitioning = match repartition.partitioning() {
+ Partitioning::Hash(partitions, size) => {
+ let mut new_partitions = vec![];
+ for partition in partitions {
+ let Some(new_partition) =
+ update_expr(partition, projection.expr(), false)?
+ else {
+ return Ok(None);
+ };
+ new_partitions.push(new_partition);
+ }
+ Partitioning::Hash(new_partitions, *size)
+ }
+ others => others.clone(),
+ };
+
+ Ok(Some(Arc::new(RepartitionExec::try_new(
+ new_projection,
+ new_partitioning,
+ )?)))
+}
+
+/// Tries to swap the projection with its input [`SortExec`]. If it can be done,
+/// it returns the new swapped version having the [`SortExec`] as the top plan.
+/// Otherwise, it returns None.
+fn try_swapping_with_sort(
+ projection: &ProjectionExec,
+ sort: &SortExec,
+) -> Result<Option<Arc<dyn ExecutionPlan>>> {
+ // If the projection does not narrow the the schema, we should not try to push it down.
+ if projection.expr().len() >= projection.input().schema().fields().len() {
+ return Ok(None);
+ }
+
+ let mut updated_exprs = vec![];
+ for sort in sort.expr() {
+ let Some(new_expr) = update_expr(&sort.expr, projection.expr(), false)? else {
+ return Ok(None);
+ };
+ updated_exprs.push(PhysicalSortExpr {
+ expr: new_expr,
+ options: sort.options,
+ });
+ }
+
+ Ok(Some(Arc::new(
+ SortExec::new(updated_exprs, make_with_child(projection, sort.input())?)
+ .with_fetch(sort.fetch()),
+ )))
+}
+
+/// Tries to swap the projection with its input [`SortPreservingMergeExec`].
+/// If this is possible, it returns the new [`SortPreservingMergeExec`] whose
+/// child is a projection. Otherwise, it returns None.
+fn try_swapping_with_sort_preserving_merge(
+ projection: &ProjectionExec,
+ spm: &SortPreservingMergeExec,
+) -> Result<Option<Arc<dyn ExecutionPlan>>> {
+ // If the projection does not narrow the the schema, we should not try to push it down.
+ if projection.expr().len() >= projection.input().schema().fields().len() {
+ return Ok(None);
+ }
+
+ let mut updated_exprs = vec![];
+ for sort in spm.expr() {
+ let Some(updated_expr) = update_expr(&sort.expr, projection.expr(), false)?
+ else {
+ return Ok(None);
+ };
+ updated_exprs.push(PhysicalSortExpr {
+ expr: updated_expr,
+ options: sort.options,
+ });
+ }
+
+ Ok(Some(Arc::new(
+ SortPreservingMergeExec::new(
+ updated_exprs,
+ make_with_child(projection, spm.input())?,
+ )
+ .with_fetch(spm.fetch()),
+ )))
+}
+
+/// Tries to push `projection` down through `union`. If possible, performs the
+/// pushdown and returns a new [`UnionExec`] as the top plan which has projections
+/// as its children. Otherwise, returns `None`.
+fn try_pushdown_through_union(
+ projection: &ProjectionExec,
+ union: &UnionExec,
+) -> Result<Option<Arc<dyn ExecutionPlan>>> {
+ // If the projection doesn't narrow the schema, we shouldn't try to push it down.
+ if projection.expr().len() >= projection.input().schema().fields().len() {
+ return Ok(None);
+ }
+
+ let new_children = union
+ .children()
+ .into_iter()
+ .map(|child| make_with_child(projection, &child))
+ .collect::<Result<Vec<_>>>()?;
+
+ Ok(Some(Arc::new(UnionExec::new(new_children))))
+}
+
+/// Tries to push `projection` down through `hash_join`. If possible, performs the
+/// pushdown and returns a new [`HashJoinExec`] as the top plan which has projections
+/// as its children. Otherwise, returns `None`.
+fn try_pushdown_through_hash_join(
+ projection: &ProjectionExec,
+ hash_join: &HashJoinExec,
+) -> Result<Option<Arc<dyn ExecutionPlan>>> {
+ // Convert projected expressions to columns. We can not proceed if this is
+ // not possible.
+ let Some(projection_as_columns) = physical_to_column_exprs(projection.expr()) else {
+ return Ok(None);
+ };
+
+ let (far_right_left_col_ind, far_left_right_col_ind) = join_table_borders(
+ hash_join.left().schema().fields().len(),
+ &projection_as_columns,
+ );
+
+ if !join_allows_pushdown(
+ &projection_as_columns,
+ hash_join.schema(),
+ far_right_left_col_ind,
+ far_left_right_col_ind,
+ ) {
+ return Ok(None);
+ }
+
+ let Some(new_on) = update_join_on(
+ &projection_as_columns[0..=far_right_left_col_ind as _],
+ &projection_as_columns[far_left_right_col_ind as _..],
+ hash_join.on(),
+ ) else {
+ return Ok(None);
+ };
+
+ let new_filter = if let Some(filter) = hash_join.filter() {
+ match update_join_filter(
+ &projection_as_columns[0..=far_right_left_col_ind as _],
+ &projection_as_columns[far_left_right_col_ind as _..],
+ filter,
+ hash_join.left(),
+ hash_join.right(),
+ ) {
+ Some(updated_filter) => Some(updated_filter),
+ None => return Ok(None),
+ }
+ } else {
+ None
+ };
+
+ let (new_left, new_right) = new_join_children(
+ projection_as_columns,
+ far_right_left_col_ind,
+ far_left_right_col_ind,
+ hash_join.left(),
+ hash_join.right(),
+ )?;
+
+ Ok(Some(Arc::new(HashJoinExec::try_new(
+ Arc::new(new_left),
+ Arc::new(new_right),
+ new_on,
+ new_filter,
+ hash_join.join_type(),
+ *hash_join.partition_mode(),
+ hash_join.null_equals_null,
+ )?)))
+}
+
+/// Tries to swap the projection with its input [`CrossJoinExec`]. If it can be done,
+/// it returns the new swapped version having the [`CrossJoinExec`] as the top plan.
+/// Otherwise, it returns None.
+fn try_swapping_with_cross_join(
+ projection: &ProjectionExec,
+ cross_join: &CrossJoinExec,
+) -> Result<Option<Arc<dyn ExecutionPlan>>> {
+ // Convert projected PhysicalExpr's to columns. If not possible, we cannot proceed.
+ let Some(projection_as_columns) = physical_to_column_exprs(projection.expr()) else {
+ return Ok(None);
+ };
+
+ let (far_right_left_col_ind, far_left_right_col_ind) = join_table_borders(
+ cross_join.left().schema().fields().len(),
+ &projection_as_columns,
+ );
+
+ if !join_allows_pushdown(
+ &projection_as_columns,
+ cross_join.schema(),
+ far_right_left_col_ind,
+ far_left_right_col_ind,
+ ) {
+ return Ok(None);
+ }
+
+ let (new_left, new_right) = new_join_children(
+ projection_as_columns,
+ far_right_left_col_ind,
+ far_left_right_col_ind,
+ cross_join.left(),
+ cross_join.right(),
+ )?;
+
+ Ok(Some(Arc::new(CrossJoinExec::new(
+ Arc::new(new_left),
+ Arc::new(new_right),
+ ))))
+}
+
+/// Tries to swap the projection with its input [`NestedLoopJoinExec`]. If it can be done,
+/// it returns the new swapped version having the [`NestedLoopJoinExec`] as the top plan.
+/// Otherwise, it returns None.
+fn try_swapping_with_nested_loop_join(
+ projection: &ProjectionExec,
+ nl_join: &NestedLoopJoinExec,
+) -> Result<Option<Arc<dyn ExecutionPlan>>> {
+ // Convert projected PhysicalExpr's to columns. If not possible, we cannot proceed.
+ let Some(projection_as_columns) = physical_to_column_exprs(projection.expr()) else {
+ return Ok(None);
+ };
+
+ let (far_right_left_col_ind, far_left_right_col_ind) = join_table_borders(
+ nl_join.left().schema().fields().len(),
+ &projection_as_columns,
+ );
+
+ if !join_allows_pushdown(
+ &projection_as_columns,
+ nl_join.schema(),
+ far_right_left_col_ind,
+ far_left_right_col_ind,
+ ) {
+ return Ok(None);
+ }
+
+ let new_filter = if let Some(filter) = nl_join.filter() {
+ match update_join_filter(
+ &projection_as_columns[0..=far_right_left_col_ind as _],
+ &projection_as_columns[far_left_right_col_ind as _..],
+ filter,
+ nl_join.left(),
+ nl_join.right(),
+ ) {
+ Some(updated_filter) => Some(updated_filter),
+ None => return Ok(None),
+ }
+ } else {
+ None
+ };
+
+ let (new_left, new_right) = new_join_children(
+ projection_as_columns,
+ far_right_left_col_ind,
+ far_left_right_col_ind,
+ nl_join.left(),
+ nl_join.right(),
+ )?;
+
+ Ok(Some(Arc::new(NestedLoopJoinExec::try_new(
+ Arc::new(new_left),
+ Arc::new(new_right),
+ new_filter,
+ nl_join.join_type(),
+ )?)))
+}
+
+/// Tries to swap the projection with its input [`SortMergeJoinExec`]. If it can be done,
+/// it returns the new swapped version having the [`SortMergeJoinExec`] as the top plan.
+/// Otherwise, it returns None.
+fn try_swapping_with_sort_merge_join(
+ projection: &ProjectionExec,
+ sm_join: &SortMergeJoinExec,
+) -> Result<Option<Arc<dyn ExecutionPlan>>> {
+ // Convert projected PhysicalExpr's to columns. If not possible, we cannot proceed.
+ let Some(projection_as_columns) = physical_to_column_exprs(projection.expr()) else {
+ return Ok(None);
+ };
+
+ let (far_right_left_col_ind, far_left_right_col_ind) = join_table_borders(
+ sm_join.left().schema().fields().len(),
+ &projection_as_columns,
+ );
+
+ if !join_allows_pushdown(
+ &projection_as_columns,
+ sm_join.schema(),
+ far_right_left_col_ind,
+ far_left_right_col_ind,
+ ) {
+ return Ok(None);
+ }
+
+ let Some(new_on) = update_join_on(
+ &projection_as_columns[0..=far_right_left_col_ind as _],
+ &projection_as_columns[far_left_right_col_ind as _..],
+ sm_join.on(),
+ ) else {
+ return Ok(None);
+ };
+
+ let (new_left, new_right) = new_join_children(
+ projection_as_columns,
+ far_right_left_col_ind,
+ far_left_right_col_ind,
+ &sm_join.children()[0],
+ &sm_join.children()[1],
+ )?;
+
+ Ok(Some(Arc::new(SortMergeJoinExec::try_new(
+ Arc::new(new_left),
+ Arc::new(new_right),
+ new_on,
+ sm_join.join_type,
+ sm_join.sort_options.clone(),
+ sm_join.null_equals_null,
+ )?)))
+}
+
+/// Tries to swap the projection with its input [`SymmetricHashJoinExec`]. If it can be done,
+/// it returns the new swapped version having the [`SymmetricHashJoinExec`] as the top plan.
+/// Otherwise, it returns None.
+fn try_swapping_with_sym_hash_join(
+ projection: &ProjectionExec,
+ sym_join: &SymmetricHashJoinExec,
+) -> Result<Option<Arc<dyn ExecutionPlan>>> {
+ // Convert projected PhysicalExpr's to columns. If not possible, we cannot proceed.
+ let Some(projection_as_columns) = physical_to_column_exprs(projection.expr()) else {
+ return Ok(None);
+ };
+
+ let (far_right_left_col_ind, far_left_right_col_ind) = join_table_borders(
+ sym_join.left().schema().fields().len(),
+ &projection_as_columns,
+ );
+
+ if !join_allows_pushdown(
+ &projection_as_columns,
+ sym_join.schema(),
+ far_right_left_col_ind,
+ far_left_right_col_ind,
+ ) {
+ return Ok(None);
+ }
+
+ let Some(new_on) = update_join_on(
+ &projection_as_columns[0..=far_right_left_col_ind as _],
+ &projection_as_columns[far_left_right_col_ind as _..],
+ sym_join.on(),
+ ) else {
+ return Ok(None);
+ };
+
+ let new_filter = if let Some(filter) = sym_join.filter() {
+ match update_join_filter(
+ &projection_as_columns[0..=far_right_left_col_ind as _],
+ &projection_as_columns[far_left_right_col_ind as _..],
+ filter,
+ sym_join.left(),
+ sym_join.right(),
+ ) {
+ Some(updated_filter) => Some(updated_filter),
+ None => return Ok(None),
+ }
+ } else {
+ None
+ };
+
+ let (new_left, new_right) = new_join_children(
+ projection_as_columns,
+ far_right_left_col_ind,
+ far_left_right_col_ind,
+ sym_join.left(),
+ sym_join.right(),
+ )?;
+
+ Ok(Some(Arc::new(SymmetricHashJoinExec::try_new(
+ Arc::new(new_left),
+ Arc::new(new_right),
+ new_on,
+ new_filter,
+ sym_join.join_type(),
+ sym_join.null_equals_null(),
+ sym_join.partition_mode(),
+ )?)))
+}
+
+/// Compare the inputs and outputs of the projection. If the projection causes
+/// any change in the fields, it returns `false`.
+fn is_projection_removable(projection: &ProjectionExec) -> bool {
+ all_alias_free_columns(projection.expr()) && {
+ let schema = projection.schema();
+ let input_schema = projection.input().schema();
+ let fields = schema.fields();
+ let input_fields = input_schema.fields();
+ fields.len() == input_fields.len()
+ && fields
+ .iter()
+ .zip(input_fields.iter())
+ .all(|(out, input)| out.eq(input))
+ }
+}
+
+/// Given the expression set of a projection, checks if the projection causes
+/// any renaming or constructs a non-`Column` physical expression.
+fn all_alias_free_columns(exprs: &[(Arc<dyn PhysicalExpr>, String)]) -> bool {
+ exprs.iter().all(|(expr, alias)| {
+ expr.as_any()
+ .downcast_ref::<Column>()
+ .map(|column| column.name() == alias)
+ .unwrap_or(false)
+ })
+}
+
+/// Updates a source provider's projected columns according to the given
+/// projection operator's expressions. To use this function safely, one must
+/// ensure that all expressions are `Column` expressions without aliases.
+fn new_projections_for_columns(
+ projection: &ProjectionExec,
+ source: &Option<Vec<usize>>,
+) -> Vec<usize> {
+ projection
+ .expr()
+ .iter()
+ .filter_map(|(expr, _)| {
+ expr.as_any()
+ .downcast_ref::<Column>()
+ .and_then(|expr| source.as_ref().map(|proj| proj[expr.index()]))
+ })
+ .collect()
+}
+
+/// The function operates in two modes:
+///
+/// 1) When `sync_with_child` is `true`:
+///
+/// The function updates the indices of `expr` if the expression resides
+/// in the input plan. For instance, given the expressions `a@1 + b@2`
+/// and `c@0` with the input schema `c@2, a@0, b@1`, the expressions are
+/// updated to `a@0 + b@1` and `c@2`.
+///
+/// 2) When `sync_with_child` is `false`:
+///
+/// The function determines how the expression would be updated if a projection
+/// was placed before the plan associated with the expression. If the expression
+/// cannot be rewritten after the projection, it returns `None`. For example,
+/// given the expressions `c@0`, `a@1` and `b@2`, and the [`ProjectionExec`] with
+/// an output schema of `a, c_new`, then `c@0` becomes `c_new@1`, `a@1` becomes
+/// `a@0`, but `b@2` results in `None` since the projection does not include `b`.
+///
+/// If the expression contains a `PhysicalExpr` variant that this function does
+/// not support, it will return `None`. An error can only be introduced if
+/// `CaseExpr::try_new` returns an error.
+fn update_expr(
+ expr: &Arc<dyn PhysicalExpr>,
+ projected_exprs: &[(Arc<dyn PhysicalExpr>, String)],
+ sync_with_child: bool,
+) -> Result<Option<Arc<dyn PhysicalExpr>>> {
+ let expr_any = expr.as_any();
+ if let Some(column) = expr_any.downcast_ref::<Column>() {
+ if sync_with_child {
+ // Update the index of `column`:
+ Ok(Some(projected_exprs[column.index()].0.clone()))
+ } else {
+ // Determine how to update `column` to accommodate `projected_exprs`:
+ Ok(projected_exprs.iter().enumerate().find_map(
+ |(index, (projected_expr, alias))| {
+ projected_expr.as_any().downcast_ref::<Column>().and_then(
+ |projected_column| {
+ column
+ .name()
+ .eq(projected_column.name())
+ .then(|| Arc::new(Column::new(alias, index)) as _)
+ },
+ )
+ },
+ ))
+ }
+ } else if let Some(binary) = expr_any.downcast_ref::<BinaryExpr>() {
+ match (
+ update_expr(binary.left(), projected_exprs, sync_with_child)?,
+ update_expr(binary.right(), projected_exprs, sync_with_child)?,
+ ) {
+ (Some(left), Some(right)) => {
+ Ok(Some(Arc::new(BinaryExpr::new(left, *binary.op(), right))))
+ }
+ _ => Ok(None),
+ }
+ } else if let Some(cast) = expr_any.downcast_ref::<CastExpr>() {
+ update_expr(cast.expr(), projected_exprs, sync_with_child).map(|maybe_expr| {
+ maybe_expr.map(|expr| {
+ Arc::new(CastExpr::new(
+ expr,
+ cast.cast_type().clone(),
+ Some(cast.cast_options().clone()),
+ )) as _
+ })
+ })
+ } else if expr_any.is::<Literal>() {
+ Ok(Some(expr.clone()))
+ } else if let Some(negative) = expr_any.downcast_ref::<NegativeExpr>() {
+ update_expr(negative.arg(), projected_exprs, sync_with_child).map(|maybe_expr| {
+ maybe_expr.map(|expr| Arc::new(NegativeExpr::new(expr)) as _)
+ })
+ } else if let Some(scalar_func) = expr_any.downcast_ref::<ScalarFunctionExpr>() {
+ scalar_func
+ .args()
+ .iter()
+ .map(|expr| update_expr(expr, projected_exprs, sync_with_child))
+ .collect::<Result<Option<Vec<_>>>>()
+ .map(|maybe_args| {
+ maybe_args.map(|new_args| {
+ Arc::new(ScalarFunctionExpr::new(
+ scalar_func.name(),
+ scalar_func.fun().clone(),
+ new_args,
+ scalar_func.return_type(),
+ scalar_func.monotonicity().clone(),
+ )) as _
+ })
+ })
+ } else if let Some(case) = expr_any.downcast_ref::<CaseExpr>() {
+ update_case_expr(case, projected_exprs, sync_with_child)
+ } else {
+ Ok(None)
+ }
+}
+
+/// Updates the indices `case` refers to according to `projected_exprs`.
+fn update_case_expr(
+ case: &CaseExpr,
+ projected_exprs: &[(Arc<dyn PhysicalExpr>, String)],
+ sync_with_child: bool,
+) -> Result<Option<Arc<dyn PhysicalExpr>>> {
+ let new_case = case
+ .expr()
+ .map(|expr| update_expr(expr, projected_exprs, sync_with_child))
+ .transpose()?
+ .flatten();
+
+ let new_else = case
+ .else_expr()
+ .map(|expr| update_expr(expr, projected_exprs, sync_with_child))
+ .transpose()?
+ .flatten();
+
+ let new_when_then = case
+ .when_then_expr()
+ .iter()
+ .map(|(when, then)| {
+ Ok((
+ update_expr(when, projected_exprs, sync_with_child)?,
+ update_expr(then, projected_exprs, sync_with_child)?,
+ ))
+ })
+ .collect::<Result<Vec<_>>>()?
+ .into_iter()
+ .filter_map(|(maybe_when, maybe_then)| match (maybe_when, maybe_then) {
+ (Some(when), Some(then)) => Some((when, then)),
+ _ => None,
+ })
+ .collect::<Vec<_>>();
+
+ if new_when_then.len() != case.when_then_expr().len()
+ || case.expr().is_some() && new_case.is_none()
+ || case.else_expr().is_some() && new_else.is_none()
+ {
+ return Ok(None);
+ }
+
+ CaseExpr::try_new(new_case, new_when_then, new_else).map(|e| Some(Arc::new(e) as _))
+}
+
+/// Creates a new [`ProjectionExec`] instance with the given child plan and
+/// projected expressions.
+fn make_with_child(
+ projection: &ProjectionExec,
+ child: &Arc<dyn ExecutionPlan>,
+) -> Result<Arc<dyn ExecutionPlan>> {
+ ProjectionExec::try_new(projection.expr().to_vec(), child.clone())
+ .map(|e| Arc::new(e) as _)
+}
+
+/// Returns `true` if all the expressions in the argument are `Column`s.
+fn all_columns(exprs: &[(Arc<dyn PhysicalExpr>, String)]) -> bool {
+ exprs.iter().all(|(expr, _)| expr.as_any().is::<Column>())
+}
+
+/// Downcasts all the expressions in `exprs` to `Column`s. If any of the given
+/// expressions is not a `Column`, returns `None`.
+fn physical_to_column_exprs(
+ exprs: &[(Arc<dyn PhysicalExpr>, String)],
+) -> Option<Vec<(Column, String)>> {
+ exprs
+ .iter()
+ .map(|(expr, alias)| {
+ expr.as_any()
+ .downcast_ref::<Column>()
+ .map(|col| (col.clone(), alias.clone()))
+ })
+ .collect()
+}
+
+/// Returns the last index before encountering a column coming from the right table when traveling
+/// through the projection from left to right, and the last index before encountering a column
+/// coming from the left table when traveling through the projection from right to left.
+/// If there is no column in the projection coming from the left side, it returns (-1, ...),
+/// if there is no column in the projection coming from the right side, it returns (..., projection length).
+fn join_table_borders(
+ left_table_column_count: usize,
+ projection_as_columns: &[(Column, String)],
+) -> (i32, i32) {
+ let far_right_left_col_ind = projection_as_columns
+ .iter()
+ .enumerate()
+ .take_while(|(_, (projection_column, _))| {
+ projection_column.index() < left_table_column_count
+ })
+ .last()
+ .map(|(index, _)| index as i32)
+ .unwrap_or(-1);
+
+ let far_left_right_col_ind = projection_as_columns
+ .iter()
+ .enumerate()
+ .rev()
+ .take_while(|(_, (projection_column, _))| {
+ projection_column.index() >= left_table_column_count
+ })
+ .last()
+ .map(|(index, _)| index as i32)
+ .unwrap_or(projection_as_columns.len() as i32);
+
+ (far_right_left_col_ind, far_left_right_col_ind)
+}
+
+/// Tries to update the equi-join `Column`'s of a join as if the the input of
+/// the join was replaced by a projection.
+fn update_join_on(
+ proj_left_exprs: &[(Column, String)],
+ proj_right_exprs: &[(Column, String)],
+ hash_join_on: &[(Column, Column)],
+) -> Option<Vec<(Column, Column)>> {
+ let (left_idx, right_idx): (Vec<_>, Vec<_>) = hash_join_on
+ .iter()
+ .map(|(left, right)| (left, right))
+ .unzip();
+
+ let new_left_columns = new_columns_for_join_on(&left_idx, proj_left_exprs);
+ let new_right_columns = new_columns_for_join_on(&right_idx, proj_right_exprs);
+
+ match (new_left_columns, new_right_columns) {
+ (Some(left), Some(right)) => Some(left.into_iter().zip(right).collect()),
+ _ => None,
+ }
+}
+
+/// This function generates a new set of columns to be used in a hash join
+/// operation based on a set of equi-join conditions (`hash_join_on`) and a
+/// list of projection expressions (`projection_exprs`).
+fn new_columns_for_join_on(
+ hash_join_on: &[&Column],
+ projection_exprs: &[(Column, String)],
+) -> Option<Vec<Column>> {
+ let new_columns = hash_join_on
+ .iter()
+ .filter_map(|on| {
+ projection_exprs
+ .iter()
+ .enumerate()
+ .find(|(_, (proj_column, _))| on.name() == proj_column.name())
+ .map(|(index, (_, alias))| Column::new(alias, index))
+ })
+ .collect::<Vec<_>>();
+ (new_columns.len() == hash_join_on.len()).then_some(new_columns)
+}
+
+/// Tries to update the column indices of a [`JoinFilter`] as if the the input of
+/// the join was replaced by a projection.
+fn update_join_filter(
+ projection_left_exprs: &[(Column, String)],
+ projection_right_exprs: &[(Column, String)],
+ join_filter: &JoinFilter,
+ join_left: &Arc<dyn ExecutionPlan>,
+ join_right: &Arc<dyn ExecutionPlan>,
+) -> Option<JoinFilter> {
+ let mut new_left_indices = new_indices_for_join_filter(
+ join_filter,
+ JoinSide::Left,
+ projection_left_exprs,
+ join_left.schema(),
+ )
+ .into_iter();
+ let mut new_right_indices = new_indices_for_join_filter(
+ join_filter,
+ JoinSide::Right,
+ projection_right_exprs,
+ join_right.schema(),
+ )
+ .into_iter();
+
+ // Check if all columns match:
+ (new_right_indices.len() + new_left_indices.len()
+ == join_filter.column_indices().len())
+ .then(|| {
+ JoinFilter::new(
+ join_filter.expression().clone(),
+ join_filter
+ .column_indices()
+ .iter()
+ .map(|col_idx| ColumnIndex {
+ index: if col_idx.side == JoinSide::Left {
+ new_left_indices.next().unwrap()
+ } else {
+ new_right_indices.next().unwrap()
+ },
+ side: col_idx.side,
+ })
+ .collect(),
+ join_filter.schema().clone(),
+ )
+ })
+}
+
+/// This function determines and returns a vector of indices representing the
+/// positions of columns in `projection_exprs` that are involved in `join_filter`,
+/// and correspond to a particular side (`join_side`) of the join operation.
+fn new_indices_for_join_filter(
+ join_filter: &JoinFilter,
+ join_side: JoinSide,
+ projection_exprs: &[(Column, String)],
+ join_child_schema: SchemaRef,
+) -> Vec<usize> {
+ join_filter
+ .column_indices()
+ .iter()
+ .filter(|col_idx| col_idx.side == join_side)
+ .filter_map(|col_idx| {
+ projection_exprs.iter().position(|(col, _)| {
+ col.name() == join_child_schema.fields()[col_idx.index].name()
+ })
+ })
+ .collect()
+}
+
+/// Checks three conditions for pushing a projection down through a join:
+/// - Projection must narrow the join output schema.
+/// - Columns coming from left/right tables must be collected at the left/right
+/// sides of the output table.
+/// - Left or right table is not lost after the projection.
+fn join_allows_pushdown(
+ projection_as_columns: &[(Column, String)],
+ join_schema: SchemaRef,
+ far_right_left_col_ind: i32,
+ far_left_right_col_ind: i32,
+) -> bool {
+ // Projection must narrow the join output:
+ projection_as_columns.len() < join_schema.fields().len()
+ // Are the columns from different tables mixed?
+ && (far_right_left_col_ind + 1 == far_left_right_col_ind)
+ // Left or right table is not lost after the projection.
+ && far_right_left_col_ind >= 0
+ && far_left_right_col_ind < projection_as_columns.len() as i32
+}
+
+/// If pushing down the projection over this join's children seems possible,
+/// this function constructs the new [`ProjectionExec`]s that will come on top
+/// of the original children of the join.
+fn new_join_children(
+ projection_as_columns: Vec<(Column, String)>,
+ far_right_left_col_ind: i32,
+ far_left_right_col_ind: i32,
+ left_child: &Arc<dyn ExecutionPlan>,
+ right_child: &Arc<dyn ExecutionPlan>,
+) -> Result<(ProjectionExec, ProjectionExec)> {
+ let new_left = ProjectionExec::try_new(
+ projection_as_columns[0..=far_right_left_col_ind as _]
+ .iter()
+ .map(|(col, alias)| {
+ (
+ Arc::new(Column::new(col.name(), col.index())) as _,
+ alias.clone(),
+ )
+ })
+ .collect_vec(),
+ left_child.clone(),
+ )?;
+ let left_size = left_child.schema().fields().len() as i32;
+ let new_right = ProjectionExec::try_new(
+ projection_as_columns[far_left_right_col_ind as _..]
+ .iter()
+ .map(|(col, alias)| {
+ (
+ Arc::new(Column::new(
+ col.name(),
+ // Align projected expressions coming from the right
+ // table with the new right child projection:
+ (col.index() as i32 - left_size) as _,
+ )) as _,
+ alias.clone(),
+ )
+ })
+ .collect_vec(),
+ right_child.clone(),
+ )?;
+
+ Ok((new_left, new_right))
+}
+
+#[cfg(test)]
+mod tests {
+ use std::sync::Arc;
+
+ use crate::datasource::file_format::file_compression_type::FileCompressionType;
+ use crate::datasource::listing::PartitionedFile;
+ use crate::datasource::physical_plan::{CsvExec, FileScanConfig};
+ use crate::physical_optimizer::output_requirements::OutputRequirementExec;
+ use crate::physical_optimizer::projection_pushdown::{
+ join_table_borders, update_expr, ProjectionPushdown,
+ };
+ use crate::physical_optimizer::utils::get_plan_string;
+ use crate::physical_optimizer::PhysicalOptimizerRule;
+ use crate::physical_plan::coalesce_partitions::CoalescePartitionsExec;
+ use crate::physical_plan::filter::FilterExec;
+ use crate::physical_plan::joins::utils::{ColumnIndex, JoinFilter};
+ use crate::physical_plan::joins::StreamJoinPartitionMode;
+ use crate::physical_plan::memory::MemoryExec;
+ use crate::physical_plan::projection::ProjectionExec;
+ use crate::physical_plan::repartition::RepartitionExec;
+ use crate::physical_plan::sorts::sort::SortExec;
+ use crate::physical_plan::sorts::sort_preserving_merge::SortPreservingMergeExec;
+ use crate::physical_plan::ExecutionPlan;
+
+ use arrow_schema::{DataType, Field, Schema, SortOptions};
+ use datafusion_common::config::ConfigOptions;
+ use datafusion_common::{JoinSide, JoinType, Result, ScalarValue, Statistics};
+ use datafusion_execution::object_store::ObjectStoreUrl;
+ use datafusion_expr::{ColumnarValue, Operator};
+ use datafusion_physical_expr::expressions::{
+ BinaryExpr, CaseExpr, CastExpr, Column, Literal, NegativeExpr,
+ };
+ use datafusion_physical_expr::{
+ Distribution, Partitioning, PhysicalExpr, PhysicalSortExpr,
+ PhysicalSortRequirement, ScalarFunctionExpr,
+ };
+ use datafusion_physical_plan::joins::SymmetricHashJoinExec;
+ use datafusion_physical_plan::union::UnionExec;
+
+ #[test]
+ fn test_update_matching_exprs() -> Result<()> {
+ let exprs: Vec<Arc<dyn PhysicalExpr>> = vec![
+ Arc::new(BinaryExpr::new(
+ Arc::new(Column::new("a", 3)),
+ Operator::Divide,
+ Arc::new(Column::new("e", 5)),
+ )),
+ Arc::new(CastExpr::new(
+ Arc::new(Column::new("a", 3)),
+ DataType::Float32,
+ None,
+ )),
+ Arc::new(NegativeExpr::new(Arc::new(Column::new("f", 4)))),
+ Arc::new(ScalarFunctionExpr::new(
+ "scalar_expr",
+ Arc::new(|_: &[ColumnarValue]| unimplemented!("not implemented")),
+ vec![
+ Arc::new(BinaryExpr::new(
+ Arc::new(Column::new("b", 1)),
+ Operator::Divide,
+ Arc::new(Column::new("c", 0)),
+ )),
+ Arc::new(BinaryExpr::new(
+ Arc::new(Column::new("c", 0)),
+ Operator::Divide,
+ Arc::new(Column::new("b", 1)),
+ )),
+ ],
+ &DataType::Int32,
+ None,
+ )),
+ Arc::new(CaseExpr::try_new(
+ Some(Arc::new(Column::new("d", 2))),
+ vec![
+ (
+ Arc::new(Column::new("a", 3)) as Arc<dyn PhysicalExpr>,
+ Arc::new(BinaryExpr::new(
+ Arc::new(Column::new("d", 2)),
+ Operator::Plus,
+ Arc::new(Column::new("e", 5)),
+ )) as Arc<dyn PhysicalExpr>,
+ ),
+ (
+ Arc::new(Column::new("a", 3)) as Arc<dyn PhysicalExpr>,
+ Arc::new(BinaryExpr::new(
+ Arc::new(Column::new("e", 5)),
+ Operator::Plus,
+ Arc::new(Column::new("d", 2)),
+ )) as Arc<dyn PhysicalExpr>,
+ ),
+ ],
+ Some(Arc::new(BinaryExpr::new(
+ Arc::new(Column::new("a", 3)),
+ Operator::Modulo,
+ Arc::new(Column::new("e", 5)),
+ ))),
+ )?),
+ ];
+ let child: Vec<(Arc<dyn PhysicalExpr>, String)> = vec![
+ (Arc::new(Column::new("c", 2)), "c".to_owned()),
+ (Arc::new(Column::new("b", 1)), "b".to_owned()),
+ (Arc::new(Column::new("d", 3)), "d".to_owned()),
+ (Arc::new(Column::new("a", 0)), "a".to_owned()),
+ (Arc::new(Column::new("f", 5)), "f".to_owned()),
+ (Arc::new(Column::new("e", 4)), "e".to_owned()),
+ ];
+
+ let expected_exprs: Vec<Arc<dyn PhysicalExpr>> = vec![
+ Arc::new(BinaryExpr::new(
+ Arc::new(Column::new("a", 0)),
+ Operator::Divide,
+ Arc::new(Column::new("e", 4)),
+ )),
+ Arc::new(CastExpr::new(
+ Arc::new(Column::new("a", 0)),
+ DataType::Float32,
+ None,
+ )),
+ Arc::new(NegativeExpr::new(Arc::new(Column::new("f", 5)))),
+ Arc::new(ScalarFunctionExpr::new(
+ "scalar_expr",
+ Arc::new(|_: &[ColumnarValue]| unimplemented!("not implemented")),
+ vec![
+ Arc::new(BinaryExpr::new(
+ Arc::new(Column::new("b", 1)),
+ Operator::Divide,
+ Arc::new(Column::new("c", 2)),
+ )),
+ Arc::new(BinaryExpr::new(
+ Arc::new(Column::new("c", 2)),
+ Operator::Divide,
+ Arc::new(Column::new("b", 1)),
+ )),
+ ],
+ &DataType::Int32,
+ None,
+ )),
+ Arc::new(CaseExpr::try_new(
+ Some(Arc::new(Column::new("d", 3))),
+ vec![
+ (
+ Arc::new(Column::new("a", 0)) as Arc<dyn PhysicalExpr>,
+ Arc::new(BinaryExpr::new(
+ Arc::new(Column::new("d", 3)),
+ Operator::Plus,
+ Arc::new(Column::new("e", 4)),
+ )) as Arc<dyn PhysicalExpr>,
+ ),
+ (
+ Arc::new(Column::new("a", 0)) as Arc<dyn PhysicalExpr>,
+ Arc::new(BinaryExpr::new(
+ Arc::new(Column::new("e", 4)),
+ Operator::Plus,
+ Arc::new(Column::new("d", 3)),
+ )) as Arc<dyn PhysicalExpr>,
+ ),
+ ],
+ Some(Arc::new(BinaryExpr::new(
+ Arc::new(Column::new("a", 0)),
+ Operator::Modulo,
+ Arc::new(Column::new("e", 4)),
+ ))),
+ )?),
+ ];
+
+ for (expr, expected_expr) in exprs.into_iter().zip(expected_exprs.into_iter()) {
+ assert!(update_expr(&expr, &child, true)?
+ .unwrap()
+ .eq(&expected_expr));
+ }
+
+ Ok(())
+ }
+
+ #[test]
+ fn test_update_projected_exprs() -> Result<()> {
+ let exprs: Vec<Arc<dyn PhysicalExpr>> = vec![
+ Arc::new(BinaryExpr::new(
+ Arc::new(Column::new("a", 3)),
+ Operator::Divide,
+ Arc::new(Column::new("e", 5)),
+ )),
+ Arc::new(CastExpr::new(
+ Arc::new(Column::new("a", 3)),
+ DataType::Float32,
+ None,
+ )),
+ Arc::new(NegativeExpr::new(Arc::new(Column::new("f", 4)))),
+ Arc::new(ScalarFunctionExpr::new(
+ "scalar_expr",
+ Arc::new(|_: &[ColumnarValue]| unimplemented!("not implemented")),
+ vec![
+ Arc::new(BinaryExpr::new(
+ Arc::new(Column::new("b", 1)),
+ Operator::Divide,
+ Arc::new(Column::new("c", 0)),
+ )),
+ Arc::new(BinaryExpr::new(
+ Arc::new(Column::new("c", 0)),
+ Operator::Divide,
+ Arc::new(Column::new("b", 1)),
+ )),
+ ],
+ &DataType::Int32,
+ None,
+ )),
+ Arc::new(CaseExpr::try_new(
+ Some(Arc::new(Column::new("d", 2))),
+ vec![
+ (
+ Arc::new(Column::new("a", 3)) as Arc<dyn PhysicalExpr>,
+ Arc::new(BinaryExpr::new(
+ Arc::new(Column::new("d", 2)),
+ Operator::Plus,
+ Arc::new(Column::new("e", 5)),
+ )) as Arc<dyn PhysicalExpr>,
+ ),
+ (
+ Arc::new(Column::new("a", 3)) as Arc<dyn PhysicalExpr>,
+ Arc::new(BinaryExpr::new(
+ Arc::new(Column::new("e", 5)),
+ Operator::Plus,
+ Arc::new(Column::new("d", 2)),
+ )) as Arc<dyn PhysicalExpr>,
+ ),
+ ],
+ Some(Arc::new(BinaryExpr::new(
+ Arc::new(Column::new("a", 3)),
+ Operator::Modulo,
+ Arc::new(Column::new("e", 5)),
+ ))),
+ )?),
+ ];
+ let projected_exprs: Vec<(Arc<dyn PhysicalExpr>, String)> = vec![
+ (Arc::new(Column::new("a", 0)), "a".to_owned()),
+ (Arc::new(Column::new("b", 1)), "b_new".to_owned()),
+ (Arc::new(Column::new("c", 2)), "c".to_owned()),
+ (Arc::new(Column::new("d", 3)), "d_new".to_owned()),
+ (Arc::new(Column::new("e", 4)), "e".to_owned()),
+ (Arc::new(Column::new("f", 5)), "f_new".to_owned()),
+ ];
+
+ let expected_exprs: Vec<Arc<dyn PhysicalExpr>> = vec![
+ Arc::new(BinaryExpr::new(
+ Arc::new(Column::new("a", 0)),
+ Operator::Divide,
+ Arc::new(Column::new("e", 4)),
+ )),
+ Arc::new(CastExpr::new(
+ Arc::new(Column::new("a", 0)),
+ DataType::Float32,
+ None,
+ )),
+ Arc::new(NegativeExpr::new(Arc::new(Column::new("f_new", 5)))),
+ Arc::new(ScalarFunctionExpr::new(
+ "scalar_expr",
+ Arc::new(|_: &[ColumnarValue]| unimplemented!("not implemented")),
+ vec![
+ Arc::new(BinaryExpr::new(
+ Arc::new(Column::new("b_new", 1)),
+ Operator::Divide,
+ Arc::new(Column::new("c", 2)),
+ )),
+ Arc::new(BinaryExpr::new(
+ Arc::new(Column::new("c", 2)),
+ Operator::Divide,
+ Arc::new(Column::new("b_new", 1)),
+ )),
+ ],
+ &DataType::Int32,
+ None,
+ )),
+ Arc::new(CaseExpr::try_new(
+ Some(Arc::new(Column::new("d_new", 3))),
+ vec![
+ (
+ Arc::new(Column::new("a", 0)) as Arc<dyn PhysicalExpr>,
+ Arc::new(BinaryExpr::new(
+ Arc::new(Column::new("d_new", 3)),
+ Operator::Plus,
+ Arc::new(Column::new("e", 4)),
+ )) as Arc<dyn PhysicalExpr>,
+ ),
+ (
+ Arc::new(Column::new("a", 0)) as Arc<dyn PhysicalExpr>,
+ Arc::new(BinaryExpr::new(
+ Arc::new(Column::new("e", 4)),
+ Operator::Plus,
+ Arc::new(Column::new("d_new", 3)),
+ )) as Arc<dyn PhysicalExpr>,
+ ),
+ ],
+ Some(Arc::new(BinaryExpr::new(
+ Arc::new(Column::new("a", 0)),
+ Operator::Modulo,
+ Arc::new(Column::new("e", 4)),
+ ))),
+ )?),
+ ];
+
+ for (expr, expected_expr) in exprs.into_iter().zip(expected_exprs.into_iter()) {
+ assert!(update_expr(&expr, &projected_exprs, false)?
+ .unwrap()
+ .eq(&expected_expr));
+ }
+
+ Ok(())
+ }
+
+ #[test]
+ fn test_join_table_borders() -> Result<()> {
+ let projections = vec![
+ (Column::new("b", 1), "b".to_owned()),
+ (Column::new("c", 2), "c".to_owned()),
+ (Column::new("e", 4), "e".to_owned()),
+ (Column::new("d", 3), "d".to_owned()),
+ (Column::new("c", 2), "c".to_owned()),
+ (Column::new("f", 5), "f".to_owned()),
+ (Column::new("h", 7), "h".to_owned()),
+ (Column::new("g", 6), "g".to_owned()),
+ ];
+ let left_table_column_count = 5;
+ assert_eq!(
+ join_table_borders(left_table_column_count, &projections),
+ (4, 5)
+ );
+
+ let left_table_column_count = 8;
+ assert_eq!(
+ join_table_borders(left_table_column_count, &projections),
+ (7, 8)
+ );
+
+ let left_table_column_count = 1;
+ assert_eq!(
+ join_table_borders(left_table_column_count, &projections),
+ (-1, 0)
+ );
+
+ let projections = vec![
+ (Column::new("a", 0), "a".to_owned()),
+ (Column::new("b", 1), "b".to_owned()),
+ (Column::new("d", 3), "d".to_owned()),
+ (Column::new("g", 6), "g".to_owned()),
+ (Column::new("e", 4), "e".to_owned()),
+ (Column::new("f", 5), "f".to_owned()),
+ (Column::new("e", 4), "e".to_owned()),
+ (Column::new("h", 7), "h".to_owned()),
+ ];
+ let left_table_column_count = 5;
+ assert_eq!(
+ join_table_borders(left_table_column_count, &projections),
+ (2, 7)
+ );
+
+ let left_table_column_count = 7;
+ assert_eq!(
+ join_table_borders(left_table_column_count, &projections),
+ (6, 7)
+ );
+
+ Ok(())
+ }
+
+ fn create_simple_csv_exec() -> Arc<dyn ExecutionPlan> {
+ let schema = Arc::new(Schema::new(vec![
+ Field::new("a", DataType::Int32, true),
+ Field::new("b", DataType::Int32, true),
+ Field::new("c", DataType::Int32, true),
+ Field::new("d", DataType::Int32, true),
+ Field::new("e", DataType::Int32, true),
+ ]));
+ Arc::new(CsvExec::new(
+ FileScanConfig {
+ object_store_url: ObjectStoreUrl::parse("test:///").unwrap(),
+ file_schema: schema.clone(),
+ file_groups: vec![vec![PartitionedFile::new("x".to_string(), 100)]],
+ statistics: Statistics::new_unknown(&schema),
+ projection: Some(vec![0, 1, 2, 3, 4]),
+ limit: None,
+ table_partition_cols: vec![],
+ output_ordering: vec![vec![]],
+ infinite_source: false,
+ },
+ false,
+ 0,
+ 0,
+ None,
+ FileCompressionType::UNCOMPRESSED,
+ ))
+ }
+
+ fn create_projecting_csv_exec() -> Arc<dyn ExecutionPlan> {
+ let schema = Arc::new(Schema::new(vec![
+ Field::new("a", DataType::Int32, true),
+ Field::new("b", DataType::Int32, true),
+ Field::new("c", DataType::Int32, true),
+ Field::new("d", DataType::Int32, true),
+ ]));
+ Arc::new(CsvExec::new(
+ FileScanConfig {
+ object_store_url: ObjectStoreUrl::parse("test:///").unwrap(),
+ file_schema: schema.clone(),
+ file_groups: vec![vec![PartitionedFile::new("x".to_string(), 100)]],
+ statistics: Statistics::new_unknown(&schema),
+ projection: Some(vec![3, 2, 1]),
+ limit: None,
+ table_partition_cols: vec![],
+ output_ordering: vec![vec![]],
+ infinite_source: false,
+ },
+ false,
+ 0,
+ 0,
+ None,
+ FileCompressionType::UNCOMPRESSED,
+ ))
+ }
+
+ fn create_projecting_memory_exec() -> Arc<dyn ExecutionPlan> {
+ let schema = Arc::new(Schema::new(vec![
+ Field::new("a", DataType::Int32, true),
+ Field::new("b", DataType::Int32, true),
+ Field::new("c", DataType::Int32, true),
+ Field::new("d", DataType::Int32, true),
+ Field::new("e", DataType::Int32, true),
+ ]));
+
+ Arc::new(MemoryExec::try_new(&[], schema, Some(vec![2, 0, 3, 4])).unwrap())
+ }
+
+ #[test]
+ fn test_csv_after_projection() -> Result<()> {
+ let csv = create_projecting_csv_exec();
+ let projection: Arc<dyn ExecutionPlan> = Arc::new(ProjectionExec::try_new(
+ vec![
+ (Arc::new(Column::new("b", 2)), "b".to_string()),
+ (Arc::new(Column::new("d", 0)), "d".to_string()),
+ ],
+ csv.clone(),
+ )?);
+ let initial = get_plan_string(&projection);
+ let expected_initial = [
+ "ProjectionExec: expr=[b@2 as b, d@0 as d]",
+ " CsvExec: file_groups={1 group: [[x]]}, projection=[d, c, b], has_header=false",
+ ];
+ assert_eq!(initial, expected_initial);
+
+ let after_optimize =
+ ProjectionPushdown::new().optimize(projection, &ConfigOptions::new())?;
+
+ let expected = [
+ "CsvExec: file_groups={1 group: [[x]]}, projection=[b, d], has_header=false",
+ ];
+ assert_eq!(get_plan_string(&after_optimize), expected);
+
+ Ok(())
+ }
+
+ #[test]
+ fn test_memory_after_projection() -> Result<()> {
+ let memory = create_projecting_memory_exec();
+ let projection: Arc<dyn ExecutionPlan> = Arc::new(ProjectionExec::try_new(
+ vec![
+ (Arc::new(Column::new("d", 2)), "d".to_string()),
+ (Arc::new(Column::new("e", 3)), "e".to_string()),
+ (Arc::new(Column::new("a", 1)), "a".to_string()),
+ ],
+ memory.clone(),
+ )?);
+ let initial = get_plan_string(&projection);
+ let expected_initial = [
+ "ProjectionExec: expr=[d@2 as d, e@3 as e, a@1 as a]",
+ " MemoryExec: partitions=0, partition_sizes=[]",
+ ];
+ assert_eq!(initial, expected_initial);
+
+ let after_optimize =
+ ProjectionPushdown::new().optimize(projection, &ConfigOptions::new())?;
+
+ let expected = ["MemoryExec: partitions=0, partition_sizes=[]"];
+ assert_eq!(get_plan_string(&after_optimize), expected);
+ assert_eq!(
+ after_optimize
+ .clone()
+ .as_any()
+ .downcast_ref::<MemoryExec>()
+ .unwrap()
+ .projection()
+ .clone()
+ .unwrap(),
+ vec![3, 4, 0]
+ );
+
+ Ok(())
+ }
+
+ #[test]
+ fn test_projection_after_projection() -> Result<()> {
+ let csv = create_simple_csv_exec();
+ let child_projection: Arc<dyn ExecutionPlan> = Arc::new(ProjectionExec::try_new(
+ vec![
+ (Arc::new(Column::new("c", 2)), "c".to_string()),
+ (Arc::new(Column::new("e", 4)), "new_e".to_string()),
+ (Arc::new(Column::new("a", 0)), "a".to_string()),
+ (Arc::new(Column::new("b", 1)), "new_b".to_string()),
+ ],
+ csv.clone(),
+ )?);
+ let top_projection: Arc<dyn ExecutionPlan> = Arc::new(ProjectionExec::try_new(
+ vec![
+ (Arc::new(Column::new("new_b", 3)), "new_b".to_string()),
+ (
+ Arc::new(BinaryExpr::new(
+ Arc::new(Column::new("c", 0)),
+ Operator::Plus,
+ Arc::new(Column::new("new_e", 1)),
+ )),
+ "binary".to_string(),
+ ),
+ (Arc::new(Column::new("new_b", 3)), "newest_b".to_string()),
+ ],
+ child_projection.clone(),
+ )?);
+
+ let initial = get_plan_string(&top_projection);
+ let expected_initial = [
+ "ProjectionExec: expr=[new_b@3 as new_b, c@0 + new_e@1 as binary, new_b@3 as newest_b]",
+ " ProjectionExec: expr=[c@2 as c, e@4 as new_e, a@0 as a, b@1 as new_b]",
+ " CsvExec: file_groups={1 group: [[x]]}, projection=[a, b, c, d, e], has_header=false"
+ ];
+ assert_eq!(initial, expected_initial);
+
+ let after_optimize =
+ ProjectionPushdown::new().optimize(top_projection, &ConfigOptions::new())?;
+
+ let expected = [
+ "ProjectionExec: expr=[b@1 as new_b, c@2 + e@4 as binary, b@1 as newest_b]",
+ " CsvExec: file_groups={1 group: [[x]]}, projection=[a, b, c, d, e], has_header=false"
+ ];
+ assert_eq!(get_plan_string(&after_optimize), expected);
+
+ Ok(())
+ }
+
+ #[test]
+ fn test_output_req_after_projection() -> Result<()> {
+ let csv = create_simple_csv_exec();
+ let sort_req: Arc<dyn ExecutionPlan> = Arc::new(OutputRequirementExec::new(
+ csv.clone(),
+ Some(vec![
+ PhysicalSortRequirement {
+ expr: Arc::new(Column::new("b", 1)),
+ options: Some(SortOptions::default()),
+ },
+ PhysicalSortRequirement {
+ expr: Arc::new(BinaryExpr::new(
+ Arc::new(Column::new("c", 2)),
+ Operator::Plus,
+ Arc::new(Column::new("a", 0)),
+ )),
+ options: Some(SortOptions::default()),
+ },
+ ]),
+ Distribution::HashPartitioned(vec![
+ Arc::new(Column::new("a", 0)),
+ Arc::new(Column::new("b", 1)),
+ ]),
+ ));
+ let projection: Arc<dyn ExecutionPlan> = Arc::new(ProjectionExec::try_new(
+ vec![
+ (Arc::new(Column::new("c", 2)), "c".to_string()),
+ (Arc::new(Column::new("a", 0)), "new_a".to_string()),
+ (Arc::new(Column::new("b", 1)), "b".to_string()),
+ ],
+ sort_req.clone(),
+ )?);
+
+ let initial = get_plan_string(&projection);
+ let expected_initial = [
+ "ProjectionExec: expr=[c@2 as c, a@0 as new_a, b@1 as b]",
+ " OutputRequirementExec",
+ " CsvExec: file_groups={1 group: [[x]]}, projection=[a, b, c, d, e], has_header=false"
+ ];
+ assert_eq!(initial, expected_initial);
+
+ let after_optimize =
+ ProjectionPushdown::new().optimize(projection, &ConfigOptions::new())?;
+
+ let expected: [&str; 3] = [
+ "OutputRequirementExec",
+ " ProjectionExec: expr=[c@2 as c, a@0 as new_a, b@1 as b]",
+ " CsvExec: file_groups={1 group: [[x]]}, projection=[a, b, c, d, e], has_header=false"
+ ];
+
+ assert_eq!(get_plan_string(&after_optimize), expected);
+ let expected_reqs = vec![
+ PhysicalSortRequirement {
+ expr: Arc::new(Column::new("b", 2)),
+ options: Some(SortOptions::default()),
+ },
+ PhysicalSortRequirement {
+ expr: Arc::new(BinaryExpr::new(
+ Arc::new(Column::new("c", 0)),
+ Operator::Plus,
+ Arc::new(Column::new("new_a", 1)),
+ )),
+ options: Some(SortOptions::default()),
+ },
+ ];
+ assert_eq!(
+ after_optimize
+ .as_any()
+ .downcast_ref::<OutputRequirementExec>()
+ .unwrap()
+ .required_input_ordering()[0]
+ .clone()
+ .unwrap(),
+ expected_reqs
+ );
+ let expected_distribution: Vec<Arc<dyn PhysicalExpr>> = vec![
+ Arc::new(Column::new("new_a", 1)),
+ Arc::new(Column::new("b", 2)),
+ ];
+ if let Distribution::HashPartitioned(vec) = after_optimize
+ .as_any()
+ .downcast_ref::<OutputRequirementExec>()
+ .unwrap()
+ .required_input_distribution()[0]
+ .clone()
+ {
+ assert!(vec
+ .iter()
+ .zip(expected_distribution)
+ .all(|(actual, expected)| actual.eq(&expected)));
+ } else {
+ panic!("Expected HashPartitioned distribution!");
+ };
+
+ Ok(())
+ }
+
+ #[test]
+ fn test_coalesce_partitions_after_projection() -> Result<()> {
+ let csv = create_simple_csv_exec();
+ let coalesce_partitions: Arc<dyn ExecutionPlan> =
+ Arc::new(CoalescePartitionsExec::new(csv));
+ let projection: Arc<dyn ExecutionPlan> = Arc::new(ProjectionExec::try_new(
+ vec![
+ (Arc::new(Column::new("b", 1)), "b".to_string()),
+ (Arc::new(Column::new("a", 0)), "a_new".to_string()),
+ (Arc::new(Column::new("d", 3)), "d".to_string()),
+ ],
+ coalesce_partitions,
+ )?);
+ let initial = get_plan_string(&projection);
+ let expected_initial = [
+ "ProjectionExec: expr=[b@1 as b, a@0 as a_new, d@3 as d]",
+ " CoalescePartitionsExec",
+ " CsvExec: file_groups={1 group: [[x]]}, projection=[a, b, c, d, e], has_header=false",
+ ];
+ assert_eq!(initial, expected_initial);
+
+ let after_optimize =
+ ProjectionPushdown::new().optimize(projection, &ConfigOptions::new())?;
+
+ let expected = [
+ "CoalescePartitionsExec",
+ " ProjectionExec: expr=[b@1 as b, a@0 as a_new, d@3 as d]",
+ " CsvExec: file_groups={1 group: [[x]]}, projection=[a, b, c, d, e], has_header=false",
+ ];
+ assert_eq!(get_plan_string(&after_optimize), expected);
+
+ Ok(())
+ }
+
+ #[test]
+ fn test_filter_after_projection() -> Result<()> {
+ let csv = create_simple_csv_exec();
+ let predicate = Arc::new(BinaryExpr::new(
+ Arc::new(BinaryExpr::new(
+ Arc::new(Column::new("b", 1)),
+ Operator::Minus,
+ Arc::new(Column::new("a", 0)),
+ )),
+ Operator::Gt,
+ Arc::new(BinaryExpr::new(
+ Arc::new(Column::new("d", 3)),
+ Operator::Minus,
+ Arc::new(Column::new("a", 0)),
+ )),
+ ));
+ let filter: Arc<dyn ExecutionPlan> =
+ Arc::new(FilterExec::try_new(predicate, csv)?);
+ let projection: Arc<dyn ExecutionPlan> = Arc::new(ProjectionExec::try_new(
+ vec![
+ (Arc::new(Column::new("a", 0)), "a_new".to_string()),
+ (Arc::new(Column::new("b", 1)), "b".to_string()),
+ (Arc::new(Column::new("d", 3)), "d".to_string()),
+ ],
+ filter.clone(),
+ )?);
+
+ let initial = get_plan_string(&projection);
+ let expected_initial = [
+ "ProjectionExec: expr=[a@0 as a_new, b@1 as b, d@3 as d]",
+ " FilterExec: b@1 - a@0 > d@3 - a@0",
+ " CsvExec: file_groups={1 group: [[x]]}, projection=[a, b, c, d, e], has_header=false",
+ ];
+ assert_eq!(initial, expected_initial);
+
+ let after_optimize =
+ ProjectionPushdown::new().optimize(projection, &ConfigOptions::new())?;
+
+ let expected = [
+ "FilterExec: b@1 - a_new@0 > d@2 - a_new@0",
+ " ProjectionExec: expr=[a@0 as a_new, b@1 as b, d@3 as d]",
+ " CsvExec: file_groups={1 group: [[x]]}, projection=[a, b, c, d, e], has_header=false",
+ ];
+ assert_eq!(get_plan_string(&after_optimize), expected);
+
+ Ok(())
+ }
+
+ #[test]
+ fn test_join_after_projection() -> Result<()> {
+ let left_csv = create_simple_csv_exec();
+ let right_csv = create_simple_csv_exec();
+
+ let join: Arc<dyn ExecutionPlan> = Arc::new(SymmetricHashJoinExec::try_new(
+ left_csv,
+ right_csv,
+ vec![(Column::new("b", 1), Column::new("c", 2))],
+ // b_left-(1+a_right)<=a_right+c_left
+ Some(JoinFilter::new(
+ Arc::new(BinaryExpr::new(
+ Arc::new(BinaryExpr::new(
+ Arc::new(Column::new("b_left_inter", 0)),
+ Operator::Minus,
+ Arc::new(BinaryExpr::new(
+ Arc::new(Literal::new(ScalarValue::Int32(Some(1)))),
+ Operator::Plus,
+ Arc::new(Column::new("a_right_inter", 1)),
+ )),
+ )),
+ Operator::LtEq,
+ Arc::new(BinaryExpr::new(
+ Arc::new(Column::new("a_right_inter", 1)),
+ Operator::Plus,
+ Arc::new(Column::new("c_left_inter", 2)),
+ )),
+ )),
+ vec![
+ ColumnIndex {
+ index: 1,
+ side: JoinSide::Left,
+ },
+ ColumnIndex {
+ index: 0,
+ side: JoinSide::Right,
+ },
+ ColumnIndex {
+ index: 2,
+ side: JoinSide::Left,
+ },
+ ],
+ Schema::new(vec![
+ Field::new("b_left_inter", DataType::Int32, true),
+ Field::new("a_right_inter", DataType::Int32, true),
+ Field::new("c_left_inter", DataType::Int32, true),
+ ]),
+ )),
+ &JoinType::Inner,
+ true,
+ StreamJoinPartitionMode::SinglePartition,
+ )?);
+ let projection: Arc<dyn ExecutionPlan> = Arc::new(ProjectionExec::try_new(
+ vec![
+ (Arc::new(Column::new("c", 2)), "c_from_left".to_string()),
+ (Arc::new(Column::new("b", 1)), "b_from_left".to_string()),
+ (Arc::new(Column::new("a", 0)), "a_from_left".to_string()),
+ (Arc::new(Column::new("a", 5)), "a_from_right".to_string()),
+ (Arc::new(Column::new("c", 7)), "c_from_right".to_string()),
+ ],
+ join,
+ )?);
+ let initial = get_plan_string(&projection);
+ let expected_initial = [
+ "ProjectionExec: expr=[c@2 as c_from_left, b@1 as b_from_left, a@0 as a_from_left, a@5 as a_from_right, c@7 as c_from_right]",
+ " SymmetricHashJoinExec: mode=SinglePartition, join_type=Inner, on=[(b@1, c@2)], filter=b_left_inter@0 - 1 + a_right_inter@1 <= a_right_inter@1 + c_left_inter@2",
+ " CsvExec: file_groups={1 group: [[x]]}, projection=[a, b, c, d, e], has_header=false",
+ " CsvExec: file_groups={1 group: [[x]]}, projection=[a, b, c, d, e], has_header=false"
+ ];
+ assert_eq!(initial, expected_initial);
+
+ let after_optimize =
+ ProjectionPushdown::new().optimize(projection, &ConfigOptions::new())?;
+
+ let expected = [
+ "SymmetricHashJoinExec: mode=SinglePartition, join_type=Inner, on=[(b_from_left@1, c_from_right@1)], filter=b_left_inter@0 - 1 + a_right_inter@1 <= a_right_inter@1 + c_left_inter@2",
+ " ProjectionExec: expr=[c@2 as c_from_left, b@1 as b_from_left, a@0 as a_from_left]",
+ " CsvExec: file_groups={1 group: [[x]]}, projection=[a, b, c, d, e], has_header=false",
+ " ProjectionExec: expr=[a@0 as a_from_right, c@2 as c_from_right]",
+ " CsvExec: file_groups={1 group: [[x]]}, projection=[a, b, c, d, e], has_header=false"
+ ];
+ assert_eq!(get_plan_string(&after_optimize), expected);
+
+ let expected_filter_col_ind = vec![
+ ColumnIndex {
+ index: 1,
+ side: JoinSide::Left,
+ },
+ ColumnIndex {
+ index: 0,
+ side: JoinSide::Right,
+ },
+ ColumnIndex {
+ index: 0,
+ side: JoinSide::Left,
+ },
+ ];
+
+ assert_eq!(
+ expected_filter_col_ind,
+ after_optimize
+ .as_any()
+ .downcast_ref::<SymmetricHashJoinExec>()
+ .unwrap()
+ .filter()
+ .unwrap()
+ .column_indices()
+ );
+
+ Ok(())
+ }
+
+ #[test]
+ fn test_repartition_after_projection() -> Result<()> {
+ let csv = create_simple_csv_exec();
+ let repartition: Arc<dyn ExecutionPlan> = Arc::new(RepartitionExec::try_new(
+ csv,
+ Partitioning::Hash(
+ vec![
+ Arc::new(Column::new("a", 0)),
+ Arc::new(Column::new("b", 1)),
+ Arc::new(Column::new("d", 3)),
+ ],
+ 6,
+ ),
+ )?);
+ let projection: Arc<dyn ExecutionPlan> = Arc::new(ProjectionExec::try_new(
+ vec![
+ (Arc::new(Column::new("b", 1)), "b_new".to_string()),
+ (Arc::new(Column::new("a", 0)), "a".to_string()),
+ (Arc::new(Column::new("d", 3)), "d_new".to_string()),
+ ],
+ repartition,
+ )?);
+ let initial = get_plan_string(&projection);
+ let expected_initial = [
+ "ProjectionExec: expr=[b@1 as b_new, a@0 as a, d@3 as d_new]",
+ " RepartitionExec: partitioning=Hash([a@0, b@1, d@3], 6), input_partitions=1",
+ " CsvExec: file_groups={1 group: [[x]]}, projection=[a, b, c, d, e], has_header=false",
+ ];
+ assert_eq!(initial, expected_initial);
+
+ let after_optimize =
+ ProjectionPushdown::new().optimize(projection, &ConfigOptions::new())?;
+
+ let expected = [
+ "RepartitionExec: partitioning=Hash([a@1, b_new@0, d_new@2], 6), input_partitions=1",
+ " ProjectionExec: expr=[b@1 as b_new, a@0 as a, d@3 as d_new]",
+ " CsvExec: file_groups={1 group: [[x]]}, projection=[a, b, c, d, e], has_header=false",
+ ];
+ assert_eq!(get_plan_string(&after_optimize), expected);
+
+ assert_eq!(
+ after_optimize
+ .as_any()
+ .downcast_ref::<RepartitionExec>()
+ .unwrap()
+ .partitioning()
+ .clone(),
+ Partitioning::Hash(
+ vec![
+ Arc::new(Column::new("a", 1)),
+ Arc::new(Column::new("b_new", 0)),
+ Arc::new(Column::new("d_new", 2)),
+ ],
+ 6,
+ ),
+ );
+
+ Ok(())
+ }
+
+ #[test]
+ fn test_sort_after_projection() -> Result<()> {
+ let csv = create_simple_csv_exec();
+ let sort_req: Arc<dyn ExecutionPlan> = Arc::new(SortExec::new(
+ vec![
+ PhysicalSortExpr {
+ expr: Arc::new(Column::new("b", 1)),
+ options: SortOptions::default(),
+ },
+ PhysicalSortExpr {
+ expr: Arc::new(BinaryExpr::new(
+ Arc::new(Column::new("c", 2)),
+ Operator::Plus,
+ Arc::new(Column::new("a", 0)),
+ )),
+ options: SortOptions::default(),
+ },
+ ],
+ csv.clone(),
+ ));
+ let projection: Arc<dyn ExecutionPlan> = Arc::new(ProjectionExec::try_new(
+ vec![
+ (Arc::new(Column::new("c", 2)), "c".to_string()),
+ (Arc::new(Column::new("a", 0)), "new_a".to_string()),
+ (Arc::new(Column::new("b", 1)), "b".to_string()),
+ ],
+ sort_req.clone(),
+ )?);
+
+ let initial = get_plan_string(&projection);
+ let expected_initial = [
+ "ProjectionExec: expr=[c@2 as c, a@0 as new_a, b@1 as b]",
+ " SortExec: expr=[b@1 ASC,c@2 + a@0 ASC]",
+ " CsvExec: file_groups={1 group: [[x]]}, projection=[a, b, c, d, e], has_header=false"
+ ];
+ assert_eq!(initial, expected_initial);
+
+ let after_optimize =
+ ProjectionPushdown::new().optimize(projection, &ConfigOptions::new())?;
+
+ let expected = [
+ "SortExec: expr=[b@2 ASC,c@0 + new_a@1 ASC]",
+ " ProjectionExec: expr=[c@2 as c, a@0 as new_a, b@1 as b]",
+ " CsvExec: file_groups={1 group: [[x]]}, projection=[a, b, c, d, e], has_header=false"
+ ];
+ assert_eq!(get_plan_string(&after_optimize), expected);
+
+ Ok(())
+ }
+
+ #[test]
+ fn test_sort_preserving_after_projection() -> Result<()> {
+ let csv = create_simple_csv_exec();
+ let sort_req: Arc<dyn ExecutionPlan> = Arc::new(SortPreservingMergeExec::new(
+ vec![
+ PhysicalSortExpr {
+ expr: Arc::new(Column::new("b", 1)),
+ options: SortOptions::default(),
+ },
+ PhysicalSortExpr {
+ expr: Arc::new(BinaryExpr::new(
+ Arc::new(Column::new("c", 2)),
+ Operator::Plus,
+ Arc::new(Column::new("a", 0)),
+ )),
+ options: SortOptions::default(),
+ },
+ ],
+ csv.clone(),
+ ));
+ let projection: Arc<dyn ExecutionPlan> = Arc::new(ProjectionExec::try_new(
+ vec![
+ (Arc::new(Column::new("c", 2)), "c".to_string()),
+ (Arc::new(Column::new("a", 0)), "new_a".to_string()),
+ (Arc::new(Column::new("b", 1)), "b".to_string()),
+ ],
+ sort_req.clone(),
+ )?);
+
+ let initial = get_plan_string(&projection);
+ let expected_initial = [
+ "ProjectionExec: expr=[c@2 as c, a@0 as new_a, b@1 as b]",
+ " SortPreservingMergeExec: [b@1 ASC,c@2 + a@0 ASC]",
+ " CsvExec: file_groups={1 group: [[x]]}, projection=[a, b, c, d, e], has_header=false"
+ ];
+ assert_eq!(initial, expected_initial);
+
+ let after_optimize =
+ ProjectionPushdown::new().optimize(projection, &ConfigOptions::new())?;
+
+ let expected = [
+ "SortPreservingMergeExec: [b@2 ASC,c@0 + new_a@1 ASC]",
+ " ProjectionExec: expr=[c@2 as c, a@0 as new_a, b@1 as b]",
+ " CsvExec: file_groups={1 group: [[x]]}, projection=[a, b, c, d, e], has_header=false"
+ ];
+ assert_eq!(get_plan_string(&after_optimize), expected);
+
+ Ok(())
+ }
+
+ #[test]
+ fn test_union_after_projection() -> Result<()> {
+ let csv = create_simple_csv_exec();
+ let union: Arc<dyn ExecutionPlan> =
+ Arc::new(UnionExec::new(vec![csv.clone(), csv.clone(), csv]));
+ let projection: Arc<dyn ExecutionPlan> = Arc::new(ProjectionExec::try_new(
+ vec![
+ (Arc::new(Column::new("c", 2)), "c".to_string()),
+ (Arc::new(Column::new("a", 0)), "new_a".to_string()),
+ (Arc::new(Column::new("b", 1)), "b".to_string()),
+ ],
+ union.clone(),
+ )?);
+
+ let initial = get_plan_string(&projection);
+ let expected_initial = [
+ "ProjectionExec: expr=[c@2 as c, a@0 as new_a, b@1 as b]",
+ " UnionExec",
+ " CsvExec: file_groups={1 group: [[x]]}, projection=[a, b, c, d, e], has_header=false",
+ " CsvExec: file_groups={1 group: [[x]]}, projection=[a, b, c, d, e], has_header=false",
+ " CsvExec: file_groups={1 group: [[x]]}, projection=[a, b, c, d, e], has_header=false"
+ ];
+ assert_eq!(initial, expected_initial);
+
+ let after_optimize =
+ ProjectionPushdown::new().optimize(projection, &ConfigOptions::new())?;
+
+ let expected = [
+ "UnionExec",
+ " ProjectionExec: expr=[c@2 as c, a@0 as new_a, b@1 as b]",
+ " CsvExec: file_groups={1 group: [[x]]}, projection=[a, b, c, d, e], has_header=false",
+ " ProjectionExec: expr=[c@2 as c, a@0 as new_a, b@1 as b]",
+ " CsvExec: file_groups={1 group: [[x]]}, projection=[a, b, c, d, e], has_header=false",
+ " ProjectionExec: expr=[c@2 as c, a@0 as new_a, b@1 as b]",
+ " CsvExec: file_groups={1 group: [[x]]}, projection=[a, b, c, d, e], has_header=false"
+ ];
+ assert_eq!(get_plan_string(&after_optimize), expected);
+
+ Ok(())
+ }
+}
diff --git a/datafusion/physical-expr/src/expressions/cast.rs b/datafusion/physical-expr/src/expressions/cast.rs
index 9390089063..5d56af3646 100644
--- a/datafusion/physical-expr/src/expressions/cast.rs
+++ b/datafusion/physical-expr/src/expressions/cast.rs
@@ -73,6 +73,11 @@ impl CastExpr {
pub fn cast_type(&self) -> &DataType {
&self.cast_type
}
+
+ /// The cast options
+ pub fn cast_options(&self) -> &CastOptions<'static> {
+ &self.cast_options
+ }
}
impl fmt::Display for CastExpr {
diff --git a/datafusion/physical-expr/src/scalar_function.rs b/datafusion/physical-expr/src/scalar_function.rs
index 768aa04dd9..63101c03bc 100644
--- a/datafusion/physical-expr/src/scalar_function.rs
+++ b/datafusion/physical-expr/src/scalar_function.rs
@@ -108,6 +108,11 @@ impl ScalarFunctionExpr {
pub fn return_type(&self) -> &DataType {
&self.return_type
}
+
+ /// Monotonicity information of the function
+ pub fn monotonicity(&self) -> &Option<FuncMonotonicity> {
+ &self.monotonicity
+ }
}
impl fmt::Display for ScalarFunctionExpr {
diff --git a/datafusion/physical-plan/src/joins/symmetric_hash_join.rs b/datafusion/physical-plan/src/joins/symmetric_hash_join.rs
index 3617893a1c..39ac25ecb5 100644
--- a/datafusion/physical-plan/src/joins/symmetric_hash_join.rs
+++ b/datafusion/physical-plan/src/joins/symmetric_hash_join.rs
@@ -325,6 +325,11 @@ impl SymmetricHashJoinExec {
self.null_equals_null
}
+ /// Get partition mode
+ pub fn partition_mode(&self) -> StreamJoinPartitionMode {
+ self.mode
+ }
+
/// Check if order information covers every column in the filter expression.
pub fn check_if_order_information_available(&self) -> Result<bool> {
if let Some(filter) = self.filter() {
diff --git a/datafusion/physical-plan/src/joins/utils.rs b/datafusion/physical-plan/src/joins/utils.rs
index c91dc92fbc..5efeedfe65 100644
--- a/datafusion/physical-plan/src/joins/utils.rs
+++ b/datafusion/physical-plan/src/joins/utils.rs
@@ -219,7 +219,7 @@ pub fn calculate_join_output_ordering(
}
/// Information about the index and placement (left or right) of the columns
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, PartialEq)]
pub struct ColumnIndex {
/// Index of the column
pub index: usize,
diff --git a/datafusion/physical-plan/src/memory.rs b/datafusion/physical-plan/src/memory.rs
index 5f1660a225..39cd47452e 100644
--- a/datafusion/physical-plan/src/memory.rs
+++ b/datafusion/physical-plan/src/memory.rs
@@ -177,6 +177,14 @@ impl MemoryExec {
})
}
+ pub fn partitions(&self) -> &[Vec<RecordBatch>] {
+ &self.partitions
+ }
+
+ pub fn projection(&self) -> &Option<Vec<usize>> {
+ &self.projection
+ }
+
/// A memory table can be ordered by multiple expressions simultaneously.
/// [`EquivalenceProperties`] keeps track of expressions that describe the
/// global ordering of the schema. These columns are not necessarily same; e.g.
@@ -197,6 +205,10 @@ impl MemoryExec {
self.sort_information = sort_information;
self
}
+
+ pub fn original_schema(&self) -> SchemaRef {
+ self.schema.clone()
+ }
}
/// Iterator over batches
diff --git a/datafusion/physical-plan/src/projection.rs b/datafusion/physical-plan/src/projection.rs
index c5d94b08e0..bbf0d6d4b3 100644
--- a/datafusion/physical-plan/src/projection.rs
+++ b/datafusion/physical-plan/src/projection.rs
@@ -46,7 +46,7 @@ use futures::stream::{Stream, StreamExt};
use log::trace;
/// Execution plan for a projection
-#[derive(Debug)]
+#[derive(Debug, Clone)]
pub struct ProjectionExec {
/// The projection expressions stored as tuples of (expression, output column name)
pub(crate) expr: Vec<(Arc<dyn PhysicalExpr>, String)>,
diff --git a/datafusion/sqllogictest/test_files/explain.slt b/datafusion/sqllogictest/test_files/explain.slt
index d28f9fc6e3..49bb63d75d 100644
--- a/datafusion/sqllogictest/test_files/explain.slt
+++ b/datafusion/sqllogictest/test_files/explain.slt
@@ -169,9 +169,8 @@ Dml: op=[Insert Into] table=[sink_table]
------TableScan: aggregate_test_100 projection=[c1, c2, c3, c4, c5, c6, c7, c8, c9, c10, c11, c12, c13]
physical_plan
FileSinkExec: sink=CsvSink(writer_mode=Append, file_groups=[WORKSPACE_ROOT/testing/data/csv/aggregate_test_100.csv])
---ProjectionExec: expr=[c1@0 as c1, c2@1 as c2, c3@2 as c3, c4@3 as c4, c5@4 as c5, c6@5 as c6, c7@6 as c7, c8@7 as c8, c9@8 as c9, c10@9 as c10, c11@10 as c11, c12@11 as c12, c13@12 as c13]
-----SortExec: expr=[c1@0 ASC NULLS LAST]
-------CsvExec: file_groups={1 group: [[WORKSPACE_ROOT/testing/data/csv/aggregate_test_100.csv]]}, projection=[c1, c2, c3, c4, c5, c6, c7, c8, c9, c10, c11, c12, c13], has_header=true
+--SortExec: expr=[c1@0 ASC NULLS LAST]
+----CsvExec: file_groups={1 group: [[WORKSPACE_ROOT/testing/data/csv/aggregate_test_100.csv]]}, projection=[c1, c2, c3, c4, c5, c6, c7, c8, c9, c10, c11, c12, c13], has_header=true
# test EXPLAIN VERBOSE
query TT
@@ -258,6 +257,7 @@ physical_plan after coalesce_batches SAME TEXT AS ABOVE
physical_plan after OutputRequirements CsvExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/example.csv]]}, projection=[a, b, c], has_header=true
physical_plan after PipelineChecker SAME TEXT AS ABOVE
physical_plan after LimitAggregation SAME TEXT AS ABOVE
+physical_plan after ProjectionPushdown SAME TEXT AS ABOVE
physical_plan CsvExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/example.csv]]}, projection=[a, b, c], has_header=true
diff --git a/datafusion/sqllogictest/test_files/groupby.slt b/datafusion/sqllogictest/test_files/groupby.slt
index 000c3dc3b5..105f11f216 100644
--- a/datafusion/sqllogictest/test_files/groupby.slt
+++ b/datafusion/sqllogictest/test_files/groupby.slt
@@ -2084,9 +2084,7 @@ logical_plan
Projection: multiple_ordered_table.a
--Sort: multiple_ordered_table.c ASC NULLS LAST
----TableScan: multiple_ordered_table projection=[a, c]
-physical_plan
-ProjectionExec: expr=[a@0 as a]
---CsvExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/window_2.csv]]}, projection=[a, c], output_ordering=[a@0 ASC NULLS LAST], has_header=true
+physical_plan CsvExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/window_2.csv]]}, projection=[a], output_ordering=[a@0 ASC NULLS LAST], has_header=true
# Final plan shouldn't have SortExec a ASC, b ASC,
# because table already satisfies this ordering.
@@ -2097,9 +2095,7 @@ logical_plan
Projection: multiple_ordered_table.a
--Sort: multiple_ordered_table.a ASC NULLS LAST, multiple_ordered_table.b ASC NULLS LAST
----TableScan: multiple_ordered_table projection=[a, b]
-physical_plan
-ProjectionExec: expr=[a@0 as a]
---CsvExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/window_2.csv]]}, projection=[a, b], output_ordering=[a@0 ASC NULLS LAST, b@1 ASC NULLS LAST], has_header=true
+physical_plan CsvExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/window_2.csv]]}, projection=[a], output_ordering=[a@0 ASC NULLS LAST], has_header=true
# test_window_agg_sort
statement ok
@@ -3696,16 +3692,15 @@ Projection: amount_usd
----------------SubqueryAlias: r
------------------TableScan: multiple_ordered_table projection=[a, d]
physical_plan
-ProjectionExec: expr=[amount_usd@0 as amount_usd]
---ProjectionExec: expr=[LAST_VALUE(l.d) ORDER BY [l.a ASC NULLS LAST]@1 as amount_usd, row_n@0 as row_n]
-----AggregateExec: mode=Single, gby=[row_n@2 as row_n], aggr=[LAST_VALUE(l.d)], ordering_mode=Sorted
-------ProjectionExec: expr=[a@0 as a, d@1 as d, row_n@4 as row_n]
---------CoalesceBatchesExec: target_batch_size=2
-----------HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(d@1, d@1)], filter=CAST(a@0 AS Int64) >= CAST(a@1 AS Int64) - 10
-------------CsvExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/window_2.csv]]}, projection=[a, d], output_ordering=[a@0 ASC NULLS LAST], has_header=true
-------------ProjectionExec: expr=[a@0 as a, d@1 as d, ROW_NUMBER() ORDER BY [r.a ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW@2 as row_n]
---------------BoundedWindowAggExec: wdw=[ROW_NUMBER() ORDER BY [r.a ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW: Ok(Field { name: "ROW_NUMBER() ORDER BY [r.a ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW", data_type: UInt64, nullable: false, dict_id: 0, dict_is_ordered: false, metadata: {} }), frame: WindowFrame { units: Range, start_bound: Preceding(Int32(NULL)), end_bound: CurrentRow }], mode=[Sorted]
-----------------CsvExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/window_2.csv]]}, projection=[a, d], output_ordering=[a@0 ASC NULLS LAST], has_header=true
+ProjectionExec: expr=[LAST_VALUE(l.d) ORDER BY [l.a ASC NULLS LAST]@1 as amount_usd]
+--AggregateExec: mode=Single, gby=[row_n@2 as row_n], aggr=[LAST_VALUE(l.d)], ordering_mode=Sorted
+----ProjectionExec: expr=[a@0 as a, d@1 as d, row_n@4 as row_n]
+------CoalesceBatchesExec: target_batch_size=2
+--------HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(d@1, d@1)], filter=CAST(a@0 AS Int64) >= CAST(a@1 AS Int64) - 10
+----------CsvExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/window_2.csv]]}, projection=[a, d], output_ordering=[a@0 ASC NULLS LAST], has_header=true
+----------ProjectionExec: expr=[a@0 as a, d@1 as d, ROW_NUMBER() ORDER BY [r.a ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW@2 as row_n]
+------------BoundedWindowAggExec: wdw=[ROW_NUMBER() ORDER BY [r.a ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW: Ok(Field { name: "ROW_NUMBER() ORDER BY [r.a ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW", data_type: UInt64, nullable: false, dict_id: 0, dict_is_ordered: false, metadata: {} }), frame: WindowFrame { units: Range, start_bound: Preceding(Int32(NULL)), end_bound: CurrentRow }], mode=[Sorted]
+--------------CsvExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/window_2.csv]]}, projection=[a, d], output_ordering=[a@0 ASC NULLS LAST], has_header=true
# reset partition number to 8.
statement ok
diff --git a/datafusion/sqllogictest/test_files/insert.slt b/datafusion/sqllogictest/test_files/insert.slt
index 0c63a34819..8b9fd52e0d 100644
--- a/datafusion/sqllogictest/test_files/insert.slt
+++ b/datafusion/sqllogictest/test_files/insert.slt
@@ -218,9 +218,8 @@ Dml: op=[Insert Into] table=[table_without_values]
------TableScan: aggregate_test_100 projection=[c1]
physical_plan
FileSinkExec: sink=MemoryTable (partitions=1)
---ProjectionExec: expr=[c1@0 as c1]
-----SortExec: expr=[c1@0 ASC NULLS LAST]
-------CsvExec: file_groups={1 group: [[WORKSPACE_ROOT/testing/data/csv/aggregate_test_100.csv]]}, projection=[c1], has_header=true
+--SortExec: expr=[c1@0 ASC NULLS LAST]
+----CsvExec: file_groups={1 group: [[WORKSPACE_ROOT/testing/data/csv/aggregate_test_100.csv]]}, projection=[c1], has_header=true
query T
insert into table_without_values select c1 from aggregate_test_100 order by c1;
diff --git a/datafusion/sqllogictest/test_files/insert_to_external.slt b/datafusion/sqllogictest/test_files/insert_to_external.slt
index fa1d646d14..d6449bc272 100644
--- a/datafusion/sqllogictest/test_files/insert_to_external.slt
+++ b/datafusion/sqllogictest/test_files/insert_to_external.slt
@@ -423,9 +423,8 @@ Dml: op=[Insert Into] table=[table_without_values]
------TableScan: aggregate_test_100 projection=[c1]
physical_plan
FileSinkExec: sink=ParquetSink(writer_mode=PutMultipart, file_groups=[])
---ProjectionExec: expr=[c1@0 as c1]
-----SortExec: expr=[c1@0 ASC NULLS LAST]
-------CsvExec: file_groups={1 group: [[WORKSPACE_ROOT/testing/data/csv/aggregate_test_100.csv]]}, projection=[c1], has_header=true
+--SortExec: expr=[c1@0 ASC NULLS LAST]
+----CsvExec: file_groups={1 group: [[WORKSPACE_ROOT/testing/data/csv/aggregate_test_100.csv]]}, projection=[c1], has_header=true
query T
insert into table_without_values select c1 from aggregate_test_100 order by c1;
diff --git a/datafusion/sqllogictest/test_files/joins.slt b/datafusion/sqllogictest/test_files/joins.slt
index 25ab2032f0..24893297f1 100644
--- a/datafusion/sqllogictest/test_files/joins.slt
+++ b/datafusion/sqllogictest/test_files/joins.slt
@@ -1441,17 +1441,16 @@ Projection: join_t1.t1_id, join_t1.t1_name, join_t1.t1_int, join_t2.t2_id, join_
----TableScan: join_t1 projection=[t1_id, t1_name, t1_int]
----TableScan: join_t2 projection=[t2_id, t2_name, t2_int]
physical_plan
-ProjectionExec: expr=[t1_id@0 as t1_id, t1_name@1 as t1_name, t1_int@2 as t1_int, t2_id@3 as t2_id, t2_name@4 as t2_name, t2_int@5 as t2_int, CAST(t1_id@0 AS Int64) + 11 as join_t1.t1_id + Int64(11)]
---ProjectionExec: expr=[t1_id@0 as t1_id, t1_name@1 as t1_name, t1_int@2 as t1_int, t2_id@4 as t2_id, t2_name@5 as t2_name, t2_int@6 as t2_int]
-----CoalesceBatchesExec: target_batch_size=2
-------HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(join_t1.t1_id + Int64(11)@3, CAST(join_t2.t2_id AS Int64)@3)]
---------CoalescePartitionsExec
-----------ProjectionExec: expr=[t1_id@0 as t1_id, t1_name@1 as t1_name, t1_int@2 as t1_int, CAST(t1_id@0 AS Int64) + 11 as join_t1.t1_id + Int64(11)]
-------------RepartitionExec: partitioning=RoundRobinBatch(2), input_partitions=1
---------------MemoryExec: partitions=1, partition_sizes=[1]
---------ProjectionExec: expr=[t2_id@0 as t2_id, t2_name@1 as t2_name, t2_int@2 as t2_int, CAST(t2_id@0 AS Int64) as CAST(join_t2.t2_id AS Int64)]
+ProjectionExec: expr=[t1_id@0 as t1_id, t1_name@1 as t1_name, t1_int@2 as t1_int, t2_id@4 as t2_id, t2_name@5 as t2_name, t2_int@6 as t2_int, CAST(t1_id@0 AS Int64) + 11 as join_t1.t1_id + Int64(11)]
+--CoalesceBatchesExec: target_batch_size=2
+----HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(join_t1.t1_id + Int64(11)@3, CAST(join_t2.t2_id AS Int64)@3)]
+------CoalescePartitionsExec
+--------ProjectionExec: expr=[t1_id@0 as t1_id, t1_name@1 as t1_name, t1_int@2 as t1_int, CAST(t1_id@0 AS Int64) + 11 as join_t1.t1_id + Int64(11)]
----------RepartitionExec: partitioning=RoundRobinBatch(2), input_partitions=1
------------MemoryExec: partitions=1, partition_sizes=[1]
+------ProjectionExec: expr=[t2_id@0 as t2_id, t2_name@1 as t2_name, t2_int@2 as t2_int, CAST(t2_id@0 AS Int64) as CAST(join_t2.t2_id AS Int64)]
+--------RepartitionExec: partitioning=RoundRobinBatch(2), input_partitions=1
+----------MemoryExec: partitions=1, partition_sizes=[1]
statement ok
set datafusion.optimizer.repartition_joins = true;
@@ -1468,20 +1467,19 @@ Projection: join_t1.t1_id, join_t1.t1_name, join_t1.t1_int, join_t2.t2_id, join_
----TableScan: join_t1 projection=[t1_id, t1_name, t1_int]
----TableScan: join_t2 projection=[t2_id, t2_name, t2_int]
physical_plan
-ProjectionExec: expr=[t1_id@0 as t1_id, t1_name@1 as t1_name, t1_int@2 as t1_int, t2_id@3 as t2_id, t2_name@4 as t2_name, t2_int@5 as t2_int, CAST(t1_id@0 AS Int64) + 11 as join_t1.t1_id + Int64(11)]
---ProjectionExec: expr=[t1_id@0 as t1_id, t1_name@1 as t1_name, t1_int@2 as t1_int, t2_id@4 as t2_id, t2_name@5 as t2_name, t2_int@6 as t2_int]
-----CoalesceBatchesExec: target_batch_size=2
-------HashJoinExec: mode=Partitioned, join_type=Inner, on=[(join_t1.t1_id + Int64(11)@3, CAST(join_t2.t2_id AS Int64)@3)]
---------CoalesceBatchesExec: target_batch_size=2
-----------RepartitionExec: partitioning=Hash([join_t1.t1_id + Int64(11)@3], 2), input_partitions=2
-------------ProjectionExec: expr=[t1_id@0 as t1_id, t1_name@1 as t1_name, t1_int@2 as t1_int, CAST(t1_id@0 AS Int64) + 11 as join_t1.t1_id + Int64(11)]
---------------RepartitionExec: partitioning=RoundRobinBatch(2), input_partitions=1
-----------------MemoryExec: partitions=1, partition_sizes=[1]
---------CoalesceBatchesExec: target_batch_size=2
-----------RepartitionExec: partitioning=Hash([CAST(join_t2.t2_id AS Int64)@3], 2), input_partitions=2
-------------ProjectionExec: expr=[t2_id@0 as t2_id, t2_name@1 as t2_name, t2_int@2 as t2_int, CAST(t2_id@0 AS Int64) as CAST(join_t2.t2_id AS Int64)]
---------------RepartitionExec: partitioning=RoundRobinBatch(2), input_partitions=1
-----------------MemoryExec: partitions=1, partition_sizes=[1]
+ProjectionExec: expr=[t1_id@0 as t1_id, t1_name@1 as t1_name, t1_int@2 as t1_int, t2_id@4 as t2_id, t2_name@5 as t2_name, t2_int@6 as t2_int, CAST(t1_id@0 AS Int64) + 11 as join_t1.t1_id + Int64(11)]
+--CoalesceBatchesExec: target_batch_size=2
+----HashJoinExec: mode=Partitioned, join_type=Inner, on=[(join_t1.t1_id + Int64(11)@3, CAST(join_t2.t2_id AS Int64)@3)]
+------CoalesceBatchesExec: target_batch_size=2
+--------RepartitionExec: partitioning=Hash([join_t1.t1_id + Int64(11)@3], 2), input_partitions=2
+----------ProjectionExec: expr=[t1_id@0 as t1_id, t1_name@1 as t1_name, t1_int@2 as t1_int, CAST(t1_id@0 AS Int64) + 11 as join_t1.t1_id + Int64(11)]
+------------RepartitionExec: partitioning=RoundRobinBatch(2), input_partitions=1
+--------------MemoryExec: partitions=1, partition_sizes=[1]
+------CoalesceBatchesExec: target_batch_size=2
+--------RepartitionExec: partitioning=Hash([CAST(join_t2.t2_id AS Int64)@3], 2), input_partitions=2
+----------ProjectionExec: expr=[t2_id@0 as t2_id, t2_name@1 as t2_name, t2_int@2 as t2_int, CAST(t2_id@0 AS Int64) as CAST(join_t2.t2_id AS Int64)]
+------------RepartitionExec: partitioning=RoundRobinBatch(2), input_partitions=1
+--------------MemoryExec: partitions=1, partition_sizes=[1]
# Both side expr key inner join
@@ -1500,18 +1498,16 @@ Projection: join_t1.t1_id, join_t2.t2_id, join_t1.t1_name
----TableScan: join_t1 projection=[t1_id, t1_name]
----TableScan: join_t2 projection=[t2_id]
physical_plan
-ProjectionExec: expr=[t1_id@0 as t1_id, t2_id@2 as t2_id, t1_name@1 as t1_name]
---ProjectionExec: expr=[t1_id@0 as t1_id, t1_name@1 as t1_name, t2_id@3 as t2_id]
-----ProjectionExec: expr=[t1_id@2 as t1_id, t1_name@3 as t1_name, join_t1.t1_id + UInt32(12)@4 as join_t1.t1_id + UInt32(12), t2_id@0 as t2_id, join_t2.t2_id + UInt32(1)@1 as join_t2.t2_id + UInt32(1)]
-------CoalesceBatchesExec: target_batch_size=2
---------HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(join_t2.t2_id + UInt32(1)@1, join_t1.t1_id + UInt32(12)@2)]
-----------CoalescePartitionsExec
-------------ProjectionExec: expr=[t2_id@0 as t2_id, t2_id@0 + 1 as join_t2.t2_id + UInt32(1)]
---------------RepartitionExec: partitioning=RoundRobinBatch(2), input_partitions=1
-----------------MemoryExec: partitions=1, partition_sizes=[1]
-----------ProjectionExec: expr=[t1_id@0 as t1_id, t1_name@1 as t1_name, t1_id@0 + 12 as join_t1.t1_id + UInt32(12)]
-------------RepartitionExec: partitioning=RoundRobinBatch(2), input_partitions=1
---------------MemoryExec: partitions=1, partition_sizes=[1]
+ProjectionExec: expr=[t1_id@2 as t1_id, t2_id@0 as t2_id, t1_name@3 as t1_name]
+--CoalesceBatchesExec: target_batch_size=2
+----HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(join_t2.t2_id + UInt32(1)@1, join_t1.t1_id + UInt32(12)@2)]
+------CoalescePartitionsExec
+--------ProjectionExec: expr=[t2_id@0 as t2_id, t2_id@0 + 1 as join_t2.t2_id + UInt32(1)]
+----------RepartitionExec: partitioning=RoundRobinBatch(2), input_partitions=1
+------------MemoryExec: partitions=1, partition_sizes=[1]
+------ProjectionExec: expr=[t1_id@0 as t1_id, t1_name@1 as t1_name, t1_id@0 + 12 as join_t1.t1_id + UInt32(12)]
+--------RepartitionExec: partitioning=RoundRobinBatch(2), input_partitions=1
+----------MemoryExec: partitions=1, partition_sizes=[1]
statement ok
set datafusion.optimizer.repartition_joins = true;
@@ -1528,21 +1524,19 @@ Projection: join_t1.t1_id, join_t2.t2_id, join_t1.t1_name
----TableScan: join_t1 projection=[t1_id, t1_name]
----TableScan: join_t2 projection=[t2_id]
physical_plan
-ProjectionExec: expr=[t1_id@0 as t1_id, t2_id@2 as t2_id, t1_name@1 as t1_name]
---ProjectionExec: expr=[t1_id@0 as t1_id, t1_name@1 as t1_name, t2_id@3 as t2_id]
-----ProjectionExec: expr=[t1_id@2 as t1_id, t1_name@3 as t1_name, join_t1.t1_id + UInt32(12)@4 as join_t1.t1_id + UInt32(12), t2_id@0 as t2_id, join_t2.t2_id + UInt32(1)@1 as join_t2.t2_id + UInt32(1)]
+ProjectionExec: expr=[t1_id@2 as t1_id, t2_id@0 as t2_id, t1_name@3 as t1_name]
+--CoalesceBatchesExec: target_batch_size=2
+----HashJoinExec: mode=Partitioned, join_type=Inner, on=[(join_t2.t2_id + UInt32(1)@1, join_t1.t1_id + UInt32(12)@2)]
------CoalesceBatchesExec: target_batch_size=2
---------HashJoinExec: mode=Partitioned, join_type=Inner, on=[(join_t2.t2_id + UInt32(1)@1, join_t1.t1_id + UInt32(12)@2)]
-----------CoalesceBatchesExec: target_batch_size=2
-------------RepartitionExec: partitioning=Hash([join_t2.t2_id + UInt32(1)@1], 2), input_partitions=2
---------------ProjectionExec: expr=[t2_id@0 as t2_id, t2_id@0 + 1 as join_t2.t2_id + UInt32(1)]
-----------------RepartitionExec: partitioning=RoundRobinBatch(2), input_partitions=1
-------------------MemoryExec: partitions=1, partition_sizes=[1]
-----------CoalesceBatchesExec: target_batch_size=2
-------------RepartitionExec: partitioning=Hash([join_t1.t1_id + UInt32(12)@2], 2), input_partitions=2
---------------ProjectionExec: expr=[t1_id@0 as t1_id, t1_name@1 as t1_name, t1_id@0 + 12 as join_t1.t1_id + UInt32(12)]
-----------------RepartitionExec: partitioning=RoundRobinBatch(2), input_partitions=1
-------------------MemoryExec: partitions=1, partition_sizes=[1]
+--------RepartitionExec: partitioning=Hash([join_t2.t2_id + UInt32(1)@1], 2), input_partitions=2
+----------ProjectionExec: expr=[t2_id@0 as t2_id, t2_id@0 + 1 as join_t2.t2_id + UInt32(1)]
+------------RepartitionExec: partitioning=RoundRobinBatch(2), input_partitions=1
+--------------MemoryExec: partitions=1, partition_sizes=[1]
+------CoalesceBatchesExec: target_batch_size=2
+--------RepartitionExec: partitioning=Hash([join_t1.t1_id + UInt32(12)@2], 2), input_partitions=2
+----------ProjectionExec: expr=[t1_id@0 as t1_id, t1_name@1 as t1_name, t1_id@0 + 12 as join_t1.t1_id + UInt32(12)]
+------------RepartitionExec: partitioning=RoundRobinBatch(2), input_partitions=1
+--------------MemoryExec: partitions=1, partition_sizes=[1]
# Left side expr key inner join
@@ -1562,16 +1556,15 @@ Projection: join_t1.t1_id, join_t2.t2_id, join_t1.t1_name
----TableScan: join_t1 projection=[t1_id, t1_name]
----TableScan: join_t2 projection=[t2_id]
physical_plan
-ProjectionExec: expr=[t1_id@0 as t1_id, t2_id@2 as t2_id, t1_name@1 as t1_name]
---ProjectionExec: expr=[t1_id@0 as t1_id, t1_name@1 as t1_name, t2_id@3 as t2_id]
-----CoalesceBatchesExec: target_batch_size=2
-------HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(join_t1.t1_id + UInt32(11)@2, t2_id@0)]
---------CoalescePartitionsExec
-----------ProjectionExec: expr=[t1_id@0 as t1_id, t1_name@1 as t1_name, t1_id@0 + 11 as join_t1.t1_id + UInt32(11)]
-------------RepartitionExec: partitioning=RoundRobinBatch(2), input_partitions=1
---------------MemoryExec: partitions=1, partition_sizes=[1]
---------RepartitionExec: partitioning=RoundRobinBatch(2), input_partitions=1
-----------MemoryExec: partitions=1, partition_sizes=[1]
+ProjectionExec: expr=[t1_id@0 as t1_id, t2_id@3 as t2_id, t1_name@1 as t1_name]
+--CoalesceBatchesExec: target_batch_size=2
+----HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(join_t1.t1_id + UInt32(11)@2, t2_id@0)]
+------CoalescePartitionsExec
+--------ProjectionExec: expr=[t1_id@0 as t1_id, t1_name@1 as t1_name, t1_id@0 + 11 as join_t1.t1_id + UInt32(11)]
+----------RepartitionExec: partitioning=RoundRobinBatch(2), input_partitions=1
+------------MemoryExec: partitions=1, partition_sizes=[1]
+------RepartitionExec: partitioning=RoundRobinBatch(2), input_partitions=1
+--------MemoryExec: partitions=1, partition_sizes=[1]
statement ok
set datafusion.optimizer.repartition_joins = true;
@@ -1589,19 +1582,18 @@ Projection: join_t1.t1_id, join_t2.t2_id, join_t1.t1_name
----TableScan: join_t1 projection=[t1_id, t1_name]
----TableScan: join_t2 projection=[t2_id]
physical_plan
-ProjectionExec: expr=[t1_id@0 as t1_id, t2_id@2 as t2_id, t1_name@1 as t1_name]
---ProjectionExec: expr=[t1_id@0 as t1_id, t1_name@1 as t1_name, t2_id@3 as t2_id]
-----CoalesceBatchesExec: target_batch_size=2
-------HashJoinExec: mode=Partitioned, join_type=Inner, on=[(join_t1.t1_id + UInt32(11)@2, t2_id@0)]
---------CoalesceBatchesExec: target_batch_size=2
-----------RepartitionExec: partitioning=Hash([join_t1.t1_id + UInt32(11)@2], 2), input_partitions=2
-------------ProjectionExec: expr=[t1_id@0 as t1_id, t1_name@1 as t1_name, t1_id@0 + 11 as join_t1.t1_id + UInt32(11)]
---------------RepartitionExec: partitioning=RoundRobinBatch(2), input_partitions=1
-----------------MemoryExec: partitions=1, partition_sizes=[1]
---------CoalesceBatchesExec: target_batch_size=2
-----------RepartitionExec: partitioning=Hash([t2_id@0], 2), input_partitions=2
+ProjectionExec: expr=[t1_id@0 as t1_id, t2_id@3 as t2_id, t1_name@1 as t1_name]
+--CoalesceBatchesExec: target_batch_size=2
+----HashJoinExec: mode=Partitioned, join_type=Inner, on=[(join_t1.t1_id + UInt32(11)@2, t2_id@0)]
+------CoalesceBatchesExec: target_batch_size=2
+--------RepartitionExec: partitioning=Hash([join_t1.t1_id + UInt32(11)@2], 2), input_partitions=2
+----------ProjectionExec: expr=[t1_id@0 as t1_id, t1_name@1 as t1_name, t1_id@0 + 11 as join_t1.t1_id + UInt32(11)]
------------RepartitionExec: partitioning=RoundRobinBatch(2), input_partitions=1
--------------MemoryExec: partitions=1, partition_sizes=[1]
+------CoalesceBatchesExec: target_batch_size=2
+--------RepartitionExec: partitioning=Hash([t2_id@0], 2), input_partitions=2
+----------RepartitionExec: partitioning=RoundRobinBatch(2), input_partitions=1
+------------MemoryExec: partitions=1, partition_sizes=[1]
# Right side expr key inner join
@@ -1621,17 +1613,15 @@ Projection: join_t1.t1_id, join_t2.t2_id, join_t1.t1_name
----TableScan: join_t1 projection=[t1_id, t1_name]
----TableScan: join_t2 projection=[t2_id]
physical_plan
-ProjectionExec: expr=[t1_id@0 as t1_id, t2_id@2 as t2_id, t1_name@1 as t1_name]
---ProjectionExec: expr=[t1_id@0 as t1_id, t1_name@1 as t1_name, t2_id@2 as t2_id]
-----ProjectionExec: expr=[t1_id@2 as t1_id, t1_name@3 as t1_name, t2_id@0 as t2_id, join_t2.t2_id - UInt32(11)@1 as join_t2.t2_id - UInt32(11)]
-------CoalesceBatchesExec: target_batch_size=2
---------HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(join_t2.t2_id - UInt32(11)@1, t1_id@0)]
-----------CoalescePartitionsExec
-------------ProjectionExec: expr=[t2_id@0 as t2_id, t2_id@0 - 11 as join_t2.t2_id - UInt32(11)]
---------------RepartitionExec: partitioning=RoundRobinBatch(2), input_partitions=1
-----------------MemoryExec: partitions=1, partition_sizes=[1]
+ProjectionExec: expr=[t1_id@2 as t1_id, t2_id@0 as t2_id, t1_name@3 as t1_name]
+--CoalesceBatchesExec: target_batch_size=2
+----HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(join_t2.t2_id - UInt32(11)@1, t1_id@0)]
+------CoalescePartitionsExec
+--------ProjectionExec: expr=[t2_id@0 as t2_id, t2_id@0 - 11 as join_t2.t2_id - UInt32(11)]
----------RepartitionExec: partitioning=RoundRobinBatch(2), input_partitions=1
------------MemoryExec: partitions=1, partition_sizes=[1]
+------RepartitionExec: partitioning=RoundRobinBatch(2), input_partitions=1
+--------MemoryExec: partitions=1, partition_sizes=[1]
statement ok
set datafusion.optimizer.repartition_joins = true;
@@ -1649,20 +1639,18 @@ Projection: join_t1.t1_id, join_t2.t2_id, join_t1.t1_name
----TableScan: join_t1 projection=[t1_id, t1_name]
----TableScan: join_t2 projection=[t2_id]
physical_plan
-ProjectionExec: expr=[t1_id@0 as t1_id, t2_id@2 as t2_id, t1_name@1 as t1_name]
---ProjectionExec: expr=[t1_id@0 as t1_id, t1_name@1 as t1_name, t2_id@2 as t2_id]
-----ProjectionExec: expr=[t1_id@2 as t1_id, t1_name@3 as t1_name, t2_id@0 as t2_id, join_t2.t2_id - UInt32(11)@1 as join_t2.t2_id - UInt32(11)]
+ProjectionExec: expr=[t1_id@2 as t1_id, t2_id@0 as t2_id, t1_name@3 as t1_name]
+--CoalesceBatchesExec: target_batch_size=2
+----HashJoinExec: mode=Partitioned, join_type=Inner, on=[(join_t2.t2_id - UInt32(11)@1, t1_id@0)]
------CoalesceBatchesExec: target_batch_size=2
---------HashJoinExec: mode=Partitioned, join_type=Inner, on=[(join_t2.t2_id - UInt32(11)@1, t1_id@0)]
-----------CoalesceBatchesExec: target_batch_size=2
-------------RepartitionExec: partitioning=Hash([join_t2.t2_id - UInt32(11)@1], 2), input_partitions=2
---------------ProjectionExec: expr=[t2_id@0 as t2_id, t2_id@0 - 11 as join_t2.t2_id - UInt32(11)]
-----------------RepartitionExec: partitioning=RoundRobinBatch(2), input_partitions=1
-------------------MemoryExec: partitions=1, partition_sizes=[1]
-----------CoalesceBatchesExec: target_batch_size=2
-------------RepartitionExec: partitioning=Hash([t1_id@0], 2), input_partitions=2
---------------RepartitionExec: partitioning=RoundRobinBatch(2), input_partitions=1
-----------------MemoryExec: partitions=1, partition_sizes=[1]
+--------RepartitionExec: partitioning=Hash([join_t2.t2_id - UInt32(11)@1], 2), input_partitions=2
+----------ProjectionExec: expr=[t2_id@0 as t2_id, t2_id@0 - 11 as join_t2.t2_id - UInt32(11)]
+------------RepartitionExec: partitioning=RoundRobinBatch(2), input_partitions=1
+--------------MemoryExec: partitions=1, partition_sizes=[1]
+------CoalesceBatchesExec: target_batch_size=2
+--------RepartitionExec: partitioning=Hash([t1_id@0], 2), input_partitions=2
+----------RepartitionExec: partitioning=RoundRobinBatch(2), input_partitions=1
+------------MemoryExec: partitions=1, partition_sizes=[1]
# Select wildcard with expr key inner join
@@ -3347,16 +3335,15 @@ Projection: amount_usd
----------------SubqueryAlias: r
------------------TableScan: multiple_ordered_table projection=[a, d]
physical_plan
-ProjectionExec: expr=[amount_usd@0 as amount_usd]
---ProjectionExec: expr=[LAST_VALUE(l.d) ORDER BY [l.a ASC NULLS LAST]@1 as amount_usd, row_n@0 as row_n]
-----AggregateExec: mode=Single, gby=[row_n@2 as row_n], aggr=[LAST_VALUE(l.d)], ordering_mode=Sorted
-------ProjectionExec: expr=[a@0 as a, d@1 as d, row_n@4 as row_n]
---------CoalesceBatchesExec: target_batch_size=2
-----------HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(d@1, d@1)], filter=CAST(a@0 AS Int64) >= CAST(a@1 AS Int64) - 10
-------------CsvExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/window_2.csv]]}, projection=[a, d], output_ordering=[a@0 ASC NULLS LAST], has_header=true
-------------ProjectionExec: expr=[a@0 as a, d@1 as d, ROW_NUMBER() ORDER BY [r.a ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW@2 as row_n]
---------------BoundedWindowAggExec: wdw=[ROW_NUMBER() ORDER BY [r.a ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW: Ok(Field { name: "ROW_NUMBER() ORDER BY [r.a ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW", data_type: UInt64, nullable: false, dict_id: 0, dict_is_ordered: false, metadata: {} }), frame: WindowFrame { units: Range, start_bound: Preceding(Int32(NULL)), end_bound: CurrentRow }], mode=[Sorted]
-----------------CsvExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/window_2.csv]]}, projection=[a, d], output_ordering=[a@0 ASC NULLS LAST], has_header=true
+ProjectionExec: expr=[LAST_VALUE(l.d) ORDER BY [l.a ASC NULLS LAST]@1 as amount_usd]
+--AggregateExec: mode=Single, gby=[row_n@2 as row_n], aggr=[LAST_VALUE(l.d)], ordering_mode=Sorted
+----ProjectionExec: expr=[a@0 as a, d@1 as d, row_n@4 as row_n]
+------CoalesceBatchesExec: target_batch_size=2
+--------HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(d@1, d@1)], filter=CAST(a@0 AS Int64) >= CAST(a@1 AS Int64) - 10
+----------CsvExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/window_2.csv]]}, projection=[a, d], output_ordering=[a@0 ASC NULLS LAST], has_header=true
+----------ProjectionExec: expr=[a@0 as a, d@1 as d, ROW_NUMBER() ORDER BY [r.a ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW@2 as row_n]
+------------BoundedWindowAggExec: wdw=[ROW_NUMBER() ORDER BY [r.a ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW: Ok(Field { name: "ROW_NUMBER() ORDER BY [r.a ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW", data_type: UInt64, nullable: false, dict_id: 0, dict_is_ordered: false, metadata: {} }), frame: WindowFrame { units: Range, start_bound: Preceding(Int32(NULL)), end_bound: CurrentRow }], mode=[Sorted]
+--------------CsvExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/window_2.csv]]}, projection=[a, d], output_ordering=[a@0 ASC NULLS LAST], has_header=true
# run query above in multiple partitions
statement ok
diff --git a/datafusion/sqllogictest/test_files/subquery.slt b/datafusion/sqllogictest/test_files/subquery.slt
index 822a70bb5b..ef08c88a9d 100644
--- a/datafusion/sqllogictest/test_files/subquery.slt
+++ b/datafusion/sqllogictest/test_files/subquery.slt
@@ -180,19 +180,18 @@ Projection: t1.t1_id, __scalar_sq_1.SUM(t2.t2_int) AS t2_sum
--------Aggregate: groupBy=[[t2.t2_id]], aggr=[[SUM(CAST(t2.t2_int AS Int64))]]
----------TableScan: t2 projection=[t2_id, t2_int]
physical_plan
-ProjectionExec: expr=[t1_id@0 as t1_id, SUM(t2.t2_int)@1 as t2_sum]
---ProjectionExec: expr=[t1_id@2 as t1_id, SUM(t2.t2_int)@0 as SUM(t2.t2_int), t2_id@1 as t2_id]
-----CoalesceBatchesExec: target_batch_size=2
-------HashJoinExec: mode=Partitioned, join_type=Right, on=[(t2_id@1, t1_id@0)]
---------ProjectionExec: expr=[SUM(t2.t2_int)@1 as SUM(t2.t2_int), t2_id@0 as t2_id]
-----------AggregateExec: mode=FinalPartitioned, gby=[t2_id@0 as t2_id], aggr=[SUM(t2.t2_int)]
-------------CoalesceBatchesExec: target_batch_size=2
---------------RepartitionExec: partitioning=Hash([t2_id@0], 4), input_partitions=4
-----------------AggregateExec: mode=Partial, gby=[t2_id@0 as t2_id], aggr=[SUM(t2.t2_int)]
-------------------MemoryExec: partitions=4, partition_sizes=[1, 0, 0, 0]
---------CoalesceBatchesExec: target_batch_size=2
-----------RepartitionExec: partitioning=Hash([t1_id@0], 4), input_partitions=4
-------------MemoryExec: partitions=4, partition_sizes=[1, 0, 0, 0]
+ProjectionExec: expr=[t1_id@2 as t1_id, SUM(t2.t2_int)@0 as t2_sum]
+--CoalesceBatchesExec: target_batch_size=2
+----HashJoinExec: mode=Partitioned, join_type=Right, on=[(t2_id@1, t1_id@0)]
+------ProjectionExec: expr=[SUM(t2.t2_int)@1 as SUM(t2.t2_int), t2_id@0 as t2_id]
+--------AggregateExec: mode=FinalPartitioned, gby=[t2_id@0 as t2_id], aggr=[SUM(t2.t2_int)]
+----------CoalesceBatchesExec: target_batch_size=2
+------------RepartitionExec: partitioning=Hash([t2_id@0], 4), input_partitions=4
+--------------AggregateExec: mode=Partial, gby=[t2_id@0 as t2_id], aggr=[SUM(t2.t2_int)]
+----------------MemoryExec: partitions=4, partition_sizes=[1, 0, 0, 0]
+------CoalesceBatchesExec: target_batch_size=2
+--------RepartitionExec: partitioning=Hash([t1_id@0], 4), input_partitions=4
+----------MemoryExec: partitions=4, partition_sizes=[1, 0, 0, 0]
query II rowsort
SELECT t1_id, (SELECT sum(t2_int) FROM t2 WHERE t2.t2_id = t1.t1_id) as t2_sum from t1
@@ -215,19 +214,18 @@ Projection: t1.t1_id, __scalar_sq_1.SUM(t2.t2_int * Float64(1)) + Int64(1) AS t2
--------Aggregate: groupBy=[[t2.t2_id]], aggr=[[SUM(CAST(t2.t2_int AS Float64)) AS SUM(t2.t2_int * Float64(1))]]
----------TableScan: t2 projection=[t2_id, t2_int]
physical_plan
-ProjectionExec: expr=[t1_id@0 as t1_id, SUM(t2.t2_int * Float64(1)) + Int64(1)@1 as t2_sum]
---ProjectionExec: expr=[t1_id@2 as t1_id, SUM(t2.t2_int * Float64(1)) + Int64(1)@0 as SUM(t2.t2_int * Float64(1)) + Int64(1), t2_id@1 as t2_id]
-----CoalesceBatchesExec: target_batch_size=2
-------HashJoinExec: mode=Partitioned, join_type=Right, on=[(t2_id@1, t1_id@0)]
---------ProjectionExec: expr=[SUM(t2.t2_int * Float64(1))@1 + 1 as SUM(t2.t2_int * Float64(1)) + Int64(1), t2_id@0 as t2_id]
-----------AggregateExec: mode=FinalPartitioned, gby=[t2_id@0 as t2_id], aggr=[SUM(t2.t2_int * Float64(1))]
-------------CoalesceBatchesExec: target_batch_size=2
---------------RepartitionExec: partitioning=Hash([t2_id@0], 4), input_partitions=4
-----------------AggregateExec: mode=Partial, gby=[t2_id@0 as t2_id], aggr=[SUM(t2.t2_int * Float64(1))]
-------------------MemoryExec: partitions=4, partition_sizes=[1, 0, 0, 0]
---------CoalesceBatchesExec: target_batch_size=2
-----------RepartitionExec: partitioning=Hash([t1_id@0], 4), input_partitions=4
-------------MemoryExec: partitions=4, partition_sizes=[1, 0, 0, 0]
+ProjectionExec: expr=[t1_id@2 as t1_id, SUM(t2.t2_int * Float64(1)) + Int64(1)@0 as t2_sum]
+--CoalesceBatchesExec: target_batch_size=2
+----HashJoinExec: mode=Partitioned, join_type=Right, on=[(t2_id@1, t1_id@0)]
+------ProjectionExec: expr=[SUM(t2.t2_int * Float64(1))@1 + 1 as SUM(t2.t2_int * Float64(1)) + Int64(1), t2_id@0 as t2_id]
+--------AggregateExec: mode=FinalPartitioned, gby=[t2_id@0 as t2_id], aggr=[SUM(t2.t2_int * Float64(1))]
+----------CoalesceBatchesExec: target_batch_size=2
+------------RepartitionExec: partitioning=Hash([t2_id@0], 4), input_partitions=4
+--------------AggregateExec: mode=Partial, gby=[t2_id@0 as t2_id], aggr=[SUM(t2.t2_int * Float64(1))]
+----------------MemoryExec: partitions=4, partition_sizes=[1, 0, 0, 0]
+------CoalesceBatchesExec: target_batch_size=2
+--------RepartitionExec: partitioning=Hash([t1_id@0], 4), input_partitions=4
+----------MemoryExec: partitions=4, partition_sizes=[1, 0, 0, 0]
query IR rowsort
SELECT t1_id, (SELECT sum(t2_int * 1.0) + 1 FROM t2 WHERE t2.t2_id = t1.t1_id) as t2_sum from t1
@@ -287,21 +285,20 @@ Projection: t1.t1_id, __scalar_sq_1.SUM(t2.t2_int) AS t2_sum
----------Aggregate: groupBy=[[t2.t2_id]], aggr=[[SUM(CAST(t2.t2_int AS Int64))]]
------------TableScan: t2 projection=[t2_id, t2_int]
physical_plan
-ProjectionExec: expr=[t1_id@0 as t1_id, SUM(t2.t2_int)@1 as t2_sum]
---ProjectionExec: expr=[t1_id@2 as t1_id, SUM(t2.t2_int)@0 as SUM(t2.t2_int), t2_id@1 as t2_id]
-----CoalesceBatchesExec: target_batch_size=2
-------HashJoinExec: mode=Partitioned, join_type=Right, on=[(t2_id@1, t1_id@0)]
---------ProjectionExec: expr=[SUM(t2.t2_int)@1 as SUM(t2.t2_int), t2_id@0 as t2_id]
-----------CoalesceBatchesExec: target_batch_size=2
-------------FilterExec: SUM(t2.t2_int)@1 < 3
---------------AggregateExec: mode=FinalPartitioned, gby=[t2_id@0 as t2_id], aggr=[SUM(t2.t2_int)]
-----------------CoalesceBatchesExec: target_batch_size=2
-------------------RepartitionExec: partitioning=Hash([t2_id@0], 4), input_partitions=4
---------------------AggregateExec: mode=Partial, gby=[t2_id@0 as t2_id], aggr=[SUM(t2.t2_int)]
-----------------------MemoryExec: partitions=4, partition_sizes=[1, 0, 0, 0]
+ProjectionExec: expr=[t1_id@2 as t1_id, SUM(t2.t2_int)@0 as t2_sum]
+--CoalesceBatchesExec: target_batch_size=2
+----HashJoinExec: mode=Partitioned, join_type=Right, on=[(t2_id@1, t1_id@0)]
+------ProjectionExec: expr=[SUM(t2.t2_int)@1 as SUM(t2.t2_int), t2_id@0 as t2_id]
--------CoalesceBatchesExec: target_batch_size=2
-----------RepartitionExec: partitioning=Hash([t1_id@0], 4), input_partitions=4
-------------MemoryExec: partitions=4, partition_sizes=[1, 0, 0, 0]
+----------FilterExec: SUM(t2.t2_int)@1 < 3
+------------AggregateExec: mode=FinalPartitioned, gby=[t2_id@0 as t2_id], aggr=[SUM(t2.t2_int)]
+--------------CoalesceBatchesExec: target_batch_size=2
+----------------RepartitionExec: partitioning=Hash([t2_id@0], 4), input_partitions=4
+------------------AggregateExec: mode=Partial, gby=[t2_id@0 as t2_id], aggr=[SUM(t2.t2_int)]
+--------------------MemoryExec: partitions=4, partition_sizes=[1, 0, 0, 0]
+------CoalesceBatchesExec: target_batch_size=2
+--------RepartitionExec: partitioning=Hash([t1_id@0], 4), input_partitions=4
+----------MemoryExec: partitions=4, partition_sizes=[1, 0, 0, 0]
query II rowsort
SELECT t1_id, (SELECT sum(t2_int) FROM t2 WHERE t2.t2_id = t1.t1_id having sum(t2_int) < 3) as t2_sum from t1
diff --git a/datafusion/sqllogictest/test_files/union.slt b/datafusion/sqllogictest/test_files/union.slt
index 688774c906..0f255cdb9f 100644
--- a/datafusion/sqllogictest/test_files/union.slt
+++ b/datafusion/sqllogictest/test_files/union.slt
@@ -272,37 +272,36 @@ Union
------TableScan: t1 projection=[id, name]
physical_plan
UnionExec
---ProjectionExec: expr=[id@0 as id, name@1 as name]
-----CoalesceBatchesExec: target_batch_size=2
-------HashJoinExec: mode=Partitioned, join_type=LeftAnti, on=[(id@0, CAST(t2.id AS Int32)@2), (name@1, name@1)]
---------AggregateExec: mode=FinalPartitioned, gby=[id@0 as id, name@1 as name], aggr=[]
-----------CoalesceBatchesExec: target_batch_size=2
-------------RepartitionExec: partitioning=Hash([id@0, name@1], 4), input_partitions=4
---------------AggregateExec: mode=Partial, gby=[id@0 as id, name@1 as name], aggr=[]
-----------------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
-------------------MemoryExec: partitions=1, partition_sizes=[1]
+--CoalesceBatchesExec: target_batch_size=2
+----HashJoinExec: mode=Partitioned, join_type=LeftAnti, on=[(id@0, CAST(t2.id AS Int32)@2), (name@1, name@1)]
+------AggregateExec: mode=FinalPartitioned, gby=[id@0 as id, name@1 as name], aggr=[]
--------CoalesceBatchesExec: target_batch_size=2
-----------RepartitionExec: partitioning=Hash([CAST(t2.id AS Int32)@2, name@1], 4), input_partitions=4
-------------ProjectionExec: expr=[id@0 as id, name@1 as name, CAST(id@0 AS Int32) as CAST(t2.id AS Int32)]
+----------RepartitionExec: partitioning=Hash([id@0, name@1], 4), input_partitions=4
+------------AggregateExec: mode=Partial, gby=[id@0 as id, name@1 as name], aggr=[]
--------------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
----------------MemoryExec: partitions=1, partition_sizes=[1]
---ProjectionExec: expr=[CAST(id@0 AS Int32) as id, name@1 as name]
-----ProjectionExec: expr=[id@0 as id, name@1 as name]
------CoalesceBatchesExec: target_batch_size=2
---------HashJoinExec: mode=Partitioned, join_type=LeftAnti, on=[(CAST(t2.id AS Int32)@2, id@0), (name@1, name@1)]
-----------CoalesceBatchesExec: target_batch_size=2
-------------RepartitionExec: partitioning=Hash([CAST(t2.id AS Int32)@2, name@1], 4), input_partitions=4
---------------ProjectionExec: expr=[id@0 as id, name@1 as name, CAST(id@0 AS Int32) as CAST(t2.id AS Int32)]
-----------------AggregateExec: mode=FinalPartitioned, gby=[id@0 as id, name@1 as name], aggr=[]
-------------------CoalesceBatchesExec: target_batch_size=2
---------------------RepartitionExec: partitioning=Hash([id@0, name@1], 4), input_partitions=4
-----------------------AggregateExec: mode=Partial, gby=[id@0 as id, name@1 as name], aggr=[]
-------------------------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
---------------------------MemoryExec: partitions=1, partition_sizes=[1]
-----------CoalesceBatchesExec: target_batch_size=2
-------------RepartitionExec: partitioning=Hash([id@0, name@1], 4), input_partitions=4
---------------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
-----------------MemoryExec: partitions=1, partition_sizes=[1]
+--------RepartitionExec: partitioning=Hash([CAST(t2.id AS Int32)@2, name@1], 4), input_partitions=4
+----------ProjectionExec: expr=[id@0 as id, name@1 as name, CAST(id@0 AS Int32) as CAST(t2.id AS Int32)]
+------------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
+--------------MemoryExec: partitions=1, partition_sizes=[1]
+--ProjectionExec: expr=[CAST(id@0 AS Int32) as id, name@1 as name]
+----CoalesceBatchesExec: target_batch_size=2
+------HashJoinExec: mode=Partitioned, join_type=LeftAnti, on=[(CAST(t2.id AS Int32)@2, id@0), (name@1, name@1)]
+--------CoalesceBatchesExec: target_batch_size=2
+----------RepartitionExec: partitioning=Hash([CAST(t2.id AS Int32)@2, name@1], 4), input_partitions=4
+------------ProjectionExec: expr=[id@0 as id, name@1 as name, CAST(id@0 AS Int32) as CAST(t2.id AS Int32)]
+--------------AggregateExec: mode=FinalPartitioned, gby=[id@0 as id, name@1 as name], aggr=[]
+----------------CoalesceBatchesExec: target_batch_size=2
+------------------RepartitionExec: partitioning=Hash([id@0, name@1], 4), input_partitions=4
+--------------------AggregateExec: mode=Partial, gby=[id@0 as id, name@1 as name], aggr=[]
+----------------------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
+------------------------MemoryExec: partitions=1, partition_sizes=[1]
+--------CoalesceBatchesExec: target_batch_size=2
+----------RepartitionExec: partitioning=Hash([id@0, name@1], 4), input_partitions=4
+------------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
+--------------MemoryExec: partitions=1, partition_sizes=[1]
+
query IT rowsort
(
@@ -580,7 +579,6 @@ UnionExec
----------AggregateExec: mode=Partial, gby=[n@0 as n], aggr=[COUNT(*)]
------------ProjectionExec: expr=[5 as n]
--------------EmptyExec: produce_one_row=true
---ProjectionExec: expr=[x@0 as count, y@1 as n]
-----ProjectionExec: expr=[1 as x, MAX(Int64(10))@0 as y]
-------AggregateExec: mode=Single, gby=[], aggr=[MAX(Int64(10))]
---------EmptyExec: produce_one_row=true
+--ProjectionExec: expr=[1 as count, MAX(Int64(10))@0 as n]
+----AggregateExec: mode=Single, gby=[], aggr=[MAX(Int64(10))]
+------EmptyExec: produce_one_row=true