enhance: pass external segment row target from DataCoord (#51352)

issue: #45881

## What this PR does

External refresh previously read the segment row target directly from
each DataNode configuration during execution. That made the worker input
implicit and allowed task behavior to depend on the selected node.

This change:

- keeps `dataNode.externalCollection.targetRowsPerSegment` as the
existing configuration source;
- has DataCoord snapshot the configured value into the refresh task
request;
- adds `target_rows_per_segment` to the DataCoord-to-DataNode request;
- validates the transmitted value in DataNode;
- uses the request value for fragment splitting and segment balancing;
- preserves the MilvusTable one-fragment-per-segment behavior.

## Validation

- `make generated-proto-without-cpp`
- `make lint-fix`
- `make build-go`
- `go test -gcflags="all=-N -l" -race -tags dynamic,test
github.com/milvus-io/milvus/pkg/v3/util/paramtable -failfast -count=1
-ldflags="-r ${RPATH}" -run "^TestCachedParam$" -v`
- `go test -gcflags="all=-N -l" -race
-coverprofile=/tmp/refine-rowcount-config-dc.out -tags dynamic,test
github.com/milvus-io/milvus/internal/datacoord -failfast -count=1
-ldflags="-r ${RPATH}" -run
"^TestRefreshExternalCollectionTask_CreateTaskOnWorker$" -v`
- `go test -gcflags="all=-N -l" -race
-coverprofile=/tmp/refine-rowcount-config-dn.out -tags dynamic,test
github.com/milvus-io/milvus/internal/datanode/external -failfast
-count=1 -ldflags="-r ${RPATH}" -run
"^TestRefreshExternalCollectionTaskSuite$" -v`
- `git diff --check`

Signed-off-by: Wei Liu <wei.liu@zilliz.com>

internal/datacoord/task_refresh_external_collection.go

@@ -688,6 +688,7 @@ func (t *refreshExternalCollectionTask) CreateTaskOnWorker(nodeID int64, cluster
 		ExploreManifestPath:    t.GetExploreManifestPath(),
 		FileIndexBegin:         t.GetFileIndexBegin(),
 		FileIndexEnd:           t.GetFileIndexEnd(),
+		TargetRowsPerSegment:   paramtable.Get().DataNodeCfg.ExternalCollectionTargetRowsPerSegment.GetAsInt64(),
 	}
 
 	// Submit task to worker via unified task system

internal/datacoord/task_refresh_external_collection_test.go

@@ -865,6 +865,10 @@ func TestRefreshExternalCollectionTask_CreateTaskOnWorker(t *testing.T) {
 	})
 
 	t.Run("success", func(t *testing.T) {
+		const targetRowsPerSegmentKey = "dataNode.externalCollection.targetRowsPerSegment"
+		paramtable.Get().Save(targetRowsPerSegmentKey, "12345")
+		defer paramtable.Get().Reset(targetRowsPerSegmentKey)
+
 		catalog := &stubCatalog{}
 		refreshMeta, err := newExternalCollectionRefreshMeta(context.Background(), catalog)
 		assert.NoError(t, err)
@@ -912,6 +916,7 @@ func TestRefreshExternalCollectionTask_CreateTaskOnWorker(t *testing.T) {
 		assert.Equal(t, indexpb.JobState_JobStateInProgress, metaTask.GetState())
 		assert.NotNil(t, cluster.refreshReq)
 		assert.Equal(t, int64(10), cluster.refreshReq.GetPartitionID())
+		assert.Equal(t, int64(12345), cluster.refreshReq.GetTargetRowsPerSegment())
 	})
 }
 

internal/datanode/external/task_update.go

@@ -187,6 +187,12 @@ func (t *RefreshExternalCollectionTask) PreExecute(ctx context.Context) error {
 	if err != nil {
 		return merr.Wrap(err, "failed to parse external spec")
 	}
+	if t.req.GetTargetRowsPerSegment() == 0 {
+		t.req.TargetRowsPerSegment = paramtable.Get().DataNodeCfg.ExternalCollectionTargetRowsPerSegment.GetAsInt64()
+	}
+	if t.req.GetTargetRowsPerSegment() <= 0 {
+		return merr.WrapErrParameterInvalidMsg("target rows per segment must be positive")
+	}
 	t.parsedSpec = spec
 	schema := proto.Clone(t.req.GetSchema()).(*schemapb.CollectionSchema)
 	if schema.GetExternalSource() == "" {
@@ -256,8 +262,6 @@ func (t *RefreshExternalCollectionTask) fetchFragmentsFromExternalSource(ctx con
 		mlog.Int64("fileIndexBegin", t.req.GetFileIndexBegin()),
 		mlog.Int64("fileIndexEnd", t.req.GetFileIndexEnd()))
 
-	targetRowsPerSegment := paramtable.Get().DataNodeCfg.ExternalCollectionTargetRowsPerSegment.GetAsInt64()
-
 	return packed.FetchFragmentsFromExternalSourceWithRange(
 		ctx,
 		t.parsedSpec.Format,
@@ -270,7 +274,7 @@ func (t *RefreshExternalCollectionTask) fetchFragmentsFromExternalSource(ctx con
 		packed.ExternalFetchOptions{
 			CollectionID: t.req.GetCollectionID(),
 			ExternalSpec: t.req.GetExternalSpec(),
-			RowLimit:     targetRowsPerSegment,
+			RowLimit:     t.req.GetTargetRowsPerSegment(),
 		},
 	)
 }
@@ -720,8 +724,7 @@ func (t *RefreshExternalCollectionTask) balanceFragmentsToSegments(ctx context.C
 			}
 		}
 	} else {
-		// Get target rows per segment from configuration
-		targetRowsPerSegment := paramtable.Get().DataNodeCfg.ExternalCollectionTargetRowsPerSegment.GetAsInt64()
+		targetRowsPerSegment := t.req.GetTargetRowsPerSegment()
 		if totalRows < targetRowsPerSegment {
 			targetRowsPerSegment = totalRows
 		}

internal/datanode/external/task_update_test.go

@@ -51,6 +51,16 @@ type RefreshExternalCollectionTaskSuite struct {
 	taskID       int64
 }
 
+func (s *RefreshExternalCollectionTaskSuite) newTask(
+	ctx context.Context,
+	req *datapb.RefreshExternalCollectionTaskRequest,
+) *RefreshExternalCollectionTask {
+	if req != nil && req.GetTargetRowsPerSegment() == 0 {
+		req.TargetRowsPerSegment = 1_000_000
+	}
+	return NewRefreshExternalCollectionTask(ctx, req)
+}
+
 type fakeRecordReader struct {
 	records []storage.Record
 	errs    []error
@@ -90,7 +100,7 @@ func (s *RefreshExternalCollectionTaskSuite) TestNewRefreshExternalCollectionTas
 		ExternalSpec:   "test_spec",
 	}
 
-	task := NewRefreshExternalCollectionTask(ctx, req)
+	task := s.newTask(ctx, req)
 
 	s.NotNil(task)
 	s.Equal(s.collectionID, task.req.GetCollectionID())
@@ -125,7 +135,7 @@ func (s *RefreshExternalCollectionTaskSuite) TestTaskLifecycle() {
 		},
 	}
 
