plan, exec: pushing down limit to slow log executor #65740
Conversation
Signed-off-by: lance6716 <lance6716@gmail.com>
ti-chi-bot
bot
added
release-note-none
Signed-off-by: lance6716 <lance6716@gmail.com>
There was a problem hiding this comment.
Pull request overview
This PR aims to improve slow log query efficiency by pushing down LIMIT/TopN hints into the slow log memtable executor/retriever so it can stop scanning earlier (especially for dashboard-style queries ordering by time).
Changes:
- Add row-limit and descending-order hint interfaces for memtable extractors, and implement them in
SlowQueryExtractor. - Push down TopN/LIMIT hints to slow log memtable plans (logical plan and PB plan builder paths).
- Refactor slow log reverse scanning and add tests verifying limit pushdown and reverse-scan behavior.
Reviewed changes
Copilot reviewed 8 out of 8 changed files in this pull request and generated 3 comments.
Show a summary per file
| File | Description |
|---|---|
| pkg/util/util.go | Splits line-reading into borrowed vs non-aliased variants (ReadLine vs ReadLineCopy) and updates multi-line reads accordingly. |
| pkg/util/stmtsummary/v2/reader.go | Switches statement-summary reader to use non-aliased line reads. |
| pkg/planner/core/base/misc_base.go | Introduces optional memtable extractor hint interfaces (row limit, desc). |
| pkg/planner/core/memtable_predicate_extractor.go | Adds Limit hint + setters to SlowQueryExtractor. |
| pkg/planner/core/operator/logicalop/logical_mem_table.go | Pushes down TopN/LIMIT/desc hints for slow log memtables during logical optimization. |
| pkg/planner/core/pb_to_plan.go | Attempts to propagate LIMIT hint into slow log extractor when building plans from protobuf executors. |
| pkg/executor/builder.go | Wires extractor’s limit hint into the slow-query retriever. |
| pkg/executor/slow_query.go | Adds limit-aware parsing and a new reverse scanner implementation for slow log reading. |
| pkg/executor/slow_query_test.go | Adds coverage for reverse scan with limit and PB plan builder limit pushdown. |
Comments suppressed due to low confidence (1)
pkg/executor/slow_query.go:719
- After publishing an error to
taskList,parseSlowLogcontinues the loop. If the caller stops consumingtaskListafter receiving the error (without canceling the context), the producer goroutine can block on further sends and leak. Return immediately aftere.sendParsedSlowLogCh(..., err)(consistent with other parsing paths).
if err != nil {
t := slowLogTask{}
t.resultCh = make(chan parsedSlowLog, 1)
select {
case <-ctx.Done():
| if memTable, ok := p.Children()[0].(*physicalop.PhysicalMemTable); ok { | ||
| if extractor, ok := memTable.Extractor.(*SlowQueryExtractor); ok { | ||
| end := limit.Offset + limit.Count | ||
| if end < limit.Offset { | ||
| end = ^uint64(0) | ||
| } | ||
| extractor.SetRowLimitHint(end) | ||
| } |
There was a problem hiding this comment.
The limit hint is only applied when the limit’s direct child is *PhysicalMemTable. In PBPlanBuilder, TypeSelection can sit between TableScan and Limit and later be removed by predicatePushDown, so this check can miss and the hint won’t be set for queries with WHERE. Consider walking down unary children to find the underlying memtable (or applying the hint after predicatePushDown).
| if memTable, ok := p.Children()[0].(*physicalop.PhysicalMemTable); ok { | |
| if extractor, ok := memTable.Extractor.(*SlowQueryExtractor); ok { | |
| end := limit.Offset + limit.Count | |
| if end < limit.Offset { | |
| end = ^uint64(0) | |
| } | |
| extractor.SetRowLimitHint(end) | |
| } | |
| // Walk down unary children to find the underlying memtable. In PBPlanBuilder, | |
| // operators like TypeSelection can sit between Limit and MemTable and may | |
| // later be removed by predicatePushDown, so checking only the direct child | |
| // can miss and the hint won't be set for queries with WHERE. | |
| child := p.Children()[0] | |
| for child != nil { | |
| if memTable, ok := child.(*physicalop.PhysicalMemTable); ok { | |
| if extractor, ok := memTable.Extractor.(*SlowQueryExtractor); ok { | |
| end := limit.Offset + limit.Count | |
| if end < limit.Offset { | |
| end = ^uint64(0) | |
| } | |
| extractor.SetRowLimitHint(end) | |
| } | |
| break | |
| } | |
| children := child.Children() | |
| if len(children) != 1 { | |
| break | |
| } | |
| child = children[0] |
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## master #65740 +/- ##
================================================
+ Coverage 77.7867% 78.7545% +0.9677%
================================================
Files 1996 1925 -71
Lines 544730 542318 -2412
================================================
+ Hits 423728 427100 +3372
+ Misses 119343 114811 -4532
+ Partials 1659 407 -1252
Flags with carried forward coverage won't be shown. Click here to find out more.
🚀 New features to boost your workflow:
|
|
Hi @lance6716. Thanks for your PR. PRs from untrusted users cannot be marked as trusted with I understand the commands that are listed here. 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. |
Signed-off-by: lance6716 <lance6716@gmail.com>
Signed-off-by: lance6716 <lance6716@gmail.com>
Signed-off-by: lance6716 <lance6716@gmail.com>
| extractor.SetRowLimitHint(end) | ||
| } | ||
| } | ||
| } |
There was a problem hiding this comment.
This pushdown currently only handles PhysicalLimit. For ORDER BY ... LIMIT queries, the DAG commonly uses TopN (tipb.ExecType_TypeTopN -> PhysicalTopN), so SlowQueryExtractor.Limit won’t be set in that case. Consider applying the same (offset+count) row-limit hint when building PhysicalTopN as well, so the optimization works for the dashboard query pattern.
| } | |
| } | |
| if topN, ok := p.(*physicalop.PhysicalTopN); ok { | |
| if memTable, ok := p.Children()[0].(*physicalop.PhysicalMemTable); ok { | |
| if extractor, ok := memTable.Extractor.(*SlowQueryExtractor); ok { | |
| end := topN.Offset + topN.Count | |
| if end < topN.Offset { | |
| end = ^uint64(0) | |
| } | |
| extractor.SetRowLimitHint(end) | |
| } | |
| } | |
| } |
| func (s *slowLogReverseScanner) loadCompressedBlocks(ctx context.Context, file *os.File) error { | ||
| _, err := file.Seek(0, io.SeekStart) | ||
| if err != nil { | ||
| return err | ||
| } | ||
| gr, err := gzip.NewReader(file) | ||
| if err != nil { | ||
| return err | ||
| } | ||
| defer func() { _ = gr.Close() }() | ||
|
|
||
| reader := bufio.NewReader(gr) | ||
| blocks := make([]slowLogBlock, 0) | ||
| var ( | ||
| block slowLogBlock | ||
| hasStartFlag bool | ||
| ) | ||
| for { | ||
| if isCtxDone(ctx) { | ||
| return ctx.Err() | ||
| } | ||
| lineByte, err := getOneLine(reader) | ||
| if err != nil { | ||
| if err == io.EOF { | ||
| if len(log) == 0 { | ||
| decomposedSlowLogTasks := decomposeToSlowLogTasks(logs, num) | ||
| offset.length = len(decomposedSlowLogTasks) | ||
| return decomposedSlowLogTasks, nil | ||
| } | ||
| e.fileLine = 0 | ||
| reader, err = e.getPreviousReader() | ||
| if reader == nil || err != nil { | ||
| return decomposeToSlowLogTasks(logs, num), nil | ||
| if len(block) > 0 { | ||
| blocks = append(blocks, block) | ||
| } | ||
| scanPreviousFile = true | ||
| continue | ||
| s.compressedBlocks = blocks | ||
| return nil | ||
| } | ||
| return nil, err | ||
| return err | ||
| } | ||
| line = string(hack.String(lineByte)) | ||
| line := string(hack.String(lineByte)) | ||
| if !hasStartFlag && strings.HasPrefix(line, variable.SlowLogStartPrefixStr) { | ||
| hasStartFlag = true | ||
| } | ||
| if hasStartFlag { | ||
| log = append(log, line) | ||
| block = append(block, line) | ||
| if strings.HasSuffix(line, variable.SlowLogSQLSuffixStr) { | ||
| if strings.HasPrefix(line, "use") || strings.HasPrefix(line, variable.SlowLogRowPrefixStr) { | ||
| continue | ||
| } | ||
| logs = append(logs, log) | ||
| if scanPreviousFile { | ||
| break | ||
| } | ||
| log = make([]string, 0, 8) | ||
| blocks = append(blocks, block) | ||
| block = make(slowLogBlock, 0, 8) |
There was a problem hiding this comment.
loadCompressedBlocks decompresses and materializes all slow-log blocks from a .gz file into memory (blocks / s.compressedBlocks). For large rotated slow logs this can be very memory-intensive and defeats the benefit of a small LIMIT hint. Consider streaming and keeping only the last N blocks needed (e.g. based on maxBlocks/e.limit) instead of storing every block, or otherwise bounding memory usage.
Signed-off-by: lance6716 <lance6716@gmail.com>
Signed-off-by: lance6716 <lance6716@gmail.com>
Signed-off-by: lance6716 <lance6716@gmail.com>
| if e.limit > 0 { | ||
| e.parseSlowLogByBatchGetterWithLimit(ctx, sctx, batchSize, off, nextBatch, afterBatch) | ||
| return | ||
| } |
There was a problem hiding this comment.
parseSlowLogByBatchGetter switches to a fully-serial parsing path whenever e.limit > 0. Because limit hints are now pushed down for normal LIMIT queries too, this can reduce throughput for large limits (e.g. exporting many slow logs) compared to the concurrent path. Consider keeping concurrency when the limit is large (or when early-exit is unlikely), or gating the serial path behind a small-limit threshold.
|
/hold failed CI shows there's a correctness problem |
ti-chi-bot
bot
added
the
do-not-merge/hold
This comment was marked as outdated.
This comment was marked as outdated.
Signed-off-by: lance6716 <lance6716@gmail.com>
|
/unhold |
ti-chi-bot
bot
removed
the
do-not-merge/hold
ti-chi-bot
bot
added
approved
needs-1-more-lgtm
|
[APPROVALNOTIFIER] This PR is APPROVED This pull-request has been approved by: crazycs520, hawkingrei The full list of commands accepted by this bot can be found here. The pull request process is described here DetailsNeeds approval from an approver in each of these files:
Approvers can indicate their approval by writing |
ti-chi-bot
bot
added
lgtm
and removed
needs-1-more-lgtm
[LGTM Timeline notifier]Timeline:
|
|
/hold I'll check test coverage soon |
ti-chi-bot
bot
added
the
do-not-merge/hold
|
/unhold |
ti-chi-bot
bot
removed
the
do-not-merge/hold
|
/retest |
|
@lance6716: PRs from untrusted users cannot be marked as trusted with DetailsIn response to this:
Instructions 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. |
|
/retest |
|
@lance6716: PRs from untrusted users cannot be marked as trusted with DetailsIn response to this:
Instructions 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. |
|
/retest |
|
@lance6716: PRs from untrusted users cannot be marked as trusted with DetailsIn response to this:
Instructions 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. |
3 participants
What problem does this PR solve?
Issue Number: close #65739
Problem Summary:
What changed and how does it work?
check limit and break earlier in slow log executor. In order to do it, need to implement a TopN push down rule for memory table operator.
This PR is mainly written by codex, I also reviewed it roughly before remove [WIP] flag.
Check List
Tests
in a 1000 * 300MB slow log files environment:
before this PR
after this PR
Side effects
Documentation
Release note
Please refer to Release Notes Language Style Guide to write a quality release note.