Repository: xataio/pgstream
Pull request #263 (2025-04-10 09:13:57)

Add snapshot recorder url to yaml config
This PR updates the yaml config to add the snapshot recorder postgres url instead of automatically setting it to the source postgres url. This is a configurable value, making the opt in explicit ensures a more straightforward flow.


Patch:
diff --git a/Makefile b/Makefile
index f6e4de9..995b869 100644
--- a/Makefile
+++ b/Makefile
@@ -24,4 +24,4 @@ gen-migrations:
 
 generate:
 	# Generate the cli-definition.json file
-	go run tools/build-cli-definition.go
\ No newline at end of file
+	go run tools/build-cli-definition.go
diff --git a/cmd/config/config_yaml.go b/cmd/config/config_yaml.go
index 6ec960f..6bbfee7 100644
--- a/cmd/config/config_yaml.go
+++ b/cmd/config/config_yaml.go
@@ -55,12 +55,16 @@ type PostgresConfig struct {
 }
 
 type SnapshotConfig struct {
-	Mode            string                `mapstructure:"mode" yaml:"mode"`
-	Tables          []string              `mapstructure:"tables" yaml:"tables"`
-	RecorderEnabled bool                  `mapstructure:"recorder_enabled" yaml:"recorder_enabled"`
-	SnapshotWorkers int                   `mapstructure:"snapshot_workers" yaml:"snapshot_workers"`
-	Data            *SnapshotDataConfig   `mapstructure:"data" yaml:"data"`
-	Schema          *SnapshotSchemaConfig `mapstructure:"schema" yaml:"schema"`
+	Mode            string                  `mapstructure:"mode" yaml:"mode"`
+	Tables          []string                `mapstructure:"tables" yaml:"tables"`
+	Recorder        *SnapshotRecorderConfig `mapstructure:"recorder" yaml:"recorder"`
+	SnapshotWorkers int                     `mapstructure:"snapshot_workers" yaml:"snapshot_workers"`
+	Data            *SnapshotDataConfig     `mapstructure:"data" yaml:"data"`
+	Schema          *SnapshotSchemaConfig   `mapstructure:"schema" yaml:"schema"`
+}
+
+type SnapshotRecorderConfig struct {
+	PostgresURL string `mapstructure:"postgres_url" yaml:"postgres_url"`
 }
 
 type SnapshotDataConfig struct {
@@ -247,6 +251,7 @@ var (
 	errUnsupportedSearchEngine       = errors.New("unsupported search engine, must be one of 'opensearch' or 'elasticsearch'")
 	errInvalidPgdumpPgrestoreConfig  = errors.New("pgdump_pgrestore snapshot mode requires target postgres config")
 	errInvalidInjectorConfig         = errors.New("injector config can't infer schemalog url from source postgres url, schemalog_url must be provided")
+	errInvalidSnapshotRecorderConfig = errors.New("snapshot recorder config requires a postgres url")
 )
 
 func (c *YAMLConfig) toStreamConfig() (*stream.Config, error) {
@@ -348,6 +353,13 @@ func (c *YAMLConfig) parseSnapshotConfig() (*snapshotbuilder.SnapshotListenerCon
 		},
 	}
 
+	if snapshotConfig.Recorder != nil {
+		if snapshotConfig.Recorder.PostgresURL == "" {
+			return nil, errInvalidSnapshotRecorderConfig
+		}
+		streamCfg.SnapshotStoreURL = snapshotConfig.Recorder.PostgresURL
+	}
+
 	switch snapshotConfig.Mode {
 	case fullSnapshotMode, dataSnapshotMode, schemaSnapshotMode:
 	default:
@@ -371,10 +383,6 @@ func (c *YAMLConfig) parseSnapshotConfig() (*snapshotbuilder.SnapshotListenerCon
 		}
 	}
 
-	if snapshotConfig.RecorderEnabled {
-		streamCfg.SnapshotStoreURL = c.Source.Postgres.URL
-	}
-
 	return streamCfg, nil
 }
 
diff --git a/config_template.yaml b/config_template.yaml
index 7bf8e41..235f0cc 100644
--- a/config_template.yaml
+++ b/config_template.yaml
@@ -5,7 +5,8 @@ source:
     snapshot: # when mode is snapshot or snapshot_and_replication
       mode: full # options are data_and, schema or data
       tables: ["test", "test_schema.test", "another_schema.*"] # tables to snapshot, can be a list of table names or a pattern