-	task := NewRefreshExternalCollectionTask(ctx, req)
+	task := s.newTask(ctx, req)
 
 	// Test OnEnqueue
 	err := task.OnEnqueue(ctx)
@@ -166,7 +176,7 @@ func (s *RefreshExternalCollectionTaskSuite) TestPreExecuteClonesSchemaBeforeFil
 		},
 	}
 
-	task := NewRefreshExternalCollectionTask(ctx, req)
+	task := s.newTask(ctx, req)
 	err := task.PreExecute(ctx)
 
 	s.NoError(err)
@@ -178,6 +188,48 @@ func (s *RefreshExternalCollectionTaskSuite) TestPreExecuteClonesSchemaBeforeFil
 	s.Equal([]string{"id"}, task.columns)
 }
 
+func (s *RefreshExternalCollectionTaskSuite) TestPreExecuteFallsBackToConfigWhenTargetRowsPerSegmentMissing() {
+	paramtable.Init()
+	const targetRowsPerSegmentKey = "dataNode.externalCollection.targetRowsPerSegment"
+	paramtable.Get().Save(targetRowsPerSegmentKey, "12345")
+	defer paramtable.Get().Reset(targetRowsPerSegmentKey)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	req := &datapb.RefreshExternalCollectionTaskRequest{
+		CollectionID:   s.collectionID,
+		TaskID:         s.taskID,
+		ExternalSource: "test_source",
+		ExternalSpec:   `{"format":"parquet"}`,
+		Schema:         &schemapb.CollectionSchema{},
+		StorageConfig:  &indexpb.StorageConfig{StorageType: "local"},
+	}
+
+	task := NewRefreshExternalCollectionTask(ctx, req)
+	err := task.PreExecute(ctx)
+	s.NoError(err)
+	s.Equal(int64(12345), req.GetTargetRowsPerSegment())
+}
+
+func (s *RefreshExternalCollectionTaskSuite) TestPreExecuteRejectsNegativeTargetRowsPerSegment() {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	req := &datapb.RefreshExternalCollectionTaskRequest{
+		CollectionID:         s.collectionID,
+		TaskID:               s.taskID,
+		ExternalSource:       "test_source",
+		ExternalSpec:         `{"format":"parquet"}`,
+		Schema:               &schemapb.CollectionSchema{},
+		StorageConfig:        &indexpb.StorageConfig{StorageType: "local"},
+		TargetRowsPerSegment: -1,
+	}
+
+	task := NewRefreshExternalCollectionTask(ctx, req)
+	err := task.PreExecute(ctx)
+	s.Error(err)
+	s.Contains(err.Error(), "target rows per segment must be positive")
+}
+
 func (s *RefreshExternalCollectionTaskSuite) TestPreExecuteWithNilRequest() {
 	ctx, cancel := context.WithCancel(context.Background()) //nolint:gosec // cancel is deferred below
 	defer cancel()
@@ -198,7 +250,7 @@ func (s *RefreshExternalCollectionTaskSuite) TestSetAndGetState() {
 		TaskID:       s.taskID,
 	}
 
-	task := NewRefreshExternalCollectionTask(ctx, req)
+	task := s.newTask(ctx, req)
 
 	task.SetState(indexpb.JobState_JobStateInProgress, "")
 	s.Equal(indexpb.JobState_JobStateInProgress, task.GetState())
@@ -216,7 +268,7 @@ func (s *RefreshExternalCollectionTaskSuite) TestReset() {
 		TaskID:       s.taskID,
 	}
 
-	task := NewRefreshExternalCollectionTask(ctx, req)
+	task := s.newTask(ctx, req)
 	task.Reset()
 
 	s.Nil(task.ctx)
@@ -233,7 +285,7 @@ func (s *RefreshExternalCollectionTaskSuite) TestBalanceFragmentsToSegments_Empt
 		TaskID:       s.taskID,
 	}
 
-	task := NewRefreshExternalCollectionTask(ctx, req)
+	task := s.newTask(ctx, req)
 	result, err := task.balanceFragmentsToSegments(context.Background(), []packed.Fragment{})
 	s.NoError(err)
 	s.Nil(result)
@@ -251,7 +303,7 @@ func (s *RefreshExternalCollectionTaskSuite) TestBalanceFragmentsToSegments_Zero
 		CollectionID: s.collectionID,
 		TaskID:       s.taskID,
 	}
-	task := NewRefreshExternalCollectionTask(ctx, req)
+	task := s.newTask(ctx, req)
 
 	fragments := []packed.Fragment{
 		{FragmentID: 1, RowCount: 0, FilePath: "s3://bucket/zero.parquet"},
@@ -285,7 +337,7 @@ func (s *RefreshExternalCollectionTaskSuite) TestBalanceFragmentsToSegments_Sing
 		},
 	}
 
