Repository: xataio/pgstream
Pull request #102 (2024-12-10 14:31:16)

Add snapshot generator recorder
This PR adds a decorator around a snapshot generator to record the request activity. It checks if there are any existing completed snapshots for the snapshot request, and parses the result to make sure if there are existing snapshots that completed with errors, only the failed tables are retried. 

It helps keep track of when a snapshot was requested and completed, and any potential errors during the processing. This will prevent the configured initial snapshot to be re-run every time pgstream is started.


Patch:
diff --git a/pkg/snapshot/errors.go b/pkg/snapshot/errors.go
index 166ac45..3d7db79 100644
--- a/pkg/snapshot/errors.go
+++ b/pkg/snapshot/errors.go
@@ -30,6 +30,33 @@ func (e *Errors) Error() string {
 	return err.Error()
 }
 
+func NewErrors(err error) *Errors {
+	if err == nil {
+		return nil
+	}
+	var snapshotErrs *Errors
+	if errors.As(err, &snapshotErrs) {
+		return snapshotErrs
+	}
+	return &Errors{
+		Snapshot: err,
+	}
+}
+
+func (e *Errors) AddSnapshotError(err error) {
+	if e == nil {
+		e = &Errors{}
+		e.AddSnapshotError(err)
+		return
+	}
+
+	if e.Snapshot == nil {
+		e.Snapshot = err
+		return
+	}
+	e.Snapshot = errors.Join(e.Snapshot, err)
+}
+
 func (e *Errors) IsTableError(table string) bool {
 	if e == nil {
 		return false
diff --git a/pkg/snapshot/generator/snapshot_generator_recorder.go b/pkg/snapshot/generator/snapshot_generator_recorder.go
new file mode 100644
index 0000000..43dbb2a
--- /dev/null
+++ b/pkg/snapshot/generator/snapshot_generator_recorder.go
@@ -0,0 +1,129 @@
+// SPDX-License-Identifier: Apache-2.0
+
+package generator
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/xataio/pgstream/pkg/snapshot"
+	snapshotstore "github.com/xataio/pgstream/pkg/snapshot/store"
+)
+
+// SnapshotRecorder is a decorator around a snapshot generator that will record
+// the snapshot request status.
+type SnapshotRecorder struct {
+	wrapped SnapshotGenerator
+	store   snapshotstore.Store
+}
+
+const updateTimeout = time.Minute
+
+// NewSnapshotRecorder will return the generator on input wrapped with an
+// activity recorder that will keep track of the status of the snapshot
+// requests.
+func NewSnapshotRecorder(store snapshotstore.Store, generator SnapshotGenerator) *SnapshotRecorder {
+	return &SnapshotRecorder{
+		wrapped: generator,
+		store:   store,
+	}
+}
+
+func (s *SnapshotRecorder) CreateSnapshot(ctx context.Context, ss *snapshot.Snapshot) error {
+	filteredTables, err := s.filterOutExistingSnapshots(ctx, ss.SchemaName, ss.TableNames)
+	if err != nil {
+		return snapshot.NewErrors(err)
+	}
+	// no tables to snapshot
+	if len(filteredTables) == 0 {
+		return nil
+	}
+
+	ss.TableNames = filteredTables
+	req := &snapshot.Request{
+		Snapshot: *ss,
+	}
+
+	if err := s.markSnapshotInProgress(ctx, req); err != nil {
+		return snapshot.NewErrors(err)
+	}
+
+	err = s.wrapped.CreateSnapshot(ctx, ss)
+
+	return s.markSnapshotCompleted(req, err)
+}
+
+func (s *SnapshotRecorder) Close() error {
+	s.store.Close()
+	return s.wrapped.Close()
+}
+
+func (s *SnapshotRecorder) markSnapshotInProgress(ctx context.Context, req *snapshot.Request) error {
+	if err := s.store.CreateSnapshotRequest(ctx, req); err != nil {
+		return err
+	}
+	// the snapshot will start immediately
+	req.MarkInProgress()
+	return s.store.UpdateSnapshotRequest(ctx, req)
+}
+
+func (s *SnapshotRecorder) markSnapshotCompleted(req *snapshot.Request, err error) error {
+	// make sure we can update the request status in the store regardless of
+	// context cancelations
+	ctx, cancel := context.WithTimeout(context.Background(), updateTimeout)
+	defer cancel()
+
+	req.MarkCompleted(err)
+	if updateErr := s.store.UpdateSnapshotRequest(ctx, req); updateErr != nil {
+		if err == nil {
+			return snapshot.NewErrors(updateErr)
+		}
+		snapshotErr := snapshot.NewErrors(err)
+		snapshotErr.AddSnapshotError(updateErr)
+		return snapshotErr
+	}
+	return err
+}
+
+func (s *SnapshotRecorder) filterOutExistingSnapshots(ctx context.Context, schema string, tables []string) ([]string, error) {
+	snapshotRequests, err := s.store.GetSnapshotRequestsBySchema(ctx, schema)
+	if err != nil {
+		return nil, fmt.Errorf("retrieving existing snapshots for schema: %w", err)
+	}
+
+	if len(snapshotRequests) == 0 {
+		return tables, nil
+	}
+
+	existingRequests := map[string]*snapshot.Request{}
+	for _, req := range snapshotRequests {
+		for _, table := range req.Snapshot.TableNames {
+			existingRequests[table] = req
+		}
+	}
+
+	const wildcard = "*"
+	filteredTables := make([]string, 0, len(tables))
+	for _, table := range tables {
+		wildCardReq, wildcardFound := existingRequests[wildcard]
+
+		switch table {
+		case wildcard:
+			if wildcardFound {
+				filteredTables = append(filteredTables, wildCardReq.Errors.GetFailedTables()...)
+				break
+			}
+			filteredTables = append(filteredTables, table)
+		default:
+			tableReq, tableFound := existingRequests[table]
+
+			if (tableFound && tableReq.HasFailedForTable(table)) ||
+				(wildcardFound && wildCardReq.HasFailedForTable(table)) ||
+				(!tableFound && !wildcardFound) {
+				filteredTables = append(filteredTables, table)
+			}
+		}
+	}
+	return filteredTables, nil
+}
diff --git a/pkg/snapshot/snapshot.go b/pkg/snapshot/snapshot.go
index 1789830..7766647 100644
--- a/pkg/snapshot/snapshot.go
+++ b/pkg/snapshot/snapshot.go
@@ -4,7 +4,6 @@ package snapshot
 
 import (
 	"context"
-	"errors"
 )
 
 type Snapshot struct {
@@ -46,16 +45,7 @@ func (s *Snapshot) IsValid() bool {
 
 func (r *Request) MarkCompleted(err error) {
 	r.Status = StatusCompleted
-	if err == nil {
-		return
-	}
-
-	var snapshotErrs *Errors
-	if errors.As(err, &snapshotErrs) {
-		r.Errors = snapshotErrs
-	} else {
-		r.Errors = &Errors{Snapshot: err}
-	}
+	r.Errors = NewErrors(err)
 }
 
 func (r *Request) MarkInProgress() {
diff --git a/pkg/snapshot/store/mocks/mock_snapshot_store.go b/pkg/snapshot/store/mocks/mock_snapshot_store.go
index 35cfcdb..d2b8a8e 100644
--- a/pkg/snapshot/store/mocks/mock_snapshot_store.go
+++ b/pkg/snapshot/store/mocks/mock_snapshot_store.go
@@ -10,9 +10,10 @@ import (
 
 type Store struct {
 	CreateSnapshotRequestFn       func(context.Context, *snapshot.Request) error
-	UpdateSnapshotRequestFn       func(context.Context, *snapshot.Request) error
+	UpdateSnapshotRequestFn       func(context.Context, uint, *snapshot.Request) error
 	GetSnapshotRequestsByStatusFn func(ctx context.Context, status snapshot.Status) ([]*snapshot.Request, error)
 	GetSnapshotRequestsBySchemaFn func(ctx context.Context, s string) ([]*snapshot.Request, error)
+	updateSnapshotRequestCalls    uint
 }
 
 func (m *Store) CreateSnapshotRequest(ctx context.Context, s *snapshot.Request) error {
@@ -20,7 +21,8 @@ func (m *Store) CreateSnapshotRequest(ctx context.Context, s *snapshot.Request)
 }
 
 func (m *Store) UpdateSnapshotRequest(ctx context.Context, s *snapshot.Request) error {
-	return m.UpdateSnapshotRequestFn(ctx, s)
+	m.updateSnapshotRequestCalls++
+	return m.UpdateSnapshotRequestFn(ctx, m.updateSnapshotRequestCalls, s)
 }
 
 func (m *Store) GetSnapshotRequestsByStatus(ctx context.Context, status snapshot.Status) ([]*snapshot.Request, error) {
diff --git a/pkg/snapshot/store/postgres/pg_snapshot_store.go b/pkg/snapshot/store/postgres/pg_snapshot_store.go
index fb92dce..229e872 100644
--- a/pkg/snapshot/store/postgres/pg_snapshot_store.go
+++ b/pkg/snapshot/store/postgres/pg_snapshot_store.go
@@ -42,7 +42,7 @@ func (s *Store) Close() error {
 
 func (s *Store) CreateSnapshotRequest(ctx context.Context, req *snapshot.Request) error {
 	query := fmt.Sprintf(`INSERT INTO %s (schema_name, table_names, created_at, updated_at, status)
-	VALUES($1, $2,'now()','now()','requested')`, snapshotsTable())
+	VALUES($1, $2, now(), now(),'requested')`, snapshotsTable())
 	_, err := s.conn.Exec(ctx, query, req.Snapshot.SchemaName, pq.StringArray(req.Snapshot.TableNames))
 	if err != nil {
 		return fmt.Errorf("error creating snapshot request: %w", err)
@@ -51,7 +51,7 @@ func (s *Store) CreateSnapshotRequest(ctx context.Context, req *snapshot.Request
 }
 
 func (s *Store) UpdateSnapshotRequest(ctx context.Context, req *snapshot.Request) error {
-	query := fmt.Sprintf(`UPDATE %s SET status = $1, errors = $2, updated_at = 'now()'
+	query := fmt.Sprintf(`UPDATE %s SET status = $1, errors = $2, updated_at = now()
 	WHERE schema_name = $3 and table_names = $4 and status != 'completed'`, snapshotsTable())
 	_, err := s.conn.Exec(ctx, query, req.Status, req.Errors, req.Snapshot.SchemaName, pq.StringArray(req.Snapshot.TableNames))
 	if err != nil {


Test patch:
diff --git a/pkg/snapshot/generator/helper_test.go b/pkg/snapshot/generator/helper_test.go
new file mode 100644
index 0000000..f113299
--- /dev/null
+++ b/pkg/snapshot/generator/helper_test.go
@@ -0,0 +1,25 @@
+// SPDX-License-Identifier: Apache-2.0
+
+package generator
+
+import (
+	"context"
+
+	"github.com/xataio/pgstream/pkg/snapshot"
+)
+
+type mockGenerator struct {
+	createSnapshotFn func(ctx context.Context, snapshot *snapshot.Snapshot) error
+	closeFn          func() error
+}
+
+func (m *mockGenerator) CreateSnapshot(ctx context.Context, snapshot *snapshot.Snapshot) error {
+	return m.createSnapshotFn(ctx, snapshot)
+}
+
+func (m *mockGenerator) Close() error {
+	if m.closeFn != nil {
+		return m.closeFn()
+	}
+	return nil
+}
diff --git a/pkg/snapshot/generator/snapshot_generator_recorder_test.go b/pkg/snapshot/generator/snapshot_generator_recorder_test.go
new file mode 100644
index 0000000..1ec591e
--- /dev/null
+++ b/pkg/snapshot/generator/snapshot_generator_recorder_test.go
@@ -0,0 +1,540 @@
+// SPDX-License-Identifier: Apache-2.0
+
+package generator
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"github.com/xataio/pgstream/pkg/snapshot"
+	snapshotstore "github.com/xataio/pgstream/pkg/snapshot/store"
+	snapshotstoremocks "github.com/xataio/pgstream/pkg/snapshot/store/mocks"
+)
+
+func TestSnapshotRecorder_CreateSnapshot(t *testing.T) {
+	t.Parallel()
+
+	testSnapshot := snapshot.Snapshot{
+		SchemaName: "test-schema",
+		TableNames: []string{"table1", "table2"},
+	}
+
+	newTestSnapshot := func() *snapshot.Snapshot {
+		ss := testSnapshot
+		return &ss
+	}
+
+	errTest := errors.New("oh noes")
+	updateErr := errors.New("update error")
+
+	tests := []struct {
+		name      string
+		store     snapshotstore.Store
+		generator SnapshotGenerator
+
+		wantErr error
+	}{
+		{
+			name: "ok",
+			store: &snapshotstoremocks.Store{
+				GetSnapshotRequestsBySchemaFn: func(ctx context.Context, s string) ([]*snapshot.Request, error) {
+					return []*snapshot.Request{}, nil
+				},
+				CreateSnapshotRequestFn: func(ctx context.Context, r *snapshot.Request) error {
+					require.Equal(t, &snapshot.Request{
+						Snapshot: testSnapshot,
+					}, r)
+					return nil
+				},
+				UpdateSnapshotRequestFn: func(ctx context.Context, i uint, r *snapshot.Request) error {
+					switch i {
+					case 1:
+						require.Equal(t, &snapshot.Request{
+							Snapshot: testSnapshot,
+							Status:   snapshot.StatusInProgress,
+						}, r)
+						return nil
+					case 2:
+						require.Equal(t, &snapshot.Request{
+							Snapshot: testSnapshot,
+							Status:   snapshot.StatusCompleted,
+						}, r)
+						return nil
+					default:
+						return fmt.Errorf("unexpected call to UpdateSnapshotRequestFn: %d", i)
+
+					}
+				},
+			},
+			generator: &mockGenerator{
+				createSnapshotFn: func(ctx context.Context, ss *snapshot.Snapshot) error {
+					require.Equal(t, newTestSnapshot(), ss)
+					return nil
+				},
+			},
+
+			wantErr: nil,
+		},
+		{
+			name: "ok - all tables filtered out",
+			store: &snapshotstoremocks.Store{
+				GetSnapshotRequestsBySchemaFn: func(ctx context.Context, s string) ([]*snapshot.Request, error) {
+					return []*snapshot.Request{
+						{Snapshot: testSnapshot, Status: snapshot.StatusCompleted},
+					}, nil
+				},
+			},
+			generator: &mockGenerator{},
+
+			wantErr: nil,
+		},
+		{
+			name: "error - getting existing requests",
+			store: &snapshotstoremocks.Store{
+				GetSnapshotRequestsBySchemaFn: func(ctx context.Context, s string) ([]*snapshot.Request, error) {
+					return nil, errTest
+				},
+			},
+			generator: &mockGenerator{},
+
+			wantErr: &snapshot.Errors{Snapshot: fmt.Errorf("retrieving existing snapshots for schema: %w", errTest)},
+		},
+		{
+			name: "error - snapshot error on wrapped generator",
+			store: &snapshotstoremocks.Store{
+				GetSnapshotRequestsBySchemaFn: func(ctx context.Context, s string) ([]*snapshot.Request, error) {
+					return []*snapshot.Request{}, nil
+				},
+				CreateSnapshotRequestFn: func(ctx context.Context, r *snapshot.Request) error {
+					require.Equal(t, &snapshot.Request{
+						Snapshot: testSnapshot,
+					}, r)
+					return nil
+				},
+				UpdateSnapshotRequestFn: func(ctx context.Context, i uint, r *snapshot.Request) error {
+					switch i {
+					case 1:
+						require.Equal(t, &snapshot.Request{
+							Snapshot: testSnapshot,
+							Status:   snapshot.StatusInProgress,
+						}, r)
+						return nil
+					case 2:
+						require.Equal(t, &snapshot.Request{
+							Snapshot: testSnapshot,
+							Status:   snapshot.StatusCompleted,
+							Errors:   &snapshot.Errors{Snapshot: errTest},
+						}, r)
+						return nil
+					default:
+						return fmt.Errorf("unexpected call to UpdateSnapshotRequestFn: %d", i)
+
+					}
+				},
+			},
+			generator: &mockGenerator{
+				createSnapshotFn: func(ctx context.Context, ss *snapshot.Snapshot) error {
+					return &snapshot.Errors{Snapshot: errTest}
+				},
+			},
+
+			wantErr: &snapshot.Errors{Snapshot: errTest},
+		},
+		{
+			name: "error - recording snapshot request",
+			store: &snapshotstoremocks.Store{
+				GetSnapshotRequestsBySchemaFn: func(ctx context.Context, s string) ([]*snapshot.Request, error) {
+					return []*snapshot.Request{}, nil
+				},
+				CreateSnapshotRequestFn: func(ctx context.Context, r *snapshot.Request) error {
+					return errTest
+				},
+			},
+			generator: &mockGenerator{
+				createSnapshotFn: func(ctx context.Context, ss *snapshot.Snapshot) error {
+					return errors.New("createSnapshotFn: should not be called")
+				},
+			},
+
+			wantErr: &snapshot.Errors{Snapshot: errTest},
+		},
+		{
+			name: "error - updating snapshot request in progress",
+			store: &snapshotstoremocks.Store{
+				GetSnapshotRequestsBySchemaFn: func(ctx context.Context, s string) ([]*snapshot.Request, error) {
+					return []*snapshot.Request{}, nil
+				},
+				CreateSnapshotRequestFn: func(ctx context.Context, r *snapshot.Request) error {
+					return nil
+				},
+				UpdateSnapshotRequestFn: func(ctx context.Context, i uint, r *snapshot.Request) error {
+					switch i {
+					case 1:
+						require.Equal(t, &snapshot.Request{
+							Snapshot: testSnapshot,
+							Status:   snapshot.StatusInProgress,
+						}, r)
+						return errTest
+					default:
+						return fmt.Errorf("unexpected call to UpdateSnapshotRequestFn: %d", i)
+
+					}
+				},
+			},
+			generator: &mockGenerator{},
+
+			wantErr: &snapshot.Errors{Snapshot: errTest},
+		},
+		{
+			name: "error - updating snapshot request completed without errors",
+			store: &snapshotstoremocks.Store{
+				GetSnapshotRequestsBySchemaFn: func(ctx context.Context, s string) ([]*snapshot.Request, error) {
+					return []*snapshot.Request{}, nil
+				},
+				CreateSnapshotRequestFn: func(ctx context.Context, r *snapshot.Request) error {
+					require.Equal(t, &snapshot.Request{
+						Snapshot: testSnapshot,
+					}, r)
+					return nil
+				},
+				UpdateSnapshotRequestFn: func(ctx context.Context, i uint, r *snapshot.Request) error {
+					switch i {
+					case 1:
+						require.Equal(t, &snapshot.Request{
+							Snapshot: testSnapshot,
+							Status:   snapshot.StatusInProgress,
+						}, r)
+						return nil
+					case 2:
+						require.Equal(t, &snapshot.Request{
+							Snapshot: testSnapshot,
+							Status:   snapshot.StatusCompleted,
+						}, r)
+						return errTest
+					default:
+						return fmt.Errorf("unexpected call to UpdateSnapshotRequestFn: %d", i)
+
+					}
+				},
+			},
+			generator: &mockGenerator{
+				createSnapshotFn: func(ctx context.Context, ss *snapshot.Snapshot) error {
+					require.Equal(t, newTestSnapshot(), ss)
+					return nil
+				},
+			},
+
+			wantErr: &snapshot.Errors{Snapshot: errTest},
+		},
+		{
+			name: "error - updating snapshot request completed with snapshot and table errors",
+			store: &snapshotstoremocks.Store{
+				GetSnapshotRequestsBySchemaFn: func(ctx context.Context, s string) ([]*snapshot.Request, error) {
+					return []*snapshot.Request{}, nil
+				},
+				CreateSnapshotRequestFn: func(ctx context.Context, r *snapshot.Request) error {
+					require.Equal(t, &snapshot.Request{
+						Snapshot: testSnapshot,
+					}, r)
+					return nil
+				},
+				UpdateSnapshotRequestFn: func(ctx context.Context, i uint, r *snapshot.Request) error {
+					switch i {
+					case 1:
+						require.Equal(t, &snapshot.Request{
+							Snapshot: testSnapshot,
+							Status:   snapshot.StatusInProgress,
+						}, r)
+						return nil
+					case 2:
+						require.Equal(t, &snapshot.Request{
+							Snapshot: testSnapshot,
+							Status:   snapshot.StatusCompleted,
+							Errors: &snapshot.Errors{
+								Snapshot: errTest,
+								Tables: []snapshot.TableError{
+									{Table: "table1", ErrorMsg: errTest.Error()},
+								},
+							},
+						}, r)
+						return updateErr
+					default:
+						return fmt.Errorf("unexpected call to UpdateSnapshotRequestFn: %d", i)
+
+					}
+				},
+			},
+			generator: &mockGenerator{
+				createSnapshotFn: func(ctx context.Context, ss *snapshot.Snapshot) error {
+					require.Equal(t, newTestSnapshot(), ss)
+					return &snapshot.Errors{
+						Snapshot: errTest,
+						Tables: []snapshot.TableError{
+							{Table: "table1", ErrorMsg: errTest.Error()},
+						},
+					}
+				},
+			},
+
+			wantErr: &snapshot.Errors{
+				Snapshot: errors.Join(errTest, updateErr),
+				Tables: []snapshot.TableError{
+					{Table: "table1", ErrorMsg: errTest.Error()},
+				},
+			},
+		},
+		{
+			name: "error - updating snapshot request completed with table errors",
+			store: &snapshotstoremocks.Store{
+				GetSnapshotRequestsBySchemaFn: func(ctx context.Context, s string) ([]*snapshot.Request, error) {
+					return []*snapshot.Request{}, nil
+				},
+				CreateSnapshotRequestFn: func(ctx context.Context, r *snapshot.Request) error {
+					require.Equal(t, &snapshot.Request{
+						Snapshot: testSnapshot,
+					}, r)
+					return nil
+				},
+				UpdateSnapshotRequestFn: func(ctx context.Context, i uint, r *snapshot.Request) error {
+					switch i {
+					case 1:
+						require.Equal(t, &snapshot.Request{
+							Snapshot: testSnapshot,
+							Status:   snapshot.StatusInProgress,
+						}, r)
+						return nil
+					case 2:
+						require.Equal(t, &snapshot.Request{
+							Snapshot: testSnapshot,
+							Status:   snapshot.StatusCompleted,
+							Errors: &snapshot.Errors{
+								Tables: []snapshot.TableError{
+									{Table: "table1", ErrorMsg: errTest.Error()},
+								},
+							},
+						}, r)
+						return updateErr
+					default:
+						return fmt.Errorf("unexpected call to UpdateSnapshotRequestFn: %d", i)
+
+					}
+				},
+			},
+			generator: &mockGenerator{
+				createSnapshotFn: func(ctx context.Context, ss *snapshot.Snapshot) error {
+					require.Equal(t, newTestSnapshot(), ss)
+					return &snapshot.Errors{
+						Tables: []snapshot.TableError{
+							{Table: "table1", ErrorMsg: errTest.Error()},
+						},
+					}
+				},
+			},
+
+			wantErr: &snapshot.Errors{
+				Snapshot: updateErr,
+				Tables: []snapshot.TableError{
+					{Table: "table1", ErrorMsg: errTest.Error()},
+				},
+			},
+		},
+	}
+
+	for _, tc := range tests {
+		tc := tc
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			sr := NewSnapshotRecorder(tc.store, tc.generator)
+			defer sr.Close()
+
+			err := sr.CreateSnapshot(context.Background(), newTestSnapshot())
+			require.Equal(t, tc.wantErr, err)
+		})
+	}
+}
+
+func TestSnapshotRecorder_filterOutExistingSnapshots(t *testing.T) {
+	t.Parallel()
+
+	testSchema := "test-schema"
+	testTables := []string{"table1", "table2"}
+	errTest := errors.New("oh noes")
+
+	tests := []struct {
+		name   string
+		store  snapshotstore.Store
+		tables []string
+
+		wantTables []string
+		wantErr    error
+	}{
+		{
+			name: "ok - no existing snapshots",
+			store: &snapshotstoremocks.Store{
+				GetSnapshotRequestsBySchemaFn: func(ctx context.Context, s string) ([]*snapshot.Request, error) {
+					return []*snapshot.Request{}, nil
+				},
+			},
+			wantTables: testTables,
+		},
+		{
+			name: "ok - no existing snapshots with wildcard",
+			store: &snapshotstoremocks.Store{
+				GetSnapshotRequestsBySchemaFn: func(ctx context.Context, s string) ([]*snapshot.Request, error) {
+					return []*snapshot.Request{
+						{
+							Snapshot: snapshot.Snapshot{
+								SchemaName: testSchema,
+								TableNames: []string{"table2"},
+							},
+							Status: snapshot.StatusCompleted,
+						},
+					}, nil
+				},
+			},
+			tables:     []string{"*"},
+			wantTables: []string{"*"},
+		},
+		{
+			name: "ok - existing snapshots",
+			store: &snapshotstoremocks.Store{
+				GetSnapshotRequestsBySchemaFn: func(ctx context.Context, s string) ([]*snapshot.Request, error) {
+					return []*snapshot.Request{
+						{
+							Snapshot: snapshot.Snapshot{
+								SchemaName: testSchema,
+								TableNames: []string{"table2"},
+							},
+							Status: snapshot.StatusCompleted,
+						},
+					}, nil
+				},
+			},
+			wantTables: []string{"table1"},
+		},
+		{
+			name: "ok - existing wildcard snapshot with wildcard table",
+			store: &snapshotstoremocks.Store{
+				GetSnapshotRequestsBySchemaFn: func(ctx context.Context, s string) ([]*snapshot.Request, error) {
+					return []*snapshot.Request{
+						{
+							Snapshot: snapshot.Snapshot{
+								SchemaName: testSchema,
+								TableNames: []string{"*"},
+							},
+							Status: snapshot.StatusCompleted,
+						},
+					}, nil
+				},
+			},
+			tables:     []string{"*"},
+			wantTables: []string{},
+		},
+		{
+			name: "ok - existing wildcard snapshot",
+			store: &snapshotstoremocks.Store{
+				GetSnapshotRequestsBySchemaFn: func(ctx context.Context, s string) ([]*snapshot.Request, error) {
+					return []*snapshot.Request{
+						{
+							Snapshot: snapshot.Snapshot{
+								SchemaName: testSchema,
+								TableNames: []string{"*"},
+							},
+							Status: snapshot.StatusCompleted,
+						},
+					}, nil
+				},
+			},
+			wantTables: []string{},
+		},
+		{
+			name: "ok - existing failed snapshots",
+			store: &snapshotstoremocks.Store{
+				GetSnapshotRequestsBySchemaFn: func(ctx context.Context, s string) ([]*snapshot.Request, error) {
+					return []*snapshot.Request{
+						{
+							Snapshot: snapshot.Snapshot{
+								SchemaName: testSchema,
+								TableNames: []string{"table2"},
+							},
+							Errors: &snapshot.Errors{Snapshot: errTest},
+							Status: snapshot.StatusCompleted,
+						},
+					}, nil
+				},
+			},
+			wantTables: []string{"table1", "table2"},
+		},
+		{
+			name: "ok - existing failed wildcard snapshot",
+			store: &snapshotstoremocks.Store{
+				GetSnapshotRequestsBySchemaFn: func(ctx context.Context, s string) ([]*snapshot.Request, error) {
+					return []*snapshot.Request{
+						{
+							Snapshot: snapshot.Snapshot{
+								SchemaName: testSchema,
+								TableNames: []string{"*"},
+							},
+							Errors: &snapshot.Errors{Tables: []snapshot.TableError{{Table: "table2", ErrorMsg: errTest.Error()}}},
+							Status: snapshot.StatusCompleted,
+						},
+					}, nil
+				},
+			},
+			tables:     []string{"*"},
+			wantTables: []string{"table2"},
+		},
+		{
+			name: "ok - existing failed table on wildcard snapshot",
+			store: &snapshotstoremocks.Store{
+				GetSnapshotRequestsBySchemaFn: func(ctx context.Context, s string) ([]*snapshot.Request, error) {
+					return []*snapshot.Request{
+						{
+							Snapshot: snapshot.Snapshot{
+								SchemaName: testSchema,
+								TableNames: []string{"*"},
+							},
+							Errors: &snapshot.Errors{Tables: []snapshot.TableError{{Table: "table2", ErrorMsg: errTest.Error()}}},
+							Status: snapshot.StatusCompleted,
+						},
+					}, nil
+				},
+			},
+			wantTables: []string{"table2"},
+		},
+		{
+			name: "error - retrieving existing snapshots",
+			store: &snapshotstoremocks.Store{
+				GetSnapshotRequestsBySchemaFn: func(ctx context.Context, s string) ([]*snapshot.Request, error) {
+					return nil, errTest
+				},
+			},
+			wantTables: nil,
+			wantErr:    errTest,
+		},
+	}
+
+	for _, tc := range tests {
+		tc := tc
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			sr := SnapshotRecorder{
+				store: tc.store,
+			}
+
+			tables := testTables
+			if tc.tables != nil {
+				tables = tc.tables
+			}
+
+			filteredTables, err := sr.filterOutExistingSnapshots(context.Background(), testSchema, tables)
+			require.ErrorIs(t, err, tc.wantErr)
+			require.Equal(t, tc.wantTables, filteredTables)
+		})
+	}
+}
diff --git a/pkg/snapshot/store/postgres/pg_snapshot_store_test.go b/pkg/snapshot/store/postgres/pg_snapshot_store_test.go
index 076e53d..25db8b0 100644
--- a/pkg/snapshot/store/postgres/pg_snapshot_store_test.go
+++ b/pkg/snapshot/store/postgres/pg_snapshot_store_test.go
@@ -39,7 +39,7 @@ func TestStore_CreateSnapshotRequest(t *testing.T) {
 			querier: &postgresmocks.Querier{
 				ExecFn: func(ctx context.Context, _ uint, s string, a ...any) (postgres.CommandTag, error) {
 					wantQuery := fmt.Sprintf(`INSERT INTO %s (schema_name, table_names, created_at, updated_at, status)
-	VALUES($1, $2,'now()','now()','requested')`, snapshotsTable())
+	VALUES($1, $2, now(), now(),'requested')`, snapshotsTable())
 					require.Equal(t, wantQuery, s)
 					wantAttr := []any{testSnapshot.SchemaName, pq.StringArray(testSnapshot.TableNames)}
 					require.Equal(t, wantAttr, a)
@@ -101,7 +101,7 @@ func TestStore_UpdateSnapshotRequest(t *testing.T) {
 			name: "ok - update without error",
 			querier: &postgresmocks.Querier{
 				ExecFn: func(ctx context.Context, _ uint, s string, a ...any) (postgres.CommandTag, error) {
-					wantQuery := fmt.Sprintf(`UPDATE %s SET status = $1, errors = $2, updated_at = 'now()'
+					wantQuery := fmt.Sprintf(`UPDATE %s SET status = $1, errors = $2, updated_at = now()
 	WHERE schema_name = $3 and table_names = $4 and status != 'completed'`,
 						snapshotsTable())
 					require.Equal(t, wantQuery, s)
@@ -117,7 +117,7 @@ func TestStore_UpdateSnapshotRequest(t *testing.T) {
 			name: "ok - update with error",
 			querier: &postgresmocks.Querier{
 				ExecFn: func(ctx context.Context, _ uint, s string, a ...any) (postgres.CommandTag, error) {
-					wantQuery := fmt.Sprintf(`UPDATE %s SET status = $1, errors = $2, updated_at = 'now()'
+					wantQuery := fmt.Sprintf(`UPDATE %s SET status = $1, errors = $2, updated_at = now()
 	WHERE schema_name = $3 and table_names = $4 and status != 'completed'`,
 						snapshotsTable())
 					require.Equal(t, wantQuery, s)