inmem: allow passing triggers (AST) data without conversion (#7959)

Fixes #7958

Signed-off-by: Anders Eknert <[email protected]>

Author: anderseknert

v1/storage/inmem/inmem.go

@@ -349,36 +349,47 @@ 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 {
 				jsn, err := ast.ValueToInterface(astData, illegalResolver{})
 				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)
+		}
 	}
 }
 

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 {

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())

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