-	task := NewRefreshExternalCollectionTask(ctx, req)
+	task := s.newTask(ctx, req)
 	task.preallocatedIDRange = req.GetPreAllocatedSegmentIds()
 	task.nextAllocID = task.preallocatedIDRange.Begin
 	task.parsedSpec = &externalspec.ExternalSpec{Format: "parquet"}
@@ -332,6 +384,7 @@ func (s *RefreshExternalCollectionTaskSuite) TestBalanceFragmentsToSegments_Mult
 	req := &datapb.RefreshExternalCollectionTaskRequest{
 		CollectionID:           s.collectionID,
 		TaskID:                 s.taskID,
+		TargetRowsPerSegment:   500000,
 		PreAllocatedSegmentIds: &datapb.IDRange{Begin: 100, End: 200},
 		StorageConfig:          &indexpb.StorageConfig{StorageType: "local"},
 		ExternalSource:         "s3://bucket/data/",
@@ -343,7 +396,7 @@ func (s *RefreshExternalCollectionTaskSuite) TestBalanceFragmentsToSegments_Mult
 		},
 	}
 
-	task := NewRefreshExternalCollectionTask(ctx, req)
+	task := s.newTask(ctx, req)
 	task.preallocatedIDRange = req.GetPreAllocatedSegmentIds()
 	task.nextAllocID = task.preallocatedIDRange.Begin
 	task.parsedSpec = &externalspec.ExternalSpec{Format: "parquet"}
@@ -368,6 +421,7 @@ func (s *RefreshExternalCollectionTaskSuite) TestBalanceFragmentsToSegments_Mult
 
 	result, err := task.balanceFragmentsToSegments(context.Background(), fragments)
 	s.NoError(err)
+	s.Len(result, 4)
 
 	// Verify all segments have StorageVersion=V3 and fake binlogs
 	for i, seg := range result {
@@ -413,7 +467,7 @@ func (s *RefreshExternalCollectionTaskSuite) TestPreExecuteContextCanceled() {
 		TaskID:       s.taskID,
 	}
 
-	task := NewRefreshExternalCollectionTask(ctx, req)
+	task := s.newTask(ctx, req)
 	cancel()
 
 	err := task.PreExecute(ctx)
@@ -428,7 +482,7 @@ func (s *RefreshExternalCollectionTaskSuite) TestExecuteContextCanceled() {
 		TaskID:       s.taskID,
 	}
 
-	task := NewRefreshExternalCollectionTask(ctx, req)
+	task := s.newTask(ctx, req)
 	cancel()
 
 	err := task.Execute(ctx)
@@ -443,7 +497,7 @@ func (s *RefreshExternalCollectionTaskSuite) TestBalanceFragmentsToSegmentsConte
 		TaskID:       s.taskID,
 	}
 
-	task := NewRefreshExternalCollectionTask(ctx, req)
+	task := s.newTask(ctx, req)
 	cancel()
 
 	result, err := task.balanceFragmentsToSegments(ctx, []packed.Fragment{{FragmentID: 1, RowCount: 10}})
@@ -463,7 +517,7 @@ func (s *RefreshExternalCollectionTaskSuite) TestOrganizeSegments_AllFragmentsEx
 		},
 	}
 
-	task := NewRefreshExternalCollectionTask(ctx, req)
+	task := s.newTask(ctx, req)
 
 	// Simulate current segment fragments mapping (use FilePath as identifier)
 	currentSegmentFragments := packed.SegmentFragments{
@@ -526,7 +580,7 @@ func (s *RefreshExternalCollectionTaskSuite) TestOrganizeSegments_SameFragmentsM
 			}, "parquet"),
 		}},
 	}
-	task := NewRefreshExternalCollectionTask(ctx, req)
+	task := s.newTask(ctx, req)
 	task.parsedSpec = &externalspec.ExternalSpec{Format: "parquet"}
 	task.columns = []string{"id", "vec", "text", "score"}
 
@@ -625,7 +679,7 @@ func (s *RefreshExternalCollectionTaskSuite) TestOrganizeSegments_PatchedSegment
 			}, "parquet"),
 		}},
 	}
-	task := NewRefreshExternalCollectionTask(ctx, req)
+	task := s.newTask(ctx, req)
 	task.parsedSpec = &externalspec.ExternalSpec{Format: "parquet"}
 	task.columns = []string{"id", "text", "score"}
 
@@ -686,7 +740,7 @@ func (s *RefreshExternalCollectionTaskSuite) TestOrganizeSegments_SameFragmentsA
 			}, "parquet"),
 		}},
 	}
-	task := NewRefreshExternalCollectionTask(ctx, req)
+	task := s.newTask(ctx, req)
 	current := packed.SegmentFragments{
 		10: []packed.Fragment{{FilePath: "s3://bucket/data/a.parquet", StartRow: 0, EndRow: 100, RowCount: 100}},
 	}
@@ -717,7 +771,7 @@ func (s *RefreshExternalCollectionTaskSuite) TestOrganizeSegments_PartialFragmen
 		PreAllocatedSegmentIds: &datapb.IDRange{Begin: 100, End: 200},
 	}
 
-	task := NewRefreshExternalCollectionTask(ctx, req)
+	task := s.newTask(ctx, req)
 
 	// S1 has file1, S2 has file2, S3 has file3
 	currentSegmentFragments := packed.SegmentFragments{
@@ -751,7 +805,7 @@ func (s *RefreshExternalCollectionTaskSuite) TestOrganizeSegments_NewFragmentsUs
 			{ID: 1, CollectionID: s.collectionID, NumOfRows: 1000},
 		},
 	}
