Repository navigation
pkg/executor: move window executors into windows subpackage - #67992
hawkingrei wants to merge 12 commits into
Conversation
|
Adding the "do-not-merge/release-note-label-needed" label because no release-note block was detected, please follow our release note process to remove it. DetailsInstructions for interacting with me using PR comments are available here. If you have questions or suggestions related to my behavior, please file an issue against the kubernetes-sigs/prow repository. |
ti-chi-bot
Bot
added
do-not-merge/release-note-label-needed
|
[APPROVALNOTIFIER] This PR is NOT APPROVED This pull-request has been approved by: The full list of commands accepted by this bot can be found here. DetailsNeeds approval from an approver in each of these files:Approvers can indicate their approval by writing |
📝 WalkthroughWalkthroughIntroduces Changes
Estimated code review effort🎯 4 (Complex) | ⏱️ ~60 minutes Possibly related PRs
Suggested labels
Suggested reviewers
Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches🧪 Generate unit tests (beta)
⚔️ Resolve merge conflicts
Warning There were issues while running some tools. Please review the errors and either fix the tool's configuration or disable the tool if it's a critical failure. 🔧 golangci-lint (2.11.4)Command failed Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
|
@hawkingrei: The following tests failed, say
Full PR test history. Your PR dashboard. DetailsInstructions for interacting with me using PR comments are available here. If you have questions or suggestions related to my behavior, please file an issue against the kubernetes-sigs/prow repository. I understand the commands that are listed here. |
|
@hawkingrei: The following tests failed, say
Full PR test history. Your PR dashboard. DetailsInstructions for interacting with me using PR comments are available here. If you have questions or suggestions related to my behavior, please file an issue against the kubernetes-sigs/prow repository. I understand the commands that are listed here. |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 5
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
pkg/executor/windows/pipelined_window.go (1)
176-180:⚠️ Potential issue | 🔴 CriticalDo not use
len(e.rows)to infer partition state after row dropping.Lines 397-405 can drop every retained row before the executor has observed the next partition boundary. In that state, Line 178 makes
newPartition=false, so the next partition can be merged into the previous partition’s window state. Use logical partition progress, not retained-buffer length.Suggested fix
func (e *PipelinedWindowExec) getRowsInPartition(ctx context.Context) (err error) { - e.newPartition = true - if len(e.rows) == 0 { - e.newPartition = false - } + e.newPartition = e.rowCnt > 0Also applies to: 397-405
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@pkg/executor/windows/pipelined_window.go` around lines 176 - 180, The code in getRowsInPartition incorrectly derives e.newPartition from len(e.rows) which breaks when retained rows are dropped; instead determine partition boundaries from logical partition progress: add or use a persistent partition identifier (e.g., e.lastPartitionKey or e.partitionSeen flag) and compare the incoming/current row's partition key (the same key used in the partitioning logic around lines 397-405) against that stored identifier to set e.newPartition; update that stored partition identifier whenever you observe a partition boundary or drop rows so partition state is driven by the partition key comparisons (in getRowsInPartition and the dropping logic) rather than the length of e.rows.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@pkg/executor/builder.go`:
- Around line 5474-5477: BuildStream's returned error isn't checked before using
streamWindowExec, which can be nil and cause a panic; modify the code around the
call to windowexec.BuildStream (the variables streamWindowExec and err) to
immediately check err (and/or nil-check streamWindowExec) and return the error
if non-nil before calling streamWindowExec.OpenSelf(), so OpenSelf is only
invoked on a valid executor.
In `@pkg/executor/windows/pipelined_window.go`:
- Around line 103-108: OpenSelf currently resets buffers but leaves the
executor's groupChecker state intact, which can cause stale split/exhaustion
state when the executor is reopened; update PipelinedWindowExec.OpenSelf to
reset or reinitialize the groupChecker (e.g., call a Reset/Init method or assign
a fresh groupChecker instance) so that any previous split/exhaustion flags are
cleared before reuse. Ensure you reference the groupChecker field on
PipelinedWindowExec and perform the reset alongside the other field
initializations in OpenSelf to prevent stale child-row behavior on subsequent
Open() calls.
In `@pkg/planner/core/casetest/windows/testdata/window_push_down_suite_in.json`:
- Line 21: The test query "select a, b, row_number() over(partition by a order
by b) as rn from stream_window_t" produces order-sensitive rows but lacks a
final ordering guarantee; update that SQL by appending " order by a, b" so the
test becomes deterministic, and change the corresponding generated expected SQL
in the same JSON testdata entry to include the same "order by a, b" clause; keep
the change minimal (only the query string and its expected SQL) so the test
output order is fixed.
In `@pkg/planner/core/casetest/windows/window_push_down_test.go`:
- Around line 40-44: The test currently classifies statements by lowercasing
stmt and only treating prefixes "select" and "explain" as queries, causing CTEs
starting with "with" to be mis-routed to tk.MustExec(tt); update the conditional
in window_push_down_test.go (the stmt variable and the branch that calls
tk.MustExec(tt)) to also treat statements starting with "with" as
query/assertion cases (or alternatively invoke the SQL parser to detect
query-type statements) so CTEs follow the query/plan/warning assertion path
instead of being executed-only.
In `@pkg/planner/core/task.go`:
- Around line 2152-2176: In attach2Task4PhysicalWindow ensure stream-window
CopTask rejection also covers index-merge plans: currently the isStreamWindow
branch only rejects when cop.IndexPlan != nil && cop.TablePlan != nil; extend
that guard to also reject when cop.IdxMergePartPlans (or the equivalent field
representing index-merge parts on physicalop.CopTask) is non-nil/non-empty so
the function (attach2Task4PhysicalWindow) returns base.InvalidTask for
index-merge CopTasks as well; keep the existing MPP TiFlash check (the p.StoreTp
dispatch) unchanged since PhysicalStreamWindow defaults to TiKV.
---
Outside diff comments:
In `@pkg/executor/windows/pipelined_window.go`:
- Around line 176-180: The code in getRowsInPartition incorrectly derives
e.newPartition from len(e.rows) which breaks when retained rows are dropped;
instead determine partition boundaries from logical partition progress: add or
use a persistent partition identifier (e.g., e.lastPartitionKey or
e.partitionSeen flag) and compare the incoming/current row's partition key (the
same key used in the partitioning logic around lines 397-405) against that
stored identifier to set e.newPartition; update that stored partition identifier
whenever you observe a partition boundary or drop rows so partition state is
driven by the partition key comparisons (in getRowsInPartition and the dropping
logic) rather than the length of e.rows.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository UI
Review profile: CHILL
Plan: Pro
Run ID: e2902cec-65f7-442f-bf75-52a60afa9a56
📒 Files selected for processing (26)
docs/agents/planner/README.mddocs/agents/planner/stream_window_notes.mdpkg/executor/BUILD.bazelpkg/executor/builder.gopkg/executor/join/test/indexjoin/BUILD.bazelpkg/executor/join/test/indexjoin/index_lookup_join_test.gopkg/executor/windows/BUILD.bazelpkg/executor/windows/builder.gopkg/executor/windows/partition_topn_window.gopkg/executor/windows/pipelined_window.gopkg/executor/windows/window.gopkg/executor/windows/window_test.gopkg/planner/core/casetest/rule/rule_outer2inner_test.gopkg/planner/core/casetest/windows/testdata/window_push_down_suite_in.jsonpkg/planner/core/casetest/windows/testdata/window_push_down_suite_out.jsonpkg/planner/core/casetest/windows/testdata/window_push_down_suite_xut.jsonpkg/planner/core/casetest/windows/window_push_down_test.gopkg/planner/core/exhaust_physical_plans.gopkg/planner/core/operator/physicalop/base_physical_plan.gopkg/planner/core/operator/physicalop/physical_window.gopkg/planner/core/optimizer.gopkg/planner/core/plan.gopkg/planner/core/stringer.gopkg/planner/core/task.gotests/integrationtest/r/planner/core/casetest/rule/rule_derive_topn_from_window.resulttests/realtikvtest/addindextest4/BUILD.bazel
| streamWindowExec, err := windowexec.BuildStream(builder.ctx, &v.PhysicalWindow, childExec) | ||
| err = streamWindowExec.OpenSelf() | ||
| if err != nil { | ||
| return nil, err |
There was a problem hiding this comment.
Check BuildStream’s error before calling OpenSelf().
If BuildStream returns an error with a nil executor, Line 5475 panics and masks the original build error.
Proposed fix
streamWindowExec, err := windowexec.BuildStream(builder.ctx, &v.PhysicalWindow, childExec)
+ if err != nil {
+ return nil, err
+ }
err = streamWindowExec.OpenSelf()
if err != nil {
return nil, err📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| streamWindowExec, err := windowexec.BuildStream(builder.ctx, &v.PhysicalWindow, childExec) | |
| err = streamWindowExec.OpenSelf() | |
| if err != nil { | |
| return nil, err | |
| streamWindowExec, err := windowexec.BuildStream(builder.ctx, &v.PhysicalWindow, childExec) | |
| if err != nil { | |
| return nil, err | |
| } | |
| err = streamWindowExec.OpenSelf() | |
| if err != nil { | |
| return nil, err | |
| } |
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@pkg/executor/builder.go` around lines 5474 - 5477, BuildStream's returned
error isn't checked before using streamWindowExec, which can be nil and cause a
panic; modify the code around the call to windowexec.BuildStream (the variables
streamWindowExec and err) to immediately check err (and/or nil-check
streamWindowExec) and return the error if non-nil before calling
streamWindowExec.OpenSelf(), so OpenSelf is only invoked on a valid executor.
| func (e *PipelinedWindowExec) OpenSelf() error { | ||
| e.done, e.newPartition, e.whole, e.initializedSlidingWindow = false, false, false, false | ||
| e.dataIdx, e.curRowIdx, e.dropped, e.rowToConsume, e.accumulated = 0, 0, 0, 0, 0 | ||
| e.lastStartRow, e.lastEndRow, e.stagedStartRow, e.stagedEndRow, e.rowStart, e.rowCnt = 0, 0, 0, 0, 0, 0 | ||
| e.rows, e.data = make([]chunk.Row, 0), make([]dataInfo, 0) | ||
| return e.BaseExecutor.Open(ctx) | ||
| return nil |
There was a problem hiding this comment.
Reset the group checker when reopening local state.
OpenSelf() clears executor-local buffers, but groupChecker still carries split/exhaustion state from the previous run. Reusing this executor via Open() or the IndexJoin OpenSelf() path can then skip fetching or split against stale child rows.
Suggested fix
func (e *PipelinedWindowExec) OpenSelf() error {
+ e.groupChecker.Reset()
+ e.childResult = nil
e.done, e.newPartition, e.whole, e.initializedSlidingWindow = false, false, false, false
e.dataIdx, e.curRowIdx, e.dropped, e.rowToConsume, e.accumulated = 0, 0, 0, 0, 0
e.lastStartRow, e.lastEndRow, e.stagedStartRow, e.stagedEndRow, e.rowStart, e.rowCnt = 0, 0, 0, 0, 0, 0📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| func (e *PipelinedWindowExec) OpenSelf() error { | |
| e.done, e.newPartition, e.whole, e.initializedSlidingWindow = false, false, false, false | |
| e.dataIdx, e.curRowIdx, e.dropped, e.rowToConsume, e.accumulated = 0, 0, 0, 0, 0 | |
| e.lastStartRow, e.lastEndRow, e.stagedStartRow, e.stagedEndRow, e.rowStart, e.rowCnt = 0, 0, 0, 0, 0, 0 | |
| e.rows, e.data = make([]chunk.Row, 0), make([]dataInfo, 0) | |
| return e.BaseExecutor.Open(ctx) | |
| return nil | |
| func (e *PipelinedWindowExec) OpenSelf() error { | |
| e.groupChecker.Reset() | |
| e.childResult = nil | |
| e.done, e.newPartition, e.whole, e.initializedSlidingWindow = false, false, false, false | |
| e.dataIdx, e.curRowIdx, e.dropped, e.rowToConsume, e.accumulated = 0, 0, 0, 0, 0 | |
| e.lastStartRow, e.lastEndRow, e.stagedStartRow, e.stagedEndRow, e.rowStart, e.rowCnt = 0, 0, 0, 0, 0, 0 | |
| e.rows, e.data = make([]chunk.Row, 0), make([]dataInfo, 0) | |
| return nil |
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@pkg/executor/windows/pipelined_window.go` around lines 103 - 108, OpenSelf
currently resets buffers but leaves the executor's groupChecker state intact,
which can cause stale split/exhaustion state when the executor is reopened;
update PipelinedWindowExec.OpenSelf to reset or reinitialize the groupChecker
(e.g., call a Reset/Init method or assign a fresh groupChecker instance) so that
any previous split/exhaustion flags are cleared before reuse. Ensure you
reference the groupChecker field on PipelinedWindowExec and perform the reset
alongside the other field initializations in OpenSelf to prevent stale child-row
behavior on subsequent Open() calls.
| "insert into stream_window_t values (1, 1, 11), (1, 2, 12), (2, 1, 21), (2, 3, 23)", | ||
| "explain format = 'plan_tree' select a, row_number() over(partition by a) as rn from stream_window_t", | ||
| "explain format = 'plan_tree' select a, b, row_number() over(partition by a order by b) as rn from stream_window_t", | ||
| "select a, b, row_number() over(partition by a order by b) as rn from stream_window_t", |
There was a problem hiding this comment.
Add an outer ORDER BY to make the result assertion deterministic.
The expected rows are order-sensitive, but the query has no final ordering guarantee. Add order by a, b and update the generated expected SQL accordingly. As per coding guidelines, “Keep test changes minimal and deterministic; avoid broad golden/testdata churn unless required.”
🧪 Proposed fix
- "select a, b, row_number() over(partition by a order by b) as rn from stream_window_t",
+ "select a, b, row_number() over(partition by a order by b) as rn from stream_window_t order by a, b",🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@pkg/planner/core/casetest/windows/testdata/window_push_down_suite_in.json` at
line 21, The test query "select a, b, row_number() over(partition by a order by
b) as rn from stream_window_t" produces order-sensitive rows but lacks a final
ordering guarantee; update that SQL by appending " order by a, b" so the test
becomes deterministic, and change the corresponding generated expected SQL in
the same JSON testdata entry to include the same "order by a, b" clause; keep
the change minimal (only the query string and its expected SQL) so the test
output order is fixed.
| stmt := strings.ToLower(strings.TrimSpace(tt)) | ||
| if !strings.HasPrefix(stmt, "select") && !strings.HasPrefix(stmt, "explain") { | ||
| tk.MustExec(tt) | ||
| continue | ||
| } |
There was a problem hiding this comment.
Treat CTEs as query statements here.
This gate sends WITH ... SELECT cases through tk.MustExec and skips result/plan/warning assertions. The same window suite includes CTE SELECT inputs, so include with in the query branch or classify statements via the parser.
🐛 Proposed fix
stmt := strings.ToLower(strings.TrimSpace(tt))
- if !strings.HasPrefix(stmt, "select") && !strings.HasPrefix(stmt, "explain") {
+ if !strings.HasPrefix(stmt, "select") &&
+ !strings.HasPrefix(stmt, "explain") &&
+ !strings.HasPrefix(stmt, "with") {
tk.MustExec(tt)
continue
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| stmt := strings.ToLower(strings.TrimSpace(tt)) | |
| if !strings.HasPrefix(stmt, "select") && !strings.HasPrefix(stmt, "explain") { | |
| tk.MustExec(tt) | |
| continue | |
| } | |
| stmt := strings.ToLower(strings.TrimSpace(tt)) | |
| if !strings.HasPrefix(stmt, "select") && | |
| !strings.HasPrefix(stmt, "explain") && | |
| !strings.HasPrefix(stmt, "with") { | |
| tk.MustExec(tt) | |
| continue | |
| } |
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@pkg/planner/core/casetest/windows/window_push_down_test.go` around lines 40 -
44, The test currently classifies statements by lowercasing stmt and only
treating prefixes "select" and "explain" as queries, causing CTEs starting with
"with" to be mis-routed to tk.MustExec(tt); update the conditional in
window_push_down_test.go (the stmt variable and the branch that calls
tk.MustExec(tt)) to also treat statements starting with "with" as
query/assertion cases (or alternatively invoke the SQL parser to detect
query-type statements) so CTEs follow the query/plan/warning assertion path
instead of being executed-only.
| func attach2Task4PhysicalWindow(pp base.PhysicalPlan, tasks ...base.Task) base.Task { | ||
| p := pp.(*physicalop.PhysicalWindow) | ||
| var ( | ||
| p *physicalop.PhysicalWindow | ||
| isStreamWindow bool | ||
| ) | ||
| switch v := pp.(type) { | ||
| case *physicalop.PhysicalWindow: | ||
| p = v | ||
| case *physicalop.PhysicalStreamWindow: | ||
| p = &v.PhysicalWindow | ||
| isStreamWindow = true | ||
| default: | ||
| return base.InvalidTask | ||
| } | ||
| if mpp, ok := tasks[0].Copy().(*physicalop.MppTask); ok && p.StoreTp == kv.TiFlash { | ||
| return attach2TaskForMPP4PhysicalWindow(p, mpp) | ||
| } | ||
| if isStreamWindow { | ||
| if cop, ok := tasks[0].(*physicalop.CopTask); ok && cop.IndexPlan != nil && cop.TablePlan != nil { | ||
| return base.InvalidTask | ||
| } | ||
| } | ||
| t := tasks[0].ConvertToRootTask(p.SCtx()) | ||
| return attachPlan2Task(p.Self, t) | ||
| } |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Check how PhysicalStreamWindow is constructed and whether StoreTp is ever set to TiFlash.
ast-grep --pattern 'PhysicalStreamWindow{$$$}'
rg -nP -C3 '\bPhysicalStreamWindow\b'
# Check if any planner code admits StreamWindow over index-merge cop tasks.
rg -nP -C3 'IdxMergePartPlans' -g '**/exhaust_physical_plans*.go' -g '**/task.go'Repository: pingcap/tidb
Length of output: 18499
🏁 Script executed:
# Check how StoreTp is initialized on PhysicalWindow and whether PhysicalStreamWindow gets it
rg -nP 'StoreTp\s*=' pkg/planner/core/operator/physicalop/physical_window.go | head -20
rg -nP 'StoreTp.*TiFlash' pkg/planner/core/operator/physicalop/
# Check where stream windows are enumerated in exhaustPhysicalPlans
rg -nP -B5 -A5 'PhysicalStreamWindow' pkg/planner/core/exhaust_physical_plans.go | head -60
# Check if there are guards against stream window with index merge in exhaust plans
rg -nP -C3 'CanUseStreamWindow' pkg/planner/core/operator/physicalop/physical_window.go | head -40Repository: pingcap/tidb
Length of output: 883
🏁 Script executed:
# Check the Clone method for PhysicalStreamWindow to see if StoreTp is preserved
rg -nP -A10 'func \(p \*PhysicalStreamWindow\) Clone' pkg/planner/core/operator/physicalop/physical_window.go
# Check what CanUseStreamWindow checks - does it block index-merge scenarios?
rg -nP -A20 'func CanUseStreamWindow' pkg/planner/core/operator/physicalop/physical_window.go
# Check if stream windows are enumerated over index-merge cop tasks
rg -nP -B5 -A10 'idxMerge.*StreamWindow|StreamWindow.*idxMerge' pkg/planner/core/exhaust_physical_plans.goRepository: pingcap/tidb
Length of output: 1509
🏁 Script executed:
# Check where StoreTp is set to TiFlash on PhysicalWindow
rg -nP -B3 -A3 'StoreTp.*=.*kv\.TiFlash' pkg/planner/core/operator/physicalop/physical_window.go
# Check the exhaustPhysicalPlans logic for windows - does it enumerate stream windows?
rg -nP -B5 -A15 'func.*exhaust.*Window' pkg/planner/core/exhaust_physical_plans.go
# Check if stream windows can be enumerated when there's an index-merge plan below
rg -nP 'enumPhysicalWindowAgg|enumWindow' pkg/planner/core/exhaust_physical_plans.goRepository: pingcap/tidb
Length of output: 38
🏁 Script executed:
# Check all places where StoreTp is set to TiFlash on window operators
rg -nP 'StoreTp' pkg/planner/core/operator/physicalop/physical_window.go | head -20
# Check the whole exhaustPhysicalPlans file to understand window enumeration
head -100 pkg/planner/core/exhaust_physical_plans.go | tail -50
# Search for where the stream window is actually created - context around line 525
sed -n '475,545p' pkg/planner/core/operator/physicalop/physical_window.goRepository: pingcap/tidb
Length of output: 5759
🏁 Script executed:
# Check the full PhysicalStreamWindow initialization - does StoreTp ever get set?
sed -n '520,555p' pkg/planner/core/operator/physicalop/physical_window.go
# Check what the zero-value of StoreTp is (kv.TiKV or kv.TiFlash?)
rg -nP 'const.*TiKV|const.*TiFlash' | head -20
# Look for any assignment to StoreTp on stream window after it's created
rg -nP -C5 'streamWindow.*StoreTp|StoreTp.*streamWindow' pkg/planner/core/operator/physicalop/
# Check if PhysicalWindow or PhysicalStreamWindow has any default initialization
rg -nP -B3 -A3 'NewPhysicalWindow|NewPhysicalStreamWindow' pkg/planner/core/operator/physicalop/physical_window.goRepository: pingcap/tidb
Length of output: 6256
🏁 Script executed:
# Check if stream windows are ever created when there's an index-merge cop below
# Look at the LogicalWindow and how it gets its child property from a CopTask
rg -nP -B5 -A10 'admitIndexJoinProp' pkg/planner/core/operator/physicalop/physical_window.go
# Check if there's any code that creates a stream window from a CopTask with index merge
rg -nP 'attach2Task4PhysicalWindow' pkg/planner/core/task.go | head -5
sed -n '2152,2180p' pkg/planner/core/task.go
# Check what conditions lead to CopTask with IdxMergePartPlans
rg -nP -B3 'IdxMergePartPlans.*=' pkg/planner/core/task.go | head -30Repository: pingcap/tidb
Length of output: 4059
🏁 Script executed:
# Check if Clone on PhysicalWindow preserves StoreTp and whether stream windows can inherit it
rg -nP -B3 -A8 'func \(p \*PhysicalWindow\) Clone' pkg/planner/core/operator/physicalop/physical_window.go
# Check if there's any code path where a PhysicalStreamWindow gets StoreTp set to TiFlash after creation
rg -nP 'streamWindow.*StoreTp|\.StoreTp\s*=' pkg/planner/core/operator/physicalop/physical_window.go
# Look at the actual constructor to understand what the zero value of StoreTp would be
rg -nP 'type.*StoreType|const.*StoreType' pkg/kv/type.go
# Check where stream windows actually get used - in indexjoin scenarios
rg -nP -B5 -A10 'admitIndexJoinInnerChildPattern' pkg/planner/core/exhaust_physical_plans.go | head -60Repository: pingcap/tidb
Length of output: 1252
🏁 Script executed:
# Check if LogicalWindow is admitted in index join inner patterns
sed -n '596,650p' pkg/planner/core/exhaust_physical_plans.go
# Check what kv.StoreType is and its zero value
rg -nP 'type StoreType|const.*iota' pkg/kv/ | head -20
# Check if stream windows can actually be created for index join inner side
rg -nP 'LogicalWindow.*index|indexJoin.*window' pkg/planner/core/exhaust_physical_plans.goRepository: pingcap/tidb
Length of output: 2225
🏁 Script executed:
# Check if there are any tests showing stream windows with index merge
rg -nP 'stream.*window|window.*stream' pkg/planner/core/casetest/ -i | grep -i 'indexmerge\|idx.*merge' | head -20
# Look for tests with stream windows and CopTasks
rg -nP 'PhysicalStreamWindow' pkg/planner/core/ -g '**/*test.go' | head -20
# Search for any actual usage patterns of stream windows with cop tasks
rg -nP -B5 -A5 'CopTask.*StreamWindow|StreamWindow.*CopTask' pkg/planner/core/ -g '**/*.go'Repository: pingcap/tidb
Length of output: 38
🏁 Script executed:
# Check if stream windows are ever created in MPP-enabled scenarios
# Look for where tryToGetMppWindows creates plans and whether they can become stream windows
sed -n '475,510p' pkg/planner/core/operator/physicalop/physical_window.go
# Check the actual value of kv.TiKV vs kv.TiFlash
rg -nP 'const.*TiKV|const.*TiFlash' pkg/kv/kv.go
# Look for tests that verify stream windows with cop tasks
fd '.*test\.go' -x grep -l 'StreamWindow' {} \; 2>/dev/null | head -10Repository: pingcap/tidb
Length of output: 1533
🏁 Script executed:
# Look at the actual StoreType constants definition
sed -n '389,410p' pkg/kv/kv.go
# Check if there are any existing tests that show stream windows with index merge cop
rg -nP -B10 -A10 'attach2Task4PhysicalWindow' pkg/planner/core/ -g '**/*test.go' | head -40
# Verify if PhysicalStreamWindow ever gets StoreTp assigned after creation
rg -nP 'streamWindow\.' pkg/planner/core/operator/physicalop/physical_window.go | grep -E 'StoreTp|SetStoreTp'Repository: pingcap/tidb
Length of output: 668
The first concern about PhysicalStreamWindow carrying StoreTp == kv.TiFlash is not a real issue.
PhysicalStreamWindow is created without explicitly setting StoreTp (see physical_window.go:525–534), so it defaults to kv.TiKV (the zero value of StoreType). Only the regular PhysicalWindow variant used in the MPP path explicitly sets StoreTp: kv.TiFlash (line 488). The dispatch on line 2166 will never reach attach2TaskForMPP4PhysicalWindow with a PhysicalStreamWindow because p.StoreTp will be kv.TiKV, not kv.TiFlash.
The second concern about the index-merge guard is valid but minor in practice. The code rejects stream windows only when a CopTask has both IndexPlan and TablePlan (a double-read), but does not reject index-merge CopTasks with IdxMergePartPlans. Index-merge is equally incompatible with stream windows for similar ordering reasons. However, in practice, this gap is unlikely to cause issues because admitIndexJoinInnerChildPattern (exhaust_physical_plans.go:625–629) only admits stream windows in index-join inner contexts, where the CopTask child is typically a simple DataSource + Selection/Projection, making index-merge construction within an index-join inner side extremely rare.
A defensive improvement to tighten the guard remains reasonable, but the current code's practical safety is high due to the constrained admission context.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@pkg/planner/core/task.go` around lines 2152 - 2176, In
attach2Task4PhysicalWindow ensure stream-window CopTask rejection also covers
index-merge plans: currently the isStreamWindow branch only rejects when
cop.IndexPlan != nil && cop.TablePlan != nil; extend that guard to also reject
when cop.IdxMergePartPlans (or the equivalent field representing index-merge
parts on physicalop.CopTask) is non-nil/non-empty so the function
(attach2Task4PhysicalWindow) returns base.InvalidTask for index-merge CopTasks
as well; keep the existing MPP TiFlash check (the p.StoreTp dispatch) unchanged
since PhysicalStreamWindow defaults to TiKV.
What problem does this PR solve?
Issue Number: ref #67989
Problem Summary:
#67696 mixes executor-side window implementation changes with planner work. For the first follow-up slice, we want to submit the executor part independently and make the window executor code easier to extend and test.
What changed and how does it work?
Check List
Tests
Test commands:
Local note:
fails in this environment because Gazelle hits the existing local replacement issue:
Side effects
Documentation
Release note
None
Summary by CodeRabbit
New Features
StreamWindowexecution mode for window functions, enabling more efficient ordered-input processing forrow_number()operations.Documentation
StreamWindowplanner and executor integration, scope constraints, and optimization patterns.Tests
IndexJoinwithStreamWindowinner side and window executor metadata correctness.