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/07 05:33:58 UTC
(skywalking-banyandb) branch perf/trace-merge-2c4g-baseline created (now d116c4431)
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a change to branch perf/trace-merge-2c4g-baseline
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
at d116c4431 docs(trace): record adaptive merge acceptance
This branch includes the following new commits:
new 656225f94 feat(trace): adapt merge sampling batch limits
new d116c4431 docs(trace): record adaptive merge acceptance
The 2 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) 02/02: docs(trace): record adaptive merge acceptance
Posted by ha...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a commit to branch perf/trace-merge-2c4g-baseline
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
commit d116c4431310715fcf37d9498aa4e33fcd6c1005
Author: Hongtao Gao <ha...@gmail.com>
AuthorDate: Fri Aug 7 05:33:24 2026 +0000
docs(trace): record adaptive merge acceptance
---
docs/design/trace-pipeline-merge-optimization-plan.md | 19 ++++++++++++++++++-
1 file changed, 18 insertions(+), 1 deletion(-)
diff --git a/docs/design/trace-pipeline-merge-optimization-plan.md b/docs/design/trace-pipeline-merge-optimization-plan.md
index 61aa1fd3e..288220463 100644
--- a/docs/design/trace-pipeline-merge-optimization-plan.md
+++ b/docs/design/trace-pipeline-merge-optimization-plan.md
@@ -190,7 +190,7 @@ causes complete-trace batch boundaries. Both byte and trace-count limits remain
- [x] Every sampled merge reports peak charged staging bytes, batch bytes and trace count, flush reason, and peak
concurrent staged bytes.
-- [ ] Cold first-merge cost and subsequent same-process reuse are reported separately without changing the budget.
+- [x] Cold first-merge cost and subsequent same-process reuse are reported separately without changing the budget.
- [ ] The serialized 24-hour workload validates naturally varying merge sizes under the canonical two-CPU, 4 GiB
container and its resource-derived 256 MiB budget.
- [ ] At least one naturally budget-limited merge, or explicit evidence that the production-shaped workload never
@@ -225,6 +225,23 @@ core and secondary-index ledger. Compared with the preceding five-run retain-all
peak heap from 167.1 MiB to 106.8 MiB and peak RSS from 199.7 MiB to 151.6 MiB. It is evidence that adaptive batching
controls the live set, not a replacement for the required alternating acceptance series.
+The balanced DR/RD acceptance series subsequently ran five fresh Docker processes per mode against the same selection.
+All ten runs preserved 147,126 rows, all three logical ledgers, the mature-only selection, and complete-trace decisions.
+The retain-all runs consistently made five sampler calls and peaked at 27.25 MiB of charged staged memory. Their median
+was 266.32 MiB allocated, 3,200,439 allocations, 109.28 MiB peak heap, and 158.93 MiB peak RSS. Relative to the previous
+one-call retain-all medians, this is a 12.4% allocated-byte reduction, 2.3% allocation-count reduction, 34.6% peak-heap
+reduction, and 20.4% peak-RSS reduction. Correctness and live-heap control pass, but the allocated-byte,
+allocation-count, and RSS targets remain open. The measured wall and CPU deltas remain non-blocking and are not credited
+as a pipeline speedup.
+
+A separate five-merge retain-all diagnostic used five identical seed clones sequentially in one two-CPU, 4 GiB process,
+without changing the 256 MiB hard ceiling or the metadata-derived preferred limit. The cold merge allocated 263.10 MiB.
+The four reuse merges allocated 225.93-238.53 MiB while CPU time stayed between 6.56 and 6.64 seconds. Reused arena
+capacities are included in the authoritative runtime charge, so the warm merges used six complete-trace sampler calls
+instead of the cold merge's five. Peak staged memory remained 27.25 MiB and every ledger remained correct. Absolute RSS
+in this diagnostic includes the history of earlier merges in the same process and must not be interpreted as an isolated
+per-merge RSS delta.
+
---
### Phase 5 — Decision Path Overhead
(skywalking-banyandb) 01/02: feat(trace): adapt merge sampling batch limits
Posted by ha...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a commit to branch perf/trace-merge-2c4g-baseline
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
commit 656225f94ae248e5b1f4ff4268e3d50492baeda4
Author: Hongtao Gao <ha...@gmail.com>
AuthorDate: Fri Aug 7 05:00:03 2026 +0000
feat(trace): adapt merge sampling batch limits
---
banyand/cmd/trace-merge-benchmark/main.go | 9 +
banyand/internal/benchmark/tracebaseline/html.go | 4 +-
.../internal/benchmark/tracebaseline/html_test.go | 5 +-
banyand/internal/benchmark/tracebaseline/server.go | 39 +++-
banyand/trace/finalizer.go | 26 +--
.../trace/fragment_guard_implementation_test.go | 4 +
banyand/trace/merge_benchmark_observer.go | 207 ++++++++++++++------
banyand/trace/merge_benchmark_observer_test.go | 41 ++++
banyand/trace/merger.go | 209 ++++++++++++++++++---
banyand/trace/pipeline_chain_filter_test.go | 85 +++++++++
banyand/trace/pipeline_chain_test.go | 8 +-
.../trace-pipeline-merge-optimization-plan.md | 61 ++++--
.../trace-pipeline-merge-performance-test.md | 12 +-
scripts/run-trace-merge-baseline.sh | 42 ++++-
scripts/run-trace-merge-controlled-seed.sh | 11 +-
15 files changed, 620 insertions(+), 143 deletions(-)
diff --git a/banyand/cmd/trace-merge-benchmark/main.go b/banyand/cmd/trace-merge-benchmark/main.go
index 83170e24f..0e036703a 100644
--- a/banyand/cmd/trace-merge-benchmark/main.go
+++ b/banyand/cmd/trace-merge-benchmark/main.go
@@ -24,8 +24,10 @@ import (
"os"
"os/signal"
"syscall"
+ "time"
"github.com/apache/skywalking-banyandb/banyand/internal/benchmark/tracebaseline"
+ "github.com/apache/skywalking-banyandb/pkg/timestamp"
)
func main() {
@@ -159,6 +161,7 @@ func render(arguments []string) {
func serve(ctx context.Context, arguments []string) {
flags := flag.NewFlagSet("serve", flag.ExitOnError)
var options tracebaseline.ServerOptions
+ var segmentMinTimeNanos, segmentMaxTimeNanos int64
flags.StringVar(&options.Root, "root", "", "data-node shard root")
flags.StringVar(&options.SocketPath, "socket", "", "Unix control socket")
flags.StringVar(&options.OutputPath, "output", "", "run report JSON path")
@@ -182,7 +185,13 @@ func serve(ctx context.Context, arguments []string) {
flags.StringVar(&options.ExecutionIdentity.CloneMethod, "clone-method", "", "clone method recorded in the environment envelope (e.g. os.CopyFS, hardlink)")
flags.StringVar(&options.ExecutionIdentity.BinarySHA256, "binary-sha256", "", "measured data-node binary checksum")
flags.StringVar(&options.ExecutionIdentity.PluginSHA256, "plugin-sha256", "", "plugin .so checksum for retain-all pipeline mode")
+ flags.StringVar(&options.PluginPath, "plugin", "", "native sampler plugin .so path")
+ flags.Int64Var(&segmentMinTimeNanos, "segment-min-time-nanos", 0, "inclusive minimum fixture timestamp for sampler coverage")
+ flags.Int64Var(&segmentMaxTimeNanos, "segment-max-time-nanos", 0, "inclusive maximum fixture timestamp for sampler coverage")
_ = flags.Parse(arguments)
+ if segmentMinTimeNanos != 0 || segmentMaxTimeNanos != 0 {
+ options.SegmentTimeRange = timestamp.NewInclusiveTimeRange(time.Unix(0, segmentMinTimeNanos), time.Unix(0, segmentMaxTimeNanos))
+ }
options.ExpectedLedger = map[string]string{
tracebaseline.LedgerCore: expectedCoreLedger, tracebaseline.LedgerLatency: expectedLatencyLedger,
tracebaseline.LedgerStartTime: expectedStartTimeLedger,
diff --git a/banyand/internal/benchmark/tracebaseline/html.go b/banyand/internal/benchmark/tracebaseline/html.go
index 802e68b37..2159c1e86 100644
--- a/banyand/internal/benchmark/tracebaseline/html.go
+++ b/banyand/internal/benchmark/tracebaseline/html.go
@@ -55,7 +55,7 @@ const baselineHTML = `<!doctype html>
<article class="card span4" data-index="04"><div class="section-title"><h2>Selection states</h2><small>mixed selections appear in both sets</small></div><svg id="class-diagram" viewBox="0 0 400 330" aria-label="overlapping hot and mature merge diagram"></svg></article>
<article class="card span12" data-index="05"><div class="section-title"><h2>Inventory waveform</h2><small>parts after each serialized merge-idle barrier</small></div><svg id="backlog-diagram" viewBox="0 0 1200 300" aria-label="part inventory area chart"></svg></article>
<article class="card span7" data-index="06"><div class="section-title"><h2>Controlled pipeline stability</h2><small>alternating disabled / retain-all mature merges</small></div><svg id="controlled-variance-diagram" viewBox="0 0 760 300" aria-label="controlled pipeline wall and CPU variance"></svg><div class="legend"><span><i class="dot" style="background:var(--cyan)"></i>disabled</span><span><i class="dot" style="background:var(--amber)"></i>retain-all</span></div></article>
-<article class="card span5" data-index="07"><div class="section-title"><h2>Controlled evidence</h2><small>five processes per variant</small></div><table><thead><tr><th>RUN</th><th>MODE</th><th>WALL</th><th>CPU</th><th>STAGE CAP</th><th>MAX TRACES</th><th>RESULT</th></tr></thead><tbody id="controlled-run-table"></tbody></table></article>
+<article class="card span5" data-index="07"><div class="section-title"><h2>Controlled evidence</h2><small>hard ceiling → metadata plan → measured staging</small></div><table><thead><tr><th>RUN</th><th>MODE</th><th>WALL</th><th>CPU</th><th>HARD CAP</th><th>BATCH LIMIT</th><th>ESTIMATE / PEAK</th><th>CALLS</th><th>RESULT</th></tr></thead><tbody id="controlled-run-table"></tbody></table></article>
<article class="card span12" data-index="08"><div class="section-title"><h2>Correctness circuit</h2><small>results are rejected when a required gate opens</small></div><div class="gate-grid" id="gates"></div></article>
<article class="card span12" data-index="09"><div class="section-title"><h2>Measured runs</h2><small>absolute one-shard values</small></div><table><thead><tr><th>RUN</th><th>MODE</th><th>PRIMARY</th><th>COOLDOWN</th><th>CPU</th><th>PEAK RSS</th><th>READ / WRITE</th><th>HOT-CONTAINING / MATURE-CONTAINING</th><th>LEDGER</th></tr></thead><tbody id="run-table"></tbody></table><div class="legend">Mixed selections may appear in both hot and mature counts. Sampling execution is reported indepen [...]
</div><footer class="foot"><span id="identity"></span><span>Standalone HTML · source data embedded below</span></footer>
@@ -75,7 +75,7 @@ function line(x1,y1,x2,y2,c='gridline'){return '<line class="'+c+'" x1="'+x1+'"
{document.getElementById('class-diagram').innerHTML='<circle cx="145" cy="145" r="105" fill="#60d8e822" stroke="#60d8e8" stroke-width="3"/><circle cx="255" cy="145" r="105" fill="#9cff5722" stroke="#9cff57" stroke-width="3"/>'+text(105,135,totalHot,'value','middle')+text(105,153,'HOT','label','middle')+text(295,135,totalMature,'value','middle')+text(295,153,'MATURE','label','middle')+text(200,135,totalMixed,'value','middle')+text(200,153,'MIXED','label','middle')+text(200,295,'counts des [...]
{const st=(tp[0]||runs[0]||{}).status||[],mx=Math.max(1,...st.map(x=>x.coreParts||0));let pts=st.map((p,i)=>(45+i/Math.max(1,st.length-1)*1110)+','+(250-(p.coreParts||0)/mx*205)).join(' '),h='<polygon class="area" points="45,250 '+pts+' 1155,250"/>';for(let i=0;i<5;i++){let y=250-i*50;h+=line(45,y,1155,y)+text(35,y+3,fmt(mx*i/4,0),'label','end')}h+=text(45,280,'00:00 LOGICAL','label')+text(1155,280,'24:00 + DRAIN','label','end');document.getElementById('backlog-diagram').innerHTML=h}
{const svg=document.getElementById('controlled-variance-diagram'),metrics=[['WALL s',controlledRuns.map(r=>ns(r.event?.resources?.elapsedNanos||0))],['CPU s',controlledRuns.map(r=>ns(r.event?.resources?.cpuNanos||0))]],colors=controlledRuns.map(r=>r.pipelineMode==='retain-all'?'#ffbf45':'#60d8e8');let h='';metrics.forEach((metric,row)=>{const y=85+row*125,values=metric[1],mx=Math.max(1,...values);h+=text(20,y,metric[0],'value')+line(125,y,720,y,'axis');values.forEach((value,index)=>{cons [...]
-document.getElementById('controlled-run-table').innerHTML=controlledRuns.map(run=>'<tr><td>'+run.runID+'</td><td>'+run.pipelineMode+'</td><td>'+fmt(ns(run.event?.resources?.elapsedNanos||0),3)+'s</td><td>'+fmt(ns(run.event?.resources?.cpuNanos||0),3)+'s</td><td>'+fmt(mb(run.stagingLimits?.stageBytes||0),1)+' MiB</td><td>'+fmt(run.stagingLimits?.maxTraceCount||0,0)+'</td><td style="color:'+(run.correct?'var(--green)':'var(--red)')+'">'+(run.correct?'PASS':'FAIL')+'</td></tr>').join('');
+document.getElementById('controlled-run-table').innerHTML=controlledRuns.map(run=>'<tr><td>'+run.runID+'</td><td>'+run.pipelineMode+'</td><td>'+fmt(ns(run.event?.resources?.elapsedNanos||0),3)+'s</td><td>'+fmt(ns(run.event?.resources?.cpuNanos||0),3)+'s</td><td>'+fmt(mb(run.event?.stagingHardLimit||run.stagingLimits?.stageBytes||0),1)+' MiB</td><td>'+fmt(mb(run.event?.decisionBatchLimit||0),1)+' MiB</td><td>'+fmt(mb(run.event?.estimatedStagingBytes||0),1)+' / '+fmt(mb(run.event?.peakStag [...]
const gates=readiness.gates||[],overallReady=readiness.ready===true;document.getElementById('stamp-result').textContent=overallReady?'VALID':'HOLD';document.getElementById('gates').innerHTML=gates.map(g=>'<div class="gate '+(g.passed?'pass':'fail')+'"><i></i><b>'+g.name+'</b><span>'+g.detail+'</span></div>').join('');
document.getElementById('run-table').innerHTML=runs.map(r=>{let cpu=ns((r.primary?.end?.cpuNanos||0)-(r.primary?.start?.cpuNanos||0)),rw=fmt(mb((r.primary?.end?.readBytes||0)-(r.primary?.start?.readBytes||0)),1)+' / '+fmt(mb((r.primary?.end?.writeBytes||0)-(r.primary?.start?.writeBytes||0)),1);return '<tr><td>'+r.runID+'</td><td>'+r.mode+'</td><td>'+fmt(ns(r.primary?.wallNanos),2)+'s</td><td>'+fmt(ns(r.cooldown?.wallNanos),3)+'s</td><td>'+fmt(cpu,2)+'s</td><td>'+fmt(mb(r.primary?.end?.rs [...]
</script></main></body></html>`
diff --git a/banyand/internal/benchmark/tracebaseline/html_test.go b/banyand/internal/benchmark/tracebaseline/html_test.go
index 2b71634c0..18d2ed6d7 100644
--- a/banyand/internal/benchmark/tracebaseline/html_test.go
+++ b/banyand/internal/benchmark/tracebaseline/html_test.go
@@ -46,8 +46,9 @@ func TestRenderHTMLUsesDiagramsAsPrimaryReportSurface(t *testing.T) {
require.Contains(t, html, "id=\"flow-diagram\"")
require.Contains(t, html, "id=\"controlled-variance-diagram\"")
require.Contains(t, html, "id=\"controlled-run-table\"")
- require.Contains(t, html, "STAGE CAP")
- require.Contains(t, html, "MAX TRACES")
+ require.Contains(t, html, "HARD CAP")
+ require.Contains(t, html, "BATCH LIMIT")
+ require.Contains(t, html, "ESTIMATE / PEAK")
require.Contains(t, html, "disabledEnabledAlternating")
require.Contains(t, html, "SAME TEST BOUNDARY")
require.Contains(t, html, "CORRECT OUTPUT")
diff --git a/banyand/internal/benchmark/tracebaseline/server.go b/banyand/internal/benchmark/tracebaseline/server.go
index f121fef64..b20f207cd 100644
--- a/banyand/internal/benchmark/tracebaseline/server.go
+++ b/banyand/internal/benchmark/tracebaseline/server.go
@@ -36,6 +36,8 @@ import (
"github.com/apache/skywalking-banyandb/banyand/internal/benchmark/controller"
"github.com/apache/skywalking-banyandb/banyand/internal/benchmark/tracefixture"
storagetrace "github.com/apache/skywalking-banyandb/banyand/trace"
+ "github.com/apache/skywalking-banyandb/pkg/pipeline/sdk"
+ "github.com/apache/skywalking-banyandb/pkg/timestamp"
)
// ServerOptions configures the measured data-node process.
@@ -59,6 +61,8 @@ type ServerOptions struct {
Mode string
OutputPath string
SocketPath string
+ PluginPath string
+ SegmentTimeRange timestamp.TimeRange
Acceleration float64
ExpectedRows uint64
MaxInputPartID uint64
@@ -92,13 +96,21 @@ func Serve(ctx context.Context, options ServerOptions) (serveResultErr error) {
if eventErr != nil {
return fmt.Errorf("cannot create merge event file: %w", eventErr)
}
+ sampler, samplerErr := openServerSampler(options)
+ if samplerErr != nil {
+ return errors.Join(samplerErr, eventFile.Close())
+ }
receiver, receiverErr := storagetrace.NewBenchmarkMergeReceiver( //nolint:contextcheck // The storage constructor has no context parameter.
options.Root, storagetrace.BenchmarkMergeReceiverOptions{
LogicalNow: time.Unix(0, 1), MergeGrace: 2 * time.Hour, EventWriter: eventFile, MaxInputPartID: options.MaxInputPartID,
- Attribution: options.Attribution,
+ Attribution: options.Attribution, Sampler: sampler, SegmentTimeRange: options.SegmentTimeRange,
})
if receiverErr != nil {
- return errors.Join(fmt.Errorf("cannot open measured merge receiver: %w", receiverErr), eventFile.Close())
+ var samplerCloseErr error
+ if sampler != nil {
+ samplerCloseErr = sampler.Close()
+ }
+ return errors.Join(fmt.Errorf("cannot open measured merge receiver: %w", receiverErr), samplerCloseErr, eventFile.Close())
}
server := &benchmarkServer{receiver: receiver, options: options}
defer func() {
@@ -146,6 +158,29 @@ func Serve(ctx context.Context, options ServerOptions) (serveResultErr error) {
return nil
}
+func openServerSampler(options ServerOptions) (sdk.Sampler, error) {
+ if options.PluginPath == "" {
+ return nil, nil
+ }
+ if options.SegmentTimeRange.Start.IsZero() || options.SegmentTimeRange.End.IsZero() ||
+ !options.SegmentTimeRange.Start.Before(options.SegmentTimeRange.End) {
+ return nil, fmt.Errorf("valid segment time range is required when the merge benchmark plugin is enabled")
+ }
+ sampler, openErr := sdk.OpenSampler(options.PluginPath, "NewSampler", nil)
+ if openErr != nil {
+ return nil, fmt.Errorf("cannot load merge benchmark sampler: %w", openErr)
+ }
+ pluginSHA256, shaErr := fileSHA256(options.PluginPath)
+ if shaErr != nil {
+ return nil, errors.Join(shaErr, sampler.Close())
+ }
+ if options.ExecutionIdentity.PluginSHA256 != "" && options.ExecutionIdentity.PluginSHA256 != pluginSHA256 {
+ return nil, errors.Join(fmt.Errorf("merge benchmark plugin checksum %s does not match expected %s",
+ pluginSHA256, options.ExecutionIdentity.PluginSHA256), sampler.Close())
+ }
+ return sampler, nil
+}
+
func (bs *benchmarkServer) health(responseWriter http.ResponseWriter, _ *http.Request) {
responseWriter.WriteHeader(http.StatusNoContent)
}
diff --git a/banyand/trace/finalizer.go b/banyand/trace/finalizer.go
index 343489a2a..de6910eb2 100644
--- a/banyand/trace/finalizer.go
+++ b/banyand/trace/finalizer.go
@@ -139,8 +139,9 @@ func (tst *tsTable) runFinalizeRound(samplers []sdk.Sampler, graceNs int64) (boo
return false, nil
}
- stageBudget := resolveStageBudget(tst.option)
- guard := tst.newTraceFragmentGuardSession(parts, guardGrace, stageBudget)
+ stagingHardLimit := resolveStageBudget(tst.option)
+ stagingPlan := planAdaptiveDecisionBatch(parts, stagingHardLimit)
+ guard := tst.newTraceFragmentGuardSession(parts, guardGrace, stagingHardLimit)
if guard == nil {
tst.incPipelineGuardBypassed()
return false, nil
@@ -150,15 +151,18 @@ func (tst *tsTable) runFinalizeRound(samplers []sdk.Sampler, graceNs int64) (boo
// fragment gap. The chain fails open on any Decide error.
chain := newMergeChain(tst.group, "", samplers, tst.option.decideTimeoutCircuitBreak)
filter := &mergeFilter{
- chain: chain,
- guard: guard,
- ctx: tst.loopCloser.Ctx(),
- owner: tst,
- timeout: tst.option.decideTimeout,
- stageBudget: stageBudget,
- traceBudget: resolveTraceBudget(tst.option),
- maxTraceCount: maxStagedTraceCountFromBudget(stageBudget),
- forceSlow: projectionRequiresSlowPath(chain.projection),
+ chain: chain,
+ guard: guard,
+ ctx: tst.loopCloser.Ctx(),
+ owner: tst,
+ timeout: tst.option.decideTimeout,
+ stagingHardLimit: stagingHardLimit,
+ decisionBatchLimit: stagingPlan.BatchLimit,
+ estimatedStagingBytes: stagingPlan.EstimatedBytes,
+ plannedStagingBatches: stagingPlan.PlannedBatches,
+ traceBudget: resolveTraceBudget(tst.option),
+ maxTraceCount: maxStagedTraceCountFromBudget(stagingPlan.BatchLimit),
+ forceSlow: projectionRequiresSlowPath(chain.projection),
}
merged := make(map[uint64]struct{}, len(parts))
diff --git a/banyand/trace/fragment_guard_implementation_test.go b/banyand/trace/fragment_guard_implementation_test.go
index a26cad225..f66d2ecea 100644
--- a/banyand/trace/fragment_guard_implementation_test.go
+++ b/banyand/trace/fragment_guard_implementation_test.go
@@ -709,6 +709,10 @@ func TestRuntimeTraceFragmentGuardUsesFragmentGapForSegmentInterior(t *testing.T
filter := tst.buildHotMergeFilter(parts)
require.NotNil(t, filter, "the 2h maturity grace must not be reused as the 1ms boundary expansion")
require.NotNil(t, filter.guard)
+ require.Equal(t, resolveStageBudget(tst.option), filter.stagingHardLimit)
+ require.Positive(t, filter.decisionBatchLimit)
+ require.LessOrEqual(t, filter.decisionBatchLimit, filter.stagingHardLimit)
+ require.Equal(t, maxStagedTraceCountFromBudget(filter.decisionBatchLimit), filter.maxTraceCount)
filter.guard.Close()
tst.option.maxTraceFragmentGap = 2 * time.Second
diff --git a/banyand/trace/merge_benchmark_observer.go b/banyand/trace/merge_benchmark_observer.go
index 03c8efe4a..88c3ce472 100644
--- a/banyand/trace/merge_benchmark_observer.go
+++ b/banyand/trace/merge_benchmark_observer.go
@@ -46,6 +46,8 @@ const (
type mergeSamplingReason string
+type mergeStagingFlushReason string
+
const (
mergeReasonPipelineDisabled mergeSamplingReason = "pipeline_disabled"
mergeReasonNoSampler mergeSamplingReason = "no_sampler"
@@ -59,6 +61,14 @@ const (
mergeReasonOther mergeSamplingReason = "other"
)
+const (
+ mergeStagingFlushEndOfMerge mergeStagingFlushReason = "end_of_merge"
+ mergeStagingFlushByteLimit mergeStagingFlushReason = "byte_limit"
+ mergeStagingFlushTraceLimit mergeStagingFlushReason = "trace_limit"
+ mergeStagingFlushByteAndTraceLimit mergeStagingFlushReason = "byte_and_trace_limit"
+ mergeStagingFlushOversizedTrace mergeStagingFlushReason = "oversized_trace"
+)
+
type mergeBenchmarkPhase string
const (
@@ -74,10 +84,14 @@ type (
BenchmarkMergeReason = mergeSamplingReason
// BenchmarkMergePhase identifies the benchmark phase in which a merge began.
BenchmarkMergePhase = mergeBenchmarkPhase
+ // BenchmarkMergeStagingFlushReason identifies why a complete-trace staging batch was flushed.
+ BenchmarkMergeStagingFlushReason = mergeStagingFlushReason
// BenchmarkMergeChild records a secondary-index merge linked to its core merge.
BenchmarkMergeChild = mergeBenchmarkChild
// BenchmarkMergeResources records process resource deltas attributed to a merge.
BenchmarkMergeResources = mergeBenchmarkResources
+ // BenchmarkMergeStagingBatch records one complete-trace staging flush.
+ BenchmarkMergeStagingBatch = mergeBenchmarkStagingBatch
// BenchmarkMergeEvent records one core merge and its secondary-index children.
BenchmarkMergeEvent = mergeBenchmarkEvent
// BenchmarkMergeAggregate groups merge measurements by phase, sampling state, reason, and lane.
@@ -99,6 +113,16 @@ const (
BenchmarkMergeSamplingEnabledNoEvaluation BenchmarkMergeSampling = mergeSamplingEnabledNoEvaluation
// BenchmarkMergeSamplingNotExecuted means sampling was ineligible for the merge.
BenchmarkMergeSamplingNotExecuted BenchmarkMergeSampling = mergeSamplingNotExecuted
+ // BenchmarkMergeStagingFlushEndOfMerge means the final complete-trace batch was flushed when input ended.
+ BenchmarkMergeStagingFlushEndOfMerge BenchmarkMergeStagingFlushReason = mergeStagingFlushEndOfMerge
+ // BenchmarkMergeStagingFlushByteLimit means the charged staging-byte limit triggered the flush.
+ BenchmarkMergeStagingFlushByteLimit BenchmarkMergeStagingFlushReason = mergeStagingFlushByteLimit
+ // BenchmarkMergeStagingFlushTraceLimit means the logical trace-count limit triggered the flush.
+ BenchmarkMergeStagingFlushTraceLimit BenchmarkMergeStagingFlushReason = mergeStagingFlushTraceLimit
+ // BenchmarkMergeStagingFlushByteAndTraceLimit means both staging limits were reached at the flush boundary.
+ BenchmarkMergeStagingFlushByteAndTraceLimit BenchmarkMergeStagingFlushReason = mergeStagingFlushByteAndTraceLimit
+ // BenchmarkMergeStagingFlushOversizedTrace means an oversized trace forced the preceding complete batch to flush.
+ BenchmarkMergeStagingFlushOversizedTrace BenchmarkMergeStagingFlushReason = mergeStagingFlushOversizedTrace
// BenchmarkMergeReasonPipelineDisabled means the native pipeline was disabled.
BenchmarkMergeReasonPipelineDisabled BenchmarkMergeReason = mergeReasonPipelineDisabled
// BenchmarkMergeReasonNoSampler means the group had no sampler.
@@ -165,44 +189,58 @@ type mergeBenchmarkResources struct {
Overlapped bool `json:"overlapped"`
}
+type mergeBenchmarkStagingBatch struct {
+ Reason mergeStagingFlushReason `json:"reason"`
+ Bytes uint64 `json:"bytes"`
+ Traces uint64 `json:"traces"`
+}
+
type mergeBenchmarkEvent struct {
- Sampling mergeSamplingClassification `json:"sampling"`
- Reason mergeSamplingReason `json:"reason,omitempty"`
- InitialReason mergeSamplingReason `json:"initialReason,omitempty"`
- Phase mergeBenchmarkPhase `json:"phase"`
- Type string `json:"type"`
- Lane string `json:"lane"`
- Error string `json:"error,omitempty"`
- RecordingError string `json:"recordingError,omitempty"`
- SelectionSHA256 string `json:"selectionSHA256"`
- InputPartIDs []uint64 `json:"inputPartIDs"`
- Children []mergeBenchmarkChild `json:"children,omitempty"`
- Resources mergeBenchmarkResources `json:"resources"`
- QueueNanos int64 `json:"queueNanos"`
- Sequence uint64 `json:"sequence"`
- OutputPartID uint64 `json:"outputPartID,omitempty"`
- InputBytes uint64 `json:"inputBytes"`
- OutputBytes uint64 `json:"outputBytes"`
- InputRows uint64 `json:"inputRows"`
- OutputRows uint64 `json:"outputRows"`
- HotInputRows uint64 `json:"hotInputRows"`
- MatureInputRows uint64 `json:"matureInputRows"`
- MinTimestamp int64 `json:"minTimestamp"`
- MaxTimestamp int64 `json:"maxTimestamp"`
- LogicalNow int64 `json:"logicalNow"`
- MaturityFrontier int64 `json:"maturityFrontier"`
- PluginCalls uint64 `json:"pluginCalls"`
- TracesEvaluated uint64 `json:"tracesEvaluated"`
- TracesRetained uint64 `json:"tracesRetained"`
- TracesDropped uint64 `json:"tracesDropped"`
- OversizedTraces uint64 `json:"oversizedTraces"`
- HotInputParts uint32 `json:"hotInputParts"`
- MatureInputParts uint32 `json:"matureInputParts"`
- InputMinDepth uint32 `json:"inputMinDepth"`
- InputMaxDepth uint32 `json:"inputMaxDepth"`
- OutputDepth uint32 `json:"outputDepth"`
- Version uint32 `json:"version"`
- LosslessRetry bool `json:"losslessRetry"`
+ Error string `json:"error,omitempty"`
+ Reason mergeSamplingReason `json:"reason,omitempty"`
+ InitialReason mergeSamplingReason `json:"initialReason,omitempty"`
+ Phase mergeBenchmarkPhase `json:"phase"`
+ Type string `json:"type"`
+ Lane string `json:"lane"`
+ Sampling mergeSamplingClassification `json:"sampling"`
+ RecordingError string `json:"recordingError,omitempty"`
+ SelectionSHA256 string `json:"selectionSHA256"`
+ InputPartIDs []uint64 `json:"inputPartIDs"`
+ Children []mergeBenchmarkChild `json:"children,omitempty"`
+ StagingBatches []mergeBenchmarkStagingBatch `json:"stagingBatches,omitempty"`
+ Resources mergeBenchmarkResources `json:"resources"`
+ OutputRows uint64 `json:"outputRows"`
+ TracesEvaluated uint64 `json:"tracesEvaluated"`
+ InputBytes uint64 `json:"inputBytes"`
+ OutputBytes uint64 `json:"outputBytes"`
+ InputRows uint64 `json:"inputRows"`
+ Sequence uint64 `json:"sequence"`
+ HotInputRows uint64 `json:"hotInputRows"`
+ MatureInputRows uint64 `json:"matureInputRows"`
+ MinTimestamp int64 `json:"minTimestamp"`
+ MaxTimestamp int64 `json:"maxTimestamp"`
+ LogicalNow int64 `json:"logicalNow"`
+ MaturityFrontier int64 `json:"maturityFrontier"`
+ PluginCalls uint64 `json:"pluginCalls"`
+ OutputPartID uint64 `json:"outputPartID,omitempty"`
+ TracesRetained uint64 `json:"tracesRetained"`
+ TracesDropped uint64 `json:"tracesDropped"`
+ OversizedTraces uint64 `json:"oversizedTraces"`
+ EstimatedStagingBytes uint64 `json:"estimatedStagingBytes"`
+ StagingHardLimit uint64 `json:"stagingHardLimit"`
+ DecisionBatchLimit uint64 `json:"decisionBatchLimit"`
+ PlannedStagingBatches uint64 `json:"plannedStagingBatches"`
+ ChargedStagingBytes uint64 `json:"chargedStagingBytes"`
+ PeakStagedBytes uint64 `json:"peakStagedBytes"`
+ QueueNanos int64 `json:"queueNanos"`
+ HotInputParts uint32 `json:"hotInputParts"`
+ MatureInputParts uint32 `json:"matureInputParts"`
+ InputMinDepth uint32 `json:"inputMinDepth"`
+ InputMaxDepth uint32 `json:"inputMaxDepth"`
+ OutputDepth uint32 `json:"outputDepth"`
+ Version uint32 `json:"version"`
+ DecisionMaxTraceCount int `json:"decisionMaxTraceCount"`
+ LosslessRetry bool `json:"losslessRetry"`
}
type mergeBenchmarkAggregate struct {
@@ -230,9 +268,10 @@ type mergeBenchmarkAggregate struct {
}
type mergeBenchmarkSnapshot struct {
- Error string `json:"error,omitempty"`
- Events []mergeBenchmarkEvent `json:"events"`
- Aggregates []mergeBenchmarkAggregate `json:"aggregates"`
+ Error string `json:"error,omitempty"`
+ Events []mergeBenchmarkEvent `json:"events"`
+ Aggregates []mergeBenchmarkAggregate `json:"aggregates"`
+ PeakConcurrentStagedBytes uint64 `json:"peakConcurrentStagedBytes"`
}
type mergeBenchmarkAggregateKey struct {
@@ -243,16 +282,18 @@ type mergeBenchmarkAggregateKey struct {
}
type mergeBenchmarkObserver struct {
- writer io.Writer
- recordErr error
- aggregates map[mergeBenchmarkAggregateKey]*mergeBenchmarkAggregate
- phase mergeBenchmarkPhase
- events []mergeBenchmarkEvent
- sequence atomic.Uint64
- active atomic.Int64
- overlapGen atomic.Uint64
- mu sync.Mutex
- attribution bool
+ writer io.Writer
+ recordErr error
+ aggregates map[mergeBenchmarkAggregateKey]*mergeBenchmarkAggregate
+ phase mergeBenchmarkPhase
+ events []mergeBenchmarkEvent
+ sequence atomic.Uint64
+ active atomic.Int64
+ currentStagedBytes atomic.Int64
+ peakConcurrentStagedBytes atomic.Uint64
+ overlapGen atomic.Uint64
+ mu sync.Mutex
+ attribution bool
}
func newMergeBenchmarkObserver(writer io.Writer, options mergeBenchmarkObserverOptions) *mergeBenchmarkObserver {
@@ -345,7 +386,10 @@ func (mbo *mergeBenchmarkObserver) snapshot() mergeBenchmarkSnapshot {
}
mbo.mu.Lock()
defer mbo.mu.Unlock()
- snapshot := mergeBenchmarkSnapshot{Events: append([]mergeBenchmarkEvent(nil), mbo.events...)}
+ snapshot := mergeBenchmarkSnapshot{
+ Events: append([]mergeBenchmarkEvent(nil), mbo.events...),
+ PeakConcurrentStagedBytes: mbo.peakConcurrentStagedBytes.Load(),
+ }
if mbo.recordErr != nil {
snapshot.Error = mbo.recordErr.Error()
}
@@ -364,11 +408,54 @@ func (mbo *mergeBenchmarkObserver) snapshot() mergeBenchmarkSnapshot {
}
type mergeEvaluationObservation struct {
- pluginCalls atomic.Uint64
- evaluated atomic.Uint64
- retained atomic.Uint64
- dropped atomic.Uint64
- oversized atomic.Uint64
+ observer *mergeBenchmarkObserver
+ stagingBatches []mergeBenchmarkStagingBatch
+ pluginCalls atomic.Uint64
+ evaluated atomic.Uint64
+ retained atomic.Uint64
+ dropped atomic.Uint64
+ oversized atomic.Uint64
+ currentStagedBytes atomic.Uint64
+ peakStagedBytes atomic.Uint64
+ mu sync.Mutex
+}
+
+func (meo *mergeEvaluationObservation) observeStagedBytes(value uint64) {
+ if meo == nil {
+ return
+ }
+ previous := meo.currentStagedBytes.Swap(value)
+ updateAtomicMaximum(&meo.peakStagedBytes, value)
+ if meo.observer == nil || previous == value {
+ return
+ }
+ var concurrent int64
+ if value > previous {
+ concurrent = meo.observer.currentStagedBytes.Add(int64(value - previous))
+ } else {
+ concurrent = meo.observer.currentStagedBytes.Add(-int64(previous - value))
+ }
+ if concurrent > 0 {
+ updateAtomicMaximum(&meo.observer.peakConcurrentStagedBytes, uint64(concurrent))
+ }
+}
+
+func (meo *mergeEvaluationObservation) recordStagingBatch(reason mergeStagingFlushReason, bytes, traces uint64) {
+ if meo == nil || traces == 0 {
+ return
+ }
+ meo.mu.Lock()
+ meo.stagingBatches = append(meo.stagingBatches, mergeBenchmarkStagingBatch{Reason: reason, Bytes: bytes, Traces: traces})
+ meo.mu.Unlock()
+}
+
+func (meo *mergeEvaluationObservation) stagingSnapshot() []mergeBenchmarkStagingBatch {
+ if meo == nil {
+ return nil
+ }
+ meo.mu.Lock()
+ defer meo.mu.Unlock()
+ return append([]mergeBenchmarkStagingBatch(nil), meo.stagingBatches...)
}
type mergeBenchmarkOperation struct {
@@ -498,7 +585,7 @@ func (mbo *mergeBenchmarkObserver) beginSeed(seed *mergeBenchmarkSeed, reason me
}
startedAt := time.Now()
operation := &mergeBenchmarkOperation{
- observer: mbo, evaluation: &mergeEvaluationObservation{}, initialReason: reason, startedAt: startedAt, event: seed.event,
+ observer: mbo, evaluation: &mergeEvaluationObservation{observer: mbo}, initialReason: reason, startedAt: startedAt, event: seed.event,
}
operation.event.InitialReason = reason
operation.event.QueueNanos = startedAt.Sub(seed.dispatched).Nanoseconds()
@@ -523,6 +610,12 @@ func (mbo *mergeBenchmarkObserver) finish(operation *mergeBenchmarkOperation, ou
event.TracesRetained = operation.evaluation.retained.Load()
event.TracesDropped = operation.evaluation.dropped.Load()
event.OversizedTraces = operation.evaluation.oversized.Load()
+ event.PeakStagedBytes = operation.evaluation.peakStagedBytes.Load()
+ event.StagingBatches = operation.evaluation.stagingSnapshot()
+ for batchIdx := range event.StagingBatches {
+ event.ChargedStagingBytes = saturatingAddUint64(event.ChargedStagingBytes, event.StagingBatches[batchIdx].Bytes)
+ }
+ operation.evaluation.observeStagedBytes(0)
event.LosslessRetry = losslessRetry
overlapped := operation.startedOverlap || mbo.active.Load() > 1 || mbo.overlapGen.Load() != operation.overlapGen
mbo.active.Add(-1)
diff --git a/banyand/trace/merge_benchmark_observer_test.go b/banyand/trace/merge_benchmark_observer_test.go
index 5f2f550a6..8b72d50b6 100644
--- a/banyand/trace/merge_benchmark_observer_test.go
+++ b/banyand/trace/merge_benchmark_observer_test.go
@@ -56,6 +56,19 @@ func TestMergeBenchmarkCountsActualSamplerInvocations(t *testing.T) {
require.Zero(t, bypassed.evaluated.Load())
}
+func TestMergeBenchmarkObserverReportsConcurrentStagingPeak(t *testing.T) {
+ observer := newMergeBenchmarkObserver(nil, mergeBenchmarkObserverOptions{})
+ first := &mergeEvaluationObservation{observer: observer}
+ second := &mergeEvaluationObservation{observer: observer}
+
+ first.observeStagedBytes(100)
+ second.observeStagedBytes(200)
+ first.observeStagedBytes(0)
+ second.observeStagedBytes(0)
+
+ require.Equal(t, uint64(300), observer.snapshot().PeakConcurrentStagedBytes)
+}
+
func TestBenchmarkSidxPartTotalsReportsMetadataFailure(t *testing.T) {
_, _, totalsErr := benchmarkSidxPartTotals(fs.NewLocalFileSystem(), map[uint64]string{1: filepath.Join(t.TempDir(), "missing")})
require.ErrorContains(t, totalsErr, "cannot read secondary-index part metadata")
@@ -93,6 +106,18 @@ func TestMergeBenchmarkObserverClassifiesHotAndMatureMerges(t *testing.T) {
require.Equal(t, uint64(2), event.TracesEvaluated)
require.Positive(t, samplerCalls)
require.Equal(t, samplerCalls, int64(event.PluginCalls))
+ require.Positive(t, event.EstimatedStagingBytes)
+ require.Positive(t, event.StagingHardLimit)
+ require.Positive(t, event.DecisionBatchLimit)
+ require.LessOrEqual(t, event.DecisionBatchLimit, event.StagingHardLimit)
+ require.Positive(t, event.PlannedStagingBatches)
+ require.Positive(t, event.DecisionMaxTraceCount)
+ require.Positive(t, event.PeakStagedBytes)
+ require.Len(t, event.StagingBatches, 1)
+ require.Equal(t, mergeStagingFlushEndOfMerge, event.StagingBatches[0].Reason)
+ require.Equal(t, event.TracesEvaluated, event.StagingBatches[0].Traces)
+ require.Positive(t, event.StagingBatches[0].Bytes)
+ require.Equal(t, event.StagingBatches[0].Bytes, event.ChargedStagingBytes)
require.Zero(t, event.HotInputParts)
require.Equal(t, uint32(2), event.MatureInputParts)
require.Zero(t, event.HotInputRows)
@@ -109,6 +134,22 @@ func TestMergeBenchmarkObserverClassifiesHotAndMatureMerges(t *testing.T) {
})
}
+func TestMergeBenchmarkObserverReportsBudgetTriggeredBatches(t *testing.T) {
+ previousBudget := testStageBudgetOverride
+ testStageBudgetOverride = 1
+ t.Cleanup(func() { testStageBudgetOverride = previousBudget })
+
+ event, samplerCalls, _ := runObservedBenchmarkMerge(t, true)
+ require.Equal(t, int64(2), samplerCalls)
+ require.Equal(t, uint64(2), event.PluginCalls)
+ require.Len(t, event.StagingBatches, 2)
+ require.Equal(t, mergeStagingFlushByteAndTraceLimit, event.StagingBatches[0].Reason)
+ require.Equal(t, uint64(1), event.StagingBatches[0].Traces)
+ require.Equal(t, mergeStagingFlushEndOfMerge, event.StagingBatches[1].Reason)
+ require.Equal(t, uint64(1), event.StagingBatches[1].Traces)
+ require.GreaterOrEqual(t, event.PeakStagedBytes, event.StagingBatches[0].Bytes)
+}
+
func TestClassifyMergeObservation(t *testing.T) {
tests := []struct {
name string
diff --git a/banyand/trace/merger.go b/banyand/trace/merger.go
index 193cd8ae2..f4ff236c1 100644
--- a/banyand/trace/merger.go
+++ b/banyand/trace/merger.go
@@ -78,17 +78,117 @@ const (
// defaultMaxStagedTraceCount independently bounds the number of logical
// traces exposed to one sampler call even when their byte footprint is tiny.
defaultMaxStagedTraceCount = 64 * 1024
+ // adaptiveBatchHardLimitDivisor keeps the preferred decision working set
+ // below the resource-derived safety ceiling while leaving room for a whole
+ // trace to cross the preferred boundary.
+ adaptiveBatchHardLimitDivisor = 8
+ // adaptiveBatchArenaMultiplier lets a completed batch reuse the bounded raw
+ // arena cache during the following batch without making the cache itself the
+ // staging safety limit.
+ adaptiveBatchArenaMultiplier = 2
)
+type adaptiveDecisionBatchPlan struct {
+ EstimatedBytes uint64
+ HardLimit uint64
+ BatchLimit uint64
+ PlannedBatches uint64
+}
+
+func planAdaptiveDecisionBatch(parts []*partWrapper, hardLimit uint64) adaptiveDecisionBatchPlan {
+ var compressedBytes, uncompressedBytes, blocksCount uint64
+ for _, partData := range parts {
+ if partData == nil || partData.p == nil {
+ continue
+ }
+ metadata := &partData.p.partMetadata
+ compressedBytes = saturatingAddUint64(compressedBytes, metadata.CompressedSizeBytes)
+ uncompressedBytes = saturatingAddUint64(uncompressedBytes, metadata.UncompressedSpanSizeBytes)
+ blocksCount = saturatingAddUint64(blocksCount, metadata.BlocksCount)
+ }
+ payloadBytes := max(compressedBytes, uncompressedBytes)
+ structuralBytes := saturatingMultiplyUint64(blocksCount, estimatedStagedBlockStructuralBytes())
+ estimatedBytes := saturatingAddUint64(payloadBytes, structuralBytes)
+ nominalLimit := nominalDecisionBatchLimit(hardLimit)
+ if estimatedBytes == 0 {
+ return adaptiveDecisionBatchPlan{HardLimit: hardLimit, BatchLimit: nominalLimit, PlannedBatches: 1}
+ }
+ plannedBatches := roundedQuotientUint64(estimatedBytes, nominalLimit)
+ batchLimit := ceilDivUint64(estimatedBytes, plannedBatches)
+ minimumLimit := min(uint64(defaultStageBudgetFloor), hardLimit)
+ batchLimit = min(max(batchLimit, minimumLimit), hardLimit)
+ return adaptiveDecisionBatchPlan{
+ EstimatedBytes: estimatedBytes,
+ HardLimit: hardLimit,
+ BatchLimit: batchLimit,
+ PlannedBatches: plannedBatches,
+ }
+}
+
+func estimatedStagedBlockStructuralBytes() uint64 {
+ return uint64(unsafe.Sizeof(stagedTrace{})) + uint64(unsafe.Sizeof(blockMetadata{})) +
+ uint64(unsafe.Sizeof(stagedDataBlockBuffer{})) + uint64(unsafe.Sizeof(stagedTraceGroup{})) +
+ stagedTraceDecisionBytes() + stagedEvaluationSlotBytes(true)
+}
+
+func nominalDecisionBatchLimit(hardLimit uint64) uint64 {
+ if hardLimit == 0 {
+ return defaultSmallMergeThreshold
+ }
+ poolWindow := uint64(maxPooledStagedArenaBytes * adaptiveBatchArenaMultiplier)
+ resourceWindow := hardLimit / adaptiveBatchHardLimitDivisor
+ nominalLimit := max(uint64(defaultStageBudgetFloor), poolWindow, resourceWindow)
+ return min(nominalLimit, hardLimit)
+}
+
+func roundedQuotientUint64(numerator, denominator uint64) uint64 {
+ if denominator == 0 {
+ return 1
+ }
+ quotient := numerator / denominator
+ remainder := numerator % denominator
+ if remainder >= ceilDivUint64(denominator, 2) {
+ quotient++
+ }
+ return max(uint64(1), quotient)
+}
+
+func ceilDivUint64(numerator, denominator uint64) uint64 {
+ if denominator == 0 {
+ return 0
+ }
+ quotient := numerator / denominator
+ if numerator%denominator != 0 {
+ quotient++
+ }
+ return quotient
+}
+
+func saturatingAddUint64(left, right uint64) uint64 {
+ if ^uint64(0)-left < right {
+ return ^uint64(0)
+ }
+ return left + right
+}
+
+func saturatingMultiplyUint64(left, right uint64) uint64 {
+ if left == 0 || right == 0 {
+ return 0
+ }
+ if left > ^uint64(0)/right {
+ return ^uint64(0)
+ }
+ return left * right
+}
+
// testStageBudgetOverride forces the staging budget when non-zero. Test-only
// seam (mirrors forceSlowMerge); production always derives it from the memory
// limit via stageBudgetFromLimit.
var testStageBudgetOverride uint64
-// resolveStageBudget returns the soft cap on bytes the in-merge retention filter
-// may stage before a trace-boundary chunk flush, derived from the protector's
-// memory limit. There is no operator flag; the budget self-tunes from the limit
-// and the merge concurrency.
+// resolveStageBudget returns the resource-derived hard staging ceiling. A
+// selected merge derives its smaller preferred decision-batch limit separately
+// from part metadata. There is no operator flag.
func resolveStageBudget(opt option) uint64 {
if testStageBudgetOverride > 0 {
return testStageBudgetOverride
@@ -132,7 +232,7 @@ func stageBudgetFromLimit(limit uint64) uint64 {
func maxStagedTraceCountFromBudget(stageBudget uint64) int {
minimumTraceBytes := uint64(unsafe.Sizeof(stagedTrace{})) + uint64(unsafe.Sizeof(stagedTraceGroup{})) +
- stagedTraceDecisionBytes() + stagedEvaluationSlotBytes(false)
+ stagedTraceDecisionBytes() + stagedEvaluationSlotBytes(true)
if stageBudget == 0 || minimumTraceBytes == 0 {
return defaultMaxStagedTraceCount
}
@@ -479,23 +579,27 @@ func (tst *tsTable) buildHotMergeFilterDecisionAt(parts []*partWrapper, logicalN
tst.incPipelineGuardBypassed()
return nil, mergeReasonGrace
}
- stageBudget := resolveStageBudget(tst.option)
- guard := tst.newTraceFragmentGuardSession(parts, tst.option.maxTraceFragmentGap, stageBudget)
+ stagingHardLimit := resolveStageBudget(tst.option)
+ stagingPlan := planAdaptiveDecisionBatch(parts, stagingHardLimit)
+ guard := tst.newTraceFragmentGuardSession(parts, tst.option.maxTraceFragmentGap, stagingHardLimit)
if guard == nil {
tst.incPipelineGuardBypassed()
return nil, mergeReasonGuardUnavailable
}
chain := newMergeChain(tst.group, "", samplers, tst.option.decideTimeoutCircuitBreak)
return &mergeFilter{
- chain: chain,
- guard: guard,
- ctx: tst.loopCloser.Ctx(),
- owner: tst,
- timeout: tst.option.decideTimeout,
- stageBudget: stageBudget,
- traceBudget: resolveTraceBudget(tst.option),
- maxTraceCount: maxStagedTraceCountFromBudget(stageBudget),
- forceSlow: projectionRequiresSlowPath(chain.projection),
+ chain: chain,
+ guard: guard,
+ ctx: tst.loopCloser.Ctx(),
+ owner: tst,
+ timeout: tst.option.decideTimeout,
+ stagingHardLimit: stagingHardLimit,
+ decisionBatchLimit: stagingPlan.BatchLimit,
+ estimatedStagingBytes: stagingPlan.EstimatedBytes,
+ plannedStagingBatches: stagingPlan.PlannedBatches,
+ traceBudget: resolveTraceBudget(tst.option),
+ maxTraceCount: maxStagedTraceCountFromBudget(stagingPlan.BatchLimit),
+ forceSlow: projectionRequiresSlowPath(chain.projection),
}, ""
}
@@ -555,6 +659,11 @@ func (tst *tsTable) mergePartsThenSendIntroductionObserved(creator snapshotCreat
operation.setInitialReason(initialReason)
if filter != nil && operation != nil {
filter.observation = operation.evaluation
+ operation.event.EstimatedStagingBytes = filter.estimatedStagingBytes
+ operation.event.StagingHardLimit = filter.stagingHardLimit
+ operation.event.DecisionBatchLimit = filter.decisionBatchLimit
+ operation.event.PlannedStagingBatches = filter.plannedStagingBatches
+ operation.event.DecisionMaxTraceCount = filter.maxTraceCount
}
losslessRetry := false
defer func() {
@@ -1282,16 +1391,19 @@ type stagedTraceGroup struct {
// mergeBlocks. When nil, mergeBlocks behaves exactly as before (no staging, no
// decode changes).
type mergeFilter struct {
- chain *mergeChain
- guard *traceFragmentGuardSession
- observation *mergeEvaluationObservation
- ctx context.Context
- owner *tsTable
- timeout time.Duration
- stageBudget uint64 // soft cap on staged bytes; a trace-boundary chunk flush fires once exceeded (0 disables chunking)
- traceBudget uint64 // hard cap on one trace's staged bytes; exceeding it retains the trace without evaluation (0 disables bypass)
- maxTraceCount int // hard cap on logical traces sent to one Decide call (0 disables the count cap)
- forceSlow bool // forces the slow assembly path when the chain projects row data
+ chain *mergeChain
+ guard *traceFragmentGuardSession
+ observation *mergeEvaluationObservation
+ ctx context.Context
+ owner *tsTable
+ timeout time.Duration
+ stagingHardLimit uint64 // resource-derived safety ceiling for one merge's staged bytes
+ decisionBatchLimit uint64 // metadata-derived preferred bytes per complete-trace Decide batch
+ estimatedStagingBytes uint64 // selected-part metadata estimate used to plan decision batches
+ plannedStagingBatches uint64 // estimated number of balanced decision batches
+ traceBudget uint64 // hard cap on one trace's staged bytes; exceeding it retains the trace without evaluation (0 disables bypass)
+ maxTraceCount int // hard cap on logical traces sent to one Decide call (0 disables the count cap)
+ forceSlow bool // forces the slow assembly path when the chain projects row data
}
func (st *stagedTrace) metadata() *blockMetadata {
@@ -1759,6 +1871,9 @@ func (tes *traceEvaluationStager) ensureBuffers() {
}
func (tes *traceEvaluationStager) releaseBuffers() {
+ if tes.filter != nil && tes.filter.observation != nil {
+ tes.filter.observation.observeStagedBytes(0)
+ }
for stagedIdx := range tes.staged {
releaseStagedTrace(&tes.staged[stagedIdx])
}
@@ -1776,10 +1891,13 @@ func (tes *traceEvaluationStager) releaseBuffers() {
tes.groups = nil
}
-func (tes *traceEvaluationStager) flush() {
+func (tes *traceEvaluationStager) flush(reason mergeStagingFlushReason) {
if len(tes.staged) == 0 {
return
}
+ if tes.filter.observation != nil {
+ tes.filter.observation.recordStagingBatch(reason, tes.stagedBytes, uint64(len(tes.groups)))
+ }
flushStaged(tes.bw, tes.filter, tes.staged, tes.groups, !tes.invalidOrder && !tes.invalidMetadata, tes.droppedSet)
clear(tes.staged)
tes.staged = tes.staged[:0]
@@ -1793,6 +1911,9 @@ func (tes *traceEvaluationStager) flush() {
tes.stagedTraceCount = 0
tes.invalidOrder = false
tes.invalidMetadata = false
+ if tes.filter.observation != nil {
+ tes.filter.observation.observeStagedBytes(0)
+ }
}
func (tes *traceEvaluationStager) writeBypassed(bypassed []stagedTrace) {
@@ -1849,6 +1970,9 @@ func (tes *traceEvaluationStager) stage(st stagedTrace) {
}
tes.stagedBytes += batchBytes
tes.currentTraceBytes += traceBytes
+ if tes.filter.observation != nil {
+ tes.filter.observation.observeStagedBytes(tes.stagedBytes)
+ }
group := &tes.groups[len(tes.groups)-1]
group.end = len(tes.staged)
group.accountedBytes = tes.currentTraceBytes
@@ -1869,6 +1993,13 @@ func (tes *traceEvaluationStager) stage(st stagedTrace) {
return
}
if tes.currentTraceStart > 0 {
+ if tes.filter.observation != nil {
+ prefixBytes := stagedBatchFixedBytes()
+ for groupIdx := range tes.groups[:len(tes.groups)-1] {
+ prefixBytes += tes.groups[groupIdx].accountedBytes
+ }
+ tes.filter.observation.recordStagingBatch(mergeStagingFlushOversizedTrace, prefixBytes, uint64(len(tes.groups)-1))
+ }
flushStaged(tes.bw, tes.filter, tes.staged[:tes.currentTraceStart], tes.groups[:len(tes.groups)-1],
!tes.invalidOrder && !tes.invalidMetadata, tes.droppedSet)
}
@@ -1887,6 +2018,9 @@ func (tes *traceEvaluationStager) stage(st stagedTrace) {
tes.stagedTraceCount = 0
tes.invalidOrder = false
tes.invalidMetadata = false
+ if tes.filter.observation != nil {
+ tes.filter.observation.observeStagedBytes(0)
+ }
if tes.filter.owner != nil {
tes.filter.owner.incPipelineOversizedTracesBypassed(1)
}
@@ -1901,18 +2035,31 @@ func (tes *traceEvaluationStager) flushBefore(nextTraceID string, pendingBlock *
}
pendingCompletesStagedTrace := !pendingBlockIsEmpty && pendingBlock.bm.traceID == tes.lastStagedTraceID
if !pendingCompletesStagedTrace {
- tes.flush()
+ tes.flush(tes.budgetFlushReason())
}
}
func (tes *traceEvaluationStager) flushAfter(completedTraceID, nextTraceID string) {
if tes.batchBudgetReached() && completedTraceID != nextTraceID {
- tes.flush()
+ tes.flush(tes.budgetFlushReason())
+ }
+}
+
+func (tes *traceEvaluationStager) budgetFlushReason() mergeStagingFlushReason {
+ bytesReached := tes.filter.decisionBatchLimit > 0 && tes.stagedBytes >= tes.filter.decisionBatchLimit
+ tracesReached := tes.filter.maxTraceCount > 0 && tes.stagedTraceCount >= tes.filter.maxTraceCount
+ switch {
+ case bytesReached && tracesReached:
+ return mergeStagingFlushByteAndTraceLimit
+ case bytesReached:
+ return mergeStagingFlushByteLimit
+ default:
+ return mergeStagingFlushTraceLimit
}
}
func (tes *traceEvaluationStager) batchBudgetReached() bool {
- bytesReached := tes.filter.stageBudget > 0 && tes.stagedBytes >= tes.filter.stageBudget
+ bytesReached := tes.filter.decisionBatchLimit > 0 && tes.stagedBytes >= tes.filter.decisionBatchLimit
tracesReached := tes.filter.maxTraceCount > 0 && tes.stagedTraceCount >= tes.filter.maxTraceCount
return bytesReached || tracesReached
}
@@ -2074,7 +2221,7 @@ func mergeBlocks(closeCh <-chan struct{}, bw *blockWriter, br *blockReader, conf
}
releaseDecoder()
if evaluationStager != nil {
- evaluationStager.flush()
+ evaluationStager.flush(mergeStagingFlushEndOfMerge)
}
var pm partMetadata
var tf traceIDFilter
diff --git a/banyand/trace/pipeline_chain_filter_test.go b/banyand/trace/pipeline_chain_filter_test.go
index db9459663..c6c157c16 100644
--- a/banyand/trace/pipeline_chain_filter_test.go
+++ b/banyand/trace/pipeline_chain_filter_test.go
@@ -534,6 +534,91 @@ func TestStageBudgetFromLimit(t *testing.T) {
}
}
+func TestAdaptiveDecisionBatchPlanUsesSelectedPartMetadata(t *testing.T) {
+ const hardLimit = uint64(256 << 20)
+ newPart := func(compressed, uncompressed uint64) *partWrapper {
+ return &partWrapper{p: &part{partMetadata: partMetadata{
+ CompressedSizeBytes: compressed,
+ UncompressedSpanSizeBytes: uncompressed,
+ }}}
+ }
+
+ tests := []struct {
+ name string
+ parts []*partWrapper
+ wantEstimate uint64
+ wantBatchLimit uint64
+ wantPlannedBatches uint64
+ }{
+ {
+ name: "missing metadata uses nominal reusable window",
+ parts: []*partWrapper{newPart(0, 0)},
+ wantEstimate: 0,
+ wantBatchLimit: 32 << 20,
+ wantPlannedBatches: 1,
+ },
+ {
+ name: "small merge stays in one decision batch",
+ parts: []*partWrapper{newPart(8<<20, 20<<20)},
+ wantEstimate: 20 << 20,
+ wantBatchLimit: 20 << 20,
+ wantPlannedBatches: 1,
+ },
+ {
+ name: "compressed size covers absent uncompressed metadata",
+ parts: []*partWrapper{newPart(12<<20, 0)},
+ wantEstimate: 12 << 20,
+ wantBatchLimit: 16 << 20,
+ wantPlannedBatches: 1,
+ },
+ {
+ name: "production shaped merge is balanced around reusable window",
+ parts: []*partWrapper{
+ newPart(18<<20, 50<<20),
+ newPart(17<<20, 48<<20),
+ },
+ wantEstimate: 98 << 20,
+ wantBatchLimit: ceilDivUint64(98<<20, 3),
+ wantPlannedBatches: 3,
+ },
+ }
+
+ for _, testCase := range tests {
+ t.Run(testCase.name, func(t *testing.T) {
+ plan := planAdaptiveDecisionBatch(testCase.parts, hardLimit)
+ require.Equal(t, hardLimit, plan.HardLimit)
+ require.Equal(t, testCase.wantEstimate, plan.EstimatedBytes)
+ require.Equal(t, testCase.wantBatchLimit, plan.BatchLimit)
+ require.Equal(t, testCase.wantPlannedBatches, plan.PlannedBatches)
+ })
+ }
+}
+
+func TestAdaptiveDecisionBatchPlanSaturatesCorruptMetadata(t *testing.T) {
+ parts := []*partWrapper{{p: &part{partMetadata: partMetadata{
+ CompressedSizeBytes: ^uint64(0),
+ UncompressedSpanSizeBytes: ^uint64(0),
+ BlocksCount: ^uint64(0),
+ }}}}
+
+ plan := planAdaptiveDecisionBatch(parts, 256<<20)
+ require.Equal(t, ^uint64(0), plan.EstimatedBytes)
+ require.Positive(t, plan.PlannedBatches)
+ require.LessOrEqual(t, plan.BatchLimit, plan.HardLimit)
+}
+
+func TestAdaptiveDecisionBatchPlanChargesBlockStructure(t *testing.T) {
+ const payloadBytes = uint64(20 << 20)
+ parts := []*partWrapper{{p: &part{partMetadata: partMetadata{
+ CompressedSizeBytes: 8 << 20,
+ UncompressedSpanSizeBytes: payloadBytes,
+ BlocksCount: 10_000,
+ }}}}
+
+ plan := planAdaptiveDecisionBatch(parts, 256<<20)
+ require.Greater(t, plan.EstimatedBytes, payloadBytes)
+}
+
func TestInMergeFilter_DisabledFlag_LegacyIdentity(t *testing.T) {
svcs, teardown := setUpWithPipelineFlags(t,
"--trace-pipeline-merge-grace-default=0",
diff --git a/banyand/trace/pipeline_chain_test.go b/banyand/trace/pipeline_chain_test.go
index b635cfe5a..dd3783e1c 100644
--- a/banyand/trace/pipeline_chain_test.go
+++ b/banyand/trace/pipeline_chain_test.go
@@ -242,10 +242,10 @@ func TestMergeFilter_AssemblesEveryPhysicalBlockBeforeDecide(t *testing.T) {
sampler := &wholeTraceErrorSampler{}
chain := newMergeChain("g", "s", []sdk.Sampler{sampler}, 0)
filter := &mergeFilter{
- chain: chain,
- timeout: time.Second,
- stageBudget: 1,
- forceSlow: true,
+ chain: chain,
+ timeout: time.Second,
+ decisionBatchLimit: 1,
+ forceSlow: true,
}
parts := appendTrace(splitTraceParts("error"), "trace-next", "success")
diff --git a/docs/design/trace-pipeline-merge-optimization-plan.md b/docs/design/trace-pipeline-merge-optimization-plan.md
index 9233feaa7..61aa1fd3e 100644
--- a/docs/design/trace-pipeline-merge-optimization-plan.md
+++ b/docs/design/trace-pipeline-merge-optimization-plan.md
@@ -10,16 +10,16 @@ The controlled mature-merge comparison uses the same frozen production selection
retain-all variants. Each variant ran in five fresh resource-limited Docker processes. All ten runs passed the selection,
maturity, row-count, secondary-index, and logical-ledger correctness gates.
-Each retain-all run made two plugin calls, evaluated and retained 33,353 complete trace IDs, dropped no traces, and
+Each retain-all run made one plugin call, evaluated and retained 33,353 complete trace IDs, dropped no traces, and
preserved all 147,126 span rows. Compressed logical write amplification over the core and both secondary indexes remained
identical at 1.0009 times.
Compared with the pipeline-disabled medians, the native retain-all path currently adds approximately:
-- 121.9 MiB of allocated bytes, an increase of 56.2%;
-- 1.07 million allocations, an increase of 38.0%;
-- 91.5 MiB of peak Go heap, an increase of 83.4%; and
-- 98.4 MiB of peak RSS, an increase of 72.8%.
+- 113.5 MiB of allocated bytes, an increase of 59.6%;
+- 466,941 allocations, an increase of 16.6%;
+- 87.1 MiB of peak Go heap, an increase of 108.9%; and
+- 85.8 MiB of peak RSS, an increase of 75.3%.
The differential allocation profile points primarily to `stageRawTrace`, `blockMetadata.copyFrom`,
`traceEvaluationStager.stage`, and `assembleStagedEvaluationBatch`. Profile values overlap and must not be added together.
@@ -174,27 +174,56 @@ performance gate.
---
-### Phase 4 — Batch Size Tuning
+### Phase 4 — Adaptive Budget Validation
-**Scope.** Empirically select the smallest staging batch size that meets memory and throughput goals without excess
-plugin-call overhead.
+**Scope.** Validate the existing resource-derived staging budget under cold, steady-state, and naturally varying merge
+sizes. The budget is a memory-safety ceiling rather than a calibrated performance knob; this phase does not add an
+operator or benchmark override.
**Entries from original plan.** P5.
-**Hard invariants.** Logical evaluated trace count, per-trace verdict, and merged output byte-identical across batch
-sizes. Both byte and trace-count limits enforced.
+**Hard invariants.** The production budget formula remains derived solely from the detected cgroup memory and CPU
+limits. Logical evaluated trace count, per-trace verdict, and merged output remain correct when natural input variation
+causes complete-trace batch boundaries. Both byte and trace-count limits remain enforced.
**Exit gate.**
-- [ ] 16 / 32 / 64 / 128 / 256 / 512 MiB sweep completed against the unchanged frozen seed.
-- [ ] Each run reports peak heap, peak RSS, allocated bytes, allocation count, plugin calls, traces per call, CPU time,
- wall time, logical write amplification.
-- [ ] Selected batch size documented with both byte and trace-count limits; alternative sizes ruled out with evidence.
+- [x] Every sampled merge reports peak charged staging bytes, batch bytes and trace count, flush reason, and peak
+ concurrent staged bytes.
+- [ ] Cold first-merge cost and subsequent same-process reuse are reported separately without changing the budget.
+- [ ] The serialized 24-hour workload validates naturally varying merge sizes under the canonical two-CPU, 4 GiB
+ container and its resource-derived 256 MiB budget.
+- [ ] At least one naturally budget-limited merge, or explicit evidence that the production-shaped workload never
+ reaches the limit, is documented without manufacturing smaller batches.
+- [ ] Core and secondary-index ledgers, complete-trace decisions, peak heap, peak RSS, allocations, plugin calls, CPU
+ time, wall time, and logical write amplification remain reported.
**Dependencies.** Phase 3.
-**Boundary rationale.** Cannot tune a knob whose dependent structure is still changing. Phase 3 must be stable before
-doing a controlled sweep.
+**Boundary rationale.** A fixed budget selected from one seed would not generalize to varying production inputs or
+container sizes. Phase 3 must be stable before its resource-derived ceiling and cross-merge reuse can be validated.
+
+**Adaptive decision-batch implementation.** The resource-derived staging and per-trace limits remain hard safety
+ceilings. Each MERGE or FINALIZE selection now derives a separate preferred decision-batch limit from the selected core
+parts' `CompressedSizeBytes`, `UncompressedSpanSizeBytes`, and `BlocksCount`. The estimator uses the greater of compressed
+and uncompressed payload bytes plus a runtime-structure estimate per stored block. It then balances the estimated input
+around a nominal reusable window derived from the hard limit and the bounded BanyanDB staging-arena pool. A two-CPU,
+4 GiB process therefore keeps its 256 MiB hard ceiling while targeting approximately 32 MiB decision batches. Small
+estimated merges remain a single call; larger merges are split only at complete-trace boundaries. Exact runtime charged
+bytes remain authoritative, and the oversized single-trace fail-open limit is unchanged.
+
+The fragment guard continues to receive the hard ceiling rather than the preferred decision limit. The per-decision
+trace-count bound is derived from the preferred limit. Benchmark events report the metadata estimate, hard ceiling,
+preferred limit, planned batch count, effective trace-count limit, actual batch sizes and reasons, and peak staged bytes.
+The metadata estimate is advisory: an underestimate causes additional complete-trace batches, while the runtime byte
+accounting and hard per-trace bypass preserve safety and correctness.
+
+An initial two-CPU, 4 GiB controlled diagnostic estimated 85,716,327 bytes from the frozen selection and resolved a
+28,572,109-byte preferred limit under the unchanged 268,435,456-byte hard ceiling. Runtime accounting produced five
+complete-trace batches, peaked at 28,575,614 staged bytes, evaluated and retained all 33,353 traces, and preserved every
+core and secondary-index ledger. Compared with the preceding five-run retain-all medians, this single diagnostic reduced
+peak heap from 167.1 MiB to 106.8 MiB and peak RSS from 199.7 MiB to 151.6 MiB. It is evidence that adaptive batching
+controls the live set, not a replacement for the required alternating acceptance series.
---
diff --git a/docs/design/trace-pipeline-merge-performance-test.md b/docs/design/trace-pipeline-merge-performance-test.md
index 71c08e8af..842e0f218 100644
--- a/docs/design/trace-pipeline-merge-performance-test.md
+++ b/docs/design/trace-pipeline-merge-performance-test.md
@@ -276,9 +276,9 @@ controller is pinned away from the data-node CPUs and reports its own scheduling
The canonical container profile is:
-- four pinned logical CPUs with a four-CPU quota;
-- `GOMAXPROCS=4`;
-- 8 GiB memory limit;
+- two pinned logical CPUs with a two-CPU quota;
+- `GOMAXPROCS=2`;
+- 4 GiB memory limit;
- memory swap limit equal to the memory limit, preventing additional swap use;
- a fixed PID limit of 512;
- the same read-only seed mount and a new writable fixture volume for every run;
@@ -289,7 +289,7 @@ must have identical limits. Every report records both the requested Docker limit
inside the container. A mismatch invalidates the run because merge concurrency and staging budgets are derived from the
detected cgroup resources.
-The four-CPU limit is an upper resource envelope, not a utilization target. Low utilization from the single shard is a
+The two-CPU limit is an upper resource envelope, not a utilization target. Low utilization from the single shard is a
valid result and must not be corrected by cloning the workload, increasing shard count, or forcing extra merges. This
design measures plugin impact on the downloaded shard; it does not establish whole-node saturation capacity.
@@ -973,7 +973,9 @@ independently in addition to their combined totals.
- traces retained by duration, error, tag, and healthy-hash rules, plus traces reaching each rule;
- plugin calls and traces per call;
- grace-bypassed merge and trace counts;
-- staged bytes and peak staged bytes;
+- metadata-estimated staging bytes, resource-derived staging hard limit, adaptive decision-batch limit, planned batch
+ count, and effective trace-count limit;
+- actual batch bytes, traces, flush reason, peak staged bytes, and peak concurrent staged bytes;
- fragment-guard candidate parts, Bloom probes, deferrals, and budget exhaustion;
- plugin errors, panics, timeouts, and malformed verdicts;
- sampled and unsampled merge counts, including every unsampled reason.
diff --git a/scripts/run-trace-merge-baseline.sh b/scripts/run-trace-merge-baseline.sh
index 273ac42c7..63b086daf 100755
--- a/scripts/run-trace-merge-baseline.sh
+++ b/scripts/run-trace-merge-baseline.sh
@@ -17,11 +17,14 @@
set -euo pipefail
ROOT=$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)
-FIXTURE=${FIXTURE:-"$ROOT/.scratch/trace-pipeline-merge-performance/generated-fixture-ticket05"}
+FIXTURE=${FIXTURE:-"$ROOT/.scratch/trace-pipeline-merge-performance/generated-fixture-2x-v2-timestamps"}
OUTPUT=${OUTPUT:-"$ROOT/.scratch/trace-pipeline-merge-performance/baseline-report"}
IMAGE=${IMAGE:-golang:1.25.12}
-DATA_CPUS=${DATA_CPUS:-0-3}
-CONTROLLER_CPU=${CONTROLLER_CPU:-4}
+DATA_CPUS=${DATA_CPUS:-0-1}
+DATA_CPU_LIMIT=${DATA_CPU_LIMIT:-2}
+DATA_MEMORY=${DATA_MEMORY:-4g}
+DATA_GOMAXPROCS=${DATA_GOMAXPROCS:-2}
+CONTROLLER_CPU=${CONTROLLER_CPU:-2}
REPETITIONS=${REPETITIONS:-5}
PILOT_ACCELERATION=${PILOT_ACCELERATION:-}
CONTROLLED_ALTERNATING=${CONTROLLED_ALTERNATING:-1}
@@ -29,6 +32,7 @@ CONTROLLED_REPETITIONS=${CONTROLLED_REPETITIONS:-10}
CONTROLLED_SEED=${CONTROLLED_SEED:-"$OUTPUT/controlled-seed"}
DEFAULT_CONTROLLED_PLUGIN="$ROOT/.scratch/trace-pipeline-merge-performance/plugins/alwayskeepsampler.so"
CONTROLLED_PLUGIN=${CONTROLLED_PLUGIN:-"$DEFAULT_CONTROLLED_PLUGIN"}
+FULL_PIPELINE=${FULL_PIPELINE:-disabled}
BIN="$ROOT/.scratch/trace-pipeline-merge-performance/bin/trace-merge-benchmark"
COMMIT=$(git -C "$ROOT" rev-parse HEAD)
@@ -40,10 +44,15 @@ print(d['scheduleSHA256'])
print(d['rowCount'])
print(d['writeCount'])
print(d['coreCompressedBytes'] + sum(d['indexCompressedBytes'].values()))
-print(d['writeIntensity'])
+print(d.get('writeIntensity', 1))
print(d['logicalLedgerSHA256']['core'])
print(d['logicalLedgerSHA256']['latency'])
print(d['logicalLedgerSHA256']['start_time'])
+import datetime
+day_start=datetime.datetime.fromisoformat(d['dayStart'].replace('Z','+00:00'))
+day_start_ns=int(day_start.timestamp()*1_000_000_000)
+print(day_start_ns-int(d['mergeGrace']))
+print(day_start_ns+int(d['dayDuration'])+int(d['mergeGrace']))
PY
)
FIXTURE_SHA=${META[0]}
@@ -55,10 +64,12 @@ WRITE_INTENSITY=${META[5]}
CORE_LEDGER_SHA=${META[6]}
LATENCY_LEDGER_SHA=${META[7]}
START_TIME_LEDGER_SHA=${META[8]}
+SEGMENT_MIN_TIME_NANOS=${META[9]}
+SEGMENT_MAX_TIME_NANOS=${META[10]}
mkdir -p "$OUTPUT" "$(dirname "$BIN")"
go build -o "$BIN" "$ROOT/banyand/cmd/trace-merge-benchmark"
-if [[ "$CONTROLLED_ALTERNATING" == "1" && ! -f "$CONTROLLED_PLUGIN" && "$CONTROLLED_PLUGIN" == "$DEFAULT_CONTROLLED_PLUGIN" ]]; then
+if [[ ("$CONTROLLED_ALTERNATING" == "1" || "$FULL_PIPELINE" == "retain-all") && "$CONTROLLED_PLUGIN" == "$DEFAULT_CONTROLLED_PLUGIN" ]]; then
mkdir -p "$(dirname "$CONTROLLED_PLUGIN")"
go build -buildmode=plugin -o "$CONTROLLED_PLUGIN" "$ROOT/test/plugins/_alwayskeepsampler"
fi
@@ -108,8 +119,13 @@ run_once() {
chmod 0777 /run'
local attribution_flag=()
if [[ "$attribution" == true ]]; then attribution_flag=(--attribution); fi
- docker run -d --name "$container" --cpuset-cpus="$DATA_CPUS" --cpus=4 --memory=8g --memory-swap=8g --pids-limit=512 \
- -e GOMAXPROCS=4 -e IMAGE_DIGEST="$IMAGE_DIGEST" \
+ local plugin_flag=()
+ if [[ "$FULL_PIPELINE" == "retain-all" ]]; then
+ plugin_flag=(--plugin=/workspace/${CONTROLLED_PLUGIN#"$ROOT/"} --plugin-sha256="$PLUGIN_SHA" \
+ --segment-min-time-nanos="$SEGMENT_MIN_TIME_NANOS" --segment-max-time-nanos="$SEGMENT_MAX_TIME_NANOS")
+ fi
+ docker run -d --name "$container" --cpuset-cpus="$DATA_CPUS" --cpus="$DATA_CPU_LIMIT" --memory="$DATA_MEMORY" --memory-swap="$DATA_MEMORY" --pids-limit=512 \
+ -e GOMAXPROCS="$DATA_GOMAXPROCS" -e IMAGE_DIGEST="$IMAGE_DIGEST" \
-v "$ROOT":/workspace:ro -v "$run_dir":/run -w /workspace "$IMAGE" \
/workspace/.scratch/trace-pipeline-merge-performance/bin/trace-merge-benchmark serve \
--root=/run/data --socket=/run/control.sock --output=/run/report.json --profiles=/run/profiles \
@@ -119,6 +135,7 @@ run_once() {
--expected-start-time-ledger="$START_TIME_LEDGER_SHA" \
--image-digest="$IMAGE_DIGEST" --filesystem="$DATA_FILESYSTEM" --storage-device="$DATA_STORAGE_DEVICE" \
--clone-method="$DATA_CLONE_METHOD" --binary-sha256="$BINARY_SHA" \
+ "${plugin_flag[@]}" \
"${attribution_flag[@]}" >"$run_dir/container.id"
if ! docker run --rm --cpuset-cpus="$CONTROLLER_CPU" --cpus=1 --memory=1g --memory-swap=1g --pids-limit=128 \
-v "$ROOT":/workspace:ro -v "$run_dir":/run -w /workspace "$IMAGE" \
@@ -131,6 +148,7 @@ run_once() {
fi
docker logs "$container" >"$run_dir/data-node.log" 2>&1 || true
docker stop -t 20 "$container" >/dev/null
+ docker run --rm -v "$run_dir":/run "$IMAGE" chmod -R 0777 /run
active_container=""
}
@@ -162,8 +180,8 @@ run_controlled_once() {
plugin_flag=(--plugin=/workspace/${CONTROLLED_PLUGIN#"$ROOT/"})
plugin_identity_flag=(--plugin-sha256="$PLUGIN_SHA")
fi
- docker run -d --name "$container" --cpuset-cpus="$DATA_CPUS" --cpus=4 --memory=8g --memory-swap=8g --pids-limit=512 \
- -e GOMAXPROCS=4 -e IMAGE_DIGEST="$IMAGE_DIGEST" \
+ docker run -d --name "$container" --cpuset-cpus="$DATA_CPUS" --cpus="$DATA_CPU_LIMIT" --memory="$DATA_MEMORY" --memory-swap="$DATA_MEMORY" --pids-limit=512 \
+ -e GOMAXPROCS="$DATA_GOMAXPROCS" -e IMAGE_DIGEST="$IMAGE_DIGEST" \
-v "$ROOT":/workspace:ro -v "$run_dir":/run -w /workspace "$IMAGE" \
/workspace/.scratch/trace-pipeline-merge-performance/bin/trace-merge-benchmark run-controlled \
--seed-manifest=/run/seed.json --data=/run/data --output=/run/report.json \
@@ -192,9 +210,15 @@ run_controlled_once() {
fi
docker logs "$container" >"$run_dir/data-node.log" 2>&1 || true
docker rm -f "$container" >/dev/null 2>&1 || true
+ docker run --rm -v "$run_dir":/run "$IMAGE" chmod -R 0777 /run
active_container=""
}
+if [[ "$FULL_PIPELINE" != "disabled" && "$FULL_PIPELINE" != "retain-all" ]]; then
+ echo "FULL_PIPELINE must be disabled or retain-all" >&2
+ exit 1
+fi
+
if [[ -n "$PILOT_ACCELERATION" ]]; then
run_once pilot serial 1 false
echo "pilot report: $OUTPUT/pilot/report.json"
diff --git a/scripts/run-trace-merge-controlled-seed.sh b/scripts/run-trace-merge-controlled-seed.sh
index 6b19cefca..9e73ec5a3 100755
--- a/scripts/run-trace-merge-controlled-seed.sh
+++ b/scripts/run-trace-merge-controlled-seed.sh
@@ -21,10 +21,13 @@
set -euo pipefail
ROOT=$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)
-FIXTURE=${FIXTURE:-"$ROOT/.scratch/trace-pipeline-merge-performance/generated-fixture-ticket05"}
+FIXTURE=${FIXTURE:-"$ROOT/.scratch/trace-pipeline-merge-performance/generated-fixture-2x-v2-timestamps"}
OUTPUT=${OUTPUT:-"$ROOT/.scratch/trace-pipeline-merge-performance/baseline-report"}
IMAGE=${IMAGE:-golang:1.25.12}
-DATA_CPUS=${DATA_CPUS:-0-3}
+DATA_CPUS=${DATA_CPUS:-0-1}
+DATA_CPU_LIMIT=${DATA_CPU_LIMIT:-2}
+DATA_MEMORY=${DATA_MEMORY:-4g}
+DATA_GOMAXPROCS=${DATA_GOMAXPROCS:-2}
SEED_OUT=${SEED_OUT:-"$OUTPUT/controlled-seed"}
BIN="$ROOT/.scratch/trace-pipeline-merge-performance/bin/trace-merge-benchmark"
@@ -44,8 +47,8 @@ docker run --rm -v "$FIXTURE":/fixture:ro -v "$run_dir":/run "$IMAGE" bash -c \
'mkdir -p /run/source /run/data/sidx/latency /run/data/sidx/start_time; cp -a /fixture/shard/. /run/source/; chmod 0777 /run'
container="banyandb-trace-seed-capture"
docker rm -f "$container" >/dev/null 2>&1 || true
-docker run -d --name "$container" --cpuset-cpus="$DATA_CPUS" --cpus=4 --memory=8g --memory-swap=8g --pids-limit=512 \
- -e GOMAXPROCS=4 \
+docker run -d --name "$container" --cpuset-cpus="$DATA_CPUS" --cpus="$DATA_CPU_LIMIT" --memory="$DATA_MEMORY" --memory-swap="$DATA_MEMORY" --pids-limit=512 \
+ -e GOMAXPROCS="$DATA_GOMAXPROCS" \
-v "$ROOT":/workspace:ro -v "$run_dir":/run -w /workspace "$IMAGE" \
/workspace/.scratch/trace-pipeline-merge-performance/bin/trace-merge-benchmark capture-seed \
--source=/run/source --data=/run/data --schedule=/workspace/${FIXTURE#"$ROOT/"}/schedule.json \