-	task := NewRefreshExternalCollectionTask(ctx, req)
+	task := s.newTask(ctx, req)
 
 	currentSegmentFragments := packed.SegmentFragments{
 		1: []packed.Fragment{{FragmentID: 101, FilePath: "/data/file1.parquet", StartRow: 0, EndRow: 1000, RowCount: 1000}},
@@ -806,7 +860,7 @@ func (s *RefreshExternalCollectionTaskSuite) TestOrganizeSegments_RewritesSegmen
 			},
 		},
 	}
-	task := NewRefreshExternalCollectionTask(ctx, req)
+	task := s.newTask(ctx, req)
 
 	currentSegmentFragments := packed.SegmentFragments{
 		1: []packed.Fragment{{FragmentID: 101, FilePath: "/data/file1.parquet", StartRow: 0, EndRow: 1000, RowCount: 1000}},
@@ -869,7 +923,7 @@ func (s *RefreshExternalCollectionTaskSuite) TestOrganizeSegments_KeepsSegmentWi
 			},
 		},
 	}
-	task := NewRefreshExternalCollectionTask(ctx, req)
+	task := s.newTask(ctx, req)
 
 	currentSegmentFragments := packed.SegmentFragments{
 		1: []packed.Fragment{{FragmentID: 101, FilePath: "/data/file1.parquet", StartRow: 0, EndRow: 1000, RowCount: 1000}},
@@ -922,7 +976,7 @@ func (s *RefreshExternalCollectionTaskSuite) TestOrganizeSegments_FunctionOutput
 			},
 		},
 	}
-	task := NewRefreshExternalCollectionTask(ctx, req)
+	task := s.newTask(ctx, req)
 
 	currentSegmentFragments := packed.SegmentFragments{
 		1: []packed.Fragment{{FragmentID: 101, FilePath: "/data/file1.parquet", StartRow: 0, EndRow: 1000, RowCount: 1000}},
@@ -1125,7 +1179,7 @@ func (s *RefreshExternalCollectionTaskSuite) TestBuildFunctionExecutionSchemaBra
 }
 
 func (s *RefreshExternalCollectionTaskSuite) TestSegmentHasFunctionOutputColumnsBranches() {
-	task := NewRefreshExternalCollectionTask(context.Background(), &datapb.RefreshExternalCollectionTaskRequest{})
+	task := s.newTask(context.Background(), &datapb.RefreshExternalCollectionTaskRequest{})
 
 	hasColumns, err := task.segmentHasFunctionOutputColumns(&datapb.SegmentInfo{}, nil)
 	s.NoError(err)
@@ -1165,7 +1219,7 @@ func (s *RefreshExternalCollectionTaskSuite) TestOrganizeSegmentsFunctionOutputC
 			},
 		},
 	}
-	task := NewRefreshExternalCollectionTask(ctx, req)
+	task := s.newTask(ctx, req)
 
 	result, err := task.organizeSegments(ctx, nil, nil)
 	s.Error(err)
@@ -1183,7 +1237,7 @@ func (s *RefreshExternalCollectionTaskSuite) TestOrganizeSegmentsBalanceError()
 		CollectionID: s.collectionID,
 		TaskID:       s.taskID,
 	}
-	task := NewRefreshExternalCollectionTask(ctx, req)
+	task := s.newTask(ctx, req)
 	mockBalance := mockey.Mock(mockey.GetMethod(task, "balanceFragmentsToSegments")).
 		Return(nil, fmt.Errorf("balance failed")).Build()
 	defer mockBalance.UnPatch()
@@ -1234,7 +1288,7 @@ func (s *RefreshExternalCollectionTaskSuite) TestOrganizeSegmentsContextCanceled
 				TaskID:          s.taskID,
 				CurrentSegments: tc.currentSegments,
 			}
-			task := NewRefreshExternalCollectionTask(ctx, req)
+			task := s.newTask(ctx, req)
 
 			var calls int
 			mockEnsure := mockey.Mock(ensureContext).To(func(ctx context.Context) error {
@@ -1359,7 +1413,7 @@ func (s *RefreshExternalCollectionTaskSuite) TestCreateManifestForSegment() {
 		PartitionID:   2000,
 		StorageConfig: &indexpb.StorageConfig{RootPath: "files", StorageType: "local"},
 	}
-	task := NewRefreshExternalCollectionTask(ctx, req)
+	task := s.newTask(ctx, req)
 	task.parsedSpec = &externalspec.ExternalSpec{Format: "parquet"}
 	task.columns = []string{"col1"}
 
@@ -1416,7 +1470,7 @@ func (s *RefreshExternalCollectionTaskSuite) TestCreateManifestWithFunctionsUses
 			},
 		},
 	}
-	task := NewRefreshExternalCollectionTask(ctx, req)
+	task := s.newTask(ctx, req)
 	task.parsedSpec = &externalspec.ExternalSpec{Format: "parquet"}
 
 	var gotBasePath string
@@ -2167,7 +2221,7 @@ func (s *RefreshExternalCollectionTaskSuite) TestTaskAccessorsAndCloneHelpers()
 	s.Error(ensureContext(ctx))
 
 	taskCtx := context.Background()
-	task := NewRefreshExternalCollectionTask(taskCtx, &datapb.RefreshExternalCollectionTaskRequest{TaskID: 10})
+	task := s.newTask(taskCtx, &datapb.RefreshExternalCollectionTaskRequest{TaskID: 10})
 	task.updatedSegments = []*datapb.SegmentInfo{{ID: 100}}
 	s.Equal(taskCtx, task.Ctx())
 	s.Equal(task.updatedSegments, task.GetUpdatedSegments())
@@ -2212,7 +2266,7 @@ func (s *RefreshExternalCollectionTaskSuite) TestExecuteWithMockedSteps() {
 		FileIndexBegin:         0,
 		FileIndexEnd:           1,
 	}
-	task := NewRefreshExternalCollectionTask(ctx, req)
+	task := s.newTask(ctx, req)
 	task.parsedSpec = &externalspec.ExternalSpec{Format: "parquet"}
 	task.columns = []string{"text"}
 	fragments := []packed.Fragment{{FragmentID: 1, FilePath: "file", StartRow: 0, EndRow: 10, RowCount: 10}}
