enhance: make query view fanout target configurable

configs/milvus.yaml

@@ -572,6 +572,7 @@ queryCoord:
   cleanExcludeSegmentInterval: 60 # the time duration of clean pipeline exclude segment which used for filter invalid data, in seconds
   queryView:
     fullReconsileInterval: 60 # Interval in seconds for periodic QueryView full reconciliation. Set to 0 to disable it.
+    targetRowsPerShardNode: 100000 # Target number of sealed rows per QueryNode used to derive the free fanout budget for each QueryView shard. Changes take effect on the next reconciliation.
   ip:  # TCP/IP address of queryCoord. If not specified, use the first unicastable address
   port: 19531 # TCP port of queryCoord
   grpc:

internal/views/coord/balancer/snapshot_builder.go

@@ -10,6 +10,7 @@ import (
 	"github.com/milvus-io/milvus/internal/views/coord/loadmgr"
 	"github.com/milvus-io/milvus/internal/views/qviews"
 	"github.com/milvus-io/milvus/pkg/v3/proto/viewpb"
+	"github.com/milvus-io/milvus/pkg/v3/util/paramtable"
 )
 
 // SnapshotBuilder assembles a BalancerSnapshot from the various sources:
@@ -131,7 +132,7 @@ func (b *SnapshotBuilder) build(ctx context.Context, pending triggerBatch) (*Bal
 
 	// 4. Assemble the scoped snapshot consumed by BalancePolicy.
 	snap := &BalancerSnapshot{
-		Config:                b.config,
+		Config:                b.currentBalanceConfig(),
 		LoadConfigSnapshot:    loadSnapshot,
 		ShardViewSnapshot:     targetSnapshot,
 		DataViewSnapshot:      dataViewSnapshot,
@@ -152,6 +153,15 @@ func (b *SnapshotBuilder) build(ctx context.Context, pending triggerBatch) (*Bal
 	return snap, targetShards
 }
 
+func (b *SnapshotBuilder) currentBalanceConfig() *BalanceConfig {
+	if b == nil || b.config == nil {
+		return nil
+	}
+	config := *b.config
+	config.TargetRowsPerShardNode = paramtable.Get().QueryCoordCfg.QueryViewTargetRowsPerShardNode.GetAsInt64()
+	return &config
+}
+
 // takeRowCountDirtyShards atomically swaps the observer-owned dirty set. Marks
 // arriving after the swap remain in the newly installed set for the next Build.
 func (b *SnapshotBuilder) takeRowCountDirtyShards() []qviews.ShardID {

internal/views/coord/balancer/snapshot_builder_test.go

@@ -16,8 +16,32 @@ import (
 	"github.com/milvus-io/milvus/internal/views/qviews"
 	"github.com/milvus-io/milvus/pkg/v3/proto/querypb"
 	"github.com/milvus-io/milvus/pkg/v3/proto/viewpb"
+	"github.com/milvus-io/milvus/pkg/v3/util/paramtable"
 )
 
+func TestSnapshotBuilderReadsBalanceConfigForEachCycle(t *testing.T) {
+	params := paramtable.Get()
+	item := &params.QueryCoordCfg.QueryViewTargetRowsPerShardNode
+	require.NoError(t, params.Reset(item.Key))
+	t.Cleanup(func() { require.NoError(t, params.Reset(item.Key)) })
+
+	builder := NewSnapshotBuilder(
+		emptyLoadConfigStore(t),
+		emptyRegistry(t),
+		&fakeNodeProvider{infos: map[int64]*NodeInfo{}},
+		&fakeDataViewProvider{},
+		DefaultBalanceConfig(),
+	)
+
+	first := buildFullSnapshot(builder).Config
+	require.Equal(t, int64(100_000), first.TargetRowsPerShardNode)
+
+	require.NoError(t, params.Save(item.Key, "250000"))
+	second := buildFullSnapshot(builder).Config
+	require.Equal(t, int64(250_000), second.TargetRowsPerShardNode)
+	require.NotSame(t, first, second)
+}
+
 // --- fake providers used throughout the tests ---
 
 type fakeNodeProvider struct {
@@ -608,7 +632,7 @@ func TestSnapshotBuilder_ScopedRefreshUsesCachedNonTargetAndMatchesFullPlan(t *t
 	assert.Empty(t, provider.segmentRequests, "cached non-target rows must not trigger metadata I/O")
 
 	oracle := &BalancerSnapshot{
-		Config:             builder.config,
+		Config:             builder.currentBalanceConfig(),
 		LoadConfigSnapshot: store.Snapshot(),
 		ShardViewSnapshot:  registry.Snapshot(),
 		DataViewSnapshot:   provider.DataViewSnapshot(context.Background()),

pkg/util/paramtable/component_param.go

@@ -3182,6 +3182,7 @@ type queryCoordConfig struct {
 	AutoWarmupForNonPKIsolationCollection ParamItem `refreshable:"false"`
 	EnableSQNServeSegments                ParamItem `refreshable:"false"`
 	QueryViewFullReconsileInterval        ParamItem `refreshable:"false"`
+	QueryViewTargetRowsPerShardNode       ParamItem `refreshable:"true"`
 }
 
 func (p *queryCoordConfig) init(base *BaseTable) {
@@ -3924,6 +3925,23 @@ Set to 0 to disable the penalty period.`,
 		Export:       true,
 	}
 	p.QueryViewFullReconsileInterval.Init(base.mgr)
+
+	p.QueryViewTargetRowsPerShardNode = ParamItem{
+		Key:          "queryCoord.queryView.targetRowsPerShardNode",
+		Version:      "3.0.0",
+		DefaultValue: "100000",
+		Doc:          "Target number of sealed rows per QueryNode used to derive the free fanout budget for each QueryView shard. Must be positive. Changes take effect on the next reconciliation.",
+		Export:       true,
+		Formatter: func(v string) string {
+			if getAsInt64(v) <= 0 {
+				mlog.Warn(context.TODO(), "queryCoord.queryView.targetRowsPerShardNode must be positive, using default 100000",
+					mlog.String("configured", v))
+				return "100000"
+			}
+			return v
+		},
+	}
+	p.QueryViewTargetRowsPerShardNode.Init(base.mgr)
 }
 
 // /////////////////////////////////////////////////////////////////////////////

pkg/util/paramtable/component_param_test.go

@@ -287,6 +287,25 @@ func TestComponentParam_QueryViewFullReconsileInterval(t *testing.T) {
 	assert.Zero(t, item.GetAsDuration(time.Second))
 }
 
+func TestComponentParam_QueryViewTargetRowsPerShardNode(t *testing.T) {
+	Init()
+	params := Get()
+	item := &params.QueryCoordCfg.QueryViewTargetRowsPerShardNode
+	params.Reset(item.Key)
+	t.Cleanup(func() { params.Reset(item.Key) })
+
+	assert.Equal(t, "queryCoord.queryView.targetRowsPerShardNode", item.Key)
+	assert.Equal(t, "100000", item.DefaultValue)
+	assert.True(t, item.Export)
+	assert.EqualValues(t, 100_000, item.GetAsInt64())
+	assert.NoError(t, params.Save(item.Key, "250000"))
+	assert.EqualValues(t, 250_000, item.GetAsInt64())
+	assert.NoError(t, params.Save(item.Key, "0"))
+	assert.EqualValues(t, 100_000, item.GetAsInt64())
+	assert.NoError(t, params.Save(item.Key, "invalid"))
+	assert.EqualValues(t, 100_000, item.GetAsInt64())
+}
+
 func TestComponentParam_TransformLogCatchupConcurrencyPerStream(t *testing.T) {
 	Init()
 	params := Get()