enhance: optimize QueryNode scheduler recovery performance from heavy load (#49825)

issue: #49707

- query node scheduler: make read tasks context-aware, use an unbuffered
add path, and cleanup canceled or near-deadline queued tasks before
queue limit checks
- query grouping: cap grouped NQ at 16 and add an NQ merge ratio guard
to avoid merging small queries into much larger groups
- REST timeout: propagate request timeout as a context deadline so
downstream query tasks receive timeout timestamps
- scheduler metrics: add ready NQ plus queue and execution duration
metrics while removing obsolete receive queue configuration
- proxy config: reduce the proxy task queue default to 256

---------

Signed-off-by: chyezh <chyezh@outlook.com>

configs/milvus.yaml

@@ -553,8 +553,7 @@ queryNode:
     maxDiskUsagePercentageForMmapAlloc: 50 # disk percentage used in mmap chunk manager
   indexOffsetCacheEnabled: false # enable index offset cache for some scalar indexes, now is just for bitmap index, enable this param can improve performance for retrieving raw data from index
   scheduler:
-    receiveChanSize: 10240
-    unsolvedQueueSize: 10240
+    unsolvedQueueSize: 1024
     # maxReadConcurrentRatio is the concurrency ratio of read task (search task and query task).
     # Max read concurrency would be the value of hardware.GetCPUNum * maxReadConcurrentRatio.
     # It defaults to 2.0, which means max read concurrency would be the value of hardware.GetCPUNum * 2.
@@ -570,13 +569,16 @@ queryNode:
       # 	Scheduling is fair on task granularity.
       # 	The policy is based on the username for authentication.
       # 	And an empty username is considered the same user.
-      # 	When there are no multi-users, the policy decay into FIFO"
+      # 	When there are no multi-users, the policy decay into FIFO.
       name: fifo
       taskQueueExpire: 60 # Control how long (many seconds) that queue retains since queue is empty
+      taskDeadlineAdvance: 50ms # Advance duration for cleaning queued query node read tasks before their context deadline. It supports duration strings such as 50ms and 1s. A bare number is interpreted as milliseconds for compatibility.
       enableCrossUserGrouping: false # Enable Cross user grouping when using user-task-polling policy. (Disable it if user's task can not merge each other)
       maxPendingTaskPerUser: 1024 # Max pending task per user in scheduler
   grouping:
-    maxNQ: 1000
+    maxNQ: 16
+    nqMergeRatio: 3 # Maximum ratio between merged total NQ and the smaller task NQ when grouping query node read tasks.
+    maxDeadlineMergeGap: 50ms # Maximum allowed gap between task context deadlines when grouping query node read tasks. Tasks with only one deadline cannot be grouped. Set a negative duration to disable this check.
     topKMergeRatio: 20
   search:
     enableResultZeroCopy: false # When true, delegator passes reduced SearchResultData directly instead of re-marshaling to SlicedBlob. Toggle at runtime for instant fallback.

internal/distributed/proxy/httpserver/handler_v2_test.go

@@ -636,6 +636,26 @@ func TestRestfulSizeMiddlewarePreservesRequestContextCancel(t *testing.T) {
 	assert.True(t, errors.Is(<-observed, context.Canceled))
 }
 
+func TestTimeoutMiddlewarePassesDeadline(t *testing.T) {
+	ginHandler := gin.Default()
+	app := ginHandler.Group("")
+	path := "/middleware/timeout/deadline"
+	app.POST(path, timeoutMiddleware(func(c *gin.Context) {
+		deadline, ok := c.Request.Context().Deadline()
+		assert.True(t, ok)
+		assert.LessOrEqual(t, time.Until(deadline), 3*time.Second)
+		assert.Greater(t, time.Until(deadline), time.Second)
+	}))
+
+	req := httptest.NewRequest(http.MethodPost, path, nil)
+	req.Header.Set(mhttp.HTTPHeaderRequestTimeout, "3")
+	w := httptest.NewRecorder()
+
+	ginHandler.ServeHTTP(w, req)
+
+	assert.Equal(t, http.StatusOK, w.Code)
+}
+
 func TestDocInDocOutCreateCollection(t *testing.T) {
 	paramtable.Init()
 	// disable rate limit

internal/distributed/proxy/httpserver/timeout_middleware.go

@@ -240,16 +240,16 @@ func timeoutMiddleware(handler gin.HandlerFunc) gin.HandlerFunc {
 	}
 	bufPool := &BufferPool{}
 	return func(gCtx *gin.Context) {
-		topCtx, cancel := context.WithCancel(gCtx.Request.Context())
-		defer cancel()
-		req := gCtx.Request.WithContext(topCtx)
-		gCtx.Request = req
-
 		timeout := paramtable.Get().HTTPCfg.RequestTimeoutMs.GetAsDuration(time.Millisecond)
 		timeoutSecond, err := strconv.ParseInt(gCtx.Request.Header.Get(mhttp.HTTPHeaderRequestTimeout), 10, 64)
 		if err == nil {
 			timeout = time.Duration(timeoutSecond) * time.Second
 		}
+		topCtx, cancel := context.WithTimeout(gCtx.Request.Context(), timeout)
+		defer cancel()
+		req := gCtx.Request.WithContext(topCtx)
+		gCtx.Request = req
+
 		finish := make(chan struct{}, 1)
 		panicChan := make(chan interface{}, 1)
 

internal/querynodev2/tasks/query_stream_task.go

@@ -54,6 +54,10 @@ func (t *QueryStreamTask) IsGpuIndex() bool {
 	return false
 }
 
+func (t *QueryStreamTask) Context() context.Context {
+	return t.ctx
+}
+
 // PreExecute the task, only call once.
 func (t *QueryStreamTask) PreExecute() error {
 	return nil
@@ -89,10 +93,6 @@ func (t *QueryStreamTask) Done(err error) {
 	t.notifier <- err
 }
 
-func (t *QueryStreamTask) Canceled() error {
-	return t.ctx.Err()
-}
-
 func (t *QueryStreamTask) Wait() error {
 	return <-t.notifier
 }

internal/querynodev2/tasks/query_task.go

@@ -68,6 +68,10 @@ func (t *QueryTask) IsGpuIndex() bool {
 	return false
 }
 
+func (t *QueryTask) Context() context.Context {
+	return t.ctx
+}
+
 // PreExecute the task, only call once.
 func (t *QueryTask) PreExecute() error {
 	// Update task wait time metric before execute
@@ -182,10 +186,6 @@ func (t *QueryTask) Done(err error) {
 	t.notifier <- err
 }
 
-func (t *QueryTask) Canceled() error {
-	return t.ctx.Err()
-}
-
 func (t *QueryTask) Wait() error {
 	return <-t.notifier
 }

internal/querynodev2/tasks/search_task.go

@@ -96,6 +96,10 @@ func (t *SearchTask) IsGpuIndex() bool {
 	return t.collection.IsGpuIndex()
 }
 
+func (t *SearchTask) Context() context.Context {
+	return t.ctx
+}
+
 func (t *SearchTask) PreExecute() error {
 	// Update task wait time metric before execute
 	nodeID := strconv.FormatInt(t.GetNodeID(), 10)
@@ -375,8 +379,7 @@ func (t *SearchTask) Merge(other *SearchTask) bool {
 		t.req.GetReq().GetMvccTimestamp() != other.req.GetReq().GetMvccTimestamp() ||
 		t.req.GetReq().GetDslType() != other.req.GetReq().GetDslType() ||
 		t.req.GetDmlChannels()[0] != other.req.GetDmlChannels()[0] ||
-		nq+otherNq > paramtable.Get().QueryNodeCfg.MaxGroupNQ.GetAsInt64() ||
-		diffTopk && ratio > paramtable.Get().QueryNodeCfg.TopKMergeRatio.GetAsFloat() ||
+		(diffTopk && ratio > paramtable.Get().QueryNodeCfg.TopKMergeRatio.GetAsFloat()) ||
 		!funcutil.SliceSetEqual(t.req.GetReq().GetPartitionIDs(), other.req.GetReq().GetPartitionIDs()) ||
 		!funcutil.SliceSetEqual(t.req.GetSegmentIDs(), other.req.GetSegmentIDs()) ||
 		!bytes.Equal(t.req.GetReq().GetSerializedExprPlan(), other.req.GetReq().GetSerializedExprPlan()) {
@@ -407,10 +410,6 @@ func (t *SearchTask) Done(err error) {
 	}
 }
 
-func (t *SearchTask) Canceled() error {
-	return t.ctx.Err()
-}
-
 func (t *SearchTask) Wait() error {
 	return <-t.notifier
 }
@@ -430,6 +429,19 @@ func (t *SearchTask) NQ() int64 {
 	return t.nq
 }
 
+func (t *SearchTask) MinNQ() int64 {
+	if len(t.originNqs) == 0 {
+		return t.nq
+	}
+	minNQ := t.originNqs[0]
+	for _, nq := range t.originNqs[1:] {
+		if nq < minNQ {
+			minNQ = nq
+		}
+	}
+	return minNQ
+}
+
 func (t *SearchTask) MergeWith(other scheduler.Task) bool {
 	switch other := other.(type) {
 	case *SearchTask:

internal/querynodev2/tasks/search_task_test.go

@@ -318,6 +318,21 @@ func (s *SearchTaskSuite) TestMergeFilterOnly() {
 	})
 }
 
+func (s *SearchTaskSuite) TestSearchTaskMinNQ() {
+	s.Run("fallback_to_total_nq_without_origin", func() {
+		task := &SearchTask{nq: 8}
+		s.Equal(int64(8), task.MinNQ())
+	})
+
+	s.Run("minimum_origin_nq", func() {
+		task := &SearchTask{
+			nq:        11,
+			originNqs: []int64{5, 2, 4},
+		}
+		s.Equal(int64(2), task.MinNQ())
+	})
+}
+
 func TestSearchTask(t *testing.T) {
 	suite.Run(t, new(SearchTaskSuite))
 }

internal/util/searchutil/scheduler/concurrent_safe_scheduler.go

@@ -1,8 +1,12 @@
 package scheduler
 
 import (
+	"context"
+	"fmt"
 	"sync"
+	"time"
 
+	"github.com/cockroachdb/errors"
 	"go.uber.org/atomic"
 	"go.uber.org/zap"
 
@@ -11,22 +15,25 @@ import (
 	"github.com/milvus-io/milvus/pkg/v3/metrics"
 	"github.com/milvus-io/milvus/pkg/v3/util/conc"
 	"github.com/milvus-io/milvus/pkg/v3/util/lifetime"
+	"github.com/milvus-io/milvus/pkg/v3/util/merr"
 	"github.com/milvus-io/milvus/pkg/v3/util/metricsinfo"
 	"github.com/milvus-io/milvus/pkg/v3/util/paramtable"
 )
 
 const (
 	maxReceiveChanBatchConsumeNum = 100
+
+	readTaskQueueOutcomeScheduled = "scheduled"
+	readTaskQueueOutcomeExpired   = "expired"
 )
 
 // newScheduler create a scheduler with given schedule policy.
 func newScheduler(policy schedulePolicy) Scheduler {
 	maxReadConcurrency := paramtable.Get().QueryNodeCfg.MaxReadConcurrency.GetAsInt()
-	maxReceiveChanSize := paramtable.Get().QueryNodeCfg.MaxReceiveChanSize.GetAsInt()
 	log.Info("query node use concurrent safe scheduler", zap.Int("max_concurrency", maxReadConcurrency))
 	return &scheduler{
 		policy:           policy,
-		receiveChan:      make(chan addTaskReq, maxReceiveChanSize),
+		receiveChan:      make(chan addTaskReq),
 		execChan:         make(chan Task),
 		pool:             conc.NewPool[any](maxReadConcurrency, conc.WithPreAlloc(true)),
 		gpuPool:          conc.NewPool[any](paramtable.Get().QueryNodeCfg.MaxGpuReadConcurrency.GetAsInt(), conc.WithPreAlloc(true)),
@@ -66,17 +73,20 @@ func (s *scheduler) Add(task Task) (err error) {
 
 	errCh := make(chan error, 1)
 
-	// TODO: add operation should be fast, is UnsolveLen metric unnesscery?
-	metrics.QueryNodeReadTaskUnsolveLen.WithLabelValues(paramtable.GetStringNodeID()).Inc()
-
-	// start a new in queue span and send task to add chan
-	s.receiveChan <- addTaskReq{
+	req := addTaskReq{
 		task: task,
 		err:  errCh,
 	}
-	err = <-errCh
 
-	metrics.QueryNodeReadTaskUnsolveLen.WithLabelValues(paramtable.GetStringNodeID()).Dec()
+	// start a new in queue span and send task to add chan
+	ctx := task.Context()
+	select {
+	case s.receiveChan <- req:
+		err = <-errCh
+	case <-ctx.Done():
+		err = ctx.Err()
+	}
+
 	return
 }
 
@@ -113,23 +123,28 @@ func (s *scheduler) Stop() {
 // schedule the owned task asynchronously and continuously.
 func (s *scheduler) schedule() {
 	defer s.wg.Done()
-	var task Task
+	var task *queuedTask
 	for {
 		s.setupReadyLenMetric()
 
 		var execChan chan Task
+		var execTask Task
 		nq := int64(0)
-		task, nq, execChan = s.setupExecListener(task)
+		now := time.Now()
+		task, nq, execChan = s.setupExecListener(task, now)
+		if task.valid() {
+			execTask = task.Task
+		}
 
 		select {
 		case req, ok := <-s.receiveChan:
 			if !ok {
 				log.Info("receiveChan closed, processing remaining request")
 				// drain policy maintained task
-				for task != nil {
-					execChan <- task
+				for task.valid() {
+					execChan <- task.Task
 					s.updateWaitingTaskCounter(-1, -nq)
-					task = s.produceExecChan()
+					task = s.produceExecChan(now)
 				}
 				log.Info("all task put into exeChan, schedule worker exit")
 				close(s.execChan)
@@ -137,22 +152,22 @@ func (s *scheduler) schedule() {
 			}
 			// Receive add operation request and return the process result.
 			// And consume recv chan as much as possible.
-			s.consumeRecvChan(req, maxReceiveChanBatchConsumeNum)
-		case execChan <- task:
+			s.consumeRecvChan(req, maxReceiveChanBatchConsumeNum, now)
+		case execChan <- execTask:
 			// Task sent, drop the ownership of sent task.
 			// Update waiting task counter.
 			s.updateWaitingTaskCounter(-1, -nq)
 			// And produce new task into execChan as much as possible.
-			task = s.produceExecChan()
+			task = s.produceExecChan(now)
 		}
 	}
 }
 
 // consumeRecvChan consume the recv chan as much as possible.
-func (s *scheduler) consumeRecvChan(req addTaskReq, limit int) {
+func (s *scheduler) consumeRecvChan(req addTaskReq, limit int, now time.Time) {
 	// Check the dynamic wait task limit.
 	maxWaitTaskNum := paramtable.Get().QueryNodeCfg.MaxUnsolvedQueueSize.GetAsInt64()
-	if !s.handleAddTaskRequest(req, maxWaitTaskNum) {
+	if !s.handleAddTaskRequest(req, maxWaitTaskNum, now) {
 		return
 	}
 
@@ -163,7 +178,7 @@ func (s *scheduler) consumeRecvChan(req addTaskReq, limit int) {
 			if !ok {
 				return
 			}
-			if !s.handleAddTaskRequest(req, maxWaitTaskNum) {
+			if !s.handleAddTaskRequest(req, maxWaitTaskNum, now) {
 				return
 			}
 		default:
@@ -174,14 +189,25 @@ func (s *scheduler) consumeRecvChan(req addTaskReq, limit int) {
 
 // HandleAddTaskRequest handle a add task request.
 // Return true if the process can be continued.
-func (s *scheduler) handleAddTaskRequest(req addTaskReq, maxWaitTaskNum int64) bool {
-	if err := req.task.Canceled(); err != nil {
+func (s *scheduler) handleAddTaskRequest(req addTaskReq, maxWaitTaskNum int64, now time.Time) bool {
+	if maxWaitTaskNum > 0 && s.GetWaitingTaskTotal() >= maxWaitTaskNum {
+		s.cleanupExpiredTasks(now)
+	}
+
+	if err := req.task.Context().Err(); err != nil {
 		log.Warn("task canceled before enqueue", zap.Error(err))
 		req.err <- err
+	} else if maxWaitTaskNum > 0 && s.GetWaitingTaskTotal() >= maxWaitTaskNum {
+		err := merr.WrapErrTooManyRequests(
+			int32(maxWaitTaskNum),
+			fmt.Sprintf("limit by %s", paramtable.Get().QueryNodeCfg.MaxUnsolvedQueueSize.Key),
+		)
+		req.err <- err
 	} else {
 		// Push the task into the policy to schedule and update the counter of the ready queue.
-		nq := req.task.NQ()
-		newTaskAdded, err := s.policy.Push(req.task)
+		queued := newQueuedTask(req.task, now)
+		nq := queued.NQ()
+		newTaskAdded, err := s.policy.Push(queued)
 		if err == nil {
 			s.updateWaitingTaskCounter(int64(newTaskAdded), nq)
 		}
@@ -189,19 +215,23 @@ func (s *scheduler) handleAddTaskRequest(req addTaskReq, maxWaitTaskNum int64) b
 	}
 
 	// Continue processing if the queue isn't reach the max limit.
-	return s.GetWaitingTaskTotal() < maxWaitTaskNum
+	return maxWaitTaskNum <= 0 || s.GetWaitingTaskTotal() < maxWaitTaskNum
 }
 
 // produceExecChan produce task from scheduler into exec chan as much as possible
-func (s *scheduler) produceExecChan() Task {
-	var task Task
+func (s *scheduler) produceExecChan(now time.Time) *queuedTask {
+	var task *queuedTask
 	for {
 		var execChan chan Task
+		var execTask Task
 		nq := int64(0)
-		task, nq, execChan = s.setupExecListener(task)
+		task, nq, execChan = s.setupExecListener(task, now)
+		if task.valid() {
+			execTask = task.Task
+		}
 
 		select {
-		case execChan <- task:
+		case execChan <- execTask:
 			// Update waiting task counter.
 			s.updateWaitingTaskCounter(-1, -nq)
 			// Task sent, drop the ownership of sent task.
@@ -223,7 +253,7 @@ func (s *scheduler) exec() {
 			return
 		}
 		// Skip this task if task is canceled.
-		if err := t.Canceled(); err != nil {
+		if err := t.Context().Err(); err != nil {
 			log.Warn("task canceled before executing", zap.Error(err))
 			t.Done(err)
 			continue
@@ -239,7 +269,12 @@ func (s *scheduler) exec() {
 			metrics.QueryNodeReadTaskConcurrency.WithLabelValues(paramtable.GetStringNodeID()).Inc()
 			collector.Counter.Inc(metricsinfo.ExecuteQueueType)
 
+			executeStart := time.Now()
 			err := t.Execute()
+			metrics.QueryNodeReadTaskExecuteDuration.WithLabelValues(
+				paramtable.GetStringNodeID(),
+				readTaskExecuteOutcome(err),
+			).Observe(float64(time.Since(executeStart).Microseconds()) / 1000.0)
 
 			// Update all metric after task finished.
 			metrics.QueryNodeReadTaskConcurrency.WithLabelValues(paramtable.GetStringNodeID()).Dec()
@@ -260,15 +295,39 @@ func (s *scheduler) getPool(t Task) *conc.Pool[any] {
 	return s.pool
 }
 
+func readTaskExecuteOutcome(err error) string {
+	if err == nil {
+		return metrics.SuccessLabel
+	}
+	if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
+		return metrics.CancelLabel
+	}
+	return metrics.FailLabel
+}
+
 // setupExecListener setup the execChan and next task to run.
-func (s *scheduler) setupExecListener(lastWaitingTask Task) (Task, int64, chan Task) {
+func (s *scheduler) setupExecListener(lastWaitingTask *queuedTask, now time.Time) (*queuedTask, int64, chan Task) {
 	var execChan chan Task
 	nq := int64(0)
-	if lastWaitingTask == nil {
+	if !lastWaitingTask.valid() {
 		// No task is waiting to send to execChan, schedule a new one from queue.
-		lastWaitingTask = s.policy.Pop()
+		for {
+			lastWaitingTask = s.policy.Pop(now)
+			if !lastWaitingTask.valid() {
+				break
+			}
+			if err := lastWaitingTask.Context().Err(); err != nil {
+				s.updateWaitingTaskCounter(-1, -lastWaitingTask.NQ())
+				s.recordReadTaskQueueDuration(lastWaitingTask, now, readTaskQueueOutcomeExpired)
+				lastWaitingTask.Done(err)
+				lastWaitingTask = nil
+				continue
+			}
+			s.recordReadTaskQueueDuration(lastWaitingTask, now, readTaskQueueOutcomeScheduled)
+			break
+		}
 	}
-	if lastWaitingTask != nil {
+	if lastWaitingTask.valid() {
 		// Try to sent task to execChan if there is a task ready to run.
 		execChan = s.execChan
 		nq = lastWaitingTask.NQ()
@@ -277,6 +336,17 @@ func (s *scheduler) setupExecListener(lastWaitingTask Task) (Task, int64, chan T
 	return lastWaitingTask, nq, execChan
 }
 
+func (s *scheduler) cleanupExpiredTasks(now time.Time) {
+	deadlineAdvance := paramtable.Get().QueryNodeCfg.SchedulePolicyTaskDeadlineAdvance.GetAsDurationByParse()
+	cleanupTime := now.Add(deadlineAdvance)
+	tasks := s.policy.Cleanup(cleanupTime)
+	for _, task := range tasks {
+		s.updateWaitingTaskCounter(-1, -task.NQ())
+		s.recordReadTaskQueueDuration(task, now, readTaskQueueOutcomeExpired)
+		task.Done(cleanupTaskError(task))
+	}
+}
+
 // setupReadyLenMetric update the read task ready len metric.
 func (s *scheduler) setupReadyLenMetric() {
 	waitingTaskCount := s.GetWaitingTaskTotal()
@@ -285,6 +355,17 @@ func (s *scheduler) setupReadyLenMetric() {
 	collector.Counter.Set(metricsinfo.ReadyQueueType, waitingTaskCount)
 	// Record the waiting task length of policy as ready task metric.
 	metrics.QueryNodeReadTaskReadyLen.WithLabelValues(paramtable.GetStringNodeID()).Set(float64(waitingTaskCount))
+	metrics.QueryNodeReadTaskReadyNQ.WithLabelValues(paramtable.GetStringNodeID()).Set(float64(s.GetWaitingTaskTotalNQ()))
+}
+
+func (s *scheduler) recordReadTaskQueueDuration(task *queuedTask, now time.Time, outcome string) {
+	if !task.valid() {
+		return
+	}
+	metrics.QueryNodeReadTaskQueueDuration.WithLabelValues(
+		paramtable.GetStringNodeID(),
+		outcome,
+	).Observe(float64(task.queueDuration(now).Microseconds()) / 1000.0)
 }
 
 // scheduler counter implement, concurrent safe.

internal/util/searchutil/scheduler/concurrent_safe_scheduler_test.go

@@ -7,12 +7,16 @@ import (
 	"testing"
 	"time"
 
+	"github.com/cockroachdb/errors"
+	dto "github.com/prometheus/client_model/go"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/suite"
 	"go.uber.org/atomic"
 
+	"github.com/milvus-io/milvus/pkg/v3/metrics"
 	"github.com/milvus-io/milvus/pkg/v3/util/conc"
 	"github.com/milvus-io/milvus/pkg/v3/util/lifetime"
+	"github.com/milvus-io/milvus/pkg/v3/util/merr"
 	"github.com/milvus-io/milvus/pkg/v3/util/paramtable"
 )
 
@@ -133,7 +137,333 @@ func (s *SchedulerSuite) TestConsumeRecvChan() {
 			scheduler.consumeRecvChan(addTaskReq{
 				task: task,
 				err:  make(chan error, 1),
-			}, maxReceiveChanBatchConsumeNum)
+			}, maxReceiveChanBatchConsumeNum, time.Now())
 		})
 	})
 }
+
+func (s *SchedulerSuite) TestConsumeRecvChanUsesLoopTimestampForBatch() {
+	now := time.Now()
+	scheduler := &scheduler{
+		policy:           newFIFOPolicy(),
+		receiveChan:      make(chan addTaskReq, 1),
+		schedulerCounter: schedulerCounter{},
+	}
+
+	firstErrCh := make(chan error, 1)
+	secondErrCh := make(chan error, 1)
+	secondTask := newMockTask(mockTaskConfig{nq: 1})
+	scheduler.receiveChan <- addTaskReq{
+		task: secondTask,
+		err:  secondErrCh,
+	}
+
+	scheduler.consumeRecvChan(addTaskReq{
+		task: newMockTask(mockTaskConfig{nq: 1}),
+		err:  firstErrCh,
+	}, 2, now)
+
+	s.NoError(<-firstErrCh)
+	s.NoError(<-secondErrCh)
+
+	first := scheduler.policy.Pop(now)
+	second := scheduler.policy.Pop(now)
+	s.True(first.valid())
+	s.True(second.valid())
+	s.Equal(now, first.enqueueTime)
+	s.Equal(now, second.enqueueTime)
+}
+
+func (s *SchedulerSuite) TestHandleAddTaskRequestRejectsWhenWaitingQueueFull() {
+	scheduler := &scheduler{
+		policy:           newFIFOPolicy(),
+		schedulerCounter: schedulerCounter{},
+	}
+
+	errCh := make(chan error, 1)
+	keepConsuming := scheduler.handleAddTaskRequest(addTaskReq{
+		task: newMockTask(mockTaskConfig{nq: 1}),
+		err:  errCh,
+	}, 1, time.Now())
+	s.False(keepConsuming)
+	s.NoError(<-errCh)
+	s.Equal(int64(1), scheduler.GetWaitingTaskTotal())
+
+	errCh = make(chan error, 1)
+	keepConsuming = scheduler.handleAddTaskRequest(addTaskReq{
+		task: newMockTask(mockTaskConfig{nq: 1}),
+		err:  errCh,
+	}, 1, time.Now())
+	s.False(keepConsuming)
+	s.ErrorIs(<-errCh, merr.ErrServiceTooManyRequests)
+	s.Equal(int64(1), scheduler.GetWaitingTaskTotal())
+}
+
+func (s *SchedulerSuite) TestHandleAddTaskRequestCleansExpiredTasksBeforeQueueLimit() {
+	now := time.Now()
+	scheduler := &scheduler{
+		policy:           newFIFOPolicy(),
+		schedulerCounter: schedulerCounter{},
+	}
+
+	expiredCtx, cancelExpired := context.WithDeadline(context.Background(), now.Add(-time.Millisecond))
+	defer cancelExpired()
+	expiredTask := newMockTask(mockTaskConfig{ctx: expiredCtx, nq: 1})
+	queued := newQueuedTask(expiredTask, now.Add(-time.Second))
+	added, err := scheduler.policy.Push(queued)
+	s.NoError(err)
+	scheduler.updateWaitingTaskCounter(int64(added), queued.NQ())
+
+	errCh := make(chan error, 1)
+	keepConsuming := scheduler.handleAddTaskRequest(addTaskReq{
+		task: newMockTask(mockTaskConfig{nq: 1}),
+		err:  errCh,
+	}, 1, now)
+
+	s.False(keepConsuming)
+	s.NoError(<-errCh)
+	s.ErrorIs(expiredTask.Wait(), context.DeadlineExceeded)
+	s.Equal(int64(1), scheduler.GetWaitingTaskTotal())
+}
+
+func (s *SchedulerSuite) TestHandleAddTaskRequestSkipsCleanupBeforeQueueFull() {
+	now := time.Now()
+	scheduler := &scheduler{
+		policy:           newFIFOPolicy(),
+		schedulerCounter: schedulerCounter{},
+	}
+
+	expiredCtx, cancelExpired := context.WithDeadline(context.Background(), now.Add(-time.Millisecond))
+	defer cancelExpired()
+	expiredTask := newMockTask(mockTaskConfig{ctx: expiredCtx, nq: 1})
+	queued := newQueuedTask(expiredTask, now.Add(-time.Second))
+	added, err := scheduler.policy.Push(queued)
+	s.NoError(err)
+	scheduler.updateWaitingTaskCounter(int64(added), queued.NQ())
+
+	errCh := make(chan error, 1)
+	keepConsuming := scheduler.handleAddTaskRequest(addTaskReq{
+		task: newMockTask(mockTaskConfig{nq: 1}),
+		err:  errCh,
+	}, 2, now)
+
+	s.False(keepConsuming)
+	s.NoError(<-errCh)
+	s.Equal(int64(2), scheduler.GetWaitingTaskTotal())
+	s.Equal(0, len(expiredTask.(*MockTask).notifier))
+}
+
+func (s *SchedulerSuite) TestHandleAddTaskRequestCleansTasksNearDeadlineBeforeQueueLimit() {
+	paramtable.Init()
+	old := paramtable.Get().QueryNodeCfg.SchedulePolicyTaskDeadlineAdvance.SwapTempValue("50ms")
+	defer paramtable.Get().QueryNodeCfg.SchedulePolicyTaskDeadlineAdvance.SwapTempValue(old)
+
+	now := time.Now()
+	scheduler := &scheduler{
+		policy:           newFIFOPolicy(),
+		schedulerCounter: schedulerCounter{},
+	}
+
+	ctx, cancel := context.WithDeadline(context.Background(), now.Add(30*time.Millisecond))
+	defer cancel()
+	nearDeadlineTask := newMockTask(mockTaskConfig{ctx: ctx, nq: 1})
+	queued := newQueuedTask(nearDeadlineTask, now.Add(-time.Second))
+	added, err := scheduler.policy.Push(queued)
+	s.NoError(err)
+	scheduler.updateWaitingTaskCounter(int64(added), queued.NQ())
+
+	errCh := make(chan error, 1)
+	keepConsuming := scheduler.handleAddTaskRequest(addTaskReq{
+		task: newMockTask(mockTaskConfig{nq: 1}),
+		err:  errCh,
+	}, 1, now)
+
+	s.False(keepConsuming)
+	s.NoError(<-errCh)
+	s.ErrorIs(nearDeadlineTask.Wait(), context.DeadlineExceeded)
+	s.Equal(int64(1), scheduler.GetWaitingTaskTotal())
+}
+
+func (s *SchedulerSuite) TestAddReturnsContextErrorWhenReceiveBlocks() {
+	paramtable.Init()
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Millisecond)
+	defer cancel()
+
+	scheduler := &scheduler{
+		policy:           newFIFOPolicy(),
+		receiveChan:      make(chan addTaskReq),
+		schedulerCounter: schedulerCounter{},
+		lifetime:         lifetime.NewLifetime(lifetime.Working),
+	}
+
+	err := scheduler.Add(newMockTask(mockTaskConfig{ctx: ctx, nq: 1}))
+	s.ErrorIs(err, context.DeadlineExceeded)
+}
+
+func (s *SchedulerSuite) TestHandleAddTaskRequestDoesNotRejectByQueueDelayDeadline() {
+	now := time.Now()
+	scheduler := &scheduler{
+		policy:           newFIFOPolicy(),
+		schedulerCounter: schedulerCounter{},
+	}
+
+	queued := newQueuedTask(newMockTask(mockTaskConfig{nq: 1}), now.Add(-time.Second))
+	newTaskAdded, err := scheduler.policy.Push(queued)
+	s.NoError(err)
+	scheduler.updateWaitingTaskCounter(int64(newTaskAdded), queued.NQ())
+
+	ctx, cancel := context.WithDeadline(context.Background(), now.Add(100*time.Millisecond))
+	defer cancel()
+
+	errCh := make(chan error, 1)
+	keepConsuming := scheduler.handleAddTaskRequest(addTaskReq{
+		task: newMockTask(mockTaskConfig{ctx: ctx, nq: 1}),
+		err:  errCh,
+	}, 0, now)
+
+	s.True(keepConsuming)
+	s.NoError(<-errCh)
+	s.Equal(int64(2), scheduler.GetWaitingTaskTotal())
+}
+
+func (s *SchedulerSuite) TestHandleAddTaskRequestAcceptsDeadlineWhenQueueEmpty() {
+	now := time.Now()
+	scheduler := &scheduler{
+		policy:           newFIFOPolicy(),
+		schedulerCounter: schedulerCounter{},
+	}
+
+	ctx, cancel := context.WithDeadline(context.Background(), now.Add(100*time.Millisecond))
+	defer cancel()
+
+	errCh := make(chan error, 1)
+	keepConsuming := scheduler.handleAddTaskRequest(addTaskReq{
+		task: newMockTask(mockTaskConfig{ctx: ctx, nq: 1}),
+		err:  errCh,
+	}, 0, now)
+
+	s.True(keepConsuming)
+	s.NoError(<-errCh)
+	s.Equal(int64(1), scheduler.GetWaitingTaskTotal())
+}
+
+func (s *SchedulerSuite) TestSetupExecListenerRecordsPoppedExpiredTask() {
+	paramtable.Init()
+	metrics.QueryNodeReadTaskQueueDuration.Reset()
+	defer metrics.QueryNodeReadTaskQueueDuration.Reset()
+
+	now := time.Now()
+	scheduler := &scheduler{
+		policy:           newFIFOPolicy(),
+		execChan:         make(chan Task),
+		schedulerCounter: schedulerCounter{},
+	}
+
+	expiredCtx, cancelExpired := context.WithDeadline(context.Background(), now.Add(-time.Millisecond))
+	defer cancelExpired()
+	expiredTask := newMockTask(mockTaskConfig{ctx: expiredCtx, nq: 1})
+	queued := newQueuedTask(expiredTask, now.Add(-time.Second))
+	added, err := scheduler.policy.Push(queued)
+	s.NoError(err)
+	scheduler.updateWaitingTaskCounter(int64(added), queued.NQ())
+
+	task, nq, execChan := scheduler.setupExecListener(nil, now)
+
+	s.False(task.valid())
+	s.Zero(nq)
+	s.Nil(execChan)
+	s.Equal(int64(0), scheduler.GetWaitingTaskTotal())
+	s.ErrorIs(expiredTask.Wait(), context.DeadlineExceeded)
+	s.Equal(uint64(1), readTaskQueueDurationCount(readTaskQueueOutcomeExpired))
+	s.Equal(uint64(0), readTaskQueueDurationCount(readTaskQueueOutcomeScheduled))
+}
+
+func (s *SchedulerSuite) TestExecRecordsReadTaskExecuteDuration() {
+	paramtable.Init()
+	metrics.QueryNodeReadTaskExecuteDuration.Reset()
+	defer metrics.QueryNodeReadTaskExecuteDuration.Reset()
+
+	scheduler := newScheduler(newFIFOPolicy())
+	scheduler.Start()
+	defer scheduler.Stop()
+
+	successTask := newMockTask(mockTaskConfig{
+		executeCost: time.Millisecond,
+		execution: func(ctx context.Context) error {
+			return nil
+		},
+	})
+	s.NoError(scheduler.Add(successTask))
+	s.NoError(successTask.(*MockTask).Wait())
+
+	expectedErr := errors.New("mock execute failure")
+	failedTask := newMockTask(mockTaskConfig{
+		executeCost: time.Millisecond,
+		execution: func(ctx context.Context) error {
+			return expectedErr
+		},
+	})
+	s.NoError(scheduler.Add(failedTask))
+	s.ErrorIs(failedTask.(*MockTask).Wait(), expectedErr)
+
+	canceledTask := newMockTask(mockTaskConfig{
+		executeCost: time.Millisecond,
+		execution: func(ctx context.Context) error {
+			return context.DeadlineExceeded
+		},
+	})
+	s.NoError(scheduler.Add(canceledTask))
+	s.ErrorIs(canceledTask.(*MockTask).Wait(), context.DeadlineExceeded)
+
+	s.Equal(uint64(1), readTaskExecuteDurationCount(metrics.SuccessLabel))
+	s.Equal(uint64(1), readTaskExecuteDurationCount(metrics.FailLabel))
+	s.Equal(uint64(1), readTaskExecuteDurationCount(metrics.CancelLabel))
+}
+
+func (s *SchedulerSuite) TestQueuedTaskTimingHelpers() {
+	now := time.Now()
+	invalid := &queuedTask{}
+	s.Zero(invalid.queueDuration(now))
+	s.False(invalid.cleanupReady(now))
+
+	taskWithoutEnqueueTime := newQueuedTask(newMockTask(mockTaskConfig{nq: 1}), time.Time{})
+	s.Zero(taskWithoutEnqueueTime.queueDuration(now))
+	s.False(taskWithoutEnqueueTime.cleanupReady(now))
+
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+	canceledTask := newQueuedTask(newMockTask(mockTaskConfig{ctx: ctx, nq: 1}), now.Add(-time.Millisecond))
+	s.True(canceledTask.cleanupReady(now))
+}
+
+func (s *SchedulerSuite) TestRecordReadTaskQueueDurationSkipsInvalidTask() {
+	paramtable.Init()
+	metrics.QueryNodeReadTaskQueueDuration.Reset()
+	defer metrics.QueryNodeReadTaskQueueDuration.Reset()
+
+	scheduler := &scheduler{}
+	scheduler.recordReadTaskQueueDuration(&queuedTask{}, time.Now(), readTaskQueueOutcomeScheduled)
+
+	observer := metrics.QueryNodeReadTaskQueueDuration.WithLabelValues(paramtable.GetStringNodeID(), readTaskQueueOutcomeScheduled)
+	metric := &dto.Metric{}
+	s.NoError(observer.(interface{ Write(*dto.Metric) error }).Write(metric))
+	s.Equal(uint64(0), metric.GetHistogram().GetSampleCount())
+}
+
+func readTaskExecuteDurationCount(outcome string) uint64 {
+	observer := metrics.QueryNodeReadTaskExecuteDuration.WithLabelValues(paramtable.GetStringNodeID(), outcome)
+	metric := &dto.Metric{}
+	if err := observer.(interface{ Write(*dto.Metric) error }).Write(metric); err != nil {
+		return 0
+	}
+	return metric.GetHistogram().GetSampleCount()
+}
+
+func readTaskQueueDurationCount(outcome string) uint64 {
+	observer := metrics.QueryNodeReadTaskQueueDuration.WithLabelValues(paramtable.GetStringNodeID(), outcome)
+	metric := &dto.Metric{}
+	if err := observer.(interface{ Write(*dto.Metric) error }).Write(metric); err != nil {
+		return 0
+	}
+	return metric.GetHistogram().GetSampleCount()
+}

internal/util/searchutil/scheduler/fifo_policy.go

@@ -1,6 +1,8 @@
 package scheduler
 
 import (
+	"time"
+
 	"github.com/milvus-io/milvus/pkg/v3/util/paramtable"
 )
 
@@ -18,14 +20,20 @@ type fifoPolicy struct {
 	queue *mergeTaskQueue
 }
 
+func (p *fifoPolicy) Cleanup(now time.Time) []*queuedTask {
+	return p.queue.cleanup(now)
+}
+
 // Push add a new task into scheduler, an error will be returned if scheduler reaches some limit.
-func (p *fifoPolicy) Push(task Task) (int, error) {
+func (p *fifoPolicy) Push(task *queuedTask) (int, error) {
 	pt := paramtable.Get()
 
 	// Try to merge task if task can merge.
-	if t := tryIntoMergeTask(task); t != nil {
+	if t := tryIntoMergeTask(task.Task); t != nil {
 		maxNQ := pt.QueryNodeCfg.MaxGroupNQ.GetAsInt64()
-		if p.queue.tryMerge(t, maxNQ) {
+		nqMergeRatio := pt.QueryNodeCfg.NQMergeRatio.GetAsFloat()
+		maxDeadlineMergeGap := pt.QueryNodeCfg.MaxDeadlineMergeGap.GetAsDurationByParse()
+		if p.queue.tryMerge(task, maxNQ, nqMergeRatio, maxDeadlineMergeGap) {
 			return 0, nil
 		}
 	}
@@ -36,10 +44,8 @@ func (p *fifoPolicy) Push(task Task) (int, error) {
 }
 
 // Pop get the task next ready to run.
-func (p *fifoPolicy) Pop() Task {
-	task := p.queue.front()
-	p.queue.pop()
-	return task
+func (p *fifoPolicy) Pop(now time.Time) *queuedTask {
+	return p.queue.pop()
 }
 
 // Len get ready task counts.

internal/util/searchutil/scheduler/mock_task_test.go

@@ -39,6 +39,7 @@ func newMockTask(c mockTaskConfig) Task {
 		notifier:    make(chan error, 1),
 		mergeAble:   c.mergeAble,
 		nq:          c.nq,
+		minNQ:       c.nq,
 		username:    c.username,
 		execution:   c.execution,
 		tr:          timerecord.NewTimeRecorderWithTrace(c.ctx, "searchTask"),
@@ -51,6 +52,7 @@ type MockTask struct {
 	notifier    chan error
 	mergeAble   bool
 	nq          int64
+	minNQ       int64
 	username    string
 	execution   func(ctx context.Context) error
 	tr          *timerecord.TimeRecorder
@@ -90,10 +92,6 @@ func (t *MockTask) Done(err error) {
 	t.notifier <- err
 }
 
-func (t *MockTask) Canceled() error {
-	return t.ctx.Err()
-}
-
 func (t *MockTask) Wait() error {
 	return <-t.notifier
 }
@@ -108,6 +106,9 @@ func (t *MockTask) MergeWith(t2 Task) bool {
 	case *MockTask:
 		if t.mergeAble && t2.mergeAble {
 			t.nq += t2.nq
+			if t2.MinNQ() < t.minNQ {
+				t.minNQ = t2.MinNQ()
+			}
 			t.executeCost += t2.executeCost
 			return true
 		}
@@ -122,3 +123,7 @@ func (t *MockTask) SearchResult() *internalpb.SearchResults {
 func (t *MockTask) NQ() int64 {
 	return t.nq
 }
+
+func (t *MockTask) MinNQ() int64 {
+	return t.minNQ
+}

internal/util/searchutil/scheduler/policy_test.go

@@ -1,8 +1,10 @@
 package scheduler
 
 import (
+	"context"
 	"fmt"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 
@@ -20,6 +22,54 @@ func TestFIFOPolicy(t *testing.T) {
 	testCommonPolicyOperation(t, newFIFOPolicy())
 }
 
+func TestPolicyCleanupExpiredTasks(t *testing.T) {
+	paramtable.Init()
+	for name, policy := range map[string]schedulePolicy{
+		"fifo":              newFIFOPolicy(),
+		"user-task-polling": newUserTaskPollingPolicy(),
+	} {
+		t.Run(name, func(t *testing.T) {
+			base := time.Now()
+			ctx, cancel := context.WithDeadline(context.Background(), base.Add(10*time.Millisecond))
+			defer cancel()
+
+			added, err := policy.Push(newQueuedTask(newMockTask(mockTaskConfig{ctx: ctx, nq: 1}), base))
+			assert.NoError(t, err)
+			assert.Equal(t, 1, added)
+			assert.Equal(t, 1, policy.Len())
+
+			expired := policy.Cleanup(base.Add(20 * time.Millisecond))
+			assert.Len(t, expired, 1)
+			assert.Equal(t, 0, policy.Len())
+			assert.False(t, policy.Pop(base.Add(20*time.Millisecond)).valid())
+		})
+	}
+}
+
+func TestPolicyCleanupCanceledTasks(t *testing.T) {
+	paramtable.Init()
+	for name, policy := range map[string]schedulePolicy{
+		"fifo":              newFIFOPolicy(),
+		"user-task-polling": newUserTaskPollingPolicy(),
+	} {
+		t.Run(name, func(t *testing.T) {
+			base := time.Now()
+			ctx, cancel := context.WithCancel(context.Background())
+
+			added, err := policy.Push(newQueuedTask(newMockTask(mockTaskConfig{ctx: ctx, nq: 1}), base))
+			assert.NoError(t, err)
+			assert.Equal(t, 1, added)
+
+			cancel()
+			removed := policy.Cleanup(base)
+
+			assert.Len(t, removed, 1)
+			assert.Equal(t, 0, policy.Len())
+			assert.False(t, policy.Pop(base).valid())
+		})
+	}
+}
+
 func testCrossUserMerge(t *testing.T, policy schedulePolicy) {
 	userN := 10
 	maxNQ := paramtable.Get().QueryNodeCfg.MaxGroupNQ.GetAsInt64()
@@ -32,17 +82,20 @@ func testCrossUserMerge(t *testing.T, policy schedulePolicy) {
 			nq:        maxNQ / 2,
 			mergeAble: true,
 		})
-		policy.Push(task)
+		policy.Push(newQueuedTask(task, time.Now()))
 	}
 	nAfterMerge := n / 2
 	assert.Equal(t, nAfterMerge, policy.Len())
 	for i := 1; i <= nAfterMerge; i++ {
-		assert.NotNil(t, policy.Pop())
+		assert.True(t, policy.Pop(time.Now()).valid())
 		assert.Equal(t, nAfterMerge-i, policy.Len())
 	}
 
 	// Open cross user grouping
-	paramtable.Get().QueryNodeCfg.SchedulePolicyEnableCrossUserGrouping.SwapTempValue("true")
+	oldGrouping := paramtable.Get().QueryNodeCfg.SchedulePolicyEnableCrossUserGrouping.SwapTempValue("true")
+	defer paramtable.Get().QueryNodeCfg.SchedulePolicyEnableCrossUserGrouping.SwapTempValue(oldGrouping)
+	oldNQMergeRatio := paramtable.Get().QueryNodeCfg.NQMergeRatio.SwapTempValue("0")
+	defer paramtable.Get().QueryNodeCfg.NQMergeRatio.SwapTempValue(oldNQMergeRatio)
 	for i := 1; i <= n; i++ {
 		username := fmt.Sprintf("user_%d", (i-1)%userN)
 		task := newMockTask(mockTaskConfig{
@@ -50,12 +103,12 @@ func testCrossUserMerge(t *testing.T, policy schedulePolicy) {
 			nq:        maxNQ / 4,
 			mergeAble: true,
 		})
-		policy.Push(task)
+		policy.Push(newQueuedTask(task, time.Now()))
 	}
 	nAfterMerge = n / 4
 	assert.Equal(t, nAfterMerge, policy.Len())
 	for i := 1; i <= nAfterMerge; i++ {
-		assert.NotNil(t, policy.Pop())
+		assert.True(t, policy.Pop(time.Now()).valid())
 		assert.Equal(t, nAfterMerge-i, policy.Len())
 	}
 }
@@ -64,7 +117,7 @@ func testCrossUserMerge(t *testing.T, policy schedulePolicy) {
 func testCommonPolicyOperation(t *testing.T, policy schedulePolicy) {
 	// Empty policy assertion.
 	assert.Equal(t, 0, policy.Len())
-	assert.Nil(t, policy.Pop())
+	assert.False(t, policy.Pop(time.Now()).valid())
 	assert.Equal(t, 0, policy.Len())
 
 	// Test no merge push pop.
@@ -76,12 +129,12 @@ func testCommonPolicyOperation(t *testing.T, policy schedulePolicy) {
 		task := newMockTask(mockTaskConfig{
 			username: username,
 		})
-		policy.Push(task)
+		policy.Push(newQueuedTask(task, time.Now()))
 		assert.Equal(t, i, policy.Len())
 	}
 	// Test Pop
 	for i := 1; i <= n; i++ {
-		assert.NotNil(t, policy.Pop())
+		assert.True(t, policy.Pop(time.Now()).valid())
 		assert.Equal(t, n-i, policy.Len())
 	}
 
@@ -95,11 +148,11 @@ func testCommonPolicyOperation(t *testing.T, policy schedulePolicy) {
 			nq:        maxNQ,
 			mergeAble: true,
 		})
-		policy.Push(task)
+		policy.Push(newQueuedTask(task, time.Now()))
 	}
 	assert.Equal(t, n, policy.Len())
 	for i := 1; i <= n; i++ {
-		assert.NotNil(t, policy.Pop())
+		assert.True(t, policy.Pop(time.Now()).valid())
 		assert.Equal(t, n-i, policy.Len())
 	}
 
@@ -112,12 +165,12 @@ func testCommonPolicyOperation(t *testing.T, policy schedulePolicy) {
 			nq:        maxNQ / 2,
 			mergeAble: true,
 		})
-		policy.Push(task)
+		policy.Push(newQueuedTask(task, time.Now()))
 	}
 	nAfterMerge := n / 2
 	assert.Equal(t, nAfterMerge, policy.Len())
 	for i := 1; i <= nAfterMerge; i++ {
-		assert.NotNil(t, policy.Pop())
+		assert.True(t, policy.Pop(time.Now()).valid())
 		assert.Equal(t, nAfterMerge-i, policy.Len())
 	}
 }

internal/util/searchutil/scheduler/queues.go

@@ -8,43 +8,102 @@ import (
 func newMergeTaskQueue(group string) *mergeTaskQueue {
 	return &mergeTaskQueue{
 		name:             group,
-		tasks:            make([]Task, 0),
+		tasks:            make([]*queuedTask, 0),
 		cleanupTimestamp: time.Now(),
 	}
 }
 
 type mergeTaskQueue struct {
 	name             string
-	tasks            []Task
+	tasks            []*queuedTask
+	count            int
 	cleanupTimestamp time.Time
 }
 
 // len returns the length of taskQueue.
 func (q *mergeTaskQueue) len() int {
-	return len(q.tasks)
+	return q.count
 }
 
 // push add a new task to the end of taskQueue.
-func (q *mergeTaskQueue) push(t Task) {
-	q.tasks = append(q.tasks, t)
+func (q *mergeTaskQueue) push(task *queuedTask) {
+	q.tasks = append(q.tasks, task)
+	q.count++
 }
 
-// front returns the first element of taskQueue,
-// returns nil if task queue is empty.
-func (q *mergeTaskQueue) front() Task {
-	if q.len() > 0 {
+// front returns the first element of taskQueue.
+func (q *mergeTaskQueue) front() *queuedTask {
+	q.dropRemovedPrefix()
+	if len(q.tasks) > 0 {
 		return q.tasks[0]
 	}
 	return nil
 }
 
-// pop pops the first element of taskQueue,
-func (q *mergeTaskQueue) pop() {
-	if q.len() > 0 {
-		q.tasks = q.tasks[1:]
-		if q.len() == 0 {
-			q.cleanupTimestamp = time.Now()
+// pop pops the first element of taskQueue.
+func (q *mergeTaskQueue) pop() *queuedTask {
+	for {
+		q.dropRemovedPrefix()
+		if len(q.tasks) == 0 {
+			return nil
 		}
+
+		task := q.tasks[0]
+		q.tasks = q.tasks[1:]
+		if !task.valid() {
+			continue
+		}
+
+		removed := q.markRemoved(task, time.Now())
+		if q.len() == 0 {
+			clear(q.tasks)
+			q.tasks = nil
+		}
+		return removed
+	}
+}
+
+func (q *mergeTaskQueue) cleanup(now time.Time) []*queuedTask {
+	if q.len() == 0 {
+		return nil
+	}
+
+	removed := make([]*queuedTask, 0)
+	for _, task := range q.tasks {
+		if !task.cleanupReady(now) {
+			continue
+		}
+		removed = append(removed, q.markRemoved(task, now))
+	}
+
+	if q.len() == 0 {
+		clear(q.tasks)
+		q.tasks = nil
+	}
+	return removed
+}
+
+func (q *mergeTaskQueue) markRemoved(task *queuedTask, now time.Time) *queuedTask {
+	if !task.valid() {
+		return nil
+	}
+	removed := *task
+	task.Task = nil
+	q.count--
+	if q.count == 0 {
+		q.cleanupTimestamp = now
+	}
+	return &removed
+}
+
+func (q *mergeTaskQueue) dropRemovedPrefix() {
+	for len(q.tasks) > 0 {
+		task := q.tasks[0]
+		if task.valid() {
+			return
+		}
+		q.tasks[0] = nil
+		q.tasks = q.tasks[1:]
 	}
 }
 
@@ -59,17 +118,59 @@ func (q *mergeTaskQueue) expire(d time.Duration) bool {
 	return false
 }
 
-// tryMerge try to a new task to any task in queue.
-func (q *mergeTaskQueue) tryMerge(task MergeTask, maxNQ int64) bool {
-	nqRest := maxNQ - task.NQ()
-	// No need to perform any merge if task.nq is greater than maxNQ.
-	if nqRest <= 0 {
+func canMergeNQ(task MergeTask, other MergeTask, maxNQ int64, nqMergeRatio float64) bool {
+	totalNQ := task.NQ() + other.NQ()
+	if totalNQ > maxNQ {
 		return false
 	}
-	for i := q.len() - 1; i >= 0; i-- {
-		if taskInQueue := tryIntoMergeTask(q.tasks[i]); taskInQueue != nil {
+	if nqMergeRatio <= 0 {
+		return true
+	}
+	minNQ := task.MinNQ()
+	if otherMinNQ := other.MinNQ(); otherMinNQ < minNQ {
+		minNQ = otherMinNQ
+	}
+	return minNQ > 0 && float64(totalNQ)/float64(minNQ) <= nqMergeRatio
+}
+
+func canMergeDeadline(task *queuedTask, other *queuedTask, maxDeadlineMergeGap time.Duration) bool {
+	if maxDeadlineMergeGap < 0 {
+		return true
+	}
+	deadline, ok := task.Context().Deadline()
+	otherDeadline, otherOk := other.Context().Deadline()
+	if !ok && !otherOk {
+		return true
+	}
+	if ok != otherOk {
+		return false
+	}
+	if deadline.After(otherDeadline) {
+		deadline, otherDeadline = otherDeadline, deadline
+	}
+	return otherDeadline.Sub(deadline) <= maxDeadlineMergeGap
+}
+
+// tryMerge try to a new task to any task in queue.
+func (q *mergeTaskQueue) tryMerge(task *queuedTask, maxNQ int64, nqMergeRatio float64, maxDeadlineMergeGap time.Duration) bool {
+	mergeTask := tryIntoMergeTask(task.Task)
+	if mergeTask == nil {
+		return false
+	}
+	// No need to perform any merge if task.nq is greater than maxNQ.
+	if mergeTask.NQ() >= maxNQ {
+		return false
+	}
+	for i := len(q.tasks) - 1; i >= 0; i-- {
+		taskInQueue := q.tasks[i]
+		if !taskInQueue.valid() {
+			continue
+		}
+		if taskInQueue := tryIntoMergeTask(taskInQueue.Task); taskInQueue != nil {
 			// Try to merge it if limit of nq is enough.
-			if taskInQueue.NQ() <= nqRest && taskInQueue.MergeWith(task) {
+			if (canMergeNQ(taskInQueue, mergeTask, maxNQ, nqMergeRatio) &&
+				canMergeDeadline(q.tasks[i], task, maxDeadlineMergeGap)) &&
+				taskInQueue.MergeWith(mergeTask) {
 				return true
 			}
 		}
@@ -107,7 +208,7 @@ func (q *fairPollingTaskQueue) groupLen(group string) int {
 }
 
 // tryMergeWithOtherGroup try to merge given task into exists tasks in the other group.
-func (q *fairPollingTaskQueue) tryMergeWithOtherGroup(group string, task MergeTask, maxNQ int64) bool {
+func (q *fairPollingTaskQueue) tryMergeWithOtherGroup(group string, task *queuedTask, maxNQ int64, nqMergeRatio float64, maxDeadlineMergeGap time.Duration) bool {
 	if q.count == 0 {
 		return false
 	}
@@ -120,7 +221,7 @@ func (q *fairPollingTaskQueue) tryMergeWithOtherGroup(group string, task MergeTa
 		if queue.len() == 0 || queue.name == group {
 			continue
 		}
-		if queue.tryMerge(task, maxNQ) {
+		if queue.tryMerge(task, maxNQ, nqMergeRatio, maxDeadlineMergeGap) {
 			return true
 		}
 		node = prev
@@ -129,14 +230,14 @@ func (q *fairPollingTaskQueue) tryMergeWithOtherGroup(group string, task MergeTa
 }
 
 // tryMergeWithSameGroup try to merge given task into exists tasks in the same group.
-func (q *fairPollingTaskQueue) tryMergeWithSameGroup(group string, task MergeTask, maxNQ int64) bool {
+func (q *fairPollingTaskQueue) tryMergeWithSameGroup(group string, task *queuedTask, maxNQ int64, nqMergeRatio float64, maxDeadlineMergeGap time.Duration) bool {
 	if q.count == 0 {
 		return false
 	}
 	// Applied to task with same group first.
 	if r, ok := q.route[group]; ok {
 		// Try to merge task into queue.
-		if r.Value.(*mergeTaskQueue).tryMerge(task, maxNQ) {
+		if r.Value.(*mergeTaskQueue).tryMerge(task, maxNQ, nqMergeRatio, maxDeadlineMergeGap) {
 			return true
 		}
 	}
@@ -144,7 +245,7 @@ func (q *fairPollingTaskQueue) tryMergeWithSameGroup(group string, task MergeTas
 }
 
 // push add a new task into queue, try merge first.
-func (q *fairPollingTaskQueue) push(group string, task Task) {
+func (q *fairPollingTaskQueue) push(group string, task *queuedTask) {
 	// Add a new task.
 	if r, ok := q.route[group]; ok {
 		// Add new task to the back of queue if queue exist.
@@ -167,11 +268,30 @@ func (q *fairPollingTaskQueue) push(group string, task Task) {
 	q.count++
 }
 
+func (q *fairPollingTaskQueue) cleanup(now time.Time) []*queuedTask {
+	if q.count == 0 || q.checkpoint == nil {
+		return nil
+	}
+	removed := make([]*queuedTask, 0)
+	checkpoint := q.checkpoint
+	queuesLen := q.checkpoint.Len()
+	for i := 0; i < queuesLen; i++ {
+		queue := checkpoint.Value.(*mergeTaskQueue)
+		tasks := queue.cleanup(now)
+		if len(tasks) > 0 {
+			q.count -= len(tasks)
+			removed = append(removed, tasks...)
+		}
+		checkpoint = checkpoint.Next()
+	}
+	return removed
+}
+
 // pop pop next ready task.
-func (q *fairPollingTaskQueue) pop(queueExpire time.Duration) (task Task) {
+func (q *fairPollingTaskQueue) pop(queueExpire time.Duration) *queuedTask {
 	// Return directly if there's no task exists.
 	if q.count == 0 {
-		return
+		return nil
 	}
 	checkpoint := q.checkpoint
 	queuesLen := q.checkpoint.Len()
@@ -196,14 +316,20 @@ func (q *fairPollingTaskQueue) pop(queueExpire time.Duration) (task Task) {
 			checkpoint = next
 			continue
 		}
-		task = queue.front()
-		queue.pop()
-		q.count--
+		task := queue.pop()
+		if task.valid() {
+			q.count--
+		}
+		if !task.valid() {
+			checkpoint = next
+			continue
+		}
 		checkpoint = next
-		break
+		q.checkpoint = checkpoint
+		return task
 	}
 
 	// Update checkpoint.
 	q.checkpoint = checkpoint
-	return
+	return nil
 }

internal/util/searchutil/scheduler/queues_test.go

@@ -1,6 +1,7 @@
 package scheduler
 
 import (
+	"context"
 	"fmt"
 	"testing"
 	"time"
@@ -11,9 +12,9 @@ import (
 func TestMergeTaskQueue(t *testing.T) {
 	q := newMergeTaskQueue("test_user")
 	assert.Equal(t, 0, q.len())
-	assert.Nil(t, q.front())
+	assert.False(t, q.front().valid())
 	q.pop()
-	assert.Nil(t, q.front())
+	assert.False(t, q.front().valid())
 	assert.Equal(t, 0, q.len())
 	assert.False(t, q.expire(5*time.Second))
 	time.Sleep(1 * time.Second)
@@ -25,7 +26,7 @@ func TestMergeTaskQueue(t *testing.T) {
 		task := newMockTask(mockTaskConfig{
 			username: "test_user",
 		})
-		q.push(task)
+		q.push(newQueuedTask(task, time.Now()))
 		assert.Equal(t, i, q.len())
 		assert.False(t, q.expire(time.Second))
 	}
@@ -45,14 +46,14 @@ func TestMergeTaskQueue(t *testing.T) {
 		mergeAble: true,
 		nq:        1,
 	})
-	q.push(task)
+	q.push(newQueuedTask(task, time.Now()))
 	for i := 1; i <= 20; i++ {
 		task := newMockTask(mockTaskConfig{
 			username:  "test_user",
 			mergeAble: true,
 			nq:        int64(i),
 		})
-		assert.Equal(t, i <= 10, q.tryMerge(tryIntoMergeTask(task), 56))
+		assert.Equal(t, i <= 10, q.tryMerge(newQueuedTask(task, time.Now()), 56, 0, 0))
 	}
 	for i := 1; i <= 2; i++ {
 		task := newMockTask(mockTaskConfig{
@@ -60,7 +61,7 @@ func TestMergeTaskQueue(t *testing.T) {
 			mergeAble: true,
 			nq:        int64(1),
 		})
-		q.push(task)
+		q.push(newQueuedTask(task, time.Now()))
 	}
 	for i := 1; i <= 20; i++ {
 		task := newMockTask(mockTaskConfig{
@@ -70,15 +71,148 @@ func TestMergeTaskQueue(t *testing.T) {
 		})
 		// 2 + 1 + ... + 9 < 55
 		// 1 + 10 + 11 + 12 + 13 < 55
-		assert.Equal(t, i <= 13, q.tryMerge(tryIntoMergeTask(task), 55))
+		assert.Equal(t, i <= 13, q.tryMerge(newQueuedTask(task, time.Now()), 55, 0, 0))
 	}
 }
 
+func TestMergeTaskQueueNQMergeRatio(t *testing.T) {
+	q := newMergeTaskQueue("test_user")
+	q.push(newQueuedTask(newMockTask(mockTaskConfig{
+		username:  "test_user",
+		mergeAble: true,
+		nq:        2,
+	}), time.Now()))
+
+	assert.True(t, q.tryMerge(newQueuedTask(newMockTask(mockTaskConfig{
+		username:  "test_user",
+		mergeAble: true,
+		nq:        4,
+	}), time.Now()), 16, 3, 0))
+
+	assert.False(t, q.tryMerge(newQueuedTask(newMockTask(mockTaskConfig{
+		username:  "test_user",
+		mergeAble: true,
+		nq:        4,
+	}), time.Now()), 16, 3, 0))
+	assert.Equal(t, int64(6), q.front().NQ())
+}
+
+func TestMergeTaskQueueDeadlineMergeGap(t *testing.T) {
+	now := time.Now()
+	baseDeadline := now.Add(time.Second)
+	baseCtx, baseCancel := context.WithDeadline(context.Background(), baseDeadline)
+	defer baseCancel()
+
+	newTask := func(ctx context.Context, nq int64) *queuedTask {
+		return newQueuedTask(newMockTask(mockTaskConfig{
+			ctx:       ctx,
+			username:  "test_user",
+			mergeAble: true,
+			nq:        nq,
+		}), now)
+	}
+
+	t.Run("deadline gap too large", func(t *testing.T) {
+		q := newMergeTaskQueue("test_user")
+		q.push(newTask(baseCtx, 2))
+		shortCtx, shortCancel := context.WithDeadline(context.Background(), baseDeadline.Add(-100*time.Millisecond))
+		defer shortCancel()
+
+		assert.False(t, q.tryMerge(newTask(shortCtx, 2), 16, 0, 50*time.Millisecond))
+		assert.Equal(t, int64(2), q.front().NQ())
+	})
+
+	t.Run("deadline gap within threshold", func(t *testing.T) {
+		q := newMergeTaskQueue("test_user")
+		q.push(newTask(baseCtx, 2))
+		closeCtx, closeCancel := context.WithDeadline(context.Background(), baseDeadline.Add(-40*time.Millisecond))
+		defer closeCancel()
+
+		assert.True(t, q.tryMerge(newTask(closeCtx, 2), 16, 0, 50*time.Millisecond))
+		assert.Equal(t, int64(4), q.front().NQ())
+	})
+
+	t.Run("one side missing deadline", func(t *testing.T) {
+		q := newMergeTaskQueue("test_user")
+		q.push(newTask(baseCtx, 2))
+
+		assert.False(t, q.tryMerge(newTask(context.Background(), 2), 16, 0, 50*time.Millisecond))
+		assert.Equal(t, int64(2), q.front().NQ())
+	})
+
+	t.Run("both sides missing deadline", func(t *testing.T) {
+		q := newMergeTaskQueue("test_user")
+		q.push(newTask(context.Background(), 2))
+
+		assert.True(t, q.tryMerge(newTask(context.Background(), 2), 16, 0, 50*time.Millisecond))
+		assert.Equal(t, int64(4), q.front().NQ())
+	})
+}
+
+func TestMergeTaskQueueCleanupSkipsRewriteWithoutExpiredTask(t *testing.T) {
+	q := newMergeTaskQueue("test_user")
+	now := time.Now()
+	for i := 0; i < 3; i++ {
+		task := newMockTask(mockTaskConfig{username: "test_user"})
+		q.push(newQueuedTask(task, now))
+	}
+
+	firstTask := q.tasks[0].Task
+	removed := q.cleanup(now)
+
+	assert.Empty(t, removed)
+	assert.Equal(t, 3, q.len())
+	assert.Same(t, firstTask, q.tasks[0].Task)
+}
+
+func TestMergeTaskQueueCleanupMarksExpiredTasksInPlace(t *testing.T) {
+	q := newMergeTaskQueue("test_user")
+	now := time.Now()
+	later := now.Add(time.Minute)
+
+	ctx1, cancel1 := context.WithDeadline(context.Background(), later)
+	defer cancel1()
+	ctx2, cancel2 := context.WithDeadline(context.Background(), now.Add(-time.Millisecond))
+	defer cancel2()
+	ctx3, cancel3 := context.WithDeadline(context.Background(), later)
+	defer cancel3()
+
+	q.push(newQueuedTask(newMockTask(mockTaskConfig{ctx: ctx1, username: "test_user"}), now))
+	expiredTask := newMockTask(mockTaskConfig{ctx: ctx2, username: "test_user"})
+	q.push(newQueuedTask(expiredTask, now))
+	q.push(newQueuedTask(newMockTask(mockTaskConfig{ctx: ctx3, username: "test_user"}), now))
+
+	originalQueueLen := len(q.tasks)
+	removed := q.cleanup(now)
+
+	assert.Len(t, removed, 1)
+	assert.Same(t, expiredTask, removed[0].Task)
+	assert.Equal(t, 2, q.len())
+	assert.Equal(t, originalQueueLen, len(q.tasks))
+	assert.True(t, q.pop().valid())
+	assert.True(t, q.pop().valid())
+	assert.False(t, q.pop().valid())
+	assert.Equal(t, 0, q.len())
+}
+
+func TestMergeTaskQueuePopClearsPoppedSlot(t *testing.T) {
+	q := newMergeTaskQueue("test_user")
+	task := newMockTask(mockTaskConfig{username: "test_user"})
+	q.push(newQueuedTask(task, time.Now()))
+
+	tasks := q.tasks
+	popped := q.pop()
+
+	assert.True(t, popped.valid())
+	assert.False(t, tasks[0].valid())
+	assert.Equal(t, 0, q.len())
+}
+
 func TestFairPollingTaskQueue(t *testing.T) {
 	q := newFairPollingTaskQueue()
 	assert.Equal(t, 0, q.len())
 	assert.Equal(t, 0, q.groupLen(""))
-	assert.Nil(t, q.pop(time.Second))
+	assert.False(t, q.pop(time.Second).valid())
 	assert.Equal(t, 0, q.len())
 	assert.Equal(t, 0, q.groupLen(""))
 
@@ -90,7 +224,7 @@ func TestFairPollingTaskQueue(t *testing.T) {
 		task := newMockTask(mockTaskConfig{
 			username: username,
 		})
-		q.push(username, task)
+		q.push(username, newQueuedTask(task, time.Now()))
 		assert.Equal(t, i, q.len())
 		assert.Equal(t, (i-1)/10+1, q.groupLen(username))
 	}
@@ -99,7 +233,7 @@ func TestFairPollingTaskQueue(t *testing.T) {
 	// Test Pop
 	for i := 1; i <= n; i++ {
 		task := q.pop(time.Minute)
-		assert.NotNil(t, task)
+		assert.True(t, task.valid())
 		username := task.Username()
 		expectedUserName := fmt.Sprintf("user_%d", (i-1)%userN)
 		assert.Equal(t, n-i, q.len())
@@ -113,9 +247,9 @@ func TestFairPollingTaskQueue(t *testing.T) {
 	task := newMockTask(mockTaskConfig{
 		username: username,
 	})
-	q.push(username, task)
+	q.push(username, newQueuedTask(task, time.Now()))
 	assert.Equal(t, userN+1, len(q.route))
-	assert.NotNil(t, q.pop(time.Second))
+	assert.True(t, q.pop(time.Second).valid())
 	assert.Equal(t, 1, len(q.route))
 
 	// Test Merge.
@@ -124,8 +258,8 @@ func TestFairPollingTaskQueue(t *testing.T) {
 		username:  username,
 		mergeAble: true,
 	})
-	assert.False(t, q.tryMergeWithSameGroup(username, tryIntoMergeTask(task), 1))
-	assert.False(t, q.tryMergeWithOtherGroup(username, tryIntoMergeTask(task), 1))
+	assert.False(t, q.tryMergeWithSameGroup(username, newQueuedTask(task, time.Now()), 1, 0, 0))
+	assert.False(t, q.tryMergeWithOtherGroup(username, newQueuedTask(task, time.Now()), 1, 0, 0))
 
 	// Add basic user info first.
 	for i := 1; i <= userN; i++ {
@@ -134,7 +268,7 @@ func TestFairPollingTaskQueue(t *testing.T) {
 			username:  username,
 			mergeAble: true,
 		})
-		q.push(username, task)
+		q.push(username, newQueuedTask(task, time.Now()))
 		assert.Equal(t, i, q.len())
 		assert.Equal(t, 1, q.groupLen(username))
 	}
@@ -146,7 +280,7 @@ func TestFairPollingTaskQueue(t *testing.T) {
 			username:  username,
 			mergeAble: true,
 		})
-		success := q.tryMergeWithSameGroup(username, tryIntoMergeTask(task), int64(n/userN))
+		success := q.tryMergeWithSameGroup(username, newQueuedTask(task, time.Now()), int64(n/userN), 0, 0)
 		assert.Equal(t, i+userN <= n, success)
 		assert.Equal(t, userN, q.len())
 		assert.Equal(t, 1, q.groupLen(username))
@@ -157,7 +291,7 @@ func TestFairPollingTaskQueue(t *testing.T) {
 		username:  username,
 		mergeAble: true,
 	})
-	assert.False(t, q.tryMergeWithOtherGroup(username, tryIntoMergeTask(task), int64(n/userN)))
-	assert.True(t, q.tryMergeWithOtherGroup(username, tryIntoMergeTask(task), int64(n/userN)+1))
+	assert.False(t, q.tryMergeWithOtherGroup(username, newQueuedTask(task, time.Now()), int64(n/userN), 0, 0))
+	assert.True(t, q.tryMergeWithOtherGroup(username, newQueuedTask(task, time.Now()), int64(n/userN)+1, 0, 0))
 	assert.Equal(t, 0, q.groupLen(username))
 }

internal/util/searchutil/scheduler/tasks.go

@@ -1,6 +1,11 @@
 package scheduler
 
-import "github.com/milvus-io/milvus/pkg/v3/proto/internalpb"
+import (
+	"context"
+	"time"
+
+	"github.com/milvus-io/milvus/pkg/v3/proto/internalpb"
+)
 
 const (
 	schedulePolicyNameFIFO            = "fifo"
@@ -36,8 +41,8 @@ func tryIntoMergeTask(t Task) MergeTask {
 
 type Scheduler interface {
 	// Add a new task into scheduler, follow some constraints.
-	// 1. It's a non-block operation.
-	// 2. Error will be returned if scheduler reaches some limit.
+	// 1. Error will be returned if scheduler reaches some limit.
+	// 2. Error will be returned if task context is canceled while waiting to be accepted.
 	// 3. Concurrent safe.
 	Add(task Task) error
 
@@ -58,17 +63,63 @@ type Scheduler interface {
 
 // schedulePolicy is the policy of scheduler.
 type schedulePolicy interface {
+	// Cleanup removes queued tasks whose context deadline has been reached.
+	// Removed tasks are returned to scheduler for error notification.
+	Cleanup(now time.Time) []*queuedTask
+
 	// Push add a new task into scheduler.
-	// Return the count of new task added (task may be chunked, merged or dropped)
+	// Return the count of new task added (task may be chunked or merged)
 	// 0 and an error will be returned if scheduler reaches some limit.
-	Push(task Task) (int, error)
+	Push(task *queuedTask) (int, error)
 
 	// Pop get the task next ready to run.
-	Pop() Task
+	Pop(now time.Time) *queuedTask
 
 	Len() int
 }
 
+type queuedTask struct {
+	Task
+
+	enqueueTime time.Time
+}
+
+func newQueuedTask(task Task, enqueueTime time.Time) *queuedTask {
+	return &queuedTask{
+		Task:        task,
+		enqueueTime: enqueueTime,
+	}
+}
+
+func (t *queuedTask) queueDuration(now time.Time) time.Duration {
+	if !t.valid() || t.enqueueTime.IsZero() {
+		return 0
+	}
+	return now.Sub(t.enqueueTime)
+}
+
+func (t *queuedTask) valid() bool {
+	return t != nil && t.Task != nil
+}
+
+func (t *queuedTask) cleanupReady(now time.Time) bool {
+	if !t.valid() {
+		return false
+	}
+	if t.Context().Err() != nil {
+		return true
+	}
+	deadline, ok := t.Context().Deadline()
+	return ok && !now.Before(deadline)
+}
+
+func cleanupTaskError(task *queuedTask) error {
+	if err := task.Context().Err(); err != nil {
+		return err
+	}
+	return context.DeadlineExceeded
+}
+
 // MergeTask is a Task which can be merged with other task
 type MergeTask interface {
 	Task
@@ -76,10 +127,15 @@ type MergeTask interface {
 	// MergeWith other task, return true if merge success.
 	// After success, the task merged should be dropped.
 	MergeWith(Task) bool
+
+	// MinNQ returns the minimum NQ among the original tasks in this merged task.
+	MinNQ() int64
 }
 
 // A task is execute unit of scheduler.
 type Task interface {
+	Context() context.Context
+
 	// Return the username which task is belong to.
 	// Return "" if the task do not contain any user info.
 	Username() string
@@ -96,10 +152,6 @@ type Task interface {
 	// Done notify the task finished.
 	Done(err error)
 
-	// Check if the Task is canceled.
-	// Concurrent safe.
-	Canceled() error
-
 	// Wait for task finish.
 	// Concurrent safe.
 	Wait() error

internal/util/searchutil/scheduler/user_task_polling_policy.go

@@ -22,22 +22,28 @@ type userTaskPollingPolicy struct {
 	queue *fairPollingTaskQueue
 }
 
+func (p *userTaskPollingPolicy) Cleanup(now time.Time) []*queuedTask {
+	return p.queue.cleanup(now)
+}
+
 // Push add a new task into scheduler, an error will be returned if scheduler reaches some limit.
-func (p *userTaskPollingPolicy) Push(task Task) (int, error) {
+func (p *userTaskPollingPolicy) Push(task *queuedTask) (int, error) {
 	pt := paramtable.Get()
 	username := task.Username()
 
 	// Try to merge task if task is mergeable.
-	if t := tryIntoMergeTask(task); t != nil {
+	if t := tryIntoMergeTask(task.Task); t != nil {
 		// Try to merge with same group first.
 		maxNQ := pt.QueryNodeCfg.MaxGroupNQ.GetAsInt64()
-		if p.queue.tryMergeWithSameGroup(username, t, maxNQ) {
+		nqMergeRatio := pt.QueryNodeCfg.NQMergeRatio.GetAsFloat()
+		maxDeadlineMergeGap := pt.QueryNodeCfg.MaxDeadlineMergeGap.GetAsDurationByParse()
+		if p.queue.tryMergeWithSameGroup(username, task, maxNQ, nqMergeRatio, maxDeadlineMergeGap) {
 			return 0, nil
 		}
 
 		// Try to merge with other group if option is enabled.
 		enableCrossGroupMerge := pt.QueryNodeCfg.SchedulePolicyEnableCrossUserGrouping.GetAsBool()
-		if enableCrossGroupMerge && p.queue.tryMergeWithOtherGroup(username, t, maxNQ) {
+		if enableCrossGroupMerge && p.queue.tryMergeWithOtherGroup(username, task, maxNQ, nqMergeRatio, maxDeadlineMergeGap) {
 			return 0, nil
 		}
 	}
@@ -60,7 +66,7 @@ func (p *userTaskPollingPolicy) Push(task Task) (int, error) {
 }
 
 // Pop get the task next ready to run.
-func (p *userTaskPollingPolicy) Pop() Task {
+func (p *userTaskPollingPolicy) Pop(now time.Time) *queuedTask {
 	expire := paramtable.Get().QueryNodeCfg.SchedulePolicyTaskQueueExpire.GetAsDuration(time.Second)
 	return p.queue.pop(expire)
 }

pkg/metrics/metrics.go

@@ -142,6 +142,7 @@ const (
 	cgoTypeLabelName               = `cgo_type`
 	queueTypeLabelName             = `queue_type`
 	poolNameLabelName              = "pool_name"
+	outcomeLabelName               = "outcome"
 
 	// model function/UDF labels
 	functionTypeName = "function_type_name"

pkg/metrics/querynode_metrics.go

@@ -267,16 +267,6 @@ var (
 			nodeIDLabelName,
 		})
 
-	QueryNodeReadTaskUnsolveLen = prometheus.NewGaugeVec(
-		prometheus.GaugeOpts{
-			Namespace: milvusNamespace,
-			Subsystem: typeutil.QueryNodeRole,
-			Name:      "read_task_unsolved_len",
-			Help:      "number of unsolved read tasks in unsolvedQueue",
-		}, []string{
-			nodeIDLabelName,
-		})
-
 	QueryNodeReadTaskReadyLen = prometheus.NewGaugeVec(
 		prometheus.GaugeOpts{
 			Namespace: milvusNamespace,
@@ -287,6 +277,40 @@ var (
 			nodeIDLabelName,
 		})
 
+	QueryNodeReadTaskReadyNQ = prometheus.NewGaugeVec(
+		prometheus.GaugeOpts{
+			Namespace: milvusNamespace,
+			Subsystem: typeutil.QueryNodeRole,
+			Name:      "read_task_ready_nq",
+			Help:      "total NQ of ready read tasks in scheduler queue",
+		}, []string{
+			nodeIDLabelName,
+		})
+
+	QueryNodeReadTaskQueueDuration = prometheus.NewHistogramVec(
+		prometheus.HistogramOpts{
+			Namespace: milvusNamespace,
+			Subsystem: typeutil.QueryNodeRole,
+			Name:      "read_task_queue_duration",
+			Help:      "duration in milliseconds that read tasks stay in scheduler policy queue",
+			Buckets:   subMsBuckets,
+		}, []string{
+			nodeIDLabelName,
+			outcomeLabelName,
+		})
+
+	QueryNodeReadTaskExecuteDuration = prometheus.NewHistogramVec(
+		prometheus.HistogramOpts{
+			Namespace: milvusNamespace,
+			Subsystem: typeutil.QueryNodeRole,
+			Name:      "read_task_execute_duration",
+			Help:      "duration in milliseconds that read tasks spend in scheduler execution pool",
+			Buckets:   subMsBuckets,
+		}, []string{
+			nodeIDLabelName,
+			outcomeLabelName,
+		})
+
 	QueryNodeReadTaskConcurrency = prometheus.NewGaugeVec(
 		prometheus.GaugeOpts{
 			Namespace: milvusNamespace,
@@ -947,8 +971,10 @@ func RegisterQueryNode(registry *prometheus.Registry) {
 	registry.MustRegister(QueryNodeSQSegmentLatencyInCore)
 	registry.MustRegister(QueryNodeReduceLatency)
 	registry.MustRegister(QueryNodeLoadSegmentLatency)
-	registry.MustRegister(QueryNodeReadTaskUnsolveLen)
 	registry.MustRegister(QueryNodeReadTaskReadyLen)
+	registry.MustRegister(QueryNodeReadTaskReadyNQ)
+	registry.MustRegister(QueryNodeReadTaskQueueDuration)
+	registry.MustRegister(QueryNodeReadTaskExecuteDuration)
 	registry.MustRegister(QueryNodeReadTaskConcurrency)
 	registry.MustRegister(QueryNodeEstimateCPUUsage)
 	registry.MustRegister(QueryNodeSearchGroupNQ)

pkg/util/paramtable/component_param.go

@@ -3515,11 +3515,12 @@ type queryNodeConfig struct {
 	ReadAheadPolicy     ParamItem `refreshable:"false"`
 	ChunkCacheWarmingUp ParamItem `refreshable:"true"`
 
-	MaxReceiveChanSize    ParamItem `refreshable:"false"`
 	MaxUnsolvedQueueSize  ParamItem `refreshable:"true"`
 	MaxReadConcurrency    ParamItem `refreshable:"true"`
 	MaxGpuReadConcurrency ParamItem `refreshable:"false"`
 	MaxGroupNQ            ParamItem `refreshable:"true"`
+	NQMergeRatio          ParamItem `refreshable:"true"`
+	MaxDeadlineMergeGap   ParamItem `refreshable:"true"`
 	TopKMergeRatio        ParamItem `refreshable:"true"`
 	CPURatio              ParamItem `refreshable:"true"`
 	GracefulStopTimeout   ParamItem `refreshable:"false"`
@@ -3551,6 +3552,7 @@ type queryNodeConfig struct {
 	// schedule task policy.
 	SchedulePolicyName                    ParamItem `refreshable:"false"`
 	SchedulePolicyTaskQueueExpire         ParamItem `refreshable:"true"`
+	SchedulePolicyTaskDeadlineAdvance     ParamItem `refreshable:"true"`
 	SchedulePolicyEnableCrossUserGrouping ParamItem `refreshable:"true"`
 	SchedulePolicyMaxPendingTaskPerUser   ParamItem `refreshable:"true"`
 
@@ -3612,6 +3614,20 @@ type queryNodeConfig struct {
 	ExternalCollectionRawDataFactor    ParamItem `refreshable:"true"`
 }
 
+func formatDurationWithMillisecondFallback(v string) string {
+	v = strings.TrimSpace(v)
+	if v == "" {
+		return v
+	}
+	if _, err := time.ParseDuration(v); err == nil {
+		return v
+	}
+	if _, err := strconv.ParseFloat(v, 64); err == nil {
+		return v + "ms"
+	}
+	return v
+}
+
 func (p *queryNodeConfig) init(base *BaseTable) {
 	p.IDFPreload = ParamItem{
 		Key:          "queryNode.idfOracle.preload",
@@ -4350,14 +4366,6 @@ However, this optimization may come at the cost of a slight decrease in query la
 	}
 	p.ChunkCacheWarmingUp.Init(base.mgr)
 
-	p.MaxReceiveChanSize = ParamItem{
-		Key:          "queryNode.scheduler.receiveChanSize",
-		Version:      "2.0.0",
-		DefaultValue: "10240",
-		Export:       true,
-	}
-	p.MaxReceiveChanSize.Init(base.mgr)
-
 	p.MaxReadConcurrency = ParamItem{
 		Key:          "queryNode.scheduler.maxReadConcurrentRatio",
 		Version:      "2.0.0",
@@ -4392,7 +4400,7 @@ Max read concurrency must greater than or equal to 1, and less than or equal to
 	p.MaxUnsolvedQueueSize = ParamItem{
 		Key:          "queryNode.scheduler.unsolvedQueueSize",
 		Version:      "2.0.0",
-		DefaultValue: "10240",
+		DefaultValue: "1024",
 		Export:       true,
 	}
 	p.MaxUnsolvedQueueSize.Init(base.mgr)
@@ -4400,11 +4408,30 @@ Max read concurrency must greater than or equal to 1, and less than or equal to
 	p.MaxGroupNQ = ParamItem{
 		Key:          "queryNode.grouping.maxNQ",
 		Version:      "2.0.0",
-		DefaultValue: "1000",
+		DefaultValue: "16",
 		Export:       true,
 	}
 	p.MaxGroupNQ.Init(base.mgr)
 
+	p.NQMergeRatio = ParamItem{
+		Key:          "queryNode.grouping.nqMergeRatio",
+		Version:      "2.6.17",
+		DefaultValue: "3.0",
+		Doc:          "Maximum ratio between merged total NQ and the smaller task NQ when grouping query node read tasks.",
+		Export:       true,
+	}
+	p.NQMergeRatio.Init(base.mgr)
+
+	p.MaxDeadlineMergeGap = ParamItem{
+		Key:          "queryNode.grouping.maxDeadlineMergeGap",
+		Version:      "2.6.17",
+		DefaultValue: "50ms",
+		Doc:          "Maximum allowed gap between task context deadlines when grouping query node read tasks. Tasks with only one deadline cannot be grouped. Set a negative duration to disable this check.",
+		Formatter:    formatDurationWithMillisecondFallback,
+		Export:       true,
+	}
+	p.MaxDeadlineMergeGap.Init(base.mgr)
+
 	p.TopKMergeRatio = ParamItem{
 		Key:          "queryNode.grouping.topKMergeRatio",
 		Version:      "2.0.0",
@@ -4643,7 +4670,7 @@ user-task-polling:
 	Scheduling is fair on task granularity.
 	The policy is based on the username for authentication.
 	And an empty username is considered the same user.
-	When there are no multi-users, the policy decay into FIFO"`,
+	When there are no multi-users, the policy decay into FIFO.`,
 		Export: true,
 	}
 	p.SchedulePolicyName.Init(base.mgr)
@@ -4655,6 +4682,15 @@ user-task-polling:
 		Export:       true,
 	}
 	p.SchedulePolicyTaskQueueExpire.Init(base.mgr)
+	p.SchedulePolicyTaskDeadlineAdvance = ParamItem{
+		Key:          "queryNode.scheduler.scheduleReadPolicy.taskDeadlineAdvance",
+		Version:      "2.6.17",
+		DefaultValue: "50ms",
+		Doc:          "Advance duration for cleaning queued query node read tasks before their context deadline. It supports duration strings such as 50ms and 1s. A bare number is interpreted as milliseconds for compatibility.",
+		Formatter:    formatDurationWithMillisecondFallback,
+		Export:       true,
+	}
+	p.SchedulePolicyTaskDeadlineAdvance.Init(base.mgr)
 	p.SchedulePolicyEnableCrossUserGrouping = ParamItem{
 		Key:          "queryNode.scheduler.scheduleReadPolicy.enableCrossUserGrouping",
 		Version:      "2.3.0",
@@ -4671,7 +4707,6 @@ user-task-polling:
 		Export:       true,
 	}
 	p.SchedulePolicyMaxPendingTaskPerUser.Init(base.mgr)
-
 	p.CGOPoolSizeRatio = ParamItem{
 		Key:          "queryNode.segcore.cgoPoolSizeRatio",
 		Version:      "2.3.0",

pkg/util/paramtable/component_param_test.go

@@ -218,6 +218,7 @@ func TestComponentParam(t *testing.T) {
 		t.Logf("MaxDimension: %d", Params.MaxDimension.GetAsInt64())
 
 		t.Logf("MaxTaskNum: %d", Params.MaxTaskNum.GetAsInt64())
+		assert.Equal(t, int64(1024), Params.MaxTaskNum.GetAsInt64())
 
 		t.Logf("AccessLog.Enable: %t", Params.AccessLog.Enable.GetAsBool())
 
@@ -460,8 +461,23 @@ func TestComponentParam(t *testing.T) {
 		nprobe := Params.InterimIndexNProbe.GetAsInt64()
 		assert.Equal(t, int64(16), nprobe)
 
-		assert.Equal(t, int32(10240), Params.MaxReceiveChanSize.GetAsInt32())
-		assert.Equal(t, int32(10240), Params.MaxUnsolvedQueueSize.GetAsInt32())
+		assert.Equal(t, int32(1024), Params.MaxUnsolvedQueueSize.GetAsInt32())
+		assert.Equal(t, "1024", Params.MaxUnsolvedQueueSize.DefaultValue)
+		assert.Equal(t, int64(16), Params.MaxGroupNQ.GetAsInt64())
+		assert.Equal(t, 3.0, Params.NQMergeRatio.GetAsFloat())
+		assert.Equal(t, 50*time.Millisecond, Params.MaxDeadlineMergeGap.GetAsDurationByParse())
+		defer params.Reset(Params.MaxDeadlineMergeGap.Key)
+		assert.NoError(t, params.Save(Params.MaxDeadlineMergeGap.Key, "100ms"))
+		assert.Equal(t, 100*time.Millisecond, Params.MaxDeadlineMergeGap.GetAsDurationByParse())
+		assert.NoError(t, params.Save(Params.MaxDeadlineMergeGap.Key, "100"))
+		assert.Equal(t, 100*time.Millisecond, Params.MaxDeadlineMergeGap.GetAsDurationByParse())
+		assert.Equal(t, "fifo", Params.SchedulePolicyName.GetValue())
+		assert.Equal(t, 50*time.Millisecond, Params.SchedulePolicyTaskDeadlineAdvance.GetAsDurationByParse())
+		defer params.Reset(Params.SchedulePolicyTaskDeadlineAdvance.Key)
+		assert.NoError(t, params.Save(Params.SchedulePolicyTaskDeadlineAdvance.Key, "100ms"))
+		assert.Equal(t, 100*time.Millisecond, Params.SchedulePolicyTaskDeadlineAdvance.GetAsDurationByParse())
+		assert.NoError(t, params.Save(Params.SchedulePolicyTaskDeadlineAdvance.Key, "100"))
+		assert.Equal(t, 100*time.Millisecond, Params.SchedulePolicyTaskDeadlineAdvance.GetAsDurationByParse())
 		assert.Equal(t, 10.0, Params.CPURatio.GetAsFloat())
 		assert.Equal(t, uint32(hardware.GetCPUNum()), Params.KnowhereThreadPoolSize.GetAsUint32())
 
@@ -880,6 +896,15 @@ func TestForbiddenItem(t *testing.T) {
 	assert.Equal(t, "by-dev", params.CommonCfg.ClusterPrefix.GetValue())
 }
 
+func TestFormatDurationWithMillisecondFallback(t *testing.T) {
+	assert.Equal(t, "", formatDurationWithMillisecondFallback(""))
+	assert.Equal(t, "", formatDurationWithMillisecondFallback("  "))
+	assert.Equal(t, "100ms", formatDurationWithMillisecondFallback("100"))
+	assert.Equal(t, "1.5ms", formatDurationWithMillisecondFallback("1.5"))
+	assert.Equal(t, "2s", formatDurationWithMillisecondFallback("2s"))
+	assert.Equal(t, "invalid", formatDurationWithMillisecondFallback("invalid"))
+}
+
 func TestCachedParam(t *testing.T) {
 	Init()
 	params := Get()