Bulk-inserting 4 Parquet files (~2.47 GB, 641k rows) into a 2-DataNode cluster took 759 s (~12.7 min), with ~90% of that time (~681 s) spent inside a single DataNode pod.

[saikishore143]

Observed Latency

Run Tasks Wall-clock Bottleneck
Baseline 4 tasks (4 files) 759.4 s (~12.7 min) All 4 tasks → 1 pod, 2nd pod idle
Scale-up 20 tasks (same files) 1057.5 s (~17.6 min) 15/5 split across 2 pods; heavier pod drives time

1. Is pickNode first-fit behavior intentional? Is there a plan to move to least-loaded scheduling?
2. Is there a recommended way to tune concurrencyPerCPUCore to force earlier spillover to the second pod?
3. Are there any other knobs in milvus.yaml that could reduce per-task parse/flush latency for large Parquet files?

[yhmo]

We can treat 4 parquet files as one import request like this:

from pymilvus.bulk_writer import bulk_import
resp = bulk_import(
    url=url,
    collection_name="xxx",
    files=[['1.parquet'],['2.parquet'],['3.parquet'],['4.parquet'],],
)

We can also call bulk_import() 4 times to create 4 import requests:

bulk_import(
    url=url,
    collection_name="xxx",
    files=[['1.parquet']],
)

bulk_import(
    url=url,
    collection_name="xxx",
    files=[['2.parquet'],],
)

bulk_import(
    url=url,
    collection_name="xxx",
    files=[['3.parquet'],],
)

bulk_import(
    url=url,
    collection_name="xxx",
    files=[['4.parquet'],],
)

For each import request, Milvus does two phases:

1. PreImport: scan files, estimate rows/sizes/stats
2. Import: actually read rows, hash, append system fields, sync segments

Two phases group files differently.

1. PreImport grouping is by a fixed file-count chunk size, in milvus.yaml:

dataCoord:
  import:
    filesPerPreImportTask = 2

So with 4 files in one import request, PreImport usually becomes:

• task A: 2 files
• task B: 2 files

If 4 import requests, each request has one file, Preimport becomes:

• task A: 1 file
• task B: 1 file
• task C: 1 file
• task D: 1 file

Different Preimport tasks can be scheduled to different DataNodes.
So the cluster may indeed have used both nodes in PreImport.

2. The real Import phase
   The Import phase does not preserve the PreImport file split. It regroups files here: internal/datacoord/import_util.go:381 — RegroupImportFiles(...), this method split files according to max segment size configured in milvus.yaml:

dataCoord:
  segment:
    maxSize: 1024

or diskSegmentMaxSize if DiskIndex is enabled:

dataCoord:
  segment:
    diskSegmentMaxSize: 2048

The regrouping rule is based on total estimated memory size per task, not “number of files per DataNode”.
If 4 files are in one import request, and dataCoord.segment.maxSize is 2GB, or diskindex is enabled, the RegroupImportFiles() might group 4 files into one task, which can be processed by only one data node.

If you could provide the runtime log for the milvus cluster, we can get more information. You can search for the keywords "add new import task" in the log to observe how many import tasks were created.

A quick workaround: call bulk_import() 4 times to create 4 import requests.

[xiaofan-luan]

I think this is a issue.

might be able to fix this.

[saikishore143]

Thanks @yhmo , @xiaofan-luan quick update with new data from our cluster after re-running with 20 concurrent RPCs and capturing DataCoord logs.

Context and goal: We're adopting bulk_insert as a new ingestion path. End goal: bring bulk-insert wall-clock from ~30+ min down to 2–3 min. From this run the surprise was that
index build dominates per-job latency (~76% of jobTimeCost/total), and the Milvus cluster has far more capacity than the workload is using — DataCoord and DataNode utilization
both look minimal in Grafana. So the question isn't whether to scale, it's why existing capacity isn't being absorbed by IndexBuild fan-out.

What we ran

Both the original 4-RPC run and the 20-RPC scale-up already use the second form — N separate do_bulk_insert RPCs, one file each:

# Production transform and perf probe — for every parquet:
utility.do_bulk_insert(collection_name=..., partition_name=..., files=[one_key])

datacoord.log
meta.json
datacoord.log
meta.json
summary.csv

20 RPCs → 20 jobIDs. RegroupImportFiles has no opportunity to merge across files within any request, so the collapse-to-one-task scenario doesn't apply to our runs as-shaped.

Cluster shape (capacity, from Grafana during run window)

• DataCoord: 2 replicas, 24 cores / 110 GiB each = ~48 cores / 220 GiB total
• DataNode (data-worker): 2 replicas, 14 cores / 118 GiB each, 56 import slots each per peek slots done
• IndexCoord: 2 replicas, 10 cores / 16 GiB each
• IndexNode (index-worker): 26 replicas, 14 cores / 60 GiB each = ~364 cores / ~1.56 TiB total
• Milvus: 2.5
• Collection shardsNum (vchannels): 4

Observed CPU/memory utilization across DataCoord, DataNode, and IndexCoord during the run was single-digit % of limits. We don't yet have per-pod IndexNode CPU traces for the run window —
that's a follow-up.

Input

• 4 distinct Snappy parquets × 5 repeats = 20 files, each ~640–690 MB / 128K–175K rows
• Total: ~3.2M rows / ~12 GB compressed

Per-RPC fan-out (from import segments info log lines)

Each do_bulk_insert RPC produces 4 segments — one per vchannel:

jobID=...090451   175,093 rows
  channel v0  segmentID=...283271  Rows=43,916  Size=193,887,250  (185 MiB)
  channel v1  segmentID=...283272  Rows=43,817  Size=193,450,158  (184 MiB)
  channel v2  segmentID=...283269  Rows=43,652  Size=192,721,701  (184 MiB)
  channel v3  segmentID=...283270  Rows=43,708  Size=192,968,934  (184 MiB)
  totalSize=773,028,043 (737 MiB)

Same shape across the 4 jobIDs we captured: per-segment size 170–185 MiB, segments-per-job = vchannel count = 4. For 20 RPCs that's ~80 segments → ~80 IndexBuild tasks queued
against 26 IndexNode pods (~3 tasks per pod if perfectly fanned out).

Per-job timing — from Milvus's own jobTimeCost/

• jobID ...090451 — buildIndex=26m 49s · total=33m 49s · residual=7m 00s
• jobID ...090474 — buildIndex=25m 27s · total=33m 51s · residual=8m 23s
• jobID ...090439 — buildIndex=25m 47s · total=33m 57s · residual=8m 09s
• jobID ...090463 — buildIndex=25m 29s · total=33m 58s · residual=8m 29s

buildIndex is 75–80% of total (~26 min of ~34 min) per import job. The 6–8 min residual covers queueing + preimport + import + sort combined; we couldn't split it further because
the captured DataCoord log window missed the earlier jobTimeCost/<stage> lines.

20-RPC wall-clock distribution (driver side)

• Submit window: 0.53 s (tight, DataCoord saw them concurrent)
• Time-to-Started (all 20): 5.55 – 5.78 s
• min per-task elapsed: 972 s
• p50: 1,215 s
• max: 2,045 s (= wall-clock)
• Σ elapsed / wall-clock: 16.8×

2.1× spread on identical-sized inputs — queueing/contention somewhere downstream of submit, not file size.

What this changes about the original question

The original framing was "single-pod pinning on DataNode" because the 4-RPC run's Grafana showed only one DataNode pod active. With the 20-RPC data + the cluster capacity above, that
framing is incomplete:

1. Index build, not data ingestion, dominates per-job latency. ~26 min out of ~34 min per job runs on IndexNode, not DataNode. Even a perfect 2-pod scatter on DataNode would leave ~26
   min/job of build untouched. This is the main blocker to our 2–3 min target.
2. IndexNode capacity is not the bottleneck either — dispatch might be. 26 IndexNode pods (~364 cores) against ~80 IndexBuild tasks should comfortably fan out 3-deep per pod. If
   buildIndex=26 min is genuinely pure CPU build on a saturated pod, that's an intrinsic-cost question; if it's queue wait + a much shorter actual build, that's a dispatch / scheduler
   question.
