You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@skywalking.apache.org by ha...@apache.org on 2026/06/23 04:23:24 UTC
(skywalking-banyandb) branch main updated: Support migration copy/verify/analysis index mode measure (#1187)
This is an automated email from the ASF dual-hosted git repository.
hanahmily 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 c6183d7ff Support migration copy/verify/analysis index mode measure (#1187)
c6183d7ff is described below
commit c6183d7ff43e322bc82341fb78ad64bd95a29601
Author: mrproliu <74...@qq.com>
AuthorDate: Tue Jun 23 12:23:19 2026 +0800
Support migration copy/verify/analysis index mode measure (#1187)
* Support migration copy/verify/analysis index mode measure
---
banyand/cmd/migration/MIGRATION.md | 2 +-
banyand/cmd/migration/README.md | 24 +-
banyand/cmd/migration/analyze.go | 75 ++
banyand/cmd/migration/copy.go | 17 +-
banyand/cmd/migration/scripts/verify-data.sh | 24 +
banyand/cmd/migration/verify.go | 172 ++-
banyand/internal/migration/orchestrator.go | 23 +
banyand/internal/migration/orchestrator_test.go | 100 ++
banyand/measure/migration_analyze.go | 17 +
banyand/measure/migration_copy.go | 18 -
banyand/measure/migration_copy_test.go | 58 -
banyand/measure/migration_executor.go | 46 +-
banyand/measure/migration_indexmode_copy.go | 1111 ++++++++++++++++
banyand/measure/migration_indexmode_copy_test.go | 1470 ++++++++++++++++++++++
banyand/measure/migration_indexmode_e2e_test.go | 538 ++++++++
banyand/measure/migration_schema.go | 44 +-
banyand/measure/migration_schema_test.go | 93 ++
banyand/measure/migration_verify.go | 56 +-
banyand/metadata/schema/reader/loaders.go | 6 +
19 files changed, 3764 insertions(+), 130 deletions(-)
diff --git a/banyand/cmd/migration/MIGRATION.md b/banyand/cmd/migration/MIGRATION.md
index 929be93d5..b46af0740 100644
--- a/banyand/cmd/migration/MIGRATION.md
+++ b/banyand/cmd/migration/MIGRATION.md
@@ -14,7 +14,7 @@ Two end-to-end procedures, depending on where the source data lives:
For the tool's CLI surface, plan schema, and on-disk output layout, see [`README.md`](README.md).
-> **Scope:** the tool migrates **measure** and **stream** data. The catalog of each group is auto-detected from the schema-property catalog (falling back to the backup directory layout for catalog detection only) and each group is routed to its catalog's executor — one plan may mix both catalogs; the runbook steps below are identical for both, substituting the stream data roots / `--stream-root-path` where a measure path is shown. Trace / property catalogs and `sw_metadata` are out of sc [...]
+> **Scope:** the tool migrates **measure** (normal *and* index-mode, e.g. `sw_metadata`) and **stream** data. The catalog of each group is auto-detected from the schema-property catalog (falling back to the backup directory layout for catalog detection only) and each group is routed to its catalog's executor — one plan may mix both catalogs; the runbook steps below are identical for both, substituting the stream data roots / `--stream-root-path` where a measure path is shown. Trace / pro [...]
---
diff --git a/banyand/cmd/migration/README.md b/banyand/cmd/migration/README.md
index 7130d3ee3..14fc1f5a2 100644
--- a/banyand/cmd/migration/README.md
+++ b/banyand/cmd/migration/README.md
@@ -1,6 +1,8 @@
# BanyanDB data migration (measure & stream)
-A one-shot CLI that rewrites **measure** or **stream** data — from either a local backup snapshot or a set of live PVC mount paths — into one or more target data roots, with rows re-bucketed onto each target stage's `SegmentInterval` grid. The catalog of each group is auto-detected (authoritatively from the source's schema-property catalog via `banyand/metadata/schema/reader`; when a backup snapshot carries no catalog, the backup directory layout — `<node>/<date>/{measure,stream}/<group> [...]
+A one-shot CLI that rewrites **measure** or **stream** data — from either a local backup snapshot or a set of live PVC mount paths — into one or more target data roots, with rows re-bucketed onto each target stage's `SegmentInterval` grid. The catalog of each group is auto-detected (authoritatively from the source's schema-property catalog via `banyand/metadata/schema/reader`; when a backup snapshot carries no catalog, the backup directory layout — `<node>/<date>/{measure,stream}/<group> [...]
+
+Index-mode specifics: an index-mode measure (`IndexMode: true`, e.g. SkyWalking's `sw_metadata` group) stores no columnar parts — every data point is one inverted document under the segment-level `seg/sidx/`, upserted by series per segment (one max-version doc per `(series, segment)`). The tool detects these groups from the catalog and routes them through a dedicated path: each source sidx is either **byte-copied** whole (when all its docs align to one not-yet-written target segment) or [...]
Stream specifics: stream has no field columns, no version, and the tool never de-duplicates rows, so source and target row counts are expected to match **exactly**. Each stream segment carries two index layers — the segment-level series index (`sidx/`, union-built and broadcast like measure) and the per-shard element index (`idx/`). On the fast path (a source segment maps to a single target segment) both are byte-copied; on the slow path (a source segment splits across target segments) t [...]
@@ -11,8 +13,8 @@ The CLI exposes three subcommands; all share a single required flag `--copy-conf
| Subcommand | Purpose |
|---|---|
| `copy` | Rewrite measure or stream 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. |
+| `verify` | Re-read the same plan and report per-(entry, group) source-vs-target row counts, segment grid alignment, and union-sidx doc counts. For index-mode groups it instead reconciles sidx doc counts, distinct doc-id (series) counts, per-`(segment, series)` full-field value digests, and per-segment 1:1 alignment. 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. For index-mode groups it reports `(series, timestamp)` version-duplicate and value-conflict keys (explains why target doc count `≤` source). 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.
@@ -20,9 +22,9 @@ Each source row is routed to a grid-aligned target segment using the `SegmentInt
## What it does
-1. Loads the schemas of the plan's catalog from the source's schema-property bluge catalog — measure schemas (tag families + `IndexMode` bit) or stream schemas (tag families + index-rule bindings + entity layout) — plus each group's `ResourceOpts`.
-2. Measure plans only: 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). Stream has no `IndexMode`.
-3. For each group, walks every source segment under `<source>/<node>/.../measure/<group>/` (or `.../stream/<group>/`) and:
+1. Loads the schemas of the plan's catalog from the source's schema-property bluge catalog — measure schemas (tag families + `IndexMode` bit + index rules) or stream schemas (tag families + index-rule bindings + entity layout) — plus each group's `ResourceOpts`.
+2. Classifies each measure group as **normal** or **index-mode** (a group is index-mode when its only non-`_top_n_result` measures all carry `IndexMode: true`). Normal groups take the union-sidx path below; index-mode groups take the sidx rebuild/byte-copy path (see "Index-mode specifics" above) and are **excluded** from union-sidx broadcast. Stream has no `IndexMode`.
+3. For each **normal** measure / stream group, walks every source segment under `<source>/<node>/.../measure/<group>/` (or `.../stream/<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:
@@ -39,7 +41,7 @@ Entries run sequentially. The first error aborts the run. The whole `staging_dir
## Limitations
- Every command **requires** a `schema-property` catalog under at least one source node (typically a hot / schema-server node): group catalogs, segment alignment and stream index rebuilds all resolve from it. A source without the catalog fails fast.
-- All groups containing any `IndexMode` measure are refused up front (measure groups only — stream has no `IndexMode`).
+- Index-mode measure groups are supported (sidx rebuild / byte-copy path). A group that mixes an index-mode measure with a non-`_top_n_result` normal measure that has actual shard parts is refused — split such a group before migrating.
- Source and target shard topology must match (no reshard semantics).
- The tool assumes nothing else is writing to the target paths.
@@ -130,6 +132,14 @@ A stream target carries one extra layer — the per-shard element index:
└── sidx/ # broadcast union sidx (bluge index)
```
+An **index-mode** measure target carries no `shard-N/` parts at all — every segment holds only the rebuilt/byte-copied series index:
+
+```
+<entry.target>/<group>/seg-YYYYMMDD[HH]/
+ ├── metadata # storage.SegmentMetadata: segment version + endTime
+ └── sidx/ # index-mode docs (bluge); byte-copied or rebuilt per segment
+```
+
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`.
---
diff --git a/banyand/cmd/migration/analyze.go b/banyand/cmd/migration/analyze.go
index d8d25157f..c4a25b4d4 100644
--- a/banyand/cmd/migration/analyze.go
+++ b/banyand/cmd/migration/analyze.go
@@ -18,6 +18,7 @@
package main
import (
+ "context"
"fmt"
"path/filepath"
"time"
@@ -87,6 +88,13 @@ target part).`,
if catalog == commonv1.Catalog_CATALOG_STREAM {
return runStreamAnalyze(entries, entry, entryIdx, srcRoots, groupName, sampleCap)
}
+ isIndexMode, imErr := measure.IsIndexModeGroup(cls.SchemaRoot, groupName)
+ if imErr != nil {
+ return imErr
+ }
+ if isIndexMode {
+ return runIndexModeAnalyze(cls.SchemaRoot, entries, entry, entryIdx, srcRoots, groupName, sampleCap)
+ }
return runMeasureAnalyze(entries, entry, entryIdx, srcRoots, groupName, sampleCap)
},
}
@@ -212,6 +220,73 @@ func runMeasureAnalyze(entries []migration.ResolvedEntry, entry migration.Resolv
return nil
}
+// runIndexModeAnalyze runs the index-mode-flavored analyze logic: it reads the
+// group's sidx documents (not part block metadata) and reports total docs,
+// distinct (series, timestamp) keys, version duplicates and value conflicts —
+// a (series, timestamp) appearing in >1 doc with differing value digests.
+func runIndexModeAnalyze(schemaRoot string, entries []migration.ResolvedEntry, entry migration.ResolvedEntry,
+ entryIdx int, srcRoots []string, groupName string, sampleCap int,
+) error {
+ fmt.Printf("== analyze entry [%d/%d] stage=%s nodes=%v group=%s (index-mode) ==\n",
+ entryIdx+1, len(entries), entry.Stage, entry.Nodes, groupName)
+ for _, r := range srcRoots {
+ fmt.Printf(" src: %s\n", r)
+ }
+ start := time.Now()
+ res, err := measure.AnalyzeIndexModeGroup(context.Background(), schemaRoot, groupName, srcRoots, sampleCap)
+ if err != nil {
+ return err
+ }
+ elapsed := time.Since(start)
+
+ fmt.Printf(" sidx dirs scanned : %d\n", res.PartsScanned)
+ fmt.Printf(" total docs on disk : %d\n", res.TotalRows)
+ fmt.Printf(" distinct (sid, ts) : %d\n", res.UniqueKeys)
+ fmt.Printf(" version-duplicate docs : %d (same (sid, ts) seen in >1 doc)\n", res.DuplicateRows)
+ fmt.Printf(" (sid, ts) keys with >1 doc : %d\n", res.KeysWithDuplicates)
+ fmt.Printf(" VALUE-CONFLICT keys : %d (same (sid, ts), differing value digest)\n", res.ValueConflictKeys)
+ fmt.Printf(" elapsed : %v\n", elapsed)
+ if res.KeysWithDuplicates > 0 {
+ shown := len(res.SamplesByVersion)
+ fmt.Printf(" sample (sid, ts) with >1 doc (showing %d / %d keys):\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\n", v.Version)
+ }
+ }
+ if uint64(shown) < res.KeysWithDuplicates {
+ fmt.Printf(" ... %d more (raise --sample to see them)\n", res.KeysWithDuplicates-uint64(shown))
+ }
+ }
+ if res.ValueConflictKeys > 0 {
+ shown := len(res.ValueConflicts)
+ fmt.Printf(" VALUE CONFLICTS (showing %d / %d keys; same (sid, ts) but differing data values):\n",
+ shown, res.ValueConflictKeys)
+ for _, c := range res.ValueConflicts {
+ ts := time.Unix(0, c.Timestamp).UTC().Format(time.RFC3339)
+ fmt.Printf(" sid=%d ts=%s\n", c.SeriesID, ts)
+ for i := range c.Versions {
+ fmt.Printf(" version=%d digest=%016x\n", c.Versions[i], c.Digests[i])
+ }
+ }
+ if uint64(shown) < res.ValueConflictKeys {
+ fmt.Printf(" ... %d more (raise --sample to see them)\n", res.ValueConflictKeys-uint64(shown))
+ }
+ }
+ fmt.Println()
+ fmt.Println("Interpretation:")
+ fmt.Println(" - index-mode data lives in the segment sidx, one upserted doc per (series, segment); the (sid, ts) keys below are the analyze granularity.")
+ fmt.Println(" - version-duplicate docs are normal across source segments (target keeps max version).")
+ fmt.Println(" - VALUE-CONFLICT keys are (sid, ts) whose data values differ between docs. For metadata")
+ fmt.Println(" measures this is common and expected: a series is re-upserted over time and its tag")
+ fmt.Println(" values evolve with each version, so the same (sid, ts) carries different values across")
+ fmt.Println(" segments — the migration keeps the max-version one. Only investigate if you expected")
+ fmt.Println(" these values to be immutable (then a conflict would indicate a real inconsistency).")
+ return nil
+}
+
// runStreamAnalyze runs the stream-flavored analyze logic.
func runStreamAnalyze(entries []migration.ResolvedEntry, entry migration.ResolvedEntry, entryIdx int, srcRoots []string, groupName string, sampleCap int) error {
fmt.Printf("== analyze entry [%d/%d] stage=%s nodes=%v group=%s (stream) ==\n",
diff --git a/banyand/cmd/migration/copy.go b/banyand/cmd/migration/copy.go
index 5ecae10a1..a618b68c0 100644
--- a/banyand/cmd/migration/copy.go
+++ b/banyand/cmd/migration/copy.go
@@ -63,11 +63,18 @@ index-rule bindings) and group resource opts (SegmentInterval per
LifecycleStage) are read directly from the source's schema-property
bluge catalog — no liaison access is needed. Each group routes to its
catalog's executor, so one plan may mix measure and stream groups.
-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. Stream groups additionally
-finalize the per-shard element index (idx/) — byte-copied on the fast
-path, rebuilt from rows + index rules when a source segment splits.
+IndexMode measure groups take a dedicated path: their data points live
+as documents in the segment-level sidx (no shard parts), so each doc is
+routed by its own timestamp to the grid-aligned target segment's sidx —
+byte-copied when a source segment maps to a single target, rebuilt
+otherwise (regenerating the index-only _im_name / _im_entity_tag_*
+fields from the doc id so entity-tag search survives), keeping the
+highest version per series on merge. The union-sidx broadcast is skipped
+for such groups. The internal _top_n_result measure is ignored; a group
+mixing IndexMode with any other normal measure is rejected. Stream
+groups additionally finalize the per-shard element index (idx/) —
+byte-copied on the fast path, rebuilt from rows + index rules when a
+source segment splits.
The staging directory is removed in full when the run finishes
(success or failure). Run inside a data pod whose BanyanDB process is
diff --git a/banyand/cmd/migration/scripts/verify-data.sh b/banyand/cmd/migration/scripts/verify-data.sh
index bec1bccb2..710e2a370 100755
--- a/banyand/cmd/migration/scripts/verify-data.sh
+++ b/banyand/cmd/migration/scripts/verify-data.sh
@@ -110,6 +110,29 @@ EOF
printf '%s' "$resp" | { grep -o '"elementId"' || true; } | wc -l | tr -d ' '
}
+# Probe an index-mode measure. Index-mode measures store every data point as a
+# bluge document in the segment-level sidx and carry NO fields, so the query
+# omits fieldProjection and projects the entity tag (storage-only/id) that every
+# metadata measure has. Any non-zero data-point count means "present".
+probe_measure_indexmode() {
+ local group="$1" measure="$2" begin="$3" end="$4"
+ local body resp
+ body=$(cat <<EOF
+{
+ "groups": ["$group"],
+ "name": "$measure",
+ "stages": ["hot"],
+ "timeRange": {"begin": "$begin", "end": "$end"},
+ "tagProjection": {"tagFamilies":[{"name":"storage-only","tags":["id"]}]},
+ "limit": 100000
+}
+EOF
+)
+ resp=$(curl -s -XPOST "$LIAISON_HTTP/api/v1/measure/data" \
+ -H 'Content-Type: application/json' -d "$body" || true)
+ 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 " catalogs: $CATALOGS"
@@ -138,6 +161,7 @@ case " $CATALOGS " in *" measure "*)
scan probe_measure "Day tier" sw_metricsDay service_apdex_day T00:00:00Z T23:59:59Z "full day"
scan probe_measure "Hour tier" sw_metricsHour service_apdex_hour T08:00:00Z T08:59:59Z "08:00..09:00"
scan probe_measure "Minute tier" sw_metricsMinute service_apdex_minute T08:08:00Z T08:08:59Z "08:08..08:09"
+ scan probe_measure_indexmode "Index-mode metadata" sw_metadata instance_traffic_minute T00:00:00Z T23:59:59Z "full day"
;; esac
case " $CATALOGS " in *" stream "*)
diff --git a/banyand/cmd/migration/verify.go b/banyand/cmd/migration/verify.go
index 878d50cb4..7dbdfd36a 100644
--- a/banyand/cmd/migration/verify.go
+++ b/banyand/cmd/migration/verify.go
@@ -98,16 +98,22 @@ migration result.`,
// hasStream / hasMeasure track which catalogs reported, so printSummary
// can pick the right per-catalog messaging (a mixed plan prints both).
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
- hasStream bool
- hasMeasure bool
+ 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
+ imSrcDocs uint64 // index-mode source sidx docs
+ imTgtDocs uint64 // index-mode target sidx docs
+ imValueMismatches int // index-mode (segment, series) value-digest mismatches
+ imDocIDIssues int // index-mode (entry, group) failing doc-id reconcile
+ segsTotal int
+ segsMisaligned int
+ hasStream bool
+ hasMeasure bool
+ hasNormalMeasure bool // a non-index-mode measure report was absorbed (part-row data)
+ hasIndexMode bool
}
// coverageState records whether SOURCE and TARGET independently hold
@@ -204,6 +210,11 @@ func (t *verifyTally) absorbCommon(stage string, nodes []string, group string, s
func (t *verifyTally) absorb(r measure.EntryGroupReport) {
t.hasMeasure = true
+ if r.IndexMode {
+ t.absorbIndexMode(r)
+ return
+ }
+ t.hasNormalMeasure = true
segs := make([]tallySeg, len(r.TargetSegs))
for i, s := range r.TargetSegs {
segs[i] = tallySeg{rows: s.Rows, aligned: s.Aligned}
@@ -211,6 +222,35 @@ func (t *verifyTally) absorb(r measure.EntryGroupReport) {
t.absorbCommon(r.EntryStage, r.EntryNodes, r.Group, r.SrcRows, segs)
}
+// absorbIndexMode folds an index-mode (entry, group) report into the tally.
+// Index-mode data lives in sidx documents, not part rows, and a coarser target
+// grid legitimately merges several source docs of one series into one — so the
+// doc COUNT differing src↔tgt is normal and must NOT be reported as a row-count
+// mismatch. The real signals are the per-(segment, series) value-digest
+// mismatches and the doc-id reconcile, tracked separately and folded into the
+// final verdict.
+func (t *verifyTally) absorbIndexMode(r measure.EntryGroupReport) {
+ t.hasIndexMode = true
+ node := nodeLabel(r.EntryNodes, r.EntryStage)
+ 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 !slices.Contains(t.groupOrder, r.Group) {
+ t.groupOrder = append(t.groupOrder, r.Group)
+ }
+ t.coverage[node][r.Group] = coverageState{src: r.SrcDocs > 0, tgt: r.TgtDocs > 0}
+ t.imSrcDocs += r.SrcDocs
+ t.imTgtDocs += r.TgtDocs
+ t.imValueMismatches += len(r.ValueMismatches)
+ if r.TgtDistinctIDs != r.ExpectDistinctIDs {
+ t.imDocIDIssues++
+ }
+}
+
func (t *verifyTally) printSummary() {
if t.hasStream && !t.hasMeasure {
fmt.Println("== SUMMARY (stream) ==")
@@ -219,23 +259,54 @@ func (t *verifyTally) printSummary() {
}
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 {
+ // Part-row counters only mean something for stream / normal-measure data; a
+ // pure index-mode run has none, so printing "0" there reads as "nothing was
+ // copied". Suppress them unless such data was actually reported.
+ hasRowData := t.hasStream || t.hasNormalMeasure
+ if hasRowData {
+ 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)
+ }
+ imClean := t.imValueMismatches == 0 && t.imDocIDIssues == 0
+ if t.hasIndexMode {
+ imTag := "OK"
+ if !imClean {
+ imTag = "MISMATCH"
+ }
+ fmt.Printf(" index-mode src docs : %d\n", t.imSrcDocs)
+ fmt.Printf(" index-mode tgt docs : %d (tgt<src is normal: a coarser grid merges same-series docs)\n", t.imTgtDocs)
+ fmt.Printf(" index-mode reconcile : %d value mismatch(es), %d doc-id issue(s) [%s]\n",
+ t.imValueMismatches, t.imDocIDIssues, imTag)
+ }
+ if len(t.mismatches) == 0 && imClean {
+ if hasRowData && 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 {
- if t.hasStream && !t.hasMeasure {
+ switch {
+ case t.hasStream && !t.hasMeasure:
fmt.Println(" src == tgt (stream invariant: no rows dropped)")
- } else {
+ case t.hasIndexMode && !t.hasNormalMeasure:
+ fmt.Println(" src == tgt (index-mode: every series preserved, max version per segment)")
+ default:
fmt.Println(" src == tgt")
}
}
return
}
+ if !imClean {
+ // Surface the index-mode failure regardless of whether a row-count
+ // mismatch table also follows, so a mixed run never hides it.
+ fmt.Println()
+ fmt.Printf(" index-mode verify FAILED: %d value mismatch(es), %d doc-id issue(s) — "+
+ "see the per-(entry, group) reports above for the offending (segment, series).\n",
+ t.imValueMismatches, t.imDocIDIssues)
+ }
+ if len(t.mismatches) == 0 {
+ return
+ }
fmt.Println()
if t.hasStream && !t.hasMeasure {
fmt.Println(" row-count mismatches (stream must have 0 diff — any non-zero is a BUG):")
@@ -346,6 +417,10 @@ func padString(s string, w int) string {
// 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) {
+ if r.IndexMode {
+ printOneIndexModeReport(r)
+ return
+ }
fmt.Printf("== entry stage=%s target=%s group=%s ==\n",
r.EntryStage, r.EntryTarget, r.Group)
if len(r.SrcRoots) == 0 {
@@ -381,6 +456,67 @@ func printOneReport(r measure.EntryGroupReport) {
}
}
+// printOneIndexModeReport renders a single index-mode (entry, group) report:
+// document counts, distinct doc-id reconciliation and per-(segment, series)
+// value-digest mismatches. Index-mode groups carry data in the segment sidx, so
+// they are reconciled at the document level rather than by part rows.
+func printOneIndexModeReport(r measure.EntryGroupReport) {
+ fmt.Printf("== entry stage=%s target=%s group=%s (index-mode) ==\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 doc(s), %d distinct doc-id across %d dir(s)\n",
+ r.SrcDocs, r.SrcDistinctIDs, len(r.SrcRoots))
+ for _, p := range r.SrcRoots {
+ fmt.Printf(" %s\n", p)
+ }
+ }
+ fmt.Printf(" tgt : %d doc(s), %d distinct doc-id under %s\n",
+ r.TgtDocs, r.TgtDistinctIDs, r.TargetGroup)
+ idTag := "OK"
+ if r.TgtDistinctIDs != r.ExpectDistinctIDs {
+ idTag = "MISMATCH"
+ }
+ fmt.Printf(" doc-id reconcile : expected(union of src)=%d tgt=%d [%s]\n",
+ r.ExpectDistinctIDs, r.TgtDistinctIDs, idTag)
+ printIndexModeSegAligns(r.SegAligns)
+ if len(r.ValueMismatches) == 0 {
+ fmt.Printf(" value reconcile : all (segment, series) digests match [OK]\n")
+ return
+ }
+ fmt.Printf(" value reconcile : %d (segment, series) key(s) MISMATCH:\n", len(r.ValueMismatches))
+ for _, m := range r.ValueMismatches {
+ fmt.Printf(" %-9s seg=%s sid=%d\n", m.Kind, m.Segment, m.SeriesID)
+ }
+}
+
+// printIndexModeSegAligns renders the per-target-segment alignment breakdown: for
+// each segment whether it was a clean 1:1 byte-copy candidate (one source segment,
+// no collapse) or merged/split, plus the source-doc -> target-doc collapse. This
+// shows the operator why a segment did or did not take the byte-copy fast path.
+func printIndexModeSegAligns(segs []measure.IndexModeSegAlign) {
+ if len(segs) == 0 {
+ return
+ }
+ aligned := 0
+ for _, s := range segs {
+ if s.Aligned {
+ aligned++
+ }
+ }
+ fmt.Printf(" segment alignment: %d/%d target seg(s) are 1:1 aligned (byte-copy candidates); the rest were merged/split:\n",
+ aligned, len(segs))
+ for _, s := range segs {
+ tag := "merged/split"
+ if s.Aligned {
+ tag = "aligned 1:1"
+ }
+ fmt.Printf(" %-16s %-13s src-segs=%d src-docs=%d -> tgt-docs=%d (collapsed %d)\n",
+ s.Segment, tag, s.SrcSegs, s.SrcDocs, s.TgtDocs, s.SrcDocs-s.TgtDocs)
+ }
+}
+
// absorbStream records one stream (entry, group) report into the tally.
func (t *verifyTally) absorbStream(r stream.EntryGroupReport) {
t.hasStream = true
diff --git a/banyand/internal/migration/orchestrator.go b/banyand/internal/migration/orchestrator.go
index 48d818b1e..2aacbea28 100644
--- a/banyand/internal/migration/orchestrator.go
+++ b/banyand/internal/migration/orchestrator.go
@@ -31,6 +31,21 @@ import (
"github.com/apache/skywalking-banyandb/banyand/internal/storage"
)
+// unionSidxSkipper is an optional executor capability: groups it reports true
+// for skip the Phase A union-sidx build + broadcast (index-mode measure groups,
+// whose data docs must NOT be broadcast).
+type unionSidxSkipper interface {
+ SkipUnionSidx(group string) bool
+}
+
+// skipUnionSidx reports whether exec wants the given group's union-sidx build
+// and broadcast skipped. Executors that do not implement unionSidxSkipper
+// always participate in the union-sidx path (the historical behavior).
+func skipUnionSidx(exec CatalogExecutor, group string) bool {
+ s, ok := exec.(unionSidxSkipper)
+ return ok && s.SkipUnionSidx(group)
+}
+
// Result aggregates a whole run across every catalog the plan touched.
type Result struct {
PerCatalog map[commonv1.Catalog]EntryGroupResult
@@ -187,6 +202,14 @@ func (p *CopyPlan) RunCopy(ctx context.Context, stagingDir string, cls *Classifi
if ctx.Err() != nil {
return res, ctx.Err()
}
+ // Index-mode measure groups opt out of the union-sidx build +
+ // broadcast: their sidx docs are data points routed per timestamp,
+ // not series-index metadata to broadcast. Leaving groupUnionSidx[group]
+ // empty makes Phase B pass UnionSidxPath=="" through to the executor.
+ if skipUnionSidx(exec, group) {
+ logStep(prefix, "group %s: index-mode — skipping union sidx build + broadcast", group)
+ continue
+ }
srcRoots := p.CollectAllSrcGroupRoots(exec.Catalog(), group)
if len(srcRoots) == 0 {
logStep(prefix, "group %s: no source dirs across any entry — skipping union sidx build", group)
diff --git a/banyand/internal/migration/orchestrator_test.go b/banyand/internal/migration/orchestrator_test.go
index d8fcb79b4..40c8959bb 100644
--- a/banyand/internal/migration/orchestrator_test.go
+++ b/banyand/internal/migration/orchestrator_test.go
@@ -18,10 +18,12 @@
package migration
import (
+ "context"
"os"
"path/filepath"
"testing"
+ "github.com/blugelabs/bluge"
"github.com/stretchr/testify/require"
commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
@@ -77,3 +79,101 @@ func TestDirtyTargets_AllListedUpFront(t *testing.T) {
require.ElementsMatch(t, []string{dirtyA, dirtyWarmA}, dirty)
require.NotContains(t, dirty, skipped)
}
+
+// skipUnionSidxExecutor is a fake CatalogExecutor that also implements the
+// optional unionSidxSkipper capability: it reports skip for the groups in
+// skipGroups and records the UnionSidxPath each CopyEntryGroup received.
+type skipUnionSidxExecutor struct {
+ gotUnionSidxPath map[string]string
+ skipGroups map[string]bool
+}
+
+func (e *skipUnionSidxExecutor) Catalog() commonv1.Catalog { return commonv1.Catalog_CATALOG_MEASURE }
+
+func (e *skipUnionSidxExecutor) LogPrefix() string { return "[migration/measure-test]" }
+
+func (e *skipUnionSidxExecutor) Prepare(_ context.Context, _ string, _ []string) error { return nil }
+
+func (e *skipUnionSidxExecutor) CopyEntryGroup(_ context.Context, in EntryGroupInput) (EntryGroupResult, error) {
+ if e.gotUnionSidxPath == nil {
+ e.gotUnionSidxPath = map[string]string{}
+ }
+ e.gotUnionSidxPath[in.Group] = in.UnionSidxPath
+ return EntryGroupResult{}, nil
+}
+
+func (e *skipUnionSidxExecutor) VerifyEntryGroup(_ context.Context, _ EntryGroupInput, _ func(report any)) error {
+ return nil
+}
+
+func (e *skipUnionSidxExecutor) SkipUnionSidx(group string) bool { return e.skipGroups[group] }
+
+// TestRunCopy_SkipsUnionSidxForIndexModeGroup verifies that a group an executor
+// reports via SkipUnionSidx never goes through the Phase A union-sidx build and
+// receives UnionSidxPath=="" in CopyEntryGroup, while a non-skipped group with
+// real source sidx docs gets a non-empty UnionSidxPath broadcast.
+func TestRunCopy_SkipsUnionSidxForIndexModeGroup(t *testing.T) {
+ root := t.TempDir()
+ stagingDir := t.TempDir()
+
+ // Live-mode source so ResolvedEntries populates entry.Source; the per-group
+ // source dirs are <root>/<group>. Seed both groups so neither is skipped for
+ // "no source dirs". The index-mode group is left without any sidx so even if
+ // a union build were (wrongly) attempted it would produce nothing.
+ skipGroup := "sw_metadata"
+ normalGroup := "sw_metrics"
+ require.NoError(t, os.MkdirAll(filepath.Join(root, skipGroup), 0o755))
+ // Seed a real source sidx under the normal group so Phase A's union build
+ // yields a non-empty path it can broadcast.
+ doc, _ := makeSidxSourceDoc(t, "alpha", "svc-a")
+ writeSidxAt(t, filepath.Join(root, normalGroup, segPrefix+"20260101"), []*bluge.Document{doc})
+
+ dayInterval := &commonv1.IntervalRule{
+ Unit: commonv1.IntervalRule_UNIT_DAY,
+ Num: 1,
+ }
+ groupProto := func(name string) *commonv1.Group {
+ return &commonv1.Group{
+ Metadata: &commonv1.Metadata{Name: name},
+ Catalog: commonv1.Catalog_CATALOG_MEASURE,
+ ResourceOpts: &commonv1.ResourceOpts{
+ SegmentInterval: dayInterval,
+ },
+ }
+ }
+ buckets := map[commonv1.Catalog][]string{}
+ buckets[commonv1.Catalog_CATALOG_MEASURE] = []string{skipGroup, normalGroup}
+ cls := &Classified{
+ Buckets: buckets,
+ Groups: map[string]*commonv1.Group{
+ skipGroup: groupProto(skipGroup),
+ normalGroup: groupProto(normalGroup),
+ },
+ SchemaRoot: filepath.Join(root, "_schema"),
+ }
+
+ plan := &CopyPlan{
+ Source: CopySource{Live: &LiveSource{
+ SchemaPropertyPath: filepath.Join(root, "_schema"),
+ Stages: map[string][]LiveStageNode{
+ "hot": {{Node: "hot-0", Root: root}},
+ },
+ }},
+ Groups: []string{skipGroup, normalGroup},
+ Entries: []CopyEntry{
+ {Stage: "hot", Target: filepath.Join(t.TempDir(), "out"), Nodes: []string{"hot-0"}},
+ },
+ }
+
+ exec := &skipUnionSidxExecutor{skipGroups: map[string]bool{skipGroup: true}}
+ _, err := plan.RunCopy(context.Background(), stagingDir, cls, []CatalogExecutor{exec})
+ require.NoError(t, err)
+
+ gotSkip, ok := exec.gotUnionSidxPath[skipGroup]
+ require.True(t, ok, "index-mode group must still be copied")
+ require.Empty(t, gotSkip, "index-mode group must receive empty UnionSidxPath")
+
+ gotNormal, ok := exec.gotUnionSidxPath[normalGroup]
+ require.True(t, ok, "normal group must be copied")
+ require.NotEmpty(t, gotNormal, "normal group must receive a broadcast union sidx path")
+}
diff --git a/banyand/measure/migration_analyze.go b/banyand/measure/migration_analyze.go
index e5982586a..c91df8f76 100644
--- a/banyand/measure/migration_analyze.go
+++ b/banyand/measure/migration_analyze.go
@@ -48,14 +48,31 @@ import (
type AnalyzeGroupResult struct {
SamplesByVersion []AnalyzeKeyMulti
PerPartDups []AnalyzePerPart
+ ValueConflicts []AnalyzeValueConflict
TotalRows uint64
UniqueKeys uint64
DuplicateRows uint64
PerPartDupRows uint64
KeysWithDuplicates uint64 // count of (sid, ts) keys that appear in >1 physical rows
+ ValueConflictKeys uint64 // count of (sid, ts) keys with >1 row whose value digests differ
PartsScanned int
}
+// AnalyzeValueConflict is one (seriesID, timestamp) key that appears in more
+// than one index-mode sidx document whose value digests differ. Unlike a plain
+// version duplicate (where the value legitimately evolves with the version), a
+// value conflict flags documents that share the SAME key yet carry diverging
+// data values — the signal used to separate normal version dedup from an
+// abnormal value inconsistency or dropped field. Digests are version-EXCLUDED
+// data digests, so docs differing only by version digest identically and are not
+// reported as a conflict.
+type AnalyzeValueConflict struct {
+ Versions []int64
+ Digests []uint64
+ SeriesID uint64
+ Timestamp int64
+}
+
// 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
diff --git a/banyand/measure/migration_copy.go b/banyand/measure/migration_copy.go
index 573c29898..df4c6b5bc 100644
--- a/banyand/measure/migration_copy.go
+++ b/banyand/measure/migration_copy.go
@@ -76,24 +76,6 @@ func logStep(format string, args ...any) {
fmt.Fprintf(os.Stdout, time.Now().Format("2006/01/02 15:04:05")+" "+measureMigrationLogPrefix+" "+format+"\n", args...)
}
-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, ", "))
-}
-
var (
fastPathHits atomic.Int64
slowPathHits atomic.Int64
diff --git a/banyand/measure/migration_copy_test.go b/banyand/measure/migration_copy_test.go
deleted file mode 100644
index 3c8828a61..000000000
--- a/banyand/measure/migration_copy_test.go
+++ /dev/null
@@ -1,58 +0,0 @@
-// 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 (
- "strings"
- "testing"
-)
-
-// 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_executor.go b/banyand/measure/migration_executor.go
index d9b4901b2..4c86b5fa2 100644
--- a/banyand/measure/migration_executor.go
+++ b/banyand/measure/migration_executor.go
@@ -30,7 +30,9 @@ import (
// migration.CatalogExecutor: it copies / verifies measure groups one
// (entry, group) at a time under the shared orchestrator.
type MigrationExecutor struct {
- schemas map[string]map[string]*measureSchemaInfo
+ schemas map[string]map[string]*measureSchemaInfo
+ indexModeGroups map[string]bool
+ ruleByID map[uint32]indexRuleInfo
}
// NewMigrationExecutor returns the measure catalog executor.
@@ -45,12 +47,14 @@ func (e *MigrationExecutor) Catalog() commonv1.Catalog { return commonv1.Catalog
func (e *MigrationExecutor) LogPrefix() string { return measureMigrationLogPrefix }
// Prepare loads the measure schemas of the given groups from the
-// schema-property catalog and rejects IndexMode groups up front
-// (broadcasting a union sidx would corrupt cross-node dedup at query time).
+// schema-property catalog and classifies each group: an index-mode group
+// (its data lives in the segment sidx, routed per timestamp) opts out of the
+// union-sidx broadcast, while a group mixing index-mode measures with real
+// normal measures is unsupported and rejected up front.
func (e *MigrationExecutor) Prepare(_ context.Context, schemaRoot string, groups []string) error {
logStep("loading measure schemas")
//nolint:contextcheck // bluge reader.Search inside reader.WalkShard already uses its own context.
- schemas, err := loadMeasureSchemas(schemaRoot, groups)
+ schemas, err := loadMeasureSchemas(schemaRoot, groups) //nolint:contextcheck // offline bluge schema read, no cancellation
if err != nil {
return fmt.Errorf("load measure schemas: %w", err)
}
@@ -64,10 +68,21 @@ func (e *MigrationExecutor) Prepare(_ context.Context, schemaRoot string, groups
return fmt.Errorf("group %q has no measures in backup schema-property catalog — typo in groups or empty group?", g)
}
}
- if rejectErr := rejectIndexModeGroups(groups, schemas); rejectErr != nil {
- return rejectErr
+ indexModeGroups := make(map[string]bool, len(groups))
+ for _, g := range groups {
+ isIndexMode, classifyErr := classifyGroup(g, schemas[g])
+ if classifyErr != nil {
+ return classifyErr
+ }
+ indexModeGroups[g] = isIndexMode
+ }
+ ruleByID, err := loadIndexRuleInfoByID(schemaRoot, groups) //nolint:contextcheck // offline bluge schema read, no cancellation
+ if err != nil {
+ return fmt.Errorf("load index rules: %w", err)
}
e.schemas = schemas
+ e.indexModeGroups = indexModeGroups
+ e.ruleByID = ruleByID
// Reset path-hit counters so a previous in-process run's totals don't
// bleed into this one.
fastPathHits.Store(0)
@@ -76,9 +91,18 @@ func (e *MigrationExecutor) Prepare(_ context.Context, schemaRoot string, groups
return nil
}
+// SkipUnionSidx reports whether the orchestrator should skip the Phase A
+// union-sidx build + broadcast for the given group. Index-mode measure groups
+// opt out: their sidx docs are data points routed per timestamp, not
+// series-index metadata to broadcast.
+func (e *MigrationExecutor) SkipUnionSidx(group string) bool { return e.indexModeGroups[group] }
+
// CopyEntryGroup migrates one (entry, group): discovers source parts and
// rewrites them onto the target grid.
func (e *MigrationExecutor) CopyEntryGroup(ctx context.Context, in migration.EntryGroupInput) (migration.EntryGroupResult, error) {
+ if e.indexModeGroups[in.Group] {
+ return copyIndexModeGroup(ctx, in, e.ruleByID, e.schemas[in.Group])
+ }
var zero migration.EntryGroupResult
tasks, err := discoverPartTasks(ctx, in.SrcRoots)
if err != nil {
@@ -102,7 +126,15 @@ func (e *MigrationExecutor) CopyEntryGroup(ctx context.Context, in migration.Ent
// VerifyEntryGroup re-reads one (entry, group) read-only: it sums source
// rows, enumerates target segments and emits an EntryGroupReport.
-func (e *MigrationExecutor) VerifyEntryGroup(_ context.Context, in migration.EntryGroupInput, onReport func(report any)) error {
+func (e *MigrationExecutor) VerifyEntryGroup(ctx context.Context, in migration.EntryGroupInput, onReport func(report any)) error {
+ if e.indexModeGroups[in.Group] {
+ report, err := verifyIndexModeGroup(ctx, in, e.ruleByID, e.schemas[in.Group])
+ if err != nil {
+ return fmt.Errorf("index-mode verify: %w", err)
+ }
+ onReport(report)
+ return nil
+ }
fileSystem := fs.NewLocalFileSystem()
ir := in.Interval
targetGroup := in.TargetGroupRoot
diff --git a/banyand/measure/migration_indexmode_copy.go b/banyand/measure/migration_indexmode_copy.go
new file mode 100644
index 000000000..d6f2c51c0
--- /dev/null
+++ b/banyand/measure/migration_indexmode_copy.go
@@ -0,0 +1,1111 @@
+// 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/binary"
+ "fmt"
+ "hash/fnv"
+ "io"
+ "os"
+ "path/filepath"
+ "sort"
+ "strings"
+
+ "github.com/blugelabs/bluge"
+ blugesearch "github.com/blugelabs/bluge/search"
+
+ "github.com/apache/skywalking-banyandb/api/common"
+ "github.com/apache/skywalking-banyandb/banyand/internal/migration"
+ "github.com/apache/skywalking-banyandb/banyand/internal/storage"
+ "github.com/apache/skywalking-banyandb/pkg/convert"
+ "github.com/apache/skywalking-banyandb/pkg/index"
+ "github.com/apache/skywalking-banyandb/pkg/index/inverted"
+ "github.com/apache/skywalking-banyandb/pkg/logger"
+ pbv1 "github.com/apache/skywalking-banyandb/pkg/pb/v1"
+)
+
+// Reserved bluge stored-field names written by the inverted store's toDoc
+// (pkg/index/inverted/inverted.go:62-65). They are not tags and must be
+// handled explicitly rather than treated as tag fields.
+const (
+ imDocIDField = "_id"
+ imTimestampField = "_timestamp"
+ imVersionField = "_version"
+)
+
+// classifyGroup decides how to route a measure group. The internal
+// _top_n_result (TopNSchemaName) measure is auto-created by BanyanDB for every
+// measure group, so it never counts as a "real" normal measure. A group with
+// at least one index-mode measure and no real normal measure is an index-mode
+// group (isIndexMode=true); a group with no index-mode measure stays on the
+// existing normal path (false, nil); a group that mixes index-mode measures
+// with real normal measures is unsupported and returns an error listing the
+// offending normal measure names.
+func classifyGroup(group string, schemas map[string]*measureSchemaInfo) (isIndexMode bool, err error) {
+ var indexModes, realNormals []string
+ for name, m := range schemas {
+ if name == TopNSchemaName {
+ continue
+ }
+ if m.IndexMode {
+ indexModes = append(indexModes, name)
+ } else {
+ realNormals = append(realNormals, name)
+ }
+ }
+ if len(indexModes) == 0 {
+ return false, nil
+ }
+ if len(realNormals) > 0 {
+ sort.Strings(realNormals)
+ return false, fmt.Errorf("group %q mixes index-mode measures with non-_top_n_result normal measure(s) %v; "+
+ "mixed groups are unsupported", group, realNormals)
+ }
+ return true, nil
+}
+
+// collectTagNames gathers every tag name across the group's measures. A stored
+// field whose name is a known tag name is a non-indexed tag; this is what lets
+// classifyStoredField tell a 4-CHARACTER tag name apart from a 4-byte IndexRuleID
+// (both are 4 bytes wide, so name length alone is ambiguous).
+func collectTagNames(schemasBySubject map[string]*measureSchemaInfo) map[string]struct{} {
+ out := map[string]struct{}{}
+ for _, sc := range schemasBySubject {
+ if sc == nil {
+ continue
+ }
+ for _, tf := range sc.TagFamilies {
+ for _, tag := range tf.Tags {
+ out[tag] = struct{}{}
+ }
+ }
+ }
+ return out
+}
+
+// classifyStoredField reverses index.FieldKey.Marshal (pkg/index/index.go:73) for
+// one stored field name. A name that matches a schema tag name is a non-indexed
+// tag (TagName). Otherwise the name is a marshaled 4-byte IndexRuleID — an indexed
+// field — so it is kept INDEXED with that rule id; when the rule is missing from
+// ruleByID the field stays indexed (Analyzer/NoSort fall back to defaults) and the
+// id is returned as missingRuleID so the caller can warn rather than silently
+// dropping its searchability. A name that is neither a known tag nor 4 bytes wide
+// is unexpected; it is kept as a TagName best-effort.
+func classifyStoredField(name string, tagNames map[string]struct{}, ruleByID map[uint32]indexRuleInfo) (
+ key index.FieldKey, indexed bool, missingRuleID uint32,
+) {
+ if _, isTag := tagNames[name]; isTag {
+ return index.FieldKey{TagName: name}, false, 0
+ }
+ if len(name) == 4 {
+ id := convert.BytesToUint32([]byte(name))
+ if _, ok := ruleByID[id]; ok {
+ return index.FieldKey{IndexRuleID: id}, true, 0
+ }
+ return index.FieldKey{IndexRuleID: id}, true, id
+ }
+ return index.FieldKey{TagName: name}, false, 0
+}
+
+// readIndexModeDocs scans every committed bluge doc under sidxDir and rebuilds
+// each as an index.Document from TWO sources:
+//
+// A) stored fields -> regular tag fields (+ timestamp/version), and
+// B) the doc _id unmarshaled to a pbv1.Series -> regenerate the index-only
+// _im_name / _im_entity_tag_* fields, which the write path never stores.
+//
+// ruleByID restores Analyzer/NoSort on indexed fields; schemasBySubject maps
+// each series' Subject to its measure schema so the entity-derived fields are
+// regenerated with the right tag names/types and skip conditions.
+func readIndexModeDocs(ctx context.Context, sidxDir string, ruleByID map[uint32]indexRuleInfo,
+ schemasBySubject map[string]*measureSchemaInfo,
+) ([]index.Document, error) {
+ r, err := bluge.OpenReader(bluge.DefaultConfig(sidxDir))
+ if err != nil {
+ // A sidx directory can exist on disk with no committed bluge snapshot
+ // (a fresh segment whose writer never received a doc). Treat that as an
+ // empty sidx instead of failing the whole copy.
+ if strings.Contains(err.Error(), "unable to find a usable snapshot") {
+ return nil, nil
+ }
+ return nil, fmt.Errorf("open sidx reader %s: %w", sidxDir, err)
+ }
+ defer func() { _ = r.Close() }()
+ dmi, err := r.Search(ctx, bluge.NewAllMatches(bluge.NewMatchAllQuery()))
+ if err != nil {
+ return nil, fmt.Errorf("search sidx %s: %w", sidxDir, err)
+ }
+ var out []index.Document
+ var timeless []uint64
+ tagNames := collectTagNames(schemasBySubject)
+ missingRules := map[uint32]int{}
+ for {
+ match, nextErr := dmi.Next()
+ if nextErr != nil {
+ return nil, fmt.Errorf("iterate sidx %s: %w", sidxDir, nextErr)
+ }
+ if match == nil {
+ break
+ }
+ doc, buildErr := rebuildOneDoc(match, ruleByID, schemasBySubject, tagNames, missingRules)
+ if buildErr != nil {
+ return nil, fmt.Errorf("rebuild doc in %s: %w", sidxDir, buildErr)
+ }
+ // The write path always stamps a checked, non-zero _timestamp on every
+ // index-mode doc (write_standalone.go), so Timestamp==0 here means the
+ // stored _timestamp is missing or undecodable — corrupt or unexpected
+ // source data. Routing such a doc by a guessed segment boundary would
+ // silently misplace it, so collect the offenders (without retaining the
+ // corrupt docs) and surface them once the whole sidx has been scanned.
+ if doc.Timestamp == 0 {
+ timeless = append(timeless, doc.DocID)
+ continue
+ }
+ out = append(out, doc)
+ }
+ // Surface corrupt data before anything else: a missing-rule warning implies
+ // "continuing", which would be misleading when the read is about to abort.
+ if len(timeless) > 0 {
+ sample := timeless
+ if len(sample) > 10 {
+ sample = sample[:10]
+ }
+ return nil, fmt.Errorf("sidx %s: %d index-mode doc(s) have a missing or undecodable _timestamp (ts==0), "+
+ "which never occurs for valid data; the source is corrupt or unexpected — sample series IDs (first %d) %v",
+ sidxDir, len(timeless), len(sample), sample)
+ }
+ warnMissingRules(sidxDir, missingRules)
+ return out, nil
+}
+
+// warnMissingRules logs a single warning enumerating every IndexRuleID that an
+// indexed stored field decoded to but that is NOT registered in the schema-
+// property catalog, with its occurrence count. The fields are still kept indexed
+// (their Analyzer/NoSort fall back to defaults), but a non-empty list means the
+// catalog is missing index rules the data was written with, so the operator should
+// know rather than have a possibly-searchable field silently degraded.
+func warnMissingRules(sidxDir string, missingRules map[uint32]int) {
+ if len(missingRules) == 0 {
+ return
+ }
+ ids := make([]string, 0, len(missingRules))
+ total := 0
+ for id, count := range missingRules {
+ ids = append(ids, fmt.Sprintf("%d×%d", id, count))
+ total += count
+ }
+ sort.Strings(ids)
+ logger.GetLogger("measure-migration").Warn().
+ Str("sidx", sidxDir).
+ Int("distinctRules", len(missingRules)).
+ Int("totalFields", total).
+ Strs("ruleIDxCount", ids).
+ Msg("index-mode rebuild: indexed field(s) decode to an IndexRuleID missing from the schema-property " +
+ "catalog; kept indexed with default Analyzer/NoSort — restore the missing index rule(s) for exact fidelity")
+}
+
+// rebuildOneDoc reconstructs a single index.Document from a bluge match,
+// combining the two sources documented on readIndexModeDocs. tagNames is the
+// group's known tag-name set (to tell a tag from a rule id). missingRules, when
+// non-nil, accumulates (rule-id -> count) for fields that decode to an
+// IndexRuleID NOT present in ruleByID, so the caller can warn.
+func rebuildOneDoc(match *blugesearch.DocumentMatch, ruleByID map[uint32]indexRuleInfo,
+ schemasBySubject map[string]*measureSchemaInfo, tagNames map[string]struct{}, missingRules map[uint32]int,
+) (index.Document, error) {
+ var entityValues []byte
+ var ts, version int64
+ var fields []index.Field
+ visitErr := match.VisitStoredFields(func(name string, value []byte) bool {
+ switch name {
+ case imDocIDField:
+ entityValues = append([]byte(nil), value...)
+ case imTimestampField:
+ if dt, decErr := bluge.DecodeDateTime(value); decErr == nil {
+ ts = dt.UnixNano()
+ }
+ case imVersionField:
+ version = convert.BytesToInt64(value)
+ default:
+ key, indexed, missingRuleID := classifyStoredField(name, tagNames, ruleByID)
+ // NewBytesField clones value internally, so no extra copy of the
+ // reusable bluge visit buffer is needed here.
+ f := index.NewBytesField(key, value)
+ f.Store = true
+ f.Index = indexed
+ if ri, ok := ruleByID[key.IndexRuleID]; indexed && ok {
+ f.NoSort = ri.NoSort
+ f.Key.Analyzer = ri.Analyzer
+ }
+ // A field that decodes to an IndexRuleID the catalog does not know: it
+ // stays indexed (defaults), but the operator must be told rather than
+ // have a possibly-searchable field silently degraded.
+ if missingRuleID != 0 && missingRules != nil {
+ missingRules[missingRuleID]++
+ }
+ fields = append(fields, f)
+ }
+ return true
+ })
+ if visitErr != nil {
+ return index.Document{}, fmt.Errorf("visit stored fields: %w", visitErr)
+ }
+ // Source B: rebuild the index-only entity-derived fields from _id.
+ var series pbv1.Series
+ if err := series.Unmarshal(entityValues); err != nil {
+ return index.Document{}, fmt.Errorf("unmarshal series from _id: %w", err)
+ }
+ fields = appendRegeneratedEntityFields(fields, &series, schemasBySubject[series.Subject])
+ return index.Document{
+ Fields: fields,
+ EntityValues: entityValues,
+ Timestamp: ts,
+ DocID: uint64(series.ID),
+ Version: version,
+ }, nil
+}
+
+// appendRegeneratedEntityFields rebuilds the index-only fields the production
+// write path adds in appendEntityTagsToIndexFields (write_standalone.go:410):
+// _im_name (subject) and _im_entity_tag_<tag> (entity values). These are never
+// stored, so they must be regenerated from the series, or entity-tag / measure-
+// name search breaks after migration.
+//
+// The entity-tag fields replicate the production skip condition: an entity tag
+// already indexed by an index rule is not re-emitted as _im_entity_tag_*.
+func appendRegeneratedEntityFields(fields []index.Field, series *pbv1.Series, schema *measureSchemaInfo) []index.Field {
+ subj := index.NewStringField(index.FieldKey{TagName: index.IndexModeName}, series.Subject)
+ subj.Index = true
+ subj.NoSort = true
+ fields = append(fields, subj)
+ if schema == nil {
+ return fields
+ }
+ for i, tagName := range schema.EntityTagNames {
+ if i >= len(series.EntityValues) {
+ break
+ }
+ if _, indexed := schema.IndexedEntityTags[tagName]; indexed {
+ continue
+ }
+ nv := encodeTagValue(tagName, schema.TagType[tagName], series.EntityValues[i])
+ value := nv.value
+ releaseNameValue(nv)
+ if value == nil {
+ continue
+ }
+ f := index.NewBytesField(index.FieldKey{TagName: index.IndexModeEntityTagPrefix + tagName}, value)
+ f.Index = true
+ f.NoSort = true
+ fields = append(fields, f)
+ }
+ return fields
+}
+
+// ── Target idx store pool ────────────────────────────────────────────────────.
+
+// targetIdxStore lazily opens one inverted store per target sidx path
+// (BatchWaitSec:0) and closes them all at the end, so multiple source segs
+// feeding the same target sidx share a single writer. Copied (cross-package,
+// unexported) from stream/migration_element_index.go:126-158.
+type targetIdxStore struct {
+ stores map[string]index.SeriesStore
+}
+
+func newTargetIdxStore() *targetIdxStore {
+ return &targetIdxStore{stores: map[string]index.SeriesStore{}}
+}
+
+func (t *targetIdxStore) get(path string) (index.SeriesStore, error) {
+ if s, ok := t.stores[path]; ok {
+ return s, nil
+ }
+ if err := os.MkdirAll(path, storage.DirPerm); err != nil {
+ return nil, fmt.Errorf("mkdir idx %s: %w", path, err)
+ }
+ s, err := inverted.NewStore(inverted.StoreOpts{Path: path, BatchWaitSec: 0})
+ if err != nil {
+ return nil, fmt.Errorf("open idx store %s: %w", path, err)
+ }
+ t.stores[path] = s
+ return s, nil
+}
+
+func (t *targetIdxStore) closeAll() error {
+ var firstErr error
+ for path, s := range t.stores {
+ if err := s.Close(); err != nil && firstErr == nil {
+ firstErr = fmt.Errorf("close idx store %s: %w", path, err)
+ }
+ }
+ t.stores = map[string]index.SeriesStore{}
+ return firstErr
+}
+
+// ── Segment routing ──────────────────────────────────────────────────────────.
+
+// ── Slow path: rebuild + route + merge keeping max version ───────────────────.
+
+// copyIndexModeSlowDocs routes already-read docs (docsByDir, keyed by source
+// sidx dir) by their _timestamp to the aligned target seg's sidx, deduplicates
+// per series (DocID) keeping the highest Version within each target seg, and
+// upserts the survivors via UpdateSeriesBatch. The target seg's <segDir>/metadata
+// is written too, since a segment without it is treated as invalid and removed
+// at startup (storage/segment.go open loop). Routing all slow-path sources
+// through a single call is what makes the max-version dedup work ACROSS sources
+// that merge into the same target seg. Returns the number
+// of rows written.
+func copyIndexModeSlowDocs(ctx context.Context, srcSidxDirs []string, docsByDir map[string][]index.Document,
+ dstGroupRoot string, ir storage.IntervalRule, stores *targetIdxStore,
+) (rows int64, err error) {
+ // per target sidx path -> seriesID(DocID) -> survivor doc (max version).
+ perTarget := map[string]map[uint64]index.Document{}
+ // targetSegName per sidx path, to write the seg metadata once flushed.
+ segNameByPath := map[string]string{}
+ var segCache alignedSegCache
+ for _, dir := range srcSidxDirs {
+ docs := docsByDir[dir]
+ for di := range docs {
+ d := docs[di]
+ seg := segCache.segNameFor(ir, d.Timestamp)
+ dstSidx := filepath.Join(dstGroupRoot, seg, directCopySidxDirName)
+ segNameByPath[dstSidx] = seg
+ m := perTarget[dstSidx]
+ if m == nil {
+ m = map[uint64]index.Document{}
+ perTarget[dstSidx] = m
+ }
+ if prev, ok := m[d.DocID]; !ok || d.Version >= prev.Version {
+ m[d.DocID] = d
+ }
+ }
+ }
+ for dstSidx, m := range perTarget {
+ if ctx.Err() != nil {
+ return rows, ctx.Err()
+ }
+ store, getErr := stores.get(dstSidx)
+ if getErr != nil {
+ return rows, getErr
+ }
+ batch := index.Batch{Documents: make(index.Documents, 0, len(m))}
+ for _, d := range m {
+ batch.Documents = append(batch.Documents, d)
+ rows++
+ }
+ if updateErr := store.UpdateSeriesBatch(batch); updateErr != nil {
+ return rows, fmt.Errorf("write target sidx %s: %w", dstSidx, updateErr)
+ }
+ segDir := filepath.Join(dstGroupRoot, segNameByPath[dstSidx])
+ if _, metaErr := writeIndexModeSegMetadata(segDir, ir); metaErr != nil {
+ return rows, metaErr
+ }
+ }
+ return rows, nil
+}
+
+// sortedSegList renders a target-segment set as a sorted comma-separated string
+// for deterministic per-source rebuild log lines.
+func sortedSegList(segs map[string]struct{}) string {
+ names := make([]string, 0, len(segs))
+ for s := range segs {
+ names = append(names, s)
+ }
+ sort.Strings(names)
+ return strings.Join(names, ",")
+}
+
+// ── Top-level orchestration ──────────────────────────────────────────────────.
+
+// copyIndexModeGroup copies one (entry, group) for an index-mode measure group.
+// It rejects groups carrying shard parts (a non-_top_n_result normal measure
+// has data), then for each source seg sidx either byte-copies the whole sidx
+// directory (fast path: all docs align to one not-yet-written target seg) or
+// rebuilds and routes docs through the slow path. Each touched target segment
+// gets its <segDir>/metadata so the runtime loads it instead of pruning it.
+func copyIndexModeGroup(ctx context.Context, in migration.EntryGroupInput,
+ ruleByID map[uint32]indexRuleInfo, schemasBySubject map[string]*measureSchemaInfo,
+) (res migration.EntryGroupResult, err error) {
+ hasPart, err := hasShardPart(ctx, in.SrcRoots)
+ if err != nil {
+ return res, fmt.Errorf("scan for shard parts: %w", err)
+ }
+ if hasPart {
+ return res, fmt.Errorf("index-mode group %q has shard part(s) on disk "+
+ "(a non-_top_n_result normal measure has data?); refusing", in.Group)
+ }
+
+ srcSidxDirs, err := collectSourceSidxDirs(in.SrcRoots)
+ if err != nil {
+ return res, err
+ }
+ if len(srcSidxDirs) == 0 {
+ logStep("%s stage=%s group %s: skipped (no sidx in selected nodes)",
+ in.EntryTag, in.Entry.Stage, in.Group)
+ return res, nil
+ }
+
+ stores := newTargetIdxStore()
+ // Close() flushes buffered batches and lands bluge's own snapshot; a close
+ // failure means an incomplete sidx on disk, so it must surface even on the
+ // happy path.
+ defer func() {
+ if cerr := stores.closeAll(); cerr != nil && err == nil {
+ err = cerr
+ }
+ }()
+
+ // Pass 1: read every source's docs once and compute the FULL set of target
+ // segs each source routes docs into (not just "single target or not").
+ // docsByDir caches the docs so the slow path never re-reads them.
+ // targetSegsBySrc holds each source's target-seg set; segSrcCount counts how
+ // many sources route ANY doc into each target seg — split sources included —
+ // so byte-copy exclusivity is judged against every writer, not just other
+ // single-target sources.
+ docsByDir := make(map[string][]index.Document, len(srcSidxDirs))
+ targetSegsBySrc := make(map[string]map[string]struct{}, len(srcSidxDirs))
+ segSrcCount := map[string]int{}
+ for _, srcSidx := range srcSidxDirs {
+ if ctx.Err() != nil {
+ return res, ctx.Err()
+ }
+ docs, readErr := readIndexModeDocs(ctx, srcSidx, ruleByID, schemasBySubject)
+ if readErr != nil {
+ return res, readErr
+ }
+ if len(docs) == 0 {
+ continue
+ }
+ docsByDir[srcSidx] = docs
+ segs := make(map[string]struct{})
+ var segCache alignedSegCache
+ for di := range docs {
+ segs[segCache.segNameFor(in.Interval, docs[di].Timestamp)] = struct{}{}
+ }
+ targetSegsBySrc[srcSidx] = segs
+ for seg := range segs {
+ segSrcCount[seg]++
+ }
+ }
+
+ // Pass 2: a source is byte-copy eligible iff all its docs fall into ONE
+ // target seg AND that target seg has no OTHER source routing any doc into it
+ // (exclusive — split sources count) AND the target sidx is not already on
+ // disk. Every other source is routed through a SINGLE copyIndexModeSlowDocs
+ // call so max-version dedup spans all sources merging into the same target
+ // seg.
+ var slowDirs []string
+ for _, srcSidx := range srcSidxDirs {
+ docs := docsByDir[srcSidx]
+ if len(docs) == 0 {
+ continue
+ }
+ // A source byte-copies only when its docs all fall in ONE target seg that
+ // no other source touches. The single element is read inside the len==1
+ // guard so a multi-target (split) source never reads a stale seg value.
+ if segs := targetSegsBySrc[srcSidx]; len(segs) == 1 {
+ var seg string
+ for s := range segs {
+ seg = s
+ }
+ if segSrcCount[seg] == 1 {
+ dstSidx := filepath.Join(in.TargetGroupRoot, seg, directCopySidxDirName)
+ // The dirty-target pre-flight (orchestrator.RunCopy) guarantees the
+ // target group dir is empty at run start, so a missing dstSidx means
+ // no other source wrote it either — a true exclusive byte-copy.
+ if _, statErr := os.Stat(dstSidx); os.IsNotExist(statErr) {
+ n, copyErr := migration.CopyDir(srcSidx, dstSidx)
+ if copyErr != nil {
+ return res, fmt.Errorf("byte-copy sidx %s -> %s: %w", srcSidx, dstSidx, copyErr)
+ }
+ segDir := filepath.Join(in.TargetGroupRoot, seg)
+ metaBytes, metaErr := writeIndexModeSegMetadata(segDir, in.Interval)
+ if metaErr != nil {
+ return res, metaErr
+ }
+ res.Bytes += n + metaBytes
+ res.Rows += int64(len(docs))
+ logStep("%s stage=%s group %s: sidx %s -> byte-copy into seg %s (%d docs)",
+ in.EntryTag, in.Entry.Stage, in.Group, srcSidx, seg, len(docs))
+ continue
+ }
+ }
+ }
+ // Per-source visibility: log that this source segment takes the rebuild
+ // (read-docs-then-write) path and which target segment(s) its docs route
+ // into, the counterpart of the byte-copy line above.
+ logStep("%s stage=%s group %s: sidx %s -> rebuild into seg(s) %s (%d docs)",
+ in.EntryTag, in.Entry.Stage, in.Group, srcSidx,
+ sortedSegList(targetSegsBySrc[srcSidx]), len(docs))
+ slowDirs = append(slowDirs, srcSidx)
+ }
+
+ if len(slowDirs) > 0 {
+ rows, slowErr := copyIndexModeSlowDocs(ctx, slowDirs, docsByDir, in.TargetGroupRoot, in.Interval, stores)
+ if slowErr != nil {
+ return res, slowErr
+ }
+ res.Rows += rows
+ logStep("%s stage=%s group %s: %d sidx -> rebuild total (%d rows after max-version merge)",
+ in.EntryTag, in.Entry.Stage, in.Group, len(slowDirs), rows)
+ }
+
+ // Flush the slow-path stores before counting segments so every target sidx
+ // (and its bluge snapshot) is on disk. closeAll is idempotent; the deferred
+ // call becomes a no-op.
+ if closeErr := stores.closeAll(); closeErr != nil {
+ return res, closeErr
+ }
+ res.Segments = countTargetSegs(in.TargetGroupRoot)
+ return res, nil
+}
+
+// hasShardPart reports whether any source root holds a measure shard PART
+// (seg-*/shard-*/<partID>). Index-mode groups keep data only in the segment-level
+// sidx, so an actual part means a non-_top_n_result normal measure has on-disk
+// data, which the index-mode copy path refuses. An EMPTY shard-* directory is
+// benign: the segment controller can create shard-0 eagerly (or a dataless
+// _top_n_result leaves one behind) without any part landing in it, so the guard
+// keys off real parts — matching directCopyPartDirPattern, exactly what
+// discoverPartTasks consumes — not bare directory existence. It short-circuits on
+// the first part instead of building the full task list.
+func hasShardPart(ctx context.Context, srcRoots []string) (bool, error) {
+ for _, root := range srcRoots {
+ if ctx.Err() != nil {
+ return false, ctx.Err()
+ }
+ segEntries, err := os.ReadDir(root)
+ if err != nil {
+ if os.IsNotExist(err) {
+ continue
+ }
+ return false, fmt.Errorf("read %s: %w", root, err)
+ }
+ for _, segE := range segEntries {
+ if !segE.IsDir() || !strings.HasPrefix(segE.Name(), directCopySegPrefix) {
+ continue
+ }
+ shardEntries, err := os.ReadDir(filepath.Join(root, segE.Name()))
+ if err != nil {
+ return false, fmt.Errorf("read %s: %w", filepath.Join(root, segE.Name()), err)
+ }
+ for _, shardE := range shardEntries {
+ if !shardE.IsDir() || !strings.HasPrefix(shardE.Name(), directCopyShardPrefix) {
+ continue
+ }
+ shardDir := filepath.Join(root, segE.Name(), shardE.Name())
+ partEntries, err := os.ReadDir(shardDir)
+ if err != nil {
+ return false, fmt.Errorf("read %s: %w", shardDir, err)
+ }
+ for _, partE := range partEntries {
+ if partE.IsDir() && directCopyPartDirPattern.MatchString(partE.Name()) {
+ return true, nil
+ }
+ }
+ }
+ }
+ }
+ return false, nil
+}
+
+// collectSourceSidxDirs lists every "<root>/seg-*/sidx" directory across the
+// entry's source roots. A seg directory without a sidx subdir is skipped.
+func collectSourceSidxDirs(srcRoots []string) ([]string, error) {
+ var out []string
+ for _, root := range srcRoots {
+ segEntries, err := os.ReadDir(root)
+ if err != nil {
+ if os.IsNotExist(err) {
+ continue
+ }
+ return nil, fmt.Errorf("read %s: %w", root, err)
+ }
+ for _, segE := range segEntries {
+ if !segE.IsDir() || !strings.HasPrefix(segE.Name(), directCopySegPrefix) {
+ continue
+ }
+ sidxDir := filepath.Join(root, segE.Name(), directCopySidxDirName)
+ if info, statErr := os.Stat(sidxDir); statErr == nil && info.IsDir() {
+ out = append(out, sidxDir)
+ }
+ }
+ }
+ sort.Strings(out)
+ return out, nil
+}
+
+// countTargetSegs counts "seg-*" directories under the target group root.
+func countTargetSegs(dstGroupRoot string) int {
+ entries, err := os.ReadDir(dstGroupRoot)
+ if err != nil {
+ return 0
+ }
+ n := 0
+ for _, e := range entries {
+ if e.IsDir() && strings.HasPrefix(e.Name(), directCopySegPrefix) {
+ n++
+ }
+ }
+ return n
+}
+
+// writeIndexModeSegMetadata writes <segDir>/metadata for an index-mode target
+// segment. The segment's end boundary is the target grid's NextTime after the
+// segment start parsed from its directory name. Without this file the runtime
+// prunes the segment as invalid at startup.
+func writeIndexModeSegMetadata(segDir string, ir storage.IntervalRule) (int64, error) {
+ segName := filepath.Base(segDir)
+ start, err := parseDirectCopySegStart(segName, ir.Unit)
+ if err != nil {
+ return 0, fmt.Errorf("parse target seg start %s: %w", segName, err)
+ }
+ return writeDirectCopySegmentMetadata(segDir, ir.NextTime(start))
+}
+
+// ── Phase 3 / 4: value digest, verify, analyze ───────────────────────────────.
+
+// docValueDigest hashes ALL data values of one rebuilt index-mode document so
+// that any changed value or any dropped field flips the digest. It covers:
+//
+// - Version, and
+// - every field's (Key.Marshal(), value bytes).
+//
+// The regenerated index-only fields (_im_name / _im_entity_tag_*) are part of
+// Fields by the time docValueDigest runs, so a doc that lost an entity-derived
+// value produces a different digest than the intact source doc.
+// The (name, value) pairs are sorted before hashing, so the digest is
+// independent of field order — two docs equal in content hash identically even
+// if their Fields slices were assembled in a different sequence.
+func docValueDigest(d index.Document) uint64 {
+ return docDigest(d, true)
+}
+
+// docDataDigest hashes only the document's data values (every field's
+// (Key.Marshal(), bytes)) and EXCLUDES Version. analyze's value-conflict check
+// uses it to tell apart a normal version duplicate (same data, version merely
+// bumped — identical data digest) from a genuine value conflict (same (series,
+// timestamp) but diverging data — different data digest).
+func docDataDigest(d index.Document) uint64 {
+ return docDigest(d, false)
+}
+
+// docDigest is the shared field-hashing core. The (name, value) pairs are sorted
+// before hashing, so the digest is independent of field order. When
+// includeVersion is true the document Version is folded in first; the
+// regenerated index-only fields (_im_name / _im_entity_tag_*) are part of Fields
+// by the time this runs, so a doc that lost an entity-derived value digests
+// differently from the intact one.
+func docDigest(d index.Document, includeVersion bool) uint64 {
+ type nv struct {
+ name string
+ value []byte
+ }
+ pairs := make([]nv, 0, len(d.Fields))
+ for i := range d.Fields {
+ pairs = append(pairs, nv{name: d.Fields[i].Key.Marshal(), value: d.Fields[i].GetBytes()})
+ }
+ sort.Slice(pairs, func(i, j int) bool {
+ if pairs[i].name != pairs[j].name {
+ return pairs[i].name < pairs[j].name
+ }
+ return string(pairs[i].value) < string(pairs[j].value)
+ })
+ h := fnv.New64a()
+ var scratch [8]byte
+ if includeVersion {
+ binary.LittleEndian.PutUint64(scratch[:], uint64(d.Version))
+ _, _ = h.Write(scratch[:])
+ }
+ for i := range pairs {
+ binary.LittleEndian.PutUint64(scratch[:], uint64(len(pairs[i].name)))
+ _, _ = h.Write(scratch[:])
+ _, _ = io.WriteString(h, pairs[i].name)
+ binary.LittleEndian.PutUint64(scratch[:], uint64(len(pairs[i].value)))
+ _, _ = h.Write(scratch[:])
+ _, _ = h.Write(pairs[i].value)
+ }
+ return h.Sum64()
+}
+
+// imValueKey identifies one logical data point in an index-mode group: the
+// series (DocID) plus its timestamp. Only analyze keys on this; verify keys on
+// imSegKey (segment, series), matching the runtime's per-segment upsert.
+type imValueKey struct {
+ sid uint64
+ ts int64
+}
+
+// imSegKey identifies one logical index-mode data point for verify
+// reconciliation: the target segment plus the series (DocID). The runtime stores
+// index-mode data as an upsert-by-series-PER-SEGMENT, so the correct unit is
+// (segment, series) — NOT (series, timestamp). When several source segments
+// merge into one coarser target segment, each series collapses to a single
+// max-version doc, so two source docs of one series at different timestamps are
+// EXPECTED to become one; a (series, timestamp) key would mis-report the dropped
+// one as "missing".
+type imSegKey struct {
+ seg string
+ sid uint64
+}
+
+// segSurvivor is the max-version doc kept for an imSegKey: its version drives the
+// max-version reduction, its digest the value comparison.
+type segSurvivor struct {
+ version int64
+ digest uint64
+}
+
+// collectTargetSidxDirs lists every "<groupRoot>/seg-*/sidx" directory under one
+// target group root. It is the target-side analog of collectSourceSidxDirs and
+// is what the index-mode verify reads to count migrated documents.
+func collectTargetSidxDirs(groupRoot string) ([]string, error) {
+ return collectSourceSidxDirs([]string{groupRoot})
+}
+
+// readAllIndexModeDocs reads and concatenates the docs of every sidx dir in dirs,
+// reusing the double-source rebuild so callers see the regenerated index-only
+// fields too. analyze uses it to group docs by (series, timestamp) across every
+// source segment.
+func readAllIndexModeDocs(ctx context.Context, dirs []string, ruleByID map[uint32]indexRuleInfo,
+ schemasBySubject map[string]*measureSchemaInfo,
+) ([]index.Document, error) {
+ var out []index.Document
+ for _, dir := range dirs {
+ docs, err := readIndexModeDocs(ctx, dir, ruleByID, schemasBySubject)
+ if err != nil {
+ return nil, err
+ }
+ out = append(out, docs...)
+ }
+ return out, nil
+}
+
+// segDigestResult is segKeyedDigests' output: max-version digests keyed by
+// (segment, series), the distinct-series set, the total doc count, the per-target
+// -segment doc count, and (source side only) the set of source segments that
+// routed docs into each target segment — used to report per-segment alignment.
+type segDigestResult struct {
+ byKey map[imSegKey]segSurvivor
+ distinctIDs map[uint64]struct{}
+ docsPerSeg map[string]uint64
+ srcSegsPerTarget map[string]map[string]struct{}
+ total uint64
+}
+
+// segKeyedDigests reads every dir's docs and reduces them to one max-version
+// value digest per (target segment, series). When routed is true the segment is
+// where each doc ROUTES under ir — the source side, mirroring the copy router —
+// so it matches the target's already-laid-out segments and records which source
+// segment fed each target; otherwise the segment is the dir's own seg-* name (the
+// target side).
+func segKeyedDigests(ctx context.Context, dirs []string, ruleByID map[uint32]indexRuleInfo,
+ schemasBySubject map[string]*measureSchemaInfo, ir storage.IntervalRule, routed bool,
+) (*segDigestResult, error) {
+ res := &segDigestResult{
+ byKey: map[imSegKey]segSurvivor{},
+ distinctIDs: map[uint64]struct{}{},
+ docsPerSeg: map[string]uint64{},
+ srcSegsPerTarget: map[string]map[string]struct{}{},
+ }
+ var segCache alignedSegCache
+ for _, dir := range dirs {
+ docs, readErr := readIndexModeDocs(ctx, dir, ruleByID, schemasBySubject)
+ if readErr != nil {
+ return nil, readErr
+ }
+ dirSeg := filepath.Base(filepath.Dir(dir)) // seg-* name is the sidx dir's parent
+ for i := range docs {
+ d := docs[i]
+ res.total++
+ res.distinctIDs[d.DocID] = struct{}{}
+ seg := dirSeg
+ if routed {
+ seg = segCache.segNameFor(ir, d.Timestamp)
+ srcSet := res.srcSegsPerTarget[seg]
+ if srcSet == nil {
+ srcSet = map[string]struct{}{}
+ res.srcSegsPerTarget[seg] = srcSet
+ }
+ srcSet[dirSeg] = struct{}{}
+ }
+ res.docsPerSeg[seg]++
+ k := imSegKey{seg: seg, sid: d.DocID}
+ if prev, ok := res.byKey[k]; !ok || d.Version >= prev.version {
+ res.byKey[k] = segSurvivor{version: d.Version, digest: docValueDigest(d)}
+ }
+ }
+ }
+ return res, nil
+}
+
+// buildSegAligns reports, per target segment, how many source segments fed it,
+// how many source docs routed in, and how many target docs survived. A segment
+// is "aligned" — a true 1:1 byte-copy candidate, matching the copy's exclusivity
+// rule — iff exactly ONE source segment fed it AND that source segment fed NO
+// OTHER target (a clean 1:1 pairing) AND no docs collapsed. A source segment that
+// splits across several targets disqualifies every target it touches, even one it
+// is the sole contributor to.
+func buildSegAligns(src, tgt *segDigestResult) []IndexModeSegAlign {
+ // Count how many distinct target segments each source segment feeds.
+ srcSegTargets := map[string]int{}
+ for targetSeg := range src.srcSegsPerTarget {
+ for srcSeg := range src.srcSegsPerTarget[targetSeg] {
+ srcSegTargets[srcSeg]++
+ }
+ }
+ segs := make(map[string]struct{}, len(src.docsPerSeg))
+ for s := range src.docsPerSeg {
+ segs[s] = struct{}{}
+ }
+ for s := range tgt.docsPerSeg {
+ segs[s] = struct{}{}
+ }
+ out := make([]IndexModeSegAlign, 0, len(segs))
+ for s := range segs {
+ srcDocs := src.docsPerSeg[s]
+ tgtDocs := tgt.docsPerSeg[s]
+ sources := src.srcSegsPerTarget[s]
+ aligned := false
+ if len(sources) == 1 && srcDocs == tgtDocs {
+ for only := range sources {
+ aligned = srcSegTargets[only] == 1
+ }
+ }
+ out = append(out, IndexModeSegAlign{
+ Segment: s, SrcSegs: len(sources), SrcDocs: srcDocs, TgtDocs: tgtDocs, Aligned: aligned,
+ })
+ }
+ sort.Slice(out, func(i, j int) bool { return out[i].Segment < out[j].Segment })
+ return out
+}
+
+// verifyIndexModeGroup reconciles one index-mode (entry, group) at the document
+// level:
+//
+// ① counts: source/target total docs + distinct doc-id. With aligned, non-
+// merging copy the distinct doc-id count matches exactly; when several
+// source segments merge into one target segment the target collapses each
+// series to one upserted doc, so the expected target distinct doc-id count
+// is the UNION of the source doc-ids (computed here, not a sum).
+// ② values: per (series, timestamp) digest (source side reduced to max
+// version first). A key in source but not target is "missing", in target
+// but not source is "extra", present on both with differing digest is
+// "tampered" (a changed value or a dropped index-only field).
+//
+// The report reuses EntryGroupReport with its IndexMode-prefixed fields filled.
+func verifyIndexModeGroup(ctx context.Context, in migration.EntryGroupInput,
+ ruleByID map[uint32]indexRuleInfo, schemasBySubject map[string]*measureSchemaInfo,
+) (EntryGroupReport, error) {
+ report := EntryGroupReport{
+ Group: in.Group,
+ EntryStage: in.Entry.Stage,
+ EntryTarget: in.Entry.Target,
+ EntryNodes: in.Entry.Nodes,
+ SrcRoots: in.SrcRoots,
+ TargetGroup: in.TargetGroupRoot,
+ IndexMode: true,
+ }
+
+ srcSidxDirs, err := collectSourceSidxDirs(in.SrcRoots)
+ if err != nil {
+ return report, fmt.Errorf("collect source sidx dirs: %w", err)
+ }
+ tgtSidxDirs, err := collectTargetSidxDirs(in.TargetGroupRoot)
+ if err != nil {
+ return report, fmt.Errorf("collect target sidx dirs: %w", err)
+ }
+
+ // Source docs are keyed by the segment they ROUTE to (mirroring the copy
+ // router); target docs by the segment they already live in. Both reduce to
+ // one max-version doc per (segment, series), so the two sides are directly
+ // comparable even when the copy merged several source segments into one.
+ src, err := segKeyedDigests(ctx, srcSidxDirs, ruleByID, schemasBySubject, in.Interval, true)
+ if err != nil {
+ return report, fmt.Errorf("read source docs: %w", err)
+ }
+ tgt, err := segKeyedDigests(ctx, tgtSidxDirs, ruleByID, schemasBySubject, in.Interval, false)
+ if err != nil {
+ return report, fmt.Errorf("read target docs: %w", err)
+ }
+
+ report.SrcDocs = src.total
+ report.TgtDocs = tgt.total
+ report.SrcDistinctIDs = uint64(len(src.distinctIDs))
+ report.TgtDistinctIDs = uint64(len(tgt.distinctIDs))
+ // Expected target distinct doc-id is the union of source doc-ids: under
+ // upsert merge it equals the source distinct count, while a no-merge copy
+ // trivially keeps the same set — one formula covers both.
+ report.ExpectDistinctIDs = uint64(len(src.distinctIDs))
+
+ report.ValueMismatches = diffValueDigests(src.byKey, tgt.byKey)
+ report.SegAligns = buildSegAligns(src, tgt)
+ return report, nil
+}
+
+// diffValueDigests reconciles two (segment, series) → max-version digest maps and
+// returns every key that did not match, tagged missing / extra / tampered.
+// Results are sorted (segment, sid) for deterministic reporting.
+func diffValueDigests(src, tgt map[imSegKey]segSurvivor) []IndexModeValueMismatch {
+ var out []IndexModeValueMismatch
+ for k, sv := range src {
+ tv, ok := tgt[k]
+ switch {
+ case !ok:
+ out = append(out, IndexModeValueMismatch{Kind: "missing", Segment: k.seg, SeriesID: k.sid})
+ case sv.digest != tv.digest:
+ out = append(out, IndexModeValueMismatch{Kind: "tampered", Segment: k.seg, SeriesID: k.sid})
+ }
+ }
+ for k := range tgt {
+ if _, ok := src[k]; !ok {
+ out = append(out, IndexModeValueMismatch{Kind: "extra", Segment: k.seg, SeriesID: k.sid})
+ }
+ }
+ sort.Slice(out, func(i, j int) bool {
+ if out[i].Segment != out[j].Segment {
+ return out[i].Segment < out[j].Segment
+ }
+ if out[i].SeriesID != out[j].SeriesID {
+ return out[i].SeriesID < out[j].SeriesID
+ }
+ return out[i].Kind < out[j].Kind
+ })
+ return out
+}
+
+// IsIndexModeGroup loads the group's measure schemas from schemaRoot and reports
+// whether it is an index-mode group, so the analyze CLI can pick the sidx-based
+// path. It returns an error for an unsupported mixed group (the same rule
+// classifyGroup enforces during copy/verify).
+func IsIndexModeGroup(schemaRoot, group string) (bool, error) {
+ schemas, err := loadMeasureSchemas(schemaRoot, []string{group})
+ if err != nil {
+ return false, fmt.Errorf("load measure schemas: %w", err)
+ }
+ if len(schemas[group]) == 0 {
+ return false, fmt.Errorf("group %q has no measures in backup schema-property catalog", group)
+ }
+ return classifyGroup(group, schemas[group])
+}
+
+// AnalyzeIndexModeGroup is the exported entry the analyze CLI calls for an
+// index-mode group: it loads the group's schemas + index rules, collects every
+// source sidx dir under srcRoots, and runs the sidx-document analysis.
+func AnalyzeIndexModeGroup(ctx context.Context, schemaRoot, group string, srcRoots []string, sampleCap int) (AnalyzeGroupResult, error) {
+ var res AnalyzeGroupResult
+ schemas, err := loadMeasureSchemas(schemaRoot, []string{group}) //nolint:contextcheck // offline bluge schema read, no cancellation
+ if err != nil {
+ return res, fmt.Errorf("load measure schemas: %w", err)
+ }
+ ruleByID, err := loadIndexRuleInfoByID(schemaRoot, []string{group}) //nolint:contextcheck // offline bluge schema read, no cancellation
+ if err != nil {
+ return res, fmt.Errorf("load index rules: %w", err)
+ }
+ srcSidxDirs, err := collectSourceSidxDirs(srcRoots)
+ if err != nil {
+ return res, fmt.Errorf("collect source sidx dirs: %w", err)
+ }
+ return analyzeIndexModeGroup(ctx, srcSidxDirs, ruleByID, schemas[group], sampleCap)
+}
+
+// analyzeIndexModeGroup is the index-mode flavor of AnalyzeGroupRows: it reads
+// the group's sidx documents (not part block metadata) and
+// reports the same shape — total docs, distinct (series, timestamp) keys,
+// version-duplicate rows, sample keys — PLUS value conflicts. A value conflict
+// is a (series, timestamp) key that appears in more than one document whose
+// value digests differ; that distinguishes a normal version-evolved duplicate
+// from an abnormal value inconsistency or a dropped field.
+func analyzeIndexModeGroup(ctx context.Context, srcSidxDirs []string, ruleByID map[uint32]indexRuleInfo,
+ schemasBySubject map[string]*measureSchemaInfo, sampleCap int,
+) (AnalyzeGroupResult, error) {
+ var res AnalyzeGroupResult
+ docs, err := readAllIndexModeDocs(ctx, srcSidxDirs, ruleByID, schemasBySubject)
+ if err != nil {
+ return res, fmt.Errorf("read index-mode docs: %w", err)
+ }
+
+ type rowInfo struct {
+ version int64
+ digest uint64
+ }
+ byKey := make(map[imValueKey][]rowInfo, len(docs))
+ for i := range docs {
+ d := docs[i]
+ res.TotalRows++
+ k := imValueKey{sid: d.DocID, ts: d.Timestamp}
+ // Conflict detection compares data values WITHOUT version: two docs that
+ // differ only by version (normal evolution) must NOT count as a conflict.
+ byKey[k] = append(byKey[k], rowInfo{version: d.Version, digest: docDataDigest(d)})
+ }
+ res.UniqueKeys = uint64(len(byKey))
+ res.DuplicateRows = res.TotalRows - res.UniqueKeys
+ res.PartsScanned = len(srcSidxDirs)
+
+ keys := make([]imValueKey, 0, len(byKey))
+ for k := range byKey {
+ keys = append(keys, k)
+ }
+ sort.Slice(keys, func(i, j int) bool {
+ if keys[i].sid != keys[j].sid {
+ return keys[i].sid < keys[j].sid
+ }
+ return keys[i].ts < keys[j].ts
+ })
+
+ for _, k := range keys {
+ rows := byKey[k]
+ if len(rows) <= 1 {
+ continue
+ }
+ res.KeysWithDuplicates++
+ if sampleCap <= 0 || len(res.SamplesByVersion) < sampleCap {
+ versions := make([]VersionWithPath, len(rows))
+ for i := range rows {
+ versions[i] = VersionWithPath{Version: rows[i].version}
+ }
+ res.SamplesByVersion = append(res.SamplesByVersion, AnalyzeKeyMulti{
+ SeriesID: common.SeriesID(k.sid),
+ Timestamp: k.ts,
+ Versions: versions,
+ })
+ }
+ distinctDigest := rows[0].digest
+ conflict := false
+ for i := 1; i < len(rows); i++ {
+ if rows[i].digest != distinctDigest {
+ conflict = true
+ break
+ }
+ }
+ if !conflict {
+ continue
+ }
+ res.ValueConflictKeys++
+ if sampleCap <= 0 || len(res.ValueConflicts) < sampleCap {
+ vc := AnalyzeValueConflict{SeriesID: k.sid, Timestamp: k.ts}
+ for i := range rows {
+ vc.Versions = append(vc.Versions, rows[i].version)
+ vc.Digests = append(vc.Digests, rows[i].digest)
+ }
+ res.ValueConflicts = append(res.ValueConflicts, vc)
+ }
+ }
+ return res, nil
+}
diff --git a/banyand/measure/migration_indexmode_copy_test.go b/banyand/measure/migration_indexmode_copy_test.go
new file mode 100644
index 000000000..88ef4f1e4
--- /dev/null
+++ b/banyand/measure/migration_indexmode_copy_test.go
@@ -0,0 +1,1470 @@
+// 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"
+ "testing"
+ "time"
+
+ "github.com/blugelabs/bluge"
+ "github.com/stretchr/testify/require"
+ "google.golang.org/protobuf/types/known/timestamppb"
+
+ 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"
+ "github.com/apache/skywalking-banyandb/banyand/internal/migration"
+ "github.com/apache/skywalking-banyandb/banyand/internal/storage"
+ "github.com/apache/skywalking-banyandb/pkg/convert"
+ "github.com/apache/skywalking-banyandb/pkg/index"
+ "github.com/apache/skywalking-banyandb/pkg/index/inverted"
+ pbv1 "github.com/apache/skywalking-banyandb/pkg/pb/v1"
+)
+
+func TestClassifyIndexModeGroups(t *testing.T) {
+ // Case 1: index-mode measures + only the internal _top_n_result -> index-mode group.
+ isIndexMode, err := classifyGroup("sw_metadata", map[string]*measureSchemaInfo{
+ "service_traffic": {Name: "service_traffic", IndexMode: true},
+ "endpoint_traffic": {Name: "endpoint_traffic", IndexMode: true},
+ TopNSchemaName: {Name: TopNSchemaName, IndexMode: false},
+ })
+ require.NoError(t, err)
+ require.True(t, isIndexMode)
+
+ // Case 2: index-mode + a real normal measure -> error naming the conflict.
+ isIndexMode, err = classifyGroup("mixed", map[string]*measureSchemaInfo{
+ "service_traffic": {Name: "service_traffic", IndexMode: true},
+ "service_cpm_minute": {Name: "service_cpm_minute", IndexMode: false},
+ TopNSchemaName: {Name: TopNSchemaName, IndexMode: false},
+ })
+ require.Error(t, err)
+ require.False(t, isIndexMode)
+ require.Contains(t, err.Error(), "service_cpm_minute")
+ // The internal _top_n_result must never be listed among the conflicting
+ // normal measures (it is auto-created for every group).
+ require.Contains(t, err.Error(), "[service_cpm_minute]")
+
+ // Case 3: pure normal group (including _top_n_result) -> not index-mode.
+ isIndexMode, err = classifyGroup("sw_metricsMinute", map[string]*measureSchemaInfo{
+ "a": {Name: "a", IndexMode: false},
+ "b": {Name: "b", IndexMode: false},
+ TopNSchemaName: {Name: TopNSchemaName, IndexMode: false},
+ })
+ require.NoError(t, err)
+ require.False(t, isIndexMode)
+}
+
+func TestClassifyStoredField(t *testing.T) {
+ const knownRuleID = uint32(201559343)
+ ruleByID := map[uint32]indexRuleInfo{knownRuleID: {Analyzer: "", NoSort: true}}
+ // The group's tag-name set includes a 4-CHARACTER tag name ("name") to prove
+ // it is not misread as an IndexRuleID just because it is 4 bytes wide.
+ tagNames := map[string]struct{}{"instance_traffic_name": {}, "name": {}}
+
+ // A known tag name -> non-indexed TagName.
+ k, indexed, missing := classifyStoredField("instance_traffic_name", tagNames, ruleByID)
+ require.False(t, indexed)
+ require.Equal(t, "instance_traffic_name", k.TagName)
+ require.Zero(t, missing)
+
+ // A 4-character tag name in the schema -> TagName, NOT an IndexRuleID.
+ k, indexed, missing = classifyStoredField("name", tagNames, ruleByID)
+ require.False(t, indexed, "a 4-char schema tag must not be treated as a rule id")
+ require.Equal(t, "name", k.TagName)
+ require.Zero(t, k.IndexRuleID)
+ require.Zero(t, missing)
+
+ // A 4-byte name matching a KNOWN rule -> indexed, no missing rule.
+ known := string(convert.Uint32ToBytes(knownRuleID))
+ k, indexed, missing = classifyStoredField(known, tagNames, ruleByID)
+ require.True(t, indexed)
+ require.Equal(t, knownRuleID, k.IndexRuleID)
+ require.Zero(t, missing)
+
+ // A 4-byte name that DECODES to a rule id absent from the catalog: kept indexed
+ // (not demoted), and returned as missingRuleID so the caller warns.
+ const goneRuleID = uint32(999999)
+ gone := string(convert.Uint32ToBytes(goneRuleID))
+ k, indexed, missing = classifyStoredField(gone, tagNames, ruleByID)
+ require.True(t, indexed, "a field decoding to a rule id must stay indexed even if the rule is missing")
+ require.Equal(t, goneRuleID, k.IndexRuleID)
+ require.Equal(t, goneRuleID, missing, "missing rule id must be reported for the warning")
+}
+
+// indexModeDocSpec describes one production-equivalent index-mode document to
+// seed into a source sidx for the read+rebuild tests.
+type indexModeDocSpec struct {
+ storedTags map[string]string
+ subject string
+ indexedValue string
+ entityTags []string
+ entityValues []string
+ timestamp int64
+ version int64
+ indexedRuleID uint32
+ indexedNoSort bool
+}
+
+// buildIndexModeTestStore writes one document into a sidx at dir that mirrors
+// what the production write path produces for an index-mode measure: stored
+// regular tags + an indexed tag + index-only _im_name / _im_entity_tag_*.
+func buildIndexModeTestStore(t *testing.T, dir string, spec indexModeDocSpec) {
+ t.Helper()
+ series := &pbv1.Series{Subject: spec.subject}
+ for _, v := range spec.entityValues {
+ series.EntityValues = append(series.EntityValues, strTagValue(v))
+ }
+ require.NoError(t, series.Marshal())
+
+ var fields []index.Field
+ // Stored-only regular tag(s) (handleIndexMode, non-indexed branch).
+ for name, val := range spec.storedTags {
+ f := index.NewBytesField(index.FieldKey{TagName: name}, convert.StringToBytes(val))
+ f.Store = true
+ f.Index = false
+ fields = append(fields, f)
+ }
+ // One indexed stored tag (handleIndexMode, indexed branch).
+ if spec.indexedRuleID != 0 {
+ f := index.NewBytesField(index.FieldKey{IndexRuleID: spec.indexedRuleID}, convert.StringToBytes(spec.indexedValue))
+ f.Store = true
+ f.Index = true
+ f.NoSort = spec.indexedNoSort
+ fields = append(fields, f)
+ }
+ // Index-only entity-derived fields (appendEntityTagsToIndexFields).
+ subj := index.NewStringField(index.FieldKey{TagName: index.IndexModeName}, spec.subject)
+ subj.Index = true
+ subj.NoSort = true
+ fields = append(fields, subj)
+ for i, tagName := range spec.entityTags {
+ f := index.NewBytesField(index.FieldKey{TagName: index.IndexModeEntityTagPrefix + tagName},
+ convert.StringToBytes(spec.entityValues[i]))
+ f.Index = true
+ f.NoSort = true
+ fields = append(fields, f)
+ }
+
+ store, err := inverted.NewStore(inverted.StoreOpts{Path: dir, BatchWaitSec: 0})
+ require.NoError(t, err)
+ require.NoError(t, store.UpdateSeriesBatch(index.Batch{Documents: index.Documents{{
+ Fields: fields,
+ EntityValues: series.Buffer,
+ Timestamp: spec.timestamp,
+ DocID: uint64(series.ID),
+ Version: spec.version,
+ }}}))
+ require.NoError(t, store.Close())
+}
+
+// fieldByMarshaledName returns the rebuilt field whose marshaled key matches.
+func fieldByMarshaledName(d index.Document, name string) (index.Field, bool) {
+ for i := range d.Fields {
+ if d.Fields[i].Key.Marshal() == name {
+ return d.Fields[i], true
+ }
+ }
+ return index.Field{}, false
+}
+
+func TestReadAndRebuildIndexModeDoc(t *testing.T) {
+ dir := t.TempDir()
+ const ruleID = uint32(201559343)
+ buildIndexModeTestStore(t, dir, indexModeDocSpec{
+ subject: "svc",
+ entityTags: []string{"id"},
+ entityValues: []string{"abc"},
+ storedTags: map[string]string{"properties": "v1"},
+ indexedRuleID: ruleID,
+ indexedValue: "https://example.com/x",
+ indexedNoSort: false,
+ timestamp: 1000,
+ version: 7,
+ })
+
+ schemasBySubject := map[string]*measureSchemaInfo{
+ "svc": {
+ Name: "svc",
+ EntityTagNames: []string{"id"},
+ TagType: map[string]databasev1.TagType{"id": databasev1.TagType_TAG_TYPE_STRING},
+ IndexedEntityTags: map[string]struct{}{},
+ },
+ }
+ ruleByID := map[uint32]indexRuleInfo{ruleID: {Analyzer: index.AnalyzerURL, NoSort: false}}
+
+ docs, err := readIndexModeDocs(context.Background(), dir, ruleByID, schemasBySubject)
+ require.NoError(t, err)
+ require.Len(t, docs, 1)
+ d := docs[0]
+
+ require.Equal(t, int64(1000), d.Timestamp)
+ require.Equal(t, int64(7), d.Version)
+ require.NotEmpty(t, d.EntityValues)
+
+ // EntityValues must round-trip back to the original series.
+ var series pbv1.Series
+ require.NoError(t, series.Unmarshal(d.EntityValues))
+ require.Equal(t, "svc", series.Subject)
+
+ // Regular stored tag is recovered, not indexed.
+ pf, ok := fieldByMarshaledName(d, "properties")
+ require.True(t, ok, "stored tag 'properties' missing")
+ require.False(t, pf.Index)
+ require.True(t, pf.Store)
+ require.Equal(t, "v1", string(pf.GetBytes()))
+
+ // Indexed stored tag restores Analyzer/NoSort from the rule table.
+ idxf, ok := fieldByMarshaledName(d, string(convert.Uint32ToBytes(ruleID)))
+ require.True(t, ok, "indexed tag field missing")
+ require.True(t, idxf.Index)
+ require.True(t, idxf.Store)
+ require.Equal(t, index.AnalyzerURL, idxf.Key.Analyzer)
+ require.False(t, idxf.NoSort)
+
+ // Index-only fields regenerated from _id.
+ nameF, ok := fieldByMarshaledName(d, index.IndexModeName)
+ require.True(t, ok, "_im_name not regenerated")
+ require.True(t, nameF.Index)
+ require.Equal(t, "svc", string(nameF.GetBytes()))
+
+ entF, ok := fieldByMarshaledName(d, index.IndexModeEntityTagPrefix+"id")
+ require.True(t, ok, "_im_entity_tag_id not regenerated")
+ require.True(t, entF.Index)
+ require.Equal(t, "abc", string(entF.GetBytes()))
+}
+
+// TestReadAndRebuildIndexModeDoc_SkipsIndexedEntityTag asserts that an entity
+// tag already indexed by an index rule is NOT re-emitted as _im_entity_tag_*,
+// replicating the production skip condition (write_standalone.go:417).
+func TestReadAndRebuildIndexModeDoc_SkipsIndexedEntityTag(t *testing.T) {
+ dir := t.TempDir()
+ buildIndexModeTestStore(t, dir, indexModeDocSpec{
+ subject: "svc",
+ entityTags: []string{"id"},
+ entityValues: []string{"abc"},
+ storedTags: map[string]string{"properties": "v1"},
+ timestamp: 1000,
+ version: 1,
+ })
+
+ schemasBySubject := map[string]*measureSchemaInfo{
+ "svc": {
+ Name: "svc",
+ EntityTagNames: []string{"id"},
+ TagType: map[string]databasev1.TagType{"id": databasev1.TagType_TAG_TYPE_STRING},
+ // "id" is already indexed by an index rule -> must be skipped.
+ IndexedEntityTags: map[string]struct{}{"id": {}},
+ },
+ }
+
+ docs, err := readIndexModeDocs(context.Background(), dir, map[uint32]indexRuleInfo{}, schemasBySubject)
+ require.NoError(t, err)
+ require.Len(t, docs, 1)
+
+ // _im_name is still regenerated.
+ _, ok := fieldByMarshaledName(docs[0], index.IndexModeName)
+ require.True(t, ok, "_im_name should always be regenerated")
+ // _im_entity_tag_id must be absent because the tag is index-rule indexed.
+ _, ok = fieldByMarshaledName(docs[0], index.IndexModeEntityTagPrefix+"id")
+ require.False(t, ok, "_im_entity_tag_id must be skipped for index-rule-indexed entity tag")
+}
+
+func TestReadIndexModeDocs_EmptySidx(t *testing.T) {
+ docs, err := readIndexModeDocs(context.Background(), t.TempDir(), map[uint32]indexRuleInfo{}, nil)
+ require.NoError(t, err)
+ require.Empty(t, docs)
+}
+
+// ── Phase 2: copy routing tests ──────────────────────────────────────────────.
+
+// day20260621Nanos returns the local-time start-of-day nanos for 2026-06-21.
+// Segment names are formatted in time.Local, so test timestamps are anchored to
+// a local day boundary to stay aligned with formatDirectCopySegName.
+func day20260621Nanos(t *testing.T) int64 {
+ t.Helper()
+ return time.Date(2026, time.June, 21, 0, 0, 0, 0, time.Local).UnixNano()
+}
+
+// seedSourceSeg writes one index-mode doc into "<root>/<segName>/sidx".
+func seedSourceSeg(t *testing.T, root, segName string, spec indexModeDocSpec) {
+ t.Helper()
+ sidxDir := filepath.Join(root, segName, directCopySidxDirName)
+ require.NoError(t, os.MkdirAll(sidxDir, storage.DirPerm))
+ buildIndexModeTestStore(t, sidxDir, spec)
+}
+
+// seedSourceSegDocs writes several docs into one source seg's sidx in a single
+// committed batch, so a source segment can span more than one target seg (a
+// "split" source) for the merge/eligibility tests.
+func seedSourceSegDocs(t *testing.T, root, segName string, specs ...indexModeDocSpec) {
+ t.Helper()
+ sidxDir := filepath.Join(root, segName, directCopySidxDirName)
+ require.NoError(t, os.MkdirAll(sidxDir, storage.DirPerm))
+ store, err := inverted.NewStore(inverted.StoreOpts{Path: sidxDir, BatchWaitSec: 0})
+ require.NoError(t, err)
+ docs := make(index.Documents, 0, len(specs))
+ for _, spec := range specs {
+ series := &pbv1.Series{Subject: spec.subject}
+ for _, v := range spec.entityValues {
+ series.EntityValues = append(series.EntityValues, strTagValue(v))
+ }
+ require.NoError(t, series.Marshal())
+ var fields []index.Field
+ for name, val := range spec.storedTags {
+ f := index.NewBytesField(index.FieldKey{TagName: name}, convert.StringToBytes(val))
+ f.Store = true
+ fields = append(fields, f)
+ }
+ subj := index.NewStringField(index.FieldKey{TagName: index.IndexModeName}, spec.subject)
+ subj.Index = true
+ subj.NoSort = true
+ fields = append(fields, subj)
+ for i, tagName := range spec.entityTags {
+ f := index.NewBytesField(index.FieldKey{TagName: index.IndexModeEntityTagPrefix + tagName},
+ convert.StringToBytes(spec.entityValues[i]))
+ f.Index = true
+ f.NoSort = true
+ fields = append(fields, f)
+ }
+ docs = append(docs, index.Document{
+ Fields: fields, EntityValues: series.Buffer,
+ Timestamp: spec.timestamp, DocID: uint64(series.ID), Version: spec.version,
+ })
+ }
+ require.NoError(t, store.UpdateSeriesBatch(index.Batch{Documents: docs}))
+ require.NoError(t, store.Close())
+}
+
+func svcSchemas() map[string]*measureSchemaInfo {
+ return map[string]*measureSchemaInfo{
+ "svc": {
+ Name: "svc",
+ EntityTagNames: []string{"id"},
+ TagType: map[string]databasev1.TagType{"id": databasev1.TagType_TAG_TYPE_STRING},
+ IndexedEntityTags: map[string]struct{}{},
+ },
+ }
+}
+
+// countSidxDocs opens a sidx read-only and returns the number of docs.
+func countSidxDocs(t *testing.T, sidxDir string) int {
+ t.Helper()
+ docs, err := readIndexModeDocs(context.Background(), sidxDir, map[uint32]indexRuleInfo{}, svcSchemas())
+ require.NoError(t, err)
+ return len(docs)
+}
+
+func TestTargetIdxStore_GetSamePathSharesWriter(t *testing.T) {
+ pool := newTargetIdxStore()
+ path := filepath.Join(t.TempDir(), "sidx")
+
+ s1, err := pool.get(path)
+ require.NoError(t, err)
+ s2, err := pool.get(path)
+ require.NoError(t, err)
+ require.Same(t, s1, s2, "same path must share one writer")
+
+ require.NoError(t, pool.closeAll())
+ require.Empty(t, pool.stores, "closeAll must clear the pool")
+}
+
+// readDocsByDir reads every dir's index-mode docs into the docsByDir map the
+// slow path consumes, mirroring what copyIndexModeGroup's Pass 1 builds.
+func readDocsByDir(t *testing.T, dirs []string) map[string][]index.Document {
+ t.Helper()
+ out := make(map[string][]index.Document, len(dirs))
+ for _, dir := range dirs {
+ docs, err := readIndexModeDocs(context.Background(), dir, map[uint32]indexRuleInfo{}, svcSchemas())
+ require.NoError(t, err)
+ out[dir] = docs
+ }
+ return out
+}
+
+func TestSlowPath_MergeKeepsMaxVersion(t *testing.T) {
+ root := t.TempDir()
+ dst := filepath.Join(t.TempDir(), "tgt")
+ ir := storage.IntervalRule{Unit: storage.DAY, Num: 1}
+ ts := day20260621Nanos(t) + int64(time.Hour)
+
+ // Two source segs hold the SAME series (same subject+entity -> same DocID)
+ // at different versions; both route to seg-20260621.
+ seedSourceSeg(t, root, "seg-20260621a", indexModeDocSpec{
+ subject: "svc", entityTags: []string{"id"}, entityValues: []string{"abc"},
+ storedTags: map[string]string{"properties": "old"}, timestamp: ts, version: 3,
+ })
+ seedSourceSeg(t, root, "seg-20260621b", indexModeDocSpec{
+ subject: "svc", entityTags: []string{"id"}, entityValues: []string{"abc"},
+ storedTags: map[string]string{"properties": "new"}, timestamp: ts, version: 9,
+ })
+
+ stores := newTargetIdxStore()
+ srcDirs := []string{
+ filepath.Join(root, "seg-20260621a", directCopySidxDirName),
+ filepath.Join(root, "seg-20260621b", directCopySidxDirName),
+ }
+ rows, err := copyIndexModeSlowDocs(context.Background(), srcDirs, readDocsByDir(t, srcDirs), dst, ir, stores)
+ require.NoError(t, err)
+ require.Equal(t, int64(1), rows, "same series collapses to one row")
+ require.NoError(t, stores.closeAll())
+
+ targetSidx := filepath.Join(dst, "seg-20260621", directCopySidxDirName)
+ docs, err := readIndexModeDocs(context.Background(), targetSidx, map[uint32]indexRuleInfo{}, svcSchemas())
+ require.NoError(t, err)
+ require.Len(t, docs, 1)
+ require.Equal(t, int64(9), docs[0].Version, "highest version must win")
+ pf, ok := fieldByMarshaledName(docs[0], "properties")
+ require.True(t, ok)
+ require.Equal(t, "new", string(pf.GetBytes()))
+ require.FileExists(t, filepath.Join(dst, "seg-20260621", storage.SegmentMetadataFilename))
+}
+
+func TestSlowPath_SplitRoutesByTimestamp(t *testing.T) {
+ root := t.TempDir()
+ dst := filepath.Join(t.TempDir(), "tgt")
+ ir := storage.IntervalRule{Unit: storage.HOUR, Num: 1}
+ base := day20260621Nanos(t)
+
+ // One source seg holds two docs in different hours -> two target segs.
+ src := filepath.Join(root, "seg-20260621", directCopySidxDirName)
+ require.NoError(t, os.MkdirAll(src, storage.DirPerm))
+ buildIndexModeTestStore(t, src, indexModeDocSpec{
+ subject: "svc", entityTags: []string{"id"}, entityValues: []string{"h01"},
+ storedTags: map[string]string{"properties": "a"}, timestamp: base + int64(time.Hour), version: 1,
+ })
+ buildIndexModeTestStore(t, src, indexModeDocSpec{
+ subject: "svc", entityTags: []string{"id"}, entityValues: []string{"h05"},
+ storedTags: map[string]string{"properties": "b"}, timestamp: base + int64(5*time.Hour), version: 1,
+ })
+
+ stores := newTargetIdxStore()
+ srcDirs := []string{src}
+ rows, err := copyIndexModeSlowDocs(context.Background(), srcDirs, readDocsByDir(t, srcDirs), dst, ir, stores)
+ require.NoError(t, err)
+ require.Equal(t, int64(2), rows)
+ require.NoError(t, stores.closeAll())
+
+ require.Equal(t, 1, countSidxDocs(t, filepath.Join(dst, "seg-2026062101", directCopySidxDirName)))
+ require.Equal(t, 1, countSidxDocs(t, filepath.Join(dst, "seg-2026062105", directCopySidxDirName)))
+}
+
+func TestCopyIndexModeGroup_FastPathByteCopy(t *testing.T) {
+ root := t.TempDir()
+ dst := filepath.Join(t.TempDir(), "tgt")
+ ir := storage.IntervalRule{Unit: storage.DAY, Num: 1}
+ ts := day20260621Nanos(t) + int64(2*time.Hour)
+
+ src := filepath.Join(root, "seg-20260621", directCopySidxDirName)
+ require.NoError(t, os.MkdirAll(src, storage.DirPerm))
+ buildIndexModeTestStore(t, src, indexModeDocSpec{
+ subject: "svc", entityTags: []string{"id"}, entityValues: []string{"abc"},
+ storedTags: map[string]string{"properties": "v1"}, timestamp: ts, version: 1,
+ })
+
+ res, err := copyIndexModeGroup(context.Background(), migration.EntryGroupInput{
+ EntryTag: "[entry 1/1]",
+ Group: "sw_metadata",
+ TargetGroupRoot: dst,
+ SrcRoots: []string{root},
+ Interval: ir,
+ }, map[uint32]indexRuleInfo{}, svcSchemas())
+ require.NoError(t, err)
+ require.Equal(t, int64(1), res.Rows)
+ require.Equal(t, 1, res.Segments)
+
+ targetSidx := filepath.Join(dst, "seg-20260621", directCopySidxDirName)
+ require.Equal(t, 1, countSidxDocs(t, targetSidx),
+ "target sidx doc count must match source")
+ require.FileExists(t, filepath.Join(dst, "seg-20260621", storage.SegmentMetadataFilename),
+ "byte-copy must still write segment metadata")
+}
+
+// TestCopyIndexModeGroup_MergeAcrossSourcesKeepsMaxVersion drives the top-level
+// copyIndexModeGroup (the production path) with two source sidx dirs holding the
+// SAME series at different versions, both aligning to the same target seg. Since
+// neither source exclusively owns the target seg, both must go through a single
+// slow-path merge and the highest version must win. This reproduces the bug
+// where copyIndexModeGroup looped one source per copyIndexModeSlow call, leaving
+// cross-source dedup to bluge's last-writer-wins.
+func TestCopyIndexModeGroup_MergeAcrossSourcesKeepsMaxVersion(t *testing.T) {
+ rootA := t.TempDir()
+ rootB := t.TempDir()
+ dst := filepath.Join(t.TempDir(), "tgt")
+ ir := storage.IntervalRule{Unit: storage.DAY, Num: 1}
+ ts := day20260621Nanos(t) + int64(time.Hour)
+
+ // Two distinct source roots (e.g. two node replicas), same series, both
+ // aligning to seg-20260621 at different versions.
+ seedSourceSeg(t, rootA, "seg-20260621", indexModeDocSpec{
+ subject: "svc", entityTags: []string{"id"}, entityValues: []string{"abc"},
+ storedTags: map[string]string{"properties": "old"}, timestamp: ts, version: 3,
+ })
+ seedSourceSeg(t, rootB, "seg-20260621", indexModeDocSpec{
+ subject: "svc", entityTags: []string{"id"}, entityValues: []string{"abc"},
+ storedTags: map[string]string{"properties": "new"}, timestamp: ts, version: 9,
+ })
+
+ res, err := copyIndexModeGroup(context.Background(), migration.EntryGroupInput{
+ EntryTag: "[entry 1/1]",
+ Group: "sw_metadata",
+ TargetGroupRoot: dst,
+ SrcRoots: []string{rootA, rootB},
+ Interval: ir,
+ }, map[uint32]indexRuleInfo{}, svcSchemas())
+ require.NoError(t, err)
+ require.Equal(t, 1, res.Segments)
+ require.Equal(t, int64(1), res.Rows, "same series across sources collapses to one row")
+
+ targetSidx := filepath.Join(dst, "seg-20260621", directCopySidxDirName)
+ docs, err := readIndexModeDocs(context.Background(), targetSidx, map[uint32]indexRuleInfo{}, svcSchemas())
+ require.NoError(t, err)
+ require.Len(t, docs, 1)
+ require.Equal(t, int64(9), docs[0].Version, "highest version must win across sources")
+ pf, ok := fieldByMarshaledName(docs[0], "properties")
+ require.True(t, ok)
+ require.Equal(t, "new", string(pf.GetBytes()), "the max-version doc's value must survive")
+}
+
+// TestCopyIndexModeGroup_ByteCopyTargetAlsoReceivesSplitSource_KeepsMaxVersion
+// reproduces a real-data bug: a single-target source (byte-copy candidate) and a
+// SPLIT source both route the same series into the same target seg. The split
+// source was not counted as an owner of that target seg, so the single-target
+// source byte-copied it and the split source's slow-path rebuild then upserted
+// (last-writer-wins) — dropping the higher version. The target must keep the
+// max version across BOTH sources.
+func TestCopyIndexModeGroup_ByteCopyTargetAlsoReceivesSplitSource_KeepsMaxVersion(t *testing.T) {
+ root := t.TempDir()
+ dst := filepath.Join(t.TempDir(), "tgt")
+ ir := storage.IntervalRule{Unit: storage.HOUR, Num: 1}
+ base := day20260621Nanos(t)
+ h1 := base + int64(time.Hour)
+ h2 := base + int64(2*time.Hour)
+
+ // Single-target source: all docs in hour-01. Higher version of "shared".
+ seedSourceSeg(t, root, "seg-2026062101", indexModeDocSpec{
+ subject: "svc", entityTags: []string{"id"}, entityValues: []string{"shared"},
+ storedTags: map[string]string{"properties": "new"}, timestamp: h1, version: 100,
+ })
+ // Split source: spans hour-01 (lower version of "shared") and hour-02 (filler),
+ // so it routes a doc into hour-01 too — making hour-01 NOT exclusively owned.
+ seedSourceSegDocs(t, root, "seg-2026062102",
+ indexModeDocSpec{
+ subject: "svc", entityTags: []string{"id"}, entityValues: []string{"shared"},
+ storedTags: map[string]string{"properties": "old"}, timestamp: h1, version: 50,
+ },
+ indexModeDocSpec{
+ subject: "svc", entityTags: []string{"id"}, entityValues: []string{"filler"},
+ storedTags: map[string]string{"properties": "f"}, timestamp: h2, version: 1,
+ })
+
+ res, err := copyIndexModeGroup(context.Background(), migration.EntryGroupInput{
+ EntryTag: "[entry 1/1]",
+ Group: "sw_metadata",
+ TargetGroupRoot: dst,
+ SrcRoots: []string{root},
+ Interval: ir,
+ }, map[uint32]indexRuleInfo{}, svcSchemas())
+ require.NoError(t, err)
+ require.Zero(t, res.Bytes, "no source exclusively owns a target seg, so nothing byte-copies")
+
+ h1Sidx := filepath.Join(dst, "seg-2026062101", directCopySidxDirName)
+ docs, err := readIndexModeDocs(context.Background(), h1Sidx, map[uint32]indexRuleInfo{}, svcSchemas())
+ require.NoError(t, err)
+ require.Len(t, docs, 1, "hour-01 holds exactly the shared series")
+ require.Equal(t, int64(100), docs[0].Version, "max version must survive across byte-copy + split source")
+ pf, ok := fieldByMarshaledName(docs[0], "properties")
+ require.True(t, ok)
+ require.Equal(t, "new", string(pf.GetBytes()), "the max-version doc's value must survive")
+ // The split source's filler doc must still land in hour-02 (not dropped).
+ require.Equal(t, 1, countSidxDocs(t, filepath.Join(dst, "seg-2026062102", directCopySidxDirName)),
+ "the split source's hour-02 doc must be preserved")
+}
+
+// TestCopyIndexModeGroup_ExclusiveTargetsByteCopy drives copyIndexModeGroup with
+// two sources that each exclusively own a distinct target seg, so both must take
+// the byte-copy fast path (no slow-path merge). res.Bytes>0 confirms byte-copy ran.
+func TestCopyIndexModeGroup_ExclusiveTargetsByteCopy(t *testing.T) {
+ root := t.TempDir()
+ dst := filepath.Join(t.TempDir(), "tgt")
+ ir := storage.IntervalRule{Unit: storage.HOUR, Num: 1}
+ base := day20260621Nanos(t)
+
+ // Two source segs, each landing wholly in a distinct hour-aligned target seg.
+ seedSourceSeg(t, root, "seg-2026062101", indexModeDocSpec{
+ subject: "svc", entityTags: []string{"id"}, entityValues: []string{"h01"},
+ storedTags: map[string]string{"properties": "a"}, timestamp: base + int64(time.Hour), version: 1,
+ })
+ seedSourceSeg(t, root, "seg-2026062105", indexModeDocSpec{
+ subject: "svc", entityTags: []string{"id"}, entityValues: []string{"h05"},
+ storedTags: map[string]string{"properties": "b"}, timestamp: base + int64(5*time.Hour), version: 1,
+ })
+
+ res, err := copyIndexModeGroup(context.Background(), migration.EntryGroupInput{
+ EntryTag: "[entry 1/1]",
+ Group: "sw_metadata",
+ TargetGroupRoot: dst,
+ SrcRoots: []string{root},
+ Interval: ir,
+ }, map[uint32]indexRuleInfo{}, svcSchemas())
+ require.NoError(t, err)
+ require.Equal(t, int64(2), res.Rows)
+ require.Equal(t, 2, res.Segments)
+ require.Positive(t, res.Bytes, "exclusive single-target sources must byte-copy")
+
+ require.Equal(t, 1, countSidxDocs(t, filepath.Join(dst, "seg-2026062101", directCopySidxDirName)))
+ require.Equal(t, 1, countSidxDocs(t, filepath.Join(dst, "seg-2026062105", directCopySidxDirName)))
+}
+
+// TestBuildSegAligns_ByteCopyEligibility locks the alignment classification:
+// a target is "aligned" (a byte-copy candidate) ONLY when its sole feeding source
+// segment feeds no other target and no docs collapsed. A split source disqualifies
+// every target it touches — even one it solely feeds — and a merge target (>1
+// source) is never aligned.
+func TestBuildSegAligns_ByteCopyEligibility(t *testing.T) {
+ src := &segDigestResult{
+ docsPerSeg: map[string]uint64{"seg-A": 10, "seg-B": 5, "seg-C": 5, "seg-D": 8},
+ srcSegsPerTarget: map[string]map[string]struct{}{
+ "seg-A": {"srcA": {}}, // exclusive 1:1
+ "seg-B": {"srcS": {}}, // srcS also feeds seg-C => split
+ "seg-C": {"srcS": {}}, // split source
+ "seg-D": {"srcX": {}, "srcY": {}}, // merge: two sources
+ },
+ }
+ tgt := &segDigestResult{docsPerSeg: map[string]uint64{"seg-A": 10, "seg-B": 5, "seg-C": 5, "seg-D": 8}}
+
+ byName := map[string]IndexModeSegAlign{}
+ for _, a := range buildSegAligns(src, tgt) {
+ byName[a.Segment] = a
+ }
+ require.True(t, byName["seg-A"].Aligned, "exclusive 1:1 source is byte-copy aligned")
+ require.False(t, byName["seg-B"].Aligned, "a target fed by a split source is not aligned")
+ require.False(t, byName["seg-C"].Aligned, "the split source's other target is not aligned")
+ require.False(t, byName["seg-D"].Aligned, "a merge target (>1 source) is not aligned")
+
+ // A collapse (src docs > tgt docs) also disqualifies an otherwise-1:1 target.
+ src2 := &segDigestResult{
+ docsPerSeg: map[string]uint64{"seg-A": 10},
+ srcSegsPerTarget: map[string]map[string]struct{}{"seg-A": {"srcA": {}}},
+ }
+ tgt2 := &segDigestResult{docsPerSeg: map[string]uint64{"seg-A": 9}}
+ require.False(t, buildSegAligns(src2, tgt2)[0].Aligned, "a collapsing target is not 1:1 aligned")
+}
+
+// TestCopyIndexModeGroup_TimestampZeroSurfacesError verifies that a Timestamp==0
+// doc (no decodable _timestamp field) — which never occurs for valid index-mode
+// data, since the write path always stamps a checked non-zero timestamp — fails
+// the copy with an error naming the offending series, instead of being silently
+// routed to a guessed segment boundary.
+func TestCopyIndexModeGroup_TimestampZeroSurfacesError(t *testing.T) {
+ root := t.TempDir()
+ dst := filepath.Join(t.TempDir(), "tgt")
+ ir := storage.IntervalRule{Unit: storage.HOUR, Num: 1}
+
+ seedSourceSeg(t, root, "seg-2026062103", indexModeDocSpec{
+ subject: "svc", entityTags: []string{"id"}, entityValues: []string{"timeless"},
+ storedTags: map[string]string{"properties": "v"}, timestamp: 0, version: 1,
+ })
+
+ _, err := copyIndexModeGroup(context.Background(), migration.EntryGroupInput{
+ EntryTag: "[entry 1/1]",
+ Group: "sw_metadata",
+ TargetGroupRoot: dst,
+ SrcRoots: []string{root},
+ Interval: ir,
+ }, map[uint32]indexRuleInfo{}, svcSchemas())
+ require.Error(t, err, "a ts==0 doc must surface an error, not be silently routed")
+ require.Contains(t, err.Error(), "ts==0")
+ require.NoDirExists(t, dst, "no target segment may be written when the source has a ts==0 doc")
+}
+
+// ── Phase 3 / 4: value digest, verify, analyze tests ─────────────────────────.
+
+// readOneDoc reads exactly one rebuilt doc from a sidx dir, failing otherwise.
+func readOneDoc(t *testing.T, sidxDir string, schemas map[string]*measureSchemaInfo) index.Document {
+ t.Helper()
+ docs, err := readIndexModeDocs(context.Background(), sidxDir, map[uint32]indexRuleInfo{}, schemas)
+ require.NoError(t, err)
+ require.Len(t, docs, 1)
+ return docs[0]
+}
+
+func TestDocValueDigest_StableAndOrderIndependent(t *testing.T) {
+ dirA := filepath.Join(t.TempDir(), "sidx")
+ dirB := filepath.Join(t.TempDir(), "sidx")
+ require.NoError(t, os.MkdirAll(dirA, storage.DirPerm))
+ require.NoError(t, os.MkdirAll(dirB, storage.DirPerm))
+ spec := indexModeDocSpec{
+ subject: "svc", entityTags: []string{"id"}, entityValues: []string{"abc"},
+ storedTags: map[string]string{"properties": "v1", "kind": "k1"}, timestamp: 1000, version: 5,
+ }
+ buildIndexModeTestStore(t, dirA, spec)
+ buildIndexModeTestStore(t, dirB, spec)
+
+ da := readOneDoc(t, dirA, svcSchemas())
+ db := readOneDoc(t, dirB, svcSchemas())
+ require.Equal(t, docValueDigest(da), docValueDigest(db),
+ "identical content must hash equal regardless of field assembly order")
+}
+
+func TestDocValueDigest_ChangedTagValueDiffers(t *testing.T) {
+ dirA := filepath.Join(t.TempDir(), "sidx")
+ dirB := filepath.Join(t.TempDir(), "sidx")
+ require.NoError(t, os.MkdirAll(dirA, storage.DirPerm))
+ require.NoError(t, os.MkdirAll(dirB, storage.DirPerm))
+ buildIndexModeTestStore(t, dirA, indexModeDocSpec{
+ subject: "svc", entityTags: []string{"id"}, entityValues: []string{"abc"},
+ storedTags: map[string]string{"properties": "old"}, timestamp: 1000, version: 5,
+ })
+ buildIndexModeTestStore(t, dirB, indexModeDocSpec{
+ subject: "svc", entityTags: []string{"id"}, entityValues: []string{"abc"},
+ storedTags: map[string]string{"properties": "NEW"}, timestamp: 1000, version: 5,
+ })
+ require.NotEqual(t, docValueDigest(readOneDoc(t, dirA, svcSchemas())),
+ docValueDigest(readOneDoc(t, dirB, svcSchemas())),
+ "a changed tag value must change the digest")
+}
+
+func TestDocValueDigest_MissingEntityFieldDiffers(t *testing.T) {
+ dir := filepath.Join(t.TempDir(), "sidx")
+ require.NoError(t, os.MkdirAll(dir, storage.DirPerm))
+ buildIndexModeTestStore(t, dir, indexModeDocSpec{
+ subject: "svc", entityTags: []string{"id"}, entityValues: []string{"abc"},
+ storedTags: map[string]string{"properties": "v1"}, timestamp: 1000, version: 5,
+ })
+ full := readOneDoc(t, dir, svcSchemas())
+
+ // Drop the regenerated _im_entity_tag_id field to simulate a rebuild that
+ // lost an index-only entity value: the digest MUST differ from the intact
+ // doc (id present, value missing is still caught).
+ stripped := index.Document{
+ EntityValues: full.EntityValues,
+ Timestamp: full.Timestamp,
+ DocID: full.DocID,
+ Version: full.Version,
+ }
+ for i := range full.Fields {
+ if full.Fields[i].Key.Marshal() == index.IndexModeEntityTagPrefix+"id" {
+ continue
+ }
+ stripped.Fields = append(stripped.Fields, full.Fields[i])
+ }
+ require.NotEqual(t, docValueDigest(full), docValueDigest(stripped),
+ "dropping a regenerated entity field must change the digest")
+}
+
+// runCopyVerify copies srcRoots into a fresh target group root and returns the
+// resulting verify report. Used by the aligned-equal and merge tests so the
+// target is produced by the real copy path.
+func runCopyVerify(t *testing.T, srcRoots []string, ir storage.IntervalRule) EntryGroupReport {
+ t.Helper()
+ dst := filepath.Join(t.TempDir(), "tgt")
+ in := migration.EntryGroupInput{
+ EntryTag: "[entry 1/1]",
+ Group: "sw_metadata",
+ TargetGroupRoot: dst,
+ SrcRoots: srcRoots,
+ Interval: ir,
+ }
+ _, err := copyIndexModeGroup(context.Background(), in, map[uint32]indexRuleInfo{}, svcSchemas())
+ require.NoError(t, err)
+ report, err := verifyIndexModeGroup(context.Background(), in, map[uint32]indexRuleInfo{}, svcSchemas())
+ require.NoError(t, err)
+ return report
+}
+
+func TestVerifyIndexModeGroup_AlignedEqual(t *testing.T) {
+ root := t.TempDir()
+ ir := storage.IntervalRule{Unit: storage.DAY, Num: 1}
+ ts := day20260621Nanos(t) + int64(2*time.Hour)
+ seedSourceSeg(t, root, "seg-20260621", indexModeDocSpec{
+ subject: "svc", entityTags: []string{"id"}, entityValues: []string{"abc"},
+ storedTags: map[string]string{"properties": "v1"}, timestamp: ts, version: 1,
+ })
+ seedSourceSeg(t, root, "seg-20260621", indexModeDocSpec{ // overwrites? no: same seg dir, second doc series id2
+ subject: "svc", entityTags: []string{"id"}, entityValues: []string{"def"},
+ storedTags: map[string]string{"properties": "v2"}, timestamp: ts, version: 1,
+ })
+
+ report := runCopyVerify(t, []string{root}, ir)
+ require.True(t, report.IndexMode)
+ require.Equal(t, report.SrcDistinctIDs, report.TgtDistinctIDs, "aligned copy keeps distinct doc-id equal")
+ require.Equal(t, report.ExpectDistinctIDs, report.TgtDistinctIDs)
+ require.Empty(t, report.ValueMismatches, "aligned copy must reconcile all value digests")
+}
+
+func TestVerifyIndexModeGroup_MergeDistinctIDUnion(t *testing.T) {
+ rootA := t.TempDir()
+ rootB := t.TempDir()
+ ir := storage.IntervalRule{Unit: storage.DAY, Num: 1}
+ ts := day20260621Nanos(t) + int64(time.Hour)
+
+ // Same series in two source roots at different versions, both -> seg-20260621.
+ // Target upsert collapses them to ONE doc; expected distinct doc-id == union(1).
+ seedSourceSeg(t, rootA, "seg-20260621", indexModeDocSpec{
+ subject: "svc", entityTags: []string{"id"}, entityValues: []string{"abc"},
+ storedTags: map[string]string{"properties": "old"}, timestamp: ts, version: 3,
+ })
+ seedSourceSeg(t, rootB, "seg-20260621", indexModeDocSpec{
+ subject: "svc", entityTags: []string{"id"}, entityValues: []string{"abc"},
+ storedTags: map[string]string{"properties": "new"}, timestamp: ts, version: 9,
+ })
+
+ report := runCopyVerify(t, []string{rootA, rootB}, ir)
+ require.Equal(t, uint64(2), report.SrcDocs, "two source docs before merge")
+ require.Equal(t, uint64(1), report.ExpectDistinctIDs, "union of source doc-ids is 1")
+ require.Equal(t, uint64(1), report.TgtDistinctIDs, "merge collapses to one upserted doc")
+ require.Empty(t, report.ValueMismatches,
+ "source reduced to max version must match the surviving target doc")
+}
+
+// TestVerifyIndexModeGroup_DetectsMissingAndExtra exercises the two reconcile
+// branches the (segment, series) keying relies on: a series in source but not
+// target is "missing"; a series in target but not source is "extra".
+func TestVerifyIndexModeGroup_DetectsMissingAndExtra(t *testing.T) {
+ srcRoot := t.TempDir()
+ tgtGroup := filepath.Join(t.TempDir(), "tgt")
+ ir := storage.IntervalRule{Unit: storage.DAY, Num: 1}
+ ts := day20260621Nanos(t) + int64(time.Hour)
+
+ // Source seg-20260621 holds series "aaa" (routes to target seg-20260621).
+ seedSourceSeg(t, srcRoot, "seg-20260621", indexModeDocSpec{
+ subject: "svc", entityTags: []string{"id"}, entityValues: []string{"aaa"},
+ storedTags: map[string]string{"properties": "v"}, timestamp: ts, version: 1,
+ })
+ // Target seg-20260621 holds a DIFFERENT series "bbb": "aaa" is missing from
+ // the target and "bbb" is extra (not present in source).
+ seedSourceSeg(t, tgtGroup, "seg-20260621", indexModeDocSpec{
+ subject: "svc", entityTags: []string{"id"}, entityValues: []string{"bbb"},
+ storedTags: map[string]string{"properties": "v"}, timestamp: ts, version: 1,
+ })
+
+ report, err := verifyIndexModeGroup(context.Background(), migration.EntryGroupInput{
+ Group: "sw_metadata", SrcRoots: []string{srcRoot}, TargetGroupRoot: tgtGroup, Interval: ir,
+ }, map[uint32]indexRuleInfo{}, svcSchemas())
+ require.NoError(t, err)
+ kinds := map[string]int{}
+ for _, m := range report.ValueMismatches {
+ kinds[m.Kind]++
+ require.Equal(t, "seg-20260621", m.Segment)
+ }
+ require.Equal(t, 1, kinds["missing"], "source series aaa is missing from target")
+ require.Equal(t, 1, kinds["extra"], "target series bbb is extra")
+}
+
+func TestVerifyIndexModeGroup_DetectsTamper(t *testing.T) {
+ root := t.TempDir()
+ dst := filepath.Join(t.TempDir(), "tgt")
+ ir := storage.IntervalRule{Unit: storage.DAY, Num: 1}
+ ts := day20260621Nanos(t) + int64(2*time.Hour)
+
+ seedSourceSeg(t, root, "seg-20260621", indexModeDocSpec{
+ subject: "svc", entityTags: []string{"id"}, entityValues: []string{"abc"},
+ storedTags: map[string]string{"properties": "correct"}, timestamp: ts, version: 1,
+ })
+ // Hand-craft a TAMPERED target: same series/timestamp/version but a changed
+ // stored tag value. verify must flag it "tampered".
+ seedSourceSeg(t, dst, "seg-20260621", indexModeDocSpec{
+ subject: "svc", entityTags: []string{"id"}, entityValues: []string{"abc"},
+ storedTags: map[string]string{"properties": "TAMPERED"}, timestamp: ts, version: 1,
+ })
+
+ report, err := verifyIndexModeGroup(context.Background(), migration.EntryGroupInput{
+ Group: "sw_metadata", TargetGroupRoot: dst, SrcRoots: []string{root}, Interval: ir,
+ }, map[uint32]indexRuleInfo{}, svcSchemas())
+ require.NoError(t, err)
+ require.Len(t, report.ValueMismatches, 1)
+ require.Equal(t, "tampered", report.ValueMismatches[0].Kind)
+}
+
+func TestVerifyIndexModeGroup_DetectsMissingEntityField(t *testing.T) {
+ root := t.TempDir()
+ dst := filepath.Join(t.TempDir(), "tgt")
+ ts := day20260621Nanos(t) + int64(2*time.Hour)
+
+ // Source: schema declares entity tag "id", so the source doc regenerates
+ // _im_entity_tag_id from its entity value.
+ seedSourceSeg(t, root, "seg-20260621", indexModeDocSpec{
+ subject: "svc", entityTags: []string{"id"}, entityValues: []string{"abc"},
+ storedTags: map[string]string{"properties": "v1"}, timestamp: ts, version: 1,
+ })
+ // Target: SAME doc on disk, but verified with a schema that has NO entity
+ // tags, so the rebuild can't regenerate _im_entity_tag_id — the entity value
+ // is effectively missing on the target side. The digest diverges and verify
+ // flags "tampered" (id present, value missing).
+ seedSourceSeg(t, dst, "seg-20260621", indexModeDocSpec{
+ subject: "svc", entityTags: []string{"id"}, entityValues: []string{"abc"},
+ storedTags: map[string]string{"properties": "v1"}, timestamp: ts, version: 1,
+ })
+ srcSchemas := svcSchemas()
+ tgtSchemas := map[string]*measureSchemaInfo{
+ "svc": {
+ Name: "svc",
+ EntityTagNames: nil, // no entity tags -> _im_entity_tag_id not regenerated
+ TagType: map[string]databasev1.TagType{},
+ IndexedEntityTags: map[string]struct{}{},
+ },
+ }
+ // Read both sides with their own schemas to simulate the lost entity field,
+ // then diff the digests directly (verifyIndexModeGroup uses one schema map,
+ // so we exercise the digest layer it relies on).
+ srcDoc := readOneDoc(t, filepath.Join(root, "seg-20260621", directCopySidxDirName), srcSchemas)
+ tgtDoc := readOneDoc(t, filepath.Join(dst, "seg-20260621", directCopySidxDirName), tgtSchemas)
+ require.NotEqual(t, docValueDigest(srcDoc), docValueDigest(tgtDoc),
+ "a target missing the regenerated entity field must not reconcile")
+ srcMap := map[imSegKey]segSurvivor{{seg: "seg-20260621", sid: srcDoc.DocID}: {version: srcDoc.Version, digest: docValueDigest(srcDoc)}}
+ tgtMap := map[imSegKey]segSurvivor{{seg: "seg-20260621", sid: tgtDoc.DocID}: {version: tgtDoc.Version, digest: docValueDigest(tgtDoc)}}
+ mismatches := diffValueDigests(srcMap, tgtMap)
+ require.Len(t, mismatches, 1)
+ require.Equal(t, "tampered", mismatches[0].Kind)
+}
+
+func TestAnalyzeIndexModeGroup_ValueConflict(t *testing.T) {
+ root := t.TempDir()
+ ts := day20260621Nanos(t) + int64(2*time.Hour)
+
+ // Two source segs hold the SAME series at the SAME timestamp but DIFFERENT
+ // stored-tag values (and different versions). That is a value conflict: same
+ // (sid, ts), diverging value digests.
+ seedSourceSeg(t, root, "seg-20260621a", indexModeDocSpec{
+ subject: "svc", entityTags: []string{"id"}, entityValues: []string{"abc"},
+ storedTags: map[string]string{"properties": "valA"}, timestamp: ts, version: 1,
+ })
+ seedSourceSeg(t, root, "seg-20260621b", indexModeDocSpec{
+ subject: "svc", entityTags: []string{"id"}, entityValues: []string{"abc"},
+ storedTags: map[string]string{"properties": "valB"}, timestamp: ts, version: 2,
+ })
+
+ srcSidxDirs, err := collectSourceSidxDirs([]string{root})
+ require.NoError(t, err)
+ res, err := analyzeIndexModeGroup(context.Background(), srcSidxDirs, map[uint32]indexRuleInfo{}, svcSchemas(), 10)
+ require.NoError(t, err)
+ require.Equal(t, uint64(2), res.TotalRows)
+ require.Equal(t, uint64(1), res.UniqueKeys, "same (sid, ts) collapses to one key")
+ require.Equal(t, uint64(1), res.KeysWithDuplicates)
+ require.Equal(t, uint64(1), res.ValueConflictKeys, "differing values on the same key is a conflict")
+ require.Len(t, res.ValueConflicts, 1)
+ require.NotEqual(t, res.ValueConflicts[0].Digests[0], res.ValueConflicts[0].Digests[1])
+}
+
+func TestAnalyzeIndexModeGroup_NoConflictWhenIdentical(t *testing.T) {
+ root := t.TempDir()
+ ts := day20260621Nanos(t) + int64(2*time.Hour)
+ // Same series, same ts, IDENTICAL value but different versions -> a version
+ // duplicate but NOT a value conflict.
+ seedSourceSeg(t, root, "seg-20260621a", indexModeDocSpec{
+ subject: "svc", entityTags: []string{"id"}, entityValues: []string{"abc"},
+ storedTags: map[string]string{"properties": "same"}, timestamp: ts, version: 1,
+ })
+ seedSourceSeg(t, root, "seg-20260621b", indexModeDocSpec{
+ subject: "svc", entityTags: []string{"id"}, entityValues: []string{"abc"},
+ storedTags: map[string]string{"properties": "same"}, timestamp: ts, version: 2,
+ })
+ srcSidxDirs, err := collectSourceSidxDirs([]string{root})
+ require.NoError(t, err)
+ res, err := analyzeIndexModeGroup(context.Background(), srcSidxDirs, map[uint32]indexRuleInfo{}, svcSchemas(), 10)
+ require.NoError(t, err)
+ require.Equal(t, uint64(1), res.KeysWithDuplicates)
+ require.Zero(t, res.ValueConflictKeys, "identical values across versions are not a conflict")
+}
+
+func TestCopyIndexModeGroup_RejectsShardParts(t *testing.T) {
+ root := t.TempDir()
+ // A shard part on disk means a non-_top_n_result normal measure has data.
+ partDir := filepath.Join(root, "seg-20260621", "shard-0", "0000000000000001")
+ require.NoError(t, os.MkdirAll(partDir, storage.DirPerm))
+
+ _, err := copyIndexModeGroup(context.Background(), migration.EntryGroupInput{
+ Group: "sw_metadata",
+ TargetGroupRoot: filepath.Join(t.TempDir(), "tgt"),
+ SrcRoots: []string{root},
+ Interval: storage.IntervalRule{Unit: storage.DAY, Num: 1},
+ }, map[uint32]indexRuleInfo{}, svcSchemas())
+ require.Error(t, err)
+ require.Contains(t, err.Error(), "shard part")
+}
+
+func TestCopyIndexModeGroup_EmptyShardDirAllowed(t *testing.T) {
+ root := t.TempDir()
+ // The segment controller can create shard-0 eagerly (or a dataless
+ // _top_n_result leaves one behind) with NO part inside; that is not normal
+ // measure data and must not block an index-mode copy.
+ require.NoError(t, os.MkdirAll(filepath.Join(root, "seg-20260621", "shard-0"), storage.DirPerm))
+ seedSourceSeg(t, root, "seg-20260621", indexModeDocSpec{
+ subject: "svc", entityTags: []string{"id"}, entityValues: []string{"abc"},
+ storedTags: map[string]string{"properties": "v"}, timestamp: time.Now().UnixNano(), version: 1,
+ })
+
+ _, err := copyIndexModeGroup(context.Background(), migration.EntryGroupInput{
+ Group: "sw_metadata",
+ TargetGroupRoot: filepath.Join(t.TempDir(), "tgt"),
+ SrcRoots: []string{root},
+ Interval: storage.IntervalRule{Unit: storage.DAY, Num: 1},
+ }, map[uint32]indexRuleInfo{}, svcSchemas())
+ require.NoError(t, err, "an empty shard-* dir (no part) must not trip the guard")
+}
+
+// ── Phase 6: end-to-end (index rules + production write path + searchability) ──.
+//
+// These tests build a realistic index-mode measure schema that carries genuine
+// index rules — including a "url"-analyzer rule and a noSort:false rule — write
+// data points through the PRODUCTION field-construction path
+// (handleIndexMode + appendEntityTagsToIndexFields, the exact functions
+// processDataPoint calls, write_standalone.go:86-87), run the migration, and
+// then prove that after migration every data value is preserved AND every
+// index-only / analyzed / sorted field is still searchable on the target sidx.
+
+const (
+ e2eMeasureName = "service_traffic_minute"
+ // e2eRuleIDShortName binds the default-flavored rule (empty analyzer,
+ // noSort:true) and e2eRuleIDEndpointURL binds the url-analyzer, noSort:false
+ // rule. The numeric ids are arbitrary but distinct.
+ e2eRuleIDShortName = uint32(1001)
+ e2eRuleIDEndpointURL = uint32(1002)
+)
+
+// e2eIndexModeSchema returns a production-shaped index-mode measure plus its
+// index rules. The searchable tag family carries:
+// - id : entity tag (STRING), no index rule -> regenerated as
+// _im_entity_tag_id (index-only)
+// - short_name : default rule (empty analyzer, noSort:true)
+// - endpoint_url : url-analyzer rule, noSort:false
+// - properties : NO index rule -> stored-only tag
+func e2eIndexModeSchema() (*databasev1.Measure, []*databasev1.IndexRule) {
+ schema := &databasev1.Measure{
+ Metadata: &commonv1.Metadata{Name: e2eMeasureName, Group: "sw_metadata"},
+ IndexMode: true,
+ Entity: &databasev1.Entity{TagNames: []string{"id"}},
+ TagFamilies: []*databasev1.TagFamilySpec{
+ {Name: "searchable", Tags: []*databasev1.TagSpec{
+ {Name: "id", Type: databasev1.TagType_TAG_TYPE_STRING},
+ {Name: "short_name", Type: databasev1.TagType_TAG_TYPE_STRING},
+ {Name: "endpoint_url", Type: databasev1.TagType_TAG_TYPE_STRING},
+ {Name: "properties", Type: databasev1.TagType_TAG_TYPE_STRING},
+ }},
+ },
+ }
+ rules := []*databasev1.IndexRule{
+ {
+ Metadata: &commonv1.Metadata{Id: e2eRuleIDShortName, Name: "short_name", Group: "sw_metadata"},
+ Tags: []string{"short_name"},
+ Type: databasev1.IndexRule_TYPE_INVERTED,
+ Analyzer: index.AnalyzerUnspecified,
+ NoSort: true,
+ },
+ {
+ Metadata: &commonv1.Metadata{Id: e2eRuleIDEndpointURL, Name: "endpoint_url", Group: "sw_metadata"},
+ Tags: []string{"endpoint_url"},
+ Type: databasev1.IndexRule_TYPE_INVERTED,
+ Analyzer: index.AnalyzerURL,
+ NoSort: false,
+ },
+ }
+ return schema, rules
+}
+
+// e2eSchemasBySubject is the migration-side lookup view of the e2e schema,
+// mirroring what loadMeasureSchemas would produce from the backup catalog. It
+// also injects an empty _top_n_result to mimic the real per-group internal
+// measure (classifyGroup / copy ignore it, but its presence keeps the fixture
+// faithful).
+func e2eSchemasBySubject() map[string]*measureSchemaInfo {
+ return map[string]*measureSchemaInfo{
+ e2eMeasureName: {
+ Name: e2eMeasureName,
+ EntityTagNames: []string{"id"},
+ TagType: map[string]databasev1.TagType{
+ "id": databasev1.TagType_TAG_TYPE_STRING,
+ "short_name": databasev1.TagType_TAG_TYPE_STRING,
+ "endpoint_url": databasev1.TagType_TAG_TYPE_STRING,
+ "properties": databasev1.TagType_TAG_TYPE_STRING,
+ },
+ // short_name + endpoint_url are index-rule indexed; id is an entity
+ // tag with no rule -> it must regenerate _im_entity_tag_id.
+ IndexedEntityTags: map[string]struct{}{},
+ },
+ TopNSchemaName: {Name: TopNSchemaName, IndexMode: false},
+ }
+}
+
+// e2eRuleByID is the flat IndexRuleID -> {Analyzer, NoSort} table the slow-path
+// rebuild needs to restore analyzer / sort flags FieldKey.Marshal drops.
+func e2eRuleByID() map[uint32]indexRuleInfo {
+ return map[uint32]indexRuleInfo{
+ e2eRuleIDShortName: {Analyzer: index.AnalyzerUnspecified, NoSort: true},
+ e2eRuleIDEndpointURL: {Analyzer: index.AnalyzerURL, NoSort: false},
+ }
+}
+
+// e2eDataPoint is one logical index-mode row to write through the production
+// field-construction path.
+type e2eDataPoint struct {
+ id string
+ shortName string
+ endpointURL string
+ properties string
+ timestamp int64
+ version int64
+}
+
+// newE2EMeasure builds a minimal but real *measure for the e2e schema so the
+// production appendEntityTagsToIndexFields (which needs stm.indexSchema +
+// stm.schema) can run unchanged. OnIndexUpdate populates indexRuleLocators +
+// indexTagMap exactly as the runtime does on schema load.
+func newE2EMeasure(t *testing.T, schema *databasev1.Measure, rules []*databasev1.IndexRule) *measure {
+ t.Helper()
+ m := &measure{schema: schema}
+ m.OnIndexUpdate(rules)
+ return m
+}
+
+// e2eDocFor builds one index.Document via the PRODUCTION field-construction
+// functions handleIndexMode + appendEntityTagsToIndexFields — the same two calls
+// processDataPoint makes for an index-mode measure (write_standalone.go:86-87).
+// The WriteRequest carries tags positionally (spec=nil), so the schema's
+// tag-family order drives field assembly, just like the live write path.
+func e2eDocFor(t *testing.T, m *measure, dp e2eDataPoint) index.Document {
+ t.Helper()
+ req := &measurev1.WriteRequest{
+ Metadata: m.schema.Metadata,
+ DataPoint: &measurev1.DataPointValue{
+ Timestamp: timestamppb.New(time.Unix(0, dp.timestamp)),
+ Version: dp.version,
+ TagFamilies: []*modelv1.TagFamilyForWrite{{
+ Tags: []*modelv1.TagValue{
+ strTagValue(dp.id),
+ strTagValue(dp.shortName),
+ strTagValue(dp.endpointURL),
+ strTagValue(dp.properties),
+ },
+ }},
+ },
+ }
+ series := &pbv1.Series{Subject: m.schema.Metadata.Name, EntityValues: []*modelv1.TagValue{strTagValue(dp.id)}}
+ require.NoError(t, series.Marshal())
+
+ is := m.indexSchema.Load().(indexSchema)
+ fields := handleIndexMode(m.schema, req, is.indexRuleLocators, nil)
+ fields = appendEntityTagsToIndexFields(fields, m, series)
+ return index.Document{
+ DocID: uint64(series.ID),
+ EntityValues: series.Buffer,
+ Fields: fields,
+ Version: dp.version,
+ Timestamp: dp.timestamp,
+ }
+}
+
+// writeE2ESourceSeg writes the given data points into "<root>/<segName>/sidx"
+// through a real inverted.Store, exactly as the production flusher does in
+// writeCallback.Rev via segment.IndexDB().Update(indexModeDocs) (the Update path
+// resolves to UpdateSeriesBatch on the underlying store). Closing the store
+// commits the bluge snapshot so the migration can read it back.
+func writeE2ESourceSeg(t *testing.T, m *measure, root, segName string, dps ...e2eDataPoint) {
+ t.Helper()
+ sidxDir := filepath.Join(root, segName, directCopySidxDirName)
+ require.NoError(t, os.MkdirAll(sidxDir, storage.DirPerm))
+ store, err := inverted.NewStore(inverted.StoreOpts{Path: sidxDir, BatchWaitSec: 0})
+ require.NoError(t, err)
+ batch := index.Batch{Documents: make(index.Documents, 0, len(dps))}
+ for _, dp := range dps {
+ batch.Documents = append(batch.Documents, e2eDocFor(t, m, dp))
+ }
+ require.NoError(t, store.UpdateSeriesBatch(batch))
+ require.NoError(t, store.Close())
+}
+
+// countFieldMatches opens a sidx read-only with the raw bluge reader and counts
+// how many committed documents match a single-term query on the given field.
+// This is exactly how the runtime's index-mode query resolves a tag condition
+// (a term query on the marshaled FieldKey), so a non-zero count proves the field
+// is genuinely searchable — index-only fields included — after migration.
+func countFieldMatches(t *testing.T, sidxDir, field, term string) int {
+ t.Helper()
+ r, err := bluge.OpenReader(bluge.DefaultConfig(sidxDir))
+ require.NoError(t, err)
+ defer func() { require.NoError(t, r.Close()) }()
+ q := bluge.NewTermQuery(term)
+ q.SetField(field)
+ dmi, err := r.Search(context.Background(), bluge.NewAllMatches(q))
+ require.NoError(t, err)
+ n := 0
+ for {
+ match, nextErr := dmi.Next()
+ require.NoError(t, nextErr)
+ if match == nil {
+ break
+ }
+ n++
+ }
+ return n
+}
+
+// e2eSampleDataPoints returns a fixed set of index-mode rows spanning two hours,
+// each entity carrying a distinct short_name + endpoint_url so per-field
+// searchability is unambiguous.
+func e2eSampleDataPoints(base int64) []e2eDataPoint {
+ hour := int64(time.Hour)
+ return []e2eDataPoint{
+ {id: "svc-a", shortName: "alpha", endpointURL: "https://example.com/api/login", properties: "p-a", timestamp: base + hour, version: 1},
+ {id: "svc-b", shortName: "beta", endpointURL: "https://example.com/api/logout", properties: "p-b", timestamp: base + hour + int64(time.Minute), version: 1},
+ {id: "svc-c", shortName: "gamma", endpointURL: "https://shop.example.org/cart", properties: "p-c", timestamp: base + 5*hour, version: 1},
+ }
+}
+
+// assertSourceSearchable sanity-checks that the SOURCE sidx already answers the
+// per-field searchability queries, so a later target failure is unambiguously a
+// migration regression and not a bad fixture.
+func assertSourceSearchable(t *testing.T, sidxDir string) {
+ t.Helper()
+ // _im_name (measure subject) is index-only.
+ require.Equal(t, 3, countFieldMatches(t, sidxDir, index.IndexModeName, e2eMeasureName),
+ "source: all 3 docs must match by measure name (_im_name)")
+ // _im_entity_tag_id is index-only, regenerated from the series _id.
+ require.Equal(t, 1, countFieldMatches(t, sidxDir, index.IndexModeEntityTagPrefix+"id", "svc-a"),
+ "source: entity-tag search must hit exactly one doc")
+ // url-analyzer tag: "com" is a token only of the two https://example.com urls
+ // (the third doc's host is shop.example.org -> token "org", not "com").
+ require.Equal(t, 2, countFieldMatches(t, sidxDir, string(convert.Uint32ToBytes(e2eRuleIDEndpointURL)), "com"),
+ "source: url-analyzer tokenizes the host -> 2 example.com docs")
+ // default (noSort:true) keyword rule on short_name.
+ require.Equal(t, 1, countFieldMatches(t, sidxDir, string(convert.Uint32ToBytes(e2eRuleIDShortName)), "alpha"),
+ "source: keyword short_name search must hit exactly one doc")
+}
+
+func TestIndexModeMigration_AlignedByteCopy(t *testing.T) {
+ root := t.TempDir()
+ dst := filepath.Join(t.TempDir(), "tgt")
+ schema, rules := e2eIndexModeSchema()
+ m := newE2EMeasure(t, schema, rules)
+ ir := storage.IntervalRule{Unit: storage.DAY, Num: 1}
+ base := day20260621Nanos(t)
+
+ // All three docs fall in the same DAY bucket -> source seg interval ==
+ // target interval -> byte-copy fast path.
+ writeE2ESourceSeg(t, m, root, "seg-20260621", e2eSampleDataPoints(base)...)
+ srcSidx := filepath.Join(root, "seg-20260621", directCopySidxDirName)
+ assertSourceSearchable(t, srcSidx)
+
+ in := migration.EntryGroupInput{
+ EntryTag: "[entry 1/1]",
+ Group: "sw_metadata",
+ TargetGroupRoot: dst,
+ SrcRoots: []string{root},
+ Interval: ir,
+ }
+ res, err := copyIndexModeGroup(context.Background(), in, e2eRuleByID(), e2eSchemasBySubject())
+ require.NoError(t, err)
+ require.Equal(t, int64(3), res.Rows)
+ require.Equal(t, 1, res.Segments)
+ require.Positive(t, res.Bytes, "aligned single-target source must take the byte-copy fast path")
+
+ tgtSidx := filepath.Join(dst, "seg-20260621", directCopySidxDirName)
+ require.Equal(t, 3, countSidxDocsWith(t, tgtSidx, e2eRuleByID(), e2eSchemasBySubject()),
+ "target sidx doc count must match source")
+
+ // Value reconciliation: every (series, timestamp) digest matches the source.
+ report, err := verifyIndexModeGroup(context.Background(), in, e2eRuleByID(), e2eSchemasBySubject())
+ require.NoError(t, err)
+ require.Empty(t, report.ValueMismatches, "byte-copy must preserve every data value")
+ require.Equal(t, uint64(3), report.SrcDistinctIDs)
+ require.Equal(t, report.SrcDistinctIDs, report.TgtDistinctIDs)
+
+ // Searchability preserved on the target (byte-copy keeps all index flags).
+ assertSourceSearchable(t, tgtSidx)
+}
+
+func TestIndexModeMigration_SplitAndMergeRebuild(t *testing.T) {
+ root := t.TempDir()
+ dst := filepath.Join(t.TempDir(), "tgt")
+ schema, rules := e2eIndexModeSchema()
+ m := newE2EMeasure(t, schema, rules)
+ // Source data lived on a DAY grid; target is HOUR. Two source segs each map
+ // wholly into hour bucket 01 (so they BOTH route through the slow-path merge
+ // for that seg — the supported max-version dedup path), while a third source
+ // seg lands wholly in hour 05. This exercises split (01 vs 05) + merge
+ // (two sources into 01 keeping max version).
+ ir := storage.IntervalRule{Unit: storage.HOUR, Num: 1}
+ base := day20260621Nanos(t)
+ hour := int64(time.Hour)
+
+ // Source seg #1: svc-a@v1 + svc-b, both in hour 01.
+ writeE2ESourceSeg(
+ t, m, root, "seg-2026062101a",
+ e2eDataPoint{id: "svc-a", shortName: "alpha", endpointURL: "https://example.com/api/login", properties: "p-a", timestamp: base + hour, version: 1},
+ e2eDataPoint{id: "svc-b", shortName: "beta", endpointURL: "https://example.com/api/logout", properties: "p-b", timestamp: base + hour + int64(time.Minute), version: 1},
+ )
+ // Source seg #2: svc-a re-emitted at a HIGHER version (changed stored
+ // "properties"), also in hour 01 -> merges with seg #1's svc-a, max wins.
+ writeE2ESourceSeg(
+ t, m, root, "seg-2026062101b",
+ e2eDataPoint{id: "svc-a", shortName: "alpha", endpointURL: "https://example.com/api/login", properties: "p-a-v2", timestamp: base + hour, version: 5},
+ )
+ // Source seg #3: svc-c in hour 05 (a distinct split target).
+ writeE2ESourceSeg(
+ t, m, root, "seg-2026062105",
+ e2eDataPoint{id: "svc-c", shortName: "gamma", endpointURL: "https://shop.example.org/cart", properties: "p-c", timestamp: base + 5*hour, version: 1},
+ )
+
+ in := migration.EntryGroupInput{
+ EntryTag: "[entry 1/1]",
+ Group: "sw_metadata",
+ TargetGroupRoot: dst,
+ SrcRoots: []string{root},
+ Interval: ir,
+ }
+ res, err := copyIndexModeGroup(context.Background(), in, e2eRuleByID(), e2eSchemasBySubject())
+ require.NoError(t, err)
+ // hour 01: svc-a (merged) + svc-b; hour 05: svc-c -> 2 target segs.
+ require.Equal(t, 2, res.Segments)
+
+ hour01 := filepath.Join(dst, "seg-2026062101", directCopySidxDirName)
+ hour05 := filepath.Join(dst, "seg-2026062105", directCopySidxDirName)
+ require.Equal(t, 2, countSidxDocsWith(t, hour01, e2eRuleByID(), e2eSchemasBySubject()),
+ "hour 01 holds svc-a (merged) + svc-b")
+ require.Equal(t, 1, countSidxDocsWith(t, hour05, e2eRuleByID(), e2eSchemasBySubject()),
+ "hour 05 holds svc-c")
+
+ // Max-version merge: the surviving svc-a doc must carry version 5 + p-a-v2.
+ docs, err := readIndexModeDocs(context.Background(), hour01, e2eRuleByID(), e2eSchemasBySubject())
+ require.NoError(t, err)
+ var svcADoc *index.Document
+ for i := range docs {
+ var series pbv1.Series
+ require.NoError(t, series.Unmarshal(docs[i].EntityValues))
+ if series.EntityValues[0].GetStr().GetValue() == "svc-a" {
+ svcADoc = &docs[i]
+ }
+ }
+ require.NotNil(t, svcADoc, "svc-a must be present in hour 01")
+ require.Equal(t, int64(5), svcADoc.Version, "max version must win across merged sources")
+ pf, ok := fieldByMarshaledName(*svcADoc, "properties")
+ require.True(t, ok)
+ require.Equal(t, "p-a-v2", string(pf.GetBytes()), "the max-version doc's stored value must survive")
+
+ // Searchability across the split target segs: each analyzed / index-only /
+ // sorted field still resolves on the rebuilt sidx.
+ // _im_name on both hour segs.
+ require.Equal(t, 2, countFieldMatches(t, hour01, index.IndexModeName, e2eMeasureName))
+ require.Equal(t, 1, countFieldMatches(t, hour05, index.IndexModeName, e2eMeasureName))
+ // url-analyzer: "com" hits svc-a + svc-b in hour 01, "shop" hits svc-c in hour 05.
+ require.Equal(t, 2, countFieldMatches(t, hour01, string(convert.Uint32ToBytes(e2eRuleIDEndpointURL)), "com"))
+ require.Equal(t, 1, countFieldMatches(t, hour05, string(convert.Uint32ToBytes(e2eRuleIDEndpointURL)), "shop"))
+ // _im_entity_tag_id resolves the merged svc-a.
+ require.Equal(t, 1, countFieldMatches(t, hour01, index.IndexModeEntityTagPrefix+"id", "svc-a"))
+ // default keyword rule on short_name.
+ require.Equal(t, 1, countFieldMatches(t, hour01, string(convert.Uint32ToBytes(e2eRuleIDShortName)), "alpha"))
+}
+
+// TestIndexModeMigration_SearchabilityPreserved is the core Phase 6 assertion:
+// after a slow-path rebuild migration, the target sidx must reconcile every data
+// value with the source AND keep every field searchable — by measure name
+// (_im_name), by entity tag (_im_entity_tag_id, both index-only), by a
+// url-analyzer tag, and by a default keyword (noSort:true) tag — with hit counts
+// matching the source. This directly nails the "id present, searchability lost"
+// failure the rebuild path guards against.
+func TestIndexModeMigration_SearchabilityPreserved(t *testing.T) {
+ root := t.TempDir()
+ dst := filepath.Join(t.TempDir(), "tgt")
+ schema, rules := e2eIndexModeSchema()
+ m := newE2EMeasure(t, schema, rules)
+ // Target HOUR != implicit source DAY -> force the slow-path rebuild so the
+ // index-only _im_* fields and analyzer/sort flags are reconstructed (not
+ // byte-copied). If the rebuild dropped them, searchability would break here.
+ ir := storage.IntervalRule{Unit: storage.HOUR, Num: 1}
+ base := day20260621Nanos(t)
+ hour := int64(time.Hour)
+
+ // Split the three docs across TWO source segs that both map wholly into hour
+ // bucket 01. Because neither source exclusively owns that target seg, BOTH are
+ // routed through the slow-path rebuild (not byte-copy), forcing the index-only
+ // _im_* fields and analyzer/sort flags to be reconstructed. They all land in
+ // ONE target seg, so per-field source/target hit counts compare exactly.
+ srcSegA := []e2eDataPoint{
+ {id: "svc-a", shortName: "alpha", endpointURL: "https://example.com/api/login", properties: "p-a", timestamp: base + hour, version: 1},
+ {id: "svc-b", shortName: "beta", endpointURL: "https://example.com/api/logout", properties: "p-b", timestamp: base + hour + int64(time.Minute), version: 1},
+ }
+ srcSegB := []e2eDataPoint{
+ {id: "svc-c", shortName: "gamma", endpointURL: "https://shop.example.org/cart", properties: "p-c", timestamp: base + hour + int64(2*time.Minute), version: 1},
+ }
+ writeE2ESourceSeg(t, m, root, "seg-2026062101a", srcSegA...)
+ writeE2ESourceSeg(t, m, root, "seg-2026062101b", srcSegB...)
+ srcSidxA := filepath.Join(root, "seg-2026062101a", directCopySidxDirName)
+ srcSidxB := filepath.Join(root, "seg-2026062101b", directCopySidxDirName)
+
+ in := migration.EntryGroupInput{
+ EntryTag: "[entry 1/1]",
+ Group: "sw_metadata",
+ TargetGroupRoot: dst,
+ SrcRoots: []string{root},
+ Interval: ir,
+ }
+ res, err := copyIndexModeGroup(context.Background(), in, e2eRuleByID(), e2eSchemasBySubject())
+ require.NoError(t, err)
+ require.Equal(t, 1, res.Segments, "all docs share one hour bucket")
+ require.Equal(t, int64(3), res.Rows)
+ require.Zero(t, res.Bytes, "two sources into one seg force a slow-path rebuild, not a byte-copy")
+
+ tgtSidx := filepath.Join(dst, "seg-2026062101", directCopySidxDirName)
+
+ // (1) Value reconciliation: each (series, timestamp) digest equals the source.
+ report, err := verifyIndexModeGroup(context.Background(), in, e2eRuleByID(), e2eSchemasBySubject())
+ require.NoError(t, err)
+ require.Empty(t, report.ValueMismatches,
+ "slow-path rebuild must reconcile every value (incl. entity values + version)")
+ require.Equal(t, uint64(3), report.SrcDocs)
+ require.Equal(t, uint64(3), report.TgtDocs)
+ require.Equal(t, uint64(3), report.SrcDistinctIDs)
+ require.Equal(t, report.SrcDistinctIDs, report.TgtDistinctIDs, "doc-id count preserved")
+
+ // (2) Searchability: hit counts on the target equal the source per field. The
+ // source side sums both source sidx dirs so the comparison spans all docs.
+ fieldChecks := []struct {
+ name string
+ field string
+ term string
+ want int
+ }{
+ {"_im_name (measure subject, index-only)", index.IndexModeName, e2eMeasureName, 3},
+ {"_im_entity_tag_id (entity, index-only)", index.IndexModeEntityTagPrefix + "id", "svc-b", 1},
+ {"url-analyzer endpoint_url (host token)", string(convert.Uint32ToBytes(e2eRuleIDEndpointURL)), "com", 2},
+ {"url-analyzer endpoint_url (path token)", string(convert.Uint32ToBytes(e2eRuleIDEndpointURL)), "login", 1},
+ {"default keyword short_name (noSort:true)", string(convert.Uint32ToBytes(e2eRuleIDShortName)), "gamma", 1},
+ }
+ for _, c := range fieldChecks {
+ srcHits := countFieldMatches(t, srcSidxA, c.field, c.term) + countFieldMatches(t, srcSidxB, c.field, c.term)
+ tgtHits := countFieldMatches(t, tgtSidx, c.field, c.term)
+ require.Equal(t, c.want, srcHits, "source hit count mismatch for %s", c.name)
+ require.Equal(t, srcHits, tgtHits,
+ "target hit count for %s must equal source (searchability lost?)", c.name)
+ }
+}
+
+// countSidxDocsWith opens a sidx read-only with explicit rule/schema tables and
+// returns the rebuilt doc count.
+func countSidxDocsWith(t *testing.T, sidxDir string, ruleByID map[uint32]indexRuleInfo,
+ schemas map[string]*measureSchemaInfo,
+) int {
+ t.Helper()
+ docs, err := readIndexModeDocs(context.Background(), sidxDir, ruleByID, schemas)
+ require.NoError(t, err)
+ return len(docs)
+}
diff --git a/banyand/measure/migration_indexmode_e2e_test.go b/banyand/measure/migration_indexmode_e2e_test.go
new file mode 100644
index 000000000..6383fb7a7
--- /dev/null
+++ b/banyand/measure/migration_indexmode_e2e_test.go
@@ -0,0 +1,538 @@
+// 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"
+ "fmt"
+ "os"
+ "path/filepath"
+ "sort"
+ "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/proto"
+ "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"
+ "github.com/apache/skywalking-banyandb/banyand/internal/migration"
+ "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/pkg/bus"
+ "github.com/apache/skywalking-banyandb/pkg/index"
+)
+
+// End-to-end migration of an INDEX-MODE measure that carries an index rule.
+//
+// Index-mode measures store no columnar parts: every data point is one inverted
+// document under <seg>/sidx/, upserted by series per segment. The migration must
+// rebuild each doc (stored tags + regenerated index-only entity fields) AND
+// restore the index rule so the indexed tag stays searchable. This test drives
+// the whole path through the LIVE service over gRPC-equivalent pipelines:
+//
+// register group + index-mode measure + index rule + binding
+// -> write points (indexed `service_name` + entity `id`) across two segments
+// -> baseline: source answers a full scan AND an indexed-tag filter
+// -> migrate the quiescent tree
+// -> re-open the target and assert it answers BOTH queries identically.
+//
+// Every assertion that depends on asynchronous schema propagation, flushing or
+// segment discovery is wrapped in Eventually so a slow / low-CPU runner retries
+// instead of failing on a transient empty read.
+// Fixed identifiers for the index-mode e2e fixture. Package-level so the query
+// helper can reference the group/measure directly (they are invariant for this
+// spec) without taking them as always-constant parameters.
+const (
+ idxModeGroup = "e2e_indexmode_migration"
+ idxModeMeasureName = "e2e_indexmode_metric"
+)
+
+var _ = Describe("migration.RunCopy index-mode end-to-end (live service)", func() {
+ const (
+ group = idxModeGroup
+ measureName = idxModeMeasureName
+ ruleName = "e2e_indexmode_service_name_rule"
+ bindingName = "e2e_indexmode_binding"
+ idxTag = "service_name"
+ entityTag = "id"
+ pointCount = 6 // 3 per segment, unique entity id + service_name each
+ )
+
+ It("merges two source segments through the rebuild (slow) path and the re-opened target answers full-scan + indexed-tag queries identically to the source", func() {
+ ctx := context.TODO()
+ // Source writes on a DAY×1 grid (group default); the migration re-grids to
+ // DAY×2 (the "hot" lifecycle stage), so the two adjacent 1-day source
+ // segments MERGE into a single 2-day target segment. A merge makes every
+ // source ineligible for the whole-sidx byte copy, forcing the rebuild
+ // (slow) path — which this test asserts both by content and by Bytes == 0.
+ twoDay := storage.IntervalRule{Unit: storage.DAY, Num: 2}
+ oneDayProto := &commonv1.IntervalRule{Unit: commonv1.IntervalRule_UNIT_DAY, Num: 1}
+ twoDayProto := &commonv1.IntervalRule{Unit: commonv1.IntervalRule_UNIT_DAY, Num: 2}
+ ttlProto := &commonv1.IntervalRule{Unit: commonv1.IntervalRule_UNIT_DAY, Num: 30}
+ // Source: DAY×1 write grid + a "hot" stage that re-grids to DAY×2 at migration.
+ sourceOpts := &commonv1.ResourceOpts{
+ ShardNum: 1,
+ SegmentInterval: oneDayProto,
+ Ttl: ttlProto,
+ Stages: []*commonv1.LifecycleStage{{
+ Name: "hot", ShardNum: 1, SegmentInterval: twoDayProto, Ttl: ttlProto, NodeSelector: "type=hot",
+ }},
+ }
+ // Target serves the migrated DAY×2 segments.
+ targetOpts := &commonv1.ResourceOpts{ShardNum: 1, SegmentInterval: twoDayProto, Ttl: ttlProto}
+
+ workspace := GinkgoT().TempDir()
+ sourceRoot := filepath.Join(workspace, "source")
+ targetRoot := filepath.Join(workspace, "target")
+
+ // ── Source: bring up the live service, register the index-mode schema. ──
+ 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())
+ registerIndexModeE2E(svcs, sourceOpts, group, measureName, ruleName, bindingName, idxTag, entityTag)
+ Eventually(func() bool {
+ _, ok := svcs.measure.LoadGroup(group)
+ return ok
+ }).WithTimeout(30 * time.Second).Should(BeTrue())
+ // Gate writes on the index rule being APPLIED, not merely on the measure
+ // resolving. The binding event arrives on a separate channel after the
+ // measure loads, so a measure can resolve before its index-rule locators
+ // exist — and a point written in that window lands UN-indexed permanently,
+ // which no later wait can repair. GetIndexRules() reflects the same index
+ // schema the write path consults, so it is the deterministic ready signal.
+ Eventually(func() []string {
+ return measureIndexRuleNames(svcs, group, measureName)
+ }).WithTimeout(30*time.Second).WithPolling(time.Second).Should(ContainElement(ruleName),
+ "source must apply the index rule before writes, or points land un-indexed")
+
+ // Two timestamps in DIFFERENT 1-day buckets but the SAME 2-day bucket, so
+ // the source flushes two 1-day segments that the migration merges into one.
+ bucket := twoDay.Standard(time.Now().UTC().AddDate(0, 0, -6))
+ day1 := bucket.Add(8 * time.Hour)
+ day2 := bucket.Add(32 * time.Hour)
+ writeIndexModeE2EPoints(svcs, group, measureName, day1, pointCount/2, 0)
+ writeIndexModeE2EPoints(svcs, group, measureName, day2, pointCount/2, pointCount/2)
+
+ // Wait until both 1-day source segments have materialized on disk.
+ Eventually(func() int {
+ info, err := svcs.measure.CollectDataInfo(ctx, group)
+ if err != nil || info == nil {
+ return 0
+ }
+ return len(info.SegmentInfo)
+ }).WithTimeout(60*time.Second).WithPolling(time.Second).Should(Equal(2),
+ "expected 2 DAY×1 source segments on disk after the writes are flushed")
+
+ queryBegin := day1.Add(-time.Hour)
+ queryEnd := day2.Add(time.Hour)
+
+ // Baseline (pre-migration): the source must answer a full scan and an
+ // indexed-tag filter. Capturing the SOURCE's full-scan content here lets the
+ // post-migration assertions compare target == source field-for-field, and
+ // isolates a migration regression from an indexing one — if the source
+ // itself cannot serve the indexed query, the data was never indexed.
+ //
+ // srcContent is captured from the SAME query result that satisfies the
+ // assertion, so the oracle is never a separate, racy re-query.
+ var srcContent map[string]string
+ Eventually(func() map[string]string {
+ dps := queryIndexModeE2E(svcs, queryBegin, queryEnd, nil)
+ srcContent = idxModeContentByID(dps)
+ return idxModeIDToTag(dps, idxTag)
+ }).WithTimeout(30*time.Second).WithPolling(time.Second).Should(Equal(expectedIDToTag(pointCount)),
+ "source full scan should return every point with its indexed tag value")
+ Eventually(func() []string {
+ return idxModeEntities(queryIndexModeE2E(svcs, queryBegin, queryEnd, eqCriteria(idxTag, "svc-2")))
+ }).WithTimeout(30*time.Second).WithPolling(time.Second).Should(ConsistOf("id-2"),
+ "source indexed-tag filter service_name=svc-2 should return exactly id-2 (proves indexing)")
+
+ // Dump the live registry (group + measure + index rule) into a synthetic
+ // schema-property catalog the migration reads, then wait for every sidx to
+ // snapshot and stop the source so the tree is quiescent for the copy.
+ targetDataPath := filepath.Join(targetRoot, "measure", "data")
+ staging := filepath.Join(workspace, "staging")
+ schemaPropertyRoot := filepath.Join(workspace, "schema-property")
+ seedIndexModeE2ESchemaProperty(svcs, schemaPropertyRoot, group, measureName, ruleName)
+ sourceDataPath := svcs.measure.(interface{ GetDataPath() string }).GetDataPath()
+ 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)
+ sourceDown()
+
+ // ── Migrate the quiescent tree. ──
+ plan := &migration.CopyPlan{
+ Source: migration.CopySource{Live: &migration.LiveSource{
+ SchemaPropertyPath: schemaPropertyRoot,
+ Stages: map[string][]migration.LiveStageNode{
+ "hot": {{Node: "node-0", Root: sourceDataPath}},
+ },
+ }},
+ Groups: []string{group},
+ Entries: []migration.CopyEntry{
+ {Stage: "hot", Target: targetDataPath, Nodes: []string{"node-0"}},
+ },
+ }
+ executors := []migration.CatalogExecutor{measure.NewMigrationExecutor()}
+ cls, err := plan.ClassifyGroups(executors)
+ Expect(err).NotTo(HaveOccurred())
+ res, err := plan.RunCopy(ctx, staging, cls, executors)
+ Expect(err).NotTo(HaveOccurred())
+
+ // Slow-path proof: index-mode records copied Bytes ONLY on the whole-sidx
+ // byte-copy fast path; the rebuild path writes docs via UpdateSeriesBatch and
+ // adds zero bytes. Two sources merging into one target make both ineligible
+ // for byte-copy, so a pure rebuild run reports Bytes == 0 with all rows written.
+ Expect(res.Totals.Rows).To(Equal(int64(pointCount)), "every source doc should be rewritten")
+ Expect(res.Totals.Bytes).To(BeZero(),
+ "the two source segments merge into one target segment, so every source takes the rebuild (slow) path — a byte-copy would have recorded copied bytes")
+
+ // ── Target: re-open from a fresh service (DAY×2 grid), re-register, and
+ // assert it answers the SAME queries as the source. ──
+ svcs2, deferFn2 := setUpMigrationTarget(targetRoot)
+ defer deferFn2()
+ Eventually(func() bool {
+ _, ok := svcs2.measure.LoadGroup("sw_metric")
+ return ok
+ }).WithTimeout(30 * time.Second).Should(BeTrue())
+ registerIndexModeE2E(svcs2, targetOpts, group, measureName, ruleName, bindingName, idxTag, entityTag)
+ Eventually(func() bool {
+ _, ok := svcs2.measure.LoadGroup(group)
+ return ok
+ }).WithTimeout(30 * time.Second).Should(BeTrue())
+ // Wait for the index rule to be APPLIED (not just the measure resolved) so
+ // the indexed-tag filter below is asserting migration fidelity, not racing
+ // the target's own binding propagation.
+ Eventually(func() []string {
+ return measureIndexRuleNames(svcs2, group, measureName)
+ }).WithTimeout(30*time.Second).WithPolling(time.Second).Should(ContainElement(ruleName),
+ "re-opened service must apply the index rule before the indexed query runs")
+
+ // The two 1-day source segments collapsed into a single 2-day target segment.
+ Eventually(func() int {
+ info, ierr := svcs2.measure.CollectDataInfo(ctx, group)
+ if ierr != nil || info == nil {
+ return 0
+ }
+ return len(info.SegmentInfo)
+ }).WithTimeout(30*time.Second).WithPolling(time.Second).Should(Equal(1),
+ "the merge should leave exactly one DAY×2 target segment")
+
+ // Content equality: the migrated target answers the full scan with the EXACT
+ // same per-series content (every tag value + timestamp) as the source query.
+ Eventually(func() map[string]string {
+ return idxModeContentByID(queryIndexModeE2E(svcs2, queryBegin, queryEnd, nil))
+ }).WithTimeout(30*time.Second).WithPolling(time.Second).Should(Equal(srcContent),
+ "post-migration full scan content must match the source query content field-for-field")
+
+ // Indexed-tag filter: the index rule survived the migration, so the
+ // migrated sidx is still searchable by the indexed tag.
+ Eventually(func() []string {
+ return idxModeEntities(queryIndexModeE2E(svcs2, queryBegin, queryEnd, eqCriteria(idxTag, "svc-2")))
+ }).WithTimeout(30*time.Second).WithPolling(time.Second).Should(ConsistOf("id-2"),
+ "post-migration indexed-tag filter service_name=svc-2 should still return exactly id-2")
+ })
+})
+
+// registerIndexModeE2E creates a CATALOG_MEASURE group, an index-mode measure
+// whose `default` family holds an entity tag plus one searchable string tag, and
+// an INVERTED index rule (+ binding) on that searchable tag. Re-callable on both
+// the source and the re-opened target service (AlreadyExists is tolerated). opts
+// carries the group's segment grid + lifecycle stages, so the source can write on
+// one grid while the target serves the migrated (re-gridded) grid.
+func registerIndexModeE2E(svcs *services, opts *commonv1.ResourceOpts, group, measureName, ruleName, bindingName, idxTag, entityTag string) {
+ ctx := context.TODO()
+ createOK := func(err error) {
+ if err != nil && !isAlreadyExistsErr(err) {
+ Expect(err).NotTo(HaveOccurred())
+ }
+ }
+ _, err := svcs.metadataService.GroupRegistry().CreateGroup(ctx, &commonv1.Group{
+ Metadata: &commonv1.Metadata{Name: group},
+ Catalog: commonv1.Catalog_CATALOG_MEASURE,
+ ResourceOpts: opts,
+ })
+ createOK(err)
+ _, err = svcs.metadataService.MeasureRegistry().CreateMeasure(ctx, &databasev1.Measure{
+ Metadata: &commonv1.Metadata{Name: measureName, Group: group},
+ IndexMode: true,
+ Entity: &databasev1.Entity{TagNames: []string{entityTag}},
+ TagFamilies: []*databasev1.TagFamilySpec{
+ {Name: defaultTagFamily, Tags: []*databasev1.TagSpec{
+ {Name: entityTag, Type: databasev1.TagType_TAG_TYPE_STRING},
+ {Name: idxTag, Type: databasev1.TagType_TAG_TYPE_STRING},
+ }},
+ },
+ })
+ createOK(err)
+ _, err = svcs.metadataService.IndexRuleRegistry().CreateIndexRule(ctx, &databasev1.IndexRule{
+ Metadata: &commonv1.Metadata{Name: ruleName, Group: group},
+ Tags: []string{idxTag},
+ Type: databasev1.IndexRule_TYPE_INVERTED,
+ Analyzer: index.AnalyzerKeyword,
+ })
+ createOK(err)
+ _, err = svcs.metadataService.IndexRuleBindingRegistry().CreateIndexRuleBinding(ctx, &databasev1.IndexRuleBinding{
+ Metadata: &commonv1.Metadata{Name: bindingName, Group: group},
+ Rules: []string{ruleName},
+ Subject: &databasev1.Subject{Catalog: commonv1.Catalog_CATALOG_MEASURE, Name: measureName},
+ BeginAt: timestamppb.New(time.Now().Add(-time.Hour)),
+ ExpireAt: timestamppb.New(time.Now().Add(365 * 24 * time.Hour)),
+ })
+ createOK(err)
+}
+
+// writeIndexModeE2EPoints publishes `count` index-mode points starting at
+// baseTime (1s apart), each with a unique entity id-<n> and searchable tag
+// value svc-<n> so every point is its own series and the indexed filter is
+// selective.
+func writeIndexModeE2EPoints(svcs *services, group, name string, baseTime time.Time, count, offset int) {
+ bp := svcs.pipeline.NewBatchPublisher(5 * time.Second)
+ defer bp.Close()
+ for i := 0; i < count; i++ {
+ idx := offset + i
+ idVal := "id-" + strconv.Itoa(idx)
+ svcVal := "svc-" + 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)),
+ // Index-mode stores _version only when non-zero (like _timestamp);
+ // a real data point always carries one. Use a unique, monotonic
+ // version per series so the upsert/merge keeps a deterministic doc.
+ Version: int64(idx + 1),
+ TagFamilies: []*modelv1.TagFamilyForWrite{{
+ Tags: []*modelv1.TagValue{
+ {Value: &modelv1.TagValue_Str{Str: &modelv1.Str{Value: idVal}}},
+ {Value: &modelv1.TagValue_Str{Str: &modelv1.Str{Value: svcVal}}},
+ },
+ }},
+ },
+ }
+ bp.Publish(context.TODO(), data.TopicMeasureWrite,
+ bus.NewMessage(bus.MessageID(idx), &measurev1.InternalWriteRequest{
+ EntityValues: []*modelv1.TagValue{
+ {Value: &modelv1.TagValue_Str{Str: &modelv1.Str{Value: idVal}}},
+ },
+ Request: req,
+ }))
+ }
+}
+
+// seedIndexModeE2ESchemaProperty writes the live Group + Measure + IndexRule into
+// a synthetic schema-property bluge catalog so the migration restores the
+// index-mode schema (and the rule that keeps the indexed tag searchable).
+func seedIndexModeE2ESchemaProperty(svcs *services, root, group, measureName, ruleName 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())
+ ruleProto, err := svcs.metadataService.IndexRuleRegistry().GetIndexRule(ctx,
+ &commonv1.Metadata{Name: ruleName, 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)))
+ ruleJSON, err := protojson.Marshal(ruleProto)
+ Expect(err).NotTo(HaveOccurred())
+ batch.Insert(migrationE2EBlugeDoc("index-rule/"+group+"/"+ruleName,
+ schema.KindIndexRule.String(), group, string(ruleJSON)))
+ Expect(w.Batch(batch)).To(Succeed())
+}
+
+// queryIndexModeE2E runs a measure query (optionally filtered) through the
+// service pipeline and returns the data points.
+func queryIndexModeE2E(svcs *services, begin, end time.Time, criteria *modelv1.Criteria) []*measurev1.DataPoint {
+ req := &measurev1.QueryRequest{
+ Groups: []string{idxModeGroup},
+ Name: idxModeMeasureName,
+ TimeRange: &modelv1.TimeRange{Begin: timestamppb.New(begin), End: timestamppb.New(end)},
+ TagProjection: &modelv1.TagProjection{
+ TagFamilies: []*modelv1.TagProjection_TagFamily{
+ {Name: defaultTagFamily, Tags: []string{"id", "service_name"}},
+ },
+ },
+ Criteria: criteria,
+ }
+ 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("index-mode measure query returned error: %s", d.Error()))
+ default:
+ Fail(fmt.Sprintf("expected *measurev1.QueryResponse, got %T", d))
+ }
+ return nil
+}
+
+func eqCriteria(tag, value string) *modelv1.Criteria {
+ return &modelv1.Criteria{
+ Exp: &modelv1.Criteria_Condition{
+ Condition: &modelv1.Condition{
+ Name: tag,
+ Op: modelv1.Condition_BINARY_OP_EQ,
+ Value: &modelv1.TagValue{Value: &modelv1.TagValue_Str{Str: &modelv1.Str{Value: value}}},
+ },
+ },
+ }
+}
+
+// measureIndexRuleNames returns the names of the index rules currently applied
+// to the measure, or nil if it has not resolved yet. This reads the same index
+// schema the write path uses, so "the rule appears here" is the deterministic
+// signal that subsequent writes/queries are indexed.
+func measureIndexRuleNames(svcs *services, group, measureName string) []string {
+ m, err := svcs.measure.Measure(&commonv1.Metadata{Name: measureName, Group: group})
+ if err != nil || m == nil {
+ return nil
+ }
+ var names []string
+ for _, r := range m.GetIndexRules() {
+ names = append(names, r.GetMetadata().GetName())
+ }
+ return names
+}
+
+// idxModeContentByID keys each data point by its entity id and maps it to a
+// stable digest of its FULL content — every tag value, every field value, and
+// the timestamp, sorted so the digest is order-independent. Comparing the source
+// and target digests asserts the migration preserved every field of every point.
+func idxModeContentByID(dps []*measurev1.DataPoint) map[string]string {
+ out := make(map[string]string, len(dps))
+ for _, dp := range dps {
+ var id string
+ parts := make([]string, 0, 4)
+ for _, tf := range dp.GetTagFamilies() {
+ for _, tag := range tf.GetTags() {
+ v := canonValue(tag.GetValue())
+ parts = append(parts, "t:"+tf.GetName()+"."+tag.GetKey()+"="+v)
+ if tag.GetKey() == "id" {
+ if sv := tag.GetValue().GetStr(); sv != nil {
+ id = sv.GetValue()
+ }
+ }
+ }
+ }
+ for _, f := range dp.GetFields() {
+ parts = append(parts, "f:"+f.GetName()+"="+canonValue(f.GetValue()))
+ }
+ parts = append(parts, "@"+strconv.FormatInt(dp.GetTimestamp().AsTime().UnixNano(), 10))
+ sort.Strings(parts)
+ out[id] = strings.Join(parts, "|")
+ }
+ return out
+}
+
+// canonValue renders any proto value (tag or field) as a stable JSON string.
+func canonValue(m proto.Message) string {
+ b, err := protojson.Marshal(m)
+ Expect(err).NotTo(HaveOccurred())
+ return string(b)
+}
+
+// idxModeIDToTag maps each data point's entity id to its indexed tag value.
+func idxModeIDToTag(dps []*measurev1.DataPoint, idxTag string) map[string]string {
+ out := map[string]string{}
+ for _, dp := range dps {
+ var id, tagVal string
+ for _, tf := range dp.TagFamilies {
+ for _, tag := range tf.Tags {
+ switch tag.Key {
+ case "id":
+ if sv := tag.Value.GetStr(); sv != nil {
+ id = sv.Value
+ }
+ case idxTag:
+ if sv := tag.Value.GetStr(); sv != nil {
+ tagVal = sv.Value
+ }
+ }
+ }
+ }
+ if id != "" {
+ out[id] = tagVal
+ }
+ }
+ return out
+}
+
+// idxModeEntities returns the sorted entity ids of the data points.
+func idxModeEntities(dps []*measurev1.DataPoint) []string {
+ out := make([]string, 0, len(dps))
+ for _, dp := range dps {
+ for _, tf := range dp.TagFamilies {
+ for _, tag := range tf.Tags {
+ if tag.Key == "id" {
+ if sv := tag.Value.GetStr(); sv != nil {
+ out = append(out, sv.Value)
+ }
+ }
+ }
+ }
+ }
+ return out
+}
+
+// expectedIDToTag builds the id-<n> -> svc-<n> map the writes produce.
+func expectedIDToTag(count int) map[string]string {
+ out := make(map[string]string, count)
+ for i := 0; i < count; i++ {
+ out["id-"+strconv.Itoa(i)] = "svc-" + strconv.Itoa(i)
+ }
+ return out
+}
diff --git a/banyand/measure/migration_schema.go b/banyand/measure/migration_schema.go
index cb38c4259..c34853ca1 100644
--- a/banyand/measure/migration_schema.go
+++ b/banyand/measure/migration_schema.go
@@ -45,32 +45,62 @@ func loadMeasureSchemas(schemaRoot string, groups []string) (map[string]map[stri
// schema directly from its proto definition.
func measureSchemaInfoFromProto(group string, m *databasev1.Measure) *measureSchemaInfo {
si := &measureSchemaInfo{
- Group: group,
- Name: m.GetMetadata().GetName(),
- IndexMode: m.GetIndexMode(),
+ Group: group,
+ Name: m.GetMetadata().GetName(),
+ IndexMode: m.GetIndexMode(),
+ EntityTagNames: m.GetEntity().GetTagNames(),
+ TagType: make(map[string]databasev1.TagType),
}
for _, tf := range m.GetTagFamilies() {
tags := make([]string, 0, len(tf.GetTags()))
for _, t := range tf.GetTags() {
tags = append(tags, t.GetName())
+ si.TagType[t.GetName()] = t.GetType()
}
si.TagFamilies = append(si.TagFamilies, tagFamilyInfo{Name: tf.GetName(), Tags: tags})
}
return si
}
+// loadIndexRuleInfoByID returns IndexRuleID -> {Analyzer, NoSort} across the
+// given groups. The slow-path rebuild needs it because FieldKey.Marshal only
+// encodes TagName or IndexRuleID and drops Analyzer/NoSort.
+func loadIndexRuleInfoByID(schemaRoot string, groups []string) (map[uint32]indexRuleInfo, error) {
+ rules, err := reader.LoadIndexRules(schemaRoot, groups)
+ if err != nil {
+ return nil, err
+ }
+ out := make(map[uint32]indexRuleInfo)
+ for _, byGroup := range rules {
+ for _, r := range byGroup {
+ out[r.GetMetadata().GetId()] = indexRuleInfo{Analyzer: r.GetAnalyzer(), NoSort: r.GetNoSort()}
+ }
+ }
+ return out, nil
+}
+
// measureSchemaInfo is the subset of a measure's schema the migration copy
// 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
- TagFamilies []tagFamilyInfo
- IndexMode bool
+ TagType map[string]databasev1.TagType
+ IndexedEntityTags map[string]struct{}
+ Group string
+ Name string
+ TagFamilies []tagFamilyInfo
+ EntityTagNames []string
+ IndexMode bool
}
type tagFamilyInfo struct {
Name string
Tags []string
}
+
+// indexRuleInfo is the minimal index-rule view the slow-path rebuild needs: it
+// restores the Analyzer and NoSort flags that FieldKey.Marshal does not encode.
+type indexRuleInfo struct {
+ Analyzer string
+ NoSort bool
+}
diff --git a/banyand/measure/migration_schema_test.go b/banyand/measure/migration_schema_test.go
index 86ecb187b..d00b5b228 100644
--- a/banyand/measure/migration_schema_test.go
+++ b/banyand/measure/migration_schema_test.go
@@ -170,6 +170,99 @@ func synthGroupsFromNames(names ...string) []synthGroup {
return out
}
+func TestMeasureSchemaInfo_IndexModeFields(t *testing.T) {
+ m := &databasev1.Measure{
+ Metadata: &commonv1.Metadata{Name: "svc", Group: "g"},
+ IndexMode: true,
+ Entity: &databasev1.Entity{TagNames: []string{"id"}},
+ TagFamilies: []*databasev1.TagFamilySpec{
+ {Name: "searchable", Tags: []*databasev1.TagSpec{
+ {Name: "id", Type: databasev1.TagType_TAG_TYPE_STRING},
+ {Name: "weight", Type: databasev1.TagType_TAG_TYPE_INT},
+ }},
+ },
+ }
+ info := measureSchemaInfoFromProto("g", m)
+ if !info.IndexMode {
+ t.Fatal("expected IndexMode true")
+ }
+ if len(info.EntityTagNames) != 1 || info.EntityTagNames[0] != "id" {
+ t.Fatalf("EntityTagNames = %v, want [id]", info.EntityTagNames)
+ }
+ if info.TagType["id"] != databasev1.TagType_TAG_TYPE_STRING {
+ t.Fatalf("TagType[id] = %v, want STRING", info.TagType["id"])
+ }
+ if info.TagType["weight"] != databasev1.TagType_TAG_TYPE_INT {
+ t.Fatalf("TagType[weight] = %v, want INT", info.TagType["weight"])
+ }
+}
+
+// synthIndexRule describes one index rule to seed into the synthetic catalog.
+type synthIndexRule struct {
+ group string
+ name string
+ analyzer string
+ id uint32
+ noSort bool
+}
+
+// synthIndexRules appends index-rule docs to an existing synthetic backup tree
+// built by synthBackup, reusing the same schema-property shard layout.
+func synthIndexRules(t *testing.T, root string, rules []synthIndexRule) {
+ t.Helper()
+ shardPath := filepath.Join(root, "node-0", "2026-05-21",
+ backupsnapshot.SchemaPropertyCatalogName, schema.SchemaGroup, "shard-0")
+ 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 _, r := range rules {
+ rule := &databasev1.IndexRule{
+ Metadata: &commonv1.Metadata{Name: r.name, Group: r.group, Id: r.id, ModRevision: 1},
+ Analyzer: r.analyzer,
+ NoSort: r.noSort,
+ }
+ rJSON, mErr := protojson.Marshal(rule)
+ if mErr != nil {
+ t.Fatalf("marshal index rule %q: %v", r.name, mErr)
+ }
+ propID := fmt.Sprintf("index-rule/%s/%s", r.group, r.name)
+ batch.Insert(synthSchemaDocWithPropID(propID, propID,
+ schema.KindIndexRule.String(), r.group, 1, string(rJSON)))
+ }
+ if bErr := w.Batch(batch); bErr != nil {
+ t.Fatalf("write index rule batch: %v", bErr)
+ }
+}
+
+func TestLoadIndexRuleInfoByID(t *testing.T) {
+ dir := t.TempDir()
+ synthBackup(t, dir, synthGroupsFromNames("g_a"), []synthMeasure{
+ {group: "g_a", name: "m_a1", modRev: 1},
+ })
+ synthIndexRules(t, dir, []synthIndexRule{
+ {group: "g_a", name: "r_default", id: 100, analyzer: "", noSort: true},
+ {group: "g_a", name: "r_url", id: 201559343, analyzer: index.AnalyzerURL, noSort: false},
+ })
+
+ byID, err := loadIndexRuleInfoByID(synthSchemaRoot(t, dir), []string{"g_a"})
+ if err != nil {
+ t.Fatalf("loadIndexRuleInfoByID: %v", err)
+ }
+ if got := byID[100]; got.Analyzer != "" || !got.NoSort {
+ t.Fatalf("rule 100 = %+v, want {Analyzer:\"\", NoSort:true}", got)
+ }
+ if got := byID[201559343]; got.Analyzer != index.AnalyzerURL || got.NoSort {
+ t.Fatalf("rule 201559343 = %+v, want {Analyzer:url, NoSort:false}", got)
+ }
+}
+
func TestLoadMeasureSchemas_returnsRequestedGroups(t *testing.T) {
dir := t.TempDir()
groups := []string{"g_a", "g_b", "g_c"}
diff --git a/banyand/measure/migration_verify.go b/banyand/measure/migration_verify.go
index 08864ad5d..57969e2da 100644
--- a/banyand/measure/migration_verify.go
+++ b/banyand/measure/migration_verify.go
@@ -60,16 +60,54 @@ type SegmentReport struct {
// 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.
+//
+// IndexMode-prefixed fields are populated only when the (entry, group) is an
+// index-mode measure group: those groups carry their data in the segment sidx
+// (one doc per series), so they are reconciled at the document level (doc count,
+// distinct doc-id, per-(series,timestamp) value digest) instead of part rows.
type EntryGroupReport struct {
- Group string
- EntryStage string
- EntryTarget string
- TargetGroup string
- EntryNodes []string
- SrcRoots []string
- TargetSegs []SegmentReport
- SrcRows uint64
- SrcParts int
+ Group string
+ EntryStage string
+ EntryTarget string
+ TargetGroup string
+ EntryNodes []string
+ SrcRoots []string
+ TargetSegs []SegmentReport
+ ValueMismatches []IndexModeValueMismatch
+ SegAligns []IndexModeSegAlign
+ SrcRows uint64
+ SrcDocs uint64
+ TgtDocs uint64
+ SrcDistinctIDs uint64
+ TgtDistinctIDs uint64
+ ExpectDistinctIDs uint64
+ SrcParts int
+ IndexMode bool
+}
+
+// IndexModeValueMismatch is one (seriesID, timestamp) key whose value digest did
+// not reconcile between source and target during an index-mode verify. The
+// reconciliation unit is (Segment, SeriesID) — the unit the runtime upserts
+// index-mode data on. Kind is one of "missing" (in source, absent in target),
+// "extra" (in target, absent in source) or "tampered" (present on both sides but
+// the digests differ — a changed value or a dropped index-only field).
+type IndexModeValueMismatch struct {
+ Kind string
+ Segment string
+ SeriesID uint64
+}
+
+// IndexModeSegAlign reports one target segment's alignment during an index-mode
+// verify: how many source segments fed it (SrcSegs), how many source docs routed
+// in (SrcDocs) and how many target docs survived (TgtDocs). Aligned is true only
+// when exactly one source segment fed the target and no docs collapsed (a 1:1
+// byte-copy candidate); otherwise the segment was merged or split-into.
+type IndexModeSegAlign struct {
+ Segment string
+ SrcSegs int
+ SrcDocs uint64
+ TgtDocs uint64
+ Aligned bool
}
// VerifyShardParts reads <shardDir>'s newest `.snp`, confirms every
diff --git a/banyand/metadata/schema/reader/loaders.go b/banyand/metadata/schema/reader/loaders.go
index 6101d8570..eea2ff365 100644
--- a/banyand/metadata/schema/reader/loaders.go
+++ b/banyand/metadata/schema/reader/loaders.go
@@ -69,6 +69,12 @@ func LoadMeasures(schemaRoot string, groups []string) (map[string][]*databasev1.
return loadGroupScoped(schemaRoot, schema.KindMeasure, groups, func() *databasev1.Measure { return &databasev1.Measure{} })
}
+// LoadIndexRules returns the latest non-deleted index rules of the requested
+// groups, keyed by group name.
+func LoadIndexRules(schemaRoot string, groups []string) (map[string][]*databasev1.IndexRule, error) {
+ return loadGroupScoped(schemaRoot, schema.KindIndexRule, groups, func() *databasev1.IndexRule { return &databasev1.IndexRule{} })
+}
+
// loadGroupScoped loads one group-scoped Kind (stream / measure) into
// (group -> latest non-deleted protos). A doc missing its source tag is an
// error: unlike Group docs these are the payload the migration depends on.