@@ -2245,10 +2299,10 @@ func (s *RefreshExternalCollectionTaskSuite) TestExecuteErrorPathsWithMockedStep
 			TaskID:                 s.taskID,
 			PreAllocatedSegmentIds: &datapb.IDRange{Begin: 3000, End: 4000},
 		}
-		return NewRefreshExternalCollectionTask(ctx, req)
+		return s.newTask(ctx, req)
 	}
 
-	task := NewRefreshExternalCollectionTask(ctx, &datapb.RefreshExternalCollectionTaskRequest{TaskID: s.taskID})
+	task := s.newTask(ctx, &datapb.RefreshExternalCollectionTaskRequest{TaskID: s.taskID})
 	s.Error(task.Execute(ctx))
 
 	task = newTask()
@@ -2283,7 +2337,7 @@ func (s *RefreshExternalCollectionTaskSuite) TestExecuteErrorPathsWithMockedStep
 func (s *RefreshExternalCollectionTaskSuite) TestPostExecuteAndOrganizeCanceled() {
 	ctx, cancel := context.WithCancel(context.Background())
 	cancel()
-	task := NewRefreshExternalCollectionTask(ctx, &datapb.RefreshExternalCollectionTaskRequest{
+	task := s.newTask(ctx, &datapb.RefreshExternalCollectionTaskRequest{
 		TaskID:       s.taskID,
 		CollectionID: s.collectionID,
 	})
@@ -2304,7 +2358,7 @@ func (s *RefreshExternalCollectionTaskSuite) TestBuildCurrentSegmentFragments()
 		StorageConfig:   &indexpb.StorageConfig{RootPath: "files", StorageType: "local"},
 		CurrentSegments: []*datapb.SegmentInfo{{ID: 1}},
 	}
-	task := NewRefreshExternalCollectionTask(ctx, req)
+	task := s.newTask(ctx, req)
 	task.columns = []string{"text_col"}
 	expected := packed.SegmentFragments{1: []packed.Fragment{{FragmentID: 1}}}
 	var gotColumns []string
@@ -2347,7 +2401,7 @@ func (s *RefreshExternalCollectionTaskSuite) TestBalanceFragmentsToSegmentsWithF
 			},
 		},
 	}
-	task := NewRefreshExternalCollectionTask(ctx, req)
+	task := s.newTask(ctx, req)
 	task.preallocatedIDRange = req.GetPreAllocatedSegmentIds()
 	task.nextAllocID = task.preallocatedIDRange.Begin
 	task.parsedSpec = &externalspec.ExternalSpec{Format: "parquet"}
@@ -2414,7 +2468,7 @@ func (s *RefreshExternalCollectionTaskSuite) TestPreAllocatedSegmentIDs() {
 		CurrentSegments:        []*datapb.SegmentInfo{},
 	}
 
-	task := NewRefreshExternalCollectionTask(ctx, req)
+	task := s.newTask(ctx, req)
 
 	// Before Execute(), pre-allocated fields should not be initialized
 	s.Nil(task.preallocatedIDRange)
@@ -2449,7 +2503,7 @@ func (s *RefreshExternalCollectionTaskSuite) TestPreAllocatedIDAllocation() {
 		CurrentSegments:        []*datapb.SegmentInfo{},
 	}
 
-	task := NewRefreshExternalCollectionTask(ctx, req)
+	task := s.newTask(ctx, req)
 
 	// Manually initialize (simulating Execute)
 	task.preallocatedIDRange = idRange
@@ -2478,7 +2532,7 @@ func (s *RefreshExternalCollectionTaskSuite) TestMissingPreAllocatedIDs() {
 		// PreAllocatedSegmentIds is nil
 	}
 
-	task := NewRefreshExternalCollectionTask(ctx, req)
+	task := s.newTask(ctx, req)
 
 	// Execute should fail because pre-allocated IDs are missing
 	err := task.Execute(ctx)
@@ -2496,7 +2550,7 @@ func (s *RefreshExternalCollectionTaskSuite) TestPreExecute_NilSchema() {
 		StorageConfig:  &indexpb.StorageConfig{StorageType: "local"},
 		// Schema is nil
 	}
-	task := NewRefreshExternalCollectionTask(ctx, req)
+	task := s.newTask(ctx, req)
 	err := task.PreExecute(ctx)
 	s.Error(err)
 	s.Contains(err.Error(), "schema is nil")
@@ -2512,7 +2566,7 @@ func (s *RefreshExternalCollectionTaskSuite) TestPreExecute_NilStorageConfig() {
 		Schema:         &schemapb.CollectionSchema{},
 		// StorageConfig is nil
 	}
-	task := NewRefreshExternalCollectionTask(ctx, req)
+	task := s.newTask(ctx, req)
 	err := task.PreExecute(ctx)
 	s.Error(err)
 	s.Contains(err.Error(), "storage config is nil")
@@ -2528,7 +2582,7 @@ func (s *RefreshExternalCollectionTaskSuite) TestPreExecute_EmptyExternalSource(
 		StorageConfig: &indexpb.StorageConfig{StorageType: "local"},
 		// ExternalSource is empty
 	}
-	task := NewRefreshExternalCollectionTask(ctx, req)
+	task := s.newTask(ctx, req)
 	err := task.PreExecute(ctx)
 	s.Error(err)
 	s.Contains(err.Error(), "external source is empty")
@@ -2545,7 +2599,7 @@ func (s *RefreshExternalCollectionTaskSuite) TestPreExecute_InvalidExternalSpec(
 		Schema:         &schemapb.CollectionSchema{},
 		StorageConfig:  &indexpb.StorageConfig{StorageType: "local"},
 	}
-	task := NewRefreshExternalCollectionTask(ctx, req)
+	task := s.newTask(ctx, req)
 	err := task.PreExecute(ctx)
 	s.Error(err)
 	s.Contains(err.Error(), "failed to parse external spec")