3. shardsNum=4 multiplies the IndexBuild fan-out. One bulk_insert RPC → 4 segments → 4 IndexBuild tasks. For a write-heavy / search-light Day0 collection this may be more shards than
   we need.

Specific questions

1. What controls IndexBuild fan-out across IndexNode replicas in 2.5? Even with 26 IndexNode pods(scaled up from min pods 2 to 26 during our run) and ~80 segments, we observed ~26 min buildIndex per job. Is this intrinsic
   per-segment build cost on a saturated pod, or is it queue wait because IndexCoord isn't dispatching across all 26? Knobs we suspect:
   • indexCoord.scheduler.interval — dispatch tick
   • indexNode.scheduler.buildParallel — concurrent builds per IndexNode
   • Any per-collection or per-shard serial-build constraint?
2. For a Day0-heavy ingest path, what shardsNum would you recommend? Today we get 4 segments per RPC; would 1–2 shards typically net out as lower wall-clock (fewer parallel builds but
   each build larger), or higher?
3. Do you recommend any changes to reduce the overall latency around datacord and index node as my ultimate goal is to reduc e the latency to 2 mins rather than 30 mins

Sharing milvus logs and test summary for your review

[saikishore143]

Update — third 20-RPC run with full per-stage jobTimeCost/<stage> telemetry. The DataNode pinning reproduces, and we now have a clean view of where the time is going.**

Same workload as previous runs (20 RPCs × one ~640–690 MB parquet, 4-vchannel collection). This run captured DataCoord scheduler logs from before submit and pulled jobTimeCost/<stage> for
all 20 jobs.

Wall-clock: 888 s (~14m 48s), 20/20 Completed

Per-stage breakdown (median across 20 jobs)

• preimport (DataNode) — 1m 53s
• import (DataNode) — 5m 25s ← 40% of per-job time
• stats (DataNode) — 46s
• buildIndex (IndexNode) — 4m 20s ← 32% of per-job time
• total — 13m 35s (median) / 14m 45s (max)

import + buildIndex together are ~72% of per-job latency. The DataNode side (preimport + import + stats) sums to ~8 min — slightly larger than the IndexNode side (~4–5 min) for
this run.

DataNode scheduling — pinning reproduces a third time

From import_scheduler.go:204 "preimport task start to execute" and matching ImportTask lines:

• nodeID=10668: 33 tasks (82.5%)
• nodeID=10667: 7 tasks (17.5%)

Both DataNodes have identical limits (14 cores / 118 GiB, 56 slots each) and both showed full slot availability at submit time per peek slots done. Three runs in a row produced similar
skews:

• Run 1 (2026-06-29 11:27 UTC) — Grafana showed second DataNode at 0 CPU activity (no log capture)
• Run 2 (13:36 UTC) — 37/3 (92/8 split)
• Run 3 (14:10 UTC, this run) — 33/7 (83/17 split)

Pinning is consistent across runs. Grafana for this run confirms one DataNode pod was heavily loaded while the other ran at ~20%.

IndexNode side is healthy

For contrast: 30 IndexNode pods pre-warmed at submit, stayed at 30 the entire run (no HPA churn). Work fanned across 28 of 30 pods. Peak busy slots 2,704 of 6,720 = 40%
utilization. Two pods sat idle. Index dispatch is not the bottleneck for this workload size.

Why DataNode pinning matters most for our latency

If import were split evenly (50/50) instead of 83/17, the worst-case DataNode-side time would roughly halve — ~2.5 min off wall-clock with no other changes. That's by far the largest
single-lever improvement we have without architectural changes on our end.

Three runs on identical workload, no code or infra changes

• Run 1: 2,045 s wall-clock, ~26 min median buildIndex (cold IndexNode pool)
• Run 2: 1,105 s wall-clock, ~5 min median buildIndex (pool ramping)
• Run 3: 888 s wall-clock, 4m 19s median buildIndex (pool steady at 30)

