diff --git a/v1/storage/inmem/inmem.go b/v1/storage/inmem/inmem.go index cdc43424dd..9fa145a051 100644 --- a/v1/storage/inmem/inmem.go +++ b/v1/storage/inmem/inmem.go @@ -349,10 +349,25 @@ func (h *handle) Unregister(_ context.Context, txn storage.Transaction) { } func (db *store) runOnCommitTriggers(ctx context.Context, txn storage.Transaction, event storage.TriggerEvent) { - if db.returnASTValuesOnRead && len(db.triggers) > 0 { - // FIXME: Not very performant for large data. + // While it's unlikely, the API allows one trigger to be configured to want + // data conversion, and another that doesn't. So let's handle that properly. + var wantsDataConversion bool + if db.returnASTValuesOnRead && len(event.Data) > 0 { + for _, t := range db.triggers { + if !t.SkipDataConversion { + wantsDataConversion = true + break + } + } + } - dataEvents := make([]storage.DataEvent, 0, len(event.Data)) + var converted storage.TriggerEvent + if wantsDataConversion { + converted = storage.TriggerEvent{ + Policy: event.Policy, + Data: make([]storage.DataEvent, 0, len(event.Data)), + Context: event.Context, + } for _, dataEvent := range event.Data { if astData, ok := dataEvent.Data.(ast.Value); ok { @@ -360,25 +375,21 @@ func (db *store) runOnCommitTriggers(ctx context.Context, txn storage.Transactio if err != nil { panic(err) } - dataEvents = append(dataEvents, storage.DataEvent{ + converted.Data = append(converted.Data, storage.DataEvent{ Path: dataEvent.Path, Data: jsn, Removed: dataEvent.Removed, }) - } else { - dataEvents = append(dataEvents, dataEvent) } } - - event = storage.TriggerEvent{ - Policy: event.Policy, - Data: dataEvents, - Context: event.Context, - } } for _, t := range db.triggers { - t.OnCommit(ctx, txn, event) + if wantsDataConversion && !t.SkipDataConversion { + t.OnCommit(ctx, txn, converted) + } else { + t.OnCommit(ctx, txn, event) + } } } diff --git a/v1/storage/inmem/inmem_bench_test.go b/v1/storage/inmem/inmem_bench_test.go index c2872c3a05..7c43c412f7 100644 --- a/v1/storage/inmem/inmem_bench_test.go +++ b/v1/storage/inmem/inmem_bench_test.go @@ -211,8 +211,8 @@ func BenchmarkWriteAndCommit(b *testing.B) { AllStores(map[string]any{}).Bench(b, operation) } -// Go 48750 ns/op 27040 B/op 311 allocs/op (no additional cost of triggers) -// AST 191501 ns/op 37585 B/op 616 allocs/op (extra cost du to converting back to Go values.. why?) +// Go 48399 ns/op 16209 B/op 304 allocs/op (no additional cost of triggers) +// AST 191501 ns/op 26673 B/op 605 allocs/op (extra cost due to converting back to Go values) func BenchmarkWriteAndCommitWithTriggers(b *testing.B) { paths := make([]storage.Path, 100) for i := range 100 { @@ -239,6 +239,62 @@ func BenchmarkWriteAndCommitWithTriggers(b *testing.B) { Bench(b, operation) } +// AST 189485 ns/op 17011 B/op 304 allocs/op +func BenchmarkWriteAndCommitWithTriggersSkipConversion(b *testing.B) { + paths := make([]storage.Path, 100) + for i := range 100 { + paths[i] = storage.Path{strconv.Itoa(i)} + } + values := make([]ast.Value, 100) + for i := range 100 { + values[i] = ast.String(paths[i][0]) + } + + operation := func(ctx context.Context, target *target) error { + txn, _ := target.store.NewTransaction(b.Context(), storage.WriteParams) + for i := range 100 { + if err := target.store.Write(b.Context(), txn, storage.AddOp, paths[i], values[i]); err != nil { + return err + } + } + + return target.store.Commit(b.Context(), txn) + } + + triggerCount := 0 + + trigger := storage.TriggerConfig{ + SkipDataConversion: true, + OnCommit: func(ctx context.Context, txn storage.Transaction, event storage.TriggerEvent) { + if event.DataChanged() { + if len(event.Data) != 100 { + b.Fatalf("Expected 100 data changes but got: %d", len(event.Data)) + } + if _, ok := event.Data[0].Data.(ast.Value); !ok { + b.Fatalf("Expected ast.Value data but got: %T", event.Data[0].Data) + } + triggerCount++ + } + }, + } + + onlyAstStores := targets{{ + name: "AST", + store: inmem.NewFromObjectWithOpts(map[string]any{}, inmem.OptReturnASTValuesOnRead(true)), + isAST: true, + }} + + onlyAstStores. + SetupWithTxn(b, writeTxn, func(ctx context.Context, target *target) error { + return onlyError(target.store.Register(b.Context(), target.txn, trigger)) + }). + Bench(b, operation) + + if triggerCount == 0 { + b.Fatalf("Expected trigger to be called at least once") + } +} + func (t targets) VerifyRead(b *testing.B, path storage.Path, expected any) targets { b.Helper() for _, target := range t { diff --git a/v1/storage/inmem/inmem_test.go b/v1/storage/inmem/inmem_test.go index c4134c3d59..c8ec9a9e6f 100644 --- a/v1/storage/inmem/inmem_test.go +++ b/v1/storage/inmem/inmem_test.go @@ -1150,6 +1150,66 @@ func TestInMemoryTriggers(t *testing.T) { } } +func TestASTInMemoryTriggersDataConversion(t *testing.T) { + ctx := t.Context() + store := NewFromObjectWithOpts(loadSmallTestData(), OptReturnASTValuesOnRead(true)) + + noConversionTriggerCount := 0 + conversionTriggerCount := 0 + + // Register a trigger that doesn't want data conversion + err := storage.Txn(t.Context(), store, storage.WriteParams, func(txn storage.Transaction) error { + _, err := store.Register(ctx, txn, storage.TriggerConfig{ + SkipDataConversion: true, + OnCommit: func(ctx context.Context, txn storage.Transaction, event storage.TriggerEvent) { + if event.DataChanged() { + if _, ok := event.Data[0].Data.(ast.Value); !ok { + t.Fatalf("Expected ast.Value data but got: %T", event.Data[0].Data) + } + noConversionTriggerCount++ + } + }, + }) + return err + }) + if err != nil { + t.Fatalf("Failed to register trigger: %v", err) + } + + // Register a trigger that wants data conversion (skip data conversion not set, as is the default) + err = storage.Txn(t.Context(), store, storage.WriteParams, func(txn storage.Transaction) error { + _, err := store.Register(ctx, txn, storage.TriggerConfig{ + OnCommit: func(ctx context.Context, txn storage.Transaction, event storage.TriggerEvent) { + if event.DataChanged() { + if _, ok := event.Data[0].Data.(ast.Value); ok { + t.Fatalf("Expected non-ast.Value data but got: %T", event.Data[0].Data) + } + conversionTriggerCount++ + } + }, + }) + return err + }) + if err != nil { + t.Fatalf("Failed to register trigger: %v", err) + } + + err = storage.Txn(t.Context(), store, storage.WriteParams, func(txn storage.Transaction) error { + return store.Write(ctx, txn, storage.ReplaceOp, storage.MustParsePath("/a"), "hello") + }) + if err != nil { + t.Fatalf("Failed to write data: %v", err) + } + + if noConversionTriggerCount != 1 { + t.Fatalf("Expected no conversion trigger to be called once but got: %d", noConversionTriggerCount) + } + + if conversionTriggerCount != 1 { + t.Fatalf("Expected conversion trigger to be called once but got: %d", conversionTriggerCount) + } +} + func TestInMemoryTriggersUnregister(t *testing.T) { ctx := t.Context() store := NewFromObject(loadSmallTestData()) diff --git a/v1/storage/interface.go b/v1/storage/interface.go index a783caae09..8407c878c2 100644 --- a/v1/storage/interface.go +++ b/v1/storage/interface.go @@ -210,6 +210,10 @@ func (e TriggerEvent) DataChanged() bool { // TriggerConfig contains the trigger registration configuration. type TriggerConfig struct { + // SkipDataConversion when set to true, avoids converting data passed to + // trigger functions from the store to Go types, and instead passes the + // original representation (e.g., ast.Value). + SkipDataConversion bool // OnCommit is invoked when a transaction is successfully committed. The // callback is invoked with a handle to the write transaction that