comphead commented on code in PR #19857:
URL: https://github.com/apache/datafusion/pull/19857#discussion_r4106477505
##########
datafusion/physical-plan/src/joins/hash_join/stream.rs:
##########
@@ -1541,6 +1556,58 @@ fn mark_null_candidates_for_probe_batch(
Ok(())
}
+/// Keeps the candidate pairs whose multi-column `NOT IN` value tuples are not
a
+/// definite mismatch, i.e. every element pair is equal or involves a NULL.
+///
+/// Such a pair compares UNKNOWN when some element is NULL (TRUE pairs, with no
+/// NULL, are found by the hash lookup instead), whereas a pair with a definite
+/// mismatch compares FALSE whatever its NULLs are: `(NULL, 1) = (2, 3)` is
+/// `UNKNOWN AND FALSE`, which is FALSE.
+fn retain_value_mismatch_free(
+ build_value_keys: &[ArrayRef],
+ probe_value_keys: &[ArrayRef],
+ build_indices: UInt64Array,
+ probe_indices: UInt32Array,
+) -> Result<(UInt64Array, UInt32Array)> {
+ if build_indices.is_empty() {
+ return Ok((build_indices, probe_indices));
+ }
+ let comparators = build_value_keys
+ .iter()
+ .zip(probe_value_keys)
+ .map(|(build, probe)| {
+ Ok((
+ build.logical_nulls(),
+ probe.logical_nulls(),
+ // Same total order the hash join's own key comparison uses.
Review Comment:
This comment no longer holds. `make_comparator` uses IEEE 754 totalOrder, so
`-0.0` vs `+0.0` counts as a definite mismatch here. The join's own key
equality (`JoinKeyComparator::new` via `normalize_float_zero`) and SQL `=`
treat them as equal. The UNKNOWN check then drops pairs that the exact-match
path treats as equal:
```sql
CREATE TABLE fo(a DOUBLE, b DOUBLE) AS VALUES (NULL, 0.0), (1.0, 0.0), (2.0,
5.0);
CREATE TABLE fi(x DOUBLE, y DOUBLE) AS VALUES (1.0, arrow_cast('-0.0',
'Float64'));
CREATE TABLE fi_null(x DOUBLE, y DOUBLE) AS VALUES (NULL, arrow_cast('-0.0',
'Float64'));
-- returns (2, 5) and (NULL, 0). Expected only (2, 5): (NULL, 0) vs (1,
-0.0) is UNKNOWN AND TRUE
SELECT * FROM fo WHERE (a, b) NOT IN (SELECT x, y FROM fi);
-- mark for (NULL, 0) is true, expected NULL
SELECT a, b, (a, b) NOT IN (SELECT x, y FROM fi) FROM fo;
-- returns all three rows, expected only (2, 5)
SELECT * FROM fo WHERE (a, b) NOT IN (SELECT x, y FROM fi_null);
```
Normalizing both arrays with `normalize_float_zero` before `make_comparator`
fixes all three locally, and the existing SLT files still pass.
This setup also runs once per `batch_size` chunk of candidate pairs and
works on the full build arrays. With normalization added, that becomes a full
scan of each float build key per chunk. Building the comparators once per probe
batch avoids it. A batch-level alternative is `take` on both sides plus the
float-normalizing comparison in `datum::apply_cmp`, which keeps the semantics
in one place.
##########
datafusion/optimizer/src/decorrelate_predicate_subquery.rs:
##########
@@ -613,34 +666,54 @@ fn build_join(
// Projecting the constant as a column of the outer side turns the
predicate
// into a real equi-join key, which fixes all three. It only pays for
itself
// on a join that ends up null-aware.
+ //
+ // Every constant element of a tuple is projected the same way, so that all
+ // value keys lead the equi-join keys in order.
let mut projected_left = None;
- if let Some((value, right_col, mut value_name, correlation_opt)) =
in_value_expr
- && value.column_refs().is_empty()
- && null_aware
+ if null_aware
+ && in_values
+ .iter()
+ .any(|(value, _)| value.column_refs().is_empty())
{
- // The projected column is unqualified, so a left field that already
has
- // this name — however unlikely — would make the reference ambiguous.
let left_schema = left.schema();
- while left_schema.fields().iter().any(|f| f.name() == &value_name) {
- value_name.push('_');
- }
- let value_col = Column::new_unqualified(value_name);
- let projections = left_schema
+ let mut projections = left_schema
.columns()
.into_iter()
.map(Expr::from)
- .chain(std::iter::once(value.alias(value_col.name())))
.collect::<Vec<_>>();
+ let is_tuple = in_values.len() > 1;
+ for (i, (value, _)) in in_values.iter_mut().enumerate() {
+ if !value.column_refs().is_empty() {
+ continue;
+ }
+ let mut value_name = if is_tuple {
+ format!("{alias}_value_{i}")
+ } else {
+ format!("{alias}_value")
+ };
+ // The projected column is unqualified, so a left field that
already
+ // has this name — however unlikely — would make the reference
+ // ambiguous.
+ while left_schema.fields().iter().any(|f| f.name() == &value_name)
{
+ value_name.push('_');
+ }
+ let value_col = Column::new_unqualified(value_name);
+ let constant = std::mem::replace(value,
Expr::Column(value_col.clone()));
+ projections.push(constant.alias(value_col.name()));
+ }
projected_left = Some(
LogicalPlanBuilder::from(left.clone())
.project(projections)?
.build()?,
);
- // Rebuild the `IN` equality against the projected column.
- let in_predicate = Expr::eq(Expr::Column(value_col),
Expr::Column(right_col));
- join_filter = in_predicate_first(in_predicate, correlation_opt);
+ // Rebuild the `IN` equalities against the projected columns.
+ join_filter = in_predicate_first(in_values_predicate(&in_values),
in_correlation);
}
let left = projected_left.as_ref().unwrap_or(left);
+ // The `IN` equalities lead the join filter, so after they are extracted
+ // into equi-join keys the first `null_aware_value_keys` keys are the
+ // `NOT IN` value keys (see `Join::null_aware_value_keys`).
+ let null_aware_value_keys = in_values.len().max(1);
Review Comment:
`V` is fixed here, before `ExtractEquijoinPredicate` runs, and nothing later
checks that the first `V` keys of `on` are the tuple equalities.
`split_op_and_other_join_predicates` leaves an equality in the filter when
`can_hash` is false (`RunEndEncoded`, `ListView`, `Union`). A correlation key
then moves into the value-key range:
```sql
CREATE TABLE co(k INT, a INT, b INT) AS VALUES (1, 1, 2), (2, 1, 2), (3,
NULL, 5), (4, 7, 7), (5, 4, 2);
CREATE TABLE ci(k INT, x INT, y INT) AS VALUES (1, 1, 2), (3, NULL, 2), (3,
9, 5);
CREATE TABLE ree_o AS SELECT k, arrow_cast(a, 'RunEndEncoded("run_ends":
non-null Int32, "values": Int32)') AS a, b FROM co;
CREATE TABLE ree_i AS SELECT k, arrow_cast(x, 'RunEndEncoded("run_ends":
non-null Int32, "values": Int32)') AS x, y FROM ci;
SELECT * FROM ree_o WHERE (a, b) NOT IN (SELECT x, y FROM ree_i WHERE
ree_i.k = ree_o.k);
-- HashJoinExec: on=[(b@2, y@1), (k@0, k@2)], filter=a@0 = x@1,
null_aware(value_keys=2)
-- returns (3, NULL, 5). Expected it excluded: (NULL, 5) vs (9, 5) is UNKNOWN
```
The scalar form already has this gap on `main`. `a NOT IN (SELECT x ...
WHERE ree_i.k = ree_o.k)` plans `on=[(k@0, k@1)], filter=a@0 = x@1` and returns
the same wrong row. This PR widens the assumption from one key to `V`. A guard
here would cover both: if any value pair fails `find_valid_equijoin_key_pair`
or `can_hash`, error out (for example `not_impl_err!`) instead of building a
null-aware join.
##########
datafusion/physical-plan/src/joins/hash_join/exec.rs:
##########
@@ -245,8 +256,18 @@ impl NullAwareMode {
join_type: JoinType,
partition_mode: PartitionMode,
num_keys: usize,
+ num_value_keys: usize,
has_filter: bool,
) -> Result<Self> {
+ if num_value_keys == 0 || num_value_keys > num_keys {
+ return plan_err!(
+ "null_aware {join_type} join needs between 1 and {num_keys}
`NOT IN` value keys, got {num_value_keys}"
+ );
+ }
+ // `num_keys > 1` also covers a multi-column value key without any
+ // correlation: a NULL in one tuple element leaves the comparison FALSE
+ // whenever another element is a definite mismatch, so it too must be
+ // decided per build row.
let correlated = num_keys > 1 || has_filter;
Review Comment:
Sending every tuple through the correlated per-build-row path has two costs
that the scalar path does not.
Memory: `can_swap_hash_join` only swaps a null-aware join with a single key,
so the outer side is always the build side:
```sql
CREATE TABLE big AS SELECT CASE WHEN v < 0 THEN NULL ELSE v END AS a, v %
1000 AS b FROM generate_series(1, 2000000) t(v);
CREATE TABLE small AS SELECT CASE WHEN v < 0 THEN NULL ELSE v END AS x, v %
1000 AS y FROM generate_series(1, 100) t(v);
-- datafusion-cli --memory-limit 64M
SELECT count(*) FROM big WHERE a NOT IN (SELECT x FROM small);
-- RightAnti, builds on `small`: runs
SELECT count(*) FROM big WHERE (a, b) NOT IN (SELECT x, y FROM small);
-- LeftAnti, builds on `big`: Resources exhausted ... Failed to allocate
additional 75.6 MB for HashJoinInput
```
NULLs: without correlation, every NULL-valued row is paired with every row
on the other side, and each pair goes through `retain_value_mismatch_free`.
Measured on this branch with `release-nonlto`, equal row counts on both sides,
and no matches:
| rows per side | nullable, no NULLs | 10% NULL in subquery | 10% NULL in
outer |
|---|---|---|---|
| 20k | 1 ms | 96 ms | 68 ms |
| 40k | 1 ms | 270 ms | 221 ms |
| 80k | 2 ms | 1.27 s | 1.23 s |
| 160k | 3 ms | 4.00 s | 5.15 s |
<details><summary>data</summary>
```sql
CREATE TABLE o AS SELECT v AS a, v % 1000 AS b FROM generate_series(1, N)
t(v);
CREATE TABLE o_null AS SELECT CASE WHEN v % 10 = 0 THEN NULL ELSE v END AS
a, v % 1000 AS b FROM generate_series(1, N) t(v);
CREATE TABLE i_null AS SELECT CASE WHEN v % 10 = 0 THEN NULL ELSE v +
10000000 END AS x, v % 1000 + 5000 AS y FROM generate_series(1, N) t(v);
CREATE TABLE i_nullable AS SELECT CASE WHEN v < 0 THEN NULL ELSE v +
10000000 END AS x, v % 1000 + 5000 AS y FROM generate_series(1, N) t(v);
SELECT count(*) FROM o WHERE (a, b) NOT IN (SELECT x, y FROM i_nullable);
-- nullable, no NULLs
SELECT count(*) FROM o WHERE (a, b) NOT IN (SELECT x, y FROM i_null);
-- 10% NULL in subquery
SELECT count(*) FROM o_null WHERE (a, b) NOT IN (SELECT x, y FROM
i_nullable); -- 10% NULL in outer
```
</details>
The NULL-free path already beats Spark, which plans nullable multi-column
`NOT IN` as a `BroadcastNestedLoopJoinExec`. The NULL path has the same shape
as PostgreSQL's `findPartialMatch`, but PostgreSQL builds on the subquery side,
deduplicates, and decides each outer row as it streams. A multi-key null-aware
`RightAnti` could do the same. A cheaper narrowing: an element that is NOT NULL
on both sides can never make the comparison UNKNOWN, so it can act as a scope
key instead of a value key. Following up on my earlier benchmark note, a
benchmark for both shapes would help. Follow-ups are fine if the cost is
documented.
##########
datafusion/expr/src/expr.rs:
##########
@@ -1399,6 +1399,34 @@ impl InSubquery {
negated,
}
}
+
+ /// The tuple elements of a multi-column `(a, b, ...) IN (SELECT x, y,
...)`,
+ /// or `None` for a single-column `IN`. See [`in_subquery_tuple_values`].
+ pub fn tuple_values(&self) -> Option<&[Expr]> {
+ in_subquery_tuple_values(&self.expr, &self.subquery.subquery)
+ }
+}
+
+/// The tuple elements of a multi-column `(a, b, ...) IN (SELECT x, y, ...)`,
+/// given the compared expression `expr` and the `subquery` plan, or `None`
+/// for a single-column `IN`.
+///
+/// The tuple is planned as a `struct` call. It is a multi-column `IN` only
when
+/// the subquery returns more than one column; against a single column the
+/// `struct` is one value compared with a struct-typed column. The number of
+/// elements is not checked against the number of subquery columns here.
+pub fn in_subquery_tuple_values<'a>(
+ expr: &'a Expr,
+ subquery: &crate::LogicalPlan,
+) -> Option<&'a [Expr]> {
+ match expr {
+ Expr::ScalarFunction(func)
+ if func.func.name() == "struct" &&
subquery.schema().fields().len() > 1 =>
Review Comment:
Matching on the `struct` function name breaks once `simplify_expressions`
folds an all-literal tuple into a `ScalarValue::Struct` literal. On `main` this
query fails in the SQL planner with "Too many columns!". Now the planner
accepts it, and it fails in the optimizer:
```sql
SELECT * FROM o WHERE (1, 2) NOT IN (SELECT x, y FROM i);
-- Optimizer rule 'simplify_expressions' failed
-- Invalid (non-executable) plan after Optimizer rule: simplify_expressions
-- caused by: InSubquery should only return one column, but found 2: i.x, i.y
```
Recognizing a struct literal here as well (the return type would need to own
the element exprs) fixes it. An SLT case with an all-literal tuple would cover
it.
--
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]
