Call WalDataToDDLEvent once per event in the processor chain

Author: kvch

pkg/snapshot/generator/postgres/data/pg_snapshot_generator_test.go

@@ -1178,6 +1178,7 @@ func TestSnapshotGenerator_CreateSnapshot(t *testing.T) {
 			}
 			diff := cmp.Diff(events, tc.wantEvents,
 				cmpopts.IgnoreFields(wal.Data{}, "Timestamp"),
+				cmpopts.IgnoreUnexported(wal.Data{}),
 				cmpopts.SortSlices(func(a, b *wal.Event) bool { return a.Data.Table < b.Data.Table }))
 			require.Empty(t, diff, fmt.Sprintf("got: \n%v, \nwant \n%v, \ndiff: \n%s", events, tc.wantEvents, diff))
 		})
@@ -1701,6 +1702,7 @@ func TestSnapshotGenerator_snapshotTableRange(t *testing.T) {
 			}
 			diff := cmp.Diff(events, tc.wantEvents,
 				cmpopts.IgnoreFields(wal.Data{}, "Timestamp"),
+				cmpopts.IgnoreUnexported(wal.Data{}),
 				cmpopts.SortSlices(func(a, b *wal.Event) bool { return a.Data.Table < b.Data.Table }))
 			require.Empty(t, diff, fmt.Sprintf("got: \n%v, \nwant \n%v, \ndiff: \n%s", events, tc.wantEvents, diff))
 		})

pkg/wal/listener/kafka/wal_kafka_reader_test.go

@@ -44,7 +44,12 @@ func TestReader_Listen(t *testing.T) {
 		require.Equal(t, []byte("test-value"), b)
 		data, ok := a.(*wal.Data)
 		require.True(t, ok)
-		*data = *testWalEvent.Data
+		// assign fields individually rather than copying the struct by value:
+		// wal.Data holds a sync.Once for DDL-event memoisation and must not be
+		// copied (this mirrors how the real deserialiser populates fields).
+		data.Action = testWalEvent.Data.Action
+		data.Schema = testWalEvent.Data.Schema
+		data.Table = testWalEvent.Data.Table
 		return nil
 	}
 

pkg/wal/processor/filter/wal_filter_test.go

@@ -200,11 +200,9 @@ func TestFilter_ProcessWALEvent(t *testing.T) {
 			},
 			processor: &mocks.Processor{
 				ProcessWALEventFn: func(ctx context.Context, walEvent *wal.Event) error {
-					require.Equal(t, &wal.Data{
-						Action:  wal.LogicalMessageAction,
-						Prefix:  wal.DDLPrefix,
-						Content: string(ddlEventBytes),
-					}, walEvent.Data)
+					require.Equal(t, wal.LogicalMessageAction, walEvent.Data.Action)
+					require.Equal(t, wal.DDLPrefix, walEvent.Data.Prefix)
+					require.Equal(t, string(ddlEventBytes), walEvent.Data.Content)
 					return nil
 				},
 			},

pkg/wal/wal_data.go

