kumarUjjawal commented on code in PR #26042:
URL: https://github.com/apache/datafusion/pull/26042#discussion_r4197675545
##########
datafusion/physical-expr/src/equivalence/class.rs:
##########
@@ -847,6 +908,24 @@ impl EquivalenceGroup {
Ok(group)
}
+ /// Retain only expressions that evaluate to NULL when the joined row is
+ /// null-extended. Rebuilding the classes also drops constant information
+ /// that may no longer hold after an outer join.
+ fn with_null_preserving_expressions(self, null_batch: &RecordBatch) ->
Self {
+ let classes = self.classes.into_iter().filter_map(|class| {
+ let expressions = class.into_iter().filter(|expr| {
+ !is_volatile(expr)
+ && expr
+ .evaluate(null_batch)
+ .and_then(|value| value.into_array(1))
+ .is_ok_and(|array| array.null_count() == 1)
Review Comment:
`array.null_count() == 1` reads the physical null buffer, so arrays with
only logical nulls report 0 and their valid equivalences are dropped.
##########
datafusion/physical-expr/src/equivalence/class.rs:
##########
@@ -815,17 +819,74 @@ impl EquivalenceGroup {
join_type: &JoinType,
left_size: usize,
on: &[(PhysicalExprRef, PhysicalExprRef)],
+ ) -> Result<Self> {
+ self.join_with_optional_schema(right_equivalences, join_type,
left_size, None, on)
+ }
+
+ /// Combine equivalence groups for a join with its output schema available.
+ pub(crate) fn join_with_schema(
+ &self,
+ right_equivalences: &Self,
+ join_type: &JoinType,
+ left_size: usize,
+ join_schema: &SchemaRef,
+ on: &[(PhysicalExprRef, PhysicalExprRef)],
+ ) -> Result<Self> {
+ self.join_with_optional_schema(
+ right_equivalences,
+ join_type,
+ left_size,
+ Some(join_schema),
+ on,
+ )
+ }
+
+ fn join_with_optional_schema(
+ &self,
+ right_equivalences: &Self,
+ join_type: &JoinType,
+ left_size: usize,
+ join_schema: Option<&SchemaRef>,
+ on: &[(PhysicalExprRef, PhysicalExprRef)],
) -> Result<Self> {
let group = match join_type {
JoinType::Inner | JoinType::Left | JoinType::Full |
JoinType::Right => {
- let mut result = Self::new(
- self.iter().cloned().chain(
- right_equivalences
- .iter()
- .map(|cls| cls.try_with_offset(left_size as _))
- .collect::<Result<Vec<_>>>()?,
- ),
+ let mut left = self.clone();
+ let mut right = Self::new(
+ right_equivalences
+ .iter()
+ .map(|cls| cls.try_with_offset(left_size as _))
+ .collect::<Result<Vec<_>>>()?,
);
+
+ // Outer joins introduce rows where one side's columns are all
+ // NULL. Keep only equivalences that remain true for that
+ // null-extended row; expressions such as coalesce(a, 0) do
+ // not, even when they were equivalent to a column before the
+ // join.
+ if let Some(join_schema) = join_schema {
+ match join_type {
+ JoinType::Left => {
+ right =
right.with_null_preserving_expressions(&null_batch(
+ join_schema,
+ )?);
+ }
+ JoinType::Right => {
+ left =
left.with_null_preserving_expressions(&null_batch(
+ join_schema,
+ )?);
+ }
+ JoinType::Full => {
+ let batch = null_batch(join_schema)?;
Review Comment:
`null_batch` allocates a new schema and one null array per output column on
every outer-join property computation, even when the nullable side has no
equivalence classes.**
##########
datafusion/physical-expr/src/equivalence/class.rs:
##########
@@ -847,6 +908,24 @@ impl EquivalenceGroup {
Ok(group)
}
+ /// Retain only expressions that evaluate to NULL when the joined row is
+ /// null-extended. Rebuilding the classes also drops constant information
+ /// that may no longer hold after an outer join.
+ fn with_null_preserving_expressions(self, null_batch: &RecordBatch) ->
Self {
+ let classes = self.classes.into_iter().filter_map(|class| {
+ let expressions = class.into_iter().filter(|expr| {
Review Comment:
The filter keeps only members that evaluate to NULL, so it drops members
that still agree on null-extended rows because they evaluate to the same
non-NULL value.
##########
datafusion/physical-expr/src/equivalence/class.rs:
##########
@@ -957,6 +1059,18 @@ mod tests {
use crate::expressions::{BinaryExpr, Column, binary, col, lit};
use arrow::datatypes::{DataType, Field, Schema, SchemaRef};
+ #[test]
+ fn full_join_equivalences_support_empty_schema() -> Result<()> {
+ let schema = Arc::new(Schema::empty());
+ let left = EquivalenceGroup::new(vec![]);
+ let right = EquivalenceGroup::new(vec![]);
+
+ let result = left.join_with_schema(&right, &JoinType::Full, 0,
&schema, &[])?;
+
+ assert_eq!(result.iter().count(), 0);
+ Ok(())
+ }
+
use datafusion_expr::Operator;
Review Comment:
The new test `full_join_equivalences_support_empty_schema` was inserted
between the test module's `use` lines, leaving `use datafusion_expr::Operator;`
stranded after a function.
##########
datafusion/physical-expr/src/equivalence/class.rs:
##########
Review Comment:
https://github.com/apache/datafusion/blob/5f1cabcd87a9964ce654186b5056f4acf06ec2cc/datafusion/physical-expr/src/equivalence/class.rs#L816
The public `EquivalenceGroup::join` (no schema) keeps the old unsound
outer-join behavior with no doc warning or deprecation, so external callers
still get wrong equivalences.
##########
datafusion/physical-expr/src/equivalence/class.rs:
##########
@@ -847,6 +908,24 @@ impl EquivalenceGroup {
Ok(group)
}
+ /// Retain only expressions that evaluate to NULL when the joined row is
+ /// null-extended. Rebuilding the classes also drops constant information
+ /// that may no longer hold after an outer join.
+ fn with_null_preserving_expressions(self, null_batch: &RecordBatch) ->
Self {
+ let classes = self.classes.into_iter().filter_map(|class| {
+ let expressions = class.into_iter().filter(|expr| {
+ !is_volatile(expr)
+ && expr
+ .evaluate(null_batch)
Review Comment:
Planning now runs every non-volatile expression in an equivalence class,
including user and third-party ScalarUDFs, on a synthetic all-NULL batch; it
does this every time join properties are recomputed, and a panic inside
evaluate is not caught.
##########
datafusion/sqllogictest/test_files/joins.slt:
##########
@@ -2074,6 +2074,55 @@ set datafusion.explain.logical_plan_only = false;
statement ok
set datafusion.execution.target_partitions = 4;
+statement ok
+set datafusion.execution.target_partitions = 1;
+
+statement ok
+CREATE TABLE outer_join_sort_left (id BIGINT) AS VALUES (1), (2);
+
+statement ok
+CREATE TABLE outer_join_sort_right (a BIGINT, b BIGINT) AS VALUES (2, 2), (3,
3);
+
+query III
Review Comment:
The regression tests cover only LEFT and FULL OUTER JOIN; the new
`JoinType::Right` branch (class.rs:874) and the cross-join path are never
exercised.
##########
datafusion/physical-expr/src/equivalence/class.rs:
##########
@@ -902,6 +981,29 @@ impl EquivalenceGroup {
}
}
+/// Build a one-row batch with every field NULL so expressions can be checked
+/// for their behavior on an outer join's null-extended rows.
+fn null_batch(schema: &SchemaRef) -> Result<RecordBatch> {
+ let schema = Arc::new(Schema::new(
+ schema
+ .fields()
+ .iter()
+ .map(|field| Field::new(field.name(), field.data_type().clone(),
true))
Review Comment:
`null_batch` rebuilds each field with `Field::new(name, type, true)`, which
drops field metadata such as extension-type annotations.
--
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]
