sunchao commented on code in PR #26214:
URL: https://github.com/apache/datafusion/pull/26214#discussion_r4429691135
##########
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:
**[P2] Fill chunks with unmarked build rows.** This chunks the original
indices before removing rows already marked. If one live row survives in each
chunk, each survivor can be paired with an entire probe batch before its first
passing match is recorded.
I reproduced this in a `LeftMark` query with a residual predicate that
always returns true and counts the rows evaluated. Of 1,048,576 build rows, the
first probe batch marked all but one per 8,192. For the next 8,192-row probe
batch, the merge base evaluated 8,192 candidate pairs in one predicate call.
This head evaluated 1,048,576 pairs in 128 calls. The physical plan and SQL
result matched.
In a pairing control, compacting the unmarked indices across input-chunk
boundaries before pairing restored the single 8,192-pair call. The largest
individual temporary allocation stayed at 64 KiB.
##########
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:
**[P2] Avoid materializing every probe index before checking the build
marks.** A `StreamingTable` can hand the join a record batch larger than the
configured batch size. I ran a single-partition `LeftMark` with one build row
`(NULL, 1)`, an 8,192-row target, and a 256 KiB memory pool. The first probe
batch `[(1, 1)]` marks the only build row. The next batch contains 8,388,608
copies of `(1, 1)`.
An allocator trace captured a 32 MiB `Vec<u32>` here on this head. The same
SQL and source on the merge base produced the same NULL result without that
allocation. No candidate pair from the second batch is needed. Could the caller
supply a replayable probe iterator, recreated from the range or null bitmap for
each surviving build chunk? That would bound this extra memory for streaming
inputs too.
##########
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:
**[P1] Keep `DataType::Null` among the value keys too.** A nonempty Arrow
`NullArray` has a physical null count of zero even though every value is
logically NULL. `RecordBatch::try_from_iter` therefore infers a nonnullable
field, and DataFusion accepts `Null` as a hash key.
I used the same fixture shape as the new dictionary test, with outer `a =
[1, NULL], b = NullArray(2)` and inner `x = [1], y = NullArray(1)`. On this
head, `(a,b) NOT IN (SELECT x,y FROM i)` returns `TRUE` for both rows, so both
pass the `WHERE` filter. The merge base and the three-valued SQL reference
return `NULL` for both and the filter returns no rows. On this same head,
changing the subquery to `SELECT x, NULL FROM i` preserves the inner values,
keeps both value keys, and restores the expected result. Could the type guard
and regression test cover `DataType::Null`?
--
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]
