alamb commented on issue #27130:
URL: https://github.com/apache/datafusion/issues/27130#issuecomment-6660069182
## Reproducer
I could not find a way to reproduce this with SQL in `datafusion-cli`: each
statement plans a fresh `SortExec`, so two executions never share a dynamic
filter. The only SQL path that calls `reset_state` is a recursive CTE, and its
iterations run one after another (the case #26670 fixed). The Lance scenario is
a cached physical plan that is reset and executed by two queries at the same
time, so here is a Rust test that models that.
Add the following to the `tests` module at the end of
`datafusion/physical-plan/src/sorts/sort.rs` (it compiles unchanged on both
`55.0.0` and `main`):
```rust
/// Regression test for
<https://github.com/apache/datafusion/issues/27130>
///
/// Plans derived from the same TopK plan via `reset_state` must not
share
/// dynamic filter state. This models a cached `LIMIT 1` plan that is
reset
/// and executed by two queries at the same time: both copies are created
/// before either executes, and the first one to finish must not cause
the
/// other to return an empty result.
#[tokio::test]
async fn test_topk_reset_state_copies_are_independent() -> Result<()> {
let schema = Arc::new(Schema::new(vec![Field::new("a",
DataType::Int32, false)]));
let batch = RecordBatch::try_new(
Arc::clone(&schema),
vec![Arc::new(Int32Array::from(vec![3, 1, 2]))],
)?;
let input =
TestMemoryExec::try_new_exec(&[vec![batch]],
Arc::clone(&schema), None)?;
let sort: Arc<dyn ExecutionPlan> = Arc::new(
SortExec::new(
[PhysicalSortExpr::new_default(Arc::new(Column::new("a",
0)))].into(),
input,
)
.with_fetch(Some(1)),
);
// Both executions are prepared before either runs
let first = Arc::clone(&sort).reset_state()?;
let second = Arc::clone(&sort).reset_state()?;
let task_ctx = Arc::new(TaskContext::default());
let first_result = collect(first, Arc::clone(&task_ctx)).await?;
let second_result = collect(second, task_ctx).await?;
assert_snapshot!(batches_to_string(&first_result), @r"
+---+
| a |
+---+
| 1 |
+---+
");
assert_snapshot!(batches_to_string(&second_result), @r"
+---+
| a |
+---+
| 1 |
+---+
");
Ok(())
}
```
Run with
```shell
cargo test -p datafusion-physical-plan --lib
test_topk_reset_state_copies_are_independent
```
### Result on `55.0.0`
```
test sorts::sort::tests::test_topk_reset_state_copies_are_independent ... ok
```
### Result on `main` (70f2ed01b8)
The second execution returns no rows:
```
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ Snapshot Summary
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
Snapshot: topk_reset_state_copies_are_independent-2
Source: datafusion/physical-plan/src/sorts/sort.rs:4807
────────────────────────────────────────────────────────────────────────────────
Expression: batches_to_string(&second_result)
────────────────────────────────────────────────────────────────────────────────
-old snapshot
+new results
────────────┬───────────────────────────────────────────────────────────────────
1 │-+---+
2 │-| a |
3 │-+---+
4 │-| 1 |
5 │-+---+
1 │+++
2 │+++
────────────┴───────────────────────────────────────────────────────────────────
```
<details>
<summary>Root cause (speculation)</summary>
Before #26670, `SortExec::reset_state` gave the new `SortExec` a brand new
dynamic filter. Now it calls `reset()` on the existing
`DynamicFilterPhysicalExpr` and only rebuilds the `TopKDynamicFilters` wrapper,
so every reset copy (and the original) shares one `DynamicFilterPhysicalExpr`.
When the first copy finishes a `LIMIT 1` it publishes `a < 1` into that
shared expression. The second copy's `TopK` evaluates the shared expression
against its own input in `TopK::insert_batch` (the TopK pre-filters rows
itself, so no scan / filter pushdown is needed to hit this), every row is
dropped, and it emits nothing.
Note that the doc comment on `reset_plan_states` on `main` already says a
plan that uses dynamic filters may not be reset while the original or another
reset copy is still executing, so this is currently documented as a limitation.
It is still a behavior change from 55 though.
</details>
--
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]
