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/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) + }) + } +} 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_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) + } + } +} 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) + }) + } +}