sunchao commented on code in PR #26214:
URL: https://github.com/apache/datafusion/pull/26214#discussion_r4418927547
##########
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:
**[P2] Fall back when the memory pool declines the optional scope index**
This allocation makes a previously executable tuple `NOT IN` fail under a
fixed memory limit. I ran `HashJoinExec` on this head with identical inputs,
`LeftAnti`, `CollectLeft`, and a 512 KiB greedy pool. The build rows were
`(2*i, i%100)` for `0 <= i < 10000`, the probe rows `(0,0)` and `(NULL,1)`, and
both group fields were `NOT NULL`.
With the base layout (`value_keys=2`) the join returned 9,899 rows and
reported 401,092 build bytes. Changing only `value_keys` to 1 failed on the
311.1 KiB scope-map reservation with 120.3 KiB available. The promoted layout
returned the same 9,899 rows with an unlimited pool, using 719,684 build bytes.
Could a declined index reservation switch to bounded pair enumeration, still
checking scope equality and the residual filter? The index speeds up the
comparison, but the join can produce the correct result without it. A
fixed-pool regression alongside the lazy-allocation test would cover this.
##########
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:
**[P1] A correlated filter cannot prove a separate volatile expression
non-NULL**
`filter_rejects_null` matches expressions structurally. If the tuple and the
subquery filter each contain a call to `random()`, they can produce different
values for the same outer row. A non-NULL result in the filter says nothing
about the call in the tuple.
```sql
CREATE TABLE o(k INT NOT NULL, a INT) AS
SELECT 1, CAST(NULL AS INT) FROM range(0,4096);
CREATE TABLE i(x INT, y INT NOT NULL, lo INT NOT NULL) AS
SELECT 5,1,0 FROM range(0,128);
SELECT u, count(*) FROM (
SELECT (a, CASE WHEN random() < 0.5 THEN o.k END) NOT IN (
SELECT x,y FROM i WHERE (CASE WHEN random() < 0.5 THEN o.k END) > i.lo
) AS u FROM o
) q GROUP BY u;
```
The outer tuple is either `(NULL,1)` or `(NULL,NULL)`. Both compare UNKNOWN
to every in-scope `(5,1)`. With 128 independent chances for the filter to pass,
an empty subquery is exceedingly unlikely. Base `86d7a490` returned NULL for
all 4,096 rows. The head returned TRUE for 2,034. Keeping both value keys on
the head by projecting `NULLIF(y,999)` restored all 4,096 NULLs with identical
input values.
Please keep a nullable volatile expression as a value key when the only
proof comes from matching a correlated predicate. A deterministic volatile UDF
could exercise this in a test.
##########
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:
**[P1] Keep dictionary elements with possible logical NULLs among the value
keys**
An Arrow dictionary can have entirely valid keys that point to a NULL
dictionary entry. `RecordBatch::try_from_iter` infers a non-nullable field for
that array from its physical null count. A normal `MemTable` built this way
reproduces a wrong result without overriding the schema:
- Outer `a = [1, NULL]`, `b = Dictionary<Int32, Int32>(keys=[0,1],
values=[NULL,3])`.
- Inner `x = [1]`, `y = Dictionary<Int32, Int32>(keys=[0], values=[2])`.
For `SELECT a, (a,b) NOT IN (SELECT x,y FROM i) FROM o`, the row with `a=1`
is `(1,NULL) NOT IN ((1,2))`. Base `86d7a490` and an explicit three-valued
`EXISTS` reference both returned NULL. On `865e3cd9` it returns TRUE. Promoting
`b = y` puts that row outside every scope, so a `WHERE NOT IN` can now emit it.
Could the promotion require proof against logical NULLs for encoded types,
and cover this with an inferred-schema `RecordBatch` regression?
##########
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:
**[P2] Promotion still regresses a batch with a rare-key prefix and a hot
scope**
The uniform bucket from the previous review is fixed. With a mixture of rare
and hot keys, `scope_matches` can decide once against grouping and then finish
with `lookup_each_row(..., usize::MAX)`. Probe rows already marked UNKNOWN keep
walking the hot chain. The previous two-value-key path drops those rows as it
goes.
I reproduced this through SQL on both revisions (local unoptimized builds,
one partition, batch size 8,192). Both physical plans are `CollectLeft
RightAnti`, with zero output rows. The identical `EXPLAIN ANALYZE` took 74–77
ms on base `86d7a490` and 16.8–24.5 s here (two runs each). Making only the
scope column nullable, with the same values, restored the two-value-key plan on
this head and took 69–71 ms. A traversal probe for one 8,192-row batch counted
155,680 candidate pairs before promotion and 58,982,912 after.
```sql
SET datafusion.execution.target_partitions = 1;
SET datafusion.execution.batch_size = 8192;
CREATE TABLE df26214_probe(a BIGINT, k BIGINT NOT NULL, kn BIGINT) AS
SELECT CAST(NULL AS BIGINT) AS a, k, CASE WHEN k < 0 THEN NULL ELSE k END
AS kn
FROM (
SELECT CASE WHEN r < 256 THEN r + 1 WHEN r >= 7936 THEN r - 7680 + 1 ELSE
0 END AS k
FROM (SELECT value % 8192 AS r FROM range(0, 16384))
);
CREATE TABLE df26214_build(x BIGINT, k BIGINT NOT NULL, kn BIGINT) AS
SELECT CAST(2 AS BIGINT) AS x, k, CASE WHEN k < 0 THEN NULL ELSE k END AS kn
FROM (
SELECT CASE WHEN value <= 512 THEN value ELSE 0 END AS k FROM range(0,
8192)
);
EXPLAIN ANALYZE SELECT count(*) FROM df26214_probe o WHERE (a,k) NOT IN
(SELECT x,k FROM df26214_build);
EXPLAIN ANALYZE SELECT count(*) FROM df26214_probe o WHERE (a,kn) NOT IN
(SELECT x,kn FROM df26214_build);
```
Could the lookup skip a resolved probe row's chain, or reconsider grouping
as the key distribution changes? Q13's NULL positions come from `value % 2` and
its scope from `value % 16`, so its even scopes are entirely NULL and its odd
scopes contain no NULLs. A mixed hot/cold case would catch this regression.
##########
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:
Only **hashable** nonnullable elements appear in the equi-join key list. The
new `RunEndEncoded` case keeps its equality in the residual join filter. Could
the guide distinguish those two representations?
--
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]
