Repository: xataio/pgstream
Pull request #287 (2025-04-16 10:18:07)

Merge wildcard tables in filter
This PR fixes an edge case with the filtering lists, where if both schema qualified tables and wildcard ones are present, only the schema qualified are matched, skipping events that should otherwise be processed.


Patch:
diff --git a/pkg/wal/processor/filter/wal_filter.go b/pkg/wal/processor/filter/wal_filter.go
index 94bc301..e476cf6 100644
--- a/pkg/wal/processor/filter/wal_filter.go
+++ b/pkg/wal/processor/filter/wal_filter.go
@@ -159,10 +159,16 @@ func (t schemaTableMap) containsSchemaTable(schema, table string) bool {
 
 func (t schemaTableMap) getSchemaTables(schema string) map[string]struct{} {
 	tables, found := t[schema]
-	if found {
-		return tables
+	if !found {
+		return t[wildcard]
+	}
+
+	// make sure it's merged with the wildcard schema tables if any
+	for table := range t[wildcard] {
+		tables[table] = struct{}{}
 	}
-	return t[wildcard]
+
+	return tables
 }
 
 func (t schemaTableMap) add(schema, table string) {
diff --git a/pkg/wal/processor/mocks/mock_processor.go b/pkg/wal/processor/mocks/mock_processor.go
index 5ace2c1..c844c49 100644
--- a/pkg/wal/processor/mocks/mock_processor.go
+++ b/pkg/wal/processor/mocks/mock_processor.go
@@ -10,12 +10,18 @@ import (
 
 type Processor struct {
 	ProcessWALEventFn func(ctx context.Context, walEvent *wal.Event) error
+	processCalls      uint
 }
 
 func (m *Processor) ProcessWALEvent(ctx context.Context, walEvent *wal.Event) error {
+	m.processCalls++
 	return m.ProcessWALEventFn(ctx, walEvent)
 }
 
+func (m *Processor) GetProcessCalls() uint {
+	return m.processCalls
+}
+
 func (m *Processor) Name() string {
 	return "mock"
 }


Test patch:
diff --git a/pkg/wal/processor/filter/wal_filter_test.go b/pkg/wal/processor/filter/wal_filter_test.go
index c9ff0fc..7a5b9bf 100644
--- a/pkg/wal/processor/filter/wal_filter_test.go
+++ b/pkg/wal/processor/filter/wal_filter_test.go
@@ -41,8 +41,8 @@ func Test_New(t *testing.T) {
 					"users":   {},
 					"default": {},
 				},
-				"*": map[string]struct{}{
-					"*": {},
+				wildcard: map[string]struct{}{
+					wildcard: {},
 				},
 			},
 			wantErr: nil,
@@ -127,7 +127,8 @@ func TestFilter_ProcessWALEvent(t *testing.T) {
 		excluded  schemaTableMap
 		event     *wal.Event
 
-		wantErr error
+		wantProcessCalls uint
+		wantErr          error
 	}{
 		{
 			name:  "event is nil",
@@ -140,7 +141,9 @@ func TestFilter_ProcessWALEvent(t *testing.T) {
 			},
 			included: nil,
 			excluded: nil,
-			wantErr:  nil,
+
+			wantProcessCalls: 1,
+			wantErr:          nil,
 		},
 		{
 			name:  "event matches included",
@@ -157,7 +160,53 @@ func TestFilter_ProcessWALEvent(t *testing.T) {
 				},
 			},
 			excluded: nil,
-			wantErr:  nil,
+
+			wantProcessCalls: 1,
+			wantErr:          nil,
+		},
+		{
+			name:  "event matches included *.*",
+			event: testEvent,
+			processor: &mocks.Processor{
+				ProcessWALEventFn: func(ctx context.Context, walEvent *wal.Event) error {
+					require.Equal(t, testEvent, walEvent)
+					return nil
+				},
+			},
+			included: schemaTableMap{
+				testSchema: {
+					"blah": {},
+				},
+				wildcard: {
+					wildcard: {},
+				},
+			},
+			excluded: nil,
+
+			wantProcessCalls: 1,
+			wantErr:          nil,
+		},
+		{
+			name:  "event matches included *.table",
+			event: testEvent,
+			processor: &mocks.Processor{
+				ProcessWALEventFn: func(ctx context.Context, walEvent *wal.Event) error {
+					require.Equal(t, testEvent, walEvent)
+					return nil
+				},
+			},
+			included: schemaTableMap{
+				testSchema: {
+					"blah": {},
+				},
+				wildcard: {
+					testTable: {},
+				},
+			},
+			excluded: nil,
+
+			wantProcessCalls: 1,
+			wantErr:          nil,
 		},
 		{
 			name:  "event does not match included",
@@ -173,7 +222,9 @@ func TestFilter_ProcessWALEvent(t *testing.T) {
 				},
 			},
 			excluded: nil,
-			wantErr:  nil,
+
+			wantProcessCalls: 0,
+			wantErr:          nil,
 		},
 		{
 			name:  "event matches excluded",
@@ -189,7 +240,9 @@ func TestFilter_ProcessWALEvent(t *testing.T) {
 					testTable: struct{}{},
 				},
 			},
-			wantErr: nil,
+
+			wantProcessCalls: 0,
+			wantErr:          nil,
 		},
 		{
 			name:  "event does not match excluded",
@@ -206,7 +259,9 @@ func TestFilter_ProcessWALEvent(t *testing.T) {
 					"orders": struct{}{},
 				},
 			},
-			wantErr: nil,
+
+			wantProcessCalls: 1,
+			wantErr:          nil,
 		},
 	}
 
@@ -223,6 +278,7 @@ func TestFilter_ProcessWALEvent(t *testing.T) {
 
 			err := f.ProcessWALEvent(ctx, tc.event)
 			require.ErrorIs(t, err, tc.wantErr)
+			require.Equal(t, tc.wantProcessCalls, tc.processor.GetProcessCalls())
 		})
 	}
 }