mirror of
https://github.com/milvus-io/milvus.git
synced 2026-10-28 17:05:53 +08:00
Jeng.Gwan
82a1dd7714
issue: #53333 ## Summary Implements the Import V3 reshard path: source files are parsed and sorted once at first read, fragments are planned into final segments across files and tasks, and the Import V3 merge produces segments that already satisfy the sort-compaction ordering contract, so the post-import SortCompaction pass is no longer run for jobs on this path (#50593). Rollout gate: `dataCoord.import.enableImportV3` (default `false`). The execution version is selected once at job create, persisted in the import WAL (`msgpb.ImportMsg.version`) and the job catalog, and never re-selected on retry, restart or config change. L0 imports and in-flight/legacy ImportTaskV2 jobs keep the old path. ## Mechanism - **DataCoord** - Import V3 job state machine: `Pending -> [PreImporting -> AssigningIDRange] -> Resharding -> Planning -> Importing -> IndexBuilding -> Uncommitted -> Committing -> Completed` (backup/binlog imports skip the count-only preimport and ID-range stages). - Planning aggregates reshard fragments by `(vchannel, partition)` bucket across all reshard tasks, packs them into segment plans in canonical order (sequential next-fit against the expected segment size) and preallocates the formal segment IDs. Reshard file grouping is a deterministic best-fit-decreasing packing capped by `dataCoord.import.maxSizeInMBPerImportTask`. - Per-file literal `IDRange` allocation reuses the two-phase mechanism added in #53544; files that already carry a range (replicated/forwarded WAL bodies) are authoritative and are not re-allocated. - Result acceptance applies worker results to the preallocated segments with storage-version self-healing on retry (a retry converges onto the current storage version instead of failing on a stale version fence). - Orphan reconciliation in the garbage collector is extended to the Import V3 intermediate paths, plus terminal-job GC derived from durable facts. - **DataNode** - `ReshardTask` parses each source file exactly once: field normalization, PK/RowID materialization, bucket routing and per-batch sort under the shared sort order (`importutilv2.SortFieldIDs`), written as temporary fragments under `{root}/import_v3/{job}/reshard/{task}/`. - `ImportTaskV3` merges fragments into final segments with a one-head k-way merge (direct fan-in `min(dataCoord.import.fragmentMergeFanIn, len(fragments))`, clamped to `[2,1024]`, default **128**; hierarchical merge when the plan has more inputs), bounded by a working-set slot estimate and DataNode slot admission. - Count-only preimport reads parquet/numpy row counts from file metadata (bounded by `dataCoord.import.parquetFooterMaxSize`) instead of scanning. - Merge intermediates go to node-local Arrow IPC / local disk, so the object store sees each fragment exactly once. - Transient IO and local-storage errors stay retryable via the shared checker denylist; non-transient run errors fail the task. ## Behavior boundaries - No user-facing API change: job creation, CommitImport/RollbackImport, auto-commit and visibility behavior are preserved; only the internal execution pipeline is split into `ReshardTask`/`ImportTaskV3` with per-run `run_id` isolation for late results. - L0 imports, V2 backup/binlog restore jobs and legacy in-flight jobs are unchanged. - **Changes observable even with the gate off** (reviewer awareness): - `dataCoord.import.maxSizeInMBPerImportTask` default 16384 -> 8192. It is consumed by the legacy V2 file regrouping (`RegroupImportFiles`), so V2 imports on collections where `segmentMaxSize x P x V` lands in (8 GiB, 16 GiB] now produce roughly twice as many, half-sized file groups/tasks. Set the knob back explicitly to keep the old grouping. - CSV import no longer silently truncates on a mid-file read/parse error: the reader used to drop every non-EOF error (`csv/reader.go`); it now fails the job. - CSV header / parquet file-reader failures move from `ErrImportSysFailed` (2101) to `ErrImportFailed` (2100, Input) or the preserved IO code; failure reason strings change accordingly. Retry classification is unchanged. - Existing datacoord metrics gained labels: `ImportJobLatency`/`ImportJobs` (`job_version`), `ImportTaskLatency` (`task_type`). - The V2 collection-drop check no longer tries to fail `Committing` jobs. The job state outcome is unchanged (`UnfailableJobStates` already refuses `Committing -> Failed`); this only removes one futile catalog write plus a warning log per GC tick. ## CMEK / TEXT columns Column groups are encrypted with the target collection's CMEK zone; fragment readers use the read/source plugin context. **TEXT columns are deliberately written plaintext.** The TEXT read chain has no key retriever wiring yet: segcore reads V3 column groups through milvus_storage's `api::Reader` without a key retriever, and the Go TEXT-decode segment reader has no FFI entry point for one, so an encrypted TEXT segment would be written but never read back (both query and the compaction TEXT rewrite). The plugin context is still threaded all the way down into the TEXT-aware writer; only its two call sites (`internal/flushcommon/syncmgr/pack_writer_v3.go`, `internal/storage/binlog_record_writer.go`) intentionally pass `nil`, with a `TODO:` to re-enable encryption once the read chain supports decryption. This matches the design doc's Stage 3 note, and only matters when `common.storage.useLoonFFI=true` and the collection has an encryption zone. ## Design design doc: `docs/design-docs/design_docs/20260921-import-v3-reshard-design.md` ## Validation - [x] Compiles with `-tags dynamic,test` (`internal/datacoord`, `internal/datanode/importv3`, `internal/storage`, `internal/flushcommon/syncmgr`, `internal/util/initcore`). - [x] `golangci-lint --build-tags dynamic,test` clean on the touched root packages and on `pkg/util/paramtable`. - [x] Unit tests pass: `internal/datanode/importv3`, `pkg/util/paramtable`, `cmd/tools/config -run TestYamlFile`. - [x] Integration test included: `tests/integration/import/import_v3_test.go`. - [ ] E2E: to be run on CI. --- # Import V3 (Reshard) vs Import V2 — Core Performance Metrics > The only difference between V3 and V2 is `enableImportV3`; everything else is identical (both use milvus-storage V3). ## TL;DR On the same dataset of 111 GB / 37.92M rows: **end-to-end time is 14%–79% faster** (the more shards/partitions, the bigger the gain), **the write chain is 41%–89% faster**, and **S3 request counts drop by 70%–99%**; moreover, **only V3 completes at 128 / 1024 partitions** (V2 fails because of the etcd 2 MB metadata limit). ## Test setup | Item | Value | |---|---| | Environment | 10 × 8C 32G datanodes | | Data | 11 jobs / **1100 parquet files / 111.31 GB / 37,924,512 rows** | | Schema | 13 fields; `dense_embedding` = **FLOAT16_VECTOR dim=512**, FLAT+IP index; **partition key = `pk_id`** (99.65% unique) | | Tiers | 10 tiers: 1x1 / 1x2 / 1x4 / 2x4 / 2x8 / 2x16 / 2x32 / 2x64 / 2x128 / 2x1024 | | Tier naming | **NxM = num_shards × number of partitions** (e.g. `2x64` = 2 vchannels / 64 partitions) | | Parameter | Value | | --- | --- | | `dataCoord.import.fragmentSizeInMB` | 128 | | `dataCoord.import.maxSizeInMBPerImportTask` | 8192 | | `dataCoord.import.reshardResidentBucketCap` | 16 | | `dataCoord.import.reshardFlushConcurrency` | 2 | | `dataCoord.import.reshardMemoryExpansionFactor` | 1.5 | | `dataCoord.import.fragmentMergeFanIn` | 128 | | `dataCoord.import.enableImportV3` | true | | `common.storage.useLoonFFI` | true | | `dataNode.import.readBufferSizeInMB` | 16 | | `dataNode.import.reshardSpillMaxStreams` | 128 | ## Core metrics ### ① Elapsed time  **Per-stage elapsed time (seconds; V3 and V2 have different stage sets; "–" = not applicable)** | Tier | Version | create | preimport | reshard | planning | importV3\* | importV2 | sort\* | index\* | wall | |---|---|---|---|---|---|---|---|---|---|---| | 1x1 | V2 | 1 | 50 | – | – | – | 112 | 127 | 480 | 835 | | 1x1 | V3 | 4 | 9 | 75 | ~0 | 43 | – | – | 497 | 643 | | 1x2 | V2 | 1 | 50 | – | – | – | 135 | 132 | 499 | 829 | | 1x2 | V3 | 4 | 9 | 73 | ~0 | 45 | – | – | 485 | 629 | | 1x4 | V2 | 1 | 46 | – | – | – | 100 | 139 | 488 | 780 | | 1x4 | V3 | 3 | 10 | 108 | ~0 | 50 | – | – | 489 | 669 | | 2x4 | V2 | 1 | 54 | – | – | – | 134 | 228 | 419 | 848 | | 2x4 | V3 | 4 | 9 | 132 | ~0 | 49 | – | – | 491 | 689 | | 2x8 | V2 | 1 | 52 | – | – | – | 308 | 341 | 336 | 1043 | | 2x8 | V3 | 3 | 10 | 233 | ~0 | 45 | – | – | 472 | 769 | | 2x16 | V2 | 1 | 50 | – | – | – | 602 | 294 | 277 | 1238 | | 2x16 | V3 | 4 | 10 | 446 | ~0 | 50 | – | – | 480 | 1004 | | 2x32 | V2 | 1 | 50 | – | – | – | 958 | 337 | 322 | 1681 | | 2x32 | V3 | 2 | 11 | 486 | ~0 | 64 | – | – | 464 | 1032 | | 2x64 | V2 | 1 | 48 | – | – | – | 4302 | 667 | 26 | 5058 | | 2x64 | V3 | 3 | 10 | 510 | ~0 | 53 | – | – | 483 | 1069 | | 2x128 | V2 | 1 | 50 | – | – | – | ≈7000 | ≈1330 | ≈25 | ≈8400 | | 2x128 | V3 | 2 | 11 | 542 | ~0 | 61 | – | – | 450 | 1075 | | 2x1024 | V2 | 1 | 50 | – | – | – | ≈105000 | ≈10700 | ≈25 | ≈116000 | | 2x1024 | V3 | 2 | 8 | 625 | ~0 | 115 | – | – | 275 | 1036 | V3 and V2 have different stage sets: V3 = preimport → reshard → importV3; V2 = preimport → importV2 → sort. Because `index` is dispatched as soon as its segments are ready: - V3's `index` overlaps `import` - V2's `index` overlaps `sort` Therefore: (1) the stages **cannot simply be summed** — the sum does not equal `wall` either, the gap being mainly index building; (2) the starred columns are stretched by the index tail (which is why V3's `import` is inflated). The last two V2 tiers actually **Failed** (etcd 2 MB metadata limit); their importV2 / sort / wall are **estimates derived from the source code** (segment count = T·V·P, see ③), not measurements. ### ② S3 access   | Tier | V2 read (ops/GB) | V3 read (ops/GB) | read drop | V2 write (ops/GB) | V3 write (ops/GB) | write drop | |---|---|---|---|---|---|---| | 1x1 | 115,195 / 129.3 | 17,758 / 130.8 | ↓84.6% | 76,535 / 215.6 | 22,686 / 218.5 | ↓70.4% | | 1x4 | 137,522 / 129.4 | 17,167 / 130.8 | ↓87.5% | 91,785 / 214.2 | 22,293 / 218.5 | ↓75.7% | | 2x8 | 255,808 / 130.7 | 17,263 / 130.7 | ↓93.3% | 170,905 / 214.8 | 21,912 / 214.7 | ↓87.2% | | 2x16 | 331,359 / 131.6 | 17,342 / 130.6 | ↓94.8% | 224,947 / 216.5 | 22,213 / 216.9 | ↓90.1% | | 2x32 | 654,440 / 131.2 | 18,695 / 130.5 | ↓97.1% | 440,787 / 219.8 | 22,087 / 215.5 | ↓95.0% | | 2x64 | 1.30M / 132.9 | 20,919 / 130.7 | ↓98.4% | 870,798 / 222.9 | 22,437 / 214.8 | ↓97.4% | | 2x128 | ≈2.61M / — | 28,786 / 130.7 | ↓98.9% | ≈1.75M / — | 22,708 / 215.4 | ↓98.7% | | 2x1024 | ≈20.86M / — | 104,040 / 130.0 | ↓99.5% | ≈14.02M / — | 59,070 / 216.2 | ↓99.6% | The **bytes** transferred are nearly identical on both sides and **barely change across tiers** (read ≈130 GB, write ≈215 GB, both ≈ the dataset size); **the entire difference is in the number of requests** — V2's sort compaction re-reads and re-writes the full dataset, so its request count is 1–2 orders of magnitude higher, whereas V3 has no sort and keeps request counts low and flat. ### ③ Segment shape   | Tier | V2 final segments | V3 final segments | V2 segment size p50 | V3 segment size p50 | V2 sort compactions | |---|---|---|---|---|---| | 1x1 | 197 | 172 | 846 MB | 965 MB | 197 | | 1x4 | 172 | 172 | 988 MB | 978 MB | 172 | | 2x8 | 176 | 176 | 1022 MB | 978 MB | 176 | | 2x16 | 352 | 192 | 511 MB | 976 MB | 352 | | 2x32 | 1344 | 192 | 128 MB | 985 MB | 1344 | | 2x64 | 2000 | 256 | 64 MB | 656 MB | 2000 | | 2x128 | ≈5,376 | 256 | ≈32 MB | 658 MB | ≈5,376 | | 2x1024 | ≈43,008 | 2048 | ≈4 MB | 82 MB | ≈43,008 | **V3 runs 0 compactions** and directly produces well-formed, large segments; V2 runs one sort compaction per segment after import (count = number of segments), so as partitions grow the data shatters into a large number of tiny segments (e.g. 2x64: 2000 segments × 64 MB). Fewer, larger segments are friendlier to subsequent query/index. ### ④ Memory (per datanode)  | Tier | V2 per-datanode peak | V3 per-datanode peak | |---|---|---| | 1x1 | 19.1 GB | 29.8 GB | | 1x2 | 22.9 GB | 28.5 GB | | 1x4 | 21.5 GB | 29.4 GB | | 2x4 | 26.2 GB | 29.9 GB | | 2x8 | 18.1 GB | 29.6 GB | | 2x16 | 15.1 GB | 28.6 GB | | 2x32 | 10.6 GB | 26.1 GB | | 2x64 | 11.6 GB | 25.8 GB | | 2x128 | — | 25.8 GB | | 2x1024 | — | 27.5 GB |  **Notes:** - **V3 peaks at 26–30 GB per datanode, slightly above V2 (10–26 GB)** — this is the cost of reshard holding bucket data resident in memory; in exchange there is no sort rewrite and none of V2's flood of tiny segments at high partition counts. - V3's peak is **stable and slightly decreasing across tiers** (≈26–30 GB) and **does not grow with partition count (up to 1024)**; the per-datanode usage curves (taking the busiest datanode) almost coincide across tiers — i.e. V3's reshard memory model keeps usage independent of the shard/partition count. - V2's peak actually *drops* at high partition counts (2x64: only 11.6 GB), but that is because the cost is pushed into sort/IO (see ② ③), not because it is more memory-efficient. Signed-off-by: xaxys <tpnnghd@163.com>