jayzhan211 commented on code in PR #26669:
URL: https://github.com/apache/datafusion/pull/26669#discussion_r4367010391
##########
datafusion/sqllogictest/test_files/cte.slt:
##########
@@ -1727,3 +1727,107 @@ order by 1, 2;
statement ok
drop table test;
+
+# Dynamic filters produced by a recursive term must be reset between
+# iterations. Otherwise the second execution of this uncorrelated aggregate
+# reuses the first iteration's MAX bound and scans no rows.
+statement ok
+COPY (SELECT CAST(value AS DOUBLE) AS r FROM generate_series(0, 128)) TO
'test_files/scratch/cte/dynamic_filter_recursive.parquet' STORED AS PARQUET;
+
+statement ok
+CREATE EXTERNAL TABLE dynamic_filter_recursive STORED AS PARQUET LOCATION
'test_files/scratch/cte/dynamic_filter_recursive.parquet';
+
+query IR rowsort
+WITH RECURSIVE w(n, m) AS (
+ SELECT 0, CAST(NULL AS DOUBLE)
+ UNION ALL
+ SELECT w.n + 1, x.m
+ FROM w CROSS JOIN (
+ SELECT MAX(r) AS m
+ FROM dynamic_filter_recursive
+ WHERE r % 2 = 0
+ ) x
+ WHERE w.n < 3
+)
+SELECT n, m FROM w;
+----
+0 NULL
+1 128
+2 128
+3 128
+
+statement ok
+DROP TABLE dynamic_filter_recursive;
+
+# Hash join dynamic filters must remain shared with the probe scan across
+# recursive iterations (#26641). Each iteration looks up a different key, so
+# retaining a stale bound would also truncate the recursive result.
Review Comment:
nit: I don't think the second sentence holds. When I removed
`dynamic_filter.filter.reset()?` from `reset_state`, this result query still
passed. Each iteration's new accumulator publishes fresh bounds before the
probe scan opens, so only the `EXPLAIN ANALYZE` block below catches the missing
reset. Maybe reword it so the comment says which query covers what:
```diff
# Hash join dynamic filters must remain shared with the probe scan across
-# recursive iterations (#26641). Each iteration looks up a different key, so
-# retaining a stale bound would also truncate the recursive result.
+# recursive iterations (#26641). Each iteration looks up a different key;
the
+# EXPLAIN ANALYZE below checks that the filter is reset and republished, so
the
+# probe scan prunes rows on every iteration.
```
##########
datafusion/physical-plan/src/joins/hash_join/exec.rs:
##########
@@ -566,11 +566,17 @@ impl HashJoinExecBuilder {
}
/// Reset task-local runtime state while retaining the immutable prepared
build.
- pub fn reset_state(mut self) -> Self {
+ pub fn reset_state(mut self) -> Result<Self> {
Review Comment:
`HashJoinExecBuilder::reset_state` has been public since 53.0.0, so changing
it to return `Result<Self>` is a breaking change, and it isn't declared yet (no
`api change` label or upgrade-guide entry). I think we can avoid the break: the
only fallible step is remapping `lit(true)` inside
`DynamicFilterPhysicalExpr::reset()`, and `lit(true)` has no column references,
so the remap never changes anything. Since #26056 hasn't merged yet, `reset()`
could be infallible there, and then this method keeps returning `Self` (I
compile-checked this and the reset tests pass):
```diff
- pub fn reset(&self) -> Result<()> {
- let new_expr = Self::remap_children(
- &self.children,
- self.remapped_children.as_ref(),
- lit(true),
- )?;
-
+ pub fn reset(&self) {
+ // `true` has no column references, so no child remapping is needed.
let mut current = self.inner.write();
let new_generation = current.generation + 1;
*current = Inner {
expression_id: current.expression_id,
generation: new_generation,
- expr: new_expr,
+ expr: lit(true),
is_complete: false,
};
drop(current);
let _ = self.state_watch.send(FilterState::InProgress {
generation: new_generation,
});
- Ok(())
}
```
```diff
- pub fn reset_state(mut self) -> Result<Self> {
+ pub fn reset_state(mut self) -> Self {
self.exec.left_fut = Default::default();
if let Some(dynamic_filter) = &mut self.exec.dynamic_filter {
- dynamic_filter.filter.reset()?;
+ dynamic_filter.filter.reset();
```
The `?` / `.unwrap()` at the other `reset()` and `reset_state()` call sites
go away too. If you'd rather keep `Result`, could you add the `api change`
label and a note in `docs/source/library-user-guide/upgrading/56.0.0.md`?
--
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]
