From 546f807d36fc37544b074591a6eb509898932140 Mon Sep 17 00:00:00 2001 From: Sylwester Lachiewicz Date: Mon, 10 Aug 2026 16:03:08 +0200 Subject: [PATCH 1/3] bench: cover the Bloblang per-message evaluation path Only the parser and iterator had benchmarks, leaving Executor.Query and Overlay - the hot path for embedded compile-once/evaluate-many use - with no in-tree measurement. --- public/bloblang/executor_bench_test.go | 92 ++++++++++++++++++++++++++ 1 file changed, 92 insertions(+) create mode 100644 public/bloblang/executor_bench_test.go diff --git a/public/bloblang/executor_bench_test.go b/public/bloblang/executor_bench_test.go new file mode 100644 index 000000000..26e583994 --- /dev/null +++ b/public/bloblang/executor_bench_test.go @@ -0,0 +1,92 @@ +// Copyright 2025 Redpanda Data, Inc. + +package bloblang + +import ( + "testing" +) + +// benchValue is a nested map[string]any activation of the shape produced by decoding a JSON +// document: string keys throughout, including keys that look numeric but are map keys rather +// than array indices. +func benchValue() map[string]any { + return map[string]any{ + "rec": map[string]any{ + "983": map[string]any{"002": "BSK", "011": "A", "202": "H304", "010": "DE000X"}, + "985": map[string]any{"307": "RE"}, + "986": map[string]any{"010": "DE000X", "249": "2", "234": "x"}, + }, + "barrier": map[string]any{"SP": map[string]any{"224": "31.12.2030"}}, + "ulcount": 3, + } +} + +// Mappings covering the shapes that dominate a compile-once/evaluate-many workload: a compiled +// mapping run against millions of values via the public Executor. +var benchMappings = []struct { + name string + mapping string +}{ + {"field_read", `root = this.rec."983"."002"`}, + {"predicate", `root = this.rec."983"."002" != "FUT"`}, + {"and2", `root = this.rec."985"."307" == "RE" && this.rec."983"."202" == "H304"`}, + {"and3", `root = this.rec."986"."010" == this.rec."983"."010" && this.rec."986"."249" == "2" && this.rec."986"."234" != ""`}, + {"contains", `root = ["A","E"].contains(this.rec."983"."011")`}, + {"exists", `root = this.barrier.exists("SP")`}, + {"multi_assign", `root.a = this.rec."983"."002" +root.b = this.rec."985"."307" +root.c = this.rec."986"."249"`}, +} + +func BenchmarkExecutorQuery(b *testing.B) { + for _, bm := range benchMappings { + exe, err := GlobalEnvironment().Parse(bm.mapping) + if err != nil { + b.Fatalf("%v: %v", bm.name, err) + } + val := benchValue() + b.Run(bm.name, func(b *testing.B) { + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + if _, err := exe.Query(val); err != nil { + b.Fatal(err) + } + } + }) + } +} + +func BenchmarkExecutorOverlay(b *testing.B) { + exe, err := GlobalEnvironment().Parse(`root.a = this.rec."983"."002"`) + if err != nil { + b.Fatal(err) + } + val := benchValue() + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + onto := any(map[string]any{}) + if err := exe.Overlay(val, &onto); err != nil { + b.Fatal(err) + } + } +} + +// A mapping that does declare variables, so the vars-map allocation cannot simply be removed +// without a regression showing up here. +func BenchmarkExecutorQueryWithVars(b *testing.B) { + exe, err := GlobalEnvironment().Parse(`let t = this.rec."983"."002" +root = $t != "FUT"`) + if err != nil { + b.Fatal(err) + } + val := benchValue() + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + if _, err := exe.Query(val); err != nil { + b.Fatal(err) + } + } +} From d95ece5f94acc15a915ae0946f68a0dc4dbc5e1a Mon Sep 17 00:00:00 2001 From: Sylwester Lachiewicz Date: Mon, 10 Aug 2026 16:09:28 +0200 Subject: [PATCH 2/3] query: resolve all-map field paths without gabs Field reads went through gabs.Wrap(target).S(path...).Data(), which builds and discards a Container per segment, so allocations scaled with the number of field reads in a mapping. Walking map[string]any directly makes them constant. Arrays stay on gabs deliberately: its Search permits a "*" wildcard segment, and resolveFieldPath delegates the whole path back rather than reimplement that. --- internal/bloblang/query/functions.go | 28 +++++++++- .../bloblang/query/functions_path_test.go | 54 +++++++++++++++++++ 2 files changed, 81 insertions(+), 1 deletion(-) create mode 100644 internal/bloblang/query/functions_path_test.go diff --git a/internal/bloblang/query/functions.go b/internal/bloblang/query/functions.go index ab252839f..f6991a55d 100644 --- a/internal/bloblang/query/functions.go +++ b/internal/bloblang/query/functions.go @@ -76,7 +76,33 @@ func (f *fieldFunction) Exec(ctx FunctionContext) (any, error) { if len(f.path) == 0 { return target, nil } - return gabs.Wrap(target).S(f.path...).Data(), nil + return resolveFieldPath(target, f.path), nil +} + +// resolveFieldPath walks a path of field names, and is equivalent to +// gabs.Wrap(target).S(path...).Data() while avoiding the Container allocations that gabs makes +// per segment. +// +// Only the all-maps case is handled here, which is the overwhelmingly common one. Arrays are +// deliberately left to gabs: its Search permits a "*" wildcard segment over an array, and +// duplicating that behaviour is not worth the risk of drifting from it. Anything that is not a +// map[string]any is therefore delegated, walking again from the original target so the result is +// gabs' own. +func resolveFieldPath(target any, path []string) any { + current := target + for _, segment := range path { + obj, ok := current.(map[string]any) + if !ok { + // Not a map: hand the whole path back to gabs from the top. + return gabs.Wrap(target).S(path...).Data() + } + if current, ok = obj[segment]; !ok { + // gabs reports a missing key as an error, which Search swallows into a nil + // Container, and Data() on a nil Container is nil. + return nil + } + } + return current } func (f *fieldFunction) QueryTargets(ctx TargetsContext) (TargetsContext, []TargetPath) { diff --git a/internal/bloblang/query/functions_path_test.go b/internal/bloblang/query/functions_path_test.go new file mode 100644 index 000000000..9ea7ca348 --- /dev/null +++ b/internal/bloblang/query/functions_path_test.go @@ -0,0 +1,54 @@ +// Copyright 2025 Redpanda Data, Inc. + +package query + +import ( + "testing" + + "github.com/Jeffail/gabs/v2" + "github.com/stretchr/testify/assert" +) + +// resolveFieldPath must be indistinguishable from the gabs expression it replaces, including on +// every failure mode: missing keys, nil intermediates, scalars mid-path, arrays, negative and +// out-of-range indices, and the "*" wildcard that gabs' Search permits. +func TestResolveFieldPathMatchesGabs(t *testing.T) { + tests := []struct { + name string + target any + path []string + }{ + {"simple", map[string]any{"a": map[string]any{"b": "c"}}, []string{"a", "b"}}, + {"numeric_map_keys", map[string]any{"983": map[string]any{"002": "BSK"}}, []string{"983", "002"}}, + {"missing_leaf", map[string]any{"a": map[string]any{"b": "c"}}, []string{"a", "z"}}, + {"missing_intermediate", map[string]any{"a": map[string]any{"b": "c"}}, []string{"z", "b"}}, + {"nil_intermediate", map[string]any{"a": nil}, []string{"a", "b"}}, + {"nil_leaf", map[string]any{"a": nil}, []string{"a"}}, + {"past_scalar", map[string]any{"a": "scalar"}, []string{"a", "b"}}, + {"scalar_root", "scalar", []string{"a"}}, + {"nil_root", nil, []string{"a"}}, + {"array_index", map[string]any{"a": []any{"x", "y"}}, []string{"a", "1"}}, + {"array_index_zero", map[string]any{"a": []any{"x", "y"}}, []string{"a", "0"}}, + {"array_out_of_range", map[string]any{"a": []any{"x"}}, []string{"a", "5"}}, + {"array_negative", map[string]any{"a": []any{"x"}}, []string{"a", "-1"}}, + {"array_non_numeric", map[string]any{"a": []any{"x"}}, []string{"a", "b"}}, + {"array_wildcard", map[string]any{"a": []any{ + map[string]any{"b": 1}, map[string]any{"b": 2}, + }}, []string{"a", "*", "b"}}, + {"array_wildcard_leaf", map[string]any{"a": []any{"x", "y"}}, []string{"a", "*"}}, + {"wildcard_on_map", map[string]any{"a": map[string]any{"b": 1}}, []string{"a", "*"}}, + {"array_of_arrays", map[string]any{"a": []any{[]any{"deep"}}}, []string{"a", "0", "0"}}, + {"map_in_array", map[string]any{"a": []any{map[string]any{"b": "c"}}}, []string{"a", "0", "b"}}, + {"empty_string_key", map[string]any{"": "blank"}, []string{""}}, + {"typed_nil_map", map[string]any{"a": map[string]any(nil)}, []string{"a", "b"}}, + {"deep_all_maps", map[string]any{"a": map[string]any{"b": map[string]any{"c": map[string]any{"d": 4}}}}, []string{"a", "b", "c", "d"}}, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + want := gabs.Wrap(test.target).S(test.path...).Data() + got := resolveFieldPath(test.target, test.path) + assert.Equal(t, want, got) + }) + } +} From 795d647b655a4fc25c5f96c4d86dd76abce7677c Mon Sep 17 00:00:00 2001 From: Sylwester Lachiewicz Date: Mon, 10 Aug 2026 17:12:49 +0200 Subject: [PATCH 3/3] bloblang: skip the per-execution variables map when a mapping has none Query and Overlay allocated a variables map on every call, including for the majority of mappings that never assign a variable. AssignsVariables reports whether any statement targets one, and mappings that do not share a single empty map. Maps applied with the apply method are not consulted, because that method already replaces the variables map before invoking the target - a named map cannot write to its caller's variables. The shared map is empty rather than nil so that reading an undefined variable keeps reporting "variable 'x' undefined" rather than "variables were undefined". --- internal/bloblang/mapping/executor.go | 16 ++++ public/bloblang/executor.go | 24 ++++- public/bloblang/executor_vars_test.go | 128 ++++++++++++++++++++++++++ 3 files changed, 166 insertions(+), 2 deletions(-) create mode 100644 public/bloblang/executor_vars_test.go diff --git a/internal/bloblang/mapping/executor.go b/internal/bloblang/mapping/executor.go index e94ac2122..6399b53bf 100644 --- a/internal/bloblang/mapping/executor.go +++ b/internal/bloblang/mapping/executor.go @@ -224,6 +224,22 @@ func (e *Executor) QueryTargets(ctx query.TargetsContext) (query.TargetsContext, return ctx, paths } +// AssignsVariables returns true if any statement within the mapping assigns to +// a variable, and therefore whether executing it requires a writable variables +// map. +// +// Maps applied with the `apply` method are deliberately not consulted: that +// method isolates variables by replacing the map before invoking the target, so +// a named map cannot write to its caller's variables. +func (e *Executor) AssignsVariables() bool { + for _, t := range e.AssignmentTargets() { + if t.Type == TargetVariable { + return true + } + } + return false +} + // AssignmentTargets returns a slice of all targets assigned to by statements // within the mapping. func (e *Executor) AssignmentTargets() []TargetPath { diff --git a/public/bloblang/executor.go b/public/bloblang/executor.go index 2d3335c2d..e77daa576 100644 --- a/public/bloblang/executor.go +++ b/public/bloblang/executor.go @@ -15,15 +15,35 @@ import ( type Executor struct { exec *mapping.Executor emptyQueryMessage message.Batch + assignsVars bool } func newExecutor(exec *mapping.Executor) *Executor { return &Executor{ exec: exec, emptyQueryMessage: message.QuickBatch(nil), + assignsVars: exec.AssignsVariables(), } } +// sharedEmptyVars is handed to mappings that contain no variable assignments, +// so that the common case does not allocate a map per execution. +// +// It must never be written to. Only VarAssignment writes to a variables map, +// and it is reached only from a statement whose assignment target is a +// variable, which is precisely what AssignsVariables reports. Reads of an +// undefined variable are unaffected: an empty map yields the same "variable +// undefined" error as a freshly allocated one, whereas a nil map would report +// "variables were undefined" instead and change behaviour. +var sharedEmptyVars = map[string]any{} + +func (e *Executor) newVars() map[string]any { + if e.assignsVars { + return map[string]any{} + } + return sharedEmptyVars +} + // ErrRootDeleted is returned by a Bloblang query when the mapping results in // the root being deleted. It might be considered correct to do this in // situations where filtering is allowed or expected. @@ -40,7 +60,7 @@ var ErrRootDeleted = errors.New("root was deleted") func (e *Executor) Query(val any) (any, error) { res, err := e.exec.Exec(query.FunctionContext{ Maps: e.exec.Maps(), - Vars: map[string]any{}, + Vars: e.newVars(), Index: 0, MsgBatch: e.emptyQueryMessage, }.WithValue(val)) @@ -64,7 +84,7 @@ func (e *Executor) Query(val any) (any, error) { // ErrRootDeleted is returned, which can be used as a signal to filter rather // than fail the mapping. func (e *Executor) Overlay(val any, onto *any) error { - vars := map[string]any{} + vars := e.newVars() if err := e.exec.ExecOnto(query.FunctionContext{ Maps: e.exec.Maps(), diff --git a/public/bloblang/executor_vars_test.go b/public/bloblang/executor_vars_test.go new file mode 100644 index 000000000..90aa456f8 --- /dev/null +++ b/public/bloblang/executor_vars_test.go @@ -0,0 +1,128 @@ +// Copyright 2025 Redpanda Data, Inc. + +package bloblang + +import ( + "sync" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// Mappings without variable assignments share a single empty variables map, so anything that +// could write to it has to be proven absent rather than assumed. + +func TestSharedEmptyVarsNotWrittenByAppliedMap(t *testing.T) { + // The outer mapping assigns no variables, so it receives the shared map. The applied map + // does assign one, and must not be able to reach the caller's variables: the apply method + // replaces them first. + exe, err := GlobalEnvironment().Parse(`map inner { + let v = "written" + root = $v +} +root.out = this.a.apply("inner")`) + require.NoError(t, err) + + res, err := exe.Query(map[string]any{"a": "ignored"}) + require.NoError(t, err) + assert.Equal(t, map[string]any{"out": "written"}, res) + + assert.Empty(t, sharedEmptyVars, "applied map leaked a variable into the shared map") +} + +func TestUndefinedVariableErrorUnchanged(t *testing.T) { + // A mapping that reads a variable it never assigns still takes the shared map. An empty map + // reports the variable as undefined; a nil map would report that variables themselves were + // undefined, which would be a different error. + exe, err := GlobalEnvironment().Parse(`root = $nope`) + require.NoError(t, err) + + _, err = exe.Query(map[string]any{}) + require.Error(t, err) + assert.Contains(t, err.Error(), "variable 'nope' undefined") + assert.Empty(t, sharedEmptyVars) +} + +func TestVarAssigningMappingGetsPrivateVars(t *testing.T) { + exe, err := GlobalEnvironment().Parse(`let seen = this.v +root.v = $seen`) + require.NoError(t, err) + + for _, v := range []string{"one", "two", "three"} { + res, err := exe.Query(map[string]any{"v": v}) + require.NoError(t, err) + assert.Equal(t, map[string]any{"v": v}, res, "variables leaked between executions") + } + assert.Empty(t, sharedEmptyVars) +} + +func TestVarsFreeMappingIsConcurrencySafe(t *testing.T) { + // Run under -race: a write to the shared map from any path would be reported here. + exe, err := GlobalEnvironment().Parse(`root.a = this.x +root.b = this.y`) + require.NoError(t, err) + + var wg sync.WaitGroup + for i := 0; i < 64; i++ { + wg.Add(1) + go func() { + defer wg.Done() + for j := 0; j < 100; j++ { + res, err := exe.Query(map[string]any{"x": 1, "y": 2}) + assert.NoError(t, err) + assert.Equal(t, map[string]any{"a": 1, "b": 2}, res) + } + }() + } + wg.Wait() + assert.Empty(t, sharedEmptyVars) +} + +func TestOverlayVarsBehaviourUnchanged(t *testing.T) { + exe, err := GlobalEnvironment().Parse(`let n = this.n +root.doubled = $n * 2`) + require.NoError(t, err) + + onto := any(map[string]any{"kept": true}) + require.NoError(t, exe.Overlay(map[string]any{"n": 21}, &onto)) + assert.Equal(t, map[string]any{"kept": true, "doubled": int64(42)}, onto) + assert.Empty(t, sharedEmptyVars) +} + +// AssignsVariables drives the choice, so its verdict is asserted directly across the statement +// forms that can carry an assignment. +func TestAssignsVariablesDetection(t *testing.T) { + tests := []struct { + name string + mapping string + want bool + }{ + {"no vars", `root = this.a`, false}, + {"reads a var only", `root = $nope`, false}, + {"top level let", `let a = 1 +root = $a`, true}, + {"let inside root if", `root = if this.a == 1 { + this.b +} else { + this.c +}`, false}, + {"var assigned in an if statement", `if this.a == 1 { + let x = 1 + root.v = $x +}`, true}, + {"only the applied map assigns", `map inner { + let v = 1 + root = $v +} +root = this.apply("inner")`, false}, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + exe, err := GlobalEnvironment().Parse(test.mapping) + require.NoError(t, err) + assert.Equal(t, test.want, exe.assignsVars) + }) + } +}