@@ -5,6 +5,7 @@ package wal
 import (
 	"errors"
 	"slices"
+	"sync"
 	"time"
 
 	"github.com/rs/xid"
@@ -33,6 +34,10 @@ type Data struct {
 	Content string `json:"content,omitempty"`
 	// pgstream specific metadata
 	Metadata Metadata `json:"metadata"`
+
+	ddlEventOnce sync.Once
+	ddlEvent     *DDLEvent
+	ddlEventErr  error
 }
 
 // Metadata is pgstream specific properties to help identify the id/version

pkg/wal/wal_ddl.go

@@ -55,8 +55,16 @@ func (d *Data) IsDDLEvent() bool {
 	return d.Action == LogicalMessageAction && d.Prefix == DDLPrefix
 }
 
-// WalDataToDDLEvent parses the wal data content field as a DDL event
+// WalDataToDDLEvent parses the wal data content field as a DDL event.
+// The parsed result (or error) is cached on the Data value.
 func WalDataToDDLEvent(d *Data) (*DDLEvent, error) {
+	d.ddlEventOnce.Do(func() {
+		d.ddlEvent, d.ddlEventErr = parseDDLEvent(d)
+	})
+	return d.ddlEvent, d.ddlEventErr
+}
+
+func parseDDLEvent(d *Data) (*DDLEvent, error) {
 	if !d.IsDDLEvent() {
 		return nil, fmt.Errorf("%w: action=%s, prefix=%s", ErrNotDDLEvent, d.Action, d.Prefix)
 	}

pkg/wal/wal_ddl_test.go

@@ -5,9 +5,99 @@ package wal
 import (
 	"testing"
 
+	jsonlib "github.com/xataio/pgstream/internal/json"
+
 	"github.com/stretchr/testify/require"
 )
 
+const testDDLContent = `{
+	"ddl": "CREATE TABLE users (id SERIAL PRIMARY KEY, name TEXT)",
+	"schema_name": "public",
+	"command_tag": "CREATE TABLE",
+	"objects": [{
+		"type": "table",
+		"identity": "public.users",
+		"schema": "public",
+		"oid": "16384",
+		"pgstream_id": "ck7s8u4000001"
+	}]
+}`
+
+func TestWalDataToDDLEvent_memoisation(t *testing.T) {
+	t.Parallel()
+
+	data := &Data{
+		Action:  "M",
+		Prefix:  "pgstream.ddl",
+		Content: testDDLContent,
+	}
+
+	first, err := WalDataToDDLEvent(data)
+	require.NoError(t, err)
+	require.NotNil(t, first)
+
+	second, err := WalDataToDDLEvent(data)
+	require.NoError(t, err)
+
+	// repeat calls must return the exact same cached pointer, i.e. the content
+	// is parsed only once.
+	require.Same(t, first, second)
+
+	// the error path is memoised too.
+	notDDL := &Data{Action: "I"}
+	_, err1 := WalDataToDDLEvent(notDDL)
+	_, err2 := WalDataToDDLEvent(notDDL)
+	require.ErrorIs(t, err1, ErrNotDDLEvent)
+	require.ErrorIs(t, err2, ErrNotDDLEvent)
+}
+
+func BenchmarkWalDataToDDLEvent_memoised(b *testing.B) {
+	data := &Data{
+		Action:  "M",
+		Prefix:  "pgstream.ddl",
+		Content: testDDLContent,
+	}
+	// prime the memo so the benchmarked calls only hit the cached fast path.
+	if _, err := WalDataToDDLEvent(data); err != nil {
+		b.Fatal(err)
+	}
+
+	b.ReportAllocs()
+	for i := 0; i < b.N; i++ {
+		if _, err := WalDataToDDLEvent(data); err != nil {
+			b.Fatal(err)
+		}
+	}
+}
+
+func TestWalDataToDDLEvent_serialisationUnaffected(t *testing.T) {
+	t.Parallel()
+
+	// a Data whose DDL event has been memoised must serialise byte-identically
+	// to one that has not: the memo fields are unexported and excluded from JSON.
+	plain := &Data{
+		Action:  "M",
+		Prefix:  "pgstream.ddl",
+		Content: testDDLContent,
+		Schema:  "public",
+	}
+	memoised := &Data{
+		Action:  "M",
+		Prefix:  "pgstream.ddl",
+		Content: testDDLContent,
+		Schema:  "public",
+	}
+	_, err := WalDataToDDLEvent(memoised)
+	require.NoError(t, err)
+
+	plainBytes, err := jsonlib.Marshal(plain)
+	require.NoError(t, err)
+	memoisedBytes, err := jsonlib.Marshal(memoised)
+	require.NoError(t, err)
+
+	require.Equal(t, plainBytes, memoisedBytes)
+}
+
 func TestData_IsDDLEvent(t *testing.T) {
 	t.Parallel()