Skip to content

Avoid API churn with project_schema #5

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
alamb wants to merge 3 commits into
base: askalt/arc_immutable
Choose a base branch
from
Open

Avoid API churn with project_schema #5

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
2c0746c
Wrap immutable plan parts into Arc
askalt Jan 19, 2026
982668b
Avoid API churn with project_schema
alamb Jan 28, 2026
6a5fa38
Remove unecessary code
alamb Jan 28, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 7 additions & 3 deletions datafusion/common/src/stats.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -391,7 +391,11 @@ impl Statistics {
/// For example, if we had statistics for columns `{"a", "b", "c"}`,
/// projecting to `vec![2, 1]` would return statistics for columns `{"c",
/// "b"}`.
pub fn project(mut self, projection: Option<&Vec<usize>>) -> Self {
pub fn project<P: AsRef<[usize]>>(self, p: Option<&P>) -> Self {
self.project_inner(p.as_ref().map(|p| p.as_ref()))
}

pub fn project_inner(mut self, projection: Option<&[usize]>) -> Self {
let Some(projection) = projection else {
return self;
};
Expand All @@ -410,7 +414,7 @@ impl Statistics {
.map(Slot::Present)
.collect();

for idx in projection {
for idx in projection.iter() {
let next_idx = self.column_statistics.len();
let slot = std::mem::replace(
columns.get_mut(*idx).expect("projection out of bounds"),
Expand Down Expand Up @@ -1066,7 +1070,7 @@ mod tests {

#[test]
fn test_project_none() {
let projection = None;
let projection: Option<Vec<usize>> = None;
let stats = make_stats(vec![10, 20, 30]).project(projection.as_ref());
assert_eq!(stats, make_stats(vec![10, 20, 30]));
}
Expand Down
6 changes: 3 additions & 3 deletions datafusion/common/src/utils/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -68,12 +68,12 @@ use std::thread::available_parallelism;
///
/// assert_eq!(projected_schema, expected_schema);
/// ```
pub fn project_schema(
pub fn project_schema<P: AsRef<[usize]>>(
schema: &SchemaRef,
projection: Option<&Vec<usize>>,
projection: Option<P>,
) -> Result<SchemaRef> {
let schema = match projection {
Some(columns) => Arc::new(schema.project(columns)?),
Some(columns) => Arc::new(schema.project(columns.as_ref())?),
None => Arc::clone(schema),
};
Ok(schema)
Expand Down
2 changes: 1 addition & 1 deletion datafusion/core/src/physical_planner.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -984,7 +984,7 @@ impl DefaultPhysicalPlanner {
// project the output columns excluding the async functions
// The async functions are always appended to the end of the schema.
.apply_projection(Some(
(0..input.schema().fields().len()).collect(),
(0..input.schema().fields().len()).collect::<Vec<_>>(),
))?
.with_batch_size(session_state.config().batch_size())
.build()?
Expand Down
188 changes: 166 additions & 22 deletions datafusion/physical-expr/src/projection.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,7 +29,8 @@ use arrow::datatypes::{Field, Schema, SchemaRef};
use datafusion_common::stats::{ColumnStatistics, Precision};
use datafusion_common::tree_node::{Transformed, TransformedResult, TreeNode};
use datafusion_common::{
Result, ScalarValue, assert_or_internal_err, internal_datafusion_err, plan_err,
Result, ScalarValue, Statistics, assert_or_internal_err, internal_datafusion_err,
plan_err, project_schema,
};

use datafusion_physical_expr_common::metrics::ExecutionPlanMetricsSet;
Expand Down Expand Up @@ -125,7 +126,8 @@ impl From<ProjectionExpr> for (Arc<dyn PhysicalExpr>, String) {
/// indices.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ProjectionExprs {
exprs: Vec<ProjectionExpr>,
/// [`Arc`] used for a cheap clone, which improves physical plan optimization performance.
exprs: Arc<[ProjectionExpr]>,
}

impl std::fmt::Display for ProjectionExprs {
Expand All @@ -137,22 +139,24 @@ impl std::fmt::Display for ProjectionExprs {

impl From<Vec<ProjectionExpr>> for ProjectionExprs {
fn from(value: Vec<ProjectionExpr>) -> Self {
Self { exprs: value }
Self {
exprs: value.into(),
}
}
}

impl From<&[ProjectionExpr]> for ProjectionExprs {
fn from(value: &[ProjectionExpr]) -> Self {
Self {
exprs: value.to_vec(),
exprs: value.iter().cloned().collect(),
}
}
}

impl FromIterator<ProjectionExpr> for ProjectionExprs {
fn from_iter<T: IntoIterator<Item = ProjectionExpr>>(exprs: T) -> Self {
Self {
exprs: exprs.into_iter().collect::<Vec<_>>(),
exprs: exprs.into_iter().collect(),
}
}
}
Expand All @@ -164,12 +168,17 @@ impl AsRef<[ProjectionExpr]> for ProjectionExprs {
}

impl ProjectionExprs {
pub fn new<I>(exprs: I) -> Self
where
I: IntoIterator<Item = ProjectionExpr>,
{
/// Make a new [`ProjectionExprs`] from expressions iterator.
pub fn new(exprs: impl IntoIterator<Item = ProjectionExpr>) -> Self {
Self {
exprs: exprs.into_iter().collect(),
}
}

/// Make a new [`ProjectionExprs`] from expressions.
pub fn from_expressions(exprs: impl Into<Arc<[ProjectionExpr]>>) -> Self {
Self {
exprs: exprs.into_iter().collect::<Vec<_>>(),
exprs: exprs.into(),
}
}

Expand Down Expand Up @@ -285,13 +294,14 @@ impl ProjectionExprs {
{
let exprs = self
.exprs
.into_iter()
.iter()
.cloned()
.map(|mut proj| {
proj.expr = f(proj.expr)?;
Ok(proj)
})
.collect::<Result<Vec<_>>>()?;
Ok(Self::new(exprs))
.collect::<Result<Arc<_>>>()?;
Ok(Self::from_expressions(exprs))
}

/// Apply another projection on top of this projection, returning the combined projection.
Expand Down Expand Up @@ -361,7 +371,7 @@ impl ProjectionExprs {
/// applied on top of this projection.
pub fn try_merge(&self, other: &ProjectionExprs) -> Result<ProjectionExprs> {
let mut new_exprs = Vec::with_capacity(other.exprs.len());
for proj_expr in &other.exprs {
for proj_expr in other.exprs.iter() {
let new_expr = update_expr(&proj_expr.expr, &self.exprs, true)?
.ok_or_else(|| {
internal_datafusion_err!(
Expand Down Expand Up @@ -602,12 +612,12 @@ impl ProjectionExprs {
/// ```
pub fn project_statistics(
&self,
mut stats: datafusion_common::Statistics,
mut stats: Statistics,
output_schema: &Schema,
) -> Result<datafusion_common::Statistics> {
) -> Result<Statistics> {
let mut column_statistics = vec![];

for proj_expr in &self.exprs {
for proj_expr in self.exprs.iter() {
let expr = &proj_expr.expr;
let col_stats = if let Some(col) = expr.as_any().downcast_ref::<Column>() {
std::mem::take(&mut stats.column_statistics[col.index()])
Expand Down Expand Up @@ -754,12 +764,146 @@ impl Projector {
}
}

impl IntoIterator for ProjectionExprs {
type Item = ProjectionExpr;
type IntoIter = std::vec::IntoIter<ProjectionExpr>;
/// Describes an option immutable reference counted shared projection.
///
/// This structure represents projecting a set of columns by index.
/// It uses an [`Arc`] internally to make it cheap to clone.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct OptionProjectionRef {
inner: Option<Arc<[usize]>>,
}

fn into_iter(self) -> Self::IntoIter {
self.exprs.into_iter()
impl OptionProjectionRef {
/// Make a new [`OptionProjectionRef`].
pub fn new(inner: Option<impl Into<Arc<[usize]>>>) -> Self {
Self {
inner: inner.map(Into::into),
}
}

/// Project inner.
pub fn as_inner(&self) -> &Option<Arc<[usize]>> {
&self.inner
}

/// Consume self and return inner.
pub fn into_inner(self) -> Option<Arc<[usize]>> {
self.inner
}

/// Represent this projection as option slice.
pub fn as_ref(&self) -> Option<&[usize]> {
self.inner.as_deref()
}

/// Check if the projection is set.
pub fn is_some(&self) -> bool {
self.inner.is_some()
}

/// Check if the projection is not set.
pub fn is_none(&self) -> bool {
self.inner.is_none()
}

/// Apply passed `projection` to inner one.
///
/// If inner projection is [`None`] then there are no changes.
/// Otherwise, if passed `projection` is not [`None`] then it is remapped
/// according to the stored one. Otherwise, there are no changes.
///
/// # Example
///
/// If stored projection is [0, 2] and we call `apply_projection([0, 2, 3])`,
/// then the resulting projection will be [0, 3].
///
/// # Error
///
/// Returns an internal error if existing projection contains index that is
/// greater than len of the passed `projection`.
///
pub fn apply_projection<'a>(
self,
projection: impl Into<Option<&'a [usize]>>,
) -> Result<Self> {
let projection = projection.into();
let Some(existing_projection) = self.inner else {
return Ok(self);
};
let Some(new_projection) = projection else {
return Ok(Self {
inner: Some(existing_projection),
});
};
Ok(Self::new(Some(
existing_projection
.iter()
.map(|i| {
let idx = *i;
assert_or_internal_err!(
idx < new_projection.len(),
"unable to apply projection: index {} is greater than new projection len {}",
idx,
new_projection.len(),
);
Ok(new_projection[*i])
})
.collect::<Result<Arc<[usize]>>>()?,
)))
}

/// Applies an optional projection to a [`SchemaRef`], returning the
/// projected schema.
pub fn project_schema(&self, schema: &SchemaRef) -> Result<SchemaRef> {
project_schema(schema, self.inner.as_ref())
}

/// Applies an optional projection to a [`Statistics`], returning the
/// projected stats.
pub fn project_statistics(&self, stats: Statistics) -> Statistics {
stats.project(self.inner.as_ref())
}
}

impl<'a> From<&'a OptionProjectionRef> for Option<&'a [usize]> {
fn from(value: &'a OptionProjectionRef) -> Self {
value.inner.as_deref()
}
}

impl From<Vec<usize>> for OptionProjectionRef {
fn from(value: Vec<usize>) -> Self {
Self::new(Some(value))
}
}

impl From<Option<Vec<usize>>> for OptionProjectionRef {
fn from(value: Option<Vec<usize>>) -> Self {
Self::new(value)
}
}

impl FromIterator<usize> for OptionProjectionRef {
fn from_iter<T: IntoIterator<Item = usize>>(iter: T) -> Self {
Self::new(Some(iter.into_iter().collect::<Arc<[usize]>>()))
}
}

impl<T> PartialEq<Option<T>> for OptionProjectionRef
where
T: AsRef<[usize]>,
{
fn eq(&self, other: &Option<T>) -> bool {
self.as_ref() == other.as_ref().map(AsRef::as_ref)
}
}

impl<T> PartialEq<Option<T>> for &OptionProjectionRef
where
T: AsRef<[usize]>,
{
fn eq(&self, other: &Option<T>) -> bool {
self.as_ref() == other.as_ref().map(AsRef::as_ref)
}
}

Expand Down
2 changes: 1 addition & 1 deletion datafusion/physical-optimizer/src/enforce_sorting/sort_pushdown.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -723,7 +723,7 @@ fn handle_hash_join(
.collect();

let column_indices = build_join_column_index(plan);
let projected_indices: Vec<_> = if let Some(projection) = &plan.projection {
let projected_indices: Vec<_> = if let Some(projection) = plan.projection.as_ref() {
projection.iter().map(|&i| &column_indices[i]).collect()
} else {
column_indices.iter().collect()
Expand Down
2 changes: 1 addition & 1 deletion datafusion/physical-optimizer/src/projection_pushdown.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -135,7 +135,7 @@ fn try_push_down_join_filter(
);

let new_lhs_length = lhs_rewrite.data.0.schema().fields.len();
let projections = match projections {
let projections = match projections.as_ref() {
None => match join.join_type() {
JoinType::Inner | JoinType::Left | JoinType::Right | JoinType::Full => {
// Build projections that ignore the newly projected columns.
Expand Down
Loading