You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@skywalking.apache.org by li...@apache.org on 2026/06/23 02:51:56 UTC
(skywalking-banyandb) branch migration-index-measure created (now e3c4aca6e)
This is an automated email from the ASF dual-hosted git repository.
mrproliu pushed a change to branch migration-index-measure
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
at e3c4aca6e Support migration copy/verify/analysis index mode measure
This branch includes the following new commits:
new e3c4aca6e Support migration copy/verify/analysis index mode measure
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(skywalking-banyandb) 01/01: Support migration copy/verify/analysis index mode measure
Posted by li...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
mrproliu pushed a commit to branch migration-index-measure
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
commit e3c4aca6e9c4514d630b2bea3fcd080f1c8eae77
Author: mrproliu <74...@qq.com>
AuthorDate: Tue Jun 23 10:51:40 2026 +0800
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 | 44 +-
banyand/measure/migration_indexmode_copy.go | 1112 ++++++++++++++++
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(+), 129 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..c91a826fa 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, timestamp).")
+ 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..b7399a15e 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,8 +47,10 @@ 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.
@@ -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)
+ 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..67b91bbd4
--- /dev/null
+++ b/banyand/measure/migration_indexmode_copy.go
@@ -0,0 +1,1112 @@
+// 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,
+) (migration.EntryGroupResult, error) {
+ var res migration.EntryGroupResult
+ 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})
+ if err != nil {
+ return res, fmt.Errorf("load measure schemas: %w", err)
+ }
+ ruleByID, err := loadIndexRuleInfoByID(schemaRoot, []string{group})
+ 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.