@@ -2562,7 +2616,7 @@ func (s *RefreshExternalCollectionTaskSuite) TestPreExecute_UnsupportedFormat()
 		Schema:         &schemapb.CollectionSchema{},
 		StorageConfig:  &indexpb.StorageConfig{StorageType: "local"},
 	}
-	task := NewRefreshExternalCollectionTask(ctx, req)
+	task := s.newTask(ctx, req)
 	err := task.PreExecute(ctx)
 	s.Error(err)
 	s.Contains(err.Error(), "unsupported format")
@@ -2751,7 +2805,7 @@ func (s *RefreshExternalCollectionTaskSuite) TestFetchFragmentsFromExternalSourc
 		ExternalSpec:        `{"format":"parquet"}`,
 		ExploreManifestPath: "", // empty
 	}
-	task := NewRefreshExternalCollectionTask(ctx, req)
+	task := s.newTask(ctx, req)
 	task.parsedSpec = &externalspec.ExternalSpec{Format: "parquet"}
 
 	_, err := task.fetchFragmentsFromExternalSource(ctx)
@@ -2760,26 +2814,38 @@ func (s *RefreshExternalCollectionTaskSuite) TestFetchFragmentsFromExternalSourc
 }
 
 func (s *RefreshExternalCollectionTaskSuite) TestFetchFragmentsFromExternalSource_Success() {
-	paramtable.Init()
 	ctx := context.Background()
 	req := &datapb.RefreshExternalCollectionTaskRequest{
-		CollectionID:        s.collectionID,
-		TaskID:              s.taskID,
-		ExternalSource:      "s3:///bucket/path",
-		ExternalSpec:        `{"format":"parquet"}`,
-		ExploreManifestPath: "/manifests/explore.json",
-		FileIndexBegin:      0,
-		FileIndexEnd:        5,
-		StorageConfig:       &indexpb.StorageConfig{StorageType: "local", BucketName: "/tmp"},
+		CollectionID:         s.collectionID,
+		TaskID:               s.taskID,
+		ExternalSource:       "s3:///bucket/path",
+		ExternalSpec:         `{"format":"parquet"}`,
+		ExploreManifestPath:  "/manifests/explore.json",
+		FileIndexBegin:       0,
+		FileIndexEnd:         5,
+		StorageConfig:        &indexpb.StorageConfig{StorageType: "local", BucketName: "/tmp"},
+		TargetRowsPerSegment: 12345,
 	}
-	task := NewRefreshExternalCollectionTask(ctx, req)
+	task := s.newTask(ctx, req)
 	task.parsedSpec = &externalspec.ExternalSpec{Format: "parquet"}
 	task.columns = []string{"col1"}
 
 	mockFetch := mockey.Mock(packed.FetchFragmentsFromExternalSourceWithRange).
-		Return([]packed.Fragment{
-			{FragmentID: 0, FilePath: "f1.parquet", StartRow: 0, EndRow: 1000, RowCount: 1000},
-		}, nil).Build()
+		To(func(
+			ctx context.Context,
+			format string,
+			columns []string,
+			externalSource string,
+			storageConfig *indexpb.StorageConfig,
+			fileIndexBegin, fileIndexEnd int64,
+			exploreManifestPath string,
+			opts packed.ExternalFetchOptions,
+		) ([]packed.Fragment, error) {
+			s.Equal(req.GetTargetRowsPerSegment(), opts.RowLimit)
+			return []packed.Fragment{
+				{FragmentID: 0, FilePath: "f1.parquet", StartRow: 0, EndRow: 1000, RowCount: 1000},
+			}, nil
+		}).Build()
 	defer mockFetch.UnPatch()
 
 	frags, err := task.fetchFragmentsFromExternalSource(ctx)
@@ -2797,7 +2863,7 @@ func (s *RefreshExternalCollectionTaskSuite) TestBalanceFragmentsToSegments_Crea
 		ExternalSpec:   `{"format":"parquet"}`,
 		StorageConfig:  &indexpb.StorageConfig{StorageType: "local", BucketName: tmpDir, RootPath: tmpDir},
 	}
-	task := NewRefreshExternalCollectionTask(ctx, req)
+	task := s.newTask(ctx, req)
 	task.parsedSpec = &externalspec.ExternalSpec{Format: "parquet"}
 	task.columns = []string{"col1"}
 	task.preallocatedIDRange = &datapb.IDRange{Begin: 1, End: 100}
@@ -2828,7 +2894,7 @@ func (s *RefreshExternalCollectionTaskSuite) TestBalanceFragmentsToSegments_Cont
 		ExternalSpec:   `{"format":"parquet"}`,
 		StorageConfig:  &indexpb.StorageConfig{StorageType: "local", BucketName: tmpDir, RootPath: tmpDir},
 	}
-	task := NewRefreshExternalCollectionTask(ctx, req)
+	task := s.newTask(ctx, req)
 	task.parsedSpec = &externalspec.ExternalSpec{Format: "parquet"}
 	task.columns = []string{"col1"}
 	task.preallocatedIDRange = &datapb.IDRange{Begin: 1, End: 100}
@@ -2862,7 +2928,7 @@ func (s *RefreshExternalCollectionTaskSuite) TestBalanceFragmentsToSegments_CtxC
 		ExternalSpec:   `{"format":"parquet"}`,
 		StorageConfig:  &indexpb.StorageConfig{StorageType: "local", BucketName: tmpDir, RootPath: tmpDir},
 	}
-	task := NewRefreshExternalCollectionTask(ctx, req)
+	task := s.newTask(ctx, req)
 	task.parsedSpec = &externalspec.ExternalSpec{Format: "parquet"}
 	task.columns = []string{"col1"}
 	task.preallocatedIDRange = &datapb.IDRange{Begin: 1, End: 100}
@@ -2997,7 +3063,7 @@ func (s *RefreshExternalCollectionTaskSuite) TestBalanceFragmentsToSegments_Samp
 		},
 	}
 