The 57% wall-clock reduction across runs came entirely from IndexNode HPA warming up between runs. With that side stable, DataNode pinning is the remaining ceiling for the workload
as-shaped.
datacoord-import-tasks.csv
datacoord-job-segments.csv
datacoord-job-timing.csv
datacoord-peek-slots.csv
datacoord-import-tasks.csv
[datacoord-job-segments.csv](https://github.com/user-attachments/files/29470559/dat
datacoord-peek-slots.csv
meta.json
summary.csv
acoord-job-segments.csv)
datacoord-job-timing.csv

[xiaofan-luan]

@xiaocai2333 is working on it.

We won't fix this in 2.5. in 2.6 and master we can round robin assign import task.

[saikishore143]

Thank you, @xiaofan-luan and @xiaocai2333

Do we have an expected timeline for rolling this round-robin change into the 2.6 release? Also, is there any way to bypass or work around this issue in 2.5 within my environment until the fix becomes available?

My primary goal is to reduce latency on the DataNode side, and I'm open to tuning configurations as a temporary mitigation.

Additionally, do you have any recommendations for further reducing the buildIndex latency beyond the current 4.32 minutes? Any configuration changes or best practices that could help improve index build performance would be greatly appreciated.

[xiaofan-luan]

Hi @saikishore143
please connect me at james.luan@zilliz.com

definitely happy to learn about your use cases and offer some advice.

[dlanov]

The maintainer replies explained what happens but not how to tune it. Here are the specific knobs.

** Why all 4 tasks land on one DataNode (first-fit scheduling)**

The current pickNode in DataCoord uses first-fit: it assigns each import task to the first DataNode that has a free slot, evaluated sequentially. Since all 4 tasks are created nearly simultaneously and DataNode-1 has enough free slots to absorb them all, DataNode-2 never gets offered a task. This is intentional in 2.5.x and is confirmed being fixed to round-robin in 2.6.

Forcing spillover in 2.5.x — the effective knobs:

1. Use 4 separate bulk_import() calls (already mentioned by maintainer)

This is the most reliable immediate workaround. Each call creates an independent import job; the scheduler assigns them one at a time, and with 4 sequential assignments the first-fit scheduler will fill DataNode-1's slots and spill to DataNode-2.

2. Lower dataNode.importTask.maxConcurrentOperations

This is the direct lever on how many import tasks a single DataNode accepts simultaneously. Default is typically 4 × CPU cores. Halving it forces the scheduler to see DataNode-1 as "full" sooner:

  # milvus.yaml
  dataNode:
    importTask:
      maxConcurrentOperations: 2   # lower from default to force spillover

With this set to 2, a 4-task batch will split 2/2 across your two DataNodes.

3. Lower dataCoord.segment.maxSize to force more tasks

Your 4 files total ~2.47 GB. With maxSize at 2 GB (default), RegroupImportFiles() packs them into as few tasks as possible, often 1-2. Lowering maxSize creates more granular tasks that spread more naturally:

  dataCoord:
    segment:
      maxSize: 512      # MB, down from 2048
      diskSegmentMaxSize: 512

This also benefits from creating more parallel flush segments, which reduces per-task flush latency.

4. Tune parse/flush parallelism per task

  dataNode:
    importTask:
      maxImportFileSizeInGB: 1.0    # split large files at parse time
    flush:
      insertBufSize: 16777216       # 16MB insert buffer, flush more frequently

Why your 20-task run was slower:
Splitting into 20 tasks increased scheduling and coordination overhead without improving DataNode utilisation — the first-fit scheduler still front-loaded DataNode-1 with 15 tasks. More tasks only helps when combined with a concurrency cap that forces spillover. With round-robin scheduling in 2.6 this changes: more tasks will genuinely parallelise across both nodes.

Recommended config for your setup today (2.5.x):

  dataCoord:
    segment:
      maxSize: 512
      diskSegmentMaxSize: 512
  dataNode:
    importTask:
      maxConcurrentOperations: 2

Then call bulk_import() once per file (4 calls). With these settings each DataNode accepts 2 tasks, your 4 tasks split evenly, and each task processes ~620 MB instead of ~2.47 GB.