-      recorder_enabled: true # whether to record snapshot status in a table under the pgstream schema in the source database
+      recorder:
+        postgres_url: "postgresql://user:password@localhost:5432/mydatabase" # URL of the database where the snapshot status is recorded
       snapshot_workers: 4 # number of schemas to be snapshotted in parallel
       data: # when mode is full or data
         schema_workers: 4 # number of schema tables to be snapshotted in parallel
diff --git a/pg2kafka.yaml b/pg2kafka.yaml
index b1c8ce1..372d6bd 100644
--- a/pg2kafka.yaml
+++ b/pg2kafka.yaml
@@ -5,7 +5,8 @@ source:
     snapshot: # when mode is snapshot or snapshot_and_replication
       mode: full # options are data_and, schema or data
       tables: ["*"] # tables to snapshot, can be a list of table names or a pattern
-      recorder_enabled: true # whether to record snapshot status in a table under the pgstream schema in the source database
+      recorder:
+        postgres_url: "postgres://postgres:postgres@localhost:5432?sslmode=disable" # URL of the database where the snapshot status is recorded
       snapshot_workers: 1 # number of schemas to be snapshotted in parallel
       data: # when mode is full or data
         schema_workers: 4 # number of schema tables to be snapshotted in parallel
diff --git a/pg2os.yaml b/pg2os.yaml
index ea201f5..696d1c1 100644
--- a/pg2os.yaml
+++ b/pg2os.yaml
@@ -5,7 +5,8 @@ source:
     snapshot: # when mode is snapshot or snapshot_and_replication
       mode: full # options are data_and, schema or data
       tables: ["*"] # tables to snapshot, can be a list of table names or a pattern
-      recorder_enabled: true # whether to record snapshot status in a table under the pgstream schema in the source database
+      recorder:
+        postgres_url: "postgres://postgres:postgres@localhost:5432?sslmode=disable" # URL of the database where the snapshot status is recorded
       snapshot_workers: 1 # number of schemas to be snapshotted in parallel
       data: # when mode is full or data
         schema_workers: 4 # number of schema tables to be snapshotted in parallel
diff --git a/pg2pg.yaml b/pg2pg.yaml
index 0f93359..bf52bd6 100644
--- a/pg2pg.yaml
+++ b/pg2pg.yaml
@@ -5,7 +5,8 @@ source:
     snapshot: # when mode is snapshot or snapshot_and_replication
       mode: full # options are data_and, schema or data
       tables: ["*"] # tables to snapshot, can be a list of table names or a pattern
-      recorder_enabled: true # whether to record snapshot status in a table under the pgstream schema in the source database
+      recorder:
+        postgres_url: "postgres://postgres:postgres@localhost:5432?sslmode=disable" # URL of the database where the snapshot status is recorded
       snapshot_workers: 1 # number of schemas to be snapshotted in parallel
       data: # when mode is full or data
         schema_workers: 4 # number of schema tables to be snapshotted in parallel
diff --git a/snapshot2pg.yaml b/snapshot2pg.yaml
new file mode 100644
index 0000000..b692831
--- /dev/null
+++ b/snapshot2pg.yaml
@@ -0,0 +1,40 @@
+source:
+  postgres:
+    url: "postgres://postgres:postgres@localhost:5432?sslmode=disable"
+    mode: snapshot # options are replication, snapshot or snapshot_and_replication
+    snapshot: # when mode is snapshot or snapshot_and_replication
+      mode: full # options are data_and, schema or data
+      tables: ["*"] # tables to snapshot, can be a list of table names or a pattern
+      recorder:
+        postgres_url: "postgres://postgres:postgres@localhost:5432?sslmode=disable" # URL of the database where the snapshot status is recorded
+      snapshot_workers: 1 # number of schemas to be snapshotted in parallel
+      data: # when mode is full or data
+        schema_workers: 4 # number of schema tables to be snapshotted in parallel
+        table_workers: 4 # number of workers to snapshot a table in parallel
+        batch_page_size: 1000 # number of pages to read per batch
+      schema: # when mode is full or schema
+        mode: pgdump_pgrestore # options are pgdump_pgrestore or schemalog
+        pgdump_pgrestore:
+          clean_target_db: true # whether to clean the target database before restoring
+
+target:
+  postgres:
+    url: "postgres://postgres:postgres@localhost:7654?sslmode=disable"
+    batch:
+      timeout: 5000 # batch timeout in milliseconds
+      size: 100 # number of messages in a batch
+    disable_triggers: false # whether to disable triggers on the target database
+    on_conflict_action: "nothing" # options are update, nothing or error
+
+modifiers:
+  injector:
+    enabled: false # whether to inject pgstream metadata into the WAL events
+  transformations:
+  - schema: public
+    table: test
+    column_transformers:
+      name:
+        name: greenmask_firstname
+        dynamic_parameters:
+          gender:
+            column: sex


