You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@skywalking.apache.org by li...@apache.org on 2026/05/21 08:56:38 UTC
(skywalking-banyandb) branch main updated: Introduce measure migration tool (#1138)
This is an automated email from the ASF dual-hosted git repository.
mrproliu pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
The following commit(s) were added to refs/heads/main by this push:
new a3d7a97a7 Introduce measure migration tool (#1138)
a3d7a97a7 is described below
commit a3d7a97a72a1b48fdaeb97fd5cc6084b4f81552a
Author: mrproliu <74...@qq.com>
AuthorDate: Thu May 21 16:56:33 2026 +0800
Introduce measure migration tool (#1138)
---
CHANGES.md | 1 +
banyand/Dockerfile | 1 +
banyand/Makefile | 2 +-
banyand/cmd/migration/MIGRATION.md | 458 ++++++
banyand/cmd/migration/README.md | 144 ++
banyand/cmd/migration/analyze.go | 186 +++
banyand/cmd/migration/copy.go | 189 +++
banyand/cmd/migration/example/plan-backup.yaml | 41 +
banyand/cmd/migration/example/plan-live.yaml | 67 +
banyand/cmd/migration/example/runner-pod.yaml | 80 ++
banyand/cmd/migration/main.go | 88 ++
banyand/cmd/migration/plan.go | 301 ++++
banyand/cmd/migration/plan_test.go | 290 ++++
banyand/cmd/migration/scripts/verify-data.sh | 113 ++
banyand/cmd/migration/verify.go | 339 +++++
banyand/internal/storage/version.go | 19 +
banyand/measure/migration_analyze.go | 934 ++++++++++++
banyand/measure/migration_copy.go | 1816 ++++++++++++++++++++++++
banyand/measure/migration_copy_test.go | 303 ++++
banyand/measure/migration_e2e_test.go | 598 ++++++++
banyand/measure/migration_schema.go | 446 ++++++
banyand/measure/migration_schema_test.go | 308 ++++
banyand/measure/migration_verify.go | 401 ++++++
banyand/measure/migration_verify_test.go | 209 +++
24 files changed, 7333 insertions(+), 1 deletion(-)
diff --git a/CHANGES.md b/CHANGES.md
index e998bf1b1..06f686659 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -54,6 +54,7 @@ Release Notes.
- Fix the `notifiedModRevision` watermark advancement in `SchemaRegistry.processInitialResourceFromProperty`, `handleWatchEvent` (DELETE branch), and `handleDeletion`. Previously `AdvanceNotified` was gated on `cache.Update` / `cache.Delete` returning true, but those methods compare `latestUpdateAt` (property timestamp) while the watermark tracks `modRevision` (etcd revision). When the property timestamp is stale (e.g. a no-op Update that doesn't change the measure spec), the cache rej [...]
- Fix the `modRevision` contract on no-op Update RPCs (`MeasureRegistryService.Update`, etc.). Previously `updateResource` detected unchanged content via `CheckerMap` and short-circuited without writing to the property store, but the caller had already fabricated `modRevision = time.Now().UnixNano()` and returned it. The returned revision never appeared in the property watch stream, so `AwaitRevisionApplied(R)` would hang. `updateResource` now returns `(int64, error)` — the existing pr [...]
- Add end-to-end observability for liaison internal queue pipelines with per-topic metrics for queue_sub and queue_pub, along with Grafana panels and troubleshooting docs.
+ - Introduce measure migration tool.
### Bug Fixes
diff --git a/banyand/Dockerfile b/banyand/Dockerfile
index b9e5ef940..cd0d53bcb 100644
--- a/banyand/Dockerfile
+++ b/banyand/Dockerfile
@@ -27,6 +27,7 @@ COPY build/bin/linux/${TARGETARCH}/banyand-server-${BINARYTYPE} /banyand
COPY build/bin/linux/${TARGETARCH}/banyand-backup-${BINARYTYPE} /backup
COPY build/bin/linux/${TARGETARCH}/banyand-restore-${BINARYTYPE} /restore
COPY build/bin/linux/${TARGETARCH}/banyand-lifecycle-${BINARYTYPE} /lifecycle
+COPY build/bin/linux/${TARGETARCH}/banyand-migration-${BINARYTYPE} /migration
COPY --from=certs /etc/ssl/certs /etc/ssl/certs
FROM build-${TARGETOS} AS final
diff --git a/banyand/Makefile b/banyand/Makefile
index af9b07330..92bf666f6 100644
--- a/banyand/Makefile
+++ b/banyand/Makefile
@@ -17,7 +17,7 @@
#
NAME := banyand
-BINARIES := $(NAME)-server $(NAME)-backup $(NAME)-restore $(NAME)-lifecycle
+BINARIES := $(NAME)-server $(NAME)-backup $(NAME)-restore $(NAME)-lifecycle $(NAME)-migration
IMG_NAME := skywalking-banyandb
diff --git a/banyand/cmd/migration/MIGRATION.md b/banyand/cmd/migration/MIGRATION.md
new file mode 100644
index 000000000..40dd2ae98
--- /dev/null
+++ b/banyand/cmd/migration/MIGRATION.md
@@ -0,0 +1,458 @@
+# Migration manual
+
+Operator-facing procedure to re-normalize **measure** data on a BanyanDB deployment when the on-disk state has drifted from what the runtime expects. Common triggers:
+
+- Historical segments left in a layout the current banyandb version no longer reads cleanly.
+- A botched lifecycle event (failed merge / compaction / shard topology change) that left duplicated boundary rows or misaligned segments.
+- A partial restore that landed measure data outside the active stage's `SegmentInterval` grid.
+- General "clean the historical measure tree without losing rows" maintenance — re-bucket every source row onto a fresh, grid-aligned target tree that the BanyanDB runtime can pick up unchanged.
+
+Two end-to-end procedures, depending on where the source data lives:
+
+- [Part 1 — Local single-host flow](#part-1--local-single-host-flow): the source is a backup snapshot downloaded to a workstation, the target is a fresh measure-data root you ship into the cluster afterwards.
+- [Part 2 — Kubernetes live in-place flow](#part-2--kubernetes-live-in-place-flow): the source is the live cluster's own data PVCs; a helper pod rewrites each PVC's measure tree in place, the operator swaps the migrated groups onto every PVC, then the data StatefulSets restart on top of the clean tree.
+
+For the tool's CLI surface, plan schema, and on-disk output layout, see [`README.md`](README.md).
+
+> **Scope:** the tool only migrates **measure** data. Stream / trace / property catalogs and `sw_metadata` are out of scope — they stay in place (Part 2) or remain on the source backup (Part 1).
+
+---
+
+# Part 1 — Local single-host flow
+
+This part assumes a local backup snapshot is available, the snapshot contains a hot / schema-server node so the `schema-property` catalog is present, and the target measure-data root will be consumed by a BanyanDB process that is **not running** during the migration.
+
+## 1. Build the migration binary
+
+Build via the standard banyandb Makefile — it produces `build/bin/banyand-migration`:
+
+```bash
+cd <banyandb-repo-root>/banyand
+make banyand-migration
+ls build/bin/banyand-migration
+```
+
+Confirm the `copy` subcommand is wired in:
+
+```bash
+./build/bin/banyand-migration copy --help
+```
+
+## 2. Stage the backup snapshot locally
+
+Download a backup snapshot for the date you intend to migrate (use whatever transport fits your environment: `gsutil cp`, `aws s3 cp`, `rclone`, etc.). After the download, the local layout should look like:
+
+```
+/tmp/banyandb-backup/
+├── hot-0/2026-05-19/measure/<group>/seg-*/shard-*/<partID>/
+├── hot-0/2026-05-19/schema-property/_schema/shard-*/
+├── hot-1/2026-05-19/...
+├── warn-0/2026-05-19/measure/...
+├── warn-1/2026-05-19/...
+└── cold-0/2026-05-19/measure/...
+```
+
+The plan loader will pick up `measure` from every node directory and `schema-property` from whichever hot / schema-server nodes carry it.
+
+## 3. Write a migration plan
+
+Copy [`example/plan-backup.yaml`](example/plan-backup.yaml) and adapt every path / date / group / target to your environment. The shipped sample uses `/tmp/...` paths so it works out of the box on a fresh workstation:
+
+```yaml
+staging_dir: /tmp/banyandb-staging
+
+source:
+ backup:
+ root: /tmp/banyandb-backup
+ date: 2026-05-19
+
+groups:
+ - sw_metricsMinute
+ - sw_metricsHour
+ - sw_metricsDay
+
+entries:
+ - { stage: hot, target: /tmp/banyandb-copy/hot/measure-data, nodes: [hot-0, hot-1] }
+ - { stage: warm, target: /tmp/banyandb-copy/warm/measure-data, nodes: [warn-0, warn-1] }
+ - { stage: cold, target: /tmp/banyandb-copy/cold/measure-data, nodes: [cold-0] }
+```
+
+Make sure each `entries[*].target` directory is absent or empty (the tool aborts otherwise), `staging_dir` (if set) is absent or empty, and `<source.backup.root>/<source.backup.date>` exists.
+
+## 4. Run the migration
+
+```bash
+./build/bin/banyand-migration copy --copy-config /tmp/migration-plan.yaml
+```
+
+The tool prints timestamped progress per group / entry / part. A representative head of the live log:
+
+```
+[migration] GOMEMLIMIT set to 17.0GiB (85% of cgroup limit 20.0GiB) at 11:57:54
+2026/05/20 11:57:55 [migration] group sw_metricsMinute: building union sidx from 5 source dir(s) merged across all entries
+2026/05/20 11:58:44 [migration] group sw_metricsHour: building union sidx from 5 source dir(s) merged across all entries
+2026/05/20 11:59:35 [migration] group sw_metricsDay: building union sidx from 5 source dir(s) merged across all entries
+2026/05/20 12:00:49 [migration] entry [1/5] stage=hot nodes=[hot-0] target=/tmp/banyandb-copy/hot/measure-data
+2026/05/20 12:00:49 [migration] entry [1/5] stage=hot group sw_metricsMinute: 1 source dir(s), 34 parts, interval=day×1 — writing target (target="/tmp/banyandb-copy/hot/measure-data/sw_metricsMinute", union sidx="…/staging/groups/sw_metricsMinute/sidx")
+2026/05/20 12:00:49 [migration] entry [1/5] stage=hot group sw_metricsMinute: part 1/34 (2.9%) done seg-20260519/shard-0/0000000000002c4c rows=18907 targetParts=1
+...
+2026/05/20 12:00:50 [migration] entry [1/5] stage=hot group sw_metricsMinute: done rows=7706398 segs=2 srcParts=34 targetParts=34
+```
+
+On error the tool aborts and removes the `staging_dir`. The partially-written target directories remain — clean them up manually before re-running.
+
+To attach `pprof` mid-run, pass `--pprof-addr 127.0.0.1:6060` on the root command (before the `copy` subcommand) and then `go tool pprof -seconds=60 http://127.0.0.1:6060/debug/pprof/profile`.
+
+## 5. Verify the output
+
+Two complementary checks. The **file-level** one re-reads the same plan and inspects the on-disk parts; the **query-level** one exercises the data through a live BanyanDB process.
+
+### 5.1 File-level verify (`migration verify`)
+
+```bash
+./build/bin/banyand-migration verify --copy-config /tmp/migration-plan.yaml
+```
+
+The subcommand walks every (entry, group) read-only: sums source rows by opening every `src/seg-*/shard-*/<partID>/`, enumerates `target/seg-*/`, opens each part and per-segment union sidx, and flags any target seg whose start time misaligns with the stage's `SegmentInterval` grid. It prints all the numbers — there is no exit code, this is an inspector. Sample tail:
+
+```
+== SUMMARY ==
+ data coverage per (node, group):
+ ✓ = src has rows AND tgt has rows (normal copy success)
+ S = src has rows, tgt is empty (copy lost this group — investigate)
+ T = tgt has rows, src is empty (orphan target — investigate)
+ -- = neither side has rows (PVC does not hold this group; hash-sharding normal)
+ ┌────────┬──────────────────┬────────────────┬───────────────┐
+ │ node │ sw_metricsMinute │ sw_metricsHour │ sw_metricsDay │
+ ├────────┼──────────────────┼────────────────┼───────────────┤
+ │ hot-0 │ ✓ │ -- │ ✓ │
+ │ hot-1 │ ✓ │ ✓ │ -- │
+ │ warn-0 │ ✓ │ -- │ ✓ │
+ │ warn-1 │ ✓ │ ✓ │ -- │
+ │ cold-0 │ ✓ │ ✓ │ ✓ │
+ └────────┴──────────────────┴────────────────┴───────────────┘
+
+ target segments : 31
+ target segments misaligned : 0
+ source rows total : 588213170
+ target rows total : 588212817
+```
+
+### 5.2 Query-level verify (`verify-data.sh`)
+
+Start a standalone BanyanDB process pointing at one entry's target:
+
+```bash
+./build/banyandb standalone \
+ --measure-root=/tmp/banyandb-copy/hot/measure-data \
+ --grpc-addr=127.0.0.1:17912 \
+ --http-addr=127.0.0.1:17913
+```
+
+Then in another shell:
+
+```bash
+bash banyand/cmd/migration/scripts/verify-data.sh
+```
+
+The script probes the most recent `DAYS` (default 15) UTC days through the liaison HTTP API — full day for `sw_metricsDay`, the `08:00..09:00` hour for `sw_metricsHour`, the `08:08..08:09` minute for `sw_metricsMinute`. Any zero response is tagged `★MISSING`. Override `DAYS` to widen / narrow the window, or `LIAISON_HTTP` to point at a different liaison endpoint.
+
+## 6. Ship the copy into your cluster
+
+The copy-mode output is a measure-data root, byte-compatible with what a running BanyanDB pod expects. Common ways to bring it into a cluster: package the directory into a volume or image and start a pod with `--measure-root` pointing at it; or stop the BanyanDB process on the target node, swap the measure-data directory, and restart. For an end-to-end in-place flow inside Kubernetes, see [Part 2](#part-2--kubernetes-live-in-place-flow).
+
+## 7. Rollback
+
+Copy mode never touches the source backup. To roll back, just delete the target measure-data roots produced in Step 4:
+
+```bash
+rm -rf /tmp/banyandb-copy/{hot,warm,cold}/measure-data
+```
+
+---
+
+# Part 2 — Kubernetes live in-place flow
+
+End-to-end procedure to migrate measure data **in place** inside a live BanyanDB cluster running on Kubernetes. The runner pod mounts every data PVC, runs the migration into a `data-copy/` sibling on the same volume, and the operator then swaps `data/` ↔ `data-copy/` per PVC before resuming traffic.
+
+The placeholder names used below (`skywalking-showcase` namespace, `demo` helm release, `hot-{0,1} / warn-{0,1} / cold-0` nodes) match the sample manifests under [`example/`](example/); adapt them to your real namespace, release, and topology.
+
+## 1. Detach the liaison svc from its clients
+
+The Service in front of the liaison pods (e.g. `demo-banyandb-grpc`) keeps routing every gRPC client to the live cluster until you repoint its selector to a label nothing in the namespace carries. This drops all in-flight + new connections immediately while leaving the Service object itself intact:
+
+```bash
+kubectl -n skywalking-showcase patch svc demo-banyandb-grpc \
+ -p '{"spec":{"selector":{"app.kubernetes.io/component":"liaison-migration-mode"}}}'
+
+kubectl -n skywalking-showcase get endpoints demo-banyandb-grpc
+```
+
+Endpoints should drop to 0. Liaison pods stay Running so the data pods keep their gossip / internal state.
+
+## 2. Scale data StatefulSets to 0
+
+The runner pod will re-mount each data PVC; RWO requires the original data pods to release them first.
+
+```bash
+for s in demo-banyandb-data-hot demo-banyandb-data-warm demo-banyandb-data-cold ; do
+ kubectl -n skywalking-showcase scale sts $s --replicas=0
+done
+kubectl -n skywalking-showcase wait --for=delete pod -l app.kubernetes.io/component=data --timeout=5m
+```
+
+Note the original replica counts (e.g. hot=2, warm=2, cold=1) — Step 8 restores them.
+
+## 3. Launch the migration-runner pod
+
+Copy [`example/runner-pod.yaml`](example/runner-pod.yaml) and adapt it to your real cluster: PVC `claimName`s, mountPaths, resource limits, and the container `image` (use the same banyandb image your cluster runs — it already ships `banyand-migration` at `/banyand-migration`). The shipped sample mounts each measure PVC at `/mnt/<node>/measure` and each hot schema-property PVC at `/mnt/<node>/schema-property`, plus an `emptyDir` at `/work` for the plan file and per-group union sidx staging.
+
+```bash
+kubectl -n skywalking-showcase apply -f banyand/cmd/migration/example/runner-pod.yaml
+kubectl -n skywalking-showcase wait --for=condition=ready pod/bydb-migration-runner --timeout=3m
+```
+
+## 4. Push the plan into the runner
+
+Copy [`example/plan-live.yaml`](example/plan-live.yaml) and adapt the node names, paths, and target groups to your topology, then push it into the runner:
+
+```bash
+kubectl -n skywalking-showcase cp banyand/cmd/migration/example/plan-live.yaml \
+ bydb-migration-runner:/work/plan.yaml
+```
+
+The shipped sample uses the **live** source mode — every entry sources from its own PVC's `/mnt/<node>/measure/data` and targets `data-copy` on the same volume:
+
+```yaml
+source:
+ live:
+ schema_property_path: /mnt/hot-0/schema-property/data/_schema
+ stages:
+ hot:
+ - { node: hot-0, root: /mnt/hot-0/measure/data }
+ - { node: hot-1, root: /mnt/hot-1/measure/data }
+ warm:
+ - { node: warn-0, root: /mnt/warn-0/measure/data }
+ - { node: warn-1, root: /mnt/warn-1/measure/data }
+ cold:
+ - { node: cold-0, root: /mnt/cold-0/measure/data }
+
+groups: [sw_metricsMinute, sw_metricsHour, sw_metricsDay]
+
+entries:
+ - { stage: hot, target: /mnt/hot-0/measure/data-copy, nodes: [hot-0] }
+ - { stage: hot, target: /mnt/hot-1/measure/data-copy, nodes: [hot-1] }
+ - { stage: warm, target: /mnt/warn-0/measure/data-copy, nodes: [warn-0] }
+ - { stage: warm, target: /mnt/warn-1/measure/data-copy, nodes: [warn-1] }
+ - { stage: cold, target: /mnt/cold-0/measure/data-copy, nodes: [cold-0] }
+```
+
+## 5. Run the migration
+
+```bash
+kubectl -n skywalking-showcase exec bydb-migration-runner -- \
+ /banyand-migration copy --copy-config /work/plan.yaml
+```
+
+Progress is printed in the same line shape as Part 1 — one line per part, plus a per-(entry, group) `done` line, and a final `DONE in …` summary with target seg / mem-part counts, rows copied, bytes written, and fast-/slow-path totals. The tool sets `GOMEMLIMIT` to 85 % of the pod's cgroup limit on startup, so the runner pod's memory limit directly controls peak heap usage.
+
+## 6. Verify (optional but recommended)
+
+Two checks again. Run at least the file-level one before you swap.
+
+### 6.1 File-level verify (`migration verify`)
+
+```bash
+kubectl -n skywalking-showcase exec bydb-migration-runner -- \
+ /banyand-migration verify --copy-config /work/plan.yaml
+```
+
+Output mirrors Part 1's `verify` — per (entry, group) src/tgt row counts, per-target-seg row + sidxDoc breakdown, the coverage table, and finally any mismatches. Excerpt from a real run:
+
+```
+== entry stage=hot target=/mnt/hot-0/measure/data-copy group=sw_metricsMinute ==
+ src : 7706398 row(s) across 34 part(s) in 1 dir(s)
+ /mnt/hot-0/measure/data/sw_metricsMinute
+ tgt : 7706398 row(s) across 2 seg(s) under /mnt/hot-0/measure/data-copy/sw_metricsMinute
+ seg-20260519 aligned shards=1 parts=19 rows=6325262 sidxDocs=1469980
+ seg-20260520 aligned shards=1 parts=15 rows=1381136 sidxDocs=1469980
+...
+== SUMMARY ==
+ target segments : 31
+ target segments misaligned : 0
+ source rows total : 588213170
+ target rows total : 588212817
+
+ row-count mismatches (tgt - src, negative = rows dropped by copy):
+ stage node group src tgt diff
+ warm warn-1 sw_metricsHour 2980213 2980121 -92
+ cold cold-0 sw_metricsMinute 448364487 448364341 -146
+ cold cold-0 sw_metricsHour 16360573 16360458 -115
+ TOTAL -353
+```
+
+Small negative diffs (a few hundred rows out of hundreds of millions) are caused by slow-path `mustInitFromDataPoints` deduping `(seriesID, timestamp)` within each chunk flush — banyandb's merger can leave duplicated boundary rows in source parts, and only the latest version of each `(sid, ts)` survives in the target.
+
+If `verify` reports a mismatch you want to drill into, run `analyze` on the offending (entry, group) to dump the exact within-part duplicate rows + the cross-part diff:
+
+```bash
+kubectl -n skywalking-showcase exec bydb-migration-runner -- \
+ /banyand-migration analyze \
+ --copy-config /work/plan.yaml \
+ --entry-idx 3 --group sw_metricsHour --sample 5
+```
+
+It walks every src part for that (entry, group), reports total/unique/duplicate row counts, lists per-part block boundaries where consecutive blocks share the same `(sid, ts)` (the banyandb merger fingerprint), and runs an src↔tgt multiset diff that pinpoints the exact `(sid, ts, version)` rows missing from the target:
+
+```
+== analyze entry [4/5] stage=warm nodes=[warn-1] group=sw_metricsHour ==
+ parts scanned : 17
+ total rows on disk : 2980213
+ distinct (sid, ts) (global) : 2401206
+ cross-part dup rows : 579007 (NOT dropped by copy — slow path is per-part)
+ (sid, ts) keys with >1 row : 544037
+ WITHIN-part dup rows : 110 (← MATCH this against verify src-tgt diff)
+ ...
+== src-vs-target multiset diff ==
+ src rows : 2980213
+ tgt rows : 2980121
+ missing rows total : 92 (← MATCHES verify src-tgt diff for this entry+group)
+ missing rows (showing 5 / 74 keys):
+ sid=3766501475917424574 ts=2026-05-06T15:00:00Z
+ src version=511194984324783 part=…/seg-20260506/shard-0/0000000000000013
+ src version=514535618046101 part=…/seg-20260506/shard-0/0000000000000013
+ tgt version=514535618046101 part=…/data-copy/sw_metricsHour/seg-20260430/shard-0/0000000000000034
+ MISSING version=511194984324783 part=…/seg-20260506/shard-0/0000000000000013
+```
+
+### 6.2 Query-level verify (`verify-data.sh`)
+
+Deferred to Step 9 — once the swap is done and queries are live again.
+
+## 7. Swap migrated groups in place
+
+For each PVC, move the migrated groups from `data-copy/` onto `data/`, keeping the originals under `data-backup/`. The list of migrated groups comes straight from your plan's `groups:` field — using the sample plan above, that is `sw_metricsMinute sw_metricsHour sw_metricsDay`:
+
+```bash
+kubectl -n skywalking-showcase exec bydb-migration-runner -- sh -c '
+ set -eux
+ MIGRATED_GROUPS="sw_metricsMinute sw_metricsHour sw_metricsDay"
+ for n in hot-0 hot-1 warn-0 warn-1 cold-0 ; do
+ base=/mnt/$n/measure
+ mkdir -p "$base/data-backup"
+ for g in $MIGRATED_GROUPS ; do
+ [ -d "$base/data/$g" ] && mv "$base/data/$g" "$base/data-backup/$g"
+ [ -d "$base/data-copy/$g" ] && mv "$base/data-copy/$g" "$base/data/$g"
+ done
+ rmdir "$base/data-copy" 2>/dev/null || true
+ done
+'
+```
+
+A missing source/copy dir on some PVC is normal — banyandb's hash sharding means each replica only carries a subset of groups.
+
+Once the swap is confirmed (Step 9), tear the runner pod down so the data StatefulSets can re-claim their PVCs:
+
+```bash
+kubectl -n skywalking-showcase delete pod bydb-migration-runner \
+ --grace-period=0 --force
+kubectl -n skywalking-showcase wait --for=delete pod/bydb-migration-runner --timeout=2m
+```
+
+## 8. Scale data StatefulSets back up
+
+Restore the original replica counts and wait until every data pod reports Ready:
+
+```bash
+kubectl -n skywalking-showcase scale sts demo-banyandb-data-hot --replicas=2
+kubectl -n skywalking-showcase scale sts demo-banyandb-data-warm --replicas=2
+kubectl -n skywalking-showcase scale sts demo-banyandb-data-cold --replicas=1
+
+kubectl -n skywalking-showcase wait --for=condition=ready \
+ pod -l app.kubernetes.io/component=data --timeout=10m
+```
+
+After startup, each data pod's `segmentController` picks up the migrated `seg-*` directories and the merge loop will compact the per-source `partID` files into larger merged parts over time — no operator action required.
+
+## 9. Port-forward + query-level verify
+
+Port-forward to a liaison pod and run `verify-data.sh` against it:
+
+```bash
+kubectl -n skywalking-showcase port-forward \
+ pod/demo-banyandb-liaison-0 17913:17913 > /tmp/pf.log 2>&1 &
+until curl -sf -m 3 http://127.0.0.1:17913/api/healthz > /dev/null ; do sleep 2 ; done
+
+bash banyand/cmd/migration/scripts/verify-data.sh
+```
+
+For each of the most recent `DAYS` (default 15) UTC days, the script issues a `stages=[hot,warm,cold]` query against each tier at the granularity that tier stores:
+
+| Tier | Measure | Window |
+|---|---|---|
+| Day | `service_apdex_day` | full day (`00:00:00 .. 23:59:59`) |
+| Hour | `service_apdex_hour` | the `08:00` hour (`08:00:00 .. 08:59:59`) |
+| Minute | `service_apdex_minute` | the `08:08` minute (`08:08:00 .. 08:08:59`) |
+
+A trailing `★MISSING` marker is appended whenever the probe returns zero data points for that day. Investigate any `★MISSING` row before reattaching client traffic.
+
+## 10. Reattach the liaison svc selector
+
+Once Step 9 is green, point the Service back at the original liaison selector so external clients can reach the cluster again:
+
+```bash
+kubectl -n skywalking-showcase patch svc demo-banyandb-grpc -p '{
+ "spec":{"selector":{
+ "app.kubernetes.io/component":"liaison",
+ "app.kubernetes.io/instance":"demo",
+ "app.kubernetes.io/name":"banyandb"
+ }}
+}'
+
+kubectl -n skywalking-showcase get endpoints demo-banyandb-grpc
+```
+
+## 11. Rollback
+
+The migration never touches the live data dir until Step 7. To revert: detach client traffic again, scale data down, restore each PVC from `data-backup/`, scale back up, then reattach traffic.
+
+```bash
+kubectl -n skywalking-showcase patch svc demo-banyandb-grpc \
+ -p '{"spec":{"selector":{"app.kubernetes.io/component":"liaison-migration-mode"}}}'
+
+for s in demo-banyandb-data-hot demo-banyandb-data-warm demo-banyandb-data-cold ; do
+ kubectl -n skywalking-showcase scale sts $s --replicas=0
+done
+kubectl -n skywalking-showcase wait --for=delete pod -l app.kubernetes.io/component=data --timeout=5m
+
+kubectl -n skywalking-showcase apply -f banyand/cmd/migration/example/runner-pod.yaml
+kubectl -n skywalking-showcase wait --for=condition=ready pod/bydb-migration-runner --timeout=3m
+
+kubectl -n skywalking-showcase exec bydb-migration-runner -- sh -c '
+ set -eux
+ MIGRATED_GROUPS="sw_metricsMinute sw_metricsHour sw_metricsDay"
+ for n in hot-0 hot-1 warn-0 warn-1 cold-0 ; do
+ base=/mnt/$n/measure
+ for g in $MIGRATED_GROUPS ; do
+ [ -d "$base/data/$g" ] && rm -rf "$base/data/$g"
+ [ -d "$base/data-backup/$g" ] && mv "$base/data-backup/$g" "$base/data/$g"
+ done
+ rmdir "$base/data-backup" 2>/dev/null || true
+ done'
+
+kubectl -n skywalking-showcase delete pod bydb-migration-runner --grace-period=0 --force
+
+kubectl -n skywalking-showcase scale sts demo-banyandb-data-hot --replicas=2
+kubectl -n skywalking-showcase scale sts demo-banyandb-data-warm --replicas=2
+kubectl -n skywalking-showcase scale sts demo-banyandb-data-cold --replicas=1
+kubectl -n skywalking-showcase wait --for=condition=ready \
+ pod -l app.kubernetes.io/component=data --timeout=10m
+
+kubectl -n skywalking-showcase patch svc demo-banyandb-grpc -p '{
+ "spec":{"selector":{
+ "app.kubernetes.io/component":"liaison",
+ "app.kubernetes.io/instance":"demo",
+ "app.kubernetes.io/name":"banyandb"
+ }}
+}'
+```
diff --git a/banyand/cmd/migration/README.md b/banyand/cmd/migration/README.md
new file mode 100644
index 000000000..92aea5172
--- /dev/null
+++ b/banyand/cmd/migration/README.md
@@ -0,0 +1,144 @@
+# BanyanDB measure migration
+
+A one-shot CLI that rewrites **measure** data — from either a local backup snapshot or a set of live PVC mount paths — into one or more target measure-data roots, with rows re-bucketed onto each target stage's `SegmentInterval` grid. Stream / trace / property catalogs and `sw_metadata` are **out of scope**: the tool only touches measure data.
+
+For end-to-end operator runbooks (workstation flow + in-cluster live flow), see [`MIGRATION.md`](MIGRATION.md).
+
+The CLI exposes three subcommands; all share a single required flag `--copy-config <path>` pointing at the YAML plan documented below:
+
+| Subcommand | Purpose |
+|---|---|
+| `copy` | Rewrite measure data from the plan's source into each entry's target. Side-effecting. |
+| `verify` | Re-read the same plan and report per-(entry, group) source-vs-target row counts, segment grid alignment, and union-sidx doc counts. Read-only inspector. |
+| `analyze` | For one (entry, group), walk every source part and dump within-part duplicate rows, per-part block boundaries, and an exact src↔tgt multiset diff. Diagnostic. |
+
+Each source row is routed to a grid-aligned target segment using the `SegmentInterval` defined by the entry's `stage`. A per-group **union sidx** is built once from every source segment and broadcast (byte-copied) into every aligned target segment. The tool reads schemas and group `ResourceOpts` directly from the source's `schema-property` bluge catalog — no liaison or network access is required at run time.
+
+---
+
+## What it does
+
+1. Loads measure schemas (tag families + `IndexMode` bit) and group `ResourceOpts` from the source's schema-property bluge catalog.
+2. Rejects any requested group that contains an `IndexMode` measure (their field data lives inside sidx and broadcasting the union would corrupt cross-node dedup at query time).
+3. For each group, walks every source segment under `<source>/<node>/.../measure/<group>/` and:
+ - Builds one **union sidx** at `<staging_dir>/<group>/sidx/` by deduplicating every source sidx doc by SeriesID.
+ - Discovers every `seg-*/shard-*/<partID>/` directory and records a `partTask` list.
+4. For each `(entry, group)` pair:
+ - Resolves the SegmentInterval for `entry.stage`.
+ - For each source part: if all rows fall in one aligned target segment, takes the **fast path** (whole-part byte copy under a fresh target partID); otherwise the **slow path** row-level rewrite into per-target-segment buckets.
+ - Writes `metadata` (segment version + endTime) and a `.snp` snapshot file per `(target segment, shard)`.
+ - Broadcasts the group's union sidx into every aligned target segment that received any rows.
+
+Entries run sequentially. The first error aborts the run. The whole `staging_dir` is removed at the end (success or failure).
+
+---
+
+## Limitations
+
+- The source **must** include a `schema-property` catalog under at least one node (typically a hot / schema-server node). Sources lacking the catalog are rejected.
+- All groups containing any `IndexMode` measure are refused up front.
+- Source and target shard topology must match (no reshard semantics).
+- The tool assumes nothing else is writing to the target paths.
+
+---
+
+## Configuration
+
+A `CopyPlan` selects ONE source mode (`backup` or `live`) and a list of `entries` that fan-out the same source data into multiple target directories. Two ready-to-edit samples ship under [`example/`](example/): [`plan-backup.yaml`](example/plan-backup.yaml) (workstation, reads a downloaded snapshot) and [`plan-live.yaml`](example/plan-live.yaml) (in-cluster, reads live PVC mounts).
+
+```yaml
+# Optional. When unset, a fresh os.MkdirTemp("banyandb-migration-staging-*")
+# is created under TMPDIR. When set, the directory must be absent OR an
+# empty directory (otherwise the loader aborts). The whole staging tree
+# is removed at the end of the run.
+staging_dir: /tmp/banyandb-staging
+
+# Required. Exactly one of source.backup or source.live must be set.
+source:
+ # --- mode A: backup snapshot --------------------------------------
+ backup:
+ # Local path holding the backup snapshot, laid out as
+ # <root>/<node>/<date>/measure/<group>/seg-*/shard-*/<partID>/
+ # <root>/<node>/<date>/schema-property/_schema/shard-*/
+ root: /tmp/banyandb-backup
+ # Pins the snapshot date inside <root>. Format YYYY-MM-DD.
+ date: 2026-05-19
+ # --- mode B: live PVC mounts (mutually exclusive with backup) -----
+ # live:
+ # schema_property_path: /mnt/hot-0/schema-property/data/_schema
+ # stages:
+ # hot:
+ # - { node: hot-0, root: /mnt/hot-0/measure/data }
+ # warm: [...]
+ # cold: [...]
+
+# Required. Non-empty, no duplicates. Each group is processed once per entry.
+groups:
+ - sw_metricsMinute
+ - sw_metricsHour
+ - sw_metricsDay
+
+# Required. Non-empty. Each entry names one fan-out destination.
+# - stage: one of the LifecycleStages configured on every group's
+# ResourceOpts (e.g. hot / warm / cold) — its SegmentInterval
+# determines the target grid for this entry.
+# - target: the measure-data root for this entry. Must not pre-exist
+# (or be empty) for any group dir written under it.
+# - nodes: the subset of source node identifiers (backup directory
+# names, or live `stages[stage][].node` keys) feeding this
+# entry's target.
+# Targets must be globally unique across all entries.
+entries:
+ - { stage: hot, target: /tmp/banyandb-copy/hot/measure-data, nodes: [hot-0, hot-1] }
+ - { stage: warm, target: /tmp/banyandb-copy/warm/measure-data, nodes: [warn-0, warn-1] }
+ - { stage: cold, target: /tmp/banyandb-copy/cold/measure-data, nodes: [cold-0] }
+```
+
+Validation errors are surfaced at startup, before any byte is written. The common ones: missing `source` (or setting both modes), missing/duplicate `groups`, missing/duplicate `entries[*].target`, bad `date` format, pre-existing non-empty `staging_dir`, or an `entries[*].nodes` entry that does not match any node declared under `source.live.stages[stage]`.
+
+---
+
+## Output layout
+
+Each target directory is byte-compatible with what a BanyanDB data pod expects at `--measure-root`. The on-disk shape under one entry's target is:
+
+```
+<entry.target>/<group>/seg-YYYYMMDD[HH]/
+ ├── metadata # storage.SegmentMetadata: segment version + endTime
+ ├── shard-N/
+ │ ├── <16-hex-partID>/ # measure part files (one dir per copied part)
+ │ └── <16-hex-partID>.snp # part index snapshot
+ └── sidx/ # broadcast union sidx (bluge index)
+```
+
+Segment names use `seg-YYYYMMDD` for `DAY` units and `seg-YYYYMMDDHH` for `HOUR` units, matching `banyand/internal/storage.segmentController`. PartIDs are emitted as zero-padded 16-character lowercase hex, starting at `0000000000000001` and incrementing as `copy` writes parts — banyandb's runtime picks up the highest existing partID on startup so live writes resume at `max(N)+1`.
+
+---
+
+## Build & run
+
+Build via the top-level Makefile — it produces every banyandb binary, including `banyand-migration`, under `banyand/build/bin/`:
+
+```bash
+cd <banyandb-repo-root>
+make build
+ls banyand/build/bin/banyand-migration
+```
+
+Run any of the three subcommands with the same `--copy-config` plan:
+
+```bash
+./banyand/build/bin/banyand-migration copy --copy-config ./migration-plan.yaml
+./banyand/build/bin/banyand-migration verify --copy-config ./migration-plan.yaml
+./banyand/build/bin/banyand-migration analyze --copy-config ./migration-plan.yaml \
+ --entry-idx 3 --group sw_metricsHour --sample 5
+```
+
+To attach `pprof` to a long-running `copy`, pass `--pprof-addr` (persistent flag on the root command):
+
+```bash
+./banyand/build/bin/banyand-migration --pprof-addr 127.0.0.1:6060 copy --copy-config ./migration-plan.yaml &
+go tool pprof -seconds=60 http://127.0.0.1:6060/debug/pprof/profile
+```
+
+The `copy` subcommand also auto-sets `GOMEMLIMIT` to 85 % of the pod cgroup memory limit on startup (see `applyGoMemLimit` in `banyand/cmd/migration/copy.go`) so a momentary heap spike during slow-path zstd flush does not trip the kernel OOMKiller.
diff --git a/banyand/cmd/migration/analyze.go b/banyand/cmd/migration/analyze.go
new file mode 100644
index 000000000..550c27f00
--- /dev/null
+++ b/banyand/cmd/migration/analyze.go
@@ -0,0 +1,186 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package main
+
+import (
+ "fmt"
+ "path/filepath"
+ "time"
+
+ "github.com/spf13/cobra"
+
+ "github.com/apache/skywalking-banyandb/banyand/measure"
+ "github.com/apache/skywalking-banyandb/pkg/fs"
+)
+
+func newAnalyzeCmd() *cobra.Command {
+ var configPath string
+ var entryIdx int
+ var groupName string
+ var sampleCap int
+
+ cmd := &cobra.Command{
+ Use: "analyze",
+ Short: "Row-level (seriesID, timestamp) duplicate analysis for one (entry, group)",
+ Long: `analyze opens every src part for ONE (entry, group) pair in the copy
+plan, decodes block metadata + timestamps + versions (NOT tag/field
+bodies — cheap scan), and reports:
+
+ - total physical row count on disk
+ - distinct (seriesID, timestamp) key count
+ - duplicate row count (= total - distinct)
+ - first --sample (sid, ts) keys that have >1 version on disk
+
+This is the diagnostic that proves the src > tgt diff reported by
+'verify' is caused by banyandb's normal version-tracking dedup at
+write time (only the latest version per (sid, ts) survives in the
+target part).`,
+ RunE: func(_ *cobra.Command, _ []string) error {
+ plan, err := LoadCopyPlan(configPath)
+ if err != nil {
+ return err
+ }
+ cfg := plan.ToDirectCopyConfig("")
+ if entryIdx < 0 || entryIdx >= len(cfg.Entries) {
+ return fmt.Errorf("entry-idx %d out of range [0,%d)", entryIdx, len(cfg.Entries))
+ }
+ entry := cfg.Entries[entryIdx]
+
+ srcRoots := measure.ResolveEntrySrcRoots(cfg, entry, groupName)
+ if len(srcRoots) == 0 {
+ return fmt.Errorf("no source dirs resolve for entry-idx=%d group=%s (PVC likely doesn't carry this group)", entryIdx, groupName)
+ }
+ fmt.Printf("== analyze entry [%d/%d] stage=%s nodes=%v group=%s ==\n",
+ entryIdx+1, len(cfg.Entries), entry.Stage, entry.Nodes, groupName)
+ for _, r := range srcRoots {
+ fmt.Printf(" src: %s\n", r)
+ }
+ start := time.Now()
+ res, err := measure.AnalyzeGroupRows(srcRoots, fs.NewLocalFileSystem(), sampleCap)
+ if err != nil {
+ return err
+ }
+ elapsed := time.Since(start)
+
+ fmt.Printf(" parts scanned : %d\n", res.PartsScanned)
+ fmt.Printf(" total rows on disk : %d\n", res.TotalRows)
+ fmt.Printf(" distinct (sid, ts) (global) : %d\n", res.UniqueKeys)
+ fmt.Printf(" cross-part dup rows : %d (NOT dropped by copy — slow path is per-part)\n", res.DuplicateRows)
+ fmt.Printf(" (sid, ts) keys with >1 row : %d\n", res.KeysWithDuplicates)
+ fmt.Printf(" WITHIN-part dup rows : %d (← MATCH this against verify src-tgt diff)\n", res.PerPartDupRows)
+ fmt.Printf(" elapsed : %v\n", elapsed)
+ if len(res.PerPartDups) > 0 {
+ partsToShow := capToSample(len(res.PerPartDups), sampleCap)
+ fmt.Printf(" per-part dedup breakdown (showing %d / %d parts with internal duplicates):\n",
+ partsToShow, len(res.PerPartDups))
+ for i := 0; i < partsToShow; i++ {
+ pp := res.PerPartDups[i]
+ fmt.Printf(" %s/%s/%s rows=%d uniq=%d withinPartDups=%d boundaries=%d\n",
+ pp.SegName, pp.ShardName, pp.PartID, pp.Rows, pp.UniqueKeys, pp.PartialDupRows, len(pp.Boundaries))
+ fmt.Printf(" source path: %s/%s/%s/%s\n",
+ pp.SourceRoot, pp.SegName, pp.ShardName, pp.PartID)
+ boundariesToShow := capToSample(len(pp.Boundaries), sampleCap)
+ for j := 0; j < boundariesToShow; j++ {
+ b := pp.Boundaries[j]
+ ts := time.Unix(0, b.Timestamp).UTC().Format(time.RFC3339)
+ fmt.Printf(" boundary: block[%d] last == block[%d] first sid=%d ts=%s vA=%d vB=%d\n",
+ b.BlockA, b.BlockB, b.SeriesID, ts, b.VersionA, b.VersionB)
+ }
+ if boundariesToShow < len(pp.Boundaries) {
+ fmt.Printf(" ... + %d more boundaries (raise --sample to see them)\n",
+ len(pp.Boundaries)-boundariesToShow)
+ }
+ }
+ if partsToShow < len(res.PerPartDups) {
+ fmt.Printf(" ... + %d more parts (raise --sample to see them)\n",
+ len(res.PerPartDups)-partsToShow)
+ }
+ }
+ if res.KeysWithDuplicates > 0 {
+ shown := len(res.SamplesByVersion)
+ fmt.Printf(" sample (sid, ts) with >1 row on disk (showing %d / %d keys; cross-part dups, won't be deduped):\n",
+ shown, res.KeysWithDuplicates)
+ for _, s := range res.SamplesByVersion {
+ ts := time.Unix(0, s.Timestamp).UTC().Format(time.RFC3339)
+ fmt.Printf(" sid=%d ts=%s\n", s.SeriesID, ts)
+ for _, v := range s.Versions {
+ fmt.Printf(" version=%d part=%s\n", v.Version, v.Path)
+ }
+ }
+ }
+
+ targetGroupRoot := filepath.Join(entry.Target, groupName)
+ fmt.Println()
+ fmt.Printf("== src-vs-target multiset diff ==\n")
+ fmt.Printf(" tgt: %s\n", targetGroupRoot)
+ diffStart := time.Now()
+ diff, derr := measure.AnalyzeGroupDiffWithTarget(srcRoots, targetGroupRoot, fs.NewLocalFileSystem(), sampleCap)
+ if derr != nil {
+ return derr
+ }
+ fmt.Printf(" src rows : %d\n", diff.SourceRows)
+ fmt.Printf(" tgt rows : %d\n", diff.TargetRows)
+ fmt.Printf(" missing rows total : %d (← MATCHES verify src-tgt diff for this entry+group)\n", diff.MissingRows)
+ fmt.Printf(" diff scan elapsed : %v\n", time.Since(diffStart))
+ if diff.MissingKeys > 0 {
+ shown := len(diff.Missing)
+ fmt.Printf(" missing rows (showing %d / %d keys):\n", shown, diff.MissingKeys)
+ for _, m := range diff.Missing {
+ ts := time.Unix(0, m.Timestamp).UTC().Format(time.RFC3339)
+ fmt.Printf(" sid=%d ts=%s\n", m.SeriesID, ts)
+ for _, v := range m.SourceVersions {
+ fmt.Printf(" src version=%d part=%s\n", v.Version, v.Path)
+ }
+ for _, v := range m.TargetVersions {
+ fmt.Printf(" tgt version=%d part=%s\n", v.Version, v.Path)
+ }
+ for _, v := range m.MissingVersions {
+ fmt.Printf(" MISSING version=%d part=%s\n", v.Version, v.Path)
+ }
+ }
+ }
+ fmt.Println()
+ fmt.Println("Interpretation:")
+ fmt.Println(" - slow-path mustInitFromDataPoints only fires INSIDE one source part's")
+ fmt.Println(" bucket flush; cross-part dups survive copy as separate target memparts.")
+ fmt.Println(" - WITHIN-part dup rows = the count slow-path drops at flush time. If this")
+ fmt.Println(" matches `verify`'s src-tgt diff (for the same entry/group), the loss is")
+ fmt.Println(" proven to be banyandb's normal in-part (seriesID, ts) collapse.")
+ return nil
+ },
+ }
+
+ cmd.Flags().StringVar(&configPath, "copy-config", "", "path to the YAML migration copy plan (required)")
+ cmd.Flags().IntVar(&entryIdx, "entry-idx", 0, "0-based entry index in the plan to analyze")
+ cmd.Flags().StringVar(&groupName, "group", "", "group name to analyze (must appear in plan.groups) (required)")
+ cmd.Flags().IntVar(&sampleCap, "sample", 10,
+ "max number of items to print per list (applies to every sample/breakdown section; 0 = unlimited)")
+ _ = cmd.MarkFlagRequired("copy-config")
+ _ = cmd.MarkFlagRequired("group")
+ return cmd
+}
+
+// capToSample returns the number of items to actually print given the
+// observed list length and the operator-supplied --sample cap. limit=0
+// means "print everything"; otherwise we never exceed limit.
+func capToSample(have, limit int) int {
+ if limit <= 0 || have <= limit {
+ return have
+ }
+ return limit
+}
diff --git a/banyand/cmd/migration/copy.go b/banyand/cmd/migration/copy.go
new file mode 100644
index 000000000..0900c428e
--- /dev/null
+++ b/banyand/cmd/migration/copy.go
@@ -0,0 +1,189 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package main
+
+import (
+ "context"
+ "fmt"
+ "os"
+ "os/signal"
+ "runtime/debug"
+ "strconv"
+ "syscall"
+ "time"
+
+ "github.com/spf13/cobra"
+
+ "github.com/apache/skywalking-banyandb/banyand/measure"
+ "github.com/apache/skywalking-banyandb/pkg/cgroups"
+)
+
+func newCopyCmd() *cobra.Command {
+ var configPath string
+
+ cmd := &cobra.Command{
+ Use: "copy",
+ Short: "Direct-file copy of measure parts into local measure root with row-level grid alignment",
+ Long: `copy walks every source part, reads each row's actual timestamp, and routes
+the row to the grid-aligned target segment via SegmentInterval.Standard.
+A single source part can fan out to multiple target segments (one new
+part each).
+
+The run is fully driven by a YAML config (--copy-config). One config
+carries one source (backup snapshot or live PVC mounts), one group list,
+and one or more (stage, target) entries — schemas and per-group union
+sidx are loaded/built once and reused across every entry.
+
+Per group, a single union sidx is built in the staging directory by
+deduplicating every source seg-*/sidx doc by SeriesID; that union sidx
+is then broadcast (byte-copied) into every aligned target segment that
+received any rows. A fresh .snp is written per (target segment, shard)
+so tsTable.loadSnapshot picks up the new parts on next startup.
+
+Schemas (measure tag families + IndexMode bit) and group resource
+opts (SegmentInterval per LifecycleStage) are read directly from the
+source's schema-property bluge catalog — no liaison access is needed.
+Groups containing any IndexMode measure are rejected up front, since
+their fields live inside sidx and the broadcast strategy would corrupt
+cross-node deduplication at the liaison.
+
+The staging directory is removed in full when the run finishes
+(success or failure). Run inside a data pod whose BanyanDB process is
+NOT writing to any of the entry targets. Source and target shard
+topology must match.`,
+ RunE: func(_ *cobra.Command, _ []string) error {
+ applyGoMemLimit()
+
+ plan, err := LoadCopyPlan(configPath)
+ if err != nil {
+ return err
+ }
+ stagingDir, err := PrepareStagingDir(plan.StagingDir)
+ if err != nil {
+ return err
+ }
+ defer func() {
+ if rmErr := os.RemoveAll(stagingDir); rmErr != nil {
+ fmt.Fprintf(os.Stderr,
+ "warning: failed to clean staging dir %s: %v\n",
+ stagingDir, rmErr)
+ }
+ }()
+
+ cfg := plan.ToDirectCopyConfig(stagingDir)
+
+ ctx, stop := signal.NotifyContext(context.Background(),
+ os.Interrupt, syscall.SIGTERM)
+ defer stop()
+
+ res, runErr := measure.MigrationCopy(ctx, cfg)
+ printCopyResult(res)
+ return runErr
+ },
+ }
+
+ cmd.Flags().StringVar(&configPath, "copy-config", "",
+ "path to the YAML migration copy plan (required)")
+ _ = cmd.MarkFlagRequired("copy-config")
+ return cmd
+}
+
+func printCopyResult(res measure.DirectCopyResult) {
+ fmt.Println("DONE in", res.Duration)
+ fmt.Printf(" target segments : %d\n", res.Segments)
+ fmt.Printf(" source parts : %d\n", res.SourceParts)
+ fmt.Printf(" target mem-parts : %d (pre-merge; banyandb's merge loop will compact)\n", res.TargetParts)
+ fmt.Printf(" rows copied : %d\n", res.Rows)
+ fmt.Printf(" bytes written : %d\n", res.Bytes)
+ fmt.Printf(" fast-path parts : %d\n", measure.FastPathHits())
+ fmt.Printf(" slow-path parts : %d\n", measure.SlowPathHits())
+ fmt.Printf(" slow-path rows : %d\n", measure.SlowPathRows())
+}
+
+// applyGoMemLimit auto-sets the Go runtime's heap soft-cap (GOMEMLIMIT)
+// from the process cgroup memory limit when the operator did not pass
+// one explicitly.
+//
+// Why this is necessary:
+// - Go's runtime sizes the heap from live set + GOGC by default; it
+// does NOT read cgroup limits on its own. On a memory-constrained
+// pod a momentary heap spike during slow-path zstd flush can sail
+// right past the pod limit and trip the kernel OOMKiller (or
+// macOS jetsam on kind), wiping the whole migration mid-run.
+// - Historical evidence from this codebase: an unconstrained run on
+// a 16 GiB GKE pod OOM'd at 99 % through the warm tier; raising
+// the limit to 20 GiB AND setting GOMEMLIMIT to 17 GiB (85 %) was
+// the fix. The shared flush-pool + Phase A union-sidx refactor
+// reduced peak memory, but no run has been validated WITHOUT
+// GOMEMLIMIT post-refactor, so we still ship the safety net.
+//
+// Why we don't rely on operators setting GOMEMLIMIT manually:
+// - Go 1.19+ honors a GOMEMLIMIT env at startup, so an operator who
+// remembered to set it gets the same effect. This helper is a UX
+// fallback for the common case where it was forgotten — we read
+// the cgroup limit (works on kind, GKE, plain docker run) and
+// apply 85 % of it. The 15 % headroom is for the runtime's own
+// overhead + the Linux page cache.
+//
+// Resolution order: GOMEMLIMIT env wins (we just log and return);
+// otherwise read cgroup limit and call debug.SetMemoryLimit; otherwise
+// log "not set" and let the process run uncapped (bare-host case where
+// the operator presumably has enough RAM).
+func applyGoMemLimit() {
+ if raw := os.Getenv("GOMEMLIMIT"); raw != "" {
+ // Honor whatever the env already set — Go's runtime parsed it
+ // at startup. Just log.
+ fmt.Fprintf(os.Stdout, "[migration] GOMEMLIMIT honored from env: %s\n", raw)
+ return
+ }
+ limit, err := cgroups.MemoryLimit()
+ if err != nil || limit <= 0 {
+ // No reliable bound: cgroups.MemoryLimit returns -1 for "max"
+ // (unlimited) on a non-containerised host, and Go's std lib has
+ // no portable cheap accessor for host total RAM (runtime.MemStats
+ // reports memory reserved for the runtime itself, not the host).
+ // Leave GOMEMLIMIT unset on this branch and rely on the operator
+ // (or the kernel OOMKiller) to bound the process.
+ fmt.Fprintf(os.Stdout, "[migration] GOMEMLIMIT not set (no cgroup limit; running on bare host)\n")
+ return
+ }
+ memLimit := int64(float64(limit) * 0.85)
+ debug.SetMemoryLimit(memLimit)
+ fmt.Fprintf(os.Stdout, "[migration] GOMEMLIMIT set to %s (85%% of cgroup limit %s) at %s\n",
+ humanBytes(memLimit), humanBytes(limit), time.Now().Format("15:04:05"))
+}
+
+func humanBytes(n int64) string {
+ if n < 0 {
+ return strconv.FormatInt(n, 10)
+ }
+ const (
+ KiB = 1 << 10
+ MiB = 1 << 20
+ GiB = 1 << 30
+ )
+ switch {
+ case n >= GiB:
+ return fmt.Sprintf("%.1fGiB", float64(n)/GiB)
+ case n >= MiB:
+ return fmt.Sprintf("%.1fMiB", float64(n)/MiB)
+ case n >= KiB:
+ return fmt.Sprintf("%.1fKiB", float64(n)/KiB)
+ }
+ return fmt.Sprintf("%dB", n)
+}
diff --git a/banyand/cmd/migration/example/plan-backup.yaml b/banyand/cmd/migration/example/plan-backup.yaml
new file mode 100644
index 000000000..73e0f8b40
--- /dev/null
+++ b/banyand/cmd/migration/example/plan-backup.yaml
@@ -0,0 +1,41 @@
+# Licensed to Apache Software Foundation (ASF) under one or more contributor
+# license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright
+# ownership. Apache Software Foundation (ASF) licenses this file to you under
+# the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+# Sample plan for running `migration copy` against a local backup
+# snapshot on a workstation. Adapt every path / date / group / target to
+# your own environment before running. Field reference: README.md.
+#
+# Source mode: `backup` — reads measure + schema-property from a
+# downloaded snapshot laid out as
+# <root>/<node>/<date>/{measure,schema-property}/...
+
+staging_dir: /tmp/banyandb-staging
+
+source:
+ backup:
+ root: /tmp/banyandb-backup
+ date: 2026-05-19
+
+groups:
+ - sw_metricsMinute
+ - sw_metricsHour
+ - sw_metricsDay
+
+entries:
+ - { stage: hot, target: /tmp/banyandb-copy/hot/measure-data, nodes: [hot-0, hot-1] }
+ - { stage: warm, target: /tmp/banyandb-copy/warm/measure-data, nodes: [warn-0, warn-1] }
+ - { stage: cold, target: /tmp/banyandb-copy/cold/measure-data, nodes: [cold-0] }
diff --git a/banyand/cmd/migration/example/plan-live.yaml b/banyand/cmd/migration/example/plan-live.yaml
new file mode 100644
index 000000000..6ef6ffb86
--- /dev/null
+++ b/banyand/cmd/migration/example/plan-live.yaml
@@ -0,0 +1,67 @@
+# Licensed to Apache Software Foundation (ASF) under one or more contributor
+# license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright
+# ownership. Apache Software Foundation (ASF) licenses this file to you under
+# the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+# Migration plan run from INSIDE the bydb-migration-runner pod.
+# Each entry rewrites one mounted PVC's measure data in-place into a
+# `data-copy/` sibling on the same volume — after operator verification,
+# the workflow is `rm -rf data && mv data-copy data` per PVC.
+#
+# Source mode: `live` — measure data + schema-property come from the
+# runner pod's PVC mounts. The discriminated `source.live.stages` map
+# names every (node, root) pair available per stage; each entry below
+# selects a subset of those nodes via `nodes` to feed its target.
+
+staging_dir: /work/staging
+
+source:
+ live:
+ schema_property_path: /mnt/hot-0/schema-property/data/_schema
+ stages:
+ hot:
+ - { node: hot-0, root: /mnt/hot-0/measure/data }
+ - { node: hot-1, root: /mnt/hot-1/measure/data }
+ warm:
+ - { node: warn-0, root: /mnt/warn-0/measure/data }
+ - { node: warn-1, root: /mnt/warn-1/measure/data }
+ cold:
+ - { node: cold-0, root: /mnt/cold-0/measure/data }
+
+groups:
+ - sw_metricsMinute
+ - sw_metricsHour
+ - sw_metricsDay
+
+# One entry per data-PVC. The migration writes that node's `data-copy/`
+# sibling using only its own node's source (in-place rewrite), so post-
+# migration the operator can `rm -rf data && mv data-copy data` per PVC
+# independently.
+entries:
+ - stage: hot
+ target: /mnt/hot-0/measure/data-copy
+ nodes: [hot-0]
+ - stage: hot
+ target: /mnt/hot-1/measure/data-copy
+ nodes: [hot-1]
+ - stage: warm
+ target: /mnt/warn-0/measure/data-copy
+ nodes: [warn-0]
+ - stage: warm
+ target: /mnt/warn-1/measure/data-copy
+ nodes: [warn-1]
+ - stage: cold
+ target: /mnt/cold-0/measure/data-copy
+ nodes: [cold-0]
diff --git a/banyand/cmd/migration/example/runner-pod.yaml b/banyand/cmd/migration/example/runner-pod.yaml
new file mode 100644
index 000000000..730116389
--- /dev/null
+++ b/banyand/cmd/migration/example/runner-pod.yaml
@@ -0,0 +1,80 @@
+# Licensed to Apache Software Foundation (ASF) under one or more contributor
+# license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright
+# ownership. Apache Software Foundation (ASF) licenses this file to you under
+# the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+# Sample runner pod for in-place live migration. Adapt every name / PVC /
+# mount path to your real cluster topology before applying.
+#
+# This pod is meant to be launched while the BanyanDB data StatefulSets
+# are scaled to 0 so the RWO measure PVCs are free to be re-mounted here.
+# It mounts every data PVC the migration needs to read from OR write to,
+# plus a small emptyDir at /work for the plan file and per-group union
+# sidx staging.
+#
+# Key fields the operator typically tweaks:
+# - image use the same banyandb image your cluster runs;
+# it already ships the `banyand-migration`
+# binary at /banyand-migration.
+# - resources.limits.memory migration sets GOMEMLIMIT to 85% of this.
+# - volumeMounts / volumes one (mountPath, PVC) pair per node × catalog
+# the plan references. Mount measure PVCs at
+# /mnt/<node>/measure and hot schema-property
+# PVCs at /mnt/<node>/schema-property.
+apiVersion: v1
+kind: Pod
+metadata:
+ name: bydb-migration-runner
+ labels:
+ app.kubernetes.io/component: bydb-migration-runner
+spec:
+ restartPolicy: Never
+ containers:
+ - name: runner
+ image: apache/skywalking-banyandb:latest
+ command: ["sleep", "infinity"]
+ resources:
+ requests:
+ cpu: "2"
+ memory: "8Gi"
+ limits:
+ cpu: "4"
+ memory: "20Gi"
+ volumeMounts:
+ - { name: work, mountPath: /work }
+ # Measure + schema-property PVCs from each node the plan references.
+ # `subPath` is only needed when the live pod uses a sub-key per
+ # catalog (typical for the hot tier in the official helm chart).
+ - { name: hot-0-measure, mountPath: /mnt/hot-0/measure, subPath: measure }
+ - { name: hot-0-schema-property, mountPath: /mnt/hot-0/schema-property, subPath: schema-property }
+ - { name: hot-1-measure, mountPath: /mnt/hot-1/measure, subPath: measure }
+ - { name: hot-1-schema-property, mountPath: /mnt/hot-1/schema-property, subPath: schema-property }
+ # Consolidated (single-catalog-per-PVC) warm + cold tiers — no subPath.
+ - { name: warn-0, mountPath: /mnt/warn-0 }
+ - { name: warn-1, mountPath: /mnt/warn-1 }
+ - { name: cold-0, mountPath: /mnt/cold-0 }
+ volumes:
+ # Scratch space: plan file + per-group union sidx staging.
+ - name: work
+ emptyDir:
+ sizeLimit: 4Gi
+ # Replace each claimName below with the PVC name your cluster uses.
+ - { name: hot-0-measure, persistentVolumeClaim: { claimName: hot-measure-data-demo-banyandb-data-hot-0 } }
+ - { name: hot-0-schema-property, persistentVolumeClaim: { claimName: hot-schema-property-data-demo-banyandb-data-hot-0 } }
+ - { name: hot-1-measure, persistentVolumeClaim: { claimName: hot-measure-data-demo-banyandb-data-hot-1 } }
+ - { name: hot-1-schema-property, persistentVolumeClaim: { claimName: hot-schema-property-data-demo-banyandb-data-hot-1 } }
+ - { name: warn-0, persistentVolumeClaim: { claimName: warm-data-demo-banyandb-data-warm-0 } }
+ - { name: warn-1, persistentVolumeClaim: { claimName: warm-data-demo-banyandb-data-warm-1 } }
+ - { name: cold-0, persistentVolumeClaim: { claimName: cold-data-demo-banyandb-data-cold-0 } }
diff --git a/banyand/cmd/migration/main.go b/banyand/cmd/migration/main.go
new file mode 100644
index 000000000..8746711ac
--- /dev/null
+++ b/banyand/cmd/migration/main.go
@@ -0,0 +1,88 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+// Package main is the CLI entry point of the BanyanDB measure-migration
+// tool. The root command exposes three subcommands:
+//
+// copy - file-level direct copy of measure parts into a data pod's
+// measure root, with row-level grid alignment against each
+// target stage's SegmentInterval.
+// verify - read-only inspection of a previous copy run: source vs
+// target row counts, target segment grid alignment, sidx
+// doc counts. Takes the same --copy-config plan as `copy`.
+// analyze - row-level (seriesID, timestamp) duplicate analysis for
+// one (entry, group): within-part dup rows plus the
+// src-vs-target multiset diff that explains any gap
+// reported by `verify`. Read-only.
+//
+// See banyand/cmd/migration/README.md for the full operator runbook.
+package main
+
+import (
+ "fmt"
+ "log"
+ "net/http"
+ _ "net/http/pprof" //nolint:gosec // registers /debug/pprof/* handlers on http.DefaultServeMux; opt-in via --pprof-addr
+ "os"
+
+ "github.com/spf13/cobra"
+)
+
+func main() {
+ var pprofAddr string
+ root := &cobra.Command{
+ Use: "migration",
+ Short: "BanyanDB measure data migration tool",
+ Long: `migration moves measure data from a backup snapshot, or from a live
+data PVC mount, into a target measure-data root.
+
+ copy - copy backup/live measure part files directly into the target,
+ aligning rows to each entry stage's SegmentInterval. Fast,
+ preserves all tag content, requires running inside the target
+ pod (or on a host that owns the target directory exclusively).
+ verify - re-read the same plan and report per-(entry, group) source
+ vs target row counts, target segment grid alignment, and
+ sidx doc counts. Read-only; no exit code on mismatch.
+ analyze - row-level (seriesID, timestamp) duplicate analysis for one
+ (entry, group): scans src parts and the target group, reports
+ within-part dup rows and the src-vs-target multiset diff to
+ explain any row-count gap reported by 'verify'. Read-only.`,
+ PersistentPreRun: func(_ *cobra.Command, _ []string) {
+ if pprofAddr == "" {
+ return
+ }
+ go func() {
+ log.Printf("pprof listening on http://%s/debug/pprof/", pprofAddr)
+ if err := http.ListenAndServe(pprofAddr, nil); err != nil { //nolint:gosec // operator-supplied debug endpoint, no public ACL needed
+ log.Printf("pprof server exited: %v", err)
+ }
+ }()
+ },
+ }
+
+ root.PersistentFlags().StringVar(&pprofAddr, "pprof-addr", "",
+ "if non-empty, expose net/http/pprof on this host:port (e.g. 127.0.0.1:6060)")
+
+ root.AddCommand(newCopyCmd())
+ root.AddCommand(newVerifyCmd())
+ root.AddCommand(newAnalyzeCmd())
+
+ if err := root.Execute(); err != nil {
+ fmt.Fprintln(os.Stderr, err)
+ os.Exit(1)
+ }
+}
diff --git a/banyand/cmd/migration/plan.go b/banyand/cmd/migration/plan.go
new file mode 100644
index 000000000..7da776fdf
--- /dev/null
+++ b/banyand/cmd/migration/plan.go
@@ -0,0 +1,301 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package main
+
+import (
+ "fmt"
+ "os"
+ "regexp"
+
+ "sigs.k8s.io/yaml"
+
+ "github.com/apache/skywalking-banyandb/banyand/measure"
+)
+
+// CopyPlan is the YAML config schema consumed by `migration copy`.
+//
+// One Source describes WHERE the measure data + schema-property live —
+// either a local backup snapshot (source.backup) or a set of live PVC
+// mount paths (source.live). The two modes are mutually exclusive: a
+// plan must specify exactly one. Each Entry then names a destination
+// (stage + target) plus the subset of node names whose data should be
+// rewritten into that entry's target.
+type CopyPlan struct {
+ // StagingDir is where per-group union sidx artifacts land. Optional;
+ // empty means use os.MkdirTemp under TMPDIR. The directory is
+ // removed in full when the run finishes (success or failure). If
+ // specified, it must not exist yet OR must already be an empty
+ // directory — otherwise the loader aborts to avoid clobbering
+ // unrelated data.
+ StagingDir string `json:"staging_dir,omitempty"`
+
+ Source CopySource `json:"source"`
+ Groups []string `json:"groups"`
+ Entries []CopyEntry `json:"entries"`
+}
+
+// CopySource is a discriminated union: exactly one of Backup or Live
+// must be set.
+type CopySource struct {
+ Backup *BackupSource `json:"backup,omitempty"`
+ Live *LiveSource `json:"live,omitempty"`
+}
+
+// BackupSource resolves measure data + schema-property from a backup
+// snapshot rooted at Root for the given Date. The on-disk layout under
+// Root follows the backup snapshot's standard per-node, per-date
+// directory structure.
+type BackupSource struct {
+ Root string `json:"root"`
+ Date string `json:"date"`
+}
+
+// LiveSource resolves measure data + schema-property from live PVC
+// mount paths inside the migration runner pod. Each entry's `nodes`
+// list selects the (node, root) pairs to migrate for that entry.
+type LiveSource struct {
+ // Stages maps stage name to the (node, root) pairs that hold that
+ // stage's measure data. Per-stage so the migration can run
+ // stage-specific work without ambiguity.
+ Stages map[string][]LiveStageNode `json:"stages"`
+
+ // SchemaPropertyPath points the schema loader directly at a
+ // schema-property catalog directory exposed by the runner pod.
+ SchemaPropertyPath string `json:"schema_property_path"`
+}
+
+// LiveStageNode binds a logical node name to a runner-pod path that
+// holds that node's measure-data root (the directory whose children
+// are the per-group segment trees BanyanDB writes during normal
+// operation).
+type LiveStageNode struct {
+ Node string `json:"node"`
+ Root string `json:"root"`
+}
+
+// CopyEntry describes one fan-out destination plus the subset of nodes
+// whose source data feeds into it. In backup mode the nodes are backup
+// directory names (`hot-0`, `warn-1`, …); in live mode they reference
+// LiveSource.Stages[Stage][].Node entries.
+type CopyEntry struct {
+ Stage string `json:"stage"`
+ Target string `json:"target"`
+ Nodes []string `json:"nodes"`
+}
+
+var dateRE = regexp.MustCompile(`^\d{4}-\d{2}-\d{2}$`)
+
+// LoadCopyPlan reads and validates a YAML CopyPlan from path.
+func LoadCopyPlan(path string) (*CopyPlan, error) {
+ if path == "" {
+ return nil, fmt.Errorf("--copy-config is required")
+ }
+ raw, err := os.ReadFile(path)
+ if err != nil {
+ return nil, fmt.Errorf("read %s: %w", path, err)
+ }
+ var plan CopyPlan
+ if err := yaml.Unmarshal(raw, &plan); err != nil {
+ return nil, fmt.Errorf("parse %s: %w", path, err)
+ }
+ if err := plan.Validate(); err != nil {
+ return nil, fmt.Errorf("invalid config %s: %w", path, err)
+ }
+ return &plan, nil
+}
+
+// Validate enforces the structural rules described on CopyPlan.
+//
+//nolint:gocyclo // straight-line schema validation; splitting hurts readability.
+func (p *CopyPlan) Validate() error {
+ if (p.Source.Backup == nil) == (p.Source.Live == nil) {
+ return fmt.Errorf("source must set exactly one of backup or live")
+ }
+ if p.Source.Backup != nil {
+ if p.Source.Backup.Root == "" {
+ return fmt.Errorf("source.backup.root is required")
+ }
+ if p.Source.Backup.Date == "" {
+ return fmt.Errorf("source.backup.date is required")
+ }
+ if !dateRE.MatchString(p.Source.Backup.Date) {
+ return fmt.Errorf("source.backup.date %q does not match YYYY-MM-DD", p.Source.Backup.Date)
+ }
+ } else {
+ live := p.Source.Live
+ if live.SchemaPropertyPath == "" {
+ return fmt.Errorf("source.live.schema_property_path is required")
+ }
+ if len(live.Stages) == 0 {
+ return fmt.Errorf("source.live.stages must list at least one stage")
+ }
+ for stage, pairs := range live.Stages {
+ if len(pairs) == 0 {
+ return fmt.Errorf("source.live.stages[%q] must list at least one node/root pair", stage)
+ }
+ seenNode := make(map[string]bool, len(pairs))
+ for i, n := range pairs {
+ if n.Node == "" {
+ return fmt.Errorf("source.live.stages[%q][%d].node is empty", stage, i)
+ }
+ if n.Root == "" {
+ return fmt.Errorf("source.live.stages[%q][%d].root is empty", stage, i)
+ }
+ if seenNode[n.Node] {
+ return fmt.Errorf("source.live.stages[%q] contains duplicate node %q", stage, n.Node)
+ }
+ seenNode[n.Node] = true
+ }
+ }
+ }
+ if len(p.Groups) == 0 {
+ return fmt.Errorf("groups must list at least one measure group")
+ }
+ seenGroup := make(map[string]bool, len(p.Groups))
+ for _, g := range p.Groups {
+ if g == "" {
+ return fmt.Errorf("groups contains an empty entry")
+ }
+ if seenGroup[g] {
+ return fmt.Errorf("groups contains duplicate entry %q", g)
+ }
+ seenGroup[g] = true
+ }
+ if len(p.Entries) == 0 {
+ return fmt.Errorf("entries must list at least one (stage, target) destination")
+ }
+ seenTarget := make(map[string]int, len(p.Entries))
+ for i, e := range p.Entries {
+ if e.Stage == "" {
+ return fmt.Errorf("entries[%d].stage is required", i)
+ }
+ if e.Target == "" {
+ return fmt.Errorf("entries[%d].target is required", i)
+ }
+ if len(e.Nodes) == 0 {
+ return fmt.Errorf("entries[%d].nodes must list at least one node", i)
+ }
+ seenNode := make(map[string]bool, len(e.Nodes))
+ for j, n := range e.Nodes {
+ if n == "" {
+ return fmt.Errorf("entries[%d].nodes[%d] is empty", i, j)
+ }
+ if seenNode[n] {
+ return fmt.Errorf("entries[%d].nodes contains duplicate %q", i, n)
+ }
+ seenNode[n] = true
+ }
+ if p.Source.Live != nil {
+ // every entry node must match a Stages[stage][].Node
+ validNodes := map[string]bool{}
+ for _, n := range p.Source.Live.Stages[e.Stage] {
+ validNodes[n.Node] = true
+ }
+ for _, n := range e.Nodes {
+ if !validNodes[n] {
+ return fmt.Errorf("entries[%d].nodes contains %q which is not defined under source.live.stages[%q]",
+ i, n, e.Stage)
+ }
+ }
+ }
+ if prev, dup := seenTarget[e.Target]; dup {
+ return fmt.Errorf("entries[%d].target %q duplicates entries[%d].target",
+ i, e.Target, prev)
+ }
+ seenTarget[e.Target] = i
+ }
+ return nil
+}
+
+// PrepareStagingDir resolves p.StagingDir: it returns either the
+// user-specified path (after asserting it is absent or empty) or a
+// fresh os.MkdirTemp directory. Callers must defer os.RemoveAll on the
+// returned path to honor the "cleanup after run" contract.
+func PrepareStagingDir(configured string) (string, error) {
+ if configured == "" {
+ return os.MkdirTemp("", "banyandb-migration-staging-*")
+ }
+ info, err := os.Stat(configured)
+ if err == nil {
+ if !info.IsDir() {
+ return "", fmt.Errorf("staging_dir %q exists and is not a directory", configured)
+ }
+ entries, readErr := os.ReadDir(configured)
+ if readErr != nil {
+ return "", fmt.Errorf("inspect staging_dir %q: %w", configured, readErr)
+ }
+ if len(entries) > 0 {
+ return "", fmt.Errorf("staging_dir %q is not empty — refusing to use it (delete it first to confirm)", configured)
+ }
+ return configured, nil
+ }
+ if !os.IsNotExist(err) {
+ return "", fmt.Errorf("stat staging_dir %q: %w", configured, err)
+ }
+ if mkErr := os.MkdirAll(configured, 0o755); mkErr != nil {
+ return "", fmt.Errorf("create staging_dir %q: %w", configured, mkErr)
+ }
+ return configured, nil
+}
+
+// ToDirectCopyConfig converts the YAML plan into the measure package's
+// internal DirectCopyConfig — the shared input both `copy` and `verify`
+// drive. Caller supplies the staging dir; pass "" when only doing
+// read-only verification (verify never writes union-sidx artifacts).
+func (p *CopyPlan) ToDirectCopyConfig(stagingDir string) measure.DirectCopyConfig {
+ cfg := measure.DirectCopyConfig{
+ Groups: p.Groups,
+ SidxStagingDir: stagingDir,
+ }
+ entries := make([]measure.DirectCopyEntry, len(p.Entries))
+ if p.Source.Backup != nil {
+ cfg.BackupDir = p.Source.Backup.Root
+ cfg.Date = p.Source.Backup.Date
+ for i, e := range p.Entries {
+ entries[i] = measure.DirectCopyEntry{
+ Stage: e.Stage,
+ Target: e.Target,
+ Nodes: append([]string(nil), e.Nodes...),
+ }
+ }
+ } else {
+ cfg.SchemaPropertyPath = p.Source.Live.SchemaPropertyPath
+ for i, e := range p.Entries {
+ rootByNode := map[string]string{}
+ for _, n := range p.Source.Live.Stages[e.Stage] {
+ rootByNode[n.Node] = n.Root
+ }
+ srcs := make([]string, 0, len(e.Nodes))
+ for _, n := range e.Nodes {
+ srcs = append(srcs, rootByNode[n])
+ }
+ // Nodes is also propagated in live mode so log lines that
+ // reference the entry by its node identifier (hot-0, …)
+ // stay informative — Source carries the same info as paths
+ // but is noisier in operator-facing output.
+ entries[i] = measure.DirectCopyEntry{
+ Stage: e.Stage,
+ Target: e.Target,
+ Source: srcs,
+ Nodes: append([]string(nil), e.Nodes...),
+ }
+ }
+ }
+ cfg.Entries = entries
+ return cfg
+}
diff --git a/banyand/cmd/migration/plan_test.go b/banyand/cmd/migration/plan_test.go
new file mode 100644
index 000000000..264e49f11
--- /dev/null
+++ b/banyand/cmd/migration/plan_test.go
@@ -0,0 +1,290 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package main
+
+import (
+ "os"
+ "path/filepath"
+ "strings"
+ "testing"
+)
+
+// validBackupPlan returns a minimal CopyPlan in backup mode that
+// passes Validate. Tests mutate fields to assert per-rule failures.
+func validBackupPlan() *CopyPlan {
+ return &CopyPlan{
+ Source: CopySource{
+ Backup: &BackupSource{
+ Root: "/tmp/backup",
+ Date: "2026-05-19",
+ },
+ },
+ Groups: []string{"sw_metricsMinute"},
+ Entries: []CopyEntry{
+ {Stage: "hot", Target: "/tmp/hot", Nodes: []string{"hot-0"}},
+ },
+ }
+}
+
+// validLivePlan returns a minimal CopyPlan in live mode.
+func validLivePlan() *CopyPlan {
+ return &CopyPlan{
+ Source: CopySource{
+ Live: &LiveSource{
+ SchemaPropertyPath: "/mnt/hot-0/schema-property/data/_schema",
+ Stages: map[string][]LiveStageNode{
+ "hot": {{Node: "hot-0", Root: "/mnt/hot-0/measure/data"}},
+ },
+ },
+ },
+ Groups: []string{"sw_metricsMinute"},
+ Entries: []CopyEntry{
+ {Stage: "hot", Target: "/mnt/hot-0/measure/data-copy", Nodes: []string{"hot-0"}},
+ },
+ }
+}
+
+func TestValidate_okBackup(t *testing.T) {
+ if err := validBackupPlan().Validate(); err != nil {
+ t.Fatalf("expected ok, got %v", err)
+ }
+}
+
+func TestValidate_okLive(t *testing.T) {
+ if err := validLivePlan().Validate(); err != nil {
+ t.Fatalf("expected ok, got %v", err)
+ }
+}
+
+func TestValidate_sourceXOR_neither(t *testing.T) {
+ p := validBackupPlan()
+ p.Source = CopySource{}
+ if err := p.Validate(); err == nil || !strings.Contains(err.Error(), "exactly one of backup or live") {
+ t.Fatalf("expected xor error, got %v", err)
+ }
+}
+
+func TestValidate_sourceXOR_both(t *testing.T) {
+ p := validBackupPlan()
+ p.Source.Live = &LiveSource{
+ SchemaPropertyPath: "/p",
+ Stages: map[string][]LiveStageNode{
+ "hot": {{Node: "hot-0", Root: "/r"}},
+ },
+ }
+ if err := p.Validate(); err == nil || !strings.Contains(err.Error(), "exactly one of backup or live") {
+ t.Fatalf("expected xor error, got %v", err)
+ }
+}
+
+func TestValidate_backupMissingRoot(t *testing.T) {
+ p := validBackupPlan()
+ p.Source.Backup.Root = ""
+ if err := p.Validate(); err == nil || !strings.Contains(err.Error(), "source.backup.root") {
+ t.Fatalf("expected root error, got %v", err)
+ }
+}
+
+func TestValidate_backupBadDate(t *testing.T) {
+ p := validBackupPlan()
+ p.Source.Backup.Date = "2026/05/19"
+ if err := p.Validate(); err == nil || !strings.Contains(err.Error(), "YYYY-MM-DD") {
+ t.Fatalf("expected date format error, got %v", err)
+ }
+}
+
+func TestValidate_liveMissingSchemaPropertyPath(t *testing.T) {
+ p := validLivePlan()
+ p.Source.Live.SchemaPropertyPath = ""
+ if err := p.Validate(); err == nil || !strings.Contains(err.Error(), "schema_property_path") {
+ t.Fatalf("expected schema_property_path error, got %v", err)
+ }
+}
+
+func TestValidate_liveEmptyStages(t *testing.T) {
+ p := validLivePlan()
+ p.Source.Live.Stages = nil
+ if err := p.Validate(); err == nil || !strings.Contains(err.Error(), "stages") {
+ t.Fatalf("expected stages error, got %v", err)
+ }
+}
+
+func TestValidate_liveStagePairsEmpty(t *testing.T) {
+ p := validLivePlan()
+ p.Source.Live.Stages["hot"] = nil
+ if err := p.Validate(); err == nil || !strings.Contains(err.Error(), "node/root pair") {
+ t.Fatalf("expected pairs error, got %v", err)
+ }
+}
+
+func TestValidate_liveDuplicateNodeInStage(t *testing.T) {
+ p := validLivePlan()
+ p.Source.Live.Stages["hot"] = []LiveStageNode{
+ {Node: "hot-0", Root: "/a"},
+ {Node: "hot-0", Root: "/b"},
+ }
+ if err := p.Validate(); err == nil || !strings.Contains(err.Error(), "duplicate") {
+ t.Fatalf("expected duplicate error, got %v", err)
+ }
+}
+
+func TestValidate_emptyGroups(t *testing.T) {
+ p := validBackupPlan()
+ p.Groups = nil
+ if err := p.Validate(); err == nil || !strings.Contains(err.Error(), "groups") {
+ t.Fatalf("expected groups error, got %v", err)
+ }
+}
+
+func TestValidate_duplicateGroup(t *testing.T) {
+ p := validBackupPlan()
+ p.Groups = []string{"a", "a"}
+ if err := p.Validate(); err == nil || !strings.Contains(err.Error(), "duplicate") {
+ t.Fatalf("expected duplicate group error, got %v", err)
+ }
+}
+
+func TestValidate_emptyEntries(t *testing.T) {
+ p := validBackupPlan()
+ p.Entries = nil
+ if err := p.Validate(); err == nil || !strings.Contains(err.Error(), "entries") {
+ t.Fatalf("expected entries error, got %v", err)
+ }
+}
+
+func TestValidate_entryMissingStageOrTarget(t *testing.T) {
+ for _, tc := range []struct {
+ name string
+ need string
+ entry CopyEntry
+ }{
+ {"stage", "stage", CopyEntry{Stage: "", Target: "/t", Nodes: []string{"hot-0"}}},
+ {"target", "target", CopyEntry{Stage: "hot", Target: "", Nodes: []string{"hot-0"}}},
+ {"nodes", "nodes", CopyEntry{Stage: "hot", Target: "/t", Nodes: nil}},
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ p := validBackupPlan()
+ p.Entries = []CopyEntry{tc.entry}
+ if err := p.Validate(); err == nil || !strings.Contains(err.Error(), tc.need) {
+ t.Fatalf("expected %s error, got %v", tc.need, err)
+ }
+ })
+ }
+}
+
+func TestValidate_duplicateTarget(t *testing.T) {
+ p := validBackupPlan()
+ p.Entries = []CopyEntry{
+ {Stage: "hot", Target: "/same", Nodes: []string{"hot-0"}},
+ {Stage: "warm", Target: "/same", Nodes: []string{"warn-0"}},
+ }
+ if err := p.Validate(); err == nil || !strings.Contains(err.Error(), "duplicates") {
+ t.Fatalf("expected duplicate target error, got %v", err)
+ }
+}
+
+func TestValidate_liveEntryNodeNotDefined(t *testing.T) {
+ p := validLivePlan()
+ p.Entries[0].Nodes = []string{"unknown-node"}
+ if err := p.Validate(); err == nil || !strings.Contains(err.Error(), "not defined under source.live.stages") {
+ t.Fatalf("expected undefined-node error, got %v", err)
+ }
+}
+
+func TestLoadCopyPlan_roundtripLive(t *testing.T) {
+ dir := t.TempDir()
+ path := filepath.Join(dir, "plan.yaml")
+ yamlContent := `
+source:
+ live:
+ schema_property_path: /mnt/hot-0/schema-property/data/_schema
+ stages:
+ hot:
+ - { node: hot-0, root: /mnt/hot-0/measure/data }
+ - { node: hot-1, root: /mnt/hot-1/measure/data }
+groups:
+ - sw_metricsMinute
+entries:
+ - stage: hot
+ target: /mnt/hot-0/measure/data-copy
+ nodes: [hot-0]
+`
+ if err := os.WriteFile(path, []byte(yamlContent), 0o600); err != nil {
+ t.Fatalf("write: %v", err)
+ }
+ plan, err := LoadCopyPlan(path)
+ if err != nil {
+ t.Fatalf("load: %v", err)
+ }
+ if plan.Source.Live == nil {
+ t.Fatalf("expected live source, got %+v", plan.Source)
+ }
+ if got := plan.Source.Live.Stages["hot"]; len(got) != 2 || got[0].Node != "hot-0" {
+ t.Fatalf("unexpected stages: %+v", got)
+ }
+ if plan.Entries[0].Nodes[0] != "hot-0" {
+ t.Fatalf("unexpected entry node: %+v", plan.Entries[0])
+ }
+}
+
+func TestPrepareStagingDir_emptyConfiguredUsesTmpDir(t *testing.T) {
+ dir, err := PrepareStagingDir("")
+ if err != nil {
+ t.Fatalf("prepare: %v", err)
+ }
+ defer os.RemoveAll(dir)
+ if !strings.Contains(dir, "banyandb-migration-staging-") {
+ t.Fatalf("expected MkdirTemp prefix, got %s", dir)
+ }
+}
+
+func TestPrepareStagingDir_emptyExistingDirIsReused(t *testing.T) {
+ base := t.TempDir()
+ got, err := PrepareStagingDir(base)
+ if err != nil {
+ t.Fatalf("prepare: %v", err)
+ }
+ if got != base {
+ t.Fatalf("expected %s, got %s", base, got)
+ }
+}
+
+func TestPrepareStagingDir_existingNonEmptyRejected(t *testing.T) {
+ base := t.TempDir()
+ if err := os.WriteFile(filepath.Join(base, "marker"), nil, 0o600); err != nil {
+ t.Fatalf("seed: %v", err)
+ }
+ if _, err := PrepareStagingDir(base); err == nil || !strings.Contains(err.Error(), "not empty") {
+ t.Fatalf("expected non-empty rejection, got %v", err)
+ }
+}
+
+func TestPrepareStagingDir_createsNonExistingDir(t *testing.T) {
+ base := filepath.Join(t.TempDir(), "subdir-that-does-not-exist")
+ got, err := PrepareStagingDir(base)
+ if err != nil {
+ t.Fatalf("prepare: %v", err)
+ }
+ if got != base {
+ t.Fatalf("expected %s, got %s", base, got)
+ }
+ info, statErr := os.Stat(base)
+ if statErr != nil || !info.IsDir() {
+ t.Fatalf("expected dir created, stat=%v info=%+v", statErr, info)
+ }
+}
diff --git a/banyand/cmd/migration/scripts/verify-data.sh b/banyand/cmd/migration/scripts/verify-data.sh
new file mode 100755
index 000000000..3cf05fae4
--- /dev/null
+++ b/banyand/cmd/migration/scripts/verify-data.sh
@@ -0,0 +1,113 @@
+#!/usr/bin/env bash
+# Licensed to Apache Software Foundation (ASF) under one or more contributor
+# license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright
+# ownership. Apache Software Foundation (ASF) licenses this file to you under
+# the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+# Per-tier data presence check for migrated measure groups via the liaison
+# HTTP API. Probes the most recent N days; for each day issues a tier-shaped
+# query at the granularity that tier stores:
+#
+# - sw_metricsDay : full-day window (00:00:00 .. 23:59:59 UTC)
+# - sw_metricsHour : the 08:00 hour (08:00:00 .. 08:59:59 UTC)
+# - sw_metricsMinute : the 08:08 minute (08:08:00 .. 08:08:59 UTC)
+#
+# A trailing "★MISSING" marker is appended whenever the probe returns zero
+# data points. There is no per-tier row-count threshold — any non-zero
+# response counts as "present".
+#
+# Prereqs: kubectl port-forward to a liaison HTTP port (default 17913).
+#
+# Override via env:
+# LIAISON_HTTP (default: http://127.0.0.1:17913)
+# DAYS (default: 15) - number of most recent UTC days to probe
+set -euo pipefail
+
+LIAISON_HTTP="${LIAISON_HTTP:-http://127.0.0.1:17913}"
+DAYS="${DAYS:-15}"
+
+# Emit today, today-1, ..., today-(N-1) in YYYY-MM-DD UTC. GNU and BSD
+# `date` use different flags for "N days ago" — probe once and pick the
+# right invocation.
+if date -u -d "0 days ago" +%Y-%m-%d >/dev/null 2>&1; then
+ DATE_FLAVOR=gnu
+elif date -u -v-0d +%Y-%m-%d >/dev/null 2>&1; then
+ DATE_FLAVOR=bsd
+else
+ echo "ERROR: neither GNU nor BSD date supports date-arithmetic on this host" >&2
+ exit 1
+fi
+
+day_minus() {
+ local i="$1"
+ case "$DATE_FLAVOR" in
+ gnu) date -u -d "${i} days ago" +%Y-%m-%d ;;
+ bsd) date -u -v-"${i}d" +%Y-%m-%d ;;
+ esac
+}
+
+# Count occurrences of "timestamp" inside the response body — the liaison
+# /api/v1/measure/data response only carries one `"timestamp"` field per
+# data point, so this is a reliable row count without invoking jq/python.
+# `grep -o` returns exit 1 when there are zero matches; isolate it in a
+# subshell + `|| echo 0` so empty responses don't trip `set -e`/pipefail.
+probe() {
+ local group="$1" measure="$2" begin="$3" end="$4"
+ local body resp
+ body=$(cat <<EOF
+{
+ "groups": ["$group"],
+ "name": "$measure",
+ "stages": ["hot","warm","cold"],
+ "timeRange": {"begin": "$begin", "end": "$end"},
+ "tagProjection": {"tagFamilies":[{"name":"storage-only","tags":["entity_id"]}]},
+ "fieldProjection": {"names":[]},
+ "limit": 100000
+}
+EOF
+)
+ resp=$(curl -s -XPOST "$LIAISON_HTTP/api/v1/measure/data" \
+ -H 'Content-Type: application/json' -d "$body" || true)
+ # Tolerate grep's exit 1 on zero matches by wrapping it in `|| true`,
+ # so the pipeline always reaches wc with a (possibly empty) input.
+ printf '%s' "$resp" | { grep -o '"timestamp"' || true; } | wc -l | tr -d ' '
+}
+
+echo "data presence check across most recent $DAYS day(s)"
+echo " liaison: $LIAISON_HTTP"
+echo
+
+scan() {
+ local label="$1" group="$2" measure="$3" begin_suffix="$4" end_suffix="$5" window_label="$6"
+ echo "-- $label ($group / $measure, window: $window_label) --"
+ local bad=0 i=0 day n marker
+ while [ "$i" -lt "$DAYS" ]; do
+ day=$(day_minus "$i")
+ n=$(probe "$group" "$measure" "${day}${begin_suffix}" "${day}${end_suffix}")
+ marker=""
+ if [ "${n:-0}" -eq 0 ]; then
+ marker=" ★MISSING"
+ bad=$((bad + 1))
+ fi
+ printf " %s: %s%s\n" "$day" "$n" "$marker"
+ i=$((i + 1))
+ done
+ echo " $bad day(s) missing"
+ echo
+}
+
+scan "Day tier" sw_metricsDay service_apdex_day T00:00:00Z T23:59:59Z "full day"
+scan "Hour tier" sw_metricsHour service_apdex_hour T08:00:00Z T08:59:59Z "08:00..09:00"
+scan "Minute tier" sw_metricsMinute service_apdex_minute T08:08:00Z T08:08:59Z "08:08..08:09"
diff --git a/banyand/cmd/migration/verify.go b/banyand/cmd/migration/verify.go
new file mode 100644
index 000000000..47ffefa4b
--- /dev/null
+++ b/banyand/cmd/migration/verify.go
@@ -0,0 +1,339 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package main
+
+import (
+ "context"
+ "fmt"
+ "os"
+ "os/signal"
+ "strings"
+ "syscall"
+ "text/tabwriter"
+
+ "github.com/spf13/cobra"
+
+ "github.com/apache/skywalking-banyandb/banyand/measure"
+)
+
+func newVerifyCmd() *cobra.Command {
+ var configPath string
+
+ cmd := &cobra.Command{
+ Use: "verify",
+ Short: "Inspect a copy run: per-(entry, group) source vs target row counts, target segment grid alignment, sidx doc counts",
+ Long: `verify reads the same plan.yaml that the copy run consumed and walks
+each (entry, group) read-only:
+
+ - sums source row count by opening every src/seg-*/shard-*/<partID>
+ and totalling partMetadata.TotalCount
+ - enumerates target/seg-*/, opens each part and the per-seg union sidx
+ - flags whether each target seg's start time aligns to the entry
+ stage's SegmentInterval grid (IntervalRule.Standard)
+ - prints all numbers (source rows, target rows per seg, sidxDocs,
+ aligned y/n) to stdout for the operator to inspect
+
+verify never fails on mismatch — it just reports. Use it after a copy
+run, optionally after the data-copy ↔ data swap, to confirm the
+migration result.`,
+ RunE: func(_ *cobra.Command, _ []string) error {
+ plan, err := LoadCopyPlan(configPath)
+ if err != nil {
+ return err
+ }
+ // staging dir is unused by verify (read-only), pass "".
+ cfg := plan.ToDirectCopyConfig("")
+
+ ctx, stop := signal.NotifyContext(context.Background(),
+ os.Interrupt, syscall.SIGTERM)
+ defer stop()
+
+ tally := &verifyTally{}
+ runErr := measure.MigrationVerify(ctx, cfg, func(r measure.EntryGroupReport) {
+ printOneReport(r)
+ tally.absorb(r)
+ })
+ tally.printSummary()
+ return runErr
+ },
+ }
+
+ cmd.Flags().StringVar(&configPath, "copy-config", "",
+ "path to the YAML migration copy plan that was used for `migration copy` (required)")
+ _ = cmd.MarkFlagRequired("copy-config")
+ return cmd
+}
+
+// verifyTally accumulates the per-(node, group) findings the callback
+// stream emits so we can print a single roll-up SUMMARY block at the
+// end of the run — the per-report stream prints itself.
+type verifyTally struct {
+ mismatches []verifyMismatch
+ coverage map[string]map[string]coverageState // node → group → (src/tgt presence)
+ nodeOrder []string // first-seen ordering
+ groupOrder []string // first-seen ordering
+ srcRowsTotal uint64
+ tgtRowsTotal uint64
+ segsTotal int
+ segsMisaligned int
+}
+
+// coverageState records whether SOURCE and TARGET independently hold
+// any rows for one (node, group) pair. Four combinations exist, each
+// surfaced by a distinct token in the coverage table:
+//
+// src=true, tgt=true → "✓" (both present, normal copy success)
+// src=true, tgt=false → "S" (src has data but target is empty — copy lost this group)
+// src=false, tgt=true → "T" (target has data without a source — orphan / leftover)
+// src=false, tgt=false → "--" (neither — PVC hash sharding excluded this group, normal)
+type coverageState struct {
+ src bool
+ tgt bool
+}
+
+// verifyMismatch records one (node, group) where source row count
+// did NOT equal target row count — surfaced in the SUMMARY block so
+// the operator immediately sees which PVCs / groups need follow-up.
+type verifyMismatch struct {
+ Stage string
+ NodeName string
+ Group string
+ SrcRows uint64
+ TgtRows uint64
+}
+
+// entryNodeName picks the coverage-table row label for an EntryGroupReport.
+// Entries usually carry exactly one node (live mode), in which case the
+// label IS that node. When an entry references multiple nodes
+// (backup-mode plans that fan one target out across several backup-node
+// dirs), the report's SrcRows / TargetSegs are aggregated across all of
+// them — so we join the node names with `,` to make it explicit that the
+// row aggregates more than one node. Falls back to the entry stage when
+// EntryNodes is empty (defensive: validation requires at least one).
+func entryNodeName(r measure.EntryGroupReport) string {
+ switch len(r.EntryNodes) {
+ case 0:
+ return r.EntryStage
+ case 1:
+ return r.EntryNodes[0]
+ default:
+ return strings.Join(r.EntryNodes, ",")
+ }
+}
+
+func (t *verifyTally) absorb(r measure.EntryGroupReport) {
+ node := entryNodeName(r)
+ if t.coverage == nil {
+ t.coverage = make(map[string]map[string]coverageState)
+ }
+ if _, ok := t.coverage[node]; !ok {
+ t.coverage[node] = make(map[string]coverageState)
+ t.nodeOrder = append(t.nodeOrder, node)
+ }
+ if _, ok := indexOf(t.groupOrder, r.Group); !ok {
+ t.groupOrder = append(t.groupOrder, r.Group)
+ }
+
+ t.srcRowsTotal += r.SrcRows
+ var tgtRows uint64
+ for _, s := range r.TargetSegs {
+ tgtRows += s.Rows
+ if !s.Aligned {
+ t.segsMisaligned++
+ }
+ }
+ t.tgtRowsTotal += tgtRows
+ t.segsTotal += len(r.TargetSegs)
+
+ t.coverage[node][r.Group] = coverageState{
+ src: r.SrcRows > 0,
+ tgt: tgtRows > 0,
+ }
+
+ if tgtRows != r.SrcRows {
+ t.mismatches = append(t.mismatches, verifyMismatch{
+ Stage: r.EntryStage,
+ NodeName: node,
+ Group: r.Group,
+ SrcRows: r.SrcRows,
+ TgtRows: tgtRows,
+ })
+ }
+}
+
+func indexOf(xs []string, v string) (int, bool) {
+ for i, x := range xs {
+ if x == v {
+ return i, true
+ }
+ }
+ return 0, false
+}
+
+func (t *verifyTally) printSummary() {
+ fmt.Println("== SUMMARY ==")
+ t.printCoverageTable()
+ fmt.Println()
+ fmt.Printf(" target segments : %d\n", t.segsTotal)
+ fmt.Printf(" target segments misaligned : %d\n", t.segsMisaligned)
+ fmt.Printf(" source rows total : %d\n", t.srcRowsTotal)
+ fmt.Printf(" target rows total : %d\n", t.tgtRowsTotal)
+ if len(t.mismatches) == 0 {
+ if t.tgtRowsTotal > t.srcRowsTotal {
+ fmt.Printf(" diff (tgt - src) : %+d (UNEXPECTED — target has MORE rows than source)\n",
+ int64(t.tgtRowsTotal)-int64(t.srcRowsTotal))
+ } else {
+ fmt.Println(" src == tgt")
+ }
+ return
+ }
+ fmt.Println()
+ fmt.Println(" row-count mismatches (tgt - src, negative = rows dropped by copy):")
+ tw := tabwriter.NewWriter(os.Stdout, 0, 0, 2, ' ', 0)
+ fmt.Fprintln(tw, " stage\tnode\tgroup\tsrc\ttgt\tdiff")
+ var total int64
+ for _, m := range t.mismatches {
+ diff := int64(m.TgtRows) - int64(m.SrcRows)
+ total += diff
+ fmt.Fprintf(tw, " %s\t%s\t%s\t%d\t%d\t%+d\n",
+ m.Stage, m.NodeName, m.Group, m.SrcRows, m.TgtRows, diff)
+ }
+ fmt.Fprintf(tw, " %s\t\t\t\t\t%+d\n", "TOTAL", total)
+ _ = tw.Flush()
+ if total < 0 {
+ fmt.Println()
+ fmt.Println(" note: rows dropped are caused by slow-path mustInitFromDataPoints")
+ fmt.Println(" deduping (seriesID, timestamp) within each chunk flush — banyandb's")
+ fmt.Println(" merger can leave duplicated boundary rows in source parts, and")
+ fmt.Println(" only the latest version of each (sid, ts) survives in target.")
+ fmt.Println(" Run `migration analyze --entry-idx <i> --group <g>` for the exact")
+ fmt.Println(" missing (sid, ts, version) rows of any (entry, group) shown above.")
+ }
+}
+
+// printCoverageTable renders a "✓ / --" grid of node × group so the
+// operator can see at a glance which PVCs hold data for which groups
+// (banyandb's hash sharding leaves some cells empty — that's normal).
+func (t *verifyTally) printCoverageTable() {
+ if len(t.nodeOrder) == 0 || len(t.groupOrder) == 0 {
+ return
+ }
+ nodeWidth := len("node")
+ for _, n := range t.nodeOrder {
+ if w := len(n); w > nodeWidth {
+ nodeWidth = w
+ }
+ }
+ colWidth := make([]int, len(t.groupOrder))
+ for i, g := range t.groupOrder {
+ colWidth[i] = len(g)
+ if colWidth[i] < 2 {
+ colWidth[i] = 2
+ }
+ }
+ fmt.Println(" data coverage per (node, group):")
+ fmt.Println(" ✓ = src has rows AND tgt has rows (normal copy success)")
+ fmt.Println(" S = src has rows, tgt is empty (copy lost this group — investigate)")
+ fmt.Println(" T = tgt has rows, src is empty (orphan target — investigate)")
+ fmt.Println(" -- = neither side has rows (PVC does not hold this group; hash-sharding normal)")
+ border := func(left, mid, right, fill string) string {
+ s := left + strings.Repeat(fill, nodeWidth+2)
+ for _, w := range colWidth {
+ s += mid + strings.Repeat(fill, w+2)
+ }
+ return s + right
+ }
+ fmt.Println(" " + border("┌", "┬", "┐", "─"))
+ headerCells := []string{padString("node", nodeWidth)}
+ for i, g := range t.groupOrder {
+ headerCells = append(headerCells, padString(g, colWidth[i]))
+ }
+ fmt.Println(" │ " + strings.Join(headerCells, " │ ") + " │")
+ fmt.Println(" " + border("├", "┼", "┤", "─"))
+ for _, n := range t.nodeOrder {
+ rowCells := []string{padString(n, nodeWidth)}
+ for i, g := range t.groupOrder {
+ s := t.coverage[n][g]
+ var tok string
+ switch {
+ case s.src && s.tgt:
+ tok = "✓"
+ case s.src && !s.tgt:
+ tok = "S"
+ case !s.src && s.tgt:
+ tok = "T"
+ default:
+ tok = "--"
+ }
+ rowCells = append(rowCells, padString(tok, colWidth[i]))
+ }
+ fmt.Println(" │ " + strings.Join(rowCells, " │ ") + " │")
+ }
+ fmt.Println(" " + border("└", "┴", "┘", "─"))
+}
+
+// padString right-pads s with spaces so it visually occupies w cells
+// in a fixed-width terminal. We measure by RUNE count, not byte count,
+// because the coverage cells include "✓" — 3 bytes in UTF-8 but only
+// one terminal cell. Byte-based padding made the table jagged.
+func padString(s string, w int) string {
+ runeCount := len([]rune(s))
+ if runeCount >= w {
+ return s
+ }
+ return s + strings.Repeat(" ", w-runeCount)
+}
+
+// printOneReport renders a single (entry, group) report immediately.
+// Both source paths (one per node) and the target group path are
+// printed so the operator can match the row counts to a specific PVC.
+func printOneReport(r measure.EntryGroupReport) {
+ fmt.Printf("== entry stage=%s target=%s group=%s ==\n",
+ r.EntryStage, r.EntryTarget, r.Group)
+ if len(r.SrcRoots) == 0 {
+ fmt.Printf(" src : (no source dirs resolved for this entry/group)\n")
+ } else {
+ fmt.Printf(" src : %d row(s) across %d part(s) in %d dir(s)\n",
+ r.SrcRows, r.SrcParts, len(r.SrcRoots))
+ for _, p := range r.SrcRoots {
+ fmt.Printf(" %s\n", p)
+ }
+ }
+ if len(r.TargetSegs) == 0 {
+ fmt.Printf(" tgt : (no segments under %s)\n", r.TargetGroup)
+ return
+ }
+ var tgtRows uint64
+ for _, s := range r.TargetSegs {
+ tgtRows += s.Rows
+ }
+ fmt.Printf(" tgt : %d row(s) across %d seg(s) under %s\n",
+ tgtRows, len(r.TargetSegs), r.TargetGroup)
+ for _, s := range r.TargetSegs {
+ alignTag := "aligned"
+ if !s.Aligned {
+ alignTag = "MISALIGNED"
+ }
+ sidxTag := "no-sidx"
+ if s.SidxOpened {
+ sidxTag = fmt.Sprintf("sidxDocs=%d", s.SidxDocCount)
+ }
+ fmt.Printf(" %-22s %s shards=%d parts=%d rows=%d %s\n",
+ s.Seg, alignTag, s.Shards, s.Parts, s.Rows, sidxTag)
+ }
+}
diff --git a/banyand/internal/storage/version.go b/banyand/internal/storage/version.go
index 82d1adaaf..29648c261 100644
--- a/banyand/internal/storage/version.go
+++ b/banyand/internal/storage/version.go
@@ -35,6 +35,25 @@ const (
compatibleVersionsFilename = "versions.yml"
)
+// SegmentMetadataFilename is the filename used for the per-segment
+// metadata document. Exported so out-of-package tooling (e.g. the
+// migration copy CLI) writes the file to the exact location runtime
+// readers expect.
+const SegmentMetadataFilename = metadataFilename
+
+// CurrentSegmentVersion is the version string a freshly-written segment
+// must carry. Exported for the same reason as SegmentMetadataFilename.
+const CurrentSegmentVersion = currentVersion
+
+// SegmentMetadata is the on-disk shape of <segment>/metadata. Kept in
+// sync with the (unexported) segmentMeta runtime reads. Exported so
+// out-of-package writers can produce identical bytes via the same JSON
+// tags (lowercase `version` / `endTime,omitempty`).
+type SegmentMetadata struct {
+ Version string `json:"version"`
+ EndTime string `json:"endTime,omitempty"`
+}
+
var errVersionIncompatible = errors.New("version not compatible")
var compatibleVersions = readCompatibleVersions()
diff --git a/banyand/measure/migration_analyze.go b/banyand/measure/migration_analyze.go
new file mode 100644
index 000000000..3cd8b3f33
--- /dev/null
+++ b/banyand/measure/migration_analyze.go
@@ -0,0 +1,934 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package measure
+
+import (
+ "fmt"
+ "os"
+ "path/filepath"
+ "sort"
+ "strconv"
+ "strings"
+ "sync/atomic"
+ "time"
+
+ "github.com/apache/skywalking-banyandb/api/common"
+ "github.com/apache/skywalking-banyandb/pkg/bytes"
+ "github.com/apache/skywalking-banyandb/pkg/compress/zstd"
+ "github.com/apache/skywalking-banyandb/pkg/encoding"
+ "github.com/apache/skywalking-banyandb/pkg/fs"
+)
+
+// AnalyzeGroupResult is the row-level summary `migration analyze`
+// returns for one (entry, group): how many physical rows are on disk
+// versus how many distinct (seriesID, timestamp) keys they describe.
+//
+// Two diff numbers matter:
+// - DuplicateRows is the GLOBAL diff (cross-part dups). These are
+// NOT dropped by migration copy because slow-path processes one
+// source part at a time and never cross-merges buckets.
+// - PerPartDupRows is the WITHIN-part diff summed over every source
+// part. This IS the count slow-path mustInitFromDataPoints would
+// drop on flush — it should match `verify`'s src-tgt diff.
+type AnalyzeGroupResult struct {
+ SamplesByVersion []AnalyzeKeyMulti
+ PerPartDups []AnalyzePerPart
+ TotalRows uint64
+ UniqueKeys uint64
+ DuplicateRows uint64
+ PerPartDupRows uint64
+ KeysWithDuplicates uint64 // count of (sid, ts) keys that appear in >1 physical rows
+ PartsScanned int
+}
+
+// AnalyzePerPart is one row in the per-part dup breakdown — non-zero
+// PartialDupRows mean THIS specific source part has rows the slow
+// path would dedup on its bucket flush. Boundaries lists the
+// "block-N ends at ts=T, block-N+1 (same series) starts at ts=T"
+// pairs WITHIN this part — these are exactly the rows the
+// merger-fast-path bug (merger.go:306 `<=`) emits.
+type AnalyzePerPart struct {
+ PartID string
+ SegName string
+ ShardName string
+ SourceRoot string
+ Boundaries []AnalyzeBoundaryPair
+ Rows uint64
+ UniqueKeys uint64
+ PartialDupRows uint64
+}
+
+// AnalyzeBoundaryPair names a single boundary collision inside one
+// part: block at BlockA ends at (SeriesID, Timestamp), block at
+// BlockB (immediately following) begins at the same (SeriesID,
+// Timestamp). Both copies survived because the merger fast-path
+// wrote pendingBlock verbatim before doing row-level merge.
+type AnalyzeBoundaryPair struct {
+ SeriesID common.SeriesID
+ Timestamp int64
+ VersionA int64
+ VersionB int64
+ BlockA int
+ BlockB int
+}
+
+// AnalyzeKeyMulti reports one (sid, ts) pair that has more than one
+// row on disk, along with every version banyandb stored for it.
+// Useful operator-facing proof of the "newest version wins" dedup
+// banyandb's merger applies.
+type AnalyzeKeyMulti struct {
+ Versions []VersionWithPath
+ SeriesID common.SeriesID
+ Timestamp int64
+}
+
+// VersionWithPath pairs a row's version field with the absolute part
+// directory it physically lives in, so the operator can grep / cat
+// the on-disk files directly to confirm duplicates.
+type VersionWithPath struct {
+ Path string
+ Version int64
+}
+
+type analyzeKey struct {
+ sid common.SeriesID
+ ts int64
+}
+
+// pathTable interns part paths so we don't pay a string allocation
+// per row in the (sid, ts) → []rowSrc map. Each unique path lands at
+// a small uint32 index; per-row overhead drops from ~80B to 12B.
+type pathTable struct {
+ idx map[string]uint32
+ paths []string
+}
+
+func newPathTable() *pathTable {
+ return &pathTable{idx: make(map[string]uint32, 256)}
+}
+
+func (t *pathTable) intern(p string) uint32 {
+ if i, ok := t.idx[p]; ok {
+ return i
+ }
+ i := uint32(len(t.paths))
+ t.paths = append(t.paths, p)
+ t.idx[p] = i
+ return i
+}
+
+func (t *pathTable) get(i uint32) string {
+ return t.paths[int(i)]
+}
+
+// analyzeRowSrc is a (version, partPathIdx) tuple stored in the
+// (sid, ts) → []analyzeRowSrc map. partPathIdx is an index into the
+// scan's pathTable.
+type analyzeRowSrc struct {
+ version int64
+ pathIdx uint32
+}
+
+// ResolveEntrySrcRoots is the exported wrapper around resolveEntrySrcRoots
+// so cmd/migration can reach the same plan→roots logic the copy + verify
+// pipelines use internally.
+func ResolveEntrySrcRoots(cfg DirectCopyConfig, entry DirectCopyEntry, group string) []string {
+ return resolveEntrySrcRoots(cfg, entry, group)
+}
+
+// AnalyzeMissingRow names one row that exists in src but not in tgt
+// after a migration copy: (sid, ts) appears more times physically in
+// source than in target, and MissingVersions lists the version values
+// that didn't survive into the target (sorted ascending).
+type AnalyzeMissingRow struct {
+ MissingVersions []VersionWithPath
+ SourceVersions []VersionWithPath
+ TargetVersions []VersionWithPath
+ SeriesID common.SeriesID
+ Timestamp int64
+}
+
+// AnalyzeDiffResult is the source-vs-target multiset diff for one
+// (entry, group). It is filled in by AnalyzeGroupDiffWithTarget and
+// answers the question "which exact rows did `migration copy` drop?".
+type AnalyzeDiffResult struct {
+ Missing []AnalyzeMissingRow
+ SourceRows uint64
+ TargetRows uint64
+ MissingRows uint64
+ MissingKeys int
+}
+
+// AnalyzeGroupRows walks every <root>/seg-*/shard-*/<partID>/ in
+// srcRoots, opens each part read-only, decodes only block metadata +
+// timestamps (NOT tag/field bodies — keeps the scan cheap), and
+// aggregates (seriesID, timestamp) keys with their versions.
+//
+// Returns:
+// - TotalRows : every physical row encountered on disk
+// - UniqueKeys : distinct (sid, ts) pairs
+// - DuplicateRows : TotalRows - UniqueKeys
+// - SamplesByVersion[0:sampleCap]: first few (sid, ts) keys that had
+// >1 row on disk, with all version values they carried.
+//
+// Memory: roughly 32 bytes per unique key + (versions × 8B). Cold-tier
+// groups with hundreds of millions of unique keys can be too large
+// for a 20 GiB pod — call AnalyzeGroupRows on a small (entry, group)
+// pair as a sanity probe, not on every shard at once.
+func AnalyzeGroupRows(srcRoots []string, fileSystem fs.FileSystem, sampleCap int) (AnalyzeGroupResult, error) {
+ var res AnalyzeGroupResult
+ keys := make(map[analyzeKey][]analyzeRowSrc, 1<<20)
+ paths := newPathTable()
+
+ // Pre-pass: open each part's metadata only (cheap) to learn the
+ // expected total row count, so the 10s ticker can report a real %.
+ expectedTotal, err := sumGroupExpectedRows(srcRoots, fileSystem)
+ if err != nil {
+ return res, fmt.Errorf("count expected rows: %w", err)
+ }
+ progressDone := startScanProgressReporter(&res.TotalRows, uint64(expectedTotal))
+ defer close(progressDone)
+
+ var (
+ compressed []byte
+ raw []byte
+ bms []blockMetadata
+ timestamps []int64
+ versions []int64
+ )
+
+ for _, root := range srcRoots {
+ segEntries, err := os.ReadDir(root)
+ if err != nil {
+ if os.IsNotExist(err) {
+ continue
+ }
+ return res, fmt.Errorf("read src root %s: %w", root, err)
+ }
+ for _, se := range segEntries {
+ if !se.IsDir() || !strings.HasPrefix(se.Name(), directCopySegPrefix) {
+ continue
+ }
+ segDir := filepath.Join(root, se.Name())
+ shardEntries, shardReadErr := os.ReadDir(segDir)
+ if shardReadErr != nil {
+ return res, fmt.Errorf("read src seg %s: %w", segDir, shardReadErr)
+ }
+ for _, sh := range shardEntries {
+ if !sh.IsDir() || !strings.HasPrefix(sh.Name(), directCopyShardPrefix) {
+ continue
+ }
+ shardDir := filepath.Join(segDir, sh.Name())
+ partEntries, partsReadErr := os.ReadDir(shardDir)
+ if partsReadErr != nil {
+ return res, fmt.Errorf("read src shard %s: %w", shardDir, partsReadErr)
+ }
+ for _, pe := range partEntries {
+ if !pe.IsDir() || !directCopyPartDirPattern.MatchString(pe.Name()) {
+ continue
+ }
+ partID, parseErr := strconv.ParseUint(pe.Name(), 16, 64)
+ if parseErr != nil {
+ return res, fmt.Errorf("parse partID %s: %w", pe.Name(), parseErr)
+ }
+ partDupKeys := make(map[analyzeKey]int, 1<<12)
+ partPath := filepath.Join(shardDir, pe.Name())
+ rowsThisPart, boundaries, scanErr := scanOnePartIntoMap(
+ partID, shardDir, partPath, fileSystem,
+ keys, paths, partDupKeys, &res,
+ &compressed, &raw, &bms, ×tamps, &versions,
+ )
+ if scanErr != nil {
+ return res, scanErr
+ }
+ var partDups uint64
+ for _, c := range partDupKeys {
+ if c > 1 {
+ partDups += uint64(c - 1)
+ }
+ }
+ res.PerPartDupRows += partDups
+ if partDups > 0 {
+ res.PerPartDups = append(res.PerPartDups, AnalyzePerPart{
+ PartID: pe.Name(),
+ SegName: se.Name(),
+ ShardName: sh.Name(),
+ SourceRoot: root,
+ Boundaries: boundaries,
+ Rows: rowsThisPart,
+ UniqueKeys: uint64(len(partDupKeys)),
+ PartialDupRows: partDups,
+ })
+ }
+ }
+ }
+ }
+ }
+
+ res.UniqueKeys = uint64(len(keys))
+ res.DuplicateRows = res.TotalRows - res.UniqueKeys
+
+ // Single pass to count every (sid, ts) key with >1 row AND collect
+ // up to sampleCap samples. Counting always; sampling stops at cap.
+ for k, vs := range keys {
+ if len(vs) <= 1 {
+ continue
+ }
+ res.KeysWithDuplicates++
+ if sampleCap > 0 && len(res.SamplesByVersion) < sampleCap {
+ res.SamplesByVersion = append(res.SamplesByVersion, AnalyzeKeyMulti{
+ SeriesID: k.sid,
+ Timestamp: k.ts,
+ Versions: rowSrcsToVersionsWithPath(vs, paths),
+ })
+ }
+ }
+ return res, nil
+}
+
+// rowSrcsToVersionsWithPath turns the internal (version, pathIdx)
+// slice into the operator-facing []VersionWithPath, sorted by version
+// ascending. Used by every CLI emission so output stays consistent.
+func rowSrcsToVersionsWithPath(rs []analyzeRowSrc, paths *pathTable) []VersionWithPath {
+ out := make([]VersionWithPath, len(rs))
+ for i, r := range rs {
+ out[i] = VersionWithPath{Version: r.version, Path: paths.get(r.pathIdx)}
+ }
+ sort.Slice(out, func(i, j int) bool { return out[i].Version < out[j].Version })
+ return out
+}
+
+// scanOnePartIntoMap opens one part, walks every block in on-disk
+// order, and:
+//
+// - merges every row's (version, partPath) under its (seriesID,
+// timestamp) key in keys (global, across all parts). The path is
+// interned via the shared pathTable so per-row overhead stays at
+// 12B regardless of how long the path is.
+// - increments partDupKeys per-part so the caller derives the
+// within-part dedup count
+// - records every boundary collision (block N ends at ts=T,
+// block N+1 begins at ts=T, same series) — these are exactly
+// the rows the merger-fast-path bug at merger.go:306 emits
+//
+// Returns rowsInPart, boundaries, error.
+func scanOnePartIntoMap(
+ partID uint64,
+ shardDir, partPath string,
+ fileSystem fs.FileSystem,
+ keys map[analyzeKey][]analyzeRowSrc,
+ paths *pathTable,
+ partDupKeys map[analyzeKey]int,
+ res *AnalyzeGroupResult,
+ compressed, raw *[]byte,
+ bms *[]blockMetadata,
+ timestamps, versions *[]int64,
+) (partRows uint64, boundaries []AnalyzeBoundaryPair, err error) {
+ var p *part
+ defer func() {
+ if p != nil {
+ p.close()
+ }
+ if r := recover(); r != nil {
+ err = fmt.Errorf("scan part %016x panicked: %v", partID, r)
+ }
+ }()
+ p = mustOpenFilePart(partID, shardDir, fileSystem)
+ res.PartsScanned++
+
+ blockIdx := 0
+ var prevSet bool
+ var prevSid common.SeriesID
+ var prevTS, prevVersion int64
+ var prevBlockIdx int
+
+ for i := range p.primaryBlockMetadata {
+ pbm := &p.primaryBlockMetadata[i]
+ *compressed = bytes.ResizeOver(*compressed, int(pbm.size))
+ fs.MustReadData(p.primary, int64(pbm.offset), *compressed)
+ var err error
+ *raw, err = zstd.Decompress((*raw)[:0], *compressed)
+ if err != nil {
+ return partRows, boundaries, fmt.Errorf("decompress primary block: %w", err)
+ }
+ *bms, err = unmarshalBlockMetadata((*bms)[:0], *raw)
+ if err != nil {
+ return partRows, boundaries, fmt.Errorf("unmarshal block metadata: %w", err)
+ }
+ for j := range *bms {
+ bm := &(*bms)[j]
+ *timestamps, *versions = mustReadTimestampsFrom((*timestamps)[:0], (*versions)[:0], &bm.timestamps, int(bm.count), p.timestamps)
+ n := int(bm.count)
+ if n == 0 {
+ blockIdx++
+ continue
+ }
+ if prevSet && bm.seriesID == prevSid && (*timestamps)[0] == prevTS {
+ boundaries = append(boundaries, AnalyzeBoundaryPair{
+ SeriesID: prevSid,
+ Timestamp: prevTS,
+ VersionA: prevVersion,
+ VersionB: (*versions)[0],
+ BlockA: prevBlockIdx,
+ BlockB: blockIdx,
+ })
+ }
+ partPathIdx := paths.intern(partPath)
+ for k := 0; k < n; k++ {
+ ak := analyzeKey{sid: bm.seriesID, ts: (*timestamps)[k]}
+ keys[ak] = append(keys[ak], analyzeRowSrc{version: (*versions)[k], pathIdx: partPathIdx})
+ partDupKeys[ak]++
+ atomic.AddUint64(&res.TotalRows, 1)
+ partRows++
+ }
+ prevSet = true
+ prevSid = bm.seriesID
+ prevTS = (*timestamps)[n-1]
+ prevVersion = (*versions)[n-1]
+ prevBlockIdx = blockIdx
+ blockIdx++
+ }
+ }
+ return partRows, boundaries, nil
+}
+
+// AnalyzeGroupDiffWithTarget walks both source roots and the target
+// group root, builds a (sid, ts) → []version multiset for each side,
+// and returns the rows that exist in src but NOT in target — i.e. the
+// exact rows `migration copy` dropped at slow-path bucket-flush time.
+//
+// For each (sid, ts) where len(src.versions) > len(tgt.versions), the
+// missing version values are computed via multiset subtraction (every
+// version in tgt cancels one identical entry in src; anything left in
+// src is reported as missing). MissingRows summed should equal
+// `verify`'s src-tgt row diff for the same (entry, group).
+//
+// Memory: ~2x of AnalyzeGroupRows since both src and tgt maps are
+// held concurrently. Use on small/medium entries.
+func AnalyzeGroupDiffWithTarget(srcRoots []string, targetGroupRoot string, fileSystem fs.FileSystem, sampleCap int) (AnalyzeDiffResult, error) {
+ var res AnalyzeDiffResult
+
+ srcParts, srcPathByID, err := openAllPartsInRoots(srcRoots, fileSystem)
+ if err != nil {
+ return res, fmt.Errorf("open src parts: %w", err)
+ }
+ defer closeAllParts(srcParts)
+ tgtParts, tgtPathByID, err := openAllPartsInRoots([]string{targetGroupRoot}, fileSystem)
+ if err != nil {
+ return res, fmt.Errorf("open tgt parts: %w", err)
+ }
+ defer closeAllParts(tgtParts)
+
+ var srcTotal, tgtTotal int64
+ for _, p := range srcParts {
+ srcTotal += int64(p.partMetadata.TotalCount)
+ }
+ for _, p := range tgtParts {
+ tgtTotal += int64(p.partMetadata.TotalCount)
+ }
+
+ srcStream, srcCleanup := newSidGroupStream(srcParts, srcPathByID)
+ defer srcCleanup()
+ tgtStream, tgtCleanup := newSidGroupStream(tgtParts, tgtPathByID)
+ defer tgtCleanup()
+
+ progressDone := startDiffProgressReporter(&srcStream.rowsRead, &tgtStream.rowsRead, srcTotal, tgtTotal)
+ defer close(progressDone)
+
+ // blockReader emits blocks in (seriesID asc, minTimestamp asc) order
+ // but does NOT row-sort by ts WITHIN a seriesID when that series
+ // spans overlapping ts ranges across parts. So we buffer one whole
+ // seriesID's worth of rows from each side, sort by ts, then
+ // row-merge-diff at ts granularity — same algorithm as before but
+ // guaranteed sorted within each (sid, ts).
+ srcGrp, srcOK := srcStream.nextGroup()
+ tgtGrp, tgtOK := tgtStream.nextGroup()
+ for srcOK || tgtOK {
+ var cmpSid int
+ switch {
+ case !srcOK:
+ cmpSid = +1
+ case !tgtOK:
+ cmpSid = -1
+ case srcGrp.sid < tgtGrp.sid:
+ cmpSid = -1
+ case srcGrp.sid > tgtGrp.sid:
+ cmpSid = +1
+ default:
+ cmpSid = 0
+ }
+ switch {
+ case cmpSid < 0:
+ absorbSrcOnlyGroup(srcGrp, sampleCap, &res)
+ srcGrp, srcOK = srcStream.nextGroup()
+ case cmpSid > 0:
+ res.TargetRows += uint64(len(tgtGrp.rows))
+ tgtGrp, tgtOK = tgtStream.nextGroup()
+ default:
+ diffSidGroup(srcGrp, tgtGrp, sampleCap, &res)
+ srcGrp, srcOK = srcStream.nextGroup()
+ tgtGrp, tgtOK = tgtStream.nextGroup()
+ }
+ }
+ if err := srcStream.err(); err != nil {
+ return res, fmt.Errorf("src block reader: %w", err)
+ }
+ if err := tgtStream.err(); err != nil {
+ return res, fmt.Errorf("tgt block reader: %w", err)
+ }
+ return res, nil
+}
+
+// absorbSrcOnlyGroup records every src row of a series whose sid does
+// not exist on the target side: each (sid, ts) bucket of src becomes
+// a missing entry, with all its versions reported.
+func absorbSrcOnlyGroup(g *sidGroup, sampleCap int, res *AnalyzeDiffResult) {
+ res.SourceRows += uint64(len(g.rows))
+ i := 0
+ for i < len(g.rows) {
+ ts := g.rows[i].ts
+ j := i
+ for j < len(g.rows) && g.rows[j].ts == ts {
+ j++
+ }
+ srcVs := make([]VersionWithPath, 0, j-i)
+ for k := i; k < j; k++ {
+ srcVs = append(srcVs, VersionWithPath{Version: g.rows[k].version, Path: g.rows[k].partPath})
+ }
+ res.MissingKeys++
+ res.MissingRows += uint64(len(srcVs))
+ if sampleCap == 0 || len(res.Missing) < sampleCap {
+ sortVersionsAsc(srcVs)
+ missingCopy := append([]VersionWithPath(nil), srcVs...)
+ res.Missing = append(res.Missing, AnalyzeMissingRow{
+ SeriesID: g.sid,
+ Timestamp: ts,
+ SourceVersions: srcVs,
+ TargetVersions: nil,
+ MissingVersions: missingCopy,
+ })
+ }
+ i = j
+ }
+}
+
+// diffSidGroup runs the ts-by-ts multiset subtraction for one
+// matching (sid). Both groups are already ts-sorted; walk them in
+// merge fashion.
+func diffSidGroup(srcGrp, tgtGrp *sidGroup, sampleCap int, res *AnalyzeDiffResult) {
+ si, ti := 0, 0
+ for si < len(srcGrp.rows) || ti < len(tgtGrp.rows) {
+ var cmpTS int
+ switch {
+ case si >= len(srcGrp.rows):
+ cmpTS = +1
+ case ti >= len(tgtGrp.rows):
+ cmpTS = -1
+ case srcGrp.rows[si].ts < tgtGrp.rows[ti].ts:
+ cmpTS = -1
+ case srcGrp.rows[si].ts > tgtGrp.rows[ti].ts:
+ cmpTS = +1
+ default:
+ cmpTS = 0
+ }
+ switch {
+ case cmpTS < 0:
+ ts := srcGrp.rows[si].ts
+ start := si
+ for si < len(srcGrp.rows) && srcGrp.rows[si].ts == ts {
+ si++
+ }
+ srcVs := make([]VersionWithPath, 0, si-start)
+ for k := start; k < si; k++ {
+ srcVs = append(srcVs, VersionWithPath{Version: srcGrp.rows[k].version, Path: srcGrp.rows[k].partPath})
+ }
+ res.SourceRows += uint64(len(srcVs))
+ res.MissingKeys++
+ res.MissingRows += uint64(len(srcVs))
+ if sampleCap == 0 || len(res.Missing) < sampleCap {
+ sortVersionsAsc(srcVs)
+ missingCopy := append([]VersionWithPath(nil), srcVs...)
+ res.Missing = append(res.Missing, AnalyzeMissingRow{
+ SeriesID: srcGrp.sid,
+ Timestamp: ts,
+ SourceVersions: srcVs,
+ TargetVersions: nil,
+ MissingVersions: missingCopy,
+ })
+ }
+ case cmpTS > 0:
+ ts := tgtGrp.rows[ti].ts
+ for ti < len(tgtGrp.rows) && tgtGrp.rows[ti].ts == ts {
+ res.TargetRows++
+ ti++
+ }
+ default:
+ ts := srcGrp.rows[si].ts
+ startSrc := si
+ for si < len(srcGrp.rows) && srcGrp.rows[si].ts == ts {
+ si++
+ }
+ startTgt := ti
+ for ti < len(tgtGrp.rows) && tgtGrp.rows[ti].ts == ts {
+ ti++
+ }
+ srcVs := make([]VersionWithPath, 0, si-startSrc)
+ for k := startSrc; k < si; k++ {
+ srcVs = append(srcVs, VersionWithPath{Version: srcGrp.rows[k].version, Path: srcGrp.rows[k].partPath})
+ }
+ tgtVs := make([]VersionWithPath, 0, ti-startTgt)
+ for k := startTgt; k < ti; k++ {
+ tgtVs = append(tgtVs, VersionWithPath{Version: tgtGrp.rows[k].version, Path: tgtGrp.rows[k].partPath})
+ }
+ res.SourceRows += uint64(len(srcVs))
+ res.TargetRows += uint64(len(tgtVs))
+ if len(tgtVs) >= len(srcVs) {
+ continue
+ }
+ tgtCount := make(map[int64]int, len(tgtVs))
+ for _, t := range tgtVs {
+ tgtCount[t.Version]++
+ }
+ var missing []VersionWithPath
+ for _, s := range srcVs {
+ if tgtCount[s.Version] > 0 {
+ tgtCount[s.Version]--
+ continue
+ }
+ missing = append(missing, s)
+ }
+ if len(missing) == 0 {
+ continue
+ }
+ res.MissingKeys++
+ res.MissingRows += uint64(len(missing))
+ if sampleCap == 0 || len(res.Missing) < sampleCap {
+ sortVersionsAsc(srcVs)
+ sortVersionsAsc(tgtVs)
+ sortVersionsAsc(missing)
+ res.Missing = append(res.Missing, AnalyzeMissingRow{
+ SeriesID: srcGrp.sid,
+ Timestamp: ts,
+ SourceVersions: srcVs,
+ TargetVersions: tgtVs,
+ MissingVersions: missing,
+ })
+ }
+ }
+ }
+}
+
+// sidGroup is one seriesID's full row set (across all parts), with
+// rows sorted by timestamp ascending. Built by sidGroupStream.
+type sidGroup struct {
+ rows []sidRow
+ sid common.SeriesID
+}
+
+type sidRow struct {
+ partPath string
+ ts int64
+ version int64
+}
+
+func sortVersionsAsc(vs []VersionWithPath) {
+ sort.Slice(vs, func(i, j int) bool { return vs[i].Version < vs[j].Version })
+}
+
+// openAllPartsInRoots enumerates every part dir under the given roots,
+// opens each via mustOpenFilePart, and returns the *part slice + a
+// partID → full path map. Caller must closeAllParts when done.
+//
+// fileSystem is passed through to mustOpenFilePart. Directory walking
+// goes through the os package directly: fs.FileSystem's ReadDir panics
+// on missing directories, while this walker must skip roots that don't
+// exist on every PVC (hash-shard sparsity). The abstraction is
+// intentionally bypassed for the enumeration path.
+func openAllPartsInRoots(roots []string, fileSystem fs.FileSystem) ([]*part, map[uint64]string, error) {
+ var parts []*part
+ pathByID := make(map[uint64]string, 64)
+ for _, root := range roots {
+ segEntries, err := os.ReadDir(root)
+ if err != nil {
+ if os.IsNotExist(err) {
+ continue
+ }
+ closeAllParts(parts)
+ return nil, nil, fmt.Errorf("read root %s: %w", root, err)
+ }
+ for _, se := range segEntries {
+ if !se.IsDir() || !strings.HasPrefix(se.Name(), directCopySegPrefix) {
+ continue
+ }
+ segDir := filepath.Join(root, se.Name())
+ shardEntries, readErr := os.ReadDir(segDir)
+ if readErr != nil {
+ closeAllParts(parts)
+ return nil, nil, fmt.Errorf("read seg %s: %w", segDir, readErr)
+ }
+ for _, sh := range shardEntries {
+ if !sh.IsDir() || !strings.HasPrefix(sh.Name(), directCopyShardPrefix) {
+ continue
+ }
+ shardDir := filepath.Join(segDir, sh.Name())
+ partEntries, partsReadErr := os.ReadDir(shardDir)
+ if partsReadErr != nil {
+ closeAllParts(parts)
+ return nil, nil, fmt.Errorf("read shard %s: %w", shardDir, partsReadErr)
+ }
+ for _, pe := range partEntries {
+ if !pe.IsDir() || !directCopyPartDirPattern.MatchString(pe.Name()) {
+ continue
+ }
+ partID, parseErr := strconv.ParseUint(pe.Name(), 16, 64)
+ if parseErr != nil {
+ closeAllParts(parts)
+ return nil, nil, fmt.Errorf("parse partID %s: %w", pe.Name(), parseErr)
+ }
+ p := mustOpenFilePart(partID, shardDir, fileSystem)
+ parts = append(parts, p)
+ pathByID[p.partMetadata.ID] = filepath.Join(shardDir, pe.Name())
+ }
+ }
+ }
+ }
+ return parts, pathByID, nil
+}
+
+func closeAllParts(parts []*part) {
+ for _, p := range parts {
+ if p != nil {
+ p.close()
+ }
+ }
+}
+
+// sidGroupStream walks partMergeIter+blockReader on one side, but
+// only exposes rows in (sid) groups: each NextGroup call drains every
+// block that shares the current seriesID and returns them sorted by
+// timestamp ascending. That's the level of sortedness diff needs to
+// run multiset subtraction correctly (blockReader's order is
+// (sid asc, block.minTs asc) — overlapping blocks within a sid are
+// NOT row-ts-sorted, which is why the previous row-stream impl
+// double-counted "missing" rows).
+//
+// rowsRead is atomic-summed across every row emitted so the
+// progress reporter goroutine can report scan %.
+type sidGroupStream struct {
+ reader *blockReader
+ decoder *encoding.BytesBlockDecoder
+ pathByPartID map[uint64]string
+ iters []*partMergeIter
+
+ rowsRead int64
+
+ primed bool // first nextBlockMetadata has been called
+ finished bool
+}
+
+func newSidGroupStream(parts []*part, pathByPartID map[uint64]string) (*sidGroupStream, func()) {
+ iters := make([]*partMergeIter, 0, len(parts))
+ for _, p := range parts {
+ pmi := generatePartMergeIter()
+ pmi.mustInitFromPart(p)
+ iters = append(iters, pmi)
+ }
+ decoder := generateColumnValuesDecoder()
+ reader := generateBlockReader()
+ reader.init(iters)
+ s := &sidGroupStream{
+ reader: reader,
+ iters: iters,
+ decoder: decoder,
+ pathByPartID: pathByPartID,
+ }
+ cleanup := func() {
+ releaseBlockReader(reader)
+ releaseColumnValuesDecoder(decoder)
+ for _, it := range iters {
+ releasePartMergeIter(it)
+ }
+ }
+ return s, cleanup
+}
+
+func (s *sidGroupStream) err() error {
+ return s.reader.error()
+}
+
+// nextGroup buffers every block whose seriesID matches the first
+// pending block's sid, sorts the row collection by ts ascending, and
+// returns it. Returns (nil, false) on EOF.
+func (s *sidGroupStream) nextGroup() (*sidGroup, bool) {
+ if s.finished {
+ return nil, false
+ }
+ if !s.primed {
+ if !s.reader.nextBlockMetadata() {
+ s.finished = true
+ return nil, false
+ }
+ s.primed = true
+ }
+ grp := &sidGroup{sid: s.reader.block.bm.seriesID}
+ for {
+ // reader.block is positioned at a block whose sid we may want
+ // to consume. If sid changed, leave it for the next call.
+ if s.reader.block.bm.seriesID != grp.sid {
+ break
+ }
+ s.reader.loadBlockData(s.decoder)
+ path := s.pathByPartID[s.reader.pih[0].partID]
+ n := int(s.reader.block.bm.count)
+ for k := 0; k < n; k++ {
+ grp.rows = append(grp.rows, sidRow{
+ ts: s.reader.block.timestamps[k],
+ version: s.reader.block.versions[k],
+ partPath: path,
+ })
+ }
+ atomic.AddInt64(&s.rowsRead, int64(n))
+ if !s.reader.nextBlockMetadata() {
+ s.finished = true
+ break
+ }
+ }
+ sort.Slice(grp.rows, func(i, j int) bool { return grp.rows[i].ts < grp.rows[j].ts })
+ return grp, true
+}
+
+// startDiffProgressReporter spawns a goroutine that, every 10
+// seconds, prints how many rows each side has scanned + the
+// percentage against the partMetadata-derived total. Returns a
+// channel; close it (or the goroutine returns at total completion)
+// to stop the reporter. Output goes to stderr so it doesn't mix
+// with the structured stdout payload.
+func startDiffProgressReporter(srcCounter, tgtCounter *int64, srcTotal, tgtTotal int64) chan struct{} {
+ done := make(chan struct{})
+ go func() {
+ ticker := time.NewTicker(10 * time.Second)
+ defer ticker.Stop()
+ start := time.Now()
+ for {
+ select {
+ case <-done:
+ return
+ case <-ticker.C:
+ sc := atomic.LoadInt64(srcCounter)
+ tc := atomic.LoadInt64(tgtCounter)
+ srcPct := safePct(sc, srcTotal)
+ tgtPct := safePct(tc, tgtTotal)
+ fmt.Fprintf(os.Stderr,
+ "[diff %5s] src %d/%d (%.1f%%) tgt %d/%d (%.1f%%)\n",
+ time.Since(start).Truncate(time.Second),
+ sc, srcTotal, srcPct, tc, tgtTotal, tgtPct,
+ )
+ }
+ }
+ }()
+ return done
+}
+
+func safePct(part, total int64) float64 {
+ if total <= 0 {
+ return 0
+ }
+ return 100.0 * float64(part) / float64(total)
+}
+
+func safePctU(part, total uint64) float64 {
+ if total == 0 {
+ return 0
+ }
+ return 100.0 * float64(part) / float64(total)
+}
+
+// startScanProgressReporter is the single-side variant of
+// startDiffProgressReporter — used by AnalyzeGroupRows where only
+// the source tree is scanned. Prints to stderr every 10s.
+func startScanProgressReporter(counter *uint64, total uint64) chan struct{} {
+ done := make(chan struct{})
+ go func() {
+ ticker := time.NewTicker(10 * time.Second)
+ defer ticker.Stop()
+ start := time.Now()
+ for {
+ select {
+ case <-done:
+ return
+ case <-ticker.C:
+ c := atomic.LoadUint64(counter)
+ pct := safePctU(c, total)
+ fmt.Fprintf(os.Stderr,
+ "[scan %5s] %d/%d rows (%.1f%%)\n",
+ time.Since(start).Truncate(time.Second),
+ c, total, pct,
+ )
+ }
+ }
+ }()
+ return done
+}
+
+// sumGroupExpectedRows walks every <root>/seg-*/shard-*/<partID>/
+// directory under roots and sums partMetadata.TotalCount — used by
+// AnalyzeGroupRows to size the progress-reporter's % calculation
+// without holding all parts open. Each part metadata read is a small
+// file read, so this pre-pass is cheap (~ms per part).
+func sumGroupExpectedRows(roots []string, fileSystem fs.FileSystem) (int64, error) {
+ var total int64
+ for _, root := range roots {
+ segEntries, err := os.ReadDir(root)
+ if err != nil {
+ if os.IsNotExist(err) {
+ continue
+ }
+ return 0, fmt.Errorf("read root %s: %w", root, err)
+ }
+ for _, se := range segEntries {
+ if !se.IsDir() || !strings.HasPrefix(se.Name(), directCopySegPrefix) {
+ continue
+ }
+ segDir := filepath.Join(root, se.Name())
+ shardEntries, readErr := os.ReadDir(segDir)
+ if readErr != nil {
+ return 0, fmt.Errorf("read seg %s: %w", segDir, readErr)
+ }
+ for _, sh := range shardEntries {
+ if !sh.IsDir() || !strings.HasPrefix(sh.Name(), directCopyShardPrefix) {
+ continue
+ }
+ shardDir := filepath.Join(segDir, sh.Name())
+ partEntries, partsReadErr := os.ReadDir(shardDir)
+ if partsReadErr != nil {
+ return 0, fmt.Errorf("read shard %s: %w", shardDir, partsReadErr)
+ }
+ for _, pe := range partEntries {
+ if !pe.IsDir() || !directCopyPartDirPattern.MatchString(pe.Name()) {
+ continue
+ }
+ partDir := filepath.Join(shardDir, pe.Name())
+ var pm partMetadata
+ pm.mustReadMetadata(fileSystem, partDir)
+ total += int64(pm.TotalCount)
+ }
+ }
+ }
+ }
+ return total, nil
+}
diff --git a/banyand/measure/migration_copy.go b/banyand/measure/migration_copy.go
new file mode 100644
index 000000000..f8818611e
--- /dev/null
+++ b/banyand/measure/migration_copy.go
@@ -0,0 +1,1816 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package measure
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "io"
+ "os"
+ "path/filepath"
+ "regexp"
+ "runtime"
+ "runtime/debug"
+ "sort"
+ "strconv"
+ "strings"
+ "sync"
+ "sync/atomic"
+ "time"
+
+ "github.com/blugelabs/bluge"
+ "github.com/blugelabs/bluge/search"
+
+ "github.com/apache/skywalking-banyandb/api/common"
+ commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
+ "github.com/apache/skywalking-banyandb/banyand/internal/storage"
+ "github.com/apache/skywalking-banyandb/pkg/bytes"
+ "github.com/apache/skywalking-banyandb/pkg/compress/zstd"
+ "github.com/apache/skywalking-banyandb/pkg/encoding"
+ "github.com/apache/skywalking-banyandb/pkg/fs"
+ pbv1 "github.com/apache/skywalking-banyandb/pkg/pb/v1"
+ "github.com/apache/skywalking-banyandb/pkg/query/model"
+)
+
+// directCopyDayFormat / directCopyHourFormat mirror segmentController.format
+// in banyand/internal/storage/segment.go.
+const (
+ directCopyDayFormat = "20060102"
+ directCopyHourFormat = "2006010215"
+ directCopySegPrefix = "seg-"
+ directCopyShardPrefix = "shard-"
+ directCopySidxDirName = "sidx"
+ directCopySnpSuffix = ".snp"
+)
+
+var directCopyPartDirPattern = regexp.MustCompile(`^[0-9a-f]{16}$`)
+
+// nofsyncFS wraps a local FileSystem and skips per-file fsync /
+// directory sync calls. Migration is a one-shot bulk import: a crash
+// mid-run leaves the target measure-root in an unusable state anyway
+// and the operator re-runs with --clean. Per-part fsyncs were the
+// dominant wall-time cost (60%+ of CPU samples in syscall.syscall on
+// open/write/fsync) precisely because each chunked target part fans
+// out 7-10 small files. Skipping fsync trades zero-crash-safety for
+// throughput, which is the right trade-off here.
+type nofsyncFS struct {
+ fs.FileSystem
+}
+
+func (f nofsyncFS) Write(buffer []byte, name string, permission fs.Mode) (int, error) {
+ file, err := os.OpenFile(name, os.O_RDWR|os.O_CREATE|os.O_TRUNC, os.FileMode(permission))
+ if err != nil {
+ return 0, fmt.Errorf("open %s: %w", name, err)
+ }
+ n, err := file.Write(buffer)
+ if cerr := file.Close(); cerr != nil && err == nil {
+ err = cerr
+ }
+ if err != nil {
+ return 0, err
+ }
+ return n, nil
+}
+
+func (f nofsyncFS) WriteAtomic(buffer []byte, name string, permission fs.Mode) (int, error) {
+ return f.Write(buffer, name, permission)
+}
+
+func (nofsyncFS) SyncPath(string) {}
+
+// FastPathHits / SlowPathHits / SlowPathRows expose copy-path
+// instrumentation to the CLI so it can print a quick fast-vs-slow
+// breakdown after each RunDirectCopy.
+func FastPathHits() int64 { return fastPathHits.Load() }
+
+// SlowPathHits returns the number of source parts that went through the
+// row-level rewrite path.
+func SlowPathHits() int64 { return slowPathHits.Load() }
+
+// SlowPathRows returns the total source-row count across all parts that
+// were forced onto the slow path.
+func SlowPathRows() int64 { return slowPathRows.Load() }
+
+// DirectCopyConfig configures a one-shot direct-file copy migration of
+// measure data from a backup snapshot into one or more local target
+// roots. The caller (e.g. banyand/cmd/migration) is expected to have
+// stopped the BanyanDB process writing to any of the Entries' Target
+// directories before invoking MigrationCopy.
+//
+// Both measure schemas (tag families + IndexMode bit) and group resource
+// options (SegmentInterval per LifecycleStage) are read straight from
+// the backup's schema-property bluge catalog — no liaison access needed.
+//
+// Each Entry names one (Stage, Target) destination; the same source
+// rows are aligned by that stage's SegmentInterval and written under
+// Target. Schemas, ResourceOpts and per-group union sidx are loaded /
+// built once and shared across every entry.
+type DirectCopyConfig struct {
+ BackupDir string
+ Date string
+
+ // SchemaPropertyPath, when set, points the schema loader at a
+ // schema-property catalog directory directly (e.g. a live cluster's
+ // schema-property PVC mounted into the runner pod). When empty, the
+ // loader falls back to discovering the catalog under BackupDir using
+ // the backup snapshot's standard layout.
+ SchemaPropertyPath string
+
+ // SidxStagingDir is where per-group union sidx artifacts are built
+ // before being broadcast into every aligned target segment. Required:
+ // the caller owns the directory's lifecycle (creation and cleanup).
+ SidxStagingDir string
+
+ Groups []string
+ Entries []DirectCopyEntry
+}
+
+// DirectCopyEntry names one fan-out destination. Each entry is
+// processed in turn; for every group, the tool reads source parts from
+// either the explicit Source paths (preferred) or, when Source is
+// empty, derives the per-node source roots from BackupDir + Date + Nodes
+// using the backup snapshot's standard layout. Rows are aligned by
+// Stage's SegmentInterval and written under the target group root.
+//
+// Each Source entry must be a measure-data root — a directory whose
+// children are the per-group segment trees that BanyanDB writes during
+// normal operation. All Source entries are traversed and merged when
+// building the per-group union sidx.
+type DirectCopyEntry struct {
+ Stage string
+ Target string
+ Source []string
+ Nodes []string
+}
+
+// DirectCopyResult summarizes one MigrationCopy invocation, with totals
+// aggregated across every (entry, group) the run touched.
+type DirectCopyResult struct {
+ Duration time.Duration
+ Rows int64
+ Bytes int64
+ SourceParts int
+ TargetParts int
+ Segments int
+}
+
+// MigrationCopy is the package-public entry point for the migration's
+// copy strategy. See banyand/cmd/migration/copy.go for the CLI driver.
+//
+// Flow:
+// 1. Load schemas + group resource opts directly from the backup's
+// schema-property bluge catalog (once for the whole run).
+// 2. Reject up front if any requested group contains an IndexMode
+// measure — broadcasting a union sidx would let queries fan their
+// full result set across data nodes and corrupt cross-node dedup.
+// 3. Resolve every group's source directories once; build one union
+// sidx per group under SidxStagingDir — the union is source-only
+// and gets reused across every entry.
+// 4. For each (entry, group): align rows by entry.Stage's
+// SegmentInterval and write parts under the entry's target group
+// root. After all parts of one (entry, group) land, broadcast that
+// group's union sidx into every aligned target segment.
+//
+// Entries run sequentially; the first error aborts the run.
+func MigrationCopy(ctx context.Context, cfg DirectCopyConfig) (DirectCopyResult, error) {
+ var res DirectCopyResult
+ if err := validateDirectCopyConfig(&cfg); err != nil {
+ return res, err
+ }
+ if cfg.SidxStagingDir == "" {
+ return res, errors.New("sidxStagingDir is required (caller owns its lifecycle)")
+ }
+ // Reset path-hit counters so a previous in-process run's totals don't
+ // bleed into this one (CLI invokes MigrationCopy once, but tests and
+ // future callers may invoke it multiple times within one binary).
+ fastPathHits.Store(0)
+ slowPathHits.Store(0)
+ slowPathRows.Store(0)
+ start := time.Now()
+
+ logStep("loading measure schemas")
+ //nolint:contextcheck // bluge reader.Search inside walkSchemaPropertyShard already uses its own context.
+ schemas, err := loadMeasureSchemasFromSchemaCatalog(cfg.BackupDir, cfg.Date, cfg.SchemaPropertyPath, cfg.Groups)
+ if err != nil {
+ return res, fmt.Errorf("load measure schemas: %w", err)
+ }
+ // Refuse to proceed when a requested group resolved to zero
+ // measures: the slow path would otherwise silently write target
+ // parts with an empty tag projection (no tag columns), which is
+ // indistinguishable from "no tags" at read time and corrupts the
+ // query result. Typical cause: typo in --groups or stale backup.
+ for _, g := range cfg.Groups {
+ if len(schemas[g]) == 0 {
+ return res, fmt.Errorf("group %q has no measures in backup schema-property catalog — typo in groups or empty group?", g)
+ }
+ }
+ if rejectErr := rejectIndexModeGroups(cfg.Groups, schemas); rejectErr != nil {
+ return res, rejectErr
+ }
+
+ //nolint:contextcheck // bluge reader.Search inside walkSchemaPropertyShard already uses its own context.
+ resourceOpts, err := loadGroupResourceOptsFromSchema(cfg.BackupDir, cfg.Date, cfg.SchemaPropertyPath, cfg.Groups)
+ if err != nil {
+ return res, fmt.Errorf("load group resource opts: %w", err)
+ }
+
+ // Pre-flight: every (entry, group) must resolve to a SegmentInterval
+ // up front. Failing here saves an expensive union-sidx build only to
+ // abort mid-run on a typo (warn vs warm has been the historic gotcha).
+ for entryIdx, entry := range cfg.Entries {
+ for _, group := range cfg.Groups {
+ opts, ok := resourceOpts[group]
+ if !ok || opts == nil {
+ return res, fmt.Errorf("group %s: ResourceOpts not found in backup schema-property catalog", group)
+ }
+ if _, err := resolveStageInterval(opts, entry.Stage); err != nil {
+ return res, fmt.Errorf("entry %d (target=%s) group %s: %w",
+ entryIdx, entry.Target, group, err)
+ }
+ }
+ }
+
+ // Phase A: build one union sidx per group, sourced from EVERY entry's
+ // roots merged (dedup by path). Each (entry, group) in Phase B reuses
+ // the same staged sidx so two invariants hold:
+ // - Cross-replica broadcast invariant: the same series-ID set ends
+ // up under every target seg's sidx subdir, which is what liaison
+ // relies on to dedup distributed query fan-outs.
+ // - We don't rescan the same source bluge index N times per group
+ // (was 5 entries × 3 groups = 15 builds on a 5-PVC GKE plan).
+ groupUnionSidx := make(map[string]string, len(cfg.Groups))
+ for _, group := range cfg.Groups {
+ if ctx.Err() != nil {
+ res.Duration = time.Since(start)
+ return res, ctx.Err()
+ }
+ srcRoots := collectAllSrcGroupRoots(cfg, group)
+ if len(srcRoots) == 0 {
+ logStep("group %s: no source dirs across any entry — skipping union sidx build", group)
+ continue
+ }
+ logStep("group %s: building union sidx from %d source dir(s) merged across all entries",
+ group, len(srcRoots))
+ unionSidxPath, err := buildGroupUnionSidx(ctx, srcRoots,
+ filepath.Join(cfg.SidxStagingDir, "groups", group, directCopySidxDirName))
+ if err != nil {
+ res.Duration = time.Since(start)
+ return res, fmt.Errorf("group %s: build union sidx: %w", group, err)
+ }
+ groupUnionSidx[group] = unionSidxPath
+ }
+
+ // Phase B: per entry × group, align source rows by stage's
+ // SegmentInterval and write parts under the entry's target group
+ // root, broadcasting the pre-built union sidx (from Phase A) into
+ // every aligned segment.
+ for entryIdx, entry := range cfg.Entries {
+ if ctx.Err() != nil {
+ res.Duration = time.Since(start)
+ return res, ctx.Err()
+ }
+ entryTag := fmt.Sprintf("entry [%d/%d]", entryIdx+1, len(cfg.Entries))
+ logStep("%s stage=%s nodes=%v target=%s",
+ entryTag, entry.Stage, entry.Nodes, entry.Target)
+ for _, group := range cfg.Groups {
+ if ctx.Err() != nil {
+ res.Duration = time.Since(start)
+ return res, ctx.Err()
+ }
+ srcRoots := resolveEntrySrcRoots(cfg, entry, group)
+ if len(srcRoots) == 0 {
+ if len(entry.Source) > 0 {
+ logStep("%s stage=%s group %s: skipped (none of source paths %v contain this group)",
+ entryTag, entry.Stage, group, entry.Source)
+ } else {
+ logStep("%s stage=%s group %s: skipped (none of nodes %v carry this group)",
+ entryTag, entry.Stage, group, entry.Nodes)
+ }
+ continue
+ }
+ tasks, err := discoverPartTasks(ctx, srcRoots)
+ if err != nil {
+ res.Duration = time.Since(start)
+ return res, fmt.Errorf("%s stage=%s group %s: discover parts: %w",
+ entryTag, entry.Stage, group, err)
+ }
+ if len(tasks) == 0 {
+ logStep("%s stage=%s group %s: skipped (no parts in selected nodes)",
+ entryTag, entry.Stage, group)
+ continue
+ }
+ opts, ok := resourceOpts[group]
+ if !ok || opts == nil {
+ return res, fmt.Errorf("group %s: ResourceOpts not found in backup schema-property catalog", group)
+ }
+ ir, err := resolveStageInterval(opts, entry.Stage)
+ if err != nil {
+ return res, fmt.Errorf("%s stage=%s group %s: %w", entryTag, entry.Stage, group, err)
+ }
+ unionSidxPath := groupUnionSidx[group]
+ targetGroupRoot := filepath.Join(entry.Target, group)
+ logStep("%s stage=%s group %s: %d source dir(s), %d parts, interval=%v×%d — writing target (target=%q, union sidx=%q)",
+ entryTag, entry.Stage, group, len(srcRoots), len(tasks), ir.Unit, ir.Num,
+ targetGroupRoot, unionSidxPath)
+ groupRes, err := directCopyGroup(ctx,
+ entryTag,
+ group, entry,
+ targetGroupRoot,
+ ir, schemas[group],
+ tasks, unionSidxPath)
+ if err != nil {
+ res.Duration = time.Since(start)
+ return res, fmt.Errorf("%s stage=%s group %s: %w", entryTag, entry.Stage, group, err)
+ }
+ logStep("%s stage=%s group %s: done rows=%d segs=%d srcParts=%d targetParts=%d",
+ entryTag, entry.Stage, group,
+ groupRes.Rows, groupRes.Segments, groupRes.SourceParts, groupRes.TargetParts)
+ // Return idle heap to the OS so compressed-memory pressure
+ // doesn't accumulate across (entry, group) — macOS jetsam
+ // killed earlier runs at 224 GB compressed.
+ debug.FreeOSMemory()
+ res.SourceParts += groupRes.SourceParts
+ res.TargetParts += groupRes.TargetParts
+ res.Rows += groupRes.Rows
+ res.Bytes += groupRes.Bytes
+ res.Segments += groupRes.Segments
+ }
+ }
+ res.Duration = time.Since(start)
+ return res, nil
+}
+
+func logStep(format string, args ...any) {
+ fmt.Fprintf(os.Stdout, time.Now().Format("2006/01/02 15:04:05")+" [migration] "+format+"\n", args...)
+}
+
+// StageHot is the conventional sentinel for "use the group's default
+// (top-level) SegmentInterval". BanyanDB models the hot tier as the
+// implicit destination of new writes, so groups rarely list "hot" as a
+// named stage; users still need a way to migrate INTO the hot tier, and
+// this constant is that way.
+const StageHot = "hot"
+
+// resolveStageInterval picks the SegmentInterval that should drive the
+// target grid for one (group, stage) pair. stage must either be the
+// reserved "hot" sentinel (which selects opts.SegmentInterval) or
+// exactly match one of opts.GetStages() entries — silent fallback to
+// default on an unknown stage has bitten operators in the past
+// (warn-vs-warm typo wrote multi-day source rows into a 1-day grid).
+func resolveStageInterval(opts *commonv1.ResourceOpts, stage string) (storage.IntervalRule, error) {
+ var zero storage.IntervalRule
+ if stage == StageHot {
+ if opts.GetSegmentInterval() == nil {
+ return zero, fmt.Errorf("stage %q maps to default SegmentInterval but the group has none", stage)
+ }
+ return intervalRuleFromProto(opts.GetSegmentInterval())
+ }
+ for _, st := range opts.GetStages() {
+ if st.GetName() == stage {
+ if st.GetSegmentInterval() == nil {
+ return zero, fmt.Errorf("stage %q has no segmentInterval", stage)
+ }
+ return intervalRuleFromProto(st.GetSegmentInterval())
+ }
+ }
+ available := make([]string, 0, len(opts.GetStages())+1)
+ available = append(available, StageHot+" (default)")
+ for _, st := range opts.GetStages() {
+ available = append(available, st.GetName())
+ }
+ return zero, fmt.Errorf("stage %q not found in group schema; available: %s",
+ stage, strings.Join(available, ", "))
+}
+
+func intervalRuleFromProto(ir *commonv1.IntervalRule) (storage.IntervalRule, error) {
+ var zero storage.IntervalRule
+ if ir.GetNum() <= 0 {
+ return zero, fmt.Errorf("segmentInterval.num must be > 0, got %d", ir.GetNum())
+ }
+ switch ir.GetUnit() {
+ case commonv1.IntervalRule_UNIT_DAY:
+ return storage.IntervalRule{Unit: storage.DAY, Num: int(ir.GetNum())}, nil
+ case commonv1.IntervalRule_UNIT_HOUR:
+ return storage.IntervalRule{Unit: storage.HOUR, Num: int(ir.GetNum())}, nil
+ }
+ return zero, fmt.Errorf("unsupported segmentInterval.unit %v", ir.GetUnit())
+}
+
+// resolveEntrySrcRoots picks the source group dirs for one (entry,
+// group) pair. When entry.Source is set, treat each entry as a
+// measure-data root and look for "<source>/<group>"; otherwise fall
+// back to the backup-snapshot layout via Nodes. Either path can yield
+// zero dirs — the caller logs and skips.
+func resolveEntrySrcRoots(cfg DirectCopyConfig, entry DirectCopyEntry, group string) []string {
+ if len(entry.Source) > 0 {
+ var roots []string
+ for _, p := range entry.Source {
+ candidate := filepath.Join(p, group)
+ if info, err := os.Stat(candidate); err == nil && info.IsDir() {
+ roots = append(roots, candidate)
+ }
+ }
+ sort.Strings(roots)
+ return roots
+ }
+ return collectEntrySrcGroupRoots(cfg.BackupDir, cfg.Date, group, entry.Nodes)
+}
+
+// collectEntrySrcGroupRoots returns the `<backupDir>/<node>/<date>/measure/<group>`
+// directories that exist for the explicit set of nodes the entry asked
+// for. Nodes that don't carry the group (or don't exist at all) are
+// silently dropped — that's a legitimate "this stage's machines don't
+// own this group" case. Returned paths are sorted.
+func collectEntrySrcGroupRoots(backupDir, date, group string, nodes []string) []string {
+ var roots []string
+ for _, node := range nodes {
+ if node == "" {
+ continue
+ }
+ candidate := filepath.Join(backupDir, node, date, "measure", group)
+ if info, err := os.Stat(candidate); err == nil && info.IsDir() {
+ roots = append(roots, candidate)
+ }
+ }
+ sort.Strings(roots)
+ return roots
+}
+
+// collectAllSrcGroupRoots merges resolveEntrySrcRoots across every
+// entry in cfg (de-duped) so Phase A can build one union sidx per group
+// that covers the whole plan, not just one entry's node slice. The
+// returned list is sorted for determinism.
+func collectAllSrcGroupRoots(cfg DirectCopyConfig, group string) []string {
+ seen := make(map[string]struct{})
+ var roots []string
+ for _, entry := range cfg.Entries {
+ for _, r := range resolveEntrySrcRoots(cfg, entry, group) {
+ if _, dup := seen[r]; dup {
+ continue
+ }
+ seen[r] = struct{}{}
+ roots = append(roots, r)
+ }
+ }
+ sort.Strings(roots)
+ return roots
+}
+
+// rejectIndexModeGroups errors out if any requested group contains at
+// least one IndexMode measure. The union-sidx broadcast strategy is
+// only safe for non-IndexMode measures (whose query result rows come
+// from part data, not from the sidx itself); silently letting an
+// IndexMode group through would corrupt cross-node dedup at the
+// liaison.
+func rejectIndexModeGroups(groups []string, schemas map[string]map[string]*measureSchemaInfo) error {
+ var offenders []string
+ for _, g := range groups {
+ for _, m := range schemas[g] {
+ if m.IndexMode {
+ offenders = append(offenders, fmt.Sprintf("%s/%s", g, m.Name))
+ }
+ }
+ }
+ if len(offenders) == 0 {
+ return nil
+ }
+ sort.Strings(offenders)
+ return fmt.Errorf("refusing to copy IndexMode measures (their data lives inside sidx; "+
+ "the union-sidx broadcast strategy would break query correctness): %s",
+ strings.Join(offenders, ", "))
+}
+
+func validateDirectCopyConfig(cfg *DirectCopyConfig) error {
+ if cfg.SchemaPropertyPath == "" {
+ if cfg.BackupDir == "" {
+ return errors.New("backupDir is required when SchemaPropertyPath is unset")
+ }
+ if cfg.Date == "" {
+ return errors.New("date is required when SchemaPropertyPath is unset")
+ }
+ }
+ // SidxStagingDir is checked in MigrationCopy specifically — verify
+ // is read-only and does not stage anything.
+ if len(cfg.Groups) == 0 {
+ return errors.New("groups is required")
+ }
+ if len(cfg.Entries) == 0 {
+ return errors.New("entries is required")
+ }
+ seenTarget := make(map[string]int, len(cfg.Entries))
+ for i, e := range cfg.Entries {
+ if e.Stage == "" {
+ return fmt.Errorf("entries[%d].Stage is required", i)
+ }
+ if e.Target == "" {
+ return fmt.Errorf("entries[%d].Target is required", i)
+ }
+ if len(e.Source) == 0 && len(e.Nodes) == 0 {
+ return fmt.Errorf("entries[%d]: at least one of Source or Nodes must be set", i)
+ }
+ if prev, dup := seenTarget[e.Target]; dup {
+ return fmt.Errorf("entries[%d].Target %q duplicates Entries[%d].Target", i, e.Target, prev)
+ }
+ seenTarget[e.Target] = i
+ }
+ return nil
+}
+
+// fastPathHits / slowPathHits / slowPathRows track how often each
+// processOneSourcePart branch fires across one RunDirectCopy. Surfaced
+// in the final summary so operators can see whether their backup is
+// already grid-aligned (fast path dominates) or needs row-level
+// rewrite (slow path dominates).
+var (
+ fastPathHits atomic.Int64
+ slowPathHits atomic.Int64
+ slowPathRows atomic.Int64
+)
+
+type partTask struct {
+ srcSegName string
+ shardName string
+ shardDir string
+ partIDStr string
+}
+
+// discoverPartTasks walks every <srcGroupRoot>/seg-*/shard-*/<partID>
+// directory across all supplied roots and produces a flat task list
+// for the worker pool to consume without further IO contention.
+func discoverPartTasks(ctx context.Context, srcGroupRoots []string) ([]partTask, error) {
+ var tasks []partTask
+ for _, srcGroupRoot := range srcGroupRoots {
+ if ctx.Err() != nil {
+ return nil, ctx.Err()
+ }
+ segEntries, err := os.ReadDir(srcGroupRoot)
+ if err != nil {
+ if os.IsNotExist(err) {
+ continue
+ }
+ return nil, fmt.Errorf("read %s: %w", srcGroupRoot, err)
+ }
+ for _, segE := range segEntries {
+ if !segE.IsDir() || !strings.HasPrefix(segE.Name(), directCopySegPrefix) {
+ continue
+ }
+ srcSegName := segE.Name()
+ srcSegDir := filepath.Join(srcGroupRoot, srcSegName)
+ shardEntries, err := os.ReadDir(srcSegDir)
+ if err != nil {
+ return nil, fmt.Errorf("read %s: %w", srcSegDir, err)
+ }
+ for _, shardE := range shardEntries {
+ if !shardE.IsDir() || !strings.HasPrefix(shardE.Name(), directCopyShardPrefix) {
+ continue
+ }
+ shardName := shardE.Name()
+ shardDir := filepath.Join(srcSegDir, shardName)
+ partEntries, err := os.ReadDir(shardDir)
+ if err != nil {
+ return nil, fmt.Errorf("read %s: %w", shardDir, err)
+ }
+ for _, partE := range partEntries {
+ if !partE.IsDir() || !directCopyPartDirPattern.MatchString(partE.Name()) {
+ continue
+ }
+ tasks = append(tasks, partTask{
+ srcSegName: srcSegName,
+ shardName: shardName,
+ shardDir: shardDir,
+ partIDStr: partE.Name(),
+ })
+ }
+ }
+ }
+ }
+ return tasks, nil
+}
+
+// segStateForGroup tracks bookkeeping needed at end-of-group to write
+// segment-level metadata.json and per-shard .snp files.
+type segStateForGroup struct {
+ alignedTime time.Time
+ shards map[string][]uint64 // shard-N -> partIDs landed here
+}
+
+// segStateRegistry is a goroutine-safe registry of segStateForGroup,
+// used so the per-part workers can register their landed (alignedSeg,
+// shard, partID) tuples concurrently without stepping on each other.
+type segStateRegistry struct {
+ m map[string]*segStateForGroup
+ mu sync.Mutex
+}
+
+func newSegStateRegistry() *segStateRegistry {
+ return &segStateRegistry{m: map[string]*segStateForGroup{}}
+}
+
+// register attaches one target partID to (alignedSegName, shardName) and
+// guarantees a segStateForGroup exists for alignedSegName with the
+// supplied alignedTime. Concurrent callers serialize on the registry's
+// mutex; the critical section is tiny (slice append + map lookup) so
+// contention stays well below worker compute cost.
+func (r *segStateRegistry) register(alignedSegName string, alignedTime time.Time, shardName string, partID uint64) {
+ r.mu.Lock()
+ defer r.mu.Unlock()
+ ss, ok := r.m[alignedSegName]
+ if !ok {
+ ss = &segStateForGroup{
+ alignedTime: alignedTime,
+ shards: map[string][]uint64{},
+ }
+ r.m[alignedSegName] = ss
+ }
+ ss.shards[shardName] = append(ss.shards[shardName], partID)
+}
+
+// snapshot returns the underlying map for serial end-of-group
+// processing (metadata.json + .snp writes). Callers may only iterate
+// it after every worker has finished.
+func (r *segStateRegistry) snapshot() map[string]*segStateForGroup {
+ return r.m
+}
+
+// directCopyGroup walks every source part for one group, splits each
+// source part's rows by aligned target segment, flushes one memPart per
+// (alignedSeg, sourcePart) bucket, then writes per-segment metadata and
+// per-shard .snp at the end. The group's union sidx (pre-built in
+// staging) is broadcast 1:1 into every aligned target segment that
+// received any rows; an empty unionSidxPath skips the sidx broadcast.
+func directCopyGroup(ctx context.Context,
+ entryTag string,
+ group string, entry DirectCopyEntry, dstGroupRoot string,
+ ir storage.IntervalRule, groupSchemas map[string]*measureSchemaInfo,
+ tasks []partTask, unionSidxPath string,
+) (DirectCopyResult, error) {
+ var res DirectCopyResult
+
+ if err := directCopyPrepareTarget(dstGroupRoot); err != nil {
+ return res, err
+ }
+
+ tagProjection := buildTagProjectionFromGroupSchemas(groupSchemas)
+
+ segStates := newSegStateRegistry()
+ // partIDGen issues monotonic partIDs (1, 2, 3, ...) shared across
+ // every srcGroupRoot and every worker so concurrent writes never
+ // collide. Starting at 0 is safe because directCopyPrepareTarget
+ // asserts the target group dir is empty and the migration runs
+ // while banyandb is not writing to it; on restart banyandb seeds
+ // its own counter past the highest partID we wrote.
+ var partIDGen atomic.Uint64
+ fileSystem := nofsyncFS{FileSystem: fs.NewLocalFileSystem()}
+
+ totalTasks := len(tasks)
+
+ // Worker pool: N = NumCPU, each worker owns a fresh BytesBlockDecoder
+ // (the decoder caches state across reads and is not safe for concurrent
+ // use). Workers fan out over the task list; the segStateRegistry,
+ // partIDGen and result counters are the only shared mutable state.
+ workerCount := runtime.NumCPU()
+ if workerCount > totalTasks {
+ workerCount = totalTasks
+ }
+ if workerCount < 1 {
+ workerCount = 1
+ }
+
+ // Shared bounded flush pool: a fixed set of consumers drain
+ // flushCh and do the heavy encode + zstd + write per bucket. Outer
+ // workers (per source part) post one job per bucket per chunk and
+ // wait on a per-chunk WaitGroup; this caps simultaneous encode+zstd
+ // at flushWorkerCount rather than `outerWorkers × bucketsPerChunk`
+ // (which had no bound and caused OOM on cold-tier fan-out workloads).
+ flushWorkerCount := workerCount
+ flushCh := make(chan flushJob, flushWorkerCount*2)
+ var flushWg sync.WaitGroup
+ for i := 0; i < flushWorkerCount; i++ {
+ flushWg.Add(1)
+ go func() {
+ defer flushWg.Done()
+ runFlushWorker(flushCh, fileSystem)
+ }()
+ }
+ defer func() {
+ close(flushCh)
+ flushWg.Wait()
+ }()
+
+ // Buffered so producers don't rendezvous with each consumer pull.
+ taskCh := make(chan partTask, workerCount*2)
+ var (
+ wg sync.WaitGroup
+ resMu sync.Mutex
+ sourceParts int
+ targetParts int
+ rowsCopied int64
+ bytesWritten int64
+ firstErr error
+ errMu sync.Mutex
+ partsCompleted atomic.Int64
+ )
+
+ workerCtx, cancelWorkers := context.WithCancel(ctx)
+ defer cancelWorkers()
+
+ for i := 0; i < workerCount; i++ {
+ wg.Add(1)
+ go func() {
+ defer wg.Done()
+ decoder := &encoding.BytesBlockDecoder{}
+ for task := range taskCh {
+ if workerCtx.Err() != nil {
+ return
+ }
+ partResult, err := func() (pr processPartResult, err error) {
+ // processOneSourcePart's mustReadMetadata / mustOpenFilePart /
+ // b.mustReadFrom panic on corrupted parts. Convert panics into
+ // firstErr so one bad source part doesn't tear down the whole
+ // migration before the deferred staging-dir cleanup runs.
+ defer func() {
+ if r := recover(); r != nil {
+ err = fmt.Errorf("panic processing %s/%s/%s: %v",
+ task.srcSegName, task.shardName, task.partIDStr, r)
+ }
+ }()
+ return processOneSourcePart(
+ processPartInput{
+ ir: ir,
+ decoder: decoder,
+ fileSystem: fileSystem,
+ tagProjection: tagProjection,
+ shardName: task.shardName,
+ shardDir: task.shardDir,
+ partIDStr: task.partIDStr,
+ dstGroupRoot: dstGroupRoot,
+ segStates: segStates,
+ partIDGen: &partIDGen,
+ flushCh: flushCh,
+ },
+ )
+ }()
+ if err != nil {
+ errMu.Lock()
+ if firstErr == nil {
+ firstErr = err
+ cancelWorkers()
+ }
+ errMu.Unlock()
+ return
+ }
+ done := partsCompleted.Add(1)
+ logStep("%s stage=%s group %s: part %d/%d (%.1f%%) done %s/%s/%s rows=%d targetParts=%d",
+ entryTag, entry.Stage, group, done, totalTasks,
+ 100.0*float64(done)/float64(totalTasks),
+ task.srcSegName, task.shardName, task.partIDStr,
+ partResult.rows, partResult.targetParts)
+ resMu.Lock()
+ sourceParts++
+ targetParts += partResult.targetParts
+ rowsCopied += partResult.rows
+ bytesWritten += partResult.bytes
+ resMu.Unlock()
+ }
+ }()
+ }
+
+ for _, t := range tasks {
+ select {
+ case taskCh <- t:
+ case <-workerCtx.Done():
+ }
+ }
+ close(taskCh)
+ wg.Wait()
+ if firstErr != nil {
+ return res, firstErr
+ }
+ res.SourceParts = sourceParts
+ res.TargetParts = targetParts
+ res.Rows = rowsCopied
+ res.Bytes = bytesWritten
+
+ unionSidxAvailable := false
+ if unionSidxPath != "" {
+ if info, statErr := os.Stat(unionSidxPath); statErr == nil && info.IsDir() {
+ unionSidxAvailable = true
+ }
+ }
+ segSnapshot := segStates.snapshot()
+ logStep("%s stage=%s group %s: finalizing %d aligned target segments (writing metadata + snp + union sidx)",
+ entryTag, entry.Stage, group, len(segSnapshot))
+ for alignedSeg, ss := range segSnapshot {
+ res.Segments++
+ segDir := filepath.Join(dstGroupRoot, alignedSeg)
+ endTime := ir.NextTime(ss.alignedTime)
+ metaBytes, err := writeDirectCopySegmentMetadata(segDir, endTime)
+ if err != nil {
+ return res, fmt.Errorf("seg %s metadata: %w", alignedSeg, err)
+ }
+ res.Bytes += metaBytes
+ for shardName, partIDs := range ss.shards {
+ sort.Slice(partIDs, func(i, j int) bool { return partIDs[i] < partIDs[j] })
+ names := make([]string, len(partIDs))
+ for i, pid := range partIDs {
+ names[i] = fmt.Sprintf("%016x", pid)
+ }
+ snpBytes, err := writeDirectCopySnp(
+ filepath.Join(segDir, shardName), names,
+ )
+ if err != nil {
+ return res, fmt.Errorf("seg %s shard %s snp: %w", alignedSeg, shardName, err)
+ }
+ res.Bytes += snpBytes
+ }
+ if unionSidxAvailable {
+ sidxBytes, err := directCopyDir(
+ unionSidxPath,
+ filepath.Join(segDir, directCopySidxDirName),
+ )
+ if err != nil {
+ return res, fmt.Errorf("seg %s union sidx: %w", alignedSeg, err)
+ }
+ res.Bytes += sidxBytes
+ }
+ }
+
+ return res, nil
+}
+
+// buildTagProjectionFromGroupSchemas merges every measure's tag families
+// in the group into a single tagProjection set. Reading more tags than
+// the part actually contains is safe — block.mustReadFrom silently skips
+// families/tags that aren't in the block.
+func buildTagProjectionFromGroupSchemas(measures map[string]*measureSchemaInfo) []model.TagProjection {
+ if len(measures) == 0 {
+ return nil
+ }
+ families := map[string]map[string]bool{} // family -> set(tag)
+ for _, m := range measures {
+ for _, tf := range m.TagFamilies {
+ if families[tf.Name] == nil {
+ families[tf.Name] = map[string]bool{}
+ }
+ for _, t := range tf.Tags {
+ families[tf.Name][t] = true
+ }
+ }
+ }
+ out := make([]model.TagProjection, 0, len(families))
+ famNames := make([]string, 0, len(families))
+ for name := range families {
+ famNames = append(famNames, name)
+ }
+ sort.Strings(famNames)
+ for _, name := range famNames {
+ tagSet := families[name]
+ tags := make([]string, 0, len(tagSet))
+ for t := range tagSet {
+ tags = append(tags, t)
+ }
+ sort.Strings(tags)
+ out = append(out, model.TagProjection{Family: name, Names: tags})
+ }
+ return out
+}
+
+// processPartInput is the closed-over context passed to processOneSourcePart;
+// keeping the helper free of long signatures. Field order is readability-first
+// for an internal helper; fieldalignment churn here would obscure intent.
+//
+//nolint:govet // internal-only helper, readability > minor padding savings
+type processPartInput struct {
+ decoder *encoding.BytesBlockDecoder
+ fileSystem fs.FileSystem
+ segStates *segStateRegistry
+ partIDGen *atomic.Uint64
+ flushCh chan<- flushJob
+ tagProjection []model.TagProjection
+ shardName string
+ shardDir string
+ partIDStr string
+ dstGroupRoot string
+ ir storage.IntervalRule
+}
+
+type processPartResult struct {
+ rows int64
+ bytes int64
+ targetParts int
+}
+
+// processOneSourcePart processes one source part. When the part is
+// entirely inside one aligned target segment (the common case for
+// already-grid-aligned backups), it takes the fast path: copy the
+// whole part directory file-for-file and skip the decode / re-encode
+// loop. Otherwise it falls back to the slow row-level split path
+// that re-routes each row by timestamp.
+func processOneSourcePart(in processPartInput) (processPartResult, error) {
+ var pr processPartResult
+ srcPartID, err := strconv.ParseUint(in.partIDStr, 16, 64)
+ if err != nil {
+ return pr, fmt.Errorf("parse partID %s: %w", in.partIDStr, err)
+ }
+ srcPartDir := filepath.Join(in.shardDir, in.partIDStr)
+
+ // Cheap metadata-only probe to decide between fast and slow paths.
+ var srcMeta partMetadata
+ srcMeta.mustReadMetadata(in.fileSystem, srcPartDir)
+
+ alignedMin := in.ir.Standard(time.Unix(0, srcMeta.MinTimestamp))
+ alignedMax := in.ir.Standard(time.Unix(0, srcMeta.MaxTimestamp))
+ if alignedMin.Equal(alignedMax) {
+ // Fast path: entire part lands in a single target aligned segment.
+ alignedSegName := formatDirectCopySegName(alignedMin, in.ir.Unit)
+ fastPathHits.Add(1)
+ return fastCopyOnePart(in, srcPartDir, alignedSegName, alignedMin, srcMeta.TotalCount)
+ }
+
+ slowPathHits.Add(1)
+ slowPathRows.Add(int64(srcMeta.TotalCount))
+ return slowCopyOnePart(in, srcPartID)
+}
+
+// slowCopyArena is a per-call slab that holds every nameValue and the
+// supporting []*nameValue / []nameValues backing arrays produced by
+// the row-level rewrite. Holding all of them in pre-sized contiguous
+// arrays — instead of doing one heap allocation per (row, family,
+// column) — was worth ~40-50% of the wall time on a cold-grid
+// workload by collapsing the GC scan footprint.
+//
+// The arena is reset (length truncated to 0; cap retained) every time
+// a chunk is flushed, so the same underlying memory serves the next
+// chunk without further allocator pressure. Buckets that still hold
+// *nameValue pointers into the arena are flushed before reset.
+type slowCopyArena struct {
+ nvs []nameValue
+ nvPtrs []*nameValue
+ famSlots []nameValues
+}
+
+func (a *slowCopyArena) reset() {
+ a.nvs = a.nvs[:0]
+ a.nvPtrs = a.nvPtrs[:0]
+ a.famSlots = a.famSlots[:0]
+}
+
+// allocNameValue appends a nameValue to the arena and returns a stable
+// pointer into the slab. The caller must guarantee cap(nvs) is large
+// enough that the append doesn't reallocate (i.e. arena was sized for
+// the chunk). On overflow we fall back to a fresh heap alloc to stay
+// correct under pathological inputs.
+func (a *slowCopyArena) allocNameValue(name string, value []byte, valueType pbv1.ValueType) *nameValue {
+ if len(a.nvs)+1 > cap(a.nvs) {
+ return &nameValue{name: name, value: value, valueType: valueType}
+ }
+ a.nvs = append(a.nvs, nameValue{name: name, value: value, valueType: valueType})
+ return &a.nvs[len(a.nvs)-1]
+}
+
+// arenaTakePtrs carves out a contiguous []*nameValue of length n from
+// the arena's pointer backing array. Like allocNameValue, it falls back
+// to a fresh slice if the arena would have to grow (which would
+// invalidate pointers we handed out earlier in the chunk).
+func arenaTakePtrs(a *slowCopyArena, n int) []*nameValue {
+ if len(a.nvPtrs)+n > cap(a.nvPtrs) {
+ return make([]*nameValue, n)
+ }
+ start := len(a.nvPtrs)
+ a.nvPtrs = a.nvPtrs[:start+n]
+ return a.nvPtrs[start : start+n : start+n]
+}
+
+// arenaTakeFamSlots carves out a contiguous []nameValues of length n
+// for the per-row tag-family slot table.
+func arenaTakeFamSlots(a *slowCopyArena, n int) []nameValues {
+ if len(a.famSlots)+n > cap(a.famSlots) {
+ return make([]nameValues, n)
+ }
+ start := len(a.famSlots)
+ a.famSlots = a.famSlots[:start+n]
+ return a.famSlots[start : start+n : start+n]
+}
+
+// slowCopyOnePartChunkRows caps how many rows the slow path keeps
+// in-memory across all buckets before forcing a flush. Tuned on the
+// cold-grid workload:
+//
+// - 100K rows: ~36s wall but ~26 GB RSS peak with NumCPU=16
+// workers — overran a kind-on-Mac 32 GB Docker VM.
+// - 50K rows: ~50s wall (~40% slower) but ~13 GB RSS peak — fits
+// comfortably in a 20 GB pod limit and leaves headroom for the
+// in-flight flush-pool encoders.
+//
+// The arena slab below is sized off this constant; halving it directly
+// halves the long-lived per-worker allocation.
+const slowCopyOnePartChunkRows = 50_000
+
+// slowCopyArenaPool pools per-call slowCopyArena instances so
+// successive parts processed by the same worker don't re-allocate the
+// nameValue / *nameValue / nameValues backing arrays. The pool retains
+// each arena's underlying cap; reset() truncates len to 0 but keeps
+// cap so the next part reuses the slab in-place.
+var slowCopyArenaPool = sync.Pool{
+ New: func() any {
+ // SkyWalking metric measures have 4-8 tag/field columns per row
+ // in practice. Sizing the slab at 8 (was 16) halves the long-
+ // lived per-worker allocation; pathological inputs fall back
+ // to fresh heap allocs in allocNameValue / arenaTakePtrs.
+ const estColumnsPerRow = 8
+ return &slowCopyArena{
+ nvs: make([]nameValue, 0, slowCopyOnePartChunkRows*estColumnsPerRow),
+ nvPtrs: make([]*nameValue, 0, slowCopyOnePartChunkRows*estColumnsPerRow),
+ famSlots: make([]nameValues, 0, slowCopyOnePartChunkRows*2),
+ }
+ },
+}
+
+func acquireSlowCopyArena() *slowCopyArena {
+ a := slowCopyArenaPool.Get().(*slowCopyArena)
+ a.reset()
+ return a
+}
+
+func releaseSlowCopyArena(a *slowCopyArena) {
+ a.reset()
+ slowCopyArenaPool.Put(a)
+}
+
+// appendBlockRowToBuckets routes one row of one block to the correct
+// per-aligned-segment bucket and writes its (seriesID, ts, version,
+// tagFamilies, field) tuple via the arena. Block buffers are aliased
+// directly into nameValue.value — the block must outlive the bucket
+// until flush completes, which slowCopyOnePart guarantees by deferring
+// releaseBlock until after flushChunk.
+//
+// Lives outside slowCopyOnePart so when banyandb's block struct gains
+// or drops a field, only this helper needs updating.
+// alignedSegCache memoises the last (alignedSegName, segment window)
+// pair so the row loop avoids ir.Standard + formatDirectCopySegName
+// on consecutive rows that share a target segment — typically the
+// common case once banyandb data is roughly time-clustered per block.
+type alignedSegCache struct {
+ name string
+ startNanos int64 // inclusive
+ endNanos int64 // exclusive
+}
+
+func (c *alignedSegCache) segNameFor(ir storage.IntervalRule, ts int64) string {
+ if c.name != "" && ts >= c.startNanos && ts < c.endNanos {
+ return c.name
+ }
+ start := ir.Standard(time.Unix(0, ts))
+ c.name = formatDirectCopySegName(start, ir.Unit)
+ c.startNanos = start.UnixNano()
+ c.endNanos = ir.NextTime(start).UnixNano()
+ return c.name
+}
+
+func appendBlockRowToBuckets(
+ ir storage.IntervalRule,
+ b *block,
+ seriesID common.SeriesID,
+ k uint64,
+ arena *slowCopyArena,
+ buckets map[string]*dataPoints,
+ segCache *alignedSegCache,
+) {
+ ts := b.timestamps[k]
+ alignedSegName := segCache.segNameFor(ir, ts)
+
+ dp, exists := buckets[alignedSegName]
+ if !exists {
+ dp = generateDataPoints()
+ dp.reset()
+ buckets[alignedSegName] = dp
+ }
+ dp.seriesIDs = append(dp.seriesIDs, seriesID)
+ dp.timestamps = append(dp.timestamps, ts)
+ dp.versions = append(dp.versions, b.versions[k])
+
+ rowTagFamilies := arenaTakeFamSlots(arena, len(b.tagFamilies))
+ for fi := range b.tagFamilies {
+ cf := &b.tagFamilies[fi]
+ ptrs := arenaTakePtrs(arena, len(cf.columns))
+ for ci := range cf.columns {
+ c := &cf.columns[ci]
+ var v []byte
+ if uint64(len(c.values)) > k {
+ v = c.values[k]
+ }
+ ptrs[ci] = arena.allocNameValue(c.name, v, c.valueType)
+ }
+ rowTagFamilies[fi] = nameValues{name: cf.name, values: ptrs}
+ }
+ dp.tagFamilies = append(dp.tagFamilies, rowTagFamilies)
+
+ fieldPtrs := arenaTakePtrs(arena, len(b.field.columns))
+ for ci := range b.field.columns {
+ c := &b.field.columns[ci]
+ var v []byte
+ if uint64(len(c.values)) > k {
+ v = c.values[k]
+ }
+ fieldPtrs[ci] = arena.allocNameValue(c.name, v, c.valueType)
+ }
+ dp.fields = append(dp.fields, nameValues{name: b.field.name, values: fieldPtrs})
+}
+
+// slowCopyOnePart handles the row-level rewrite path: opens the source
+// part, decodes blocks, and routes rows into per-aligned-target
+// buckets. Buckets are flushed when the cumulative row count reaches
+// slowCopyOnePartChunkRows or at end of part. Live blocks are held
+// alive across multiple block iterations so bucket nameValue.value
+// slices can alias the block buffers (zero-copy) — they are all
+// released only after the buckets that reference them have been
+// flushed.
+//
+// The primaryBlockMetadata walk + b.mustReadFrom pattern intentionally
+// mirrors migration_reader.go:readPartRows (the MigrationReplay sibling).
+// partIter (used by query / merge paths) carries a sids-filter + heap
+// merge that adds no value when we want every block of one part — and
+// would force an upfront scan to enumerate seriesIDs. Keep the two
+// migration_* functions in lockstep: if banyandb changes its block
+// layout, both update together.
+func slowCopyOnePart(in processPartInput, srcPartID uint64) (processPartResult, error) {
+ var pr processPartResult
+ p := mustOpenFilePart(srcPartID, in.shardDir, in.fileSystem)
+ defer p.close()
+
+ flushBuckets := func(buckets map[string]*dataPoints) error {
+ if len(buckets) == 0 {
+ return nil
+ }
+ // Submit one job per bucket to the shared flush pool and wait
+ // on a per-chunk WaitGroup. The shared pool caps concurrent
+ // encode + zstd at flushWorkerCount instead of spawning one
+ // goroutine per (outerWorker, bucket) — that unbounded fan-out
+ // ballooned heap to 224 GB compressed on cold-tier loads and
+ // got jetsam-killed.
+ var (
+ chunkWg sync.WaitGroup
+ chunkErrMu sync.Mutex
+ chunkErr error
+ chunkBytes atomic.Int64
+ )
+ setErr := func(err error) {
+ chunkErrMu.Lock()
+ if chunkErr == nil {
+ chunkErr = err
+ }
+ chunkErrMu.Unlock()
+ }
+ addBytes := func(b int64) { chunkBytes.Add(b) }
+ for alignedSegName, dp := range buckets {
+ chunkWg.Add(1)
+ targetPartID := in.partIDGen.Add(1)
+ targetPartIDStr := fmt.Sprintf("%016x", targetPartID)
+ dstPart := filepath.Join(in.dstGroupRoot, alignedSegName, in.shardName, targetPartIDStr)
+ in.flushCh <- flushJob{
+ dp: dp,
+ alignedSeg: alignedSegName,
+ dstPart: dstPart,
+ targetPartID: targetPartID,
+ shardName: in.shardName,
+ ir: in.ir,
+ segStates: in.segStates,
+ chunkWg: &chunkWg,
+ setErr: setErr,
+ addBytes: addBytes,
+ }
+ }
+ chunkWg.Wait()
+ pr.bytes += chunkBytes.Load()
+ pr.targetParts += len(buckets)
+ return chunkErr
+ }
+
+ buckets := map[string]*dataPoints{}
+ liveBlocks := make([]*block, 0, 8)
+ chunkRows := 0
+ arena := acquireSlowCopyArena()
+ defer releaseSlowCopyArena(arena)
+ var segCache alignedSegCache
+
+ releaseLive := func() {
+ for _, b := range liveBlocks {
+ releaseBlock(b)
+ }
+ liveBlocks = liveBlocks[:0]
+ }
+ // Guarantee live blocks return to the pool even on flush errors,
+ // chunk-walk errors, or early returns below. Successful flushChunk
+ // calls drain liveBlocks first; this defer is a safety net.
+ defer releaseLive()
+ // Safety net for unflushed bucket dp's on early-return error paths
+ // (decompress / unmarshal failures below). Once flushBuckets has been
+ // called the dp's are owned by the workers (directCopyFlushBucket
+ // defers releaseDataPoints), so flushChunk clears the map afterwards
+ // to keep this defer a no-op in the normal path.
+ defer func() {
+ for _, dp := range buckets {
+ releaseDataPoints(dp)
+ }
+ }()
+ flushChunk := func() error {
+ if chunkRows == 0 {
+ return nil
+ }
+ err := flushBuckets(buckets)
+ // Workers have already returned every submitted dp to the pool
+ // via directCopyFlushBucket's defer (regardless of per-job
+ // error). Drop our map references so the outer defer doesn't
+ // double-release.
+ buckets = map[string]*dataPoints{}
+ if err != nil {
+ return err
+ }
+ releaseLive()
+ arena.reset()
+ chunkRows = 0
+ return nil
+ }
+
+ var (
+ compressed []byte
+ raw []byte
+ bms []blockMetadata
+ )
+ for i := range p.primaryBlockMetadata {
+ pbm := &p.primaryBlockMetadata[i]
+ compressed = bytes.ResizeOver(compressed, int(pbm.size))
+ fs.MustReadData(p.primary, int64(pbm.offset), compressed)
+ var err error
+ raw, err = zstd.Decompress(raw[:0], compressed)
+ if err != nil {
+ releaseLive()
+ return pr, fmt.Errorf("decompress primary block: %w", err)
+ }
+ bms, err = unmarshalBlockMetadata(bms[:0], raw)
+ if err != nil {
+ releaseLive()
+ return pr, fmt.Errorf("unmarshal block metadata: %w", err)
+ }
+ for j := range bms {
+ bm := &bms[j]
+ bm.tagProjection = in.tagProjection
+ b := generateBlock()
+ b.mustReadFrom(in.decoder, p, *bm)
+ liveBlocks = append(liveBlocks, b)
+
+ for k := uint64(0); k < bm.count; k++ {
+ appendBlockRowToBuckets(in.ir, b, bm.seriesID, k, arena, buckets, &segCache)
+ pr.rows++
+ chunkRows++
+ }
+ if chunkRows >= slowCopyOnePartChunkRows {
+ if err := flushChunk(); err != nil {
+ return pr, err
+ }
+ }
+ }
+ }
+ if err := flushChunk(); err != nil {
+ return pr, err
+ }
+ return pr, nil
+}
+
+// fastCopyOnePart handles the case where every row in a source part
+// already lands in the same target aligned segment. We can skip all
+// decoding / re-encoding and just copy the part directory verbatim
+// (under a fresh target partID so concurrent sources don't collide).
+// The caller pre-resolved alignedSegName + alignedTime from the part
+// metadata so this function does no further timestamp math.
+func fastCopyOnePart(
+ in processPartInput,
+ srcPartDir, alignedSegName string,
+ alignedTime time.Time,
+ totalCount uint64,
+) (processPartResult, error) {
+ var pr processPartResult
+ targetPartID := in.partIDGen.Add(1)
+ targetPartIDStr := fmt.Sprintf("%016x", targetPartID)
+ dstPart := filepath.Join(in.dstGroupRoot, alignedSegName, in.shardName, targetPartIDStr)
+ pr.rows = int64(totalCount)
+ pr.targetParts = 1
+
+ if err := os.MkdirAll(filepath.Dir(dstPart), storage.DirPerm); err != nil {
+ return pr, fmt.Errorf("mkdir %s: %w", filepath.Dir(dstPart), err)
+ }
+ bytesCopied, err := directCopyDir(srcPartDir, dstPart)
+ if err != nil {
+ return pr, fmt.Errorf("fast-copy %s -> %s: %w", srcPartDir, dstPart, err)
+ }
+ pr.bytes = bytesCopied
+
+ in.segStates.register(alignedSegName, alignedTime, in.shardName, targetPartID)
+ return pr, nil
+}
+
+// flushJob is one bucket's worth of work submitted to the shared flush
+// pool: write dp out as a target part, register the segment, and report
+// bytes / errors back to the submitting chunk.
+type flushJob struct {
+ dp *dataPoints
+ chunkWg *sync.WaitGroup
+ setErr func(error)
+ addBytes func(int64)
+ segStates *segStateRegistry
+ alignedSeg string
+ dstPart string
+ shardName string
+ targetPartID uint64
+ ir storage.IntervalRule
+}
+
+// runFlushWorker drains flushCh until it is closed. The flush pool size
+// is the only cap on concurrent encode + zstd + write — sized to NumCPU
+// in directCopyGroup so the host can't be drowned by a chunked slow path.
+func runFlushWorker(flushCh <-chan flushJob, fileSystem fs.FileSystem) {
+ for job := range flushCh {
+ sz, err := directCopyFlushBucket(job.dp, fileSystem, job.dstPart)
+ if err != nil {
+ job.setErr(fmt.Errorf("flush %s: %w", job.dstPart, err))
+ job.chunkWg.Done()
+ continue
+ }
+ job.addBytes(sz)
+ alignedTime, err := parseDirectCopySegStart(job.alignedSeg, job.ir.Unit)
+ if err != nil {
+ job.setErr(fmt.Errorf("parse seg start %s: %w", job.alignedSeg, err))
+ job.chunkWg.Done()
+ continue
+ }
+ job.segStates.register(job.alignedSeg, alignedTime, job.shardName, job.targetPartID)
+ job.chunkWg.Done()
+ }
+}
+
+// directCopyFlushBucket builds a memPart from dp and flushes it to disk at
+// partPath, then releases pooled objects. mustFlush internally calls
+// MkdirPanicIfExist(partPath), so the caller must only pre-create the
+// parent shard dir; this helper handles that. dp is always returned to
+// the pool — including on the MkdirAll error path.
+func directCopyFlushBucket(dp *dataPoints, fileSystem fs.FileSystem, partPath string) (int64, error) {
+ defer releaseDataPoints(dp)
+ if err := os.MkdirAll(filepath.Dir(partPath), storage.DirPerm); err != nil {
+ return 0, err
+ }
+ mp := generateMemPart()
+ mp.mustInitFromDataPoints(dp)
+ mp.mustFlush(fileSystem, partPath)
+ sz := int64(mp.partMetadata.CompressedSizeBytes)
+ releaseMemPart(mp)
+ return sz, nil
+}
+
+// directCopyPrepareTarget refuses to write into a non-empty group dir.
+// The operator must remove (or relocate) the previous target manually
+// before re-running.
+func directCopyPrepareTarget(dstGroupRoot string) error {
+ entries, err := os.ReadDir(dstGroupRoot)
+ if err != nil {
+ if os.IsNotExist(err) {
+ return os.MkdirAll(dstGroupRoot, storage.DirPerm)
+ }
+ return err
+ }
+ if len(entries) > 0 {
+ return fmt.Errorf("target %s is not empty; remove it before re-running", dstGroupRoot)
+ }
+ return nil
+}
+
+// formatDirectCopySegName mirrors segmentController.format.
+func formatDirectCopySegName(t time.Time, unit storage.IntervalUnit) string {
+ switch unit {
+ case storage.HOUR:
+ return directCopySegPrefix + t.Format(directCopyHourFormat)
+ case storage.DAY:
+ return directCopySegPrefix + t.Format(directCopyDayFormat)
+ }
+ panic(fmt.Sprintf("formatDirectCopySegName: unsupported interval unit %v", unit))
+}
+
+// parseDirectCopySegStart inverts formatDirectCopySegName.
+func parseDirectCopySegStart(segName string, unit storage.IntervalUnit) (time.Time, error) {
+ suffix := strings.TrimPrefix(segName, directCopySegPrefix)
+ switch unit {
+ case storage.HOUR:
+ return time.ParseInLocation(directCopyHourFormat, suffix, time.Local)
+ case storage.DAY:
+ return time.ParseInLocation(directCopyDayFormat, suffix, time.Local)
+ }
+ return time.Time{}, fmt.Errorf("unrecognized interval unit %v", unit)
+}
+
+// writeDirectCopySegmentMetadata writes <segDir>/metadata matching the
+// runtime's segmentMeta layout (banyand/internal/storage/version.go).
+// The filename, JSON tags, and Version string all come from the exported
+// storage helpers so a banyandb runtime that reads compatibleVersions
+// from versions.yml will accept these files without modification.
+func writeDirectCopySegmentMetadata(segDir string, endTime time.Time) (int64, error) {
+ if err := os.MkdirAll(segDir, storage.DirPerm); err != nil {
+ return 0, err
+ }
+ body := storage.SegmentMetadata{
+ Version: storage.CurrentSegmentVersion,
+ EndTime: endTime.Format(time.RFC3339Nano),
+ }
+ data, err := json.Marshal(body)
+ if err != nil {
+ return 0, err
+ }
+ if err := os.WriteFile(
+ filepath.Join(segDir, storage.SegmentMetadataFilename),
+ data, storage.FilePerm,
+ ); err != nil {
+ return 0, err
+ }
+ return int64(len(data)), nil
+}
+
+// writeDirectCopySnp writes a <epoch-hex16>.snp file listing the part
+// names for one (alignedSeg, shard) so tsTable.loadSnapshot finds them
+// at next startup.
+func writeDirectCopySnp(dstShard string, partNames []string) (int64, error) {
+ if err := os.MkdirAll(dstShard, storage.DirPerm); err != nil {
+ return 0, err
+ }
+ data, err := json.Marshal(partNames)
+ if err != nil {
+ return 0, err
+ }
+ snpPath := filepath.Join(dstShard,
+ fmt.Sprintf("%016x%s", time.Now().UnixNano(), directCopySnpSuffix))
+ if err := os.WriteFile(snpPath, data, storage.FilePerm); err != nil {
+ return 0, err
+ }
+ return int64(len(data)), nil
+}
+
+// directCopyDir replicates src into dst via byte-level file copy,
+// returning total bytes materialized. Used both for the union-sidx
+// broadcast and for the fast-path whole-part copy.
+//
+// Earlier versions of this helper attempted a recursive hard-link via
+// fs.LocalFileSystem.CreateHardLink first, falling back to byte copy
+// on cross-FS failure. Every observed deployment path (staging on
+// emptyDir → target on PVC) trips cross-FS and forces the byte path
+// anyway, so the hardlink attempt was pure overhead + noisy log
+// lines. We now always go straight to the byte copy.
+func directCopyDir(src, dst string) (int64, error) {
+ if err := os.MkdirAll(filepath.Dir(dst), storage.DirPerm); err != nil {
+ return 0, err
+ }
+ return directCopyDirByteCopy(src, dst)
+}
+
+// directCopyDirByteCopy walks src and byte-copies every file into dst.
+// Called from directCopyDir for both the union-sidx broadcast and the
+// fast-path whole-part copy.
+func directCopyDirByteCopy(src, dst string) (int64, error) {
+ var total int64
+ if err := os.MkdirAll(dst, storage.DirPerm); err != nil {
+ return 0, err
+ }
+ entries, err := os.ReadDir(src)
+ if err != nil {
+ return 0, err
+ }
+ for _, e := range entries {
+ srcPath := filepath.Join(src, e.Name())
+ dstPath := filepath.Join(dst, e.Name())
+ if e.IsDir() {
+ sub, recErr := directCopyDirByteCopy(srcPath, dstPath)
+ if recErr != nil {
+ return total, recErr
+ }
+ total += sub
+ continue
+ }
+ n, copyErr := directCopyFile(srcPath, dstPath)
+ if copyErr != nil {
+ return total, copyErr
+ }
+ total += n
+ }
+ return total, nil
+}
+
+func directCopyFile(src, dst string) (int64, error) {
+ in, err := os.Open(src)
+ if err != nil {
+ return 0, err
+ }
+ defer in.Close()
+ out, err := os.OpenFile(dst, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, storage.FilePerm)
+ if err != nil {
+ return 0, err
+ }
+ // No out.Sync(): consistent with nofsyncFS, migration is one-shot
+ // bulk import and operator re-runs with --clean on crash.
+ defer out.Close()
+ n, copyErr := io.Copy(out, in)
+ return n, copyErr
+}
+
+// Field names copied verbatim from pkg/index/inverted/inverted.go. The
+// migration tool deliberately does not depend on that package's
+// (writer-flavored) constants because we open every source sidx
+// read-only via raw bluge — no banyandb store lifecycle is needed.
+const (
+ unionSidxDocIDField = "_id"
+ unionSidxTimestampField = "_timestamp"
+ unionSidxVersionField = "_version"
+
+ // Larger batches mean fewer bluge merger ticks; the merger loop
+ // dominates CPU at small batch sizes. 50k keeps the merger work
+ // off the hot path while still fitting in the per-process heap.
+ unionSidxBatchSize = 50000
+)
+
+// buildGroupUnionSidx walks every srcGroupRoot/seg-*/sidx/ directory
+// across every supplied group root, scans every series-index doc,
+// deduplicates by SeriesID, and re-emits the surviving docs into a
+// fresh bluge index rooted at stagingPath.
+//
+// The returned path is stagingPath when at least one doc was written;
+// it is "" (without error) when no source sidx contained any doc — the
+// caller treats this as "no sidx to broadcast" and skips the per-target
+// copy. SeriesID is recovered by unmarshaling the source doc's EntityValues
+// (the bluge _id field) into pbv1.Series and reading series.ID.
+//
+// We intentionally bypass pkg/index/inverted's store wrapper here: that
+// wrapper starts background flush goroutines and registers metrics, both
+// of which are unnecessary for a one-shot offline build. Raw bluge
+// readers/writers keep the staging artifact byte-for-byte compatible
+// with what BanyanDB writes at runtime — the wrapper itself ultimately
+// produces the same .seg/.snp layout.
+func buildGroupUnionSidx(ctx context.Context, srcGroupRoots []string, stagingPath string) (string, error) {
+ if err := os.MkdirAll(stagingPath, storage.DirPerm); err != nil {
+ return "", fmt.Errorf("mkdir staging %q: %w", stagingPath, err)
+ }
+
+ writer, err := bluge.OpenWriter(bluge.DefaultConfig(stagingPath))
+ if err != nil {
+ return "", fmt.Errorf("open union sidx writer at %q: %w", stagingPath, err)
+ }
+ closed := false
+ defer func() {
+ if !closed {
+ _ = writer.Close()
+ }
+ }()
+
+ // Collect every source sidx directory across all node + segment
+ // roots up front so the worker pool can fan out cleanly.
+ var sidxPaths []string
+ for _, srcGroupRoot := range srcGroupRoots {
+ if ctx.Err() != nil {
+ return "", ctx.Err()
+ }
+ segEntries, err := os.ReadDir(srcGroupRoot)
+ if err != nil {
+ if os.IsNotExist(err) {
+ continue
+ }
+ return "", fmt.Errorf("read src group root %q: %w", srcGroupRoot, err)
+ }
+ for _, se := range segEntries {
+ if !se.IsDir() || !strings.HasPrefix(se.Name(), directCopySegPrefix) {
+ continue
+ }
+ srcSidxPath := filepath.Join(srcGroupRoot, se.Name(), directCopySidxDirName)
+ info, statErr := os.Stat(srcSidxPath)
+ if statErr != nil || !info.IsDir() {
+ continue
+ }
+ sidxPaths = append(sidxPaths, srcSidxPath)
+ }
+ }
+
+ seen := make(map[common.SeriesID]struct{}, 1_000_000)
+ var seenMu sync.Mutex
+ var writerMu sync.Mutex
+ var insertedAtomic atomic.Int64
+ var firstErr atomic.Pointer[error]
+
+ workerCount := runtime.NumCPU()
+ if workerCount > len(sidxPaths) {
+ workerCount = len(sidxPaths)
+ }
+ if workerCount < 1 {
+ workerCount = 1
+ }
+
+ pathCh := make(chan string)
+ var wg sync.WaitGroup
+ workerCtx, cancelWorkers := context.WithCancel(ctx)
+ defer cancelWorkers()
+
+ for i := 0; i < workerCount; i++ {
+ wg.Add(1)
+ go func() {
+ defer wg.Done()
+ for srcSidxPath := range pathCh {
+ if workerCtx.Err() != nil {
+ return
+ }
+ count, mergeErr := mergeOneSourceSidxInto(workerCtx, srcSidxPath, writer, seen, &seenMu, &writerMu)
+ if mergeErr != nil {
+ e := fmt.Errorf("merge %s: %w", srcSidxPath, mergeErr)
+ if firstErr.CompareAndSwap(nil, &e) {
+ cancelWorkers()
+ }
+ return
+ }
+ insertedAtomic.Add(int64(count))
+ }
+ }()
+ }
+ for _, p := range sidxPaths {
+ select {
+ case pathCh <- p:
+ case <-workerCtx.Done():
+ }
+ }
+ close(pathCh)
+ wg.Wait()
+ if errPtr := firstErr.Load(); errPtr != nil {
+ return "", *errPtr
+ }
+ inserted := int(insertedAtomic.Load())
+
+ closed = true
+ if closeErr := writer.Close(); closeErr != nil {
+ return "", fmt.Errorf("close union sidx writer: %w", closeErr)
+ }
+ if inserted == 0 {
+ // Nothing to broadcast. Remove the empty bluge directory so the
+ // caller can cleanly signal "no sidx for this group" by checking
+ // for an empty return path. Leaving a half-baked empty index
+ // behind would otherwise confuse a future bluge open at the
+ // same path.
+ _ = os.RemoveAll(stagingPath)
+ return "", nil
+ }
+ return stagingPath, nil
+}
+
+// mergeOneSourceSidxInto opens one source sidx, scans every doc, and
+// inserts unseen-SeriesID docs into dst in batched flushes. Concurrent
+// callers serialize on seenMu (dedup map) and writerMu (single bluge
+// writer); cross-source readers run in parallel.
+func mergeOneSourceSidxInto(
+ ctx context.Context,
+ srcPath string,
+ dst *bluge.Writer,
+ seen map[common.SeriesID]struct{},
+ seenMu *sync.Mutex,
+ writerMu *sync.Mutex,
+) (int, error) {
+ reader, err := bluge.OpenReader(bluge.DefaultConfig(srcPath))
+ if err != nil {
+ // A sidx directory may exist on disk while carrying no committed
+ // bluge snapshot — e.g. a fresh segment created by the runtime
+ // whose sidx writer never received a doc. Treat that as "no
+ // sidx to merge for this seg" instead of aborting the whole
+ // per-group union build.
+ if strings.Contains(err.Error(), "unable to find a usable snapshot") {
+ return 0, nil
+ }
+ return 0, fmt.Errorf("open reader: %w", err)
+ }
+ defer func() { _ = reader.Close() }()
+
+ dmi, err := reader.Search(ctx, bluge.NewAllMatches(bluge.NewMatchAllQuery()))
+ if err != nil {
+ return 0, fmt.Errorf("search: %w", err)
+ }
+
+ batch := bluge.NewBatch()
+ batched := 0
+ inserted := 0
+
+ flush := func() error {
+ writerMu.Lock()
+ defer writerMu.Unlock()
+ return dst.Batch(batch)
+ }
+
+ for {
+ next, nextErr := dmi.Next()
+ if nextErr != nil {
+ return inserted, fmt.Errorf("iterate docs: %w", nextErr)
+ }
+ if next == nil {
+ break
+ }
+
+ doc, dup, buildErr := buildDocFromMatchLocked(next, seen, seenMu)
+ if buildErr != nil {
+ return inserted, buildErr
+ }
+ if dup || doc == nil {
+ continue
+ }
+ batch.Insert(doc)
+ batched++
+ inserted++
+ if batched >= unionSidxBatchSize {
+ if err := flush(); err != nil {
+ return inserted, fmt.Errorf("flush batch: %w", err)
+ }
+ batch = bluge.NewBatch()
+ batched = 0
+ }
+ }
+ if batched > 0 {
+ if err := flush(); err != nil {
+ return inserted, fmt.Errorf("flush tail batch: %w", err)
+ }
+ }
+ return inserted, nil
+}
+
+// buildDocFromMatchLocked is the concurrency-safe variant: the caller
+// passes the dedup map's mutex; the seen check + insert is a single
+// critical section so two workers never insert the same SeriesID.
+func buildDocFromMatchLocked(
+ match *search.DocumentMatch,
+ seen map[common.SeriesID]struct{},
+ seenMu *sync.Mutex,
+) (*bluge.Document, bool, error) {
+ var entityValues []byte
+ type storedField struct {
+ name string
+ value []byte
+ }
+ var fields []storedField
+ visitErr := match.VisitStoredFields(func(field string, value []byte) bool {
+ switch field {
+ case unionSidxDocIDField:
+ entityValues = append([]byte(nil), value...)
+ default:
+ fields = append(fields, storedField{
+ name: field,
+ value: append([]byte(nil), value...),
+ })
+ }
+ return true
+ })
+ if visitErr != nil {
+ return nil, false, fmt.Errorf("visit stored fields: %w", visitErr)
+ }
+ if len(entityValues) == 0 {
+ return nil, false, nil
+ }
+ var series pbv1.Series
+ if err := series.Unmarshal(entityValues); err != nil {
+ return nil, false, nil
+ }
+ seenMu.Lock()
+ if _, dup := seen[series.ID]; dup {
+ seenMu.Unlock()
+ return nil, true, nil
+ }
+ seen[series.ID] = struct{}{}
+ seenMu.Unlock()
+
+ doc := bluge.NewDocument(string(entityValues))
+ for _, f := range fields {
+ switch f.name {
+ case unionSidxTimestampField:
+ ts, decErr := bluge.DecodeDateTime(f.value)
+ if decErr != nil {
+ return nil, false, fmt.Errorf("decode timestamp on series %d: %w", series.ID, decErr)
+ }
+ doc.AddField(bluge.NewDateTimeField(f.name, ts).StoreValue())
+ case unionSidxVersionField:
+ doc.AddField(bluge.NewStoredOnlyField(f.name, f.value))
+ default:
+ doc.AddField(bluge.NewKeywordFieldBytes(f.name, f.value).StoreValue())
+ }
+ }
+ return doc, false, nil
+}
diff --git a/banyand/measure/migration_copy_test.go b/banyand/measure/migration_copy_test.go
new file mode 100644
index 000000000..aa35b3627
--- /dev/null
+++ b/banyand/measure/migration_copy_test.go
@@ -0,0 +1,303 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package measure
+
+import (
+ "context"
+ "os"
+ "path/filepath"
+ "reflect"
+ "sort"
+ "strings"
+ "testing"
+
+ "github.com/blugelabs/bluge"
+
+ "github.com/apache/skywalking-banyandb/api/common"
+ modelv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/model/v1"
+ "github.com/apache/skywalking-banyandb/banyand/internal/storage"
+ pbv1 "github.com/apache/skywalking-banyandb/pkg/pb/v1"
+)
+
+// Src-root resolution tests.
+
+// TestCollectAllSrcGroupRoots_LiveDedup checks that the helper merges
+// every entry's source dirs across the whole plan, drops duplicates,
+// silently skips paths that don't host the group, and returns a sorted
+// slice — these properties are what Phase A relies on to build a single
+// union sidx per group.
+func TestCollectAllSrcGroupRoots_LiveDedup(t *testing.T) {
+ tmp := t.TempDir()
+ hot0 := filepath.Join(tmp, "hot-0", "measure", "data")
+ hot1 := filepath.Join(tmp, "hot-1", "measure", "data")
+ warn0 := filepath.Join(tmp, "warn-0", "measure", "data")
+ group := "sw_metricsMinute"
+
+ // Only hot-0 and warn-0 actually carry the group; hot-1 doesn't.
+ for _, root := range []string{hot0, warn0} {
+ if err := os.MkdirAll(filepath.Join(root, group), 0o755); err != nil {
+ t.Fatalf("seed %s: %v", root, err)
+ }
+ }
+ if err := os.MkdirAll(hot1, 0o755); err != nil {
+ t.Fatalf("seed hot-1: %v", err)
+ }
+
+ cfg := DirectCopyConfig{
+ Groups: []string{group},
+ Entries: []DirectCopyEntry{
+ // Two entries both pointing at hot-0 — must dedup.
+ {Stage: "hot", Target: filepath.Join(tmp, "out-a"), Source: []string{hot0}},
+ {Stage: "hot", Target: filepath.Join(tmp, "out-b"), Source: []string{hot0, hot1}},
+ {Stage: "warm", Target: filepath.Join(tmp, "out-c"), Source: []string{warn0}},
+ },
+ }
+
+ got := collectAllSrcGroupRoots(cfg, group)
+ want := []string{
+ filepath.Join(hot0, group),
+ filepath.Join(warn0, group),
+ }
+ if !reflect.DeepEqual(got, want) {
+ t.Fatalf("collectAllSrcGroupRoots merged result mismatch:\n got: %v\n want: %v", got, want)
+ }
+}
+
+// TestCollectAllSrcGroupRoots_BackupNodes covers the backup-mode path
+// (entry.Nodes drives root resolution via cfg.BackupDir).
+func TestCollectAllSrcGroupRoots_BackupNodes(t *testing.T) {
+ tmp := t.TempDir()
+ date := "2026-05-19"
+ group := "sw_metricsMinute"
+
+ for _, node := range []string{"hot-0", "warn-0"} {
+ if err := os.MkdirAll(filepath.Join(tmp, node, date, "measure", group), 0o755); err != nil {
+ t.Fatalf("seed %s: %v", node, err)
+ }
+ }
+
+ cfg := DirectCopyConfig{
+ BackupDir: tmp,
+ Date: date,
+ Groups: []string{group},
+ Entries: []DirectCopyEntry{
+ {Stage: "hot", Target: filepath.Join(tmp, "out-a"), Nodes: []string{"hot-0"}},
+ // Entry-2 references both hot-0 (dup) and a non-existent node.
+ {Stage: "hot", Target: filepath.Join(tmp, "out-b"), Nodes: []string{"hot-0", "ghost"}},
+ {Stage: "warm", Target: filepath.Join(tmp, "out-c"), Nodes: []string{"warn-0"}},
+ },
+ }
+
+ got := collectAllSrcGroupRoots(cfg, group)
+ want := []string{
+ filepath.Join(tmp, "hot-0", date, "measure", group),
+ filepath.Join(tmp, "warn-0", date, "measure", group),
+ }
+ if !reflect.DeepEqual(got, want) {
+ t.Fatalf("collectAllSrcGroupRoots backup-mode mismatch:\n got: %v\n want: %v", got, want)
+ }
+}
+
+// Union sidx build tests.
+
+// makeSidxSourceDoc builds one bluge doc whose layout mirrors what the
+// measure write path emits into <segment>/sidx/: bluge _id is the marshaled
+// pbv1.Series buffer and the doc carries one stored+indexed tag field plus
+// (optionally) a stored version field. Returns the SeriesID for assertions.
+func makeSidxSourceDoc(t *testing.T, entity, tagValue string) (*bluge.Document, common.SeriesID) {
+ t.Helper()
+ series := &pbv1.Series{
+ Subject: "m1",
+ EntityValues: []*modelv1.TagValue{{Value: &modelv1.TagValue_Str{Str: &modelv1.Str{Value: entity}}}},
+ }
+ if err := series.Marshal(); err != nil {
+ t.Fatalf("series.Marshal: %v", err)
+ }
+ doc := bluge.NewDocument(string(series.Buffer))
+ doc.AddField(bluge.NewKeywordFieldBytes("service", []byte(tagValue)).StoreValue())
+ return doc, series.ID
+}
+
+// writeSidxAt creates a bluge sidx at <seg>/sidx/ populated with docs.
+func writeSidxAt(t *testing.T, segDir string, docs []*bluge.Document) {
+ t.Helper()
+ if err := os.MkdirAll(filepath.Join(segDir, directCopySidxDirName), storage.DirPerm); err != nil {
+ t.Fatalf("mkdir sidx: %v", err)
+ }
+ w, err := bluge.OpenWriter(bluge.DefaultConfig(filepath.Join(segDir, directCopySidxDirName)))
+ if err != nil {
+ t.Fatalf("open writer: %v", err)
+ }
+ batch := bluge.NewBatch()
+ for _, d := range docs {
+ batch.Insert(d)
+ }
+ if err := w.Batch(batch); err != nil {
+ t.Fatalf("batch: %v", err)
+ }
+ if err := w.Close(); err != nil {
+ t.Fatalf("close: %v", err)
+ }
+}
+
+// countDocsInSidx opens a bluge sidx and returns the count of docs whose
+// _id unmarshals into a pbv1.Series, plus the sorted list of SeriesIDs.
+func countDocsInSidx(t *testing.T, path string) []common.SeriesID {
+ t.Helper()
+ r, err := bluge.OpenReader(bluge.DefaultConfig(path))
+ if err != nil {
+ t.Fatalf("open reader: %v", err)
+ }
+ defer r.Close()
+ dmi, err := r.Search(context.Background(), bluge.NewAllMatches(bluge.NewMatchAllQuery()))
+ if err != nil {
+ t.Fatalf("search: %v", err)
+ }
+ var ids []common.SeriesID
+ for {
+ next, e := dmi.Next()
+ if e != nil {
+ t.Fatalf("iterate: %v", e)
+ }
+ if next == nil {
+ break
+ }
+ var entity []byte
+ _ = next.VisitStoredFields(func(field string, value []byte) bool {
+ if field == unionSidxDocIDField {
+ entity = append([]byte(nil), value...)
+ }
+ return true
+ })
+ if len(entity) == 0 {
+ continue
+ }
+ var s pbv1.Series
+ if err := s.Unmarshal(entity); err != nil {
+ continue
+ }
+ ids = append(ids, s.ID)
+ }
+ sort.Slice(ids, func(i, j int) bool { return ids[i] < ids[j] })
+ return ids
+}
+
+func TestBuildGroupUnionSidx_DeduplicatesAcrossSegments(t *testing.T) {
+ srcGroupRoot := t.TempDir()
+ stagingPath := filepath.Join(t.TempDir(), "union", "sidx")
+
+ // Three source segments. Seg A carries entity {alpha, beta}, Seg B
+ // carries {beta, gamma} (beta is the cross-segment overlap that the
+ // dedup logic must collapse), Seg C carries {delta} only.
+ segA := filepath.Join(srcGroupRoot, directCopySegPrefix+"20260101")
+ segB := filepath.Join(srcGroupRoot, directCopySegPrefix+"20260102")
+ segC := filepath.Join(srcGroupRoot, directCopySegPrefix+"20260103")
+
+ docAlpha, sidAlpha := makeSidxSourceDoc(t, "alpha", "svc-a")
+ docBetaA, sidBeta := makeSidxSourceDoc(t, "beta", "svc-b-from-A")
+ docBetaB, sidBetaB := makeSidxSourceDoc(t, "beta", "svc-b-from-B")
+ docGamma, sidGamma := makeSidxSourceDoc(t, "gamma", "svc-c")
+ docDelta, sidDelta := makeSidxSourceDoc(t, "delta", "svc-d")
+
+ if sidBeta != sidBetaB {
+ t.Fatalf("setup mismatch: beta SeriesID should be stable across segments, got %d vs %d", sidBeta, sidBetaB)
+ }
+
+ writeSidxAt(t, segA, []*bluge.Document{docAlpha, docBetaA})
+ writeSidxAt(t, segB, []*bluge.Document{docBetaB, docGamma})
+ writeSidxAt(t, segC, []*bluge.Document{docDelta})
+
+ resultPath, err := buildGroupUnionSidx(context.Background(), []string{srcGroupRoot}, stagingPath)
+ if err != nil {
+ t.Fatalf("buildGroupUnionSidx: %v", err)
+ }
+ if resultPath != stagingPath {
+ t.Fatalf("expected resultPath==stagingPath, got %q vs %q", resultPath, stagingPath)
+ }
+
+ got := countDocsInSidx(t, resultPath)
+ want := []common.SeriesID{sidAlpha, sidBeta, sidGamma, sidDelta}
+ sort.Slice(want, func(i, j int) bool { return want[i] < want[j] })
+ if len(got) != len(want) {
+ t.Fatalf("expected %d unique SeriesIDs, got %d (%v)", len(want), len(got), got)
+ }
+ for i, w := range want {
+ if got[i] != w {
+ t.Fatalf("SeriesID mismatch at %d: got %d want %d", i, got[i], w)
+ }
+ }
+}
+
+func TestBuildGroupUnionSidx_EmptyGroupYieldsEmptyPath(t *testing.T) {
+ srcGroupRoot := t.TempDir()
+ stagingPath := filepath.Join(t.TempDir(), "union", "sidx")
+
+ resultPath, err := buildGroupUnionSidx(context.Background(), []string{srcGroupRoot}, stagingPath)
+ if err != nil {
+ t.Fatalf("buildGroupUnionSidx: %v", err)
+ }
+ if resultPath != "" {
+ t.Fatalf("expected empty result path for empty group, got %q", resultPath)
+ }
+}
+
+func TestBuildGroupUnionSidx_MissingSrcRootIsNoop(t *testing.T) {
+ stagingPath := filepath.Join(t.TempDir(), "union", "sidx")
+ resultPath, err := buildGroupUnionSidx(context.Background(), []string{filepath.Join(t.TempDir(), "no-such-dir")}, stagingPath)
+ if err != nil {
+ t.Fatalf("buildGroupUnionSidx with missing src: %v", err)
+ }
+ if resultPath != "" {
+ t.Fatalf("expected empty result path when src root missing, got %q", resultPath)
+ }
+}
+
+// IndexMode rejection tests.
+
+func TestRejectIndexModeGroups_FailsOnIndexMode(t *testing.T) {
+ schemas := map[string]map[string]*measureSchemaInfo{
+ "sw_metricsMinute": {
+ "service_cpm_minute": {Group: "sw_metricsMinute", Name: "service_cpm_minute", IndexMode: false},
+ },
+ "sw_metadata": {
+ "service_traffic_minute": {Group: "sw_metadata", Name: "service_traffic_minute", IndexMode: true},
+ "endpoint_traffic_minute": {Group: "sw_metadata", Name: "endpoint_traffic_minute", IndexMode: true},
+ },
+ }
+ err := rejectIndexModeGroups([]string{"sw_metricsMinute", "sw_metadata"}, schemas)
+ if err == nil {
+ t.Fatalf("expected error for IndexMode group, got nil")
+ }
+ msg := err.Error()
+ if !strings.Contains(msg, "sw_metadata/service_traffic_minute") ||
+ !strings.Contains(msg, "sw_metadata/endpoint_traffic_minute") {
+ t.Fatalf("error message %q must name both offending measures", msg)
+ }
+}
+
+func TestRejectIndexModeGroups_PassesWhenAllRegular(t *testing.T) {
+ schemas := map[string]map[string]*measureSchemaInfo{
+ "sw_metricsMinute": {
+ "a": {Group: "sw_metricsMinute", Name: "a", IndexMode: false},
+ "b": {Group: "sw_metricsMinute", Name: "b", IndexMode: false},
+ },
+ }
+ if err := rejectIndexModeGroups([]string{"sw_metricsMinute"}, schemas); err != nil {
+ t.Fatalf("expected nil for non-IndexMode groups, got: %v", err)
+ }
+}
diff --git a/banyand/measure/migration_e2e_test.go b/banyand/measure/migration_e2e_test.go
new file mode 100644
index 000000000..6c90cab9c
--- /dev/null
+++ b/banyand/measure/migration_e2e_test.go
@@ -0,0 +1,598 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package measure_test
+
+import (
+ "context"
+ "encoding/json"
+ "fmt"
+ "os"
+ "path/filepath"
+ "strconv"
+ "strings"
+ "time"
+
+ "github.com/blugelabs/bluge"
+ . "github.com/onsi/ginkgo/v2"
+ . "github.com/onsi/gomega"
+ "google.golang.org/protobuf/encoding/protojson"
+ "google.golang.org/protobuf/types/known/timestamppb"
+
+ "github.com/apache/skywalking-banyandb/api/common"
+ "github.com/apache/skywalking-banyandb/api/data"
+ commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
+ databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
+ measurev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/measure/v1"
+ modelv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/model/v1"
+ propertyv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/property/v1"
+ "github.com/apache/skywalking-banyandb/banyand/internal/storage"
+ "github.com/apache/skywalking-banyandb/banyand/measure"
+ "github.com/apache/skywalking-banyandb/banyand/metadata/schema"
+ "github.com/apache/skywalking-banyandb/banyand/metadata/service"
+ obsservice "github.com/apache/skywalking-banyandb/banyand/observability/services"
+ "github.com/apache/skywalking-banyandb/banyand/protector"
+ "github.com/apache/skywalking-banyandb/banyand/query"
+ "github.com/apache/skywalking-banyandb/banyand/queue"
+ "github.com/apache/skywalking-banyandb/pkg/bus"
+ "github.com/apache/skywalking-banyandb/pkg/fs"
+ "github.com/apache/skywalking-banyandb/pkg/test"
+)
+
+// MigrationCopy end-to-end: write through the live measure service (so the
+// schema goes through the metadata registry and parts land on disk via the
+// normal write → flush path), then run MigrationCopy on the live dataPath,
+// inspect the migrated target tree at both the **file level**
+// (EnumerateGroupTarget) AND at the **service level** by re-opening the
+// target tree from a second live measure service and calling
+// CollectDataInfo on it — that path mirrors what banyandb's runtime does
+// after Step 14 of the in-cluster swap (MIGRATION.md Part 2), so the test
+// catches misalignment that pure on-disk inspection might miss.
+
+const defaultTagFamily = "default"
+
+var _ = Describe("MigrationCopy end-to-end (live service)", func() {
+ const (
+ group = "e2e_two_day_migration"
+ measureName = "e2e_two_day_metric"
+ )
+
+ It("migrates a live measure tree and CollectDataInfo on a re-opened target matches the source", func() {
+ ctx := context.TODO()
+ twoDayInterval := storage.IntervalRule{Unit: storage.DAY, Num: 2}
+
+ // Use ginkgo-owned temp dirs for source + target rootPaths so the
+ // source data survives the deliberate mid-test source-service
+ // shutdown (which must happen before MigrationCopy runs so every
+ // per-seg seriesIndex commits its bluge writer to disk).
+ workspace := GinkgoT().TempDir()
+ sourceRoot := filepath.Join(workspace, "source")
+ targetRoot := filepath.Join(workspace, "target")
+ Expect(os.MkdirAll(sourceRoot, 0o755)).To(Succeed())
+
+ // Bring up the source measure service. Wait for the preloaded
+ // sw_metric group so we know the schema pipeline is live, then
+ // register our own DAY×2 group + measure.
+ svcs, deferFn := setUpMigrationTarget(sourceRoot)
+ sourceDown := func() {
+ if deferFn != nil {
+ deferFn()
+ deferFn = nil
+ }
+ }
+ defer sourceDown()
+ Eventually(func() bool {
+ _, ok := svcs.measure.LoadGroup("sw_metric")
+ return ok
+ }).WithTimeout(30 * time.Second).Should(BeTrue())
+ registerMigrationE2EGroup(svcs, group, measureName)
+ Eventually(func() bool {
+ _, ok := svcs.measure.LoadGroup(group)
+ return ok
+ }).WithTimeout(30 * time.Second).Should(BeTrue())
+ // Group-loaded does NOT imply measure-loaded — the measure
+ // event arrives on a separate schemaRepo channel. Wait for
+ // it explicitly, otherwise the first writes get rejected with
+ // "cannot find measure definition" and the Eventually below
+ // times out waiting for segments to land.
+ Eventually(func() error {
+ _, err := svcs.measure.Measure(&commonv1.Metadata{Name: measureName, Group: group})
+ return err
+ }).WithTimeout(30*time.Second).Should(Succeed(),
+ "source service should resolve the measure before writes are published")
+
+ // Pick two timestamps that land in DIFFERENT DAY×2 buckets:
+ // align "now" down to the current 2-day bucket, then step back 2
+ // and 4 buckets so the source produces exactly two on-disk segs.
+ nowBucket := twoDayInterval.Standard(time.Now().UTC())
+ day1 := nowBucket.AddDate(0, 0, -4).Add(8 * time.Hour)
+ day2 := nowBucket.AddDate(0, 0, -2).Add(8 * time.Hour)
+ writeMigrationE2EPoints(svcs, group, measureName, day1, 3, 0)
+ writeMigrationE2EPoints(svcs, group, measureName, day2, 3, 3)
+
+ // Wait until the flusher has materialized two segments on disk.
+ var sourceInfo *databasev1.DataInfo
+ Eventually(func() int {
+ info, err := svcs.measure.CollectDataInfo(ctx, group)
+ if err != nil || info == nil {
+ return 0
+ }
+ sourceInfo = info
+ return len(info.SegmentInfo)
+ }).WithTimeout(60*time.Second).WithPolling(time.Second).Should(Equal(2),
+ "expected 2 DAY×2 segments on disk after writes are flushed")
+
+ // Capture sourceDataPath + dump the live registry into a synthetic
+ // schema-property bluge catalog BEFORE shutting down the source
+ // (the registry calls in seedMigrationE2ESchemaProperty require a
+ // running service).
+ targetDataPath := filepath.Join(targetRoot, "measure", "data")
+ staging := filepath.Join(workspace, "staging")
+ schemaPropertyRoot := filepath.Join(workspace, "schema-property")
+ seedMigrationE2ESchemaProperty(svcs, schemaPropertyRoot, group, measureName)
+ sourceDataPath := svcs.measure.(interface{ GetDataPath() string }).GetDataPath()
+
+ // Wait until every per-seg bluge sidx dir has a `.snp` snapshot
+ // file. Measure's flushTimeout (5s) drives PersisterNapTime in
+ // the underlying inverted.Store, and the persister exits without
+ // writing the `.snp` if the writer is closed before that nap
+ // elapses — so we poll the disk until persistence is confirmed
+ // before tearing the service down. (Production tolerates the
+ // missing-snapshot case via mergeOneSourceSidxInto, but the test
+ // wants the strict version so the migrated sidx is never empty.)
+ sourceGroupRoot := filepath.Join(sourceDataPath, group)
+ Eventually(func() bool {
+ return allSidxDirsHaveSnapshot(sourceGroupRoot)
+ }).WithTimeout(60*time.Second).WithPolling(time.Second).Should(BeTrue(),
+ "every <seg>/sidx/ under %s should carry a committed .snp before migration runs",
+ sourceGroupRoot)
+
+ // Stop the source service so every tsTable + per-segment
+ // seriesIndex commits its bluge writer to disk. Migration must
+ // run against a quiescent tree (the in-cluster runbook scales
+ // the data StatefulSets to 0 for the same reason) — otherwise
+ // the migrated target's `sidx/` carries no committed snapshot
+ // and the runtime can't resolve series IDs at query time.
+ sourceDown()
+
+ cfg := measure.DirectCopyConfig{
+ SchemaPropertyPath: schemaPropertyRoot,
+ SidxStagingDir: staging,
+ Groups: []string{group},
+ Entries: []measure.DirectCopyEntry{
+ {
+ Stage: measure.StageHot,
+ Target: targetDataPath,
+ Nodes: []string{"node-0"},
+ Source: []string{sourceDataPath},
+ },
+ },
+ }
+ res, err := measure.MigrationCopy(ctx, cfg)
+ Expect(err).NotTo(HaveOccurred())
+ Expect(res.Segments).To(Equal(len(sourceInfo.SegmentInfo)),
+ "copy res should land the same number of segments as the source")
+
+ // File-level inspect: enumerate every target seg and cross-check
+ // startTime against the source CollectDataInfo and endTime
+ // against the on-disk metadata file.
+ targetGroupRoot := filepath.Join(targetDataPath, group)
+ reports, err := measure.EnumerateGroupTarget(targetGroupRoot, twoDayInterval, fs.NewLocalFileSystem())
+ Expect(err).NotTo(HaveOccurred())
+ Expect(reports).To(HaveLen(len(sourceInfo.SegmentInfo)))
+
+ sourceByStart := map[string]*databasev1.SegmentInfo{}
+ for _, si := range sourceInfo.SegmentInfo {
+ sourceByStart[si.TimeRangeStart] = si
+ }
+ for _, r := range reports {
+ Expect(r.Aligned).To(BeTrue(), "seg %s should align to DAY×2 grid", r.Seg)
+ startRFC := r.StartTime.Format(time.RFC3339Nano)
+ si, ok := sourceByStart[startRFC]
+ Expect(ok).To(BeTrue(),
+ "target seg %s startTime %s does not match any source segment (have: %v)",
+ r.Seg, startRFC, keysOf(sourceByStart))
+ Expect(r.Rows).To(BeNumerically(">=", uint64(1)),
+ "seg %s should carry at least one row", r.Seg)
+
+ raw, readErr := os.ReadFile(filepath.Join(targetGroupRoot, r.Seg, storage.SegmentMetadataFilename))
+ Expect(readErr).NotTo(HaveOccurred(), "read %s/metadata", r.Seg)
+ var meta storage.SegmentMetadata
+ Expect(json.Unmarshal(raw, &meta)).To(Succeed())
+ Expect(meta.Version).To(Equal(storage.CurrentSegmentVersion))
+ Expect(meta.EndTime).To(Equal(si.TimeRangeEnd),
+ "seg %s endTime mismatch: target=%s, source=%s",
+ r.Seg, meta.EndTime, si.TimeRangeEnd)
+
+ // Standard alignment invariant: the segment must span exactly
+ // ONE bucket on the IntervalRule grid — start = Standard(start)
+ // AND end = NextTime(start) — which is what banyandb's
+ // segmentController guarantees at runtime. Asserting this on
+ // the migrated target catches grid drift the per-seg
+ // EnumerateGroupTarget Aligned bit alone won't flag.
+ // time.Time comparisons go through BeTemporally rather than
+ // gomega.Equal: Equal uses reflect.DeepEqual which trips on
+ // matching instants that differ only in Location pointer
+ // (time.Local from parseDirectCopySegStart vs time.UTC
+ // from time.Parse), and that mismatch surfaces only on
+ // hosts whose $TZ is UTC at the OS level (CI Linux runners,
+ // not the typical macOS dev box).
+ Expect(twoDayInterval.Standard(r.StartTime)).To(BeTemporally("==", r.StartTime),
+ "seg %s start %s is not aligned to its own DAY×2 standard %s",
+ r.Seg, r.StartTime, twoDayInterval.Standard(r.StartTime))
+ endTime, parseErr := time.Parse(time.RFC3339Nano, meta.EndTime)
+ Expect(parseErr).NotTo(HaveOccurred(), "parse seg %s endTime %q", r.Seg, meta.EndTime)
+ Expect(endTime).To(BeTemporally("==", twoDayInterval.NextTime(r.StartTime)),
+ "seg %s [%s, %s) is not exactly one DAY×2 bucket wide (expected end %s)",
+ r.Seg, r.StartTime, endTime, twoDayInterval.NextTime(r.StartTime))
+ }
+
+ // Service-level inspect: re-open the migrated target from a fresh
+ // measure service (which is what banyandb does after the in-cluster
+ // data-dir swap in MIGRATION.md Part 2 Step 14 → Step 16). The
+ // new service must surface the exact same segment count / time
+ // ranges via CollectDataInfo, otherwise a query against the
+ // migrated cluster would see segment-boundary inconsistencies.
+ svcs2, deferFn2 := setUpMigrationTarget(targetRoot)
+ defer deferFn2()
+ Eventually(func() bool {
+ _, ok := svcs2.measure.LoadGroup("sw_metric")
+ return ok
+ }).WithTimeout(30 * time.Second).Should(BeTrue())
+ registerMigrationE2EGroup(svcs2, group, measureName)
+ Eventually(func() bool {
+ _, ok := svcs2.measure.LoadGroup(group)
+ return ok
+ }).WithTimeout(30 * time.Second).Should(BeTrue())
+ Eventually(func() error {
+ _, err := svcs2.measure.Measure(&commonv1.Metadata{Name: measureName, Group: group})
+ return err
+ }).WithTimeout(30*time.Second).Should(Succeed(),
+ "re-opened service should resolve the measure before the query runs")
+
+ var targetInfo *databasev1.DataInfo
+ Eventually(func() int {
+ info, ierr := svcs2.measure.CollectDataInfo(ctx, group)
+ if ierr != nil || info == nil {
+ return 0
+ }
+ targetInfo = info
+ return len(info.SegmentInfo)
+ }).WithTimeout(30*time.Second).WithPolling(time.Second).Should(
+ Equal(len(sourceInfo.SegmentInfo)),
+ "re-opened service should rediscover the same number of segments")
+
+ // Cross-check every TimeRangeStart / TimeRangeEnd against the
+ // source CollectDataInfo so any segment grid drift fails loudly.
+ targetByStart := map[string]*databasev1.SegmentInfo{}
+ for _, si := range targetInfo.SegmentInfo {
+ targetByStart[si.TimeRangeStart] = si
+ }
+ for start, src := range sourceByStart {
+ tgt, ok := targetByStart[start]
+ Expect(ok).To(BeTrue(),
+ "source segment with TimeRangeStart=%s missing from target CollectDataInfo (have: %v)",
+ start, keysOf(targetByStart))
+ Expect(tgt.TimeRangeEnd).To(Equal(src.TimeRangeEnd),
+ "segment %s: target TimeRangeEnd=%s != source TimeRangeEnd=%s",
+ start, tgt.TimeRangeEnd, src.TimeRangeEnd)
+ }
+
+ // Final check: drive a real measure query through the re-opened
+ // service's pipeline so we exercise the runtime's part-merging /
+ // block-scanning path against the migrated parts. A working
+ // query proves the migrated tree is not just on-disk-shaped but
+ // actually decodable by banyandb's reader.
+ queryBegin := day1.Add(-time.Hour)
+ queryEnd := day2.Add(time.Hour)
+ dps := queryMigrationE2E(svcs2, group, measureName, queryBegin, queryEnd)
+ Expect(dps).To(HaveLen(6),
+ "query against the migrated target should return all 6 written data points")
+ seenEntities := map[string]struct{}{}
+ for _, dp := range dps {
+ for _, tf := range dp.TagFamilies {
+ if tf.Name != defaultTagFamily {
+ continue
+ }
+ for _, tag := range tf.Tags {
+ if tag.Key == "entity_id" {
+ if sv := tag.Value.GetStr(); sv != nil {
+ seenEntities[sv.Value] = struct{}{}
+ }
+ }
+ }
+ }
+ }
+ Expect(seenEntities).To(HaveLen(6),
+ "every entity-id written into the source should appear in the post-migration query (got %v)",
+ seenEntities)
+ })
+})
+
+// queryMigrationE2E issues a measurev1.QueryRequest through the supplied
+// services bundle's local pipeline and returns the resulting data points.
+// Used to prove the migrated tree is queryable end-to-end through the
+// runtime's part-merging / block-scanning path, not just on-disk shaped.
+func queryMigrationE2E(svcs *services, group, measureName string, begin, end time.Time) []*measurev1.DataPoint {
+ req := &measurev1.QueryRequest{
+ Groups: []string{group},
+ Name: measureName,
+ TimeRange: &modelv1.TimeRange{
+ Begin: timestamppb.New(begin),
+ End: timestamppb.New(end),
+ },
+ TagProjection: &modelv1.TagProjection{
+ TagFamilies: []*modelv1.TagProjection_TagFamily{
+ {Name: defaultTagFamily, Tags: []string{"id", "entity_id"}},
+ },
+ },
+ FieldProjection: &measurev1.QueryRequest_FieldProjection{
+ Names: []string{"total", "value"},
+ },
+ }
+ feat, err := svcs.pipeline.Publish(context.Background(), data.TopicMeasureQuery,
+ bus.NewMessage(bus.MessageID(time.Now().UnixNano()), req))
+ Expect(err).NotTo(HaveOccurred())
+ msg, err := feat.Get()
+ Expect(err).NotTo(HaveOccurred())
+ switch d := msg.Data().(type) {
+ case *measurev1.QueryResponse:
+ return d.DataPoints
+ case *common.Error:
+ Fail(fmt.Sprintf("measure query returned error: %s", d.Error()))
+ default:
+ Fail(fmt.Sprintf("expected *measurev1.QueryResponse, got %T", d))
+ }
+ return nil
+}
+
+// allSidxDirsHaveSnapshot returns true when every `<groupRoot>/seg-*/sidx`
+// directory carries at least one `.snp` file (bluge's snapshot marker).
+// Used to wait deterministically for the per-seg seriesIndex persister to
+// fire before MigrationCopy scans the source — see the call site for
+// rationale.
+func allSidxDirsHaveSnapshot(groupRoot string) bool {
+ segs, err := os.ReadDir(groupRoot)
+ if err != nil {
+ return false
+ }
+ saw := 0
+ for _, seg := range segs {
+ if !seg.IsDir() || !strings.HasPrefix(seg.Name(), "seg-") {
+ continue
+ }
+ sidxDir := filepath.Join(groupRoot, seg.Name(), "sidx")
+ entries, readErr := os.ReadDir(sidxDir)
+ if readErr != nil {
+ return false
+ }
+ hasSnap := false
+ for _, e := range entries {
+ if !e.IsDir() && strings.HasSuffix(e.Name(), ".snp") {
+ hasSnap = true
+ break
+ }
+ }
+ if !hasSnap {
+ return false
+ }
+ saw++
+ }
+ return saw > 0
+}
+
+func keysOf(m map[string]*databasev1.SegmentInfo) []string {
+ out := make([]string, 0, len(m))
+ for k := range m {
+ out = append(out, k)
+ }
+ return out
+}
+
+// registerMigrationE2EGroup creates a CATALOG_MEASURE group with
+// SegmentInterval = DAY × 2 and one matching measure under it, using the
+// metadata registry of the supplied services bundle. The same call is
+// made on both the source service and the post-migration re-opened
+// service so each one's schemaRepo discovers the on-disk segments.
+func registerMigrationE2EGroup(svcs *services, group, measureName string) {
+ ctx := context.TODO()
+ _, err := svcs.metadataService.GroupRegistry().CreateGroup(ctx, &commonv1.Group{
+ Metadata: &commonv1.Metadata{Name: group},
+ Catalog: commonv1.Catalog_CATALOG_MEASURE,
+ ResourceOpts: &commonv1.ResourceOpts{
+ ShardNum: 1,
+ SegmentInterval: &commonv1.IntervalRule{Unit: commonv1.IntervalRule_UNIT_DAY, Num: 2},
+ Ttl: &commonv1.IntervalRule{Unit: commonv1.IntervalRule_UNIT_DAY, Num: 30},
+ },
+ })
+ if err != nil && !isAlreadyExistsErr(err) {
+ Expect(err).NotTo(HaveOccurred())
+ }
+ _, err = svcs.metadataService.MeasureRegistry().CreateMeasure(ctx, &databasev1.Measure{
+ Metadata: &commonv1.Metadata{Name: measureName, Group: group},
+ Entity: &databasev1.Entity{TagNames: []string{"entity_id"}},
+ TagFamilies: []*databasev1.TagFamilySpec{
+ {Name: defaultTagFamily, Tags: []*databasev1.TagSpec{
+ {Name: "id", Type: databasev1.TagType_TAG_TYPE_STRING},
+ {Name: "entity_id", Type: databasev1.TagType_TAG_TYPE_STRING},
+ }},
+ },
+ Fields: []*databasev1.FieldSpec{
+ {
+ Name: "total",
+ FieldType: databasev1.FieldType_FIELD_TYPE_INT,
+ CompressionMethod: databasev1.CompressionMethod_COMPRESSION_METHOD_ZSTD,
+ EncodingMethod: databasev1.EncodingMethod_ENCODING_METHOD_GORILLA,
+ },
+ {
+ Name: "value",
+ FieldType: databasev1.FieldType_FIELD_TYPE_INT,
+ CompressionMethod: databasev1.CompressionMethod_COMPRESSION_METHOD_ZSTD,
+ EncodingMethod: databasev1.EncodingMethod_ENCODING_METHOD_GORILLA,
+ },
+ },
+ })
+ if err != nil && !isAlreadyExistsErr(err) {
+ Expect(err).NotTo(HaveOccurred())
+ }
+}
+
+func isAlreadyExistsErr(err error) bool {
+ if err == nil {
+ return false
+ }
+ msg := err.Error()
+ return msg != "" && (containsAny(msg, "already exists", "AlreadyExists"))
+}
+
+func containsAny(s string, needles ...string) bool {
+ for _, n := range needles {
+ if n == "" {
+ continue
+ }
+ for i := 0; i+len(n) <= len(s); i++ {
+ if s[i:i+len(n)] == n {
+ return true
+ }
+ }
+ }
+ return false
+}
+
+// writeMigrationE2EPoints emits `count` data points spaced 1s apart starting
+// at `baseTime`, each carrying a unique entity_id so the writes spread
+// across all configured shards.
+func writeMigrationE2EPoints(svcs *services, group, name string, baseTime time.Time, count, entityOffset int) {
+ bp := svcs.pipeline.NewBatchPublisher(5 * time.Second)
+ defer bp.Close()
+ for i := 0; i < count; i++ {
+ idx := entityOffset + i
+ iStr := strconv.Itoa(idx)
+ req := &measurev1.WriteRequest{
+ Metadata: &commonv1.Metadata{Name: name, Group: group},
+ DataPoint: &measurev1.DataPointValue{
+ Timestamp: timestamppb.New(baseTime.Add(time.Duration(i) * time.Second)),
+ TagFamilies: []*modelv1.TagFamilyForWrite{{
+ Tags: []*modelv1.TagValue{
+ {Value: &modelv1.TagValue_Str{Str: &modelv1.Str{Value: "id-" + iStr}}},
+ {Value: &modelv1.TagValue_Str{Str: &modelv1.Str{Value: "entity-" + iStr}}},
+ },
+ }},
+ Fields: []*modelv1.FieldValue{
+ {Value: &modelv1.FieldValue_Int{Int: &modelv1.Int{Value: int64(100 + idx)}}},
+ {Value: &modelv1.FieldValue_Int{Int: &modelv1.Int{Value: int64(idx)}}},
+ },
+ },
+ }
+ bp.Publish(context.TODO(), data.TopicMeasureWrite,
+ bus.NewMessage(bus.MessageID(idx), &measurev1.InternalWriteRequest{
+ EntityValues: []*modelv1.TagValue{
+ {Value: &modelv1.TagValue_Str{Str: &modelv1.Str{Value: "entity-" + iStr}}},
+ },
+ Request: req,
+ }))
+ }
+}
+
+// seedMigrationE2ESchemaProperty writes a synthetic schema-property bluge
+// catalog at <root>/shard-0/ containing the live registry's Group + Measure
+// docs. The on-disk shape is exactly what walkSchemaPropertyShard reads in
+// production.
+func seedMigrationE2ESchemaProperty(svcs *services, root, group, measureName string) {
+ ctx := context.TODO()
+ grpProto, err := svcs.metadataService.GroupRegistry().GetGroup(ctx, group)
+ Expect(err).NotTo(HaveOccurred())
+ measureProto, err := svcs.metadataService.MeasureRegistry().GetMeasure(ctx,
+ &commonv1.Metadata{Name: measureName, Group: group})
+ Expect(err).NotTo(HaveOccurred())
+
+ shardPath := filepath.Join(root, "shard-0")
+ Expect(os.MkdirAll(shardPath, storage.DirPerm)).To(Succeed())
+ w, err := bluge.OpenWriter(bluge.DefaultConfig(shardPath))
+ Expect(err).NotTo(HaveOccurred())
+ defer func() { Expect(w.Close()).To(Succeed()) }()
+
+ batch := bluge.NewBatch()
+ grpJSON, err := protojson.Marshal(grpProto)
+ Expect(err).NotTo(HaveOccurred())
+ batch.Insert(migrationE2EBlugeDoc("group/"+group, schema.KindGroup.String(), "", string(grpJSON)))
+ measureJSON, err := protojson.Marshal(measureProto)
+ Expect(err).NotTo(HaveOccurred())
+ batch.Insert(migrationE2EBlugeDoc("measure/"+group+"/"+measureName,
+ schema.KindMeasure.String(), group, string(measureJSON)))
+ Expect(w.Batch(batch)).To(Succeed())
+}
+
+// migrationE2EBlugeDoc wraps one inner-proto JSON in a propertyv1.Property
+// and returns a bluge doc whose `_source` field carries the Property JSON.
+func migrationE2EBlugeDoc(id, kind, group, sourceJSON string) *bluge.Document {
+ tags := []*modelv1.Tag{
+ {Key: "source", Value: &modelv1.TagValue{Value: &modelv1.TagValue_Str{Str: &modelv1.Str{Value: sourceJSON}}}},
+ }
+ if group != "" {
+ tags = append(tags, &modelv1.Tag{
+ Key: "group",
+ Value: &modelv1.TagValue{Value: &modelv1.TagValue_Str{Str: &modelv1.Str{Value: group}}},
+ })
+ }
+ prop := &propertyv1.Property{
+ Id: id,
+ Metadata: &commonv1.Metadata{Name: kind, ModRevision: 1},
+ Tags: tags,
+ }
+ propJSON, err := protojson.Marshal(prop)
+ Expect(err).NotTo(HaveOccurred())
+ return bluge.NewDocument(id).AddField(bluge.NewStoredOnlyField("_source", propJSON))
+}
+
+// setUpMigrationTarget mirrors setUp() from measure_suite_test.go but pins
+// the measure root path to an existing on-disk tree (the migration target)
+// so a fresh standalone service can re-open it. The metadata service is
+// fresh so the caller must re-register every group it cares about; the
+// preloaded fixtures (sw_metric et al.) come up automatically as in setUp().
+func setUpMigrationTarget(measureRootPath string) (*services, func()) {
+ pipeline := queue.Local()
+ metadataService, err := service.NewService()
+ Expect(err).NotTo(HaveOccurred())
+ metricSvc := obsservice.NewMetricService(metadataService, pipeline, "test", nil)
+ pm := protector.NewMemory(metricSvc)
+ measureService, err := measure.NewStandalone(metadataService, pipeline, nil, metricSvc, pm)
+ Expect(err).NotTo(HaveOccurred())
+ preloadSvc := &preloadMeasureService{metaSvc: metadataService}
+ querySvc, err := query.NewService(context.TODO(), nil, measureService, nil, metadataService, pipeline, metricSvc)
+ Expect(err).NotTo(HaveOccurred())
+
+ metaPath, metaDeferFunc, err := test.NewSpace()
+ Expect(err).NotTo(HaveOccurred())
+ schemaPorts, err := test.AllocateFreePorts(1)
+ Expect(err).NotTo(HaveOccurred())
+ flags := []string{
+ "--schema-server-root-path=" + metaPath,
+ fmt.Sprintf("--schema-server-grpc-port=%d", schemaPorts[0]),
+ "--schema-server-grpc-host=127.0.0.1",
+ "--measure-root-path=" + measureRootPath,
+ }
+ moduleDeferFunc := test.SetupModules(flags, pipeline, metadataService, preloadSvc, measureService, querySvc)
+ return &services{
+ measure: measureService,
+ metadataService: metadataService,
+ pipeline: pipeline,
+ }, func() {
+ moduleDeferFunc()
+ metaDeferFunc()
+ }
+}
diff --git a/banyand/measure/migration_schema.go b/banyand/measure/migration_schema.go
new file mode 100644
index 000000000..1829b3304
--- /dev/null
+++ b/banyand/measure/migration_schema.go
@@ -0,0 +1,446 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package measure
+
+import (
+ "context"
+ "fmt"
+ "os"
+ "path/filepath"
+ "strings"
+
+ "github.com/blugelabs/bluge"
+ "github.com/pkg/errors"
+ "google.golang.org/protobuf/encoding/protojson"
+
+ commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
+ databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
+ propertyv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/property/v1"
+ backupsnapshot "github.com/apache/skywalking-banyandb/banyand/backup/snapshot"
+ "github.com/apache/skywalking-banyandb/banyand/metadata/schema"
+)
+
+// errSchemaPropertyMissing signals "no schema-property/_schema under
+// backupDir". Returned by findSchemaPropertyRoot when a caller asks for
+// a backup-tree discovery but the snapshot omitted the catalog.
+var errSchemaPropertyMissing = errors.New("schema-property catalog not found in backup")
+
+// schemaPropDoc is one decoded doc emitted by walkSchemaPropertyShard;
+// the caller decides whether the kind / group matches what it wants.
+type schemaPropDoc struct {
+ propID string
+ kindName string // "measure" / "group" / "stream" / ...
+ group string // empty for non-measure docs
+ sourceJSON string // embedded protobuf JSON ready for kind-specific Unmarshal
+ modRev int64
+ deleted bool
+}
+
+// walkSchemaPropertyShard opens one shard of the backup's
+// schema-property bluge index and invokes visit() for each doc.
+// Visitors filter on kindName and fold into their own dedup tables.
+func walkSchemaPropertyShard(shardPath string, visit func(schemaPropDoc) error) error {
+ reader, err := bluge.OpenReader(bluge.DefaultConfig(shardPath))
+ if err != nil {
+ return fmt.Errorf("open bluge reader: %w", err)
+ }
+ defer func() { _ = reader.Close() }()
+ dmi, err := reader.Search(context.Background(),
+ bluge.NewTopNSearch(schemaSearchSize, bluge.NewMatchAllQuery()))
+ if err != nil {
+ return fmt.Errorf("search schema docs: %w", err)
+ }
+ for {
+ next, err := dmi.Next()
+ if err != nil {
+ return fmt.Errorf("iterate schema docs: %w", err)
+ }
+ if next == nil {
+ return nil
+ }
+ var sourceBytes []byte
+ var deleted bool
+ if err := next.VisitStoredFields(func(field string, value []byte) bool {
+ switch field {
+ case schemaSourceField:
+ sourceBytes = append([]byte(nil), value...)
+ case schemaDeletedTag:
+ if len(value) > 0 {
+ deleted = true
+ }
+ }
+ return true
+ }); err != nil {
+ return fmt.Errorf("visit schema doc: %w", err)
+ }
+ if len(sourceBytes) == 0 {
+ continue
+ }
+ var prop propertyv1.Property
+ if err := protojson.Unmarshal(sourceBytes, &prop); err != nil {
+ // Surface the failure with the shard path + a snippet of the
+ // offending JSON so an operator hitting a corrupt catalog can
+ // see WHERE the parse broke instead of just "missing schema"
+ // later in the run.
+ return fmt.Errorf("unmarshal property doc in %s (%d source bytes): %w",
+ shardPath, len(sourceBytes), err)
+ }
+ var group, srcJSON string
+ for _, t := range prop.GetTags() {
+ sv := t.GetValue().GetStr()
+ if sv == nil {
+ continue
+ }
+ switch t.GetKey() {
+ case "group":
+ group = sv.GetValue()
+ case "source":
+ srcJSON = sv.GetValue()
+ }
+ }
+ if vErr := visit(schemaPropDoc{
+ propID: prop.GetId(),
+ kindName: prop.GetMetadata().GetName(),
+ group: group,
+ sourceJSON: srcJSON,
+ modRev: prop.GetMetadata().GetModRevision(),
+ deleted: deleted,
+ }); vErr != nil {
+ return vErr
+ }
+ }
+}
+
+const (
+ schemaSourceField = "_source"
+ schemaDeletedTag = "_deleted"
+ schemaShardPrefix = "shard-"
+ // schemaSearchSize sets the per-page document budget for the bluge
+ // scan; the schema-property index in production tops out at ~10k docs so
+ // a single page is plenty.
+ schemaSearchSize = 200000
+)
+
+// resolveSchemaRoot picks the `_schema` bluge directory to read schemas
+// from. If schemaPropertyPath is non-empty it is honored directly (used
+// by callers that mount a live cluster's schema-property PVC and want
+// to skip the backup-style `<node>/<date>/` discovery). Otherwise the
+// caller falls back to walking the backup snapshot via
+// findSchemaPropertyRoot, restricted to date (when non-empty) so the
+// schemas come from the same snapshot as the migrated measure data.
+func resolveSchemaRoot(backupDir, date, schemaPropertyPath string) (string, error) {
+ if schemaPropertyPath != "" {
+ return schemaPropertyPath, nil
+ }
+ return findSchemaPropertyRoot(backupDir, date)
+}
+
+// loadMeasureSchemasFromSchemaCatalog reads every measure schema under the
+// given groups directly from the schema-property bluge index. Returns
+// (group -> measure-name -> schema). When schemaPropertyPath is set, the
+// catalog is read straight from that bluge dir (the live PVC mount path);
+// otherwise the function walks the backup-style `<node>/<date>/` layout
+// rooted at backupDir.
+//
+// The bluge index retains historical revisions across segments (a schema
+// updated N times leaves N docs sharing the same propID with distinct
+// mod_revisions). This loader mirrors the live cluster's
+// SchemaRegistry.listSchemas dedup: propID is the dedup key, the entry
+// with the highest mod_revision wins, and the winning entry is dropped
+// entirely if it carries a non-zero _deleted marker (the schema was
+// tombstoned at that revision).
+func loadMeasureSchemasFromSchemaCatalog(backupDir, date, schemaPropertyPath string, groups []string) (map[string]map[string]*measureSchemaInfo, error) {
+ byGroup, err := fetchMeasureSchemasFromSchema(backupDir, date, schemaPropertyPath, groups)
+ if err != nil {
+ return nil, err
+ }
+ out := make(map[string]map[string]*measureSchemaInfo, len(groups))
+ for _, group := range groups {
+ list := byGroup[group]
+ byName := make(map[string]*measureSchemaInfo, len(list))
+ for _, s := range list {
+ byName[s.Name] = s
+ }
+ out[group] = byName
+ }
+ return out, nil
+}
+
+func fetchMeasureSchemasFromSchema(backupDir, date, schemaPropertyPath string, groups []string) (map[string][]*measureSchemaInfo, error) {
+ if backupDir == "" && schemaPropertyPath == "" {
+ return nil, errors.New("either backup-dir or schema-property-path is required")
+ }
+ wanted := make(map[string]bool, len(groups))
+ for _, g := range groups {
+ wanted[g] = true
+ }
+ schemaRoot, err := resolveSchemaRoot(backupDir, date, schemaPropertyPath)
+ if err != nil {
+ return nil, err
+ }
+ shards, err := os.ReadDir(schemaRoot)
+ if err != nil {
+ return nil, fmt.Errorf("read schema-property root %q: %w", schemaRoot, err)
+ }
+ candidates := make(map[string]*schemaCandidate)
+ for _, sh := range shards {
+ if !sh.IsDir() || !strings.HasPrefix(sh.Name(), schemaShardPrefix) {
+ continue
+ }
+ shardPath := filepath.Join(schemaRoot, sh.Name())
+ if loadErr := loadMeasureSchemasFromShard(shardPath, wanted, candidates); loadErr != nil {
+ return nil, fmt.Errorf("load schemas from %s: %w", shardPath, loadErr)
+ }
+ }
+ result := make(map[string][]*measureSchemaInfo)
+ for _, c := range candidates {
+ if c.deleted || c.info == nil {
+ continue
+ }
+ result[c.info.Group] = append(result[c.info.Group], c.info)
+ }
+ return result, nil
+}
+
+// schemaCandidate captures the best (highest mod_revision) doc seen for
+// one propID while iterating the schema-property bluge index. deleted is
+// true when the winning revision is a tombstone, in which case the entry
+// must be skipped entirely (an older live revision must not resurrect a
+// deleted schema).
+type schemaCandidate struct {
+ info *measureSchemaInfo
+ modRev int64
+ deleted bool
+}
+
+// findSchemaPropertyRoot walks <backup>/<node>/<date>/ and returns the
+// first schema-property/_schema directory it finds. Hot/schema-server
+// pods are the only ones whose backup carries this catalog.
+//
+// When date is non-empty, only `<node>/<date>/...` subtrees are
+// considered — this is the typical case where the caller migrates a
+// specific snapshot and must not silently pick up schemas from a
+// different (older/newer) date that happens to live under the same
+// backup root.
+func findSchemaPropertyRoot(backupDir, date string) (string, error) {
+ nodes, err := os.ReadDir(backupDir)
+ if err != nil {
+ return "", fmt.Errorf("read backup dir %q: %w", backupDir, err)
+ }
+ tryDate := func(nodeRoot, d string) (string, bool) {
+ p := filepath.Join(nodeRoot, d, backupsnapshot.SchemaPropertyCatalogName, schema.SchemaGroup)
+ info, statErr := os.Stat(p)
+ return p, statErr == nil && info.IsDir()
+ }
+ for _, node := range nodes {
+ if !node.IsDir() {
+ continue
+ }
+ nodeRoot := filepath.Join(backupDir, node.Name())
+ candidates := []string{date}
+ if date == "" {
+ dates, readErr := os.ReadDir(nodeRoot)
+ if readErr != nil {
+ continue
+ }
+ candidates = candidates[:0]
+ for _, d := range dates {
+ if d.IsDir() {
+ candidates = append(candidates, d.Name())
+ }
+ }
+ }
+ for _, c := range candidates {
+ if p, ok := tryDate(nodeRoot, c); ok {
+ return p, nil
+ }
+ }
+ }
+ dateClause := ""
+ if date != "" {
+ dateClause = fmt.Sprintf(" for date %q", date)
+ }
+ return "", errors.WithMessagef(errSchemaPropertyMissing,
+ "no %s/%s directory found under %q%s (only hot/schema-server node backups carry schemas)",
+ backupsnapshot.SchemaPropertyCatalogName, schema.SchemaGroup, backupDir, dateClause)
+}
+
+// loadGroupResourceOptsFromSchema scans the schema-property bluge index
+// for commonv1.Group docs and returns each requested group's
+// ResourceOpts (carrying SegmentInterval + Stages). Missing groups are
+// simply absent from the result map.
+func loadGroupResourceOptsFromSchema(backupDir, date, schemaPropertyPath string, groups []string) (map[string]*commonv1.ResourceOpts, error) {
+ if backupDir == "" && schemaPropertyPath == "" {
+ return nil, errors.New("either backup-dir or schema-property-path is required")
+ }
+ wanted := make(map[string]bool, len(groups))
+ for _, g := range groups {
+ wanted[g] = true
+ }
+ schemaRoot, err := resolveSchemaRoot(backupDir, date, schemaPropertyPath)
+ if err != nil {
+ return nil, err
+ }
+ shards, err := os.ReadDir(schemaRoot)
+ if err != nil {
+ return nil, fmt.Errorf("read schema-property root %q: %w", schemaRoot, err)
+ }
+ candidates := map[string]*groupOptsCandidate{}
+ for _, sh := range shards {
+ if !sh.IsDir() || !strings.HasPrefix(sh.Name(), schemaShardPrefix) {
+ continue
+ }
+ if scanErr := loadGroupResourceOptsFromShard(filepath.Join(schemaRoot, sh.Name()), wanted, candidates); scanErr != nil {
+ return nil, scanErr
+ }
+ }
+ out := map[string]*commonv1.ResourceOpts{}
+ for _, c := range candidates {
+ if c.deleted || c.opts == nil {
+ continue
+ }
+ out[c.name] = c.opts
+ }
+ return out, nil
+}
+
+// groupOptsCandidate is the analog of schemaCandidate for group
+// ResourceOpts: tracks the best mod_revision per propID and whether
+// the winning revision is a tombstone.
+type groupOptsCandidate struct {
+ opts *commonv1.ResourceOpts
+ name string
+ modRev int64
+ deleted bool
+}
+
+func loadGroupResourceOptsFromShard(
+ shardPath string,
+ wanted map[string]bool,
+ candidates map[string]*groupOptsCandidate,
+) error {
+ return walkSchemaPropertyShard(shardPath, func(d schemaPropDoc) error {
+ if d.kindName != schema.KindGroup.String() || d.sourceJSON == "" {
+ return nil
+ }
+ var g commonv1.Group
+ if uErr := protojson.Unmarshal([]byte(d.sourceJSON), &g); uErr != nil {
+ return nil
+ }
+ name := g.GetMetadata().GetName()
+ if !wanted[name] {
+ return nil
+ }
+ if cur, ok := candidates[d.propID]; ok && cur.modRev >= d.modRev {
+ return nil
+ }
+ candidates[d.propID] = &groupOptsCandidate{
+ opts: g.GetResourceOpts(),
+ name: name,
+ modRev: d.modRev,
+ deleted: d.deleted,
+ }
+ return nil
+ })
+}
+
+// loadMeasureSchemasFromShard folds every MEASURE doc in one shard
+// into the candidates map. Updates win by mod_revision; tombstones
+// drop the propID later.
+func loadMeasureSchemasFromShard(shardPath string, wantedGroups map[string]bool,
+ candidates map[string]*schemaCandidate,
+) error {
+ return walkSchemaPropertyShard(shardPath, func(d schemaPropDoc) error {
+ if d.kindName != schema.KindMeasure.String() {
+ return nil
+ }
+ if len(wantedGroups) > 0 && !wantedGroups[d.group] {
+ return nil
+ }
+ if d.sourceJSON == "" {
+ return fmt.Errorf("schema property %q missing source tag", d.propID)
+ }
+ var measure databasev1.Measure
+ if uErr := protojson.Unmarshal([]byte(d.sourceJSON), &measure); uErr != nil {
+ return fmt.Errorf("unmarshal measure %q source: %w", d.propID, uErr)
+ }
+ if existing, ok := candidates[d.propID]; ok && existing.modRev >= d.modRev {
+ return nil
+ }
+ candidates[d.propID] = &schemaCandidate{
+ info: measureSchemaInfoFromProto(d.group, &measure),
+ modRev: d.modRev,
+ deleted: d.deleted,
+ }
+ return nil
+ })
+}
+
+// measureSchemaInfoFromProto builds the lookup-table view of a measure
+// schema directly from its proto definition.
+func measureSchemaInfoFromProto(group string, m *databasev1.Measure) *measureSchemaInfo {
+ si := &measureSchemaInfo{
+ Group: group,
+ Name: m.GetMetadata().GetName(),
+ EntityTagNames: append([]string(nil), m.GetEntity().GetTagNames()...),
+ IndexMode: m.GetIndexMode(),
+ }
+ for _, tf := range m.GetTagFamilies() {
+ tags := make([]string, 0, len(tf.GetTags()))
+ for _, t := range tf.GetTags() {
+ tags = append(tags, t.GetName())
+ for _, en := range si.EntityTagNames {
+ if t.GetName() == en {
+ si.EntityFamily = tf.GetName()
+ }
+ }
+ }
+ si.TagFamilies = append(si.TagFamilies, tagFamilyInfo{Name: tf.GetName(), Tags: tags})
+ }
+ for _, f := range m.GetFields() {
+ si.Fields = append(si.Fields, fieldInfo{
+ Name: f.GetName(),
+ FieldType: f.GetFieldType(),
+ })
+ }
+ return si
+}
+
+// measureSchemaInfo is the subset of a measure's schema MigrationCopy
+// cares about: enough to drive tag-family projection and reject IndexMode
+// groups (their field values live inside sidx, not in part data, and
+// broadcasting a union sidx would break query correctness).
+type measureSchemaInfo struct {
+ Group string
+ Name string
+ EntityTagNames []string
+ EntityFamily string
+ TagFamilies []tagFamilyInfo
+ Fields []fieldInfo
+ IndexMode bool
+}
+
+type tagFamilyInfo struct {
+ Name string
+ Tags []string
+}
+
+type fieldInfo struct {
+ Name string
+ FieldType databasev1.FieldType
+}
diff --git a/banyand/measure/migration_schema_test.go b/banyand/measure/migration_schema_test.go
new file mode 100644
index 000000000..b4a11bfb4
--- /dev/null
+++ b/banyand/measure/migration_schema_test.go
@@ -0,0 +1,308 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package measure
+
+import (
+ "fmt"
+ "os"
+ "path/filepath"
+ "testing"
+
+ "github.com/blugelabs/bluge"
+ "google.golang.org/protobuf/encoding/protojson"
+
+ commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
+ databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
+ modelv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/model/v1"
+ propertyv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/property/v1"
+ backupsnapshot "github.com/apache/skywalking-banyandb/banyand/backup/snapshot"
+ "github.com/apache/skywalking-banyandb/banyand/internal/storage"
+ "github.com/apache/skywalking-banyandb/banyand/metadata/schema"
+)
+
+// synthMeasure describes one measure to seed into the synthetic
+// schema-property catalog.
+type synthMeasure struct {
+ group string
+ name string
+ modRev int64
+}
+
+// synthGroup describes one group to seed into the synthetic
+// schema-property catalog. Opts is optional; nil yields a bare group
+// metadata doc (sufficient for the measure-fetch tests).
+type synthGroup struct {
+ opts *commonv1.ResourceOpts
+ name string
+}
+
+// synthBackup builds a minimal backup tree at root with one
+// schema-property shard carrying the given groups and measures so the
+// fetch/load helpers can exercise the full bluge → proto → info path
+// without any real cluster snapshot.
+func synthBackup(t *testing.T, root string, groups []synthGroup, measures []synthMeasure) {
+ t.Helper()
+ shardPath := filepath.Join(root, "node-0", "2026-05-21",
+ backupsnapshot.SchemaPropertyCatalogName, schema.SchemaGroup, "shard-0")
+ if err := os.MkdirAll(shardPath, storage.DirPerm); err != nil {
+ t.Fatalf("mkdir shard: %v", err)
+ }
+ w, err := bluge.OpenWriter(bluge.DefaultConfig(shardPath))
+ if err != nil {
+ t.Fatalf("open bluge writer: %v", err)
+ }
+ defer func() {
+ if cErr := w.Close(); cErr != nil {
+ t.Fatalf("close bluge writer: %v", cErr)
+ }
+ }()
+
+ batch := bluge.NewBatch()
+ for _, g := range groups {
+ grp := &commonv1.Group{
+ Metadata: &commonv1.Metadata{Name: g.name, ModRevision: 1},
+ ResourceOpts: g.opts,
+ }
+ grpJSON, mErr := protojson.Marshal(grp)
+ if mErr != nil {
+ t.Fatalf("marshal group %q: %v", g.name, mErr)
+ }
+ batch.Insert(synthSchemaDoc("group/"+g.name, schema.KindGroup.String(), "", 1, string(grpJSON)))
+ }
+ for _, m := range measures {
+ measure := &databasev1.Measure{
+ Metadata: &commonv1.Metadata{Name: m.name, Group: m.group, ModRevision: m.modRev},
+ Entity: &databasev1.Entity{TagNames: []string{"entity_id"}},
+ TagFamilies: []*databasev1.TagFamilySpec{
+ {Name: "default", Tags: []*databasev1.TagSpec{
+ {Name: "entity_id", Type: databasev1.TagType_TAG_TYPE_STRING},
+ {Name: "label", Type: databasev1.TagType_TAG_TYPE_STRING},
+ }},
+ },
+ Fields: []*databasev1.FieldSpec{
+ {Name: "total", FieldType: databasev1.FieldType_FIELD_TYPE_INT},
+ },
+ }
+ mJSON, mErr := protojson.Marshal(measure)
+ if mErr != nil {
+ t.Fatalf("marshal measure %q/%q: %v", m.group, m.name, mErr)
+ }
+ docID := fmt.Sprintf("measure/%s/%s/rev-%d", m.group, m.name, m.modRev)
+ // Use a stable propID (no revision suffix) so multiple revisions
+ // of the same measure share one candidate slot at fetch time.
+ propID := fmt.Sprintf("measure/%s/%s", m.group, m.name)
+ batch.Insert(synthSchemaDocWithPropID(docID, propID,
+ schema.KindMeasure.String(), m.group, m.modRev, string(mJSON)))
+ }
+ if bErr := w.Batch(batch); bErr != nil {
+ t.Fatalf("write batch: %v", bErr)
+ }
+}
+
+// synthSchemaDoc builds a single bluge doc whose `_source` field carries
+// the inner-proto JSON wrapped in a propertyv1.Property, matching the
+// shape that schema-property bluge indexes use in production.
+func synthSchemaDoc(propID, kind, group string, modRev int64, sourceJSON string) *bluge.Document {
+ return synthSchemaDocWithPropID(propID, propID, kind, group, modRev, sourceJSON)
+}
+
+func synthSchemaDocWithPropID(docID, propID, kind, group string, modRev int64, sourceJSON string) *bluge.Document {
+ tags := []*modelv1.Tag{
+ {Key: "source", Value: &modelv1.TagValue{Value: &modelv1.TagValue_Str{Str: &modelv1.Str{Value: sourceJSON}}}},
+ }
+ if group != "" {
+ tags = append(tags, &modelv1.Tag{
+ Key: "group",
+ Value: &modelv1.TagValue{Value: &modelv1.TagValue_Str{Str: &modelv1.Str{Value: group}}},
+ })
+ }
+ prop := &propertyv1.Property{
+ Id: propID,
+ Metadata: &commonv1.Metadata{Name: kind, ModRevision: modRev},
+ Tags: tags,
+ }
+ propJSON, err := protojson.Marshal(prop)
+ if err != nil {
+ panic(fmt.Sprintf("marshal property %q: %v", propID, err))
+ }
+ return bluge.NewDocument(docID).AddField(bluge.NewStoredOnlyField("_source", propJSON))
+}
+
+// synthGroupsFromNames is a shorthand for callers that don't care
+// about ResourceOpts.
+func synthGroupsFromNames(names ...string) []synthGroup {
+ out := make([]synthGroup, len(names))
+ for i, n := range names {
+ out[i] = synthGroup{name: n}
+ }
+ return out
+}
+
+func TestFetchMeasureSchemasFromSchema_returnsRequestedGroups(t *testing.T) {
+ dir := t.TempDir()
+ groups := []string{"g_a", "g_b", "g_c"}
+ measures := []synthMeasure{
+ {group: "g_a", name: "m_a1", modRev: 1},
+ {group: "g_a", name: "m_a2", modRev: 1},
+ {group: "g_b", name: "m_b1", modRev: 1},
+ {group: "g_c", name: "m_c1", modRev: 1},
+ {group: "g_d", name: "m_d1", modRev: 1}, // outside the requested set
+ }
+ synthBackup(t, dir, synthGroupsFromNames(append(groups, "g_d")...), measures)
+
+ byGroup, err := fetchMeasureSchemasFromSchema(dir, "", "", groups)
+ if err != nil {
+ t.Fatalf("fetchMeasureSchemasFromSchema: %v", err)
+ }
+ for _, g := range groups {
+ if len(byGroup[g]) == 0 {
+ t.Fatalf("group %q: expected schemas, got 0", g)
+ }
+ }
+ for g := range byGroup {
+ found := false
+ for _, want := range groups {
+ if g == want {
+ found = true
+ break
+ }
+ }
+ if !found {
+ t.Fatalf("group %q returned but not requested", g)
+ }
+ }
+}
+
+func TestFetchMeasureSchemasFromSchema_schemaFieldsArePopulated(t *testing.T) {
+ dir := t.TempDir()
+ synthBackup(t, dir, synthGroupsFromNames("g_a"), []synthMeasure{
+ {group: "g_a", name: "m_a1", modRev: 1},
+ })
+
+ byGroup, err := fetchMeasureSchemasFromSchema(dir, "", "", []string{"g_a"})
+ if err != nil {
+ t.Fatalf("fetchMeasureSchemasFromSchema: %v", err)
+ }
+ list := byGroup["g_a"]
+ if len(list) == 0 {
+ t.Fatal("expected at least one measure schema in g_a")
+ }
+ for _, s := range list {
+ if s.Name == "" {
+ t.Fatalf("measure with empty name (Group=%q)", s.Group)
+ }
+ if s.Group != "g_a" {
+ t.Fatalf("measure %q: wrong group %q", s.Name, s.Group)
+ }
+ if len(s.EntityTagNames) == 0 {
+ t.Fatalf("measure %q: empty entity tag names", s.Name)
+ }
+ if len(s.TagFamilies) == 0 {
+ t.Fatalf("measure %q: no tag families", s.Name)
+ }
+ if s.EntityFamily == "" {
+ t.Fatalf("measure %q: missing entity family", s.Name)
+ }
+ }
+}
+
+func TestFetchMeasureSchemasFromSchema_unknownGroupReturnsEmpty(t *testing.T) {
+ dir := t.TempDir()
+ synthBackup(t, dir, synthGroupsFromNames("g_a"), []synthMeasure{
+ {group: "g_a", name: "m_a1", modRev: 1},
+ })
+
+ byGroup, err := fetchMeasureSchemasFromSchema(dir, "", "", []string{"this_group_does_not_exist"})
+ if err != nil {
+ t.Fatalf("fetchMeasureSchemasFromSchema: %v", err)
+ }
+ if got := len(byGroup["this_group_does_not_exist"]); got != 0 {
+ t.Fatalf("expected 0 schemas for unknown group, got %d", got)
+ }
+}
+
+func TestFetchMeasureSchemasFromSchema_missingBackupDir(t *testing.T) {
+ _, err := fetchMeasureSchemasFromSchema("/no/such/dir/should/exist", "", "", []string{"g_a"})
+ if err == nil {
+ t.Fatal("expected error for nonexistent backup-dir, got nil")
+ }
+}
+
+// TestFetchMeasureSchemasFromSchema_dedupesStaleRevisions asserts that the
+// loader collapses historical revisions of the same schema (different
+// mod_revisions on the same propID, retained across bluge segments) into a
+// single entry. The live cluster's SchemaRegistry already dedupes by
+// propID, and the backup loader must match that contract.
+func TestFetchMeasureSchemasFromSchema_dedupesStaleRevisions(t *testing.T) {
+ dir := t.TempDir()
+ // Same (group, name) appears at three different mod_revisions; the
+ // loader must keep only the highest one.
+ synthBackup(t, dir, synthGroupsFromNames("g_a"), []synthMeasure{
+ {group: "g_a", name: "m_a1", modRev: 1},
+ {group: "g_a", name: "m_a1", modRev: 2},
+ {group: "g_a", name: "m_a1", modRev: 3},
+ })
+
+ byGroup, err := fetchMeasureSchemasFromSchema(dir, "", "", []string{"g_a"})
+ if err != nil {
+ t.Fatalf("fetchMeasureSchemasFromSchema: %v", err)
+ }
+ seen := map[string]int{}
+ for _, s := range byGroup["g_a"] {
+ seen[s.Name]++
+ }
+ for name, count := range seen {
+ if count > 1 {
+ t.Fatalf("group g_a: measure %q appears %d times (expected 1)", name, count)
+ }
+ }
+}
+
+// TestLoadGroupResourceOptsFromSchema_returnsRequestedGroups asserts the
+// group ResourceOpts loader resolves a backup catalog through the same
+// resolveSchemaRoot path as the measure loader and carries SegmentInterval
+// from the seeded Group proto into the returned map.
+func TestLoadGroupResourceOptsFromSchema_returnsRequestedGroups(t *testing.T) {
+ dir := t.TempDir()
+ hourly := &commonv1.ResourceOpts{
+ SegmentInterval: &commonv1.IntervalRule{Unit: commonv1.IntervalRule_UNIT_HOUR, Num: 4},
+ }
+ daily := &commonv1.ResourceOpts{
+ SegmentInterval: &commonv1.IntervalRule{Unit: commonv1.IntervalRule_UNIT_DAY, Num: 1},
+ }
+ synthBackup(t, dir, []synthGroup{
+ {name: "g_hourly", opts: hourly},
+ {name: "g_daily", opts: daily},
+ {name: "g_skipped", opts: hourly}, // present in catalog, not requested
+ }, nil)
+
+ got, err := loadGroupResourceOptsFromSchema(dir, "", "",
+ []string{"g_hourly", "g_daily"})
+ if err != nil {
+ t.Fatalf("loadGroupResourceOptsFromSchema: %v", err)
+ }
+ if _, ok := got["g_skipped"]; ok {
+ t.Fatalf("non-requested group g_skipped leaked into result")
+ }
+ if si := got["g_hourly"].GetSegmentInterval(); si.GetUnit() != commonv1.IntervalRule_UNIT_HOUR || si.GetNum() != 4 {
+ t.Fatalf("g_hourly: want HOUR×4, got %v×%d", si.GetUnit(), si.GetNum())
+ }
+ if si := got["g_daily"].GetSegmentInterval(); si.GetUnit() != commonv1.IntervalRule_UNIT_DAY || si.GetNum() != 1 {
+ t.Fatalf("g_daily: want DAY×1, got %v×%d", si.GetUnit(), si.GetNum())
+ }
+}
diff --git a/banyand/measure/migration_verify.go b/banyand/measure/migration_verify.go
new file mode 100644
index 000000000..6e2a5a680
--- /dev/null
+++ b/banyand/measure/migration_verify.go
@@ -0,0 +1,401 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package measure
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "os"
+ "path/filepath"
+ "sort"
+ "strconv"
+ "strings"
+ "time"
+
+ "github.com/blugelabs/bluge"
+
+ "github.com/apache/skywalking-banyandb/banyand/internal/storage"
+ "github.com/apache/skywalking-banyandb/pkg/fs"
+)
+
+// SegmentReport summarizes one `seg-*` directory under a measure
+// group root. Populated by EnumerateGroupTarget for the migration
+// verify CLI.
+//
+// Aligned is true only when ALL of these hold:
+// - the dir name parses cleanly into a start time;
+// - start = IntervalRule.Standard(start) (start is on the grid);
+// - <seg>/metadata exists, is well-formed JSON, and carries a non-empty endTime;
+// - end = IntervalRule.NextTime(start) (the segment spans exactly one bucket);
+// - the inclusive last instant (end - 1ns) standardizes back to start (start
+// and end fall in the same IntervalRule bucket).
+type SegmentReport struct {
+ StartTime time.Time
+ EndTime time.Time
+ Seg string
+ Rows uint64
+ SidxDocCount uint64
+ Shards int
+ Parts int
+ Aligned bool
+ SidxOpened bool
+}
+
+// EntryGroupReport aggregates source row count + target per-seg report
+// for one (entry, group) pair. Both source and target are read-only;
+// the verify command never mutates the dataset.
+type EntryGroupReport struct {
+ Group string
+ EntryStage string
+ EntryTarget string
+ TargetGroup string
+ EntryNodes []string
+ SrcRoots []string
+ TargetSegs []SegmentReport
+ SrcRows uint64
+ SrcParts int
+}
+
+// VerifyShardParts reads <shardDir>'s newest `.snp`, confirms every
+// listed partID has an on-disk dir, opens each part, and returns the
+// sum of partMetadata.TotalCount plus the part count.
+//
+// Errors (instead of t.Fatalf) so this helper is usable from a CLI.
+//
+// fileSystem is passed through to mustOpenFilePart. Directory listing
+// and .snp reads go through the os package directly: fs.FileSystem's
+// ReadDir panics on missing directories, and verify is read-only, so
+// the abstraction is intentionally bypassed for the enumeration path.
+func VerifyShardParts(shardDir string, fileSystem fs.FileSystem) (uint64, int, error) {
+ entries, err := os.ReadDir(shardDir)
+ if err != nil {
+ return 0, 0, fmt.Errorf("read shard: %w", err)
+ }
+ var snpPath string
+ for _, e := range entries {
+ if !e.IsDir() && strings.HasSuffix(e.Name(), directCopySnpSuffix) {
+ candidate := filepath.Join(shardDir, e.Name())
+ if snpPath == "" || candidate > snpPath {
+ snpPath = candidate
+ }
+ }
+ }
+ if snpPath == "" {
+ return 0, 0, fmt.Errorf("no .snp file under %s", shardDir)
+ }
+ snpRaw, err := os.ReadFile(snpPath)
+ if err != nil {
+ return 0, 0, fmt.Errorf("read .snp: %w", err)
+ }
+ var partNames []string
+ if err := json.Unmarshal(snpRaw, &partNames); err != nil {
+ return 0, 0, fmt.Errorf("parse .snp: %w", err)
+ }
+
+ onDiskPartIDs := map[string]bool{}
+ for _, e := range entries {
+ if e.IsDir() && directCopyPartDirPattern.MatchString(e.Name()) {
+ onDiskPartIDs[e.Name()] = true
+ }
+ }
+ for _, name := range partNames {
+ if !onDiskPartIDs[name] {
+ return 0, 0, fmt.Errorf("snp references missing partID %s", name)
+ }
+ }
+
+ var rowsTotal uint64
+ for _, name := range partNames {
+ partID, parseErr := strconv.ParseUint(name, 16, 64)
+ if parseErr != nil {
+ return 0, 0, fmt.Errorf("parse partID %q: %w", name, parseErr)
+ }
+ var rows uint64
+ var openErr error
+ func() {
+ defer func() {
+ if r := recover(); r != nil {
+ openErr = fmt.Errorf("open part %s panicked: %v", name, r)
+ }
+ }()
+ p := mustOpenFilePart(partID, shardDir, fileSystem)
+ defer p.close()
+ rows = p.partMetadata.TotalCount
+ }()
+ if openErr != nil {
+ return 0, 0, openErr
+ }
+ rowsTotal += rows
+ }
+ return rowsTotal, len(partNames), nil
+}
+
+// CountBlugeDocs opens the bluge index at path read-only and returns the
+// total document count. Used to spot-check the broadcast union sidx
+// that `migration copy` writes into every aligned target segment.
+func CountBlugeDocs(path string) (uint64, error) {
+ reader, err := bluge.OpenReader(bluge.DefaultConfig(path))
+ if err != nil {
+ return 0, err
+ }
+ defer func() { _ = reader.Close() }()
+ // Reader.Count sums per-segment doc counts in O(numSegments), not
+ // O(numDocs) — important for `verify` because the union sidx is
+ // broadcast into every aligned target seg of a group and we call
+ // CountBlugeDocs once per seg. Iterating dmi.Next() instead would
+ // scale with total docs × broadcast fan-out and dominate verify
+ // runtime on production-sized datasets.
+ return reader.Count()
+}
+
+// EnumerateGroupTarget walks <groupRoot>/seg-* and reports per-segment
+// row total + sidx doc count + whether the segment's start time aligns
+// to the supplied IntervalRule's standard grid. A missing groupRoot
+// returns (nil, nil) — the caller decides whether that is a no-op or
+// an error (e.g. when the PVC didn't own the group, "no segs" is
+// expected, not a fault).
+func EnumerateGroupTarget(groupRoot string, intervalRule storage.IntervalRule, fileSystem fs.FileSystem) ([]SegmentReport, error) {
+ entries, err := os.ReadDir(groupRoot)
+ if err != nil {
+ if errors.Is(err, os.ErrNotExist) {
+ return nil, nil
+ }
+ return nil, fmt.Errorf("read group root %s: %w", groupRoot, err)
+ }
+ var reports []SegmentReport
+ for _, e := range entries {
+ if !e.IsDir() || !strings.HasPrefix(e.Name(), directCopySegPrefix) {
+ continue
+ }
+ segDir := filepath.Join(groupRoot, e.Name())
+ report := SegmentReport{Seg: e.Name()}
+
+ var (
+ startTime time.Time
+ endTime time.Time
+ haveStart bool
+ haveEnd bool
+ )
+ if start, parseErr := parseDirectCopySegStart(e.Name(), intervalRule.Unit); parseErr == nil {
+ startTime = start
+ haveStart = true
+ report.StartTime = start
+ }
+
+ // Read <seg>/metadata: the migrated tree's runtime contract is that
+ // every segment carries a JSON-encoded SegmentMetadata with a
+ // non-empty endTime (RFC3339Nano) — both writer paths
+ // (writeDirectCopySegmentMetadata and the live segmentController)
+ // emit it. A missing / unparseable file or a missing endTime keeps
+ // Aligned false so the verify CLI surfaces it loudly.
+ metaPath := filepath.Join(segDir, storage.SegmentMetadataFilename)
+ if raw, readErr := os.ReadFile(metaPath); readErr == nil {
+ var meta storage.SegmentMetadata
+ if jsonErr := json.Unmarshal(raw, &meta); jsonErr == nil && meta.EndTime != "" {
+ if parsed, parseErr := time.Parse(time.RFC3339Nano, meta.EndTime); parseErr == nil {
+ endTime = parsed
+ haveEnd = true
+ report.EndTime = parsed
+ }
+ }
+ }
+
+ if haveStart && haveEnd {
+ startOnGrid := intervalRule.Standard(startTime).Equal(startTime)
+ endIsNext := intervalRule.NextTime(startTime).Equal(endTime)
+ sameBucket := intervalRule.Standard(endTime.Add(-time.Nanosecond)).Equal(startTime)
+ report.Aligned = startOnGrid && endIsNext && sameBucket
+ }
+
+ shardEntries, readErr := os.ReadDir(segDir)
+ if readErr != nil {
+ return nil, fmt.Errorf("read seg %s: %w", segDir, readErr)
+ }
+ for _, sh := range shardEntries {
+ if !sh.IsDir() || !strings.HasPrefix(sh.Name(), directCopyShardPrefix) {
+ continue
+ }
+ report.Shards++
+ shardDir := filepath.Join(segDir, sh.Name())
+ rows, parts, partsErr := VerifyShardParts(shardDir, fileSystem)
+ if partsErr != nil {
+ return nil, fmt.Errorf("shard %s: %w", shardDir, partsErr)
+ }
+ report.Rows += rows
+ report.Parts += parts
+ }
+
+ sidxDir := filepath.Join(segDir, directCopySidxDirName)
+ if info, statErr := os.Stat(sidxDir); statErr == nil && info.IsDir() {
+ count, sidxErr := CountBlugeDocs(sidxDir)
+ if sidxErr != nil {
+ return nil, fmt.Errorf("seg %s sidx open: %w", e.Name(), sidxErr)
+ }
+ report.SidxOpened = true
+ report.SidxDocCount = count
+ }
+ reports = append(reports, report)
+ }
+ sort.Slice(reports, func(i, j int) bool { return reports[i].Seg < reports[j].Seg })
+ return reports, nil
+}
+
+// SumGroupSourceRows walks every <root>/seg-*/shard-*/<partID>/ across
+// the given source roots, opens each part read-only, and returns the
+// total row count plus the part count. Missing roots are silently
+// skipped — the caller has already filtered them via
+// resolveEntrySrcRoots and only existing dirs end up here in practice.
+func SumGroupSourceRows(srcRoots []string, fileSystem fs.FileSystem) (uint64, int, error) {
+ var totalRows uint64
+ var totalParts int
+ for _, root := range srcRoots {
+ segEntries, err := os.ReadDir(root)
+ if err != nil {
+ if errors.Is(err, os.ErrNotExist) {
+ continue
+ }
+ return 0, 0, fmt.Errorf("read src root %s: %w", root, err)
+ }
+ for _, se := range segEntries {
+ if !se.IsDir() || !strings.HasPrefix(se.Name(), directCopySegPrefix) {
+ continue
+ }
+ segDir := filepath.Join(root, se.Name())
+ shardEntries, readErr := os.ReadDir(segDir)
+ if readErr != nil {
+ return 0, 0, fmt.Errorf("read src seg %s: %w", segDir, readErr)
+ }
+ for _, sh := range shardEntries {
+ if !sh.IsDir() || !strings.HasPrefix(sh.Name(), directCopyShardPrefix) {
+ continue
+ }
+ shardDir := filepath.Join(segDir, sh.Name())
+ rows, parts, partsErr := VerifyShardParts(shardDir, fileSystem)
+ if partsErr != nil {
+ return 0, 0, fmt.Errorf("src shard %s: %w", shardDir, partsErr)
+ }
+ totalRows += rows
+ totalParts += parts
+ }
+ }
+ }
+ return totalRows, totalParts, nil
+}
+
+// MigrationVerify drives a `migration verify` run end-to-end against a
+// DirectCopyConfig. For every (entry, group) it:
+//
+// 1. Resolves the source roots via the same plan→cfg helper that
+// `copy` uses (so verify always sees what copy saw).
+// 2. Sums source row counts by opening each `src/seg-*/shard-*/<partID>/`.
+// 3. Enumerates `target/seg-*/`, opens every part + per-seg union sidx,
+// and flags whether each seg's start time is grid-aligned.
+// 4. Builds an EntryGroupReport and hands it to `onReport` IMMEDIATELY
+// (callback fires inside the loop, not after) so the operator sees
+// numbers as they arrive — important on cold-tier scans where a
+// single (entry, group) can take minutes to finish.
+//
+// The function is read-only. Errors returned here mean "I could not
+// read this PVC at all" (bad path, permission, broken FS); per-(entry,
+// group) data failures are surfaced through the report itself, never
+// suppress the callback for sibling pairs.
+//
+// Pairs with empty source AND empty target are skipped (no callback).
+func MigrationVerify(ctx context.Context, cfg DirectCopyConfig, onReport func(EntryGroupReport)) error {
+ if err := validateDirectCopyConfig(&cfg); err != nil {
+ return err
+ }
+ //nolint:contextcheck // bluge reader.Search inside walkSchemaPropertyShard already uses its own context.
+ resourceOpts, err := loadGroupResourceOptsFromSchema(cfg.BackupDir, cfg.Date, cfg.SchemaPropertyPath, cfg.Groups)
+ if err != nil {
+ return fmt.Errorf("load group resource opts: %w", err)
+ }
+
+ fileSystem := fs.NewLocalFileSystem()
+ for entryIdx, entry := range cfg.Entries {
+ if ctx.Err() != nil {
+ return ctx.Err()
+ }
+ entryTag := fmt.Sprintf("entry [%d/%d]", entryIdx+1, len(cfg.Entries))
+ for _, group := range cfg.Groups {
+ if ctx.Err() != nil {
+ return ctx.Err()
+ }
+ opts, ok := resourceOpts[group]
+ if !ok || opts == nil {
+ return fmt.Errorf("%s group %s: ResourceOpts not found in backup schema-property catalog", entryTag, group)
+ }
+ ir, irErr := resolveStageInterval(opts, entry.Stage)
+ if irErr != nil {
+ return fmt.Errorf("%s stage=%s group %s: %w", entryTag, entry.Stage, group, irErr)
+ }
+
+ srcRoots := resolveEntrySrcRoots(cfg, entry, group)
+ targetGroup := filepath.Join(entry.Target, group)
+ if len(srcRoots) == 0 {
+ // Confirm whether the target carries any parts so the
+ // operator can still see them (rare but possible after
+ // a partial copy). Empty src + empty tgt is silenced.
+ //nolint:contextcheck // bluge reader.Search inside CountBlugeDocs already uses its own context.
+ segs, segErr := EnumerateGroupTarget(targetGroup, ir, fileSystem)
+ if segErr != nil {
+ return fmt.Errorf("%s stage=%s group %s: target: %w", entryTag, entry.Stage, group, segErr)
+ }
+ if len(segs) == 0 {
+ continue
+ }
+ onReport(EntryGroupReport{
+ Group: group,
+ EntryStage: entry.Stage,
+ EntryTarget: entry.Target,
+ EntryNodes: entry.Nodes,
+ SrcRoots: nil,
+ TargetGroup: targetGroup,
+ TargetSegs: segs,
+ })
+ continue
+ }
+
+ srcRows, srcParts, srcErr := SumGroupSourceRows(srcRoots, fileSystem)
+ if srcErr != nil {
+ return fmt.Errorf("%s stage=%s group %s: src: %w", entryTag, entry.Stage, group, srcErr)
+ }
+
+ //nolint:contextcheck // bluge reader.Search inside CountBlugeDocs already uses its own context.
+ segs, segErr := EnumerateGroupTarget(targetGroup, ir, fileSystem)
+ if segErr != nil {
+ return fmt.Errorf("%s stage=%s group %s: target: %w", entryTag, entry.Stage, group, segErr)
+ }
+
+ onReport(EntryGroupReport{
+ Group: group,
+ EntryStage: entry.Stage,
+ EntryTarget: entry.Target,
+ EntryNodes: entry.Nodes,
+ SrcRoots: srcRoots,
+ SrcRows: srcRows,
+ SrcParts: srcParts,
+ TargetGroup: targetGroup,
+ TargetSegs: segs,
+ })
+ }
+ }
+ return nil
+}
diff --git a/banyand/measure/migration_verify_test.go b/banyand/measure/migration_verify_test.go
new file mode 100644
index 000000000..014bd4f5a
--- /dev/null
+++ b/banyand/measure/migration_verify_test.go
@@ -0,0 +1,209 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package measure
+
+import (
+ "encoding/json"
+ "os"
+ "path/filepath"
+ "testing"
+ "time"
+
+ "github.com/blugelabs/bluge"
+
+ "github.com/apache/skywalking-banyandb/banyand/internal/storage"
+ "github.com/apache/skywalking-banyandb/pkg/fs"
+)
+
+// TestEnumerateGroupTarget_AlignmentAndSidx exercises the verify
+// helper's two main signals: (a) seg name → start time → grid alignment
+// against an IntervalRule, and (b) sidx doc count. It deliberately
+// creates segs WITHOUT shard subdirs so Parts/Rows stay at 0; that path
+// is covered separately by the end-to-end migration test.
+func TestEnumerateGroupTarget_AlignmentAndSidx(t *testing.T) {
+ groupRoot := t.TempDir()
+ dayInterval := storage.IntervalRule{Unit: storage.DAY, Num: 1}
+
+ // Aligned: 2026-05-19 at midnight UTC fits the day grid AND
+ // carries a metadata file whose endTime = NextTime(start).
+ segAligned := filepath.Join(groupRoot, directCopySegPrefix+"20260519")
+ if err := os.MkdirAll(segAligned, storage.DirPerm); err != nil {
+ t.Fatalf("seed aligned: %v", err)
+ }
+ wantStart := time.Date(2026, 5, 19, 0, 0, 0, 0, time.Local)
+ wantEnd := dayInterval.NextTime(wantStart)
+ writeTestSegmentMetadata(t, segAligned, wantEnd)
+
+ // Misaligned: HOUR-format seg name decoded as DAY would parse but
+ // the start time keeps the hour offset, which IntervalRule.Standard
+ // would round down — easier path is a seg name that DAY can't
+ // parse cleanly (parseDirectCopySegStart errors → Aligned stays
+ // false because StartTime is zero, which Standard returns as well).
+ // Use a name with hour suffix.
+ segUnparseable := filepath.Join(groupRoot, directCopySegPrefix+"20260519CRAP")
+ if err := os.MkdirAll(segUnparseable, storage.DirPerm); err != nil {
+ t.Fatalf("seed unparseable: %v", err)
+ }
+
+ // Drop a sidx with two docs into the aligned seg only.
+ sidxDir := filepath.Join(segAligned, directCopySidxDirName)
+ if err := os.MkdirAll(sidxDir, storage.DirPerm); err != nil {
+ t.Fatalf("mkdir sidx: %v", err)
+ }
+ w, err := bluge.OpenWriter(bluge.DefaultConfig(sidxDir))
+ if err != nil {
+ t.Fatalf("open writer: %v", err)
+ }
+ batch := bluge.NewBatch()
+ batch.Insert(bluge.NewDocument("id-a").AddField(
+ bluge.NewKeywordFieldBytes("k", []byte("v")).StoreValue()))
+ batch.Insert(bluge.NewDocument("id-b").AddField(
+ bluge.NewKeywordFieldBytes("k", []byte("v2")).StoreValue()))
+ if batchErr := w.Batch(batch); batchErr != nil {
+ t.Fatalf("batch: %v", batchErr)
+ }
+ if closeErr := w.Close(); closeErr != nil {
+ t.Fatalf("close: %v", closeErr)
+ }
+
+ reports, err := EnumerateGroupTarget(groupRoot, dayInterval, fs.NewLocalFileSystem())
+ if err != nil {
+ t.Fatalf("EnumerateGroupTarget: %v", err)
+ }
+ if len(reports) != 2 {
+ t.Fatalf("expected 2 reports, got %d: %+v", len(reports), reports)
+ }
+
+ var aligned, unparseable *SegmentReport
+ for i := range reports {
+ switch reports[i].Seg {
+ case directCopySegPrefix + "20260519":
+ aligned = &reports[i]
+ case directCopySegPrefix + "20260519CRAP":
+ unparseable = &reports[i]
+ }
+ }
+ if aligned == nil || unparseable == nil {
+ t.Fatalf("missing reports: %+v", reports)
+ }
+ if !aligned.StartTime.Equal(wantStart) {
+ t.Errorf("aligned StartTime = %v, want %v", aligned.StartTime, wantStart)
+ }
+ if !aligned.EndTime.Equal(wantEnd) {
+ t.Errorf("aligned EndTime = %v, want %v", aligned.EndTime, wantEnd)
+ }
+ if !aligned.Aligned {
+ t.Errorf("aligned.Aligned = false, want true (start aligned, endTime = NextTime(start), both in same bucket)")
+ }
+ if !aligned.SidxOpened || aligned.SidxDocCount != 2 {
+ t.Errorf("aligned sidx: opened=%v docs=%d (want true, 2)", aligned.SidxOpened, aligned.SidxDocCount)
+ }
+ if unparseable.Aligned {
+ t.Errorf("unparseable.Aligned = true, want false (zero StartTime fails grid check)")
+ }
+ if unparseable.SidxOpened {
+ t.Errorf("unparseable.SidxOpened = true, want false (no sidx subdir)")
+ }
+}
+
+// TestEnumerateGroupTarget_AlignedRequiresMetadata pins the contract that
+// a grid-aligned start directory name is NOT enough — the segment must
+// also carry a JSON metadata file with a non-empty endTime that equals
+// IntervalRule.NextTime(start). Without it Aligned must stay false so
+// the verify CLI surfaces "this seg was never finalized".
+func TestEnumerateGroupTarget_AlignedRequiresMetadata(t *testing.T) {
+ groupRoot := t.TempDir()
+ dayInterval := storage.IntervalRule{Unit: storage.DAY, Num: 1}
+
+ // Seg dir name aligns to the DAY grid but we deliberately omit the
+ // metadata file so haveEnd stays false.
+ segNoMeta := filepath.Join(groupRoot, directCopySegPrefix+"20260601")
+ if err := os.MkdirAll(segNoMeta, storage.DirPerm); err != nil {
+ t.Fatalf("seed: %v", err)
+ }
+
+ // A second seg whose metadata file has a wrong (non-NextTime) endTime
+ // must also report Aligned=false.
+ segBadEnd := filepath.Join(groupRoot, directCopySegPrefix+"20260603")
+ if err := os.MkdirAll(segBadEnd, storage.DirPerm); err != nil {
+ t.Fatalf("seed: %v", err)
+ }
+ wrongEnd := time.Date(2026, 6, 4, 12, 0, 0, 0, time.Local) // mid-day, not on grid
+ writeTestSegmentMetadata(t, segBadEnd, wrongEnd)
+
+ reports, err := EnumerateGroupTarget(groupRoot, dayInterval, fs.NewLocalFileSystem())
+ if err != nil {
+ t.Fatalf("EnumerateGroupTarget: %v", err)
+ }
+ if len(reports) != 2 {
+ t.Fatalf("expected 2 reports, got %d", len(reports))
+ }
+ for _, r := range reports {
+ if r.Aligned {
+ t.Errorf("seg %s Aligned=true, want false (metadata missing or endTime mismatched)", r.Seg)
+ }
+ }
+}
+
+// writeTestSegmentMetadata seeds <segDir>/metadata with a JSON-encoded
+// SegmentMetadata carrying the supplied endTime. Mirrors what
+// writeDirectCopySegmentMetadata + the live segmentController emit.
+func writeTestSegmentMetadata(t *testing.T, segDir string, endTime time.Time) {
+ t.Helper()
+ body := storage.SegmentMetadata{
+ Version: storage.CurrentSegmentVersion,
+ EndTime: endTime.Format(time.RFC3339Nano),
+ }
+ raw, err := json.Marshal(body)
+ if err != nil {
+ t.Fatalf("marshal seg metadata: %v", err)
+ }
+ if writeErr := os.WriteFile(filepath.Join(segDir, storage.SegmentMetadataFilename), raw, storage.FilePerm); writeErr != nil {
+ t.Fatalf("write seg metadata: %v", writeErr)
+ }
+}
+
+// TestEnumerateGroupTarget_MissingRoot ensures a non-existent groupRoot
+// returns (nil, nil) — the caller is expected to treat this as "this
+// PVC does not host the group" rather than an error.
+func TestEnumerateGroupTarget_MissingRoot(t *testing.T) {
+ missing := filepath.Join(t.TempDir(), "no-such-group")
+ dayInterval := storage.IntervalRule{Unit: storage.DAY, Num: 1}
+ reports, err := EnumerateGroupTarget(missing, dayInterval, fs.NewLocalFileSystem())
+ if err != nil {
+ t.Fatalf("EnumerateGroupTarget on missing dir: %v", err)
+ }
+ if reports != nil {
+ t.Fatalf("expected nil reports, got %+v", reports)
+ }
+}
+
+// TestSumGroupSourceRows_MissingRootsSkipped confirms the helper
+// silently skips srcRoots whose path does not exist, matching the
+// resolveEntrySrcRoots contract (it already filters non-existent
+// dirs out before calling SumGroupSourceRows).
+func TestSumGroupSourceRows_MissingRootsSkipped(t *testing.T) {
+ missing := filepath.Join(t.TempDir(), "ghost-pvc", "measure", "data", "sw_metricsMinute")
+ rows, parts, err := SumGroupSourceRows([]string{missing}, fs.NewLocalFileSystem())
+ if err != nil {
+ t.Fatalf("expected no error on missing root, got %v", err)
+ }
+ if rows != 0 || parts != 0 {
+ t.Fatalf("expected 0/0, got rows=%d parts=%d", rows, parts)
+ }
+}