Repository: xataio/pgstream
Pull request #147 (2025-02-17 14:36:16)

Restructure snapshot generator pkg
This PR restructures the snapshot generator package to prepare for the addition of a new postgres schema implementation (using pgdump/pgrestore). 

The main changes are the following moves:
- `pkg/snapshot/generator/data/postgres` -> `pkg/snapshot/generator/postgres/data`
- `pkg/snapshot/generator/schema` -> `pkg/snapshot/generator/postgres/schema/schemalog`. 

This will enable the new implementation to be added under `pkg/snapshot/generator/postgres/schema`. Also, it makes the data/schema separation specific of the postgres implementation.


Patch:
diff --git a/cmd/config.go b/cmd/config.go
index 7cb6c49..f26b5ad 100644
--- a/cmd/config.go
+++ b/cmd/config.go
@@ -9,7 +9,7 @@ import (
 	"github.com/xataio/pgstream/pkg/backoff"
 	"github.com/xataio/pgstream/pkg/kafka"
 	pgschemalog "github.com/xataio/pgstream/pkg/schemalog/postgres"
-	pgsnapshotgenerator "github.com/xataio/pgstream/pkg/snapshot/generator/data/postgres"
+	pgsnapshotgenerator "github.com/xataio/pgstream/pkg/snapshot/generator/postgres/data"
 	"github.com/xataio/pgstream/pkg/stream"
 	"github.com/xataio/pgstream/pkg/tls"
 	kafkacheckpoint "github.com/xataio/pgstream/pkg/wal/checkpointer/kafka"
diff --git a/pkg/snapshot/generator/schema/snapshot_schema_generator.go b/pkg/snapshot/generator/postgres/schema/schemalog/snapshot_schemalog_generator.go
similarity index 96%
rename from pkg/snapshot/generator/schema/snapshot_schema_generator.go
rename to pkg/snapshot/generator/postgres/schema/schemalog/snapshot_schemalog_generator.go
index 3ea4f38..4a31f47 100644
--- a/pkg/snapshot/generator/schema/snapshot_schema_generator.go
+++ b/pkg/snapshot/generator/postgres/schema/schemalog/snapshot_schemalog_generator.go
@@ -1,6 +1,6 @@
 // SPDX-License-Identifier: Apache-2.0
 
-package schema
+package schemalog
 
 import (
 	"context"
@@ -40,7 +40,7 @@ func (s *SnapshotGenerator) CreateSnapshot(ctx context.Context, ss *snapshot.Sna
 		return s.processRow(ctx, row)
 	}()
 	if err != nil {
-		return &snapshot.Errors{Snapshot: err}
+		return snapshot.NewErrors(err)
 	}
 
 	return nil
diff --git a/pkg/stream/stream_run.go b/pkg/stream/stream_run.go
index d5cea23..470a534 100644
--- a/pkg/stream/stream_run.go
+++ b/pkg/stream/stream_run.go
@@ -11,7 +11,7 @@ import (
 	kafkainstrumentation "github.com/xataio/pgstream/pkg/kafka/instrumentation"
 	loglib "github.com/xataio/pgstream/pkg/log"
 	"github.com/xataio/pgstream/pkg/otel"
-	pgsnapshotgenerator "github.com/xataio/pgstream/pkg/snapshot/generator/data/postgres"
+	pgsnapshotgenerator "github.com/xataio/pgstream/pkg/snapshot/generator/postgres/data"
 	"github.com/xataio/pgstream/pkg/wal/checkpointer"
 	kafkacheckpoint "github.com/xataio/pgstream/pkg/wal/checkpointer/kafka"
 	pgcheckpoint "github.com/xataio/pgstream/pkg/wal/checkpointer/postgres"
diff --git a/pkg/wal/listener/postgres/config.go b/pkg/wal/listener/postgres/config.go
index f98e14b..0a64ab6 100644
--- a/pkg/wal/listener/postgres/config.go
+++ b/pkg/wal/listener/postgres/config.go
@@ -6,7 +6,7 @@ import (
 	"strings"
 
 	schemalogpg "github.com/xataio/pgstream/pkg/schemalog/postgres"
-	pgsnapshotgenerator "github.com/xataio/pgstream/pkg/snapshot/generator/data/postgres"
+	pgsnapshotgenerator "github.com/xataio/pgstream/pkg/snapshot/generator/postgres/data"
 )
 
 type SnapshotConfig struct {
diff --git a/pkg/wal/listener/postgres/wal_pg_snapshot_generator_adapter.go b/pkg/wal/listener/postgres/wal_pg_snapshot_generator_adapter.go
index 8cbe1a0..0f9ab57 100644
--- a/pkg/wal/listener/postgres/wal_pg_snapshot_generator_adapter.go
+++ b/pkg/wal/listener/postgres/wal_pg_snapshot_generator_adapter.go
@@ -13,8 +13,8 @@ import (
 	schemalogpg "github.com/xataio/pgstream/pkg/schemalog/postgres"
 	"github.com/xataio/pgstream/pkg/snapshot"
 	"github.com/xataio/pgstream/pkg/snapshot/generator"
-	pgsnapshotgenerator "github.com/xataio/pgstream/pkg/snapshot/generator/data/postgres"
-	"github.com/xataio/pgstream/pkg/snapshot/generator/schema"
+	pgsnapshotgenerator "github.com/xataio/pgstream/pkg/snapshot/generator/postgres/data"
+	pgschemasnapshot "github.com/xataio/pgstream/pkg/snapshot/generator/postgres/schema/schemalog"
 	pgsnapshotstore "github.com/xataio/pgstream/pkg/snapshot/store/postgres"
 	"github.com/xataio/pgstream/pkg/wal"
 	"golang.org/x/sync/errgroup"
@@ -51,7 +51,7 @@ func NewSnapshotGeneratorAdapter(ctx context.Context, cfg *SnapshotConfig, proce
 		return nil, fmt.Errorf("create schema log postgres store: %w", err)
 	}
 	schemaLogStore = schemalog.NewStoreCache(schemaLogStore)
-	schemaSnapshotGenerator := schema.NewSnapshotGenerator(schemaLogStore, s.processRow)
+	schemaSnapshotGenerator := pgschemasnapshot.NewSnapshotGenerator(schemaLogStore, s.processRow)
 
 	// postgres data snapshot generator
 	dataSnapshotGenerator, err := pgsnapshotgenerator.NewSnapshotGenerator(ctx, &cfg.Generator, s.processRow, opts...)


Test patch:
diff --git a/pkg/snapshot/generator/data/postgres/config.go b/pkg/snapshot/generator/postgres/data/config.go
similarity index 100%
rename from pkg/snapshot/generator/data/postgres/config.go
rename to pkg/snapshot/generator/postgres/data/config.go
diff --git a/pkg/snapshot/generator/data/postgres/helper_test.go b/pkg/snapshot/generator/postgres/data/helper_test.go
similarity index 100%
rename from pkg/snapshot/generator/data/postgres/helper_test.go
rename to pkg/snapshot/generator/postgres/data/helper_test.go
diff --git a/pkg/snapshot/generator/data/postgres/pg_schema_table_parser.go b/pkg/snapshot/generator/postgres/data/pg_schema_table_parser.go
similarity index 100%
rename from pkg/snapshot/generator/data/postgres/pg_schema_table_parser.go
rename to pkg/snapshot/generator/postgres/data/pg_schema_table_parser.go
diff --git a/pkg/snapshot/generator/data/postgres/pg_schema_table_parser_test.go b/pkg/snapshot/generator/postgres/data/pg_schema_table_parser_test.go
similarity index 100%
rename from pkg/snapshot/generator/data/postgres/pg_schema_table_parser_test.go
rename to pkg/snapshot/generator/postgres/data/pg_schema_table_parser_test.go
diff --git a/pkg/snapshot/generator/data/postgres/pg_snapshot_generator.go b/pkg/snapshot/generator/postgres/data/pg_snapshot_generator.go
similarity index 100%
rename from pkg/snapshot/generator/data/postgres/pg_snapshot_generator.go
rename to pkg/snapshot/generator/postgres/data/pg_snapshot_generator.go
diff --git a/pkg/snapshot/generator/data/postgres/pg_snapshot_generator_integration_test.go b/pkg/snapshot/generator/postgres/data/pg_snapshot_generator_integration_test.go
similarity index 100%
rename from pkg/snapshot/generator/data/postgres/pg_snapshot_generator_integration_test.go
rename to pkg/snapshot/generator/postgres/data/pg_snapshot_generator_integration_test.go
diff --git a/pkg/snapshot/generator/data/postgres/pg_snapshot_generator_test.go b/pkg/snapshot/generator/postgres/data/pg_snapshot_generator_test.go
similarity index 99%
rename from pkg/snapshot/generator/data/postgres/pg_snapshot_generator_test.go
rename to pkg/snapshot/generator/postgres/data/pg_snapshot_generator_test.go
index 4953183..c8588a8 100644
--- a/pkg/snapshot/generator/data/postgres/pg_snapshot_generator_test.go
+++ b/pkg/snapshot/generator/postgres/data/pg_snapshot_generator_test.go
@@ -389,9 +389,7 @@ func TestSnapshotGenerator_CreateSnapshot(t *testing.T) {
 				},
 			},
 
-			wantErr: &snapshot.Errors{
-				Snapshot: fmt.Errorf("exporting snapshot: %w", errTest),
-			},
+			wantErr:  snapshot.NewErrors(fmt.Errorf("exporting snapshot: %w", errTest)),
 			wantRows: []*snapshot.Row{},
 		},
 		{
diff --git a/pkg/snapshot/generator/schema/snapshot_schema_generator_test.go b/pkg/snapshot/generator/postgres/schema/schemalog/snapshot_schemalog_generator_test.go
similarity index 94%
rename from pkg/snapshot/generator/schema/snapshot_schema_generator_test.go
rename to pkg/snapshot/generator/postgres/schema/schemalog/snapshot_schemalog_generator_test.go
index 87ab72d..9244e40 100644
--- a/pkg/snapshot/generator/schema/snapshot_schema_generator_test.go
+++ b/pkg/snapshot/generator/postgres/schema/schemalog/snapshot_schemalog_generator_test.go
@@ -1,6 +1,6 @@
 // SPDX-License-Identifier: Apache-2.0
 
-package schema
+package schemalog
 
 import (
 	"context"
@@ -98,7 +98,7 @@ func TestSnapshotGenerator_CreateSnapshot(t *testing.T) {
 				return errors.New("processRow: should not be called")
 			},
 
-			wantErr: &snapshot.Errors{Snapshot: errTest},
+			wantErr: snapshot.NewErrors(errTest),
 		},
 		{
 			name: "error - processing schema row",
@@ -112,7 +112,7 @@ func TestSnapshotGenerator_CreateSnapshot(t *testing.T) {
 				return errTest
 			},
 
-			wantErr: &snapshot.Errors{Snapshot: errTest},
+			wantErr: snapshot.NewErrors(errTest),
 		},
 		{
 			name: "error - converting log entry to row",
@@ -126,7 +126,7 @@ func TestSnapshotGenerator_CreateSnapshot(t *testing.T) {
 			},
 			marshaler: func(a any) ([]byte, error) { return nil, errTest },
 
-			wantErr: &snapshot.Errors{Snapshot: fmt.Errorf("marshaling log entry schema into json: %w", errTest)},
+			wantErr: snapshot.NewErrors(fmt.Errorf("marshaling log entry schema into json: %w", errTest)),
 		},
 	}