You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@datafusion.apache.org by "adriangb (via GitHub)" <gi...@apache.org> on 2026/01/29 13:45:09 UTC
[PR] Consolidate filters and projections onto `TableScan` [datafusion]
adriangb opened a new pull request, #20061:
URL: https://github.com/apache/datafusion/pull/20061
## Related issues
Closes #19894. I think this will also help with #19387 as well.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Consolidate filters and projections onto `TableScan` [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on code in PR #20061:
URL: https://github.com/apache/datafusion/pull/20061#discussion_r2741776013
##########
datafusion/core/src/physical_planner.rs:
##########
@@ -1671,6 +1654,217 @@ impl DefaultPhysicalPlanner {
Ok(exec_node)
}
+ /// Plan a TableScan node, handling filter pushdown classification and
+ /// wrapping with FilterExec/ProjectionExec as needed.
+ async fn plan_table_scan(
+ &self,
+ scan: &TableScan,
+ session_state: &SessionState,
+ ) -> Result<Arc<dyn ExecutionPlan>> {
+ use datafusion_expr::TableProviderFilterPushDown;
+
+ let provider = source_as_provider(&scan.source)?;
+ let source_schema = scan.source.schema();
+
+ // Remove all qualifiers from filters as the provider doesn't know
+ // (nor should care) how the relation was referred to in the query
+ let filters: Vec<Expr> = unnormalize_cols(scan.filters.iter().cloned());
+
+ // Separate volatile filters (they should never be pushed down)
+ let (volatile_filters, non_volatile_filters): (Vec<Expr>, Vec<Expr>) = filters
+ .into_iter()
+ .partition(|pred: &Expr| pred.is_volatile());
+
+ // Classify filters using supports_filters_pushdown
+ let filter_refs: Vec<&Expr> = non_volatile_filters.iter().collect();
+ let supported = provider.supports_filters_pushdown(&filter_refs)?;
+
+ assert_eq_or_internal_err!(
+ non_volatile_filters.len(),
+ supported.len(),
+ "supports_filters_pushdown returned {} results for {} filters",
+ supported.len(),
+ non_volatile_filters.len()
+ );
+
+ // Separate filters into:
+ // - pushable_filters: Exact or Inexact filters to pass to the provider
+ // - post_scan_filters: Inexact, Unsupported, and volatile filters for FilterExec
+ let mut pushable_filters = Vec::new();
+ let mut post_scan_filters = Vec::new();
+
+ for (filter, support) in non_volatile_filters.into_iter().zip(supported.iter()) {
+ match support {
+ TableProviderFilterPushDown::Exact => {
+ pushable_filters.push(filter);
+ }
+ TableProviderFilterPushDown::Inexact => {
+ pushable_filters.push(filter.clone());
+ post_scan_filters.push(filter);
+ }
+ TableProviderFilterPushDown::Unsupported => {
+ post_scan_filters.push(filter);
+ }
+ }
+ }
+
+ // Add volatile filters to post_scan_filters
+ post_scan_filters.extend(volatile_filters);
+
+ // Compute required column indices for the scan
+ // We need columns from both projection expressions and post-scan filters
+ let scan_projection = self.compute_scan_projection(
+ &scan.projection,
+ &post_scan_filters,
+ &source_schema,
+ )?;
+
+ // Check if we have inexact filters - if so, we can't push limit
+ let has_inexact = supported.contains(&TableProviderFilterPushDown::Inexact);
+ let scan_limit = if has_inexact || !post_scan_filters.is_empty() {
+ None // Can't push limit when post-filtering is needed
+ } else {
+ scan.fetch
+ };
+
+ // Create the scan
+ let scan_args = ScanArgs::default()
+ .with_projection(scan_projection.as_deref())
+ .with_filters(if pushable_filters.is_empty() {
+ None
+ } else {
+ Some(&pushable_filters)
+ })
+ .with_limit(scan_limit);
+
+ let scan_result = provider.scan_with_args(session_state, scan_args).await?;
+ let mut plan: Arc<dyn ExecutionPlan> = Arc::clone(scan_result.plan());
+
+ // Create a DFSchema from the scan output for filter and projection creation
+ // The scan output schema is the physical plan's schema
+ let scan_output_schema = plan.schema();
+ let scan_df_schema = DFSchema::try_from(scan_output_schema.as_ref().clone())?;
+
+ // Wrap with FilterExec if needed
+ if !post_scan_filters.is_empty()
+ && let Some(filter_expr) = conjunction(post_scan_filters)
+ {
+ let physical_filter =
+ self.create_physical_expr(&filter_expr, &scan_df_schema, session_state)?;
+ plan = Arc::new(FilterExecBuilder::new(physical_filter, plan).build()?);
+ }
+
+ // Wrap with ProjectionExec if projection is present and differs from scan output
+ // (either non-identity, or fewer columns due to filter-only columns)
+ if let Some(ref proj_exprs) = scan.projection {
+ let needs_projection = !self
+ .is_identity_column_projection(proj_exprs, &source_schema)
+ || scan_output_schema.fields().len() != proj_exprs.len();
+
+ if needs_projection {
+ // Unnormalize projection expressions to match the scan output schema
+ let unnormalized_proj_exprs =
+ unnormalize_cols(proj_exprs.iter().cloned());
+ let physical_exprs: Vec<ProjectionExpr> = unnormalized_proj_exprs
+ .iter()
+ .map(|e| {
+ let physical =
+ self.create_physical_expr(e, &scan_df_schema, session_state)?;
+ // Use the original qualified name for the output alias
+ let name = e.schema_name().to_string();
+ Ok(ProjectionExpr {
+ expr: physical,
+ alias: name,
+ })
+ })
+ .collect::<Result<Vec<_>>>()?;
+
+ plan = Arc::new(ProjectionExec::try_new(physical_exprs, plan)?);
+ }
+ }
+
+ // Apply limit if it wasn't pushed to scan
+ if scan.fetch.is_some() && scan_limit.is_none() {
+ // Limit wasn't pushed, need to apply it here
+ // Note: This is typically handled by other optimizer rules
Review Comment:
Need to double check this. If handled by other optimizer rules why even have this clause here? Maybe we should give the same treatment to limit (handle it here instead of deferring to other optimizer rules)?
##########
datafusion/core/src/physical_planner.rs:
##########
@@ -1968,9 +2162,16 @@ fn extract_dml_filters(input: &Arc<LogicalPlan>) -> Result<Vec<Expr>> {
let mut filters = Vec::new();
input.apply(|node| {
- if let LogicalPlan::Filter(filter) = node {
- // Split AND predicates into individual expressions
- filters.extend(split_conjunction(&filter.predicate).into_iter().cloned());
+ match node {
+ LogicalPlan::Filter(filter) => {
+ // Split AND predicates into individual expressions
+ filters.extend(split_conjunction(&filter.predicate).into_iter().cloned());
+ }
Review Comment:
Maybe we can drop this since filters are effectively pushed into TableScan now?
##########
datafusion/core/src/physical_planner.rs:
##########
@@ -1671,6 +1654,217 @@ impl DefaultPhysicalPlanner {
Ok(exec_node)
}
+ /// Plan a TableScan node, handling filter pushdown classification and
+ /// wrapping with FilterExec/ProjectionExec as needed.
+ async fn plan_table_scan(
+ &self,
+ scan: &TableScan,
+ session_state: &SessionState,
+ ) -> Result<Arc<dyn ExecutionPlan>> {
+ use datafusion_expr::TableProviderFilterPushDown;
+
+ let provider = source_as_provider(&scan.source)?;
+ let source_schema = scan.source.schema();
+
+ // Remove all qualifiers from filters as the provider doesn't know
+ // (nor should care) how the relation was referred to in the query
+ let filters: Vec<Expr> = unnormalize_cols(scan.filters.iter().cloned());
+
+ // Separate volatile filters (they should never be pushed down)
+ let (volatile_filters, non_volatile_filters): (Vec<Expr>, Vec<Expr>) = filters
+ .into_iter()
+ .partition(|pred: &Expr| pred.is_volatile());
+
+ // Classify filters using supports_filters_pushdown
+ let filter_refs: Vec<&Expr> = non_volatile_filters.iter().collect();
+ let supported = provider.supports_filters_pushdown(&filter_refs)?;
+
+ assert_eq_or_internal_err!(
+ non_volatile_filters.len(),
+ supported.len(),
+ "supports_filters_pushdown returned {} results for {} filters",
+ supported.len(),
+ non_volatile_filters.len()
+ );
+
+ // Separate filters into:
+ // - pushable_filters: Exact or Inexact filters to pass to the provider
+ // - post_scan_filters: Inexact, Unsupported, and volatile filters for FilterExec
+ let mut pushable_filters = Vec::new();
+ let mut post_scan_filters = Vec::new();
+
+ for (filter, support) in non_volatile_filters.into_iter().zip(supported.iter()) {
+ match support {
+ TableProviderFilterPushDown::Exact => {
+ pushable_filters.push(filter);
+ }
+ TableProviderFilterPushDown::Inexact => {
+ pushable_filters.push(filter.clone());
+ post_scan_filters.push(filter);
+ }
+ TableProviderFilterPushDown::Unsupported => {
+ post_scan_filters.push(filter);
+ }
+ }
+ }
+
+ // Add volatile filters to post_scan_filters
+ post_scan_filters.extend(volatile_filters);
+
+ // Compute required column indices for the scan
+ // We need columns from both projection expressions and post-scan filters
+ let scan_projection = self.compute_scan_projection(
+ &scan.projection,
+ &post_scan_filters,
+ &source_schema,
+ )?;
+
+ // Check if we have inexact filters - if so, we can't push limit
+ let has_inexact = supported.contains(&TableProviderFilterPushDown::Inexact);
+ let scan_limit = if has_inexact || !post_scan_filters.is_empty() {
+ None // Can't push limit when post-filtering is needed
+ } else {
+ scan.fetch
+ };
+
+ // Create the scan
+ let scan_args = ScanArgs::default()
+ .with_projection(scan_projection.as_deref())
+ .with_filters(if pushable_filters.is_empty() {
+ None
+ } else {
+ Some(&pushable_filters)
+ })
+ .with_limit(scan_limit);
+
+ let scan_result = provider.scan_with_args(session_state, scan_args).await?;
+ let mut plan: Arc<dyn ExecutionPlan> = Arc::clone(scan_result.plan());
+
+ // Create a DFSchema from the scan output for filter and projection creation
+ // The scan output schema is the physical plan's schema
+ let scan_output_schema = plan.schema();
+ let scan_df_schema = DFSchema::try_from(scan_output_schema.as_ref().clone())?;
+
+ // Wrap with FilterExec if needed
+ if !post_scan_filters.is_empty()
+ && let Some(filter_expr) = conjunction(post_scan_filters)
+ {
+ let physical_filter =
+ self.create_physical_expr(&filter_expr, &scan_df_schema, session_state)?;
+ plan = Arc::new(FilterExecBuilder::new(physical_filter, plan).build()?);
+ }
+
+ // Wrap with ProjectionExec if projection is present and differs from scan output
+ // (either non-identity, or fewer columns due to filter-only columns)
Review Comment:
The idea for #19387 is that we might be able to push down trivial expressions here, thus avoiding the need for any physical optimizer changes/rules.
##########
datafusion/expr/src/logical_plan/plan.rs:
##########
Review Comment:
Note: check that we are correctly handling functional deps in the builder. Wonder how to handle that if the projection is expressions? Do we loose functional dependencies? Maybe check how the LogicalPlan::Projection or ProjectionExec handle this sort of thing.
##########
datafusion/core/tests/dataframe/mod.rs:
##########
@@ -4070,11 +4074,8 @@ async fn right_semi_with_alias_filter() -> Result<()> {
actual,
@r"
RightSemi Join: t1.a = t2.a [a:UInt32, b:Utf8, c:Int32]
- Projection: t1.a [a:UInt32]
- Filter: t1.c > Int32(1) [a:UInt32, c:Int32]
- TableScan: t1 projection=[a, c] [a:UInt32, c:Int32]
- Filter: t2.c > Int32(1) [a:UInt32, b:Utf8, c:Int32]
- TableScan: t2 projection=[a, b, c] [a:UInt32, b:Utf8, c:Int32]
+ TableScan: t1 projection=[a], unsupported_filters=[t1.c > Int32(1)] [a:UInt32]
Review Comment:
same note about `filters` vs `unsupported_filters` in the debug output. I think just `filters=[...]` would be fine and I assume it's a simpler implementation
##########
datafusion/sqllogictest/test_files/spark/collection/size.slt:
##########
@@ -129,4 +129,3 @@ SELECT size(column1) FROM VALUES (map(['a'], [1])), (map(['a','b'], [1,2])), (NU
1
2
-1
-
Review Comment:
Need to go through diff and restore originals for slt files with whitespace only changes.
##########
datafusion/expr/src/logical_plan/plan.rs:
##########
@@ -2807,12 +2831,199 @@ impl TableScan {
Ok(Self {
table_name,
source: table_source,
- projection,
+ projection: projection_exprs,
projected_schema,
filters,
fetch,
})
}
+
+ /// Returns the column indices if the projection is a simple column projection.
+ ///
+ /// Returns `None` if the projection contains non-column expressions or
+ /// if columns cannot be resolved to indices in the source schema.
+ pub fn projection_indices(&self) -> Option<Vec<usize>> {
+ let projection = self.projection.as_ref()?;
+ let schema = self.source.schema();
+
+ let indices: Option<Vec<usize>> = projection
+ .iter()
+ .map(|expr| {
+ if let Expr::Column(col) = expr {
+ schema.index_of(col.name()).ok()
+ } else {
+ None
+ }
+ })
+ .collect();
+
+ indices
+ }
+
+ /// Returns true if the projection is an identity projection (all columns in order)
+ /// or if there is no projection.
+ pub fn is_identity_projection(&self) -> bool {
+ match &self.projection {
+ None => true,
+ Some(exprs) => {
+ let schema = self.source.schema();
+ if exprs.len() != schema.fields().len() {
+ return false;
+ }
+ exprs.iter().enumerate().all(|(i, expr)| {
+ if let Expr::Column(col) = expr {
+ schema.index_of(col.name()).ok() == Some(i)
+ } else {
+ false
+ }
+ })
+ }
+ }
+ }
+}
+
+/// Builder for constructing [`TableScan`] with expression-based projections.
+///
+/// This builder provides more flexibility than [`TableScan::try_new`] by
+/// allowing arbitrary expressions in the projection, not just column references.
+///
+/// # Example
+/// ```ignore
+/// let scan = TableScanBuilder::new("my_table", table_source)
+/// .with_projection(vec![col("a"), col("b") + lit(1)])
+/// .with_filters(vec![col("c").gt(lit(10))])
+/// .build()?;
+/// ```
+#[derive(Clone)]
+pub struct TableScanBuilder {
+ table_name: TableReference,
+ table_source: Arc<dyn TableSource>,
+ projection: Option<Vec<Expr>>,
+ filters: Vec<Expr>,
+ fetch: Option<usize>,
+}
+
+impl TableScanBuilder {
+ /// Create a new TableScanBuilder with the given table name and source.
+ pub fn new(
+ table_name: impl Into<TableReference>,
+ table_source: Arc<dyn TableSource>,
+ ) -> Self {
+ Self {
+ table_name: table_name.into(),
+ table_source,
+ projection: None,
+ filters: vec![],
+ fetch: None,
+ }
+ }
+
+ /// Set the projection expressions.
+ ///
+ /// If `None`, all columns from the source are selected.
+ pub fn with_projection(mut self, projection: Option<Vec<Expr>>) -> Self {
+ self.projection = projection;
+ self
+ }
+
+ /// Set the filter expressions.
+ ///
+ /// These filters will be stored on the TableScan and the physical planner
+ /// will decide which filters to push down to the source based on
+ /// `supports_filters_pushdown`.
+ pub fn with_filters(mut self, filters: Vec<Expr>) -> Self {
+ self.filters = filters;
+ self
+ }
+
+ /// Set the fetch limit.
+ pub fn with_fetch(mut self, fetch: Option<usize>) -> Self {
+ self.fetch = fetch;
+ self
+ }
+
+ /// Build the TableScan.
+ ///
+ /// Computes the output schema based on the projection expressions.
+ pub fn build(self) -> Result<TableScan> {
+ if self.table_name.table().is_empty() {
+ return plan_err!("table_name cannot be empty");
+ }
+
+ let schema = self.table_source.schema();
+ let func_dependencies = FunctionalDependencies::new_from_constraints(
+ self.table_source.constraints(),
+ schema.fields.len(),
+ );
+
+ // Build the projected schema from projection expressions
+ let projected_schema = match &self.projection {
+ Some(exprs) => {
+ // Create a qualified schema for expression evaluation
+ let qualified_schema = DFSchema::try_from_qualified_schema(
+ self.table_name.clone(),
+ &schema,
+ )?;
+
+ // Derive output fields from projection expressions
+ let fields: Vec<(Option<TableReference>, FieldRef)> = exprs
+ .iter()
+ .map(|expr| {
+ let (qualifier, field) = expr.to_field(&qualified_schema)?;
+ Ok((qualifier.clone(), field))
+ })
+ .collect::<Result<Vec<_>>>()?;
+
+ // Try to compute functional dependencies for simple column projections
+ let projected_func_dependencies = if let Some(indices) =
+ self.projection_indices_internal(exprs, &schema)
+ {
+ func_dependencies
+ .project_functional_dependencies(&indices, indices.len())
+ } else {
+ FunctionalDependencies::empty()
Review Comment:
Confirming from above: if the projection is anything but simple column projections we loose functional dependencies. Is this okay? Is this what would have happened anyway if we had `Projection -> TableScan`? Does this have a negative impact on e.g. filter tricks we can do if we know dependencies between columns?
##########
datafusion/core/src/physical_planner.rs:
##########
@@ -4284,6 +4493,10 @@ digraph {
}
#[tokio::test]
+ // With the new physical planner that adds ProjectionExec to handle schema
+ // mismatches, this now proceeds past the schema check and hits the
+ // unimplemented NoOpExecutionPlan.
+ #[should_panic(expected = "NoOpExecutionPlan")]
Review Comment:
Should consider rewriting / improving this test
##########
datafusion/core/tests/sql/explain_analyze.rs:
##########
@@ -438,23 +434,15 @@ async fn csv_explain_plans() {
{
graph[label="LogicalPlan"]
2[shape=box label="Explain"]
- 3[shape=box label="Projection: aggregate_test_100.c1"]
+ 3[shape=box label="TableScan: aggregate_test_100 projection=[c1], partial_filters=[aggregate_test_100.c2 > Int8(10)]"]
Review Comment:
same note about `partial_filters` vs. just `filters`
##########
datafusion/expr/src/logical_plan/builder.rs:
##########
@@ -520,15 +520,20 @@ impl LogicalPlanBuilder {
{
let sub_plan = p.into_owned();
- if let Some(proj) = table_scan.projection {
- let projection_exprs = proj
+ if let Some(proj_exprs) = table_scan.projection {
+ // Strip qualifiers from projection expressions since the sub_plan
+ // may have unqualified column names
+ let unqualified_exprs: Vec<Expr> = proj_exprs
Review Comment:
This matches what we do with filters. When projections get pushed down into a scan there is no point in having qualifiers on them.
##########
datafusion/core/tests/dataframe/mod.rs:
##########
@@ -3137,18 +3141,18 @@ async fn test_count_wildcard_on_where_exist() -> Result<()> {
assert_snapshot!(
pretty_format_batches(&sql_results).unwrap(),
@r"
- +---------------+-----------------------------------------------------+
- | plan_type | plan |
- +---------------+-----------------------------------------------------+
- | logical_plan | LeftSemi Join: |
- | | TableScan: t1 projection=[a, b] |
- | | SubqueryAlias: __correlated_sq_1 |
- | | EmptyRelation: rows=1 |
- | physical_plan | NestedLoopJoinExec: join_type=RightSemi |
- | | PlaceholderRowExec |
- | | DataSourceExec: partitions=1, partition_sizes=[1] |
- | | |
- +---------------+-----------------------------------------------------+
+ +---------------+------------------------------------------------------------------------+
+ | plan_type | plan |
+ +---------------+------------------------------------------------------------------------+
+ | logical_plan | LeftSemi Join: |
+ | | TableScan: t1 projection=[a, b], unsupported_filters=[Boolean(true)] |
Review Comment:
Hmm a bit weird of debug output:
1. Do we want `unsupported_filters` or just `filters`? Does the formatter call `TableProvider::supports_filter_pushdown` just to include debug output?
2. Why is there a `true` filter? It will be optimized away at some point (probably in logical -> physical conversion). At the very least we should check if the test is adding a pointless filter (and consider updating the test unless it is actually part of what's being tested) or otherwise why we are ending up with `where true`.
##########
datafusion/expr/src/logical_plan/plan.rs:
##########
@@ -2807,12 +2831,199 @@ impl TableScan {
Ok(Self {
table_name,
source: table_source,
- projection,
+ projection: projection_exprs,
projected_schema,
filters,
fetch,
})
}
+
+ /// Returns the column indices if the projection is a simple column projection.
+ ///
+ /// Returns `None` if the projection contains non-column expressions or
+ /// if columns cannot be resolved to indices in the source schema.
+ pub fn projection_indices(&self) -> Option<Vec<usize>> {
+ let projection = self.projection.as_ref()?;
+ let schema = self.source.schema();
+
+ let indices: Option<Vec<usize>> = projection
+ .iter()
+ .map(|expr| {
+ if let Expr::Column(col) = expr {
+ schema.index_of(col.name()).ok()
+ } else {
+ None
+ }
+ })
+ .collect();
+
+ indices
+ }
+
+ /// Returns true if the projection is an identity projection (all columns in order)
+ /// or if there is no projection.
+ pub fn is_identity_projection(&self) -> bool {
+ match &self.projection {
+ None => true,
+ Some(exprs) => {
+ let schema = self.source.schema();
+ if exprs.len() != schema.fields().len() {
+ return false;
+ }
+ exprs.iter().enumerate().all(|(i, expr)| {
+ if let Expr::Column(col) = expr {
+ schema.index_of(col.name()).ok() == Some(i)
+ } else {
+ false
+ }
+ })
+ }
+ }
+ }
+}
+
+/// Builder for constructing [`TableScan`] with expression-based projections.
+///
+/// This builder provides more flexibility than [`TableScan::try_new`] by
+/// allowing arbitrary expressions in the projection, not just column references.
+///
+/// # Example
+/// ```ignore
+/// let scan = TableScanBuilder::new("my_table", table_source)
+/// .with_projection(vec![col("a"), col("b") + lit(1)])
+/// .with_filters(vec![col("c").gt(lit(10))])
+/// .build()?;
+/// ```
+#[derive(Clone)]
+pub struct TableScanBuilder {
+ table_name: TableReference,
+ table_source: Arc<dyn TableSource>,
+ projection: Option<Vec<Expr>>,
+ filters: Vec<Expr>,
+ fetch: Option<usize>,
+}
+
+impl TableScanBuilder {
+ /// Create a new TableScanBuilder with the given table name and source.
+ pub fn new(
+ table_name: impl Into<TableReference>,
+ table_source: Arc<dyn TableSource>,
+ ) -> Self {
+ Self {
+ table_name: table_name.into(),
+ table_source,
+ projection: None,
+ filters: vec![],
+ fetch: None,
+ }
+ }
+
+ /// Set the projection expressions.
+ ///
+ /// If `None`, all columns from the source are selected.
+ pub fn with_projection(mut self, projection: Option<Vec<Expr>>) -> Self {
+ self.projection = projection;
+ self
+ }
+
+ /// Set the filter expressions.
+ ///
+ /// These filters will be stored on the TableScan and the physical planner
+ /// will decide which filters to push down to the source based on
+ /// `supports_filters_pushdown`.
+ pub fn with_filters(mut self, filters: Vec<Expr>) -> Self {
+ self.filters = filters;
+ self
+ }
+
+ /// Set the fetch limit.
+ pub fn with_fetch(mut self, fetch: Option<usize>) -> Self {
+ self.fetch = fetch;
+ self
+ }
+
+ /// Build the TableScan.
+ ///
+ /// Computes the output schema based on the projection expressions.
+ pub fn build(self) -> Result<TableScan> {
+ if self.table_name.table().is_empty() {
+ return plan_err!("table_name cannot be empty");
+ }
+
+ let schema = self.table_source.schema();
+ let func_dependencies = FunctionalDependencies::new_from_constraints(
+ self.table_source.constraints(),
+ schema.fields.len(),
+ );
+
+ // Build the projected schema from projection expressions
+ let projected_schema = match &self.projection {
+ Some(exprs) => {
+ // Create a qualified schema for expression evaluation
+ let qualified_schema = DFSchema::try_from_qualified_schema(
+ self.table_name.clone(),
+ &schema,
+ )?;
+
+ // Derive output fields from projection expressions
+ let fields: Vec<(Option<TableReference>, FieldRef)> = exprs
+ .iter()
+ .map(|expr| {
+ let (qualifier, field) = expr.to_field(&qualified_schema)?;
+ Ok((qualifier.clone(), field))
+ })
+ .collect::<Result<Vec<_>>>()?;
+
+ // Try to compute functional dependencies for simple column projections
+ let projected_func_dependencies = if let Some(indices) =
+ self.projection_indices_internal(exprs, &schema)
+ {
+ func_dependencies
+ .project_functional_dependencies(&indices, indices.len())
+ } else {
+ FunctionalDependencies::empty()
+ };
+
+ let df_schema =
+ DFSchema::new_with_metadata(fields, schema.metadata.clone())?;
+ df_schema.with_functional_dependencies(projected_func_dependencies)?
+ }
+ None => {
+ let df_schema = DFSchema::try_from_qualified_schema(
+ self.table_name.clone(),
+ &schema,
+ )?;
+ df_schema.with_functional_dependencies(func_dependencies)?
+ }
+ };
+
+ Ok(TableScan {
+ table_name: self.table_name,
+ source: self.table_source,
+ projection: self.projection,
+ projected_schema: Arc::new(projected_schema),
+ filters: self.filters,
+ fetch: self.fetch,
+ })
+ }
+
+ /// Helper to get column indices from projection expressions
+ fn projection_indices_internal(
Review Comment:
This duplicates `TableScan::projection_indices`
Maybe we should just make this helper public in `utils.rs` and remove the method from `TableScan`?
##########
datafusion/expr/src/logical_plan/plan.rs:
##########
@@ -2807,12 +2831,199 @@ impl TableScan {
Ok(Self {
table_name,
source: table_source,
- projection,
+ projection: projection_exprs,
projected_schema,
filters,
fetch,
})
}
+
+ /// Returns the column indices if the projection is a simple column projection.
+ ///
+ /// Returns `None` if the projection contains non-column expressions or
+ /// if columns cannot be resolved to indices in the source schema.
+ pub fn projection_indices(&self) -> Option<Vec<usize>> {
+ let projection = self.projection.as_ref()?;
+ let schema = self.source.schema();
+
+ let indices: Option<Vec<usize>> = projection
+ .iter()
+ .map(|expr| {
+ if let Expr::Column(col) = expr {
+ schema.index_of(col.name()).ok()
+ } else {
+ None
+ }
+ })
+ .collect();
+
+ indices
+ }
+
+ /// Returns true if the projection is an identity projection (all columns in order)
+ /// or if there is no projection.
+ pub fn is_identity_projection(&self) -> bool {
Review Comment:
Unused function. Should remove.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Consolidate filters and projections onto `TableScan` [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on code in PR #20061:
URL: https://github.com/apache/datafusion/pull/20061#discussion_r2748074813
##########
datafusion/core/src/physical_planner.rs:
##########
@@ -1968,9 +2115,16 @@ fn extract_dml_filters(input: &Arc<LogicalPlan>) -> Result<Vec<Expr>> {
let mut filters = Vec::new();
input.apply(|node| {
- if let LogicalPlan::Filter(filter) = node {
- // Split AND predicates into individual expressions
- filters.extend(split_conjunction(&filter.predicate).into_iter().cloned());
+ match node {
+ LogicalPlan::Filter(filter) => {
+ // Split AND predicates into individual expressions
+ filters.extend(split_conjunction(&filter.predicate).into_iter().cloned());
+ }
+ LogicalPlan::TableScan(scan) => {
+ // Also extract filters from TableScan (where they may be pushed down)
+ filters.extend(scan.filters.iter().cloned());
Review Comment:
> I don't know how the DML stuff is supposed to handle more complex cases involving `EXISTS`, etc.
E.g.:
```sql
-- 1. Setup: Create tables
CREATE TEMP TABLE departments (
dept_id INT,
dept_name TEXT,
is_active BOOLEAN
);
CREATE TEMP TABLE employees (
emp_id SERIAL PRIMARY KEY,
name TEXT,
dept_name TEXT,
salary INT
);
-- 2. Seed Data
INSERT INTO departments VALUES (1, 'Engineering', true), (2, 'Marketing', false);
INSERT INTO employees (name, dept_name, salary) VALUES
('Alice', 'Engineering', 5000),
('Bob', 'Engineering', 5000),
('Charlie', 'Marketing', 4000);
-- 3. Execute the UPDATE with a subquery in the WHERE clause
-- Goal: Give a 10% raise only to people in the 'Engineering' department
UPDATE employees
SET salary = salary * 1.10
WHERE dept_name = (
SELECT dept_name
FROM departments
WHERE dept_id = 1
LIMIT 1
);
-- 4. Verify the results
SELECT * FROM employees;
```
This seems to work in postgres.
But again I'm not sure if this is directly something this PR should handle.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Consolidate filters and projections onto `TableScan` [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on code in PR #20061:
URL: https://github.com/apache/datafusion/pull/20061#discussion_r2752049723
##########
datafusion/proto/src/logical_plan/mod.rs:
##########
@@ -1023,11 +1023,10 @@ impl AsLogicalPlan for LogicalPlanNode {
let projection = match projection {
Review Comment:
I think what we need here is to extend the proto to `Vec<LogicalExprNode>`...
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Consolidate filters and projections onto `TableScan` [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on PR #20061:
URL: https://github.com/apache/datafusion/pull/20061#issuecomment-3825929064
@ethan-tyler any chance you can take a look at this as well?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Consolidate filters and projections onto `TableScan` [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on PR #20061:
URL: https://github.com/apache/datafusion/pull/20061#issuecomment-3820417065
@kosiew would you be open to reviewing this?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Consolidate filters and projections onto `TableScan` [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on PR #20061:
URL: https://github.com/apache/datafusion/pull/20061#issuecomment-3900008176
My main thought was that:
1. This would allow more closely mirroring the physical API
2. This would allow an optimizer to distinguish pushing down from pushing down into a scan for cases where the latter is the real goal.
But since I think I've solved the problem without this I will close this for now, we can laways come back to this if needed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Consolidate filters and projections onto `TableScan` [datafusion]
Posted by "kosiew (via GitHub)" <gi...@apache.org>.
kosiew commented on code in PR #20061:
URL: https://github.com/apache/datafusion/pull/20061#discussion_r2745663790
##########
datafusion/core/src/physical_planner.rs:
##########
@@ -1968,9 +2115,16 @@ fn extract_dml_filters(input: &Arc<LogicalPlan>) -> Result<Vec<Expr>> {
let mut filters = Vec::new();
input.apply(|node| {
- if let LogicalPlan::Filter(filter) = node {
- // Split AND predicates into individual expressions
- filters.extend(split_conjunction(&filter.predicate).into_iter().cloned());
+ match node {
+ LogicalPlan::Filter(filter) => {
+ // Split AND predicates into individual expressions
+ filters.extend(split_conjunction(&filter.predicate).into_iter().cloned());
+ }
+ LogicalPlan::TableScan(scan) => {
+ // Also extract filters from TableScan (where they may be pushed down)
+ filters.extend(scan.filters.iter().cloned());
Review Comment:
I think this would be a problem in scenarios like `UPDATE target FROM source ...` where the input plan contains `TableScan` nodes for both `target` and `source`. This function will extract filters from *both* tables and attempt to apply them to the `target` table (after stripping qualifiers).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Consolidate filters and projections onto `TableScan` [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on PR #20061:
URL: https://github.com/apache/datafusion/pull/20061#issuecomment-3825902207
> > Found an issue and stopping as there are other CI issues.
>
> Thank you, I will look into this.
>
> I'm struggling with the CI issue: https://github.com/apache/datafusion/actions/runs/21495116091/job/61927800829?pr=20061#step:4:7993
>
> It seems like the only diff is the inclusion of a backtrace. I can reproduce locally if I run with `RUST_BACKTRACE=1` but without it no failure. Which is confusing...
Fixed now. The issue was that we can't push `Expr::ScalarSubquery` into `TableScan`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Consolidate filters and projections onto `TableScan` [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on code in PR #20061:
URL: https://github.com/apache/datafusion/pull/20061#discussion_r2747636764
##########
datafusion/core/src/physical_planner.rs:
##########
@@ -1968,9 +2115,16 @@ fn extract_dml_filters(input: &Arc<LogicalPlan>) -> Result<Vec<Expr>> {
let mut filters = Vec::new();
input.apply(|node| {
- if let LogicalPlan::Filter(filter) = node {
- // Split AND predicates into individual expressions
- filters.extend(split_conjunction(&filter.predicate).into_iter().cloned());
+ match node {
+ LogicalPlan::Filter(filter) => {
+ // Split AND predicates into individual expressions
+ filters.extend(split_conjunction(&filter.predicate).into_iter().cloned());
+ }
+ LogicalPlan::TableScan(scan) => {
+ // Also extract filters from TableScan (where they may be pushed down)
+ filters.extend(scan.filters.iter().cloned());
Review Comment:
Hmm interesting. Is the problem the stripping of qualifiers? It seems to me that the current implementation is generally fragile: we should be taking into account the join aspect and only collect from subtrees of the join. In particular:
```
> explain format indent UPDATE "trg" SET col = 1 FROM src WHERE trg.id = src.id AND src.type = 'active' AND trg.id > 100;
+---------------+-----------------------------------------------------------------------+
| plan_type | plan |
+---------------+-----------------------------------------------------------------------+
| logical_plan | Dml: op=[Update] table=[trg] |
| | Projection: trg.id AS id, Utf8View("1") AS col |
| | Inner Join: trg.id = src.id |
| | Filter: trg.id > Int32(100) |
| | TableScan: trg projection=[id] |
| | Projection: src.id |
| | Filter: src.type = Utf8View("active") AND src.id > Int32(100) |
| | TableScan: src projection=[id, type] |
```
It seems to me that we should do something like:
1. Find the table scan corresponding to the table being updated.
2. Walk up the tree until we hit a join / subquery (?) / other blocker.
3. Collect all filters in that subtree (ideally once this PR is across the line they've all been pushed into the `TableScan` so that becomes trivial)
I don't know how the DML stuff is supposed to handle more complex cases involving `EXISTS`, etc.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Consolidate filters and projections onto `TableScan` [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on PR #20061:
URL: https://github.com/apache/datafusion/pull/20061#issuecomment-3900009677
I do still think there's something to be said for the filters only refactor here... I just don't have time or motivation to see it across the line.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Consolidate filters and projections onto `TableScan` [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on PR #20061:
URL: https://github.com/apache/datafusion/pull/20061#issuecomment-3823265637
> Found an issue and stopping as there are other CI issues.
Thank you, I will look into this.
I'm struggling with the CI issue: https://github.com/apache/datafusion/actions/runs/21495116091/job/61927800829?pr=20061#step:4:7993
It seems like the only diff is the inclusion of a backtrace. I can reproduce locally if I run with `RUST_BACKTRACE=1` but without it no failure. Which is confusing...
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Consolidate filters and projections onto `TableScan` [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb closed pull request #20061: Consolidate filters and projections onto `TableScan`
URL: https://github.com/apache/datafusion/pull/20061
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Consolidate filters and projections onto `TableScan` [datafusion]
Posted by "ethan-tyler (via GitHub)" <gi...@apache.org>.
ethan-tyler commented on code in PR #20061:
URL: https://github.com/apache/datafusion/pull/20061#discussion_r2751941932
##########
datafusion/optimizer/src/push_down_filter.rs:
##########
@@ -1126,62 +1124,41 @@ impl OptimizerRule for PushDownFilter {
}
LogicalPlan::Join(join) => push_down_join(join, Some(&filter.predicate)),
LogicalPlan::TableScan(scan) => {
+ // Move filters to the TableScan.
+ // Exclude scalar subqueries from being pushed down since those are essentially a
+ // fork in the logical plan tree that should not be processed by TableScan nodes.
let filter_predicates = split_conjunction(&filter.predicate);
- let (volatile_filters, non_volatile_filters): (Vec<&Expr>, Vec<&Expr>) =
- filter_predicates
- .into_iter()
- .partition(|pred| pred.is_volatile());
-
- // Check which non-volatile filters are supported by source
- let supported_filters = scan
- .source
- .supports_filters_pushdown(non_volatile_filters.as_slice())?;
- assert_eq_or_internal_err!(
- non_volatile_filters.len(),
- supported_filters.len(),
- "Vec returned length: {} from supports_filters_pushdown is not the same size as the filters passed, which length is: {}",
- supported_filters.len(),
- non_volatile_filters.len()
- );
-
- // Compose scan filters from non-volatile filters of `Exact` or `Inexact` pushdown type
- let zip = non_volatile_filters.into_iter().zip(supported_filters);
+ // Partition predicates: scalar subqueries must stay in a Filter node
+ let (scalar_subquery_filters, pushable_filters): (Vec<_>, Vec<_>) =
+ filter_predicates.iter().partition(|pred| {
+ pred.exists(|e| Ok(matches!(e, Expr::ScalarSubquery(_))))
+ .unwrap()
+ });
- let new_scan_filters = zip
- .clone()
- .filter(|(_, res)| res != &TableProviderFilterPushDown::Unsupported)
- .map(|(pred, _)| pred);
-
- // Add new scan filters
+ // Combine existing scan filters with pushable filter predicates
let new_scan_filters: Vec<Expr> = scan
.filters
.iter()
- .chain(new_scan_filters)
+ .chain(pushable_filters)
.unique()
Review Comment:
This refactor changes what reaches `.unique()`. Volatile predicates can now land in `TableScan.filters`, and dedup collapses repeated evaluation. `random() < 0.5 AND random() < 0.5` is not equivalent to `random() < 0.5`. I would either drop the dedup, or restrict it to `!expr.is_volatile()` and keep volatiles in order.
##########
datafusion/core/src/physical_planner.rs:
##########
Review Comment:
In `UPDATE t1 SET ... FROM t2 WHERE t2.x = 1 AND t1.id = t2.id`, `t2.x` becomes `x` and can accidentally match `t1.x`; join predicates degrade into `id = id`. I would only pass predicates that are provably evaluable against the target table schema alone before stripping qualifiers and keep the rest outside the provider call.
##########
datafusion/core/src/physical_planner.rs:
##########
Review Comment:
Same qualifier-stripping problem is happening here. `extract_update_assignments` strips qualifiers from the RHS, `SET b = t2.b` can become `b = b` and get treated as an identity assignment.
##########
datafusion/substrait/src/logical_plan/producer/rel/read_rel.rs:
##########
@@ -97,47 +101,151 @@ pub fn from_table_scan(
producer: &mut impl SubstraitProducer,
scan: &TableScan,
) -> datafusion::common::Result<Box<Rel>> {
- let projection = scan.projection.as_ref().map(|p| {
- p.iter()
- .map(|i| StructItem {
- field: *i as i32,
- child: None,
- })
- .collect()
- });
+ let table_schema = scan.source.schema();
+ let table_schema_qualified = Arc::new(DFSchema::try_from_qualified_schema(
+ scan.table_name.clone(),
+ &table_schema,
+ )?);
- let projection = projection.map(|struct_items| MaskExpression {
- select: Some(StructSelect { struct_items }),
- maintain_singular_struct: false,
- });
-
- let table_schema = scan.source.schema().to_dfschema_ref()?;
- let base_schema = to_substrait_named_struct(producer, &table_schema)?;
+ // Check if projection contains only simple column references
+ let simple_projection_indices = scan
+ .projection
+ .as_ref()
+ .and_then(|exprs| projection_indices_from_exprs(exprs, &scan.source.schema()));
+ // Build the filter expression if any
let filter_option = if scan.filters.is_empty() {
None
} else {
- let table_schema_qualified = Arc::new(
- DFSchema::try_from_qualified_schema(
- scan.table_name.clone(),
- &(scan.source.schema()),
- )
- .unwrap(),
- );
-
let combined_expr = conjunction(scan.filters.clone()).unwrap();
let filter_expr =
producer.handle_expr(&combined_expr, &table_schema_qualified)?;
Some(Box::new(filter_expr))
};
+ let base_schema = to_substrait_named_struct(
+ producer,
+ &Arc::clone(&table_schema).to_dfschema_ref()?,
+ )?;
+
+ // If projection is simple column references, use a mask on the ReadRel
+ if let Some(indices) = simple_projection_indices {
+ let projection = Some(MaskExpression {
+ select: Some(StructSelect {
+ struct_items: indices
+ .iter()
+ .map(|&i| StructItem {
+ field: i as i32,
+ child: None,
+ })
+ .collect(),
+ }),
+ maintain_singular_struct: false,
+ });
+
+ return Ok(Box::new(Rel {
+ rel_type: Some(RelType::Read(Box::new(ReadRel {
+ common: None,
+ base_schema: Some(base_schema),
+ filter: filter_option,
+ best_effort_filter: None,
+ projection,
+ advanced_extension: None,
+ read_type: Some(ReadType::NamedTable(NamedTable {
+ names: scan.table_name.to_vec(),
+ advanced_extension: None,
+ })),
+ }))),
+ }));
+ }
+
+ // Complex projection expressions - need to wrap ReadRel in ProjectRel
+ if let Some(proj_exprs) = &scan.projection {
+ // Find all columns needed by the projection expressions
+ let mut required_columns = HashSet::new();
+ for expr in proj_exprs {
+ expr_to_columns(expr, &mut required_columns)?;
+ }
+
+ // Convert column names to indices
+ let mut column_indices: Vec<usize> = required_columns
+ .iter()
+ .filter_map(|col| table_schema.index_of(col.name()).ok())
+ .collect();
+ column_indices.sort();
+ column_indices.dedup();
Review Comment:
Field references are built against the full table DFSchema but the ReadRel uses a column subset mask. If expression projections stay, the indices won't line up. Goes away if projections are restricted to column refs.
##########
datafusion/sql/src/unparser/utils.rs:
##########
@@ -385,17 +385,24 @@ pub(crate) fn try_transform_to_simple_table_scan_with_filters(
}
}
- let mut builder = LogicalPlanBuilder::scan(
+ // Use TableScanBuilder to preserve full projection expressions
+ let mut scan_builder = TableScanBuilder::new(
Review Comment:
This builder chain doesn't carry over `table_scan.fetch`, it needs a `.with_fetch(table_scan.fetch)` or plans with fetch pushed into the scan will lose LIMIT in the generated SQL.
##########
datafusion/proto/src/logical_plan/mod.rs:
##########
@@ -1023,11 +1023,10 @@ impl AsLogicalPlan for LogicalPlanNode {
let projection = match projection {
Review Comment:
Projections are serialized as column name strings and deserialized by schema index lookup. Non-column expressions won't survive the roundtrip, and even column names with dots or quoting differences are fragile. I think the cleanest path is enforcing column-ref-only projections in the builder and failing fast on anything else. Extending proto to `Vec<LogicalExprNode>` is the alternative but a bigger lift.
##########
datafusion/core/src/physical_planner.rs:
##########
@@ -1671,6 +1610,214 @@ impl DefaultPhysicalPlanner {
Ok(exec_node)
}
+ /// Plan a TableScan node, handling filter pushdown classification and
+ /// wrapping with FilterExec/ProjectionExec as needed.
+ async fn plan_table_scan(
+ &self,
+ scan: &TableScan,
+ session_state: &SessionState,
+ ) -> Result<Arc<dyn ExecutionPlan>> {
+ use datafusion_expr::TableProviderFilterPushDown;
+
+ let provider = source_as_provider(&scan.source)?;
+ let source_schema = scan.source.schema();
+
+ // Remove all qualifiers from filters as the provider doesn't know
+ // (nor should care) how the relation was referred to in the query
+ let filters: Vec<Expr> = unnormalize_cols(scan.filters.iter().cloned());
+
+ // Separate volatile filters (they should never be pushed down)
+ let (volatile_filters, non_volatile_filters): (Vec<Expr>, Vec<Expr>) = filters
+ .into_iter()
+ .partition(|pred: &Expr| pred.is_volatile());
+
+ // Classify filters using supports_filters_pushdown
+ let filter_refs: Vec<&Expr> = non_volatile_filters.iter().collect();
+ let supported = provider.supports_filters_pushdown(&filter_refs)?;
+
+ assert_eq_or_internal_err!(
+ non_volatile_filters.len(),
+ supported.len(),
+ "supports_filters_pushdown returned {} results for {} filters",
+ supported.len(),
+ non_volatile_filters.len()
+ );
+
+ // Separate filters into:
+ // - pushable_filters: Exact or Inexact filters to pass to the provider
+ // - post_scan_filters: Inexact, Unsupported, and volatile filters for FilterExec
+ let mut pushable_filters = Vec::new();
+ let mut post_scan_filters = Vec::new();
+
+ for (filter, support) in non_volatile_filters.into_iter().zip(supported.iter()) {
+ match support {
+ TableProviderFilterPushDown::Exact => {
+ pushable_filters.push(filter);
+ }
+ TableProviderFilterPushDown::Inexact => {
+ pushable_filters.push(filter.clone());
+ post_scan_filters.push(filter);
+ }
+ TableProviderFilterPushDown::Unsupported => {
+ post_scan_filters.push(filter);
+ }
+ }
+ }
+
+ // Add volatile filters to post_scan_filters
+ post_scan_filters.extend(volatile_filters);
+
+ // Compute required column indices for the scan
+ // We need columns from both projection expressions and post-scan filters
+ let scan_projection = self.compute_scan_projection(
+ &scan.projection,
+ &post_scan_filters,
+ &source_schema,
+ )?;
+
+ // Check if we have inexact filters - if so, we can't push limit
+ let has_inexact = supported.contains(&TableProviderFilterPushDown::Inexact);
+ let scan_limit = if has_inexact || !post_scan_filters.is_empty() {
+ None // Can't push limit when post-filtering is needed
+ } else {
+ scan.fetch
+ };
+
+ // Create the scan
+ let scan_args = ScanArgs::default()
+ .with_projection(scan_projection.as_deref())
+ .with_filters(if pushable_filters.is_empty() {
+ None
+ } else {
+ Some(&pushable_filters)
+ })
+ .with_limit(scan_limit);
+
+ let scan_result = provider.scan_with_args(session_state, scan_args).await?;
+ let mut plan: Arc<dyn ExecutionPlan> = Arc::clone(scan_result.plan());
+
+ // Create a DFSchema from the scan output for filter and projection creation
+ // The scan output schema is the physical plan's schema
+ let scan_output_schema = plan.schema();
+ let scan_df_schema = DFSchema::try_from(scan_output_schema.as_ref().clone())?;
+
+ // Wrap with FilterExec if needed
+ if !post_scan_filters.is_empty()
+ && let Some(filter_expr) = conjunction(post_scan_filters)
+ {
+ let num_scan_columns = scan_output_schema.fields().len();
+ plan = self.create_filter_exec(
+ &filter_expr,
+ plan,
+ &scan_df_schema,
+ session_state,
+ num_scan_columns,
+ )?;
+ }
+
+ // Wrap with ProjectionExec if projection is present and differs from scan output
+ // (either non-identity, or fewer columns due to filter-only columns)
+ if let Some(ref proj_exprs) = scan.projection {
+ let needs_projection = !self
+ .is_identity_column_projection(proj_exprs, &source_schema)
+ || scan_output_schema.fields().len() != proj_exprs.len();
+
+ if needs_projection {
+ // Unnormalize projection expressions to match the scan output schema
+ let unnormalized_proj_exprs =
+ unnormalize_cols(proj_exprs.iter().cloned());
+ plan = self.create_projection_exec(
+ &unnormalized_proj_exprs,
+ plan,
+ &scan_df_schema,
+ session_state,
+ )?;
+ }
+ }
+
+ // Apply limit if it wasn't pushed to scan
+ if let Some(fetch) = scan.fetch
+ && scan_limit.is_none()
+ {
+ plan = Arc::new(GlobalLimitExec::new(plan, 0, Some(fetch)));
+ }
+
+ Ok(plan)
+ }
+
+ /// Compute the column indices needed for the scan based on projection
+ /// expressions and post-scan filters.
+ fn compute_scan_projection(
+ &self,
+ projection: &Option<Vec<Expr>>,
+ post_filters: &[Expr],
+ source_schema: &Schema,
+ ) -> Result<Option<Vec<usize>>> {
+ // Collect all columns needed
+ let mut required_columns = HashSet::new();
+
+ // Add columns from projection expressions
+ if let Some(exprs) = projection {
+ for expr in exprs {
+ expr.apply(|e| {
+ if let Expr::Column(col) = e {
+ required_columns.insert(col.name().to_string());
+ }
+ Ok(TreeNodeRecursion::Continue)
+ })?;
+ }
+ }
+
+ // Add columns from post-scan filters
+ for filter in post_filters {
+ filter.apply(|e| {
+ if let Expr::Column(col) = e {
+ required_columns.insert(col.name().to_string());
+ }
+ Ok(TreeNodeRecursion::Continue)
+ })?;
+ }
+
+ // If no projection specified and no filters, return None (all columns)
+ if projection.is_none() && post_filters.is_empty() {
+ return Ok(None);
+ }
+
+ // If projection is None but we have filters, we need all columns
+ if projection.is_none() {
+ return Ok(None);
+ }
+
+ // Convert column names to indices
+ let indices: Vec<usize> = required_columns
+ .iter()
+ .filter_map(|name| source_schema.index_of(name).ok())
+ .sorted()
+ .collect();
+
+ if indices.is_empty() {
+ Ok(None)
+ } else {
+ Ok(Some(indices))
+ }
+ }
Review Comment:
When the required column set is empty (e.g. `SELECT 1 FROM wide_table`), this returns `None` which makes it scan all columns. IMO this is the opposite of what we want and would return `Some(vec![])`, or a single cheap column if empty projections aren't provider safe.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org