-	task := NewRefreshExternalCollectionTask(ctx, req)
+	task := s.newTask(ctx, req)
 	task.preallocatedIDRange = req.GetPreAllocatedSegmentIds()
 	task.nextAllocID = task.preallocatedIDRange.Begin
 	task.parsedSpec = &externalspec.ExternalSpec{Format: "parquet"}
@@ -3051,7 +3117,7 @@ func (s *RefreshExternalCollectionTaskSuite) TestBalanceFragmentsToSegments_Samp
 		},
 	}
 
-	task := NewRefreshExternalCollectionTask(ctx, req)
+	task := s.newTask(ctx, req)
 	task.preallocatedIDRange = req.GetPreAllocatedSegmentIds()
 	task.nextAllocID = task.preallocatedIDRange.Begin
 	task.parsedSpec = &externalspec.ExternalSpec{Format: "parquet"}
@@ -3101,7 +3167,7 @@ func (s *RefreshExternalCollectionTaskSuite) TestBalanceFragmentsToSegments_Samp
 		},
 	}
 
-	task := NewRefreshExternalCollectionTask(ctx, req)
+	task := s.newTask(ctx, req)
 	task.preallocatedIDRange = req.GetPreAllocatedSegmentIds()
 	task.nextAllocID = task.preallocatedIDRange.Begin
 	task.parsedSpec = &externalspec.ExternalSpec{Format: "parquet"}
@@ -3158,7 +3224,7 @@ func (s *RefreshExternalCollectionTaskSuite) TestBalanceFragmentsToSegments_PerS
 		},
 	}
 
-	task := NewRefreshExternalCollectionTask(ctx, req)
+	task := s.newTask(ctx, req)
 	task.preallocatedIDRange = req.GetPreAllocatedSegmentIds()
 	task.nextAllocID = task.preallocatedIDRange.Begin
 	task.parsedSpec = &externalspec.ExternalSpec{Format: "parquet"}
@@ -3184,7 +3250,7 @@ func (s *RefreshExternalCollectionTaskSuite) TestBalanceFragmentsToSegments_PerS
 	defer m2.UnPatch()
 
 	// 3 fragments each above targetRowsPerSegment → 3 segments.
-	targetRows := paramtable.Get().DataNodeCfg.ExternalCollectionTargetRowsPerSegment.GetAsInt64()
+	targetRows := req.GetTargetRowsPerSegment()
 	rowsPerFragment := targetRows * 2 // force one segment per fragment
 	fragments := []packed.Fragment{
 		{FragmentID: 1, RowCount: rowsPerFragment},
@@ -3227,7 +3293,7 @@ func (s *RefreshExternalCollectionTaskSuite) TestBalanceFragmentsToSegments_PerS
 		},
 	}
 
-	task := NewRefreshExternalCollectionTask(ctx, req)
+	task := s.newTask(ctx, req)
 	task.preallocatedIDRange = req.GetPreAllocatedSegmentIds()
 	task.nextAllocID = task.preallocatedIDRange.Begin
 	task.parsedSpec = &externalspec.ExternalSpec{Format: "parquet"}
@@ -3256,7 +3322,7 @@ func (s *RefreshExternalCollectionTaskSuite) TestBalanceFragmentsToSegments_PerS
 		}).Build()
 	defer m2.UnPatch()
 
-	targetRows := paramtable.Get().DataNodeCfg.ExternalCollectionTargetRowsPerSegment.GetAsInt64()
+	targetRows := req.GetTargetRowsPerSegment()
 	rowsPerFragment := targetRows * 2
 	fragments := []packed.Fragment{
 		{FragmentID: 1, RowCount: rowsPerFragment},
@@ -3299,7 +3365,7 @@ func (s *RefreshExternalCollectionTaskSuite) TestBalanceFragmentsToSegments_PerS
 		},
 	}
 
-	task := NewRefreshExternalCollectionTask(ctx, req)
+	task := s.newTask(ctx, req)
 	task.preallocatedIDRange = req.GetPreAllocatedSegmentIds()
 	task.nextAllocID = task.preallocatedIDRange.Begin
 	task.parsedSpec = &externalspec.ExternalSpec{Format: "parquet"}
@@ -3314,7 +3380,7 @@ func (s *RefreshExternalCollectionTaskSuite) TestBalanceFragmentsToSegments_PerS
 		Return(nil, fmt.Errorf("all samples failed")).Build()
 	defer m2.UnPatch()
 
-	targetRows := paramtable.Get().DataNodeCfg.ExternalCollectionTargetRowsPerSegment.GetAsInt64()
+	targetRows := req.GetTargetRowsPerSegment()
 	rowsPerFragment := targetRows * 2
 	fragments := []packed.Fragment{
 		{FragmentID: 1, RowCount: rowsPerFragment},
@@ -3353,7 +3419,7 @@ func (s *RefreshExternalCollectionTaskSuite) TestBalanceFragmentsToSegments_Zero
 		},
 	}
 
-	task := NewRefreshExternalCollectionTask(ctx, req)
+	task := s.newTask(ctx, req)
 	task.preallocatedIDRange = req.GetPreAllocatedSegmentIds()
 	task.nextAllocID = task.preallocatedIDRange.Begin
 	task.parsedSpec = &externalspec.ExternalSpec{Format: "parquet"}
@@ -3413,7 +3479,7 @@ func (s *RefreshExternalCollectionTaskSuite) TestBalanceFragmentsToSegments_Insu
 		Schema:                 &schemapb.CollectionSchema{},
 	}
 
-	task := NewRefreshExternalCollectionTask(ctx, req)
+	task := s.newTask(ctx, req)
 	task.preallocatedIDRange = req.GetPreAllocatedSegmentIds()
 	task.nextAllocID = task.preallocatedIDRange.Begin
 	task.parsedSpec = &externalspec.ExternalSpec{Format: "parquet"}
