Repository navigation
pkg/planner, pkg/executor: add stream window demo path - #67696
hawkingrei wants to merge 12 commits into
Conversation
|
Skipping CI for Draft Pull Request. |
ti-chi-bot
Bot
added
do-not-merge/needs-triage-completed
do-not-merge/needs-linked-issue
do-not-merge/work-in-progress
|
Skipping CI for Draft Pull Request. |
|
Note Reviews pausedIt looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the Use the following commands to manage reviews:
Use the checkboxes below for quick actions:
📝 WalkthroughWalkthroughAdds a stream-window execution path: planner/operator support for Changes
Sequence Diagram(s)sequenceDiagram
participant Planner
participant Exhaustor
participant Builder
participant IndexJoinBuilder
participant Executor
participant ChildExec
Planner->>Exhaustor: ExhaustPhysicalPlans4LogicalWindow
Exhaustor->>Planner: CanUseStreamWindow(row_number) -> PhysicalStreamWindow
Planner-->>Builder: PhysicalStreamWindow plan
Builder->>Builder: buildStreamWindow() (force pipelined)
Builder->>Executor: build child PipelinedWindowExec
Builder->>Executor: wrap -> StreamWindowExec
Note over IndexJoinBuilder: index-join inner path may detect rn upper bound
IndexJoinBuilder->>Builder: build child executor
IndexJoinBuilder->>IndexJoinBuilder: detect rn upper bound -> build PartitionTopNWindowExec
Executor->>ChildExec: Open child
Executor->>Executor: StreamWindowExec.Open() -> PipelinedWindowExec.OpenSelf()
Executor->>Executor: Next(): emit ordered rows with computed rn / PartitionTopN emits top-K per partition
sequenceDiagram
participant PartitionTopNExec
participant VecGroupChecker
participant ChildExec
participant OutputChunk
PartitionTopNExec->>PartitionTopNExec: Open()/OpenSelf(): init counters
loop per Next()
PartitionTopNExec->>PartitionTopNExec: if no current group -> loadNextGroup()
PartitionTopNExec->>ChildExec: fetch child batch
ChildExec-->>PartitionTopNExec: return rows
PartitionTopNExec->>VecGroupChecker: split into groups
alt group available
PartitionTopNExec->>PartitionTopNExec: emit up to K rows for group (append rn)
PartitionTopNExec-->>OutputChunk: return chunk
else exhausted
PartitionTopNExec-->>OutputChunk: return empty/end
end
end
Estimated code review effort🎯 4 (Complex) | ⏱️ ~45 minutes Possibly related PRs
Suggested labels
Suggested reviewers
Poem
🚥 Pre-merge checks | ✅ 2 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (2 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
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 |
ti-chi-bot
Bot
added
size/L
ti-chi-bot
Bot
added
size/XL
|
[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 |
ti-chi-bot
Bot
added
size/XXL
ti-chi-bot
Bot
removed
the
do-not-merge/work-in-progress
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 4
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
pkg/executor/builder.go (1)
5769-5776:⚠️ Potential issue | 🔴 CriticalPropagate
UpdateCompareColsfailures intob.err.These branches currently
return nilonUpdateCompareColserrors without assigningb.err. That can leave the builder with a nil executor and no recorded failure, which then turns into a later, less actionable error path.Suggested fix
err = exec.start.UpdateCompareCols(b.ctx, exec.orderByCols) if err != nil { + b.err = err return nil } err = exec.end.UpdateCompareCols(b.ctx, exec.orderByCols) if err != nil { + b.err = err return nil } @@ err = tmpProcessor.start.UpdateCompareCols(b.ctx, orderByCols) if err != nil { + b.err = err return nil } err = tmpProcessor.end.UpdateCompareCols(b.ctx, orderByCols) if err != nil { + b.err = err return nil }Also applies to: 5808-5815
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@pkg/executor/builder.go` around lines 5769 - 5776, The code currently returns nil when exec.start.UpdateCompareCols or exec.end.UpdateCompareCols fails, but does not set b.err; update these error branches to assign the error to the builder error field (b.err = err) before returning nil so failures propagate; apply the same fix for both occurrences (the block calling exec.start.UpdateCompareCols and exec.end.UpdateCompareCols and the similar block at the second location referenced) so b.err holds the failure instead of silently dropping it.
🧹 Nitpick comments (1)
pkg/planner/core/operator/physicalop/physical_window.go (1)
551-556: Add a doc comment forCanUseStreamWindow.This is a new exported helper, so it should carry a doc comment describing the semantic gate, not just restating the name.
As per coding guidelines, "Keep exported-symbol doc comments, and prefer semantic constraints over name restatement."
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@pkg/planner/core/operator/physicalop/physical_window.go` around lines 551 - 556, Add a descriptive doc comment immediately above the exported function CanUseStreamWindow explaining the semantic gate it represents: that it returns true only when the LogicalWindow contains exactly one window function and that function is ROW_NUMBER (ast.WindowFuncRowNumber), meaning the logical window can be implemented as a streaming window rather than a blocking one; avoid restating the function name and instead describe the constraint and the implication for planner/physical execution.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@docs/agents/planner/stream_window_notes.md`:
- Around line 83-85: The duplicated validation command lines containing "go test
./pkg/planner/core/casetest/windows -run TestWindowPushDownPlans -record
--tags=intest" and "go test ./pkg/planner/core/casetest/windows -run
TestWindowPushDownPlans --tags=intest" should be de-duplicated and the header
clarified; either remove the second occurrence (lines 84-85) if they repeat the
earlier listing, or change the header "Validation commands used during the
demo:" to specify which demo/step these commands apply to (e.g., "Validation
commands for demo 1" or "Validation commands for step 2"), so the purpose and
scope of the commands are unambiguous.
- Around line 80-81: Replace the non-standard `--tags=intest` flag with the
standard Go test short form `-tags=intest` in the two command lines shown (the
lines containing "go test ./pkg/executor/join/test/indexjoin -run
TestIndexJoinWithStreamWindowInner --tags=intest" and "go test
./pkg/planner/core/casetest/windows -run TestWindowPushDownPlans --tags=intest")
so they match other docs entries that use `-tags=intest` (or
`-tags=intest,deadlock`).
In `@pkg/executor/partition_topn_window.go`:
- Around line 53-61: OpenSelf currently resets executor cursors but leaves the
VecGroupChecker state, which can cause stale offsets in loadNextGroup() when
childResult is nil; update PartitionTopNWindowExec.OpenSelf to also reset the
groupChecker (e.g., set e.groupChecker to its zero value or call its Reset/Init
method) so any state from a prior run is cleared when reopening the executor.
In `@pkg/planner/core/casetest/windows/window_push_down_test.go`:
- Around line 40-42: The test misclassifies CTEs because the stmt detection only
checks for "select" and "explain" prefixes; update the branch that decides to
call tk.MustExec(tt) so statements beginning with "with" (case-insensitive,
including "with recursive") are treated as queries too. Concretely, modify the
condition that uses stmt := strings.ToLower(strings.TrimSpace(tt)) and the
subsequent if that currently checks strings.HasPrefix(stmt, "select") and
strings.HasPrefix(stmt, "explain") to also consider strings.HasPrefix(stmt,
"with") (and/or "with recursive") so CTEs follow the query/plan/warning
assertion path rather than hitting tk.MustExec(tt).
---
Outside diff comments:
In `@pkg/executor/builder.go`:
- Around line 5769-5776: The code currently returns nil when
exec.start.UpdateCompareCols or exec.end.UpdateCompareCols fails, but does not
set b.err; update these error branches to assign the error to the builder error
field (b.err = err) before returning nil so failures propagate; apply the same
fix for both occurrences (the block calling exec.start.UpdateCompareCols and
exec.end.UpdateCompareCols and the similar block at the second location
referenced) so b.err holds the failure instead of silently dropping it.
---
Nitpick comments:
In `@pkg/planner/core/operator/physicalop/physical_window.go`:
- Around line 551-556: Add a descriptive doc comment immediately above the
exported function CanUseStreamWindow explaining the semantic gate it represents:
that it returns true only when the LogicalWindow contains exactly one window
function and that function is ROW_NUMBER (ast.WindowFuncRowNumber), meaning the
logical window can be implemented as a streaming window rather than a blocking
one; avoid restating the function name and instead describe the constraint and
the implication for planner/physical execution.
🪄 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: b9b2604b-fde4-4951-a5b3-e4948f0d8101
📒 Files selected for processing (20)
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/partition_topn_window.gopkg/executor/pipelined_window.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/realtikvtest/addindextest4/BUILD.bazel
| - `go test ./pkg/executor/join/test/indexjoin -run TestIndexJoinWithStreamWindowInner --tags=intest` | ||
| - `go test ./pkg/planner/core/casetest/windows -run TestWindowPushDownPlans --tags=intest` |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Description: Verify that test paths and test names referenced in the documentation exist
# Check if the test directories exist
fd --type d "indexjoin" pkg/executor/join/test/
fd --type d "windows" pkg/planner/core/casetest/
# Search for the test functions referenced in the commands
rg -l "TestIndexJoinWithStreamWindowInner" --type go pkg/executor/join/test/
rg -l "TestWindowPushDownPlans" --type go pkg/planner/core/casetest/Repository: pingcap/tidb
Length of output: 242
🏁 Script executed:
#!/bin/bash
# Verify exact test function names (case-sensitive)
rg "func TestIndexJoinWithStreamWindowInner" pkg/executor/join/test/indexjoin/ --type go
rg "func TestWindowPushDownPlans" pkg/planner/core/casetest/windows/ --type go
# Check for build tag usage in these test files
rg "//\+build|//go:build" pkg/executor/join/test/indexjoin/index_lookup_join_test.go
rg "//\+build|//go:build" pkg/planner/core/casetest/windows/window_push_down_test.goRepository: pingcap/tidb
Length of output: 274
🏁 Script executed:
#!/bin/bash
# Check first 10 lines of each test file for build tags
head -n 10 pkg/executor/join/test/indexjoin/index_lookup_join_test.go
echo "---"
head -n 10 pkg/planner/core/casetest/windows/window_push_down_test.go
# Also search for 'intest' tag usage in the codebase
rg "intest" --type go -c | head -20Repository: pingcap/tidb
Length of output: 1603
🏁 Script executed:
#!/bin/bash
# Search more specifically for build tag usage in the test files
rg "//.*build.*intest" pkg/executor/join/test/indexjoin/index_lookup_join_test.go pkg/planner/core/casetest/windows/window_push_down_test.go
# Check if there's any mention of intest tag in the test files
rg "intest" pkg/executor/join/test/indexjoin/index_lookup_join_test.go pkg/planner/core/casetest/windows/window_push_down_test.goRepository: pingcap/tidb
Length of output: 38
🏁 Script executed:
#!/bin/bash
# Check the actual content around line 1-20 of both test files to see if there are build tags
echo "=== index_lookup_join_test.go ==="
head -n 20 pkg/executor/join/test/indexjoin/index_lookup_join_test.go | cat -n
echo ""
echo "=== window_push_down_test.go ==="
head -n 20 pkg/planner/core/casetest/windows/window_push_down_test.go | cat -n
# Check if intest is used as a build tag anywhere in the repo
echo ""
echo "=== Searching for intest build tag usage ==="
rg "//\+build.*intest|//go:build.*intest" --type go | head -10Repository: pingcap/tidb
Length of output: 2098
🏁 Script executed:
#!/bin/bash
# Verify correct Go test flag syntax for build tags
# The flag should be -tags not --tags
echo "=== Checking Go test documentation ==="
go help test 2>&1 | grep -A5 "tags"Repository: pingcap/tidb
Length of output: 400
🏁 Script executed:
#!/bin/bash
# Check Go test command syntax more directly
go test -help 2>&1 | grep -i "tags"Repository: pingcap/tidb
Length of output: 38
🏁 Script executed:
#!/bin/bash
# Search for examples of go test commands with tags in the documentation
rg "go test.*-tags" --type md | head -10
# Also check if the actual commands work by examining the structure
echo "=== Checking command structure ==="
echo "Command format check:"
echo "go test ./pkg/executor/join/test/indexjoin -run TestIndexJoinWithStreamWindowInner --tags=intest"
echo ""
echo "Expected Go syntax: go test <package> -run <pattern> -tags <value>"Repository: pingcap/tidb
Length of output: 1463
🏁 Script executed:
#!/bin/bash
# Verify both flag formats work with Go test
echo "=== Testing both flag formats ==="
# Test with short form (standard Go)
go test -help 2>&1 | grep -E "^ -tags" || echo "Short form -tags found in help"
# Check if long form --tags works by searching examples
rg "\-\-tags=" docs/agents/ --type md | wc -l
rg "\-tags=" docs/agents/ --type md | wc -l
# Get consistency pattern from similar docs files
echo ""
echo "=== Consistency check: flag format in stream_window_notes.md vs other docs ==="
rg "go test.*tags" docs/agents/planner/ --type md | head -5Repository: pingcap/tidb
Length of output: 935
Change --tags=intest to -tags=intest for consistency with Go test syntax and other similar commands in the repository.
The test paths and function names are correct and copy-pasteable from the repository root. However, the commands use the non-standard flag format --tags=intest instead of the standard Go test format -tags=intest. Other similar commands throughout docs/agents/planner/ consistently use -tags=intest,deadlock (short form with equals).
Affected commands:
go test ./pkg/executor/join/test/indexjoin -run TestIndexJoinWithStreamWindowInner -tags=intestgo test ./pkg/planner/core/casetest/windows -run TestWindowPushDownPlans -tags=intest
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@docs/agents/planner/stream_window_notes.md` around lines 80 - 81, Replace the
non-standard `--tags=intest` flag with the standard Go test short form
`-tags=intest` in the two command lines shown (the lines containing "go test
./pkg/executor/join/test/indexjoin -run TestIndexJoinWithStreamWindowInner
--tags=intest" and "go test ./pkg/planner/core/casetest/windows -run
TestWindowPushDownPlans --tags=intest") so they match other docs entries that
use `-tags=intest` (or `-tags=intest,deadlock`).
| Validation commands used during the demo: | ||
| - `go test ./pkg/planner/core/casetest/windows -run TestWindowPushDownPlans -record --tags=intest` | ||
| - `go test ./pkg/planner/core/casetest/windows -run TestWindowPushDownPlans --tags=intest` |
There was a problem hiding this comment.
Remove or clarify duplicate validation commands.
Lines 84-85 duplicate the commands already listed at lines 80-81. The header "Validation commands used during the demo:" is ambiguous—it's unclear which demo this refers to (the first demo from section 1, or "this step" from section 2).
📝 Suggested fix
If these commands are meant to reference the first demo section, consider removing them here since they're already documented at lines 80-81. If they serve a different purpose, clarify the header to explain the distinction.
-Validation commands used during the demo:
-- `go test ./pkg/planner/core/casetest/windows -run TestWindowPushDownPlans -record --tags=intest`
-- `go test ./pkg/planner/core/casetest/windows -run TestWindowPushDownPlans --tags=intest`📝 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.
| Validation commands used during the demo: | |
| - `go test ./pkg/planner/core/casetest/windows -run TestWindowPushDownPlans -record --tags=intest` | |
| - `go test ./pkg/planner/core/casetest/windows -run TestWindowPushDownPlans --tags=intest` |
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@docs/agents/planner/stream_window_notes.md` around lines 83 - 85, The
duplicated validation command lines containing "go test
./pkg/planner/core/casetest/windows -run TestWindowPushDownPlans -record
--tags=intest" and "go test ./pkg/planner/core/casetest/windows -run
TestWindowPushDownPlans --tags=intest" should be de-duplicated and the header
clarified; either remove the second occurrence (lines 84-85) if they repeat the
earlier listing, or change the header "Validation commands used during the
demo:" to specify which demo/step these commands apply to (e.g., "Validation
commands for demo 1" or "Validation commands for step 2"), so the purpose and
scope of the commands are unambiguous.
| // OpenSelf initializes the executor state without opening children. | ||
| func (e *PartitionTopNWindowExec) OpenSelf() error { | ||
| e.childResult = nil | ||
| e.groupStart = 0 | ||
| e.groupEnd = 0 | ||
| e.groupRank = 0 | ||
| e.exhausted = false | ||
| e.resumeLastPartition = false | ||
| return nil |
There was a problem hiding this comment.
Reset groupChecker in OpenSelf.
OpenSelf clears the executor-local cursors, but it leaves VecGroupChecker state from the previous run intact. If this executor is reopened after an incomplete prior scan, loadNextGroup() can observe stale group offsets and skip the initial fetch on a nil childResult.
Suggested fix
func (e *PartitionTopNWindowExec) OpenSelf() error {
+ if e.groupChecker != nil {
+ e.groupChecker.Reset()
+ }
e.childResult = nil
e.groupStart = 0
e.groupEnd = 0
e.groupRank = 0
e.exhausted = false📝 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.
| // OpenSelf initializes the executor state without opening children. | |
| func (e *PartitionTopNWindowExec) OpenSelf() error { | |
| e.childResult = nil | |
| e.groupStart = 0 | |
| e.groupEnd = 0 | |
| e.groupRank = 0 | |
| e.exhausted = false | |
| e.resumeLastPartition = false | |
| return nil | |
| // OpenSelf initializes the executor state without opening children. | |
| func (e *PartitionTopNWindowExec) OpenSelf() error { | |
| if e.groupChecker != nil { | |
| e.groupChecker.Reset() | |
| } | |
| e.childResult = nil | |
| e.groupStart = 0 | |
| e.groupEnd = 0 | |
| e.groupRank = 0 | |
| e.exhausted = false | |
| e.resumeLastPartition = false | |
| return nil |
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@pkg/executor/partition_topn_window.go` around lines 53 - 61, OpenSelf
currently resets executor cursors but leaves the VecGroupChecker state, which
can cause stale offsets in loadNextGroup() when childResult is nil; update
PartitionTopNWindowExec.OpenSelf to also reset the groupChecker (e.g., set
e.groupChecker to its zero value or call its Reset/Init method) so any state
from a prior run is cleared when reopening the executor.
| stmt := strings.ToLower(strings.TrimSpace(tt)) | ||
| if !strings.HasPrefix(stmt, "select") && !strings.HasPrefix(stmt, "explain") { | ||
| tk.MustExec(tt) |
There was a problem hiding this comment.
CTE queries are misclassified as exec-only statements
At Line 41, only select/explain are treated as query paths. WITH ... SELECT now goes through MustExec, which skips result/plan/warning assertions and can silently reduce coverage.
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) | |
| stmt := strings.ToLower(strings.TrimSpace(tt)) | |
| if !strings.HasPrefix(stmt, "select") && | |
| !strings.HasPrefix(stmt, "explain") && | |
| !strings.HasPrefix(stmt, "with") { | |
| tk.MustExec(tt) |
🤖 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 -
42, The test misclassifies CTEs because the stmt detection only checks for
"select" and "explain" prefixes; update the branch that decides to call
tk.MustExec(tt) so statements beginning with "with" (case-insensitive, including
"with recursive") are treated as queries too. Concretely, modify the condition
that uses stmt := strings.ToLower(strings.TrimSpace(tt)) and the subsequent if
that currently checks strings.HasPrefix(stmt, "select") and
strings.HasPrefix(stmt, "explain") to also consider strings.HasPrefix(stmt,
"with") (and/or "with recursive") so CTEs follow the query/plan/warning
assertion path rather than hitting tk.MustExec(tt).
There was a problem hiding this comment.
🧹 Nitpick comments (1)
pkg/planner/core/operator/physicalop/physical_window.go (1)
125-151: Refactor duplicated clone logic to avoid future drift.
PhysicalStreamWindow.Cloneduplicates the full deep-copy path fromPhysicalWindow.Clone. Consider extracting the common field-clone routine so both code paths stay consistent over time.As per coding guidelines, "Code SHOULD remain maintainable for future readers with basic TiDB familiarity..." and "Follow existing package-local conventions first and keep style consistent with nearby files."
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@pkg/planner/core/operator/physicalop/physical_window.go` around lines 125 - 151, The clone logic in PhysicalStreamWindow.Clone duplicates deep-copy steps already implemented in PhysicalWindow.Clone; extract the shared field-copying into a single helper (e.g., cloneWindowFields or a method like (*PhysicalWindow).copyCloneFields(target, newCtx)) that performs: SetSCtx, calls PhysicalSchemaProducer.CloneWithSelf, deep-clones PartitionBy, OrderBy, WindowFuncDescs and Frame, and returns the partially-cloned struct (or error). Replace the body of both PhysicalStreamWindow.Clone and PhysicalWindow.Clone to call this new helper and then perform any remaining type-specific adjustments so both clones share the exact same field-copy routine.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Nitpick comments:
In `@pkg/planner/core/operator/physicalop/physical_window.go`:
- Around line 125-151: The clone logic in PhysicalStreamWindow.Clone duplicates
deep-copy steps already implemented in PhysicalWindow.Clone; extract the shared
field-copying into a single helper (e.g., cloneWindowFields or a method like
(*PhysicalWindow).copyCloneFields(target, newCtx)) that performs: SetSCtx, calls
PhysicalSchemaProducer.CloneWithSelf, deep-clones PartitionBy, OrderBy,
WindowFuncDescs and Frame, and returns the partially-cloned struct (or error).
Replace the body of both PhysicalStreamWindow.Clone and PhysicalWindow.Clone to
call this new helper and then perform any remaining type-specific adjustments so
both clones share the exact same field-copy routine.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository UI
Review profile: CHILL
Plan: Pro
Run ID: 5a9f606c-7c2a-42d1-b554-b48a2ba03643
📒 Files selected for processing (1)
pkg/planner/core/operator/physicalop/physical_window.go
There was a problem hiding this comment.
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
pkg/executor/builder.go (1)
5711-5829:⚠️ Potential issue | 🟠 MajorPropagate
UpdateCompareColsfailures intob.err.Both RANGE-frame branches return
nilonUpdateCompareColserrors without recording the cause. After this refactor, that can leave callers with a nil executor and a misleading follow-up failure instead of the real frame/order error.💡 Proposed fix
if v.Frame == nil { exec.start = &logicalop.FrameBound{ Type: ast.Preceding, UnBounded: true, @@ exec.orderByCols = orderByCols exec.expectedCmpResult = cmpResult exec.isRangeFrame = true err = exec.start.UpdateCompareCols(b.ctx, exec.orderByCols) if err != nil { + b.err = err return nil } err = exec.end.UpdateCompareCols(b.ctx, exec.orderByCols) if err != nil { + b.err = err return nil } } } @@ err = tmpProcessor.start.UpdateCompareCols(b.ctx, orderByCols) if err != nil { + b.err = err return nil } err = tmpProcessor.end.UpdateCompareCols(b.ctx, orderByCols) if err != nil { + b.err = err return nil }As per coding guidelines, "Keep error handling actionable and contextual; avoid silently swallowing errors."
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@pkg/executor/builder.go` around lines 5711 - 5829, In buildWindowExecWithChildExec the calls to UpdateCompareCols on both the pipelined branch (exec.start.UpdateCompareCols / exec.end.UpdateCompareCols) and the non-pipelined range branch (tmpProcessor.start.UpdateCompareCols / tmpProcessor.end.UpdateCompareCols) discard errors by returning nil without recording them; change those error paths to assign the error to b.err (e.g., b.err = err) before returning nil so callers see the original failure and context, leaving all other logic unchanged and keeping the same early-return behavior in buildWindowExecWithChildExec.
🧹 Nitpick comments (1)
pkg/planner/core/operator/physicalop/physical_window.go (1)
125-151: Extract the shared window-clone logic.
PhysicalStreamWindow.Cloneis now effectively a copy ofPhysicalWindow.Clone. The next field added toPhysicalWindowhas to be remembered in both places, which is easy to miss and can quietly desync the two operators.As per coding guidelines, "Code SHOULD remain maintainable for future readers with basic TiDB familiarity, including readers who are not experts in the specific subsystem/feature."
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@pkg/planner/core/operator/physicalop/physical_window.go` around lines 125 - 151, PhysicalStreamWindow.Clone duplicates the same field-copy logic found in PhysicalWindow.Clone; extract the common cloning of PartitionBy, OrderBy, WindowFuncDescs and Frame into a single helper (e.g., cloneWindowFields or a method on a shared embedding like PhysicalWindow.cloneFields) and call that helper from both PhysicalStreamWindow.Clone and PhysicalWindow.Clone; keep the existing handling of SetSCtx and the call to PhysicalSchemaProducer.CloneWithSelf in each Clone method but move the loops that allocate and append PartitionBy/OrderBy/WindowFuncDescs and the Frame nil-check into the shared helper so future field additions only need one update.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Outside diff comments:
In `@pkg/executor/builder.go`:
- Around line 5711-5829: In buildWindowExecWithChildExec the calls to
UpdateCompareCols on both the pipelined branch (exec.start.UpdateCompareCols /
exec.end.UpdateCompareCols) and the non-pipelined range branch
(tmpProcessor.start.UpdateCompareCols / tmpProcessor.end.UpdateCompareCols)
discard errors by returning nil without recording them; change those error paths
to assign the error to b.err (e.g., b.err = err) before returning nil so callers
see the original failure and context, leaving all other logic unchanged and
keeping the same early-return behavior in buildWindowExecWithChildExec.
---
Nitpick comments:
In `@pkg/planner/core/operator/physicalop/physical_window.go`:
- Around line 125-151: PhysicalStreamWindow.Clone duplicates the same field-copy
logic found in PhysicalWindow.Clone; extract the common cloning of PartitionBy,
OrderBy, WindowFuncDescs and Frame into a single helper (e.g., cloneWindowFields
or a method on a shared embedding like PhysicalWindow.cloneFields) and call that
helper from both PhysicalStreamWindow.Clone and PhysicalWindow.Clone; keep the
existing handling of SetSCtx and the call to
PhysicalSchemaProducer.CloneWithSelf in each Clone method but move the loops
that allocate and append PartitionBy/OrderBy/WindowFuncDescs and the Frame
nil-check into the shared helper so future field additions only need one update.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository UI
Review profile: CHILL
Plan: Pro
Run ID: 063bdafc-fb70-4989-a322-08f06f975f30
📒 Files selected for processing (3)
pkg/executor/builder.gopkg/planner/core/exhaust_physical_plans.gopkg/planner/core/operator/physicalop/physical_window.go
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (2)
pkg/planner/core/operator/physicalop/physical_window.go (2)
125-151: Code duplication withPhysicalWindow.Clone.This method duplicates the cloning logic from
PhysicalWindow.Clone. SincePhysicalStreamWindowembedsPhysicalWindow, consider extracting a helper to clone thePhysicalWindowfields and reusing it in both methods. This reduces the risk of divergence ifPhysicalWindowgains new fields.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@pkg/planner/core/operator/physicalop/physical_window.go` around lines 125 - 151, The cloning logic in PhysicalStreamWindow.Clone duplicates PhysicalWindow.Clone; extract a helper that clones the embedded PhysicalWindow fields (PartitionBy, OrderBy, WindowFuncDescs, Frame and the call to PhysicalSchemaProducer.CloneWithSelf) so both PhysicalWindow.Clone and PhysicalStreamWindow.Clone call that helper instead of repeating code; implement the helper to accept the source (p), destination (cloned) and newCtx (for SetSCtx/CloneWithSelf) and reuse it from both Clone methods.
65-71: Define a TypeStreamWindow constant in plancodec for consistency.
PhysicalWindow.Inituses the constantplancodec.TypeWindow, butPhysicalStreamWindow.Inituses the string literal"StreamWindow". Add aTypeStreamWindow = "StreamWindow"constant inpkg/util/plancodec/id.goand use it here for consistency with the existing pattern (similar toTypeStreamAggandTypeHashAgg).🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@pkg/planner/core/operator/physicalop/physical_window.go` around lines 65 - 71, Add a new constant TypeStreamWindow = "StreamWindow" in the plancodec id definitions (same file that defines TypeStreamAgg/TypeHashAgg), then update PhysicalStreamWindow.Init to call NewBasePhysicalPlan(ctx, plancodec.TypeStreamWindow, &p, offset) instead of using the hardcoded "StreamWindow" literal; target the plancodec id file for the constant and the PhysicalStreamWindow.Init method to apply the change.
🤖 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/planner/core/operator/physicalop/physical_window.go`:
- Around line 618-621: The current check only matches when the left operand is
the row_number column (col, ok = args[0].(*expression.Column) and
col.EqualByExprAndID(nil, rowNumberCol)), so update the logic in
physical_window.go to accept either ordering by checking both args: try casting
args[0] to *expression.Column and comparing with rowNumberCol, and if that
fails, cast args[1] to *expression.Column and compare; ensure you still return
(0, false) when neither side matches, and preserve use of EqualByExprAndID for
the comparisons so both `row_number = constant` and `constant = row_number` are
handled.
---
Nitpick comments:
In `@pkg/planner/core/operator/physicalop/physical_window.go`:
- Around line 125-151: The cloning logic in PhysicalStreamWindow.Clone
duplicates PhysicalWindow.Clone; extract a helper that clones the embedded
PhysicalWindow fields (PartitionBy, OrderBy, WindowFuncDescs, Frame and the call
to PhysicalSchemaProducer.CloneWithSelf) so both PhysicalWindow.Clone and
PhysicalStreamWindow.Clone call that helper instead of repeating code; implement
the helper to accept the source (p), destination (cloned) and newCtx (for
SetSCtx/CloneWithSelf) and reuse it from both Clone methods.
- Around line 65-71: Add a new constant TypeStreamWindow = "StreamWindow" in the
plancodec id definitions (same file that defines TypeStreamAgg/TypeHashAgg),
then update PhysicalStreamWindow.Init to call NewBasePhysicalPlan(ctx,
plancodec.TypeStreamWindow, &p, offset) instead of using the hardcoded
"StreamWindow" literal; target the plancodec id file for the constant and the
PhysicalStreamWindow.Init method to apply the change.
🪄 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: dede3a07-90d9-4960-8e59-76494f76f7c4
📒 Files selected for processing (5)
docs/agents/planner/stream_window_notes.mdpkg/executor/builder.gopkg/executor/join/test/indexjoin/index_lookup_join_test.gopkg/planner/core/operator/physicalop/physical_window.gopkg/planner/core/testdata/plan_suite_unexported_out.json
✅ Files skipped from review due to trivial changes (2)
- docs/agents/planner/stream_window_notes.md
- pkg/executor/builder.go
🚧 Files skipped from review as they are similar to previous changes (1)
- pkg/executor/join/test/indexjoin/index_lookup_join_test.go
| col, ok = args[0].(*expression.Column) | ||
| if !ok || !col.EqualByExprAndID(nil, rowNumberCol) { | ||
| return 0, false | ||
| } |
There was a problem hiding this comment.
Equality check only handles column on left side.
The equality pattern matching at lines 618-621 only checks for row_number = constant, not constant = row_number. While SQL optimizers typically normalize the order, this asymmetry could miss cases.
🔧 Proposed fix to handle both orderings
col, ok = args[0].(*expression.Column)
-if !ok || !col.EqualByExprAndID(nil, rowNumberCol) {
- return 0, false
+if ok && col.EqualByExprAndID(nil, rowNumberCol) {
+ constant, ok := args[1].(*expression.Constant)
+ if ok {
+ if value, ok := constant.Value.GetValue().(int64); ok && value > 0 {
+ return uint64(value), true
+ }
+ }
+ return 0, false
}
-constant, ok := args[1].(*expression.Constant)
-if !ok {
- return 0, false
+// Try the other order: constant = row_number
+constant, ok := args[0].(*expression.Constant)
+if !ok {
+ return 0, false
}
-value, ok := constant.Value.GetValue().(int64)
-if !ok || value <= 0 {
- return 0, false
+col, ok = args[1].(*expression.Column)
+if !ok || !col.EqualByExprAndID(nil, rowNumberCol) {
+ return 0, false
}
-return uint64(value), true
+value, ok := constant.Value.GetValue().(int64)
+if !ok || value <= 0 {
+ return 0, false
+}
+return uint64(value), true🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@pkg/planner/core/operator/physicalop/physical_window.go` around lines 618 -
621, The current check only matches when the left operand is the row_number
column (col, ok = args[0].(*expression.Column) and col.EqualByExprAndID(nil,
rowNumberCol)), so update the logic in physical_window.go to accept either
ordering by checking both args: try casting args[0] to *expression.Column and
comparing with rowNumberCol, and if that fails, cast args[1] to
*expression.Column and compare; ensure you still return (0, false) when neither
side matches, and preserve use of EqualByExprAndID for the comparisons so both
`row_number = constant` and `constant = row_number` are handled.
Codecov Report❌ Patch coverage is Please upload reports for the commit cec2918 to get more accurate results. Additional details and impacted files@@ Coverage Diff @@
## master #67696 +/- ##
================================================
- Coverage 77.5947% 77.1122% -0.4826%
================================================
Files 1981 1965 -16
Lines 548157 549153 +996
================================================
- Hits 425341 423464 -1877
- Misses 122006 125687 +3681
+ Partials 810 2 -808
Flags with carried forward coverage won't be shown. Click here to find out more.
🚀 New features to boost your workflow:
|
|
/retest |
|
@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. |
| isOrderedWindow = true | ||
| default: | ||
| return base.InvalidTask | ||
| } | ||
| if mpp, ok := tasks[0].Copy().(*physicalop.MppTask); ok && p.StoreTp == kv.TiFlash { |
There was a problem hiding this comment.
is any compatible issues when mpp is true and isOrderedWindow is true
| @@ -48,6 +48,10 @@ func optimizeByShuffle(tsk base.Task, ctx base.PlanContext) base.Task { | |||
| if shuffle := optimizeByShuffle4Window(p, ctx); shuffle != nil { | |||
| return shuffle.Attach2Task(tsk) | |||
| } | |||
| case *physicalop.PhysicalOrderedWindow: | |||
| if shuffle := optimizeByShuffle4Window(&p.PhysicalWindow, ctx); shuffle != nil { | |||
There was a problem hiding this comment.
why orderedwindow need shuffle?
| @@ -793,6 +786,9 @@ func setupFineGrainedShuffleInternal(ctx context.Context, sctx base.PlanContext, | |||
| // which will break data partition. | |||
| helper.updateTarget(window, &x.BasePhysicalPlan) | |||
| setupFineGrainedShuffleInternal(ctx, sctx, x.Children()[0], helper, streamCountInfo, tiflashServerCountInfo) | |||
| case *physicalop.PhysicalOrderedWindow: | |||
There was a problem hiding this comment.
This code path is for mpp, why need to handle PhysicalOrderedWindow here?
| @@ -597,7 +620,17 @@ func admitIndexJoinInnerChildPattern(p base.LogicalPlan, indexJoinProp *property | |||
| if !checkIndexJoinInnerTaskWithAgg(x, indexJoinProp) { | |||
| return false | |||
| } | |||
|
|
|||
| case *logicalop.LogicalWindow: | |||
| if !p.SCtx().GetSessionVars().EnableINLJoinInnerMultiPattern { | |||
There was a problem hiding this comment.
This variable seems a general one, so there is no specific variable to enable/disable this feature?
| return nil, err | ||
| } | ||
| cloned.PhysicalSchemaProducer = *base | ||
| cloned.PartitionBy = make([]property.SortItem, 0, len(p.PartitionBy)) |
There was a problem hiding this comment.
it looks to me that all the following fields is from PhysicalWindow, so i think you should find a way to reuse the clone mehtod of PhysicalWindow
| @@ -462,10 +514,34 @@ func ExhaustPhysicalPlans4LogicalWindow(lp base.LogicalPlan, prop *property.Phys | |||
| var byItems []property.SortItem | |||
| byItems = append(byItems, lw.PartitionBy...) | |||
| byItems = append(byItems, lw.OrderBy...) | |||
There was a problem hiding this comment.
what if there is some shared items in lw.PartitionBy and lw.OrderBy
| if len(lw.WindowFuncDescs) != 1 { | ||
| return false | ||
| } | ||
| if lw.WindowFuncDescs[0].Name != ast.WindowFuncRowNumber { |
There was a problem hiding this comment.
why the first window function must be row_number
| } | ||
|
|
||
| // Next implements the Executor Next interface. | ||
| func (e *PartitionTopNWindowExec) Next(ctx context.Context, req *chunk.Chunk) error { |
There was a problem hiding this comment.
PartitionTopNWindowExec assumes that its input is ordered?
|
[FORMAT CHECKER NOTIFICATION] Notice: To remove the 📖 For more info, you can check the "Contribute Code" section in the development guide. |
What problem does this PR solve?
Issue Number: ref #66906
Problem Summary:
This PR adds a first demo path for stream window planning/execution. The goal is to show that a window operator can reuse an existing
PARTITION BY + ORDER BYproperty from the child plan instead of always falling back toSort + Window.What changed and how does it work?
PhysicalStreamWindowand only enumerate it when the child naturally provides the required orderStreamWindow, so this plan shape really means "reuse existing order"row_number()for nowStreamWindowExecas an explicit executor type and reuse the existing pipelined window implementation underneathwindow_push_down_suitefor:(partition by a)(partition by a order by b)withidx_ab(a, b)Windowforrank()Windowwhen theORDER BYkey is not covered by the available orderCheck List
Tests
Side effects
Documentation
Release note
Summary by CodeRabbit
New Features
Documentation
Tests