viirya commented on code in PR #26214:
URL: https://github.com/apache/datafusion/pull/26214#discussion_r4425954148
##########
datafusion/optimizer/src/decorrelate_predicate_subquery.rs:
##########
@@ -1037,6 +1042,38 @@ fn build_join(
}
let left = projected_left.as_ref().unwrap_or(left);
+ // The first `V` keys of a null-aware join are its `NOT IN` value keys.
+ //
+ // A tuple element that cannot be NULL in scope on either side never makes
+ // the tuple comparison UNKNOWN: compared with a definite value it is
either
+ // equal, which leaves the other elements to decide, or a definite
mismatch,
+ // which makes the comparison FALSE whatever the other elements are. It
+ // therefore only selects which subquery rows a NULL in another element is
+ // compared with, which is what a correlation scope key does. Moving such
+ // elements after the ones that can be NULL turns them into scope keys, so
+ // the hash join finds the rows a NULL-valued row is compared with through
a
+ // hash lookup instead of pairing it with every row of the other side
+ // (<https://github.com/apache/datafusion/issues/26088>).
+ //
+ // A null-aware tuple has at least one element that can be NULL, so `V` is
+ // at least one. With exactly one, the join is a scalar `NOT IN` on that
+ // element, correlated by the other elements.
+ let null_aware_value_keys = if null_aware && in_values.len() > 1 {
+ let (nullable, non_nullable): (Vec<_>, Vec<_>) = in_values
+ .into_iter()
+ .zip(elements_may_be_null)
+ .partition(|(_, may_be_null)| *may_be_null);
Review Comment:
I could reproduce this, but the promotion is not what makes it wrong. The
same data already gives wrong answers through paths this PR does not touch. On
this branch, with your `o` and `i`:
- `SELECT a, b IS NULL FROM o` returns `false` for the `a = 1` row.
- `SELECT a, b NOT IN (SELECT y FROM i) FROM o` returns `true` for that row,
where it should be NULL. The scalar `NOT IN` decides null-awareness from the
same schema nullability, so it has the same problem.
The root cause is a field declared non-nullable over an array that holds a
logical NULL. DataFusion relies on `nullable` to skip NULL handling in many
places, so guarding only the tuple elements here would leave the scalar form
and `IS NULL` inconsistent with it. I'd rather not special-case encoded types
in this PR. Would a separate issue about this mismatch (non-nullable fields
inferred for dictionaries whose values hold NULLs) work for you?
##########
datafusion/optimizer/src/decorrelate_predicate_subquery.rs:
##########
@@ -953,16 +955,19 @@ fn build_join(
None => vec![],
};
- // True if an `IN` value or the subquery column it is compared with can be
- // NULL inside the scope of an outer row.
- let mut value_may_be_null = false;
- for in_value in &in_values {
- value_may_be_null |= in_value.may_be_null_in_scope(
- left.schema(),
- sub_query_alias.schema(),
- &pull_up.correlated_filters,
- )?;
- }
+ // For each `IN` value, true if it or the subquery column it is compared
+ // with can be NULL inside the scope of an outer row.
+ let elements_may_be_null = in_values
+ .iter()
+ .map(|in_value| {
+ in_value.may_be_null_in_scope(
Review Comment:
Good catch, fixed in 98c253010. `key_may_be_null_in_scope` no longer accepts
a filter proof for a volatile key, so a nullable volatile element stays a value
key.
The scalar form goes through the same check, added in #25338, so it had the
same problem: `CASE WHEN random() < 0.5 THEN o.k END NOT IN (SELECT y FROM i
WHERE (CASE WHEN random() < 0.5 THEN o.k END) > i.lo)` returned TRUE for about
half of the rows on this branch. The fix covers both forms.
`mark_join_for_volatile_in_predicate_correlation_is_null_aware` builds this
shape with a volatile test UDF and checks that the join stays null-aware. It
fails without the fix.
##########
datafusion/optimizer/src/decorrelate_predicate_subquery.rs:
##########
@@ -1037,6 +1042,38 @@ fn build_join(
}
let left = projected_left.as_ref().unwrap_or(left);
+ // The first `V` keys of a null-aware join are its `NOT IN` value keys.
+ //
+ // A tuple element that cannot be NULL in scope on either side never makes
+ // the tuple comparison UNKNOWN: compared with a definite value it is
either
+ // equal, which leaves the other elements to decide, or a definite
mismatch,
+ // which makes the comparison FALSE whatever the other elements are. It
+ // therefore only selects which subquery rows a NULL in another element is
+ // compared with, which is what a correlation scope key does. Moving such
+ // elements after the ones that can be NULL turns them into scope keys, so
+ // the hash join finds the rows a NULL-valued row is compared with through
a
+ // hash lookup instead of pairing it with every row of the other side
+ // (<https://github.com/apache/datafusion/issues/26088>).
+ //
+ // A null-aware tuple has at least one element that can be NULL, so `V` is
+ // at least one. With exactly one, the join is a scalar `NOT IN` on that
+ // element, correlated by the other elements.
+ let null_aware_value_keys = if null_aware && in_values.len() > 1 {
+ let (nullable, non_nullable): (Vec<_>, Vec<_>) = in_values
+ .into_iter()
+ .zip(elements_may_be_null)
+ .partition(|(_, may_be_null)| *may_be_null);
+ let null_aware_value_keys = nullable.len();
Review Comment:
Confirmed, and fixed in 44471f17c (Q14: e3d0754c9). The rare prefix kept the
matches per row low, so the sampled estimate rejected grouping, and the
fallback then looked up every remaining probe row without a limit.
The decision against grouping is no longer final. The lookup per probe row
now goes on only until it wastes as many matches again. Those wasted matches
pay for a sort too, so the join then sorts the rest of the probe rows by hash
and pairs them by group, whatever the estimate says. The same averaging also
affected marks on build rows (`LeftAnti`), where the groups held fewer than two
rows on average.
On your tables (one partition, batch size 8,192, `ci` test profile), the
promoted plan went from 5.2 s before the fix to 22 ms, against 21 ms for the
two-value-key plan.
Added:
- `hot_scope_among_rare_ones_is_grouped`, a unit test for both mark sides
that bounds the matches walked.
- Q14 in the `null_aware_join` benchmark: a scope key with a hot value for
3/4 of the rows and rare values for the rest, 50% NULLs in the other element,
and the outer side probed. At 10k rows it takes 23 ms, against 899 ms with the
old rule and 709 ms with both elements as value keys.
##########
datafusion/physical-plan/src/joins/hash_join/not_in.rs:
##########
@@ -672,6 +652,54 @@ fn correlated_row_state(
))
}
+/// Hash table over the correlation scope keys (`on[value_keys..]`) of all
+/// build rows, used by correlated `NOT IN` (see [`NotInScope::Correlated`]).
+///
+/// Only the probe rows with a NULL value key look it up, so it is built on
+/// the first such probe row rather than with the build side: a probe side
+/// without NULLs never pays for a second hash table over the build side.
+pub(super) struct BuildScopeMap {
+ map: OnceLock<Box<dyn JoinHashMapType>>,
+ /// Charged for the map once it is built. Its lock also makes the probe
+ /// partitions build the map only once.
+ reservation: Mutex<MemoryReservation>,
+ metrics: BuildProbeJoinMetrics,
+}
+
+impl BuildScopeMap {
+ /// Returns the map, building it on first use from `scope_values`, the
+ /// build side's scope keys (`JoinLeftData::values()[value_keys..]`).
+ fn get_or_try_init(
+ &self,
+ scope_values: &[ArrayRef],
+ random_state: &RandomState,
+ ) -> Result<&dyn JoinHashMapType> {
+ if let Some(map) = self.map.get() {
+ return Ok(map.as_ref());
+ }
+ let mut reservation = self.reservation.lock();
+ if let Some(map) = self.map.get() {
+ return Ok(map.as_ref());
+ }
+ // Scope-only NULL marking uses a HashMap (the primary join map may
+ // use ArrayMap for full-key matches, but scope keys have arbitrary
+ // shape).
+ let num_rows = scope_values[0].len();
+ let mut map = new_join_hashmap(num_rows, &mut reservation,
&self.metrics)?;
Review Comment:
Fixed in 0fa56a135. When the memory pool declines the scope map,
`BuildScopeMap` records that. The NULL-valued probe rows are then paired with
every build row instead, and each pair is checked for scope equality before the
value tuples and the residual filter. This is the same enumeration as the
`value_keys=2` layout. The smaller map over the NULL-valued build rows, which
is built with the build side, falls back the same way.
Added:
- `null_aware_scope_map_falls_back_when_memory_is_short`: your setup, with a
pool that holds the build side but not the scope map. It returns 9,899 rows,
and `build_mem_used` equals that of the run without a NULL probe value.
- `null_aware_null_build_scope_map_falls_back_when_memory_is_short`, for the
map over the NULL-valued build rows.
##########
docs/source/library-user-guide/upgrading/56.0.0.md:
##########
@@ -934,9 +934,9 @@ Wire compatibility is directional:
### Null-aware joins record their `NOT IN` value key count
Multi-column `(a, b) NOT IN (SELECT x, y ...)` is now planned as a null-aware
-join. Its first equi-join keys are the tuple elements, followed by any
-correlation keys, so a null-aware join now records how many of its leading keys
-are `NOT IN` value keys:
+join. Its first equi-join keys are the tuple elements that can be NULL,
+followed by the elements that cannot and any correlation keys, so a null-aware
Review Comment:
Done in 53cd54d96. The guide now says that the elements that cannot be NULL
and any equality correlations become equi-join keys when their types can be
hashed, and otherwise stay in the join filter.
--
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]