@@ -3442,7 +3508,7 @@ func (s *RefreshExternalCollectionTaskSuite) TestBalanceFragmentsToSegments_Mani
 		Schema:                 &schemapb.CollectionSchema{},
 	}
 
-	task := NewRefreshExternalCollectionTask(ctx, req)
+	task := s.newTask(ctx, req)
 	task.preallocatedIDRange = req.GetPreAllocatedSegmentIds()
 	task.nextAllocID = task.preallocatedIDRange.Begin
 	task.parsedSpec = &externalspec.ExternalSpec{Format: "parquet"}
@@ -3477,7 +3543,7 @@ func (s *RefreshExternalCollectionTaskSuite) TestBalanceFragmentsToSegments_Cont
 		Schema:                 &schemapb.CollectionSchema{},
 	}
 
-	task := NewRefreshExternalCollectionTask(ctx, req)
+	task := s.newTask(ctx, req)
 	task.preallocatedIDRange = req.GetPreAllocatedSegmentIds()
 	task.nextAllocID = task.preallocatedIDRange.Begin
 	task.parsedSpec = &externalspec.ExternalSpec{Format: "parquet"}
@@ -3520,7 +3586,7 @@ func (s *RefreshExternalCollectionTaskSuite) TestBalanceFragmentsToSegments_Ensu
 		Schema:                 &schemapb.CollectionSchema{},
 	}
 
-	task := NewRefreshExternalCollectionTask(ctx, req)
+	task := s.newTask(ctx, req)
 	task.preallocatedIDRange = req.GetPreAllocatedSegmentIds()
 	task.nextAllocID = task.preallocatedIDRange.Begin
 	task.parsedSpec = &externalspec.ExternalSpec{Format: "parquet"}
@@ -3561,7 +3627,7 @@ func (s *RefreshExternalCollectionTaskSuite) TestBalanceFragmentsToSegments_Empt
 		},
 	}
 
-	task := NewRefreshExternalCollectionTask(ctx, req)
+	task := s.newTask(ctx, req)
 	task.preallocatedIDRange = req.GetPreAllocatedSegmentIds()
 	task.nextAllocID = task.preallocatedIDRange.Begin
 	task.parsedSpec = &externalspec.ExternalSpec{Format: "parquet"}
@@ -3619,7 +3685,7 @@ func (s *RefreshExternalCollectionTaskSuite) TestBalanceFragmentsToSegments_Cont
 				ExternalSpec:           `{"format":"parquet"}`,
 				Schema:                 &schemapb.CollectionSchema{},
 			}
-			task := NewRefreshExternalCollectionTask(ctx, req)
+			task := s.newTask(ctx, req)
 			task.preallocatedIDRange = req.GetPreAllocatedSegmentIds()
 			task.nextAllocID = task.preallocatedIDRange.Begin
 			task.parsedSpec = &externalspec.ExternalSpec{Format: "parquet"}
@@ -3654,7 +3720,7 @@ func (s *RefreshExternalCollectionTaskSuite) TestBalanceFragmentsToSegments_Cont
 		ExternalSpec:           `{"format":"parquet"}`,
 		Schema:                 &schemapb.CollectionSchema{},
 	}
-	task := NewRefreshExternalCollectionTask(ctx, req)
+	task := s.newTask(ctx, req)
 	task.preallocatedIDRange = req.GetPreAllocatedSegmentIds()
 	task.nextAllocID = task.preallocatedIDRange.Begin
 	task.parsedSpec = &externalspec.ExternalSpec{Format: "parquet"}
@@ -3686,7 +3752,7 @@ func (s *RefreshExternalCollectionTaskSuite) TestBalanceFragmentsToSegments_Func
 			},
 		},
 	}
-	task := NewRefreshExternalCollectionTask(ctx, req)
+	task := s.newTask(ctx, req)
 	task.preallocatedIDRange = req.GetPreAllocatedSegmentIds()
 	task.nextAllocID = task.preallocatedIDRange.Begin
 	task.parsedSpec = &externalspec.ExternalSpec{Format: "parquet"}
@@ -3812,7 +3878,7 @@ func (s *RefreshExternalCollectionTaskSuite) TestBalanceFragmentsToSegments_Adds
 		},
 	}
 
-	task := NewRefreshExternalCollectionTask(ctx, req)
+	task := s.newTask(ctx, req)
 	task.preallocatedIDRange = req.GetPreAllocatedSegmentIds()
 	task.nextAllocID = task.preallocatedIDRange.Begin
 	task.parsedSpec = &externalspec.ExternalSpec{Format: "parquet"}
@@ -3865,7 +3931,7 @@ func (s *RefreshExternalCollectionTaskSuite) TestBalanceFragmentsToSegments_Pass
 		},
 	}
 
-	task := NewRefreshExternalCollectionTask(ctx, req)
+	task := s.newTask(ctx, req)
 	task.preallocatedIDRange = req.GetPreAllocatedSegmentIds()
 	task.nextAllocID = task.preallocatedIDRange.Begin
 	// Parse the spec so the balance path has an in-memory struct to work with.
@@ -3933,7 +3999,7 @@ func (s *RefreshExternalCollectionTaskSuite) TestBalanceFragmentsToSegments_NilS
 		},
 	}
 
-	task := NewRefreshExternalCollectionTask(ctx, req)
+	task := s.newTask(ctx, req)
 	task.preallocatedIDRange = req.GetPreAllocatedSegmentIds()
 	task.nextAllocID = task.preallocatedIDRange.Begin
 	task.parsedSpec = &externalspec.ExternalSpec{Format: "parquet"}

pkg/proto/data_coord.proto

@@ -1447,6 +1447,7 @@ message RefreshExternalCollectionTaskRequest {
   int64 file_index_begin = 12;            // File range start for this task
   int64 file_index_end = 13;              // File range end for this task
   int64 partitionID = 14;                 // target partition for manifest path
+  int64 target_rows_per_segment = 15;     // target row count for segment organization
 }
 
 message RefreshExternalCollectionTaskResponse {