diff --git a/internal/reference/reference_bench_test.go b/internal/reference/reference_bench_test.go new file mode 100644 index 0000000..2faa83f --- /dev/null +++ b/internal/reference/reference_bench_test.go @@ -0,0 +1,266 @@ +// Copyright © 2026 Meroxa, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package reference + +import ( + "testing" + + "github.com/conduitio/conduit-commons/opencdc" +) + +// benchRecordRaw and benchRecordStructured mirror the shapes processors see in +// production: a handful of metadata entries plus a raw or structured payload. +// They are used as the fixed starting point for every benchmark iteration so +// the measured cost is only reference resolution/get/set, not record setup. + +func benchRecordRaw() opencdc.Record { + return opencdc.Record{ + Position: opencdc.Position("bench-position-0000000000"), + Operation: opencdc.OperationUpdate, + Metadata: opencdc.Metadata{ + "collection": "orders", + "version": "v1", + "opencdc.collection": "orders", // dotted key, only reachable via bracket syntax + }, + Key: opencdc.RawData("order-12345"), + Payload: opencdc.Change{ + Before: opencdc.RawData(`{"id":12345,"status":"pending"}`), + After: opencdc.RawData(`{"id":12345,"status":"shipped"}`), + }, + } +} + +func benchRecordStructured() opencdc.Record { + return opencdc.Record{ + Position: opencdc.Position("bench-position-0000000000"), + Operation: opencdc.OperationUpdate, + Metadata: opencdc.Metadata{ + "collection": "orders", + "version": "v1", + "opencdc.collection": "orders", // dotted key, only reachable via bracket syntax + }, + Key: opencdc.StructuredData{ + "id": 12345, + }, + Payload: opencdc.Change{ + Before: opencdc.StructuredData{ + "id": 12345, + "status": "pending", + "customer": map[string]any{ + "id": 987, + "name": "Jane Doe", + }, + }, + After: opencdc.StructuredData{ + "id": 12345, + "status": "shipped", + "customer": map[string]any{ + "id": 987, + "name": "Jane Doe", + }, + }, + }, + } +} + +// BenchmarkNewResolver measures the one-time cost of parsing a reference +// string into a Resolver (lexing + validation walk). Processors typically pay +// this cost once at Configure time, not per record, but it is included as a +// baseline to contrast against the per-record Resolve/Get/Set cost below. +func BenchmarkNewResolver(b *testing.B) { + inputs := []string{ + ".Key", + ".Metadata.collection", + `.Metadata["opencdc.collection"]`, + ".Payload.After.status", + ".Payload.After.customer.name", + } + + for _, input := range inputs { + b.Run(input, func(b *testing.B) { + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + _, err := NewResolver(input) + if err != nil { + b.Fatal(err) + } + } + }) + } +} + +// BenchmarkResolver_Resolve measures the per-record cost of resolving an +// already-parsed reference against a record and reading the field's value. +// This is the hot path: a processor resolves the same reference for every +// record it processes. +// +// References to a whole top-level field (.Key, .Payload.After) are valid +// against both raw and structured payloads, so those run against both record +// shapes. References into a structured sub-field (e.g. +// .Payload.After.status) only resolve against a structured payload -- a +// processor configured that way against a raw-payload record would get a +// resolve error on every record, which is exercised separately. +func BenchmarkResolver_Resolve(b *testing.B) { + wholeFieldCases := []struct { + name string + ref string + }{ + {"Key", ".Key"}, + {"Metadata/Field", ".Metadata.collection"}, + {"Metadata/BracketField", `.Metadata["opencdc.collection"]`}, + {"Payload.After", ".Payload.After"}, + } + + for _, bc := range wholeFieldCases { + resolver, err := NewResolver(bc.ref) + if err != nil { + b.Fatal(err) + } + + b.Run("Raw/"+bc.name, func(b *testing.B) { + rec := benchRecordRaw() + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + ref, err := resolver.Resolve(&rec) + if err != nil { + b.Fatal(err) + } + _ = ref.Get() + } + }) + + b.Run("Structured/"+bc.name, func(b *testing.B) { + rec := benchRecordStructured() + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + ref, err := resolver.Resolve(&rec) + if err != nil { + b.Fatal(err) + } + _ = ref.Get() + } + }) + } + + structuredFieldCases := []struct { + name string + ref string + }{ + {"Payload.After/StructuredField", ".Payload.After.status"}, + {"Payload.After/NestedStructuredField", ".Payload.After.customer.name"}, + } + + for _, bc := range structuredFieldCases { + resolver, err := NewResolver(bc.ref) + if err != nil { + b.Fatal(err) + } + + b.Run("Structured/"+bc.name, func(b *testing.B) { + rec := benchRecordStructured() + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + ref, err := resolver.Resolve(&rec) + if err != nil { + b.Fatal(err) + } + _ = ref.Get() + } + }) + } +} + +// BenchmarkResolver_ResolveAndSet measures resolving a reference and writing a +// new value through it, the pattern used by field-set/field-rename style +// processors on every record. +func BenchmarkResolver_ResolveAndSet(b *testing.B) { + metadataCase := struct { + name string + ref string + val any + }{"Metadata/Field", ".Metadata.processed_by", "bench-processor"} + + { + resolver, err := NewResolver(metadataCase.ref) + if err != nil { + b.Fatal(err) + } + + b.Run("Raw/"+metadataCase.name, func(b *testing.B) { + rec := benchRecordRaw() + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + ref, err := resolver.Resolve(&rec) + if err != nil { + b.Fatal(err) + } + if err := ref.Set(metadataCase.val); err != nil { + b.Fatal(err) + } + } + }) + + b.Run("Structured/"+metadataCase.name, func(b *testing.B) { + rec := benchRecordStructured() + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + ref, err := resolver.Resolve(&rec) + if err != nil { + b.Fatal(err) + } + if err := ref.Set(metadataCase.val); err != nil { + b.Fatal(err) + } + } + }) + } + + structuredFieldCases := []struct { + name string + ref string + val any + }{ + {"Payload.After/StructuredField", ".Payload.After.status", "processed"}, + {"Payload.After/NestedStructuredField", ".Payload.After.customer.name", "John Smith"}, + } + + for _, bc := range structuredFieldCases { + resolver, err := NewResolver(bc.ref) + if err != nil { + b.Fatal(err) + } + + b.Run("Structured/"+bc.name, func(b *testing.B) { + rec := benchRecordStructured() + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + ref, err := resolver.Resolve(&rec) + if err != nil { + b.Fatal(err) + } + if err := ref.Set(bc.val); err != nil { + b.Fatal(err) + } + } + }) + } +} diff --git a/processor_func_bench_test.go b/processor_func_bench_test.go new file mode 100644 index 0000000..48b886e --- /dev/null +++ b/processor_func_bench_test.go @@ -0,0 +1,178 @@ +// Copyright © 2026 Meroxa, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package sdk + +import ( + "context" + "fmt" + "testing" + + "github.com/conduitio/conduit-commons/opencdc" +) + +// benchBatchSize is the number of records processed per Process call in the +// benchmarks below. It is representative of a typical pipeline batch, not a +// single record; ns/op and allocs/op should be divided by this constant to +// get a per-record figure. +const benchBatchSize = 100 + +func benchBatchRaw(n int) []opencdc.Record { + records := make([]opencdc.Record, n) + for i := range records { + records[i] = opencdc.Record{ + Position: opencdc.Position("bench-position"), + Operation: opencdc.OperationUpdate, + Metadata: opencdc.Metadata{ + "collection": "orders", + "version": "v1", + }, + Key: opencdc.RawData("order-12345"), + Payload: opencdc.Change{ + Before: opencdc.RawData(`{"id":12345,"status":"pending"}`), + After: opencdc.RawData(`{"id":12345,"status":"shipped"}`), + }, + } + } + return records +} + +func benchBatchStructured(n int) []opencdc.Record { + records := make([]opencdc.Record, n) + for i := range records { + records[i] = opencdc.Record{ + Position: opencdc.Position("bench-position"), + Operation: opencdc.OperationUpdate, + Metadata: opencdc.Metadata{ + "collection": "orders", + "version": "v1", + }, + Key: opencdc.StructuredData{ + "id": 12345, + }, + Payload: opencdc.Change{ + Before: opencdc.StructuredData{ + "id": 12345, + "status": "pending", + "customer": map[string]any{ + "id": 987, + "name": "Jane Doe", + }, + }, + After: opencdc.StructuredData{ + "id": 12345, + "status": "shipped", + "customer": map[string]any{ + "id": 987, + "name": "Jane Doe", + }, + }, + }, + } + } + return records +} + +// BenchmarkProcessorFunc_Process_MetadataFieldSet benchmarks a representative +// field-set style processor (built with the ProcessorFunc adapter, the SDK's +// concrete, reusable Processor implementation) that resolves a reference once +// per record and sets a metadata field on it. This is the same +// resolve-then-set pattern used by real field-set/field-rename processors, +// exercised through the full Processor.Process call including the +// ProcessedRecord wrapping. The reference works against both raw and +// structured payloads since it targets metadata, not the payload itself. +func BenchmarkProcessorFunc_Process_MetadataFieldSet(b *testing.B) { + resolver, err := NewReferenceResolver(".Metadata.processed_by") + if err != nil { + b.Fatal(err) + } + + proc := NewProcessorFunc( + Specification{Name: "bench-metadata-field-set"}, + func(_ context.Context, record opencdc.Record) (opencdc.Record, error) { + ref, err := resolver.Resolve(&record) + if err != nil { + return record, fmt.Errorf("failed to resolve reference: %w", err) + } + if err := ref.Set("bench-processor"); err != nil { + return record, fmt.Errorf("failed to set field: %w", err) + } + return record, nil + }, + ) + + ctx := context.Background() + + b.Run("Raw", func(b *testing.B) { + records := benchBatchRaw(benchBatchSize) + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + out := proc.Process(ctx, records) + if len(out) != benchBatchSize { + b.Fatalf("expected %d processed records, got %d", benchBatchSize, len(out)) + } + } + }) + + b.Run("Structured", func(b *testing.B) { + records := benchBatchStructured(benchBatchSize) + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + out := proc.Process(ctx, records) + if len(out) != benchBatchSize { + b.Fatalf("expected %d processed records, got %d", benchBatchSize, len(out)) + } + } + }) +} + +// BenchmarkProcessorFunc_Process_StructuredFieldSet benchmarks a processor +// that resolves a nested structured-payload reference and overwrites its +// value on every record in the batch, the pattern used by processors that +// rewrite a specific field deep in a record's payload (e.g. masking, +// normalization, enrichment). +func BenchmarkProcessorFunc_Process_StructuredFieldSet(b *testing.B) { + resolver, err := NewReferenceResolver(".Payload.After.customer.name") + if err != nil { + b.Fatal(err) + } + + proc := NewProcessorFunc( + Specification{Name: "bench-structured-field-set"}, + func(_ context.Context, record opencdc.Record) (opencdc.Record, error) { + ref, err := resolver.Resolve(&record) + if err != nil { + return record, fmt.Errorf("failed to resolve reference: %w", err) + } + if err := ref.Set("REDACTED"); err != nil { + return record, fmt.Errorf("failed to set field: %w", err) + } + return record, nil + }, + ) + + ctx := context.Background() + records := benchBatchStructured(benchBatchSize) + + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + out := proc.Process(ctx, records) + if len(out) != benchBatchSize { + b.Fatalf("expected %d processed records, got %d", benchBatchSize, len(out)) + } + } +}