kumarUjjawal commented on code in PR #24821:
URL: https://github.com/apache/datafusion/pull/24821#discussion_r3930694906
##########
datafusion/physical-plan/src/filter.rs:
##########
@@ -1131,6 +1138,66 @@ fn collect_equality_columns(predicate: &Arc<dyn
PhysicalExpr>) -> (HashSet<usize
(eq_values.into_keys().collect(), infeasible)
}
+/// Removes `col IS NOT NULL` conjuncts whose column is provably non-null in
+/// the input statistics (`null_count == Exact(0)`). Returns the original
+/// predicate (Arc clone) when nothing can be dropped.
+///
+/// This is the statistics-driven inverse of the null-rejecting analysis in
+/// `collect_null_rejecting_columns`: there, a surviving `IS NOT NULL` conjunct
+/// *derives* `null_count = Exact(0)` for the output; here, an input that
+/// already proves `Exact(0)` nulls makes the conjunct vacuously true, so
+/// evaluating it per batch is pure overhead (measured: 100% selectivity
+/// join-key filters over NOT NULL data still pay full evaluation + batch
+/// copy per scan).
+fn simplify_not_null_conjuncts(
+ predicate: Arc<dyn PhysicalExpr>,
+ input: &Arc<dyn ExecutionPlan>,
+) -> Arc<dyn PhysicalExpr> {
+ // Only worth consulting statistics when the predicate has a bare
+ // `Column IS NOT NULL` conjunct to drop.
+ let conjuncts = split_conjunction(&predicate);
+ let has_bare_not_null = conjuncts.iter().any(|e| {
+ e.downcast_ref::<IsNotNullExpr>()
+ .and_then(|n| n.arg().downcast_ref::<Column>())
+ .is_some()
+ });
+ if !has_bare_not_null {
+ return predicate;
+ }
+
+ // Global (all-partitions) statistics; per-partition refinement is not
+ // needed for a proof that holds across every partition.
+ let Ok(stats) = StatisticsContext::new().compute(input.as_ref(),
&StatisticsArgs::new()) else {
+ return predicate;
+ };
+ let schema = input.schema();
+ let provably_non_null = |col: &Column| {
+ schema.index_of(col.name()).is_ok_and(|idx| {
+ stats
+ .column_statistics
+ .get(idx)
+ .is_some_and(|cs| matches!(cs.null_count, Precision::Exact(0)))
+ })
+ };
+
+ let mut kept: Vec<Arc<dyn PhysicalExpr>> =
Vec::with_capacity(conjuncts.len());
+ for expr in conjuncts {
+ let drop = expr
+ .downcast_ref::<IsNotNullExpr>()
+ .and_then(|n| n.arg().downcast_ref::<Column>())
+ .is_some_and(provably_non_null);
+ if !drop {
+ kept.push(Arc::clone(expr));
+ }
+ }
+ match kept.len() {
+ 0 => Arc::new(Literal::new(ScalarValue::Boolean(Some(true)))),
Review Comment:
Replacing the predicate with `lit(true)` does not make this operator a
pass-through. `FilterExecStream` still creates a Boolean array, calls
`filter_record_batch`, records metrics, and uses the batch coalescer. Arrow
already handles an all-true mask using zero-copy array slices, so the claimed
full-batch-copy saving is also inaccurate. Please either remove/bypass the
`FilterExec` in this case or provide a current-head benchmark showing that this
path improves the stated workload.
##########
datafusion/physical-plan/src/filter.rs:
##########
@@ -1131,6 +1138,66 @@ fn collect_equality_columns(predicate: &Arc<dyn
PhysicalExpr>) -> (HashSet<usize
(eq_values.into_keys().collect(), infeasible)
}
+/// Removes `col IS NOT NULL` conjuncts whose column is provably non-null in
+/// the input statistics (`null_count == Exact(0)`). Returns the original
+/// predicate (Arc clone) when nothing can be dropped.
+///
+/// This is the statistics-driven inverse of the null-rejecting analysis in
+/// `collect_null_rejecting_columns`: there, a surviving `IS NOT NULL` conjunct
+/// *derives* `null_count = Exact(0)` for the output; here, an input that
+/// already proves `Exact(0)` nulls makes the conjunct vacuously true, so
+/// evaluating it per batch is pure overhead (measured: 100% selectivity
+/// join-key filters over NOT NULL data still pay full evaluation + batch
+/// copy per scan).
+fn simplify_not_null_conjuncts(
+ predicate: Arc<dyn PhysicalExpr>,
+ input: &Arc<dyn ExecutionPlan>,
+) -> Arc<dyn PhysicalExpr> {
+ // Only worth consulting statistics when the predicate has a bare
+ // `Column IS NOT NULL` conjunct to drop.
+ let conjuncts = split_conjunction(&predicate);
+ let has_bare_not_null = conjuncts.iter().any(|e| {
+ e.downcast_ref::<IsNotNullExpr>()
+ .and_then(|n| n.arg().downcast_ref::<Column>())
+ .is_some()
+ });
+ if !has_bare_not_null {
+ return predicate;
+ }
+
+ // Global (all-partitions) statistics; per-partition refinement is not
+ // needed for a proof that holds across every partition.
+ let Ok(stats) = StatisticsContext::new().compute(input.as_ref(),
&StatisticsArgs::new()) else {
+ return predicate;
+ };
+ let schema = input.schema();
+ let provably_non_null = |col: &Column| {
+ schema.index_of(col.name()).is_ok_and(|idx| {
Review Comment:
`Column::name()` is only used for display; physical evaluation uses
`Column::index()`. Looking up statistics by name can select a different field
when names are duplicated, such as after a join. If the first field has
`null_count = Exact(0)` but the referenced same-named field contains NULLs,
this removes the predicate and returns incorrect rows. Please use `col.index()`
and add a regression with duplicate field names.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