Test patch:
diff --git a/cmd/config/config_yaml_test.go b/cmd/config/config_yaml_test.go
index 3aedd11..692b904 100644
--- a/cmd/config/config_yaml_test.go
+++ b/cmd/config/config_yaml_test.go
@@ -57,6 +57,22 @@ func TestYAMLConfig_toStreamConfig_ErrorCases(t *testing.T) {
 
 			wantErr: errUnsupportedSnapshotMode,
 		},
+		{
+			name: "err - invalid snapshot recorder config",
+			config: YAMLConfig{
+				Source: SourceConfig{
+					Postgres: &PostgresConfig{
+						Mode: snapshotMode,
+						Snapshot: &SnapshotConfig{
+							Mode:     fullSnapshotMode,
+							Recorder: &SnapshotRecorderConfig{},
+						},
+					},
+				},
+			},
+
+			wantErr: errInvalidSnapshotRecorderConfig,
+		},
 		{
 			name: "err - invalid postgres snapshot schema mode",
 			config: YAMLConfig{
diff --git a/cmd/config/helper_test.go b/cmd/config/helper_test.go
index 11cc4c4..fbcc860 100644
--- a/cmd/config/helper_test.go
+++ b/cmd/config/helper_test.go
@@ -25,6 +25,7 @@ func validateTestStreamConfig(t *testing.T, streamConfig *stream.Config) {
 	assert.Equal(t, uint(1000), streamConfig.Listener.Postgres.Snapshot.Generator.BatchPageSize)
 	assert.NotNil(t, streamConfig.Listener.Postgres.Snapshot.Schema.DumpRestore)
 	assert.True(t, streamConfig.Listener.Postgres.Snapshot.Schema.DumpRestore.CleanTargetDB)
+	assert.Equal(t, "postgresql://user:password@localhost:5432/mytargetdatabase", streamConfig.Listener.Postgres.Snapshot.SnapshotStoreURL)
 
 	assert.NotNil(t, streamConfig.Listener.Kafka)
 	assert.Equal(t, []string{"localhost:9092"}, streamConfig.Listener.Kafka.Reader.Conn.Servers)
diff --git a/cmd/config/test/test_config.env b/cmd/config/test/test_config.env
index e1197e2..44c67f5 100644
--- a/cmd/config/test/test_config.env
+++ b/cmd/config/test/test_config.env
@@ -4,7 +4,7 @@
 PGSTREAM_POSTGRES_LISTENER_URL="postgresql://user:password@localhost:5432/mydatabase"
 PGSTREAM_POSTGRES_REPLICATION_SLOT_NAME="pgstream_mydatabase_slot"
 PGSTREAM_POSTGRES_LISTENER_INITIAL_SNAPSHOT_ENABLED=true
-PGSTREAM_POSTGRES_INITIAL_SNAPSHOT_STORE_URL="postgresql://user:password@localhost:5432/mydatabase"
+PGSTREAM_POSTGRES_INITIAL_SNAPSHOT_STORE_URL="postgresql://user:password@localhost:5432/mytargetdatabase"
 PGSTREAM_POSTGRES_INITIAL_SNAPSHOT_TABLES="test test_schema.test another_schema.*"
 PGSTREAM_POSTGRES_INITIAL_SNAPSHOT_WORKERS=4
 PGSTREAM_POSTGRES_INITIAL_SNAPSHOT_SCHEMA_WORKERS=4
diff --git a/cmd/config/test/test_config.yaml b/cmd/config/test/test_config.yaml
index 4625bfb..f4dd321 100644
--- a/cmd/config/test/test_config.yaml
+++ b/cmd/config/test/test_config.yaml
@@ -5,7 +5,8 @@ source:
     snapshot: # when mode is snapshot or snapshot_and_replication
       mode: full # options are data_and, schema or data
       tables: ["test", "test_schema.test", "another_schema.*"] # tables to snapshot, can be a list of table names or a pattern
-      recorder_enabled: true # whether to record snapshot status in a table under the pgstream schema in the source database
+      recorder:
+        postgres_url: "postgresql://user:password@localhost:5432/mytargetdatabase" # URL of the database where the snapshot status is recorded
       snapshot_workers: 4 # number of schemas to be snapshotted in parallel
       data: # when mode is full or data
         schema_workers: 4 # number of schema tables to be snapshotted in parallel