You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@skywalking.apache.org by ha...@apache.org on 2026/08/10 05:09:07 UTC
(skywalking-banyandb) branch main updated: perf(trace): halve sampler tag-rule cost via an escape-free var-array decode path (#1271)
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
The following commit(s) were added to refs/heads/main by this push:
new 3d6282605 perf(trace): halve sampler tag-rule cost via an escape-free var-array decode path (#1271)
3d6282605 is described below
commit 3d62826054c5a0eef77b62573741835995b5292d
Author: Gao Hongtao <ha...@gmail.com>
AuthorDate: Mon Aug 10 13:09:01 2026 +0800
perf(trace): halve sampler tag-rule cost via an escape-free var-array decode path (#1271)
---
CHANGES.md | 3 +
.../trace-pipeline-merge-optimization-plan.md | 56 ++-
pkg/encoding/vararray/vararray.go | 54 ++-
pkg/encoding/vararray/vararray_test.go | 421 +++++++++++++++++++++
.../skywalking/internal/tracesampler/sampler.go | 222 ++++++++---
.../tracesampler/sampler_bench_ext_test.go | 7 +-
.../internal/tracesampler/sampler_test.go | 150 ++++++++
test/cases/schema/clamp.go | 34 +-
8 files changed, 864 insertions(+), 83 deletions(-)
diff --git a/CHANGES.md b/CHANGES.md
index 8bd6049e1..506414a9d 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -173,6 +173,9 @@ Release Notes.
- Bump Go and ui/mcp dependencies to clear Dependabot advisories: `golang.org/x/net` v0.52.0→v0.56.0 (CVE-2026-25680), `opencontainers/runc` v1.3.3→v1.3.6 (CVE-2026-41579). Refresh mcp/ui lockfiles and license attribution.
- Defer the sw-trace-sampler tag-array decode to first use: 60% per-trace cost reduction on the realistic-mix path (`duration+err+tag` 2,710 → 1,015 ns/trace, 2,380 → 664 allocs/op on the SkyWalking schema).
- Alias decoded entry strings via `pkg/convert.BytesToString`: 408 → 1 alloc/op on SkyWalking tag paths and 1,536 → 1 on Zipkin tag paths (the realistic 3-row × 8-entry trace); 24-40% ns/trace reduction on tag-using paths. Also fixes a pre-existing nil-deref in `keepTrace` when the array column is absent.
+- Decode escape-free trace tag rows without copying them. The sw/zipkin trace sampler copied every tag row before decoding, because the var-array decode rewrites its input in place and the engine hands the same batch to every link of a sampler chain. That copy is only needed for a row that actually contains an escape, so rows without one now decode straight from the block with the entries aliasing it — safe because `assembleTraceBlock` already deep-copies each value into block-owned stor [...]
+- Precompute each tag rule's `"tagKey="` entry prefix at admission instead of rebuilding it on every `matchEntries` call (once per rule per trace).
+- Add an escape-free fast path to `vararray.UnmarshalVarArray`, the var-array decoder shared by sidx, trace query, index filtering, the dump tool and the sampler SDK. An entry containing no escape byte is located with two `bytes.IndexByte` scans instead of a per-byte loop, dropping that case from O(len) to O(1) (-91% at a 256-byte payload) and leaving the buffer untouched. An entry that does carry an escape resumes the in-place loop at the first escape rather than restarting, since the b [...]
## 0.10.0
diff --git a/docs/design/trace-pipeline-merge-optimization-plan.md b/docs/design/trace-pipeline-merge-optimization-plan.md
index ffe1e9e51..3fa8f49f0 100644
--- a/docs/design/trace-pipeline-merge-optimization-plan.md
+++ b/docs/design/trace-pipeline-merge-optimization-plan.md
@@ -379,13 +379,67 @@ fixes are kept separate from framework fixes.
- [ ] Independent measurements for projected tag/span decoding, complete-trace latency calculation, error and status
rules, trace-ID hashing, regex/tag rules, verdict allocation, decision throughput across representative batch sizes.
-- [ ] Each plugin-local cost attributed to a specific function call site.
+ *Every axis is measured (table below), but on SYNTHETIC batches, not the mature seed — see "Remaining gap".*
+- [x] Each plugin-local cost attributed to a specific function call site.
**Dependencies.** None (parallel workstream).
**Boundary rationale.** Outside the merge framework; can run any time after the mature seed is frozen. Parallel
workstream.
+**Per-axis measurements.** All figures are ns per logical trace from
+`plugins/skywalking/internal/tracesampler` (`BenchmarkDecide*`), median of 3 runs at `-benchtime 2000x`, taken after
+the escape-free decode work landed. Both first-party schemas are reported because they exercise different paths: the
+segment schema reads a dedicated `is_error` column, while Zipkin has none and detects errors inside the flattened tag
+array.
+
+| Axis | Benchmark | sw | zipkin |
+| --- | --- | ---: | ---: |
+| Projected tag decoding, 2 / 8 / 32 entries | `_ArrayEntries` (unescaped) | 70 / 149 / 496 | 67 / 147 / 491 |
+| Same, escaped values | `_ArrayEntries` (escaped) | 157 / 377 / 1261 | 159 / 380 / 1265 |
+| Complete-trace latency envelope | `durationOnly` | 238 | 241 |
+| Same, per row count 1 / 4 / 16 / 64 | `_RowCount` | 342 / 1176 / 4405 / 17333 | 494 / 1718 / 6632 / 25917 |
+| Error rule, dedicated column vs in-array | `errorsOnly` | 123 | 382 |
+| Trace-ID hashing | `sampleOnly` | 17 | 15 |
+| Tag rules: equals / regex / 5xx regex | `_RegexRule` | 373 / 384 / 387 | 376 / 386 / 385 |
+| Tag rules by match position 0 / 16 / 31 | `_TagMatchPosition` | 971 / 1038 / 1106 | 1034 / 1194 / 1327 |
+| Verdict allocation, 1 / 16 / 64 / 256 traces | `_KeepSliceAllocation` | 204 / 135 / 135 / 127 | 517 / 384 / 381 / 384 |
+| Decision throughput, 1 / 16 / 64 / 128 / 256 | `_BatchSize` | 1136 / 922 / 902 / 897 / 905 | 1655 / 1327 / 1315 / 1325 / 1332 |
+
+Two shape facts fall out. Verdict allocation is one `allocs/op` at every batch size — the `make([]bool, len(batch.Traces))`
+in `Decide` — with `B/op` equal to the trace count, so it is already at its floor and per-trace cost only falls as the
+batch amortizes it. Decision throughput is flat from 16 traces upward; the `traces=1` column is that same fixed cost
+divided by one, not a batching penalty.
+
+**Cost attribution to call sites.** From `-cpuprofile` at `-benchtime 30000x`, one profile per axis. Percentages are
+flat unless marked cumulative.
+
+| Axis | Dominant call site | Share |
+| --- | --- | ---: |
+| Latency envelope | `runtime.duffcopy` under `sdk.(*TagColumn).At` (`hasSlowTrace` 91.8% cum) | 57.1% |
+| Error rule, dedicated column | `runtime.duffcopy` under `sdk.(*TagColumn).At` (`At` 51.9% cum) | 44.4% |
+| Error rule, in-array | `indexbytebody` (`arrayEntries` 55.3% cum, `matchEntries` 39.5% cum) | 39.5% |
+| Tag rules | `indexbytebody` (`arrayEntries` 66.7% cum, `matchEntries` 21.3% cum) | 40.0% |
+| Trace-ID hashing | `hash/fnv.(*sum64a).Write` | 60.0% |
+| Verdict allocation | `make([]bool, …)` in `Decide` | 1 alloc/op |
+
+The `duffcopy` entries are one finding, not two: `sdk.Value` is **112 bytes** and `TagColumn.At` returns it by value, so
+every per-row column read on the duration and dedicated-error paths copies 112 bytes. That copy — not the decode — is
+the majority cost of both, and it is an SDK-side shape, so fixing it (an `AtInto(*Value)`, or accessors that avoid
+materializing the struct) is a framework change rather than a plugin-local one and belongs outside this workstream.
+
+On the array paths the remaining cost is the delimiter scan itself plus the string comparisons in `matchEntries`; the
+decode overhead that used to dominate here was removed by the escape-free path (`tagRulesOnly` 68.4 → 24.4 µs per
+3000-trace batch, -64%).
+
+**Remaining gap (why this phase is not closed).** The hard invariant requires plugins to be measured "against
+representative complete-trace batches from the mature seed". These benchmarks build their batches synthetically in
+`benchBatch`/`benchEntries` — realistic in SHAPE (production tag keys, both first-party configs, complete traces) but
+not drawn from the frozen selection. The machinery to close this exists — `tracefixture.EvaluateSampler` already drives
+a sampler over complete generated traces via `buildSamplerBlock` — so the work is to add a seed-backed benchmark
+alongside the synthetic ones, not to build a fixture. Until then the first exit-gate item stays open and no
+production-ratio claim should be made from the numbers above.
+
---
### Phase 10 — Final Integration
diff --git a/pkg/encoding/vararray/vararray.go b/pkg/encoding/vararray/vararray.go
index 242d27194..13fb0227c 100644
--- a/pkg/encoding/vararray/vararray.go
+++ b/pkg/encoding/vararray/vararray.go
@@ -56,12 +56,19 @@ func MarshalVarArray(dest, src []byte) []byte {
// UnmarshalVarArray unmarshals a variable-length array from src starting at idx.
//
-// WARNING: This function mutates src. Decoding is performed in-place by
-// overwriting bytes in src[idx:next) to remove escape characters. The decoded
-// value is the view src[idx:end] into the same backing array; copy it (for
-// example, with bytes.Clone or append([]byte(nil), ...)) if you need to
-// preserve the original encoded buffer or keep the decoded value independent of
-// subsequent in-place decoding on the same buffer.
+// WARNING: This function mutates src when the entry contains an escape.
+// Decoding is then performed in-place by overwriting bytes in src[idx:next) to
+// remove escape characters. The decoded value is the view src[idx:end] into the
+// same backing array; copy it (for example, with bytes.Clone or
+// append([]byte(nil), ...)) if you need to preserve the original encoded buffer
+// or keep the decoded value independent of subsequent in-place decoding on the
+// same buffer.
+//
+// An entry with no escape byte is decoded without writing to src at all. Do NOT
+// read that as "src is safe from mutation": whether a write happens is a
+// property of the DATA, not of the call. A caller may only skip its defensive
+// copy behind its own escape check (e.g. bytes.IndexByte(row, Escape) < 0) —
+// never unconditionally.
//
// It returns:
// - end: the index of the first byte after the decoded value (exclusive)
@@ -82,9 +89,40 @@ func UnmarshalVarArray(src []byte, idx int) (int, int, error) {
// Empty value; value is src[idx:idx], next starts after the delimiter.
return idx, idx + 1, nil
}
+ // Fast path for an entry that carries no escape. The loop below advances
+ // writeIdx and readIdx in lockstep until the first escape is consumed, so
+ // until then every `src[writeIdx] = b` writes a byte back over itself. An
+ // escape-free entry therefore leaves src byte-identical and returns the
+ // delimiter position — which two IndexByte scans find without touching the
+ // buffer at all. The escape search is bounded by the delimiter so an escape
+ // belonging to a LATER entry cannot drag this one onto the slow path.
+ //
+ // The trade is deliberate: an escape-free entry drops from O(len) to O(1)
+ // (-91% at a 256-byte payload). An entry that DOES carry an escape pays the
+ // two scans but skips the loop up to the first escape, which for a value with
+ // one escape is a large win too (-89% at 256 bytes); only escape-dense values
+ // (an escape every few bytes, where the scans buy no skip) pay a net cost.
+ // End-to-end the sampler tag-rule benchmarks move -35%.
+ //
+ // A missing delimiter falls through: the loop reports "invalid escape
+ // character" for a trailing escape and "invalid variable array" otherwise,
+ // and that distinction is part of the contract.
+ start := idx
+ if rel := bytes.IndexByte(src[idx:], EntityDelimiter); rel >= 0 {
+ esc := bytes.IndexByte(src[idx:idx+rel], Escape)
+ if esc < 0 {
+ return idx + rel, idx + rel + 1, nil
+ }
+ // Escaped entry, so the value must be shifted — but the scan is not
+ // wasted. Every byte before the first escape is neither escape nor
+ // delimiter, so the loop would only self-assign it. Resume at the escape
+ // with writeIdx still in lockstep. For a value carrying a single escape
+ // this skips nearly the whole entry.
+ start = idx + esc
+ }
// Decode in-place: read index i, write index j.
- writeIdx := idx
- for readIdx := idx; readIdx < len(src); readIdx++ {
+ writeIdx := start
+ for readIdx := start; readIdx < len(src); readIdx++ {
b := src[readIdx]
switch {
case b == Escape:
diff --git a/pkg/encoding/vararray/vararray_test.go b/pkg/encoding/vararray/vararray_test.go
new file mode 100644
index 000000000..2f0bd9260
--- /dev/null
+++ b/pkg/encoding/vararray/vararray_test.go
@@ -0,0 +1,421 @@
+// 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 vararray
+
+import (
+ "bytes"
+ "errors"
+ "fmt"
+ "testing"
+
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+)
+
+// unmarshalReference is a verbatim copy of the pre-fast-path UnmarshalVarArray
+// implementation. It is the oracle for the differential tests: any change to
+// UnmarshalVarArray must agree with this function on the returned triple AND on
+// the post-call contents of src, for every input.
+//
+// Do not "simplify" this function. It is deliberately frozen.
+func unmarshalReference(src []byte, idx int) (int, int, error) {
+ if idx >= len(src) {
+ return 0, 0, errors.New("empty entity value")
+ }
+ if src[idx] == EntityDelimiter {
+ return idx, idx + 1, nil
+ }
+ writeIdx := idx
+ for readIdx := idx; readIdx < len(src); readIdx++ {
+ b := src[readIdx]
+ switch {
+ case b == Escape:
+ if readIdx+1 >= len(src) {
+ return 0, 0, errors.New("invalid escape character")
+ }
+ readIdx++
+ src[writeIdx] = src[readIdx]
+ writeIdx++
+ case b == EntityDelimiter:
+ return writeIdx, readIdx + 1, nil
+ default:
+ src[writeIdx] = b
+ writeIdx++
+ }
+ }
+ return 0, 0, errors.New("invalid variable array")
+}
+
+// unmarshalRestartAtIdx is the rejected fast-path variant: it takes the same
+// escape-free shortcut, but on an escaped entry it restarts the in-place loop at
+// idx instead of resuming at the first escape, so the two scans buy it nothing.
+// Kept so the choice between the two shapes stays measurable in ONE process
+// (BenchmarkUnmarshalVarArray_AB) rather than resting on a comparison across
+// benchmark runs, which is what first led to picking the wrong one.
+func unmarshalRestartAtIdx(src []byte, idx int) (int, int, error) {
+ if idx >= len(src) {
+ return 0, 0, errors.New("empty entity value")
+ }
+ if src[idx] == EntityDelimiter {
+ return idx, idx + 1, nil
+ }
+ if rel := bytes.IndexByte(src[idx:], EntityDelimiter); rel >= 0 &&
+ bytes.IndexByte(src[idx:idx+rel], Escape) < 0 {
+ return idx + rel, idx + rel + 1, nil
+ }
+ writeIdx := idx
+ for readIdx := idx; readIdx < len(src); readIdx++ {
+ b := src[readIdx]
+ switch {
+ case b == Escape:
+ if readIdx+1 >= len(src) {
+ return 0, 0, errors.New("invalid escape character")
+ }
+ readIdx++
+ src[writeIdx] = src[readIdx]
+ writeIdx++
+ case b == EntityDelimiter:
+ return writeIdx, readIdx + 1, nil
+ default:
+ src[writeIdx] = b
+ writeIdx++
+ }
+ }
+ return 0, 0, errors.New("invalid variable array")
+}
+
+// TestUnmarshalRestartAtIdx_MatchesReference keeps the rejected variant honest:
+// it must stay a valid implementation, otherwise benchmarking against it is
+// meaningless.
+func TestUnmarshalRestartAtIdx_MatchesReference(t *testing.T) {
+ enumerate(6, func(src []byte) {
+ for idx := 0; idx <= len(src); idx++ {
+ got := append([]byte(nil), src...)
+ want := append([]byte(nil), src...)
+ gotEnd, gotNext, gotErr := unmarshalRestartAtIdx(got, idx)
+ wantEnd, wantNext, wantErr := unmarshalReference(want, idx)
+ input := fmt.Sprintf("src=%q idx=%d", src, idx)
+ require.Equal(t, wantErr == nil, gotErr == nil, input)
+ require.Equal(t, wantEnd, gotEnd, input)
+ require.Equal(t, wantNext, gotNext, input)
+ require.True(t, bytes.Equal(want, got), input)
+ }
+ })
+}
+
+func TestUnmarshalVarArray_Table(t *testing.T) {
+ tests := []struct {
+ name string
+ src string
+ wantVal string
+ errMsg string
+ idx int
+ wantEnd int
+ wantNxt int
+ wantErr bool
+ }{
+ {name: "idx beyond length", src: "abc", idx: 3, wantErr: true, errMsg: "empty entity value"},
+ {name: "empty src", src: "", idx: 0, wantErr: true, errMsg: "empty entity value"},
+ {name: "empty value", src: "|", idx: 0, wantEnd: 0, wantNxt: 1, wantVal: ""},
+ {name: "no delimiter", src: "abc", idx: 0, wantErr: true, errMsg: "invalid variable array"},
+ {name: "invalid escape at end", src: "abc\\", idx: 0, wantErr: true, errMsg: "invalid escape character"},
+ {name: "lone escape", src: "\\", idx: 0, wantErr: true, errMsg: "invalid escape character"},
+ {name: "simple value", src: "abc|", idx: 0, wantEnd: 3, wantNxt: 4, wantVal: "abc"},
+ {name: "escaped delimiter", src: "a\\|b|", idx: 0, wantEnd: 3, wantNxt: 5, wantVal: "a|b"},
+ {name: "escaped escape", src: "c\\\\d|", idx: 0, wantEnd: 3, wantNxt: 5, wantVal: "c\\d"},
+ {name: "value is a single escaped delimiter", src: "\\||", idx: 0, wantEnd: 1, wantNxt: 3, wantVal: "|"},
+ {name: "consecutive escaped escapes", src: "\\\\\\\\|", idx: 0, wantEnd: 2, wantNxt: 5, wantVal: "\\\\"},
+ // The escape here sits AFTER the terminating delimiter of the entry being
+ // decoded. A decoder that searches the whole remaining buffer for an escape
+ // before deciding whether the entry is escape-free would mis-handle this.
+ {name: "escape after this entry's delimiter", src: "ab|c\\|d|", idx: 0, wantEnd: 2, wantNxt: 3, wantVal: "ab"},
+ {name: "second entry carries the escape", src: "ab|c\\|d|", idx: 3, wantEnd: 6, wantNxt: 8, wantVal: "c|d"},
+ {name: "empty entry mid sequence", src: "a||b|", idx: 1, wantEnd: 1, wantNxt: 2, wantVal: ""},
+ {name: "second empty entry mid sequence", src: "a||b|", idx: 2, wantEnd: 2, wantNxt: 3, wantVal: ""},
+ {name: "entry after empty entry", src: "a||b|", idx: 3, wantEnd: 4, wantNxt: 5, wantVal: "b"},
+ }
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ src := []byte(tt.src)
+ end, next, err := UnmarshalVarArray(src, tt.idx)
+ if tt.wantErr {
+ require.Error(t, err)
+ assert.Equal(t, tt.errMsg, err.Error())
+ assert.Equal(t, 0, end)
+ assert.Equal(t, 0, next)
+ return
+ }
+ require.NoError(t, err)
+ assert.Equal(t, tt.wantEnd, end)
+ assert.Equal(t, tt.wantNxt, next)
+ assert.Equal(t, tt.wantVal, string(src[tt.idx:end]))
+ })
+ }
+}
+
+// TestUnmarshalVarArray_MutationContract pins the documented in-place behavior.
+// Nothing else in the repository asserts what src looks like after the call, yet
+// three call sites deep-copy their input specifically because of it.
+func TestUnmarshalVarArray_MutationContract(t *testing.T) {
+ t.Run("escape-free entry leaves src byte-identical", func(t *testing.T) {
+ const encoded = "abc|def|"
+ src := []byte(encoded)
+ _, _, err := UnmarshalVarArray(src, 0)
+ require.NoError(t, err)
+ assert.Equal(t, encoded, string(src), "an entry with no escape must not modify the buffer")
+ })
+
+ t.Run("empty entry leaves src byte-identical", func(t *testing.T) {
+ const encoded = "|abc|"
+ src := []byte(encoded)
+ _, _, err := UnmarshalVarArray(src, 0)
+ require.NoError(t, err)
+ assert.Equal(t, encoded, string(src))
+ })
+
+ t.Run("escaped entry shifts bytes left in place", func(t *testing.T) {
+ // "a\|b|" decodes to "a|b": the escape is removed by shifting the tail
+ // left one byte, so the buffer is rewritten and the trailing byte is stale.
+ src := []byte("a\\|b|")
+ end, next, err := UnmarshalVarArray(src, 0)
+ require.NoError(t, err)
+ assert.Equal(t, "a|b", string(src[0:end]))
+ assert.Equal(t, 5, next)
+ assert.Equal(t, "a|bb|", string(src), "decoded bytes are written over the source in place")
+ })
+}
+
+func TestUnmarshalVarArray_Iteration(t *testing.T) {
+ var encoded []byte
+ values := [][]byte{[]byte("a|b"), []byte("c\\d"), []byte(""), []byte("plain"), []byte("|"), []byte("\\")}
+ for _, v := range values {
+ encoded = MarshalVarArray(encoded, v)
+ }
+ var decoded [][]byte
+ for idx := 0; idx < len(encoded); {
+ end, next, err := UnmarshalVarArray(encoded, idx)
+ require.NoError(t, err)
+ decoded = append(decoded, append([]byte(nil), encoded[idx:end]...))
+ idx = next
+ }
+ require.Len(t, decoded, len(values))
+ for i := range values {
+ // Compared as strings: an empty entry decodes to a nil slice rather than
+ // an empty one, which is the same value for every caller of this codec.
+ assert.Equal(t, string(values[i]), string(decoded[i]), "entry %d", i)
+ }
+}
+
+// alphabet is restricted to one ordinary byte plus the two bytes with special
+// meaning, so short exhaustive enumeration covers every structural arrangement
+// of delimiters and escapes rather than sampling them randomly.
+var alphabet = []byte{'a', EntityDelimiter, Escape}
+
+// enumerate calls fn with every string over alphabet of length 1..maxLen.
+func enumerate(maxLen int, fn func(src []byte)) {
+ for length := 1; length <= maxLen; length++ {
+ word := make([]byte, length)
+ var rec func(pos int)
+ rec = func(pos int) {
+ if pos == length {
+ fn(word)
+ return
+ }
+ for _, c := range alphabet {
+ word[pos] = c
+ rec(pos + 1)
+ }
+ }
+ rec(0)
+ }
+}
+
+// TestUnmarshalVarArray_DifferentialExhaustive compares UnmarshalVarArray against
+// the frozen reference over every string of length 1..6 built from {'a', '|', '\'},
+// at every start index. Both the returned triple and the post-call buffer must match.
+func TestUnmarshalVarArray_DifferentialExhaustive(t *testing.T) {
+ cases := 0
+ enumerate(6, func(src []byte) {
+ for idx := 0; idx <= len(src); idx++ {
+ got := append([]byte(nil), src...)
+ want := append([]byte(nil), src...)
+ gotEnd, gotNext, gotErr := UnmarshalVarArray(got, idx)
+ wantEnd, wantNext, wantErr := unmarshalReference(want, idx)
+
+ input := fmt.Sprintf("src=%q idx=%d", src, idx)
+ if wantErr != nil {
+ require.Error(t, gotErr, input)
+ require.Equal(t, wantErr.Error(), gotErr.Error(), input)
+ } else {
+ require.NoError(t, gotErr, input)
+ }
+ require.Equal(t, wantEnd, gotEnd, "end mismatch: %s", input)
+ require.Equal(t, wantNext, gotNext, "next mismatch: %s", input)
+ require.True(t, bytes.Equal(want, got),
+ "post-call buffer mismatch: %s reference=%q actual=%q", input, want, got)
+ cases++
+ }
+ })
+ t.Logf("compared %d (input, idx) pairs", cases)
+}
+
+func FuzzUnmarshalVarArray_Differential(f *testing.F) {
+ f.Add([]byte("abc|"), 0)
+ f.Add([]byte("a\\|b|"), 0)
+ f.Add([]byte("ab|c\\|d|"), 3)
+ f.Add([]byte("\\\\|"), 0)
+ f.Add([]byte(""), 0)
+ f.Add([]byte("no-delimiter"), 0)
+ f.Fuzz(func(t *testing.T, src []byte, idx int) {
+ if idx < 0 || idx > len(src) {
+ t.Skip()
+ }
+ got := append([]byte(nil), src...)
+ want := append([]byte(nil), src...)
+ gotEnd, gotNext, gotErr := UnmarshalVarArray(got, idx)
+ wantEnd, wantNext, wantErr := unmarshalReference(want, idx)
+
+ if (gotErr == nil) != (wantErr == nil) {
+ t.Fatalf("error mismatch for src=%q idx=%d: got %v, reference %v", src, idx, gotErr, wantErr)
+ }
+ if gotErr != nil && gotErr.Error() != wantErr.Error() {
+ t.Fatalf("error text mismatch for src=%q idx=%d: got %q, reference %q", src, idx, gotErr, wantErr)
+ }
+ if gotEnd != wantEnd || gotNext != wantNext {
+ t.Fatalf("index mismatch for src=%q idx=%d: got (%d,%d), reference (%d,%d)", src, idx, gotEnd, gotNext, wantEnd, wantNext)
+ }
+ if !bytes.Equal(got, want) {
+ t.Fatalf("post-call buffer mismatch for src=%q idx=%d: got %q, reference %q", src, idx, got, want)
+ }
+ })
+}
+
+func FuzzVarArrayRoundTrip(f *testing.F) {
+ f.Add([]byte("plain"), []byte("a|b"), []byte("c\\d"))
+ f.Add([]byte(""), []byte("|"), []byte("\\"))
+ f.Add([]byte("\\\\"), []byte("||"), []byte(""))
+ f.Fuzz(func(t *testing.T, a, b, c []byte) {
+ values := [][]byte{a, b, c}
+ var encoded []byte
+ for _, v := range values {
+ encoded = MarshalVarArray(encoded, v)
+ }
+ var decoded [][]byte
+ for idx := 0; idx < len(encoded); {
+ end, next, err := UnmarshalVarArray(encoded, idx)
+ if err != nil {
+ t.Fatalf("round-trip decode failed for %q: %v", values, err)
+ }
+ decoded = append(decoded, append([]byte(nil), encoded[idx:end]...))
+ idx = next
+ }
+ if len(decoded) != len(values) {
+ t.Fatalf("entry count mismatch for %q: got %d, want %d", values, len(decoded), len(values))
+ }
+ for i := range values {
+ if !bytes.Equal(decoded[i], values[i]) {
+ t.Fatalf("entry %d mismatch: got %q, want %q", i, decoded[i], values[i])
+ }
+ }
+ })
+}
+
+// benchEntry builds a var-array holding one entry of the requested payload length.
+// escapeEvery controls escape density: 0 means none (the common case for trace tag
+// values, which rarely contain '|' or '\'), 1 means a single escape near the end
+// (a realistic worst case for a value that does), and n>1 puts a delimiter at every
+// nth byte (a pathological density that maximizes in-place shifting).
+func benchEntry(payloadLen, escapeEvery int) []byte {
+ payload := make([]byte, payloadLen)
+ for i := range payload {
+ payload[i] = 'a'
+ }
+ switch {
+ case escapeEvery == 1:
+ payload[payloadLen-1] = EntityDelimiter
+ case escapeEvery > 1:
+ for i := escapeEvery - 1; i < payloadLen; i += escapeEvery {
+ payload[i] = EntityDelimiter
+ }
+ }
+ return MarshalVarArray(nil, payload)
+}
+
+// benchKinds are the escape densities the A/B benchmark sweeps.
+var benchKinds = []struct {
+ name string
+ escapeEvery int
+}{
+ {name: "clean", escapeEvery: 0},
+ {name: "oneEscape", escapeEvery: 1},
+ {name: "escapeDense", escapeEvery: 4},
+}
+
+// BenchmarkUnmarshalVarArray_AB runs the current implementation and the frozen
+// reference over the same inputs inside ONE process, so the comparison cannot be
+// confounded by machine drift between two separate benchmark runs.
+func BenchmarkUnmarshalVarArray_AB(b *testing.B) {
+ impls := []struct {
+ fn func([]byte, int) (int, int, error)
+ name string
+ }{
+ {name: "ref", fn: unmarshalReference},
+ {name: "fast", fn: UnmarshalVarArray},
+ {name: "restartAtIdx", fn: unmarshalRestartAtIdx},
+ }
+ for _, payloadLen := range []int{8, 16, 64, 256} {
+ for _, kind := range benchKinds {
+ src := benchEntry(payloadLen, kind.escapeEvery)
+ for _, impl := range impls {
+ b.Run(fmt.Sprintf("%s/len=%d/%s", kind.name, payloadLen, impl.name), func(b *testing.B) {
+ scratch := make([]byte, len(src))
+ b.ReportAllocs()
+ b.ResetTimer()
+ for i := 0; i < b.N; i++ {
+ copy(scratch, src)
+ if _, _, err := impl.fn(scratch, 0); err != nil {
+ b.Fatal(err)
+ }
+ }
+ })
+ }
+ }
+ }
+}
+
+func BenchmarkUnmarshalVarArray(b *testing.B) {
+ for _, payloadLen := range []int{8, 16, 64, 256} {
+ for _, kind := range benchKinds {
+ // The per-iteration restore is inside the timed region. It is a
+ // constant across implementations, so the arm still compares like
+ // for like; it is required because an escaped entry mutates src.
+ src := benchEntry(payloadLen, kind.escapeEvery)
+ b.Run(fmt.Sprintf("%s/len=%d", kind.name, payloadLen), func(b *testing.B) {
+ scratch := make([]byte, len(src))
+ b.ReportAllocs()
+ b.ResetTimer()
+ for i := 0; i < b.N; i++ {
+ copy(scratch, src)
+ if _, _, err := UnmarshalVarArray(scratch, 0); err != nil {
+ b.Fatal(err)
+ }
+ }
+ })
+ }
+ }
+}
diff --git a/plugins/skywalking/internal/tracesampler/sampler.go b/plugins/skywalking/internal/tracesampler/sampler.go
index cd3eb7a48..cf70a7cd4 100644
--- a/plugins/skywalking/internal/tracesampler/sampler.go
+++ b/plugins/skywalking/internal/tracesampler/sampler.go
@@ -169,7 +169,13 @@ func (s Schema) firstClassColumn(tagKey, errorColumn string) string {
// rule is one sure-keep tag predicate. Exactly one matcher is honored, checked
// in the order exists, equals, in, regex.
type rule struct {
- re *regexp.Regexp
+ re *regexp.Regexp
+ // prefix is the "tagKey=" that an entry must carry for this rule to apply,
+ // precomputed at admission because matchEntries runs per rule per trace and
+ // the concatenation showed up as runtime.concatstrings in the tag-rule
+ // profile. Empty when the rule targets the array column itself, which is the
+ // case that matches a bare entry rather than a "key=value" one.
+ prefix string
Regex string `json:"regex"`
TagKey string `json:"tagKey"`
Equals string `json:"equals"`
@@ -400,7 +406,10 @@ func New(configJSON []byte, schema Schema) (sdk.Sampler, error) {
s.errorTagInArray = schema.ErrorTagInArray
// The in-array error signal is just an exists rule on the error key, evaluated
// by the same matcher as keepTagRules so the two can never diverge.
+ // Built here rather than through validateRules, so its prefix has to be
+ // set explicitly — matchEntries reads the precomputed field for every rule.
s.errorRule = rule{TagKey: s.errorTag, Exists: true}
+ s.errorRule.setPrefix(s.arrayColumn)
}
if c.DurationThresholdMs < 0 {
return nil, fmt.Errorf("tracesampler: durationThresholdMs must be >= 0, got %d", c.DurationThresholdMs)
@@ -484,10 +493,20 @@ func validateRules(rs []rule, schema Schema, errorColumn string) error {
}
r.re = re
}
+ r.setPrefix(schema.ArrayTagColumn)
}
return nil
}
+// setPrefix precomputes the rule's entry prefix against the flattened array
+// column. A rule naming the array column itself matches a bare entry, so it has
+// no prefix; every other rule matches "tagKey=value" entries.
+func (r *rule) setPrefix(arrayColumn string) {
+ if r.TagKey != arrayColumn {
+ r.prefix = r.TagKey + "="
+ }
+}
+
// Kind reports the sampler kind, satisfying the generic sdk.Plugin interface
// that sdk.Sampler embeds.
func (s *Sampler) Kind() sdk.Kind { return sdk.KindSampler }
@@ -566,26 +585,33 @@ func (s *Sampler) keepTrace(b *sdk.TraceBlock) bool {
// release) keeps this variable on the stack — the pool owns the backing
// storage, so taking the pointer's address is unnecessary.
//
- // stableBufPtr is the *[]byte that backs the per-row entry strings. The
- // strings alias its backing array via convert.BytesToString, so the array
- // must outlive every matchEntries call below. The lazy decode acquires
- // both pointers on the first call site that needs them; both are released
- // at the single exit below. A nil stableBufPtr means no entries were
- // decoded (skip path) and there is nothing to release.
+ // stableBufPtr is the *[]byte that backs the entry strings of any ESCAPED
+ // row, which arrayEntries acquires and hands back only when it needed one.
+ // Those strings alias its backing array via convert.BytesToString, so the
+ // array must outlive every matchEntries call below; it is released at the
+ // single exit. A nil stableBufPtr means no row needed unescaping (the
+ // common case) or nothing was decoded at all, and there is nothing to
+ // release.
+ //
+ // decoded, not a nil-check on entriesPtr, is what makes the decode fire
+ // once: arrayEntries returns a nil entriesPtr on error, so keying the
+ // second call site off entriesPtr would decode twice — and overwrite (leak)
+ // the stable buffer of the first attempt.
var (
entriesPtr *[]string
stableBufPtr *[]byte
entriesErr error
+ decoded bool
)
keep := false
// Error keep.
if s.keepErrors {
if s.errorTagInArray {
- if entriesPtr == nil {
- stableBufPtr = acquireStableBuf()
- entriesPtr, entriesErr = arrayEntries(b.Tag(s.arrayColumn), stableBufPtr)
+ if !decoded {
+ decoded = true
+ entriesPtr, stableBufPtr, entriesErr = arrayEntries(b.Tag(s.arrayColumn))
}
- if entriesErr != nil || matchEntries(*entriesPtr, &s.errorRule, s.arrayColumn) {
+ if entriesErr != nil || matchEntries(*entriesPtr, &s.errorRule) {
keep = true
}
} else {
@@ -598,15 +624,15 @@ func (s *Sampler) keepTrace(b *sdk.TraceBlock) bool {
// Sure-keep tag rules. The decode is hoisted out of the loop so the
// lazy fetch fires exactly once even when multiple rules are configured.
if !keep && len(s.rules) > 0 {
- if entriesPtr == nil {
- stableBufPtr = acquireStableBuf()
- entriesPtr, entriesErr = arrayEntries(b.Tag(s.arrayColumn), stableBufPtr)
+ if !decoded {
+ decoded = true
+ entriesPtr, stableBufPtr, entriesErr = arrayEntries(b.Tag(s.arrayColumn))
}
if entriesErr != nil {
keep = true
} else {
for i := range s.rules {
- if matchEntries(*entriesPtr, &s.rules[i], s.arrayColumn) {
+ if matchEntries(*entriesPtr, &s.rules[i]) {
keep = true
break
}
@@ -618,10 +644,11 @@ func (s *Sampler) keepTrace(b *sdk.TraceBlock) bool {
if !keep && s.healthySampleRate > 0 && sampleFraction(b.TraceID) < s.healthySampleRate {
keep = true
}
- if entriesPtr != nil {
+ // Release only what a decode acquired. Keying this off decoded rather than
+ // off the pointers is what keeps the flag honest: it is the same flag both
+ // call sites above test, so a future third site cannot quietly skip it.
+ if decoded {
releaseEntries(entriesPtr)
- }
- if stableBufPtr != nil {
releaseStableBuf(stableBufPtr)
}
return keep
@@ -826,25 +853,36 @@ func releaseStableBuf(b *[]byte) {
// the address of a []string local would impose. releaseEntries takes the
// pointer by value too.
//
-// The stableBuf pointer is the caller-owned backing buffer for the decoded
-// entry strings. Each entry's bytes are appended to *stableBuf and then
-// aliased via convert.BytesToString: the string header allocation that
-// string(buf[idx:end]) would impose is skipped, but only because *stableBuf
-// is never reset — only appended to — across the lifetime of the strings.
-// releaseStableBuf is the caller's responsibility, called after the strings
-// are no longer needed. arrayEntries returns without touching stableBuf's
-// pool lifecycle; passing nil for stableBuf falls back to the safe-but-
-// allocating string() conversion, which the tests rely on for the no-column
-// path that never produces any entry.
-func arrayEntries(col *sdk.TagColumn, stableBuf *[]byte) (*[]string, error) {
+// The second return value is the stable buffer backing the decoded entry
+// strings, or nil when no row needed one. Only an ESCAPED row needs it: its
+// bytes are unescaped into a scratch copy, so the resulting strings cannot alias
+// either that scratch (it is reused per row) or the block (it still holds the
+// escaped form), and are instead appended to the stable buffer and aliased via
+// convert.BytesToString — skipping the string header allocation that
+// string(buf[idx:end]) would impose. The buffer is only ever appended to, never
+// reset, so earlier strings stay valid. An escape-free row aliases the block
+// directly and needs no buffer at all, which is the common case, so the buffer
+// is acquired lazily and a whole column of unescaped rows returns nil here.
+//
+// The caller MUST releaseStableBuf whatever is returned, and only after the
+// strings are no longer read — they alias it.
+//
+// errInvalidVarArray mirrors the error vararray.UnmarshalVarArray returns for an
+// entry with no terminating delimiter. The escape-free split path reports the
+// same condition, so it must read identically in logs whichever path decoded
+// the row.
+var errInvalidVarArray = errors.New("invalid variable array")
+
+func arrayEntries(col *sdk.TagColumn) (*[]string, *[]byte, error) {
if col == nil {
// Always return a non-nil pool element so the caller can dereference
// without a nil check. matchEntries on an empty slice returns false,
// which matches the pre-pool behavior of "no entries" → no match.
- return acquireEntries(), nil
+ return acquireEntries(), nil, nil
}
- // Size the copy buffer to the longest row up front so it is allocated exactly once
- // and never grows mid-loop. This needs no decode — only the raw byte lengths.
+ // Early exit for a column whose every row is empty: it can yield no entry, so
+ // the row loop below would do nothing. This needs no decode — only the raw
+ // byte lengths.
widest := 0
for _, raw := range col.Values {
if len(raw) > widest {
@@ -852,7 +890,7 @@ func arrayEntries(col *sdk.TagColumn, stableBuf *[]byte) (*[]string, error) {
}
}
if widest == 0 {
- return acquireEntries(), nil
+ return acquireEntries(), nil, nil
}
// Acquire the result slice from the pool. The backing array is reused across
// calls; the length is reset to zero so stale entries from the previous decode
@@ -863,19 +901,85 @@ func arrayEntries(col *sdk.TagColumn, stableBuf *[]byte) (*[]string, error) {
// references it. The error path here releases locally because the slice
// never escapes to keepTrace.
out := acquireEntries()
- // Acquire the per-row copy buffer from the pool. If its cap is smaller than
- // the longest row of this trace, append will allocate a fresh backing array;
- // the pool then sees a larger buffer on release and the steady-state cap
- // converges to the widest row the schema produces.
- bufPtr := acquireCopyBuf()
- defer releaseCopyBuf(bufPtr)
- buf := *bufPtr
- stable := *stableBuf
+ // The per-row copy buffer is acquired LAZILY, on the first row that actually
+ // carries an escape. A column whose rows are all escape-free never needs one,
+ // and the pool checkout/release is not free: releaseCopyBuf's Bounded.Put was
+ // 8.7% flat in the tag-rule CPU profile. If its cap is smaller than the row
+ // being copied, append allocates a fresh backing array; the pool then sees a
+ // larger buffer on release and the steady-state cap converges to the widest
+ // escaped row the schema produces.
+ // The stable buffer is acquired lazily too, and only an escaped row needs it.
+ var (
+ bufPtr *[]byte
+ buf []byte
+ stablePtr *[]byte
+ stable []byte
+ )
+ // failed distinguishes the two owners of the stable buffer: on success the
+ // CALLER releases it (the returned strings still alias it), on failure this
+ // function does. Either way the grown slice header has to be written back
+ // first, or the pool takes back the pre-append header and the growth is lost
+ // — so both the persist and the release live here rather than at each of the
+ // error sites, where the persist was previously missed.
+ failed := false
+ defer func() {
+ if bufPtr != nil {
+ // Persist any growth so the next caller sees the wider cap.
+ *bufPtr = buf
+ releaseCopyBuf(bufPtr)
+ }
+ if stablePtr != nil {
+ *stablePtr = stable
+ if failed {
+ releaseStableBuf(stablePtr)
+ }
+ }
+ }()
for row := range col.Values {
if col.Values[row] == nil {
continue
}
- // Decode a COPY of the row, never the engine's buffer. vararray.UnmarshalVarArray
+ // A row with no escape byte is decoded straight out of the block, with the
+ // entries aliasing it — no row copy, no stable-buffer append.
+ //
+ // Two facts make that safe, and BOTH are required:
+ //
+ // 1. No write. vararray.UnmarshalVarArray only shifts bytes when it
+ // consumes an escape; with none present it does not touch the buffer,
+ // so the corruption hazard described below cannot arise.
+ // 2. No recycling. assembleTraceBlock (banyand/trace/pipeline_chain.go)
+ // deep-copies every value into block-owned storage precisely "so the
+ // merge loop may recycle/overwrite those buffers while an abandoned
+ // Decide goroutine still reads the block". The bytes therefore outlive
+ // this call even on the timeout path, so aliasing them cannot observe
+ // recycled memory.
+ //
+ // The escape check is per row and must stay that way: it is a property of
+ // the DATA, not of the column or the schema.
+ if raw := col.Values[row]; bytes.IndexByte(raw, vararray.Escape) < 0 {
+ // The row is known escape-free, so entries are exactly the
+ // delimiter-separated runs and UnmarshalVarArray's own escape scan
+ // would be redundant work on every entry — it was 34.6% flat in the
+ // tag-rule profile. Split on the delimiter directly instead. This is
+ // the ONE place that may assume the encoding, and only because the
+ // check above proves no escape can be present: with an escape, a
+ // delimiter byte may be escaped and this loop would split mid-value.
+ idx := 0
+ for idx < len(raw) {
+ rel := bytes.IndexByte(raw[idx:], vararray.EntityDelimiter)
+ if rel < 0 {
+ // Same condition, and the same message, that
+ // UnmarshalVarArray reports for an unterminated entry.
+ releaseEntries(out)
+ failed = true
+ return nil, nil, fmt.Errorf("str array: %w", errInvalidVarArray)
+ }
+ *out = append(*out, convert.BytesToString(raw[idx:idx+rel]))
+ idx += rel + 1
+ }
+ continue
+ }
+ // Escaped row: decode a COPY, never the block's buffer. vararray.UnmarshalVarArray
// shifts bytes left past every escape in place, TraceBlock slices are documented
// read-only, and the engine hands the SAME TraceBatch to every link of a chain
// (sdk.applyChainLink). Decoding in place would therefore corrupt the value for
@@ -889,33 +993,30 @@ func arrayEntries(col *sdk.TagColumn, stableBuf *[]byte) (*[]string, error) {
// (which is only ever appended to, never reset, so prior bytes stay valid)
// and then alias stable via convert.BytesToString — skipping the per-entry
// string-header allocation that string(buf[idx:end]) would impose.
+ if bufPtr == nil {
+ bufPtr = acquireCopyBuf()
+ buf = *bufPtr
+ stablePtr = acquireStableBuf()
+ stable = *stablePtr
+ }
buf = append(buf[:0], col.Values[row]...)
idx := 0
for idx < len(buf) {
end, next, err := vararray.UnmarshalVarArray(buf, idx)
if err != nil {
releaseEntries(out)
- return nil, fmt.Errorf("str array: %w", err)
+ failed = true
+ return nil, nil, fmt.Errorf("str array: %w", err)
}
start := len(stable)
stable = append(stable, buf[idx:end]...)
- if stableBuf != nil {
- *out = append(*out, convert.BytesToString(stable[start:]))
- } else {
- // Defensive fallback: copy out of stable when the caller did not
- // provide a stable buffer. Tests pass nil for the no-column path,
- // which never reaches this branch.
- *out = append(*out, string(stable[start:]))
- }
+ *out = append(*out, convert.BytesToString(stable[start:]))
idx = next
}
}
- // Persist any growth of buf and stable into their pool pointers so the next
- // caller sees the wider cap. Without this the growth would be lost when the
- // local variables were reassigned only via append.
- *bufPtr = buf
- *stableBuf = stable
- return out, nil
+ // Both pointers stay nil when no row carried an escape, which is the common
+ // case; the deferred block above persists whatever growth did happen.
+ return out, stablePtr, nil
}
// errNoErrorColumn reports that the schema's error column is absent from the block.
@@ -969,11 +1070,8 @@ func (s *Sampler) hasErrorColumn(b *sdk.TraceBlock) (bool, error) {
// no "="). The comparison is exact, so a longer key such as "error_rate=0" never
// satisfies an exists rule on "error". keepErrors routes through here too, so the
// two cannot drift apart.
-func matchEntries(entries []string, r *rule, arrayColumn string) bool {
- prefix := ""
- if r.TagKey != arrayColumn {
- prefix = r.TagKey + "="
- }
+func matchEntries(entries []string, r *rule) bool {
+ prefix := r.prefix
for _, entry := range entries {
if r.Exists && entry == r.TagKey {
return true
diff --git a/plugins/skywalking/internal/tracesampler/sampler_bench_ext_test.go b/plugins/skywalking/internal/tracesampler/sampler_bench_ext_test.go
index 561783027..6b92d4cc4 100644
--- a/plugins/skywalking/internal/tracesampler/sampler_bench_ext_test.go
+++ b/plugins/skywalking/internal/tracesampler/sampler_bench_ext_test.go
@@ -20,9 +20,10 @@
// benchPlugins / benchEntries / benchBatch / mustSampler / runDecide / benchSink
// helpers are reused rather than re-declared.
//
-// Layer 1 only covers synthetic-data engine microbenchmarks. No production-ratio
-// claim, no calibrated fixture, no .so loading. See
-// docs/design/phase-9-sw-trace-sampler-bench.md §Layer 1.
+// These cover synthetic-data engine microbenchmarks only: no production-ratio
+// claim, no calibrated fixture, no .so loading. See the Phase 9 workstream in
+// docs/design/trace-pipeline-merge-optimization-plan.md, whose exit gate asks for
+// per-call-site attribution of the plugin-local costs these measure.
package tracesampler
import (
diff --git a/plugins/skywalking/internal/tracesampler/sampler_test.go b/plugins/skywalking/internal/tracesampler/sampler_test.go
index 6d82dd0c5..e89766061 100644
--- a/plugins/skywalking/internal/tracesampler/sampler_test.go
+++ b/plugins/skywalking/internal/tracesampler/sampler_test.go
@@ -25,6 +25,7 @@ import (
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
+ "github.com/apache/skywalking-banyandb/pkg/encoding/vararray"
"github.com/apache/skywalking-banyandb/pkg/pb/v1/valuetype"
"github.com/apache/skywalking-banyandb/pkg/pipeline/sdk"
"github.com/apache/skywalking-banyandb/pkg/pipeline/sdk/sdktest"
@@ -718,6 +719,155 @@ func TestDecide_HealthySampleRate(t *testing.T) {
assert.Equal(t, []bool{false}, verdict.Keep)
}
+// TestArrayEntries_LeavesBlockBytesIntact pins the precondition that lets
+// arrayEntries alias the block's own bytes instead of copying every row: the
+// decode must leave col.Values byte-identical.
+//
+// An escape-free row is decoded in place and the entries alias it, which is only
+// sound because UnmarshalVarArray does not write when there is no escape to
+// remove. An escaped row is still copied first, because that decode DOES write.
+// If either half regresses, the engine hands the SAME batch to every link of a
+// chain, so a later rule would read rewritten bytes and its verdict would depend
+// on its position in the chain — silent and data-dependent.
+//
+// The mixed column is deliberate: the choice is per row, not per column.
+func TestArrayEntries_LeavesBlockBytesIntact(t *testing.T) {
+ rows := [][]string{
+ {"db.type=PostgreSQL", "http.method=GET"}, // escape-free
+ {`db.statement=SELECT a|b`, `url=C:\tmp`}, // both bytes needing an escape
+ {"plain"}, // escape-free, single entry
+ }
+ col := &sdk.TagColumn{Name: "tags", ValueType: valuetype.ValueTypeStrArr}
+ for _, entries := range rows {
+ var encoded []byte
+ for _, e := range entries {
+ encoded = vararray.MarshalVarArray(encoded, []byte(e))
+ }
+ col.Values = append(col.Values, encoded)
+ }
+ before := make([][]byte, len(col.Values))
+ for i, v := range col.Values {
+ before[i] = append([]byte(nil), v...)
+ }
+
+ entriesPtr, stablePtr, err := arrayEntries(col)
+ require.NoError(t, err)
+ defer releaseEntries(entriesPtr)
+ defer releaseStableBuf(stablePtr)
+
+ assert.Equal(t, []string{
+ "db.type=PostgreSQL", "http.method=GET",
+ `db.statement=SELECT a|b`, `url=C:\tmp`,
+ "plain",
+ }, *entriesPtr, "every row's entries must decode, escaped or not")
+
+ for i := range col.Values {
+ assert.Equal(t, before[i], col.Values[i],
+ "row %d was rewritten; a later chain link would read corrupted bytes", i)
+ }
+}
+
+// TestArrayEntries_SplitPathMatchesCodec pins the one place the sampler assumes
+// the var-array encoding. The escape-free branch splits on the delimiter itself
+// instead of calling UnmarshalVarArray, to skip that function's redundant
+// per-entry escape scan. It must therefore agree with the codec exactly — same
+// entries, same error — on every row shape, or a rule's verdict would depend on
+// which branch decoded the row.
+func TestArrayEntries_SplitPathMatchesCodec(t *testing.T) {
+ rows := []struct {
+ name string
+ encoded []byte
+ }{
+ {"empty", nil},
+ {"single entry", []byte("abc|")},
+ {"two entries", []byte("a|b|")},
+ {"empty entry first", []byte("|a|")},
+ {"empty entry last", []byte("a||")},
+ {"all empty entries", []byte("|||")},
+ {"unterminated", []byte("abc")},
+ {"unterminated after a good entry", []byte("a|bc")},
+ }
+ for _, row := range rows {
+ t.Run(row.name, func(t *testing.T) {
+ // Reference: decode via the codec, which is what the escaped branch does.
+ var (
+ want []string
+ wantErr error
+ )
+ ref := append([]byte(nil), row.encoded...)
+ for idx := 0; idx < len(ref); {
+ end, next, err := vararray.UnmarshalVarArray(ref, idx)
+ if err != nil {
+ wantErr = err
+ break
+ }
+ want = append(want, string(ref[idx:end]))
+ idx = next
+ }
+
+ col := &sdk.TagColumn{
+ Name: "tags", ValueType: valuetype.ValueTypeStrArr,
+ Values: [][]byte{append([]byte(nil), row.encoded...)},
+ }
+ got, stablePtr, err := arrayEntries(col)
+
+ if wantErr != nil {
+ require.Error(t, err, "codec rejected this row, so the split path must too")
+ assert.Contains(t, err.Error(), wantErr.Error(), "both paths must report the same condition")
+ return
+ }
+ require.NoError(t, err)
+ defer releaseEntries(got)
+ defer releaseStableBuf(stablePtr)
+ // Compared by content: a row yielding nothing gives the codec a nil
+ // slice and arrayEntries a zero-length pooled one, which is the same
+ // value to every caller (matchEntries iterates the length).
+ if len(want) == 0 {
+ assert.Empty(t, *got)
+ return
+ }
+ assert.Equal(t, want, *got)
+ })
+ }
+}
+
+// TestArrayEntries_ErrorAfterGrownStableBuffer drives the error path in the state
+// that makes buffer bookkeeping easy to get wrong: an escaped row has already
+// appended to (and possibly reallocated) the stable buffer when a later row turns
+// out to be malformed, so the local slice header has diverged from the pooled one
+// and must be written back before the buffer is released.
+//
+// The assertions here only pin the observable contract — the error surfaces and
+// nothing panics or double-releases. That the grown header is persisted is
+// structural: the persist and the release both live in one deferred block, so an
+// error site cannot skip it.
+func TestArrayEntries_ErrorAfterGrownStableBuffer(t *testing.T) {
+ escaped := vararray.MarshalVarArray(nil, []byte(`db.statement=SELECT a|b`))
+ col := &sdk.TagColumn{
+ Name: "tags", ValueType: valuetype.ValueTypeStrArr,
+ Values: [][]byte{
+ escaped, // grows the stable buffer
+ []byte("unterminated"), // no trailing delimiter -> error
+ },
+ }
+ got, stablePtr, err := arrayEntries(col)
+ require.Error(t, err)
+ assert.Contains(t, err.Error(), "invalid variable array")
+ assert.Nil(t, got, "the entries slice is released on the error path")
+ assert.Nil(t, stablePtr, "the stable buffer is released here, not handed to the caller")
+
+ // The same shape must also fail open through the public path rather than drop.
+ s, newErr := New([]byte(`{"keepTagRules":[{"tagKey":"db.statement","exists":true}]}`), segmentSchema)
+ require.NoError(t, newErr)
+ batch := &sdk.TraceBatch{Traces: []sdk.TraceBlock{{
+ TraceID: "malformed-after-escaped",
+ Tags: []sdk.TagColumn{*col},
+ }}}
+ verdict, decideErr := s.Decide(batch)
+ require.NoError(t, decideErr)
+ assert.Equal(t, []bool{true}, verdict.Keep, "a decode error must fail open (keep), never drop")
+}
+
// TestDecide_FailOpenOnDecodeError proves a malformed tag value keeps the trace
// (fail open) rather than dropping it or erroring the whole batch: an is_error
// column whose raw int64 is not 8 bytes fails to decode, and keepErrors keeps
diff --git a/test/cases/schema/clamp.go b/test/cases/schema/clamp.go
index bdfa2e00b..5e94702a8 100644
--- a/test/cases/schema/clamp.go
+++ b/test/cases/schema/clamp.go
@@ -245,19 +245,35 @@ var _ = g.Describe("Schema time-range clamp", func() {
"write at T_data1 must return STATUS_SUCCEED")
g.By("Sanity-checking that querying group1 alone returns the datum (legacy unclamped baseline — pass modRevision=0)")
- // In distributed mode the write→query path is asynchronous: the
- // data-node ack returns when mustAddMemPart's applied channel
- // fires, but the query fan-out can still race the new memPart
- // becoming visible. Mirrors the deletion.go retry pattern.
- test.EventuallyConsistently(func() int {
+ // TWO distinct races have to settle before this query can succeed, and the
+ // budget has to cover both of them, not just the second:
+ //
+ // 1. Topology. AwaitRevision confirms the schema CACHE has the group; it
+ // does NOT confirm that the distributed query coordinator has resolved
+ // it, which is a separate path. Until it does, the query fails outright
+ // with "group ... not found" — even though the write to that same group
+ // has already been acked.
+ // 2. Visibility. The data-node ack fires when mustAddMemPart's applied
+ // channel does, but the query fan-out can still race the new memPart
+ // becoming visible.
+ //
+ // The probe carries the error instead of collapsing it to a sentinel. A
+ // swallowed error made a topology failure report as "Expected 0 to equal 1",
+ // which blames visibility and hides the real cause — the "group not found"
+ // only showed up by reading interleaved server logs in the CI output.
+ type baselineProbe struct {
+ Err string
+ Count int
+ }
+ test.EventuallyConsistently(func() baselineProbe {
baselineResp, baselineErr := queryMeasureRange(ctx, clients.MeasureWriteClient, group1, measureName,
tData1.Add(-time.Hour), time.Now().Add(time.Hour), 0)
if baselineErr != nil {
- return -1
+ return baselineProbe{Err: baselineErr.Error()}
}
- return len(baselineResp.GetDataPoints())
- }, 5*time.Second, 50*time.Millisecond).Should(gm.Equal(1),
- "baseline single-group query must return the written datum — otherwise is not falsifying anything")
+ return baselineProbe{Count: len(baselineResp.GetDataPoints())}
+ }, 10*time.Second, 200*time.Millisecond).Should(gm.Equal(baselineProbe{Count: 1}),
+ "baseline single-group query must return the written datum — otherwise the clamp assertion below is not falsifying anything")
g.By("Creating newer group2 and measure (CreatedAt2 > T_data1)")
_, createGroup2Err := clients.GroupClient.Create(ctx, &databasev1.GroupRegistryServiceCreateRequest{