viirya commented on code in PR #26214:
URL: https://github.com/apache/datafusion/pull/26214#discussion_r4431021369
##########
datafusion/optimizer/src/decorrelate_predicate_subquery.rs:
##########
@@ -631,6 +632,34 @@ impl InValue {
scope_filters,
))
}
+
+ /// Can either side hold a NULL that its schema does not show? See
+ /// [`type_may_hide_null`].
+ fn may_hide_null(
+ &self,
+ left_schema: &DFSchema,
+ right_schema: &DFSchema,
+ ) -> Result<bool> {
+ Ok(type_may_hide_null(&self.value.get_type(left_schema)?)
+ ||
type_may_hide_null(&self.subquery_column.get_type(right_schema)?))
+ }
+}
+
+/// Can an array of this type hold a NULL in a field declared non-nullable?
+///
+/// An array of these types can be logically NULL at a slot without a null
+/// bit: a dictionary key that points to a NULL value, a run of NULLs, or a
+/// NULL in the selected union child. The null count covers only the null
+/// bits, so a field inferred from such an array, as
+/// `RecordBatch::try_from_iter` does, is non-nullable although the array
+/// holds NULLs.
+fn type_may_hide_null(data_type: &DataType) -> bool {
+ matches!(
+ data_type,
+ DataType::Dictionary(_, _)
Review Comment:
Confirmed and fixed in 1e61fa76c. `type_may_hide_null` now covers
`DataType::Null` too. With that, it holds every type whose arrays can be
logically NULL without a null bit: `Null`, dictionary, run-end encoded and
union.
`multi_column_not_in_sees_null_type_values` uses your fixture. It expects
NULL for both rows and an empty `WHERE` result, and fails without the fix.
##########
datafusion/physical-plan/src/joins/hash_join/not_in.rs:
##########
@@ -1136,24 +1234,50 @@ where
build_lookups: 0,
#[cfg(test)]
mark_checks: 0,
+ #[cfg(test)]
+ max_decided_rows: 0,
f,
}
}
/// Pairs each of `probe_rows` with each of `build_rows`. The pairing stops
/// when every row that the pairs decide is marked.
+ ///
+ /// When the pairs decide the build rows, the build rows are paired in
+ /// chunks of at most `batch_size` rows. The rows held at once then stay
+ /// bounded by the batch size rather than by the build side, which matters
+ /// when this pairing stands in for a scope map that the memory pool
+ /// declined.
fn cross_product(
&mut self,
build_rows: impl Iterator<Item = u64>,
probe_rows: impl Iterator<Item = u32>,
) -> Result<()> {
- // Drop the rows that are already marked before the first pair.
- self.needs_refresh = true;
match self.marks {
UnknownMarks::Build(_) => {
- self.pair_unmarked(&mut build_rows.collect(), probe_rows, |b,
p| (b, p))
+ let probe_rows: Vec<u32> = probe_rows.collect();
Review Comment:
Fixed in fc6765b22. `cross_product` now takes a closure that produces the
probe rows again for each build chunk. The callers build it from the probe
range or the null bitmap, so no probe index vector is collected when the marks
are on the build rows, whatever the probe batch size.
With marks on the probe rows, the probe rows are still collected once, as on
the merge base, because they are the rows that get dropped as they are marked.
##########
datafusion/physical-plan/src/joins/hash_join/not_in.rs:
##########
@@ -1136,24 +1234,50 @@ where
build_lookups: 0,
#[cfg(test)]
mark_checks: 0,
+ #[cfg(test)]
+ max_decided_rows: 0,
f,
}
}
/// Pairs each of `probe_rows` with each of `build_rows`. The pairing stops
/// when every row that the pairs decide is marked.
+ ///
+ /// When the pairs decide the build rows, the build rows are paired in
+ /// chunks of at most `batch_size` rows. The rows held at once then stay
+ /// bounded by the batch size rather than by the build side, which matters
+ /// when this pairing stands in for a scope map that the memory pool
+ /// declined.
fn cross_product(
&mut self,
build_rows: impl Iterator<Item = u64>,
probe_rows: impl Iterator<Item = u32>,
) -> Result<()> {
- // Drop the rows that are already marked before the first pair.
- self.needs_refresh = true;
match self.marks {
UnknownMarks::Build(_) => {
- self.pair_unmarked(&mut build_rows.collect(), probe_rows, |b,
p| (b, p))
+ let probe_rows: Vec<u32> = probe_rows.collect();
+ if probe_rows.is_empty() {
+ return Ok(());
+ }
+ let mut build_rows = build_rows.peekable();
+ let mut chunk = Vec::with_capacity(self.batch_size);
+ while build_rows.peek().is_some() {
+ chunk.clear();
+ chunk.extend(build_rows.by_ref().take(self.batch_size));
Review Comment:
Right, and fixed in fc6765b22 too. A chunk is now filled while the marks
bitmap is held, skipping the marked build rows, until it holds `batch_size`
unmarked rows. A build side with one unmarked row per batch then pairs them all
in one chunk, as the merge base did.
`cross_product_fills_chunks_with_unmarked_build_rows` has one unmarked build
row in each of 64 batches of 64 rows. It checks for one call with 64 pairs.
Without the fix, it sees 64 calls with 4,096 pairs.
##########
datafusion/optimizer/src/decorrelate_predicate_subquery.rs:
##########
@@ -631,6 +632,34 @@ impl InValue {
scope_filters,
))
}
+
+ /// Can either side hold a NULL that its schema does not show? See
+ /// [`type_may_hide_null`].
+ fn may_hide_null(
Review Comment:
Good point, fixed in 1e61fa76c. Both of your variants do promote `b`: the
plan shows `CAST(o.b AS Int64) = __correlated_sq_1.y` as a scope key. On this
data the query then fails before it returns a result: the `ProjectionExec` that
computes `CAST(o.b AS Int64)` rejects the NULL in a field declared
non-nullable. That error comes from the same inference and does not depend on
the key order, so it belongs with the separate inference issue.
`may_hide_null` now checks the key's type and the type of every column the
key reads. Those columns are looked up in the outer plan, and for the subquery
side in the subquery plan under its output projection. A column that cannot be
found counts as hiding a NULL. This also covers functions other than casts.
Because the query fails on this data,
`multi_column_not_in_keeps_cast_dictionary_as_value_key` checks the plan
instead. It covers a cast on either side and expects
`null_aware(value_keys=2)`. It fails without the fix.
##########
datafusion/optimizer/src/decorrelate_predicate_subquery.rs:
##########
@@ -1037,6 +1079,51 @@ 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.
+ //
+ // An element whose type can hide a NULL from its schema (see
+ // `type_may_hide_null`) stays a value key too. As a scope key, such a NULL
+ // would put its row outside every scope, which makes the comparison FALSE
+ // instead of UNKNOWN; a value key still sees it.
+ let null_aware_value_keys = if null_aware && in_values.len() > 1 {
+ let keeps_value_key = in_values
+ .iter()
+ .zip(elements_may_be_null)
+ .map(|(in_value, may_be_null)| {
+ Ok(may_be_null
+ || in_value.may_hide_null(left.schema(),
sub_query_alias.schema())?)
+ })
+ .collect::<Result<Vec<_>>>()?;
+ let (nullable, non_nullable): (Vec<_>, Vec<_>) = in_values
+ .into_iter()
+ .zip(keeps_value_key)
+ .partition(|(_, keeps_value_key)| *keeps_value_key);
+ let null_aware_value_keys = nullable.len();
+ in_values = nullable
+ .into_iter()
+ .chain(non_nullable)
+ .map(|(in_value, _)| in_value)
+ .collect();
Review Comment:
Done in 1e61fa76c, with the stable sort as you suggest.
--
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]
