Skip to content

Add option for ignoring processing errors #559

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 4 commits into from
Oct 14, 2025
Merged

Add option for ignoring processing errors #559

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
ea0bd44
Add option for ignoring processing errors
tsg Oct 11, 2025
21c35b4
Simplify logging
eminano Oct 14, 2025
17eff9d
Add tests
eminano Oct 14, 2025
135cb35
Update docs
eminano Oct 14, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 9 additions & 5 deletions cmd/config/config_env.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -49,6 +49,8 @@ func init() {
viper.BindEnv("PGSTREAM_POSTGRES_SNAPSHOT_WORKERS")
viper.BindEnv("PGSTREAM_POSTGRES_SNAPSHOT_STORE_URL")
viper.BindEnv("PGSTREAM_POSTGRES_SNAPSHOT_STORE_REPEATABLE")
viper.BindEnv("PGSTREAM_POSTGRES_SNAPSHOT_IGNORE_ROW_PROCESSING_ERRORS")
viper.BindEnv("PGSTREAM_POSTGRES_SNAPSHOT_LOG_ROW_ON_ERROR")
viper.BindEnv("PGSTREAM_POSTGRES_SNAPSHOT_USE_SCHEMALOG")
viper.BindEnv("PGSTREAM_POSTGRES_SNAPSHOT_INCLUDE_GLOBAL_DB_OBJECTS")
viper.BindEnv("PGSTREAM_POSTGRES_SNAPSHOT_ROLE")
Expand Down Expand Up @@ -219,11 +221,13 @@ func parseSnapshotConfig(pgURL string) (*snapshotbuilder.SnapshotListenerConfig,

cfg := &snapshotbuilder.SnapshotListenerConfig{
Generator: pgsnapshotgenerator.Config{
URL: pgURL,
BatchBytes: viper.GetUint64("PGSTREAM_POSTGRES_SNAPSHOT_BATCH_BYTES"),
SchemaWorkers: viper.GetUint("PGSTREAM_POSTGRES_SNAPSHOT_SCHEMA_WORKERS"),
TableWorkers: viper.GetUint("PGSTREAM_POSTGRES_SNAPSHOT_TABLE_WORKERS"),
SnapshotWorkers: viper.GetUint("PGSTREAM_POSTGRES_SNAPSHOT_WORKERS"),
URL: pgURL,
BatchBytes: viper.GetUint64("PGSTREAM_POSTGRES_SNAPSHOT_BATCH_BYTES"),
SchemaWorkers: viper.GetUint("PGSTREAM_POSTGRES_SNAPSHOT_SCHEMA_WORKERS"),
TableWorkers: viper.GetUint("PGSTREAM_POSTGRES_SNAPSHOT_TABLE_WORKERS"),
SnapshotWorkers: viper.GetUint("PGSTREAM_POSTGRES_SNAPSHOT_WORKERS"),
IgnoreRowProcessingErrors: viper.GetBool("PGSTREAM_POSTGRES_SNAPSHOT_IGNORE_ROW_PROCESSING_ERRORS"),
LogRowOnError: viper.GetBool("PGSTREAM_POSTGRES_SNAPSHOT_LOG_ROW_ON_ERROR"),
},
Adapter: adapter.SnapshotConfig{
Tables: viper.GetStringSlice("PGSTREAM_POSTGRES_SNAPSHOT_TABLES"),
Expand Down
10 changes: 7 additions & 3 deletions cmd/config/config_yaml.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -87,9 +87,11 @@ type SnapshotRecorderConfig struct {
}

type SnapshotDataConfig struct {
SchemaWorkers int `mapstructure:"schema_workers" yaml:"schema_workers"`
TableWorkers int `mapstructure:"table_workers" yaml:"table_workers"`
BatchBytes uint64 `mapstructure:"batch_bytes" yaml:"batch_bytes"`
SchemaWorkers int `mapstructure:"schema_workers" yaml:"schema_workers"`
TableWorkers int `mapstructure:"table_workers" yaml:"table_workers"`
BatchBytes uint64 `mapstructure:"batch_bytes" yaml:"batch_bytes"`
IgnoreRowProcessingErrors bool `mapstructure:"ignore_row_processing_errors" yaml:"ignore_row_processing_errors"`
LogRowOnError bool `mapstructure:"log_row_on_error" yaml:"log_row_on_error"`
}

type SnapshotSchemaConfig struct {
Expand Down Expand Up @@ -493,6 +495,8 @@ func (c *YAMLConfig) parseDataSnapshotConfig() pgsnapshotgenerator.Config {
cfg.BatchBytes = snapshotCfg.Data.BatchBytes
cfg.SchemaWorkers = uint(snapshotCfg.Data.SchemaWorkers)
cfg.TableWorkers = uint(snapshotCfg.Data.TableWorkers)
cfg.IgnoreRowProcessingErrors = snapshotCfg.Data.IgnoreRowProcessingErrors
cfg.LogRowOnError = snapshotCfg.Data.LogRowOnError
}

return cfg
Expand Down
2 changes: 2 additions & 0 deletions config_template.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,8 @@ source:
schema_workers: 4 # number of schema tables to be snapshotted in parallel. Defaults to 4
table_workers: 4 # number of workers to snapshot a table in parallel. Defaults to 4
batch_bytes: 83886080 # bytes to read per batch (defaults to 80MiB)
ignore_row_processing_errors: false # if true, log errors during row processing and continue. Defaults to false. Warning: can result in consistency errors.
log_row_on_error: false # if true, log the row data as JSON when an error occurs. Defaults to false
schema: # when mode is full or schema
mode: pgdump_pgrestore # options are pgdump_pgrestore or schemalog
pgdump_pgrestore:
Expand Down
Loading
Loading