Merge pull request #3144 from rockwotj/pg_snap

pgcdc: fix snapshot consistency

Author: rockwotj

CHANGELOG.md

@@ -13,13 +13,20 @@ All notable changes to this project will be documented in this file.
 - Field `avro.preserve_logical_types` for processor `schema_registry_decode` was added to preserve logical types instead of decoding them as their primitive representation. (@rockwotj)
 - Processor `schema_registry_decode` now adds metadata `schema_id` for the schema's ID in the schema registry. (@rockwotj)
 - Field `schema_evolution.processors` added to `snowpipe_streaming` to support side effects or enrichment during schema evolution. (@rockwotj)
+- Field `unchanged_toast_value` added to `postgres_cdc` to control the value substituted for unchanged toast values when a table does not have full replica identity. (@rockwotj)
+
+### Fixed
+
+- Fix a snapshot stream consistency issue with `postgres_cdc` where data could be missed if writes were happening during the snapshot phase. (@rockwotj)
+- Fix an issue where `@table` metadata was quoted for the snapshot phase in `postgres_cdc`. (@rockwotj)
 
 ### Changed
 
 - Field `avro_raw_json` was deprecated in favor of `avro.raw_unions` for processor `schema_registry_decode`. (@rockwotj)
 - The `snowpipe_streaming` output now has better error handling for authentication failures when uploading to cloud storage. (@rockwotj)
 - Field `schema_evolution.new_column_type_mapping` for `snowpipe_streaming` is deprecated and can be replaced with `schema_evolution.processors`. (@rockwotj)
 - Increased the default values for `max_message_bytes` and `broker_write_max_bytes` by using IEC units instead of SI units. This better matches defaults in Redpanda and Kafka. (@rockwotj)
+- Dropped support for postgres 10 and 11 in `postgres_cdc`. (@rockwotj)
 
 ## 4.45.1 - 2025-01-17
 

docs/modules/components/pages/inputs/pg_stream.adoc

@@ -88,6 +88,7 @@ input:
     pg_standby_timeout: 10s
     pg_wal_monitor_interval: 3s
     max_parallel_snapshot_tables: 1
+    unchanged_toast_value: null
     auto_replay_nacks: true
     batching:
       count: 0
@@ -285,6 +286,21 @@ Int specifies a number of tables that will be processed in parallel during the s
 
 *Default*: `1`
 
+=== `unchanged_toast_value`
+
+The value to emit when there are unchanged TOAST values in the stream. This occurs for updates and deletes where REPLICA IDENTITY is not FULL.
+
+
+*Type*: `unknown`
+
+*Default*: `null`
+
+```yml
+# Examples
+
+unchanged_toast_value: __redpanda_connect_unchanged_toast_value__
+```
+
 === `auto_replay_nacks`
 
 Whether messages that are rejected (nacked) at the output level should be automatically replayed indefinitely, eventually resulting in back pressure if the cause of the rejections is persistent. If set to `false` these messages will instead be deleted. Disabling auto replays can greatly improve memory efficiency of high throughput streams as the original shape of the data can be discarded immediately upon consumption and mutation.

docs/modules/components/pages/inputs/postgres_cdc.adoc

@@ -83,6 +83,7 @@ input:
     pg_standby_timeout: 10s
     pg_wal_monitor_interval: 3s
     max_parallel_snapshot_tables: 1
+    unchanged_toast_value: null
     auto_replay_nacks: true
     batching:
       count: 0
@@ -280,6 +281,21 @@ Int specifies a number of tables that will be processed in parallel during the s
 
 *Default*: `1`
 
+=== `unchanged_toast_value`
+
+The value to emit when there are unchanged TOAST values in the stream. This occurs for updates and deletes where REPLICA IDENTITY is not FULL.
+
+
+*Type*: `unknown`
+
+*Default*: `null`
+
+```yml
+# Examples
+
+unchanged_toast_value: __redpanda_connect_unchanged_toast_value__
+```
+
 === `auto_replay_nacks`
 
 Whether messages that are rejected (nacked) at the output level should be automatically replayed indefinitely, eventually resulting in back pressure if the cause of the rejections is persistent. If set to `false` these messages will instead be deleted. Disabling auto replays can greatly improve memory efficiency of high throughput streams as the original shape of the data can be discarded immediately upon consumption and mutation.

internal/impl/postgresql/input_pg_stream.go

@@ -40,6 +40,7 @@ const (
 	fieldSlotName                  = "slot_name"
 	fieldBatching                  = "batching"
 	fieldMaxParallelSnapshotTables = "max_parallel_snapshot_tables"
+	fieldUnchangedToastValue       = "unchanged_toast_value"
 
 	shutdownTimeout = 5 * time.Second
 )
@@ -111,6 +112,11 @@ This input adds the following metadata fields to each message:
 		Field(service.NewIntField(fieldMaxParallelSnapshotTables).
 			Description("Int specifies a number of tables that will be processed in parallel during the snapshot processing stage").
 			Default(1)).
+		Field(service.NewAnyField(fieldUnchangedToastValue).
+			Description("The value to emit when there are unchanged TOAST values in the stream. This occurs for updates and deletes where REPLICA IDENTITY is not FULL.").
+			Default(nil).
+			Example("__redpanda_connect_unchanged_toast_value__").
+			Advanced()).
 		Field(service.NewAutoRetryNacksToggleField()).
 		Field(service.NewBatchPolicyField(fieldBatching))
 }
@@ -131,6 +137,7 @@ func newPgStreamInput(conf *service.ParsedConfig, mgr *service.Resources) (s ser
 		maxParallelSnapshotTables int
 		pgStandbyTimeout          time.Duration
 		batching                  service.BatchPolicy
+		unchangedToastValue       any
 	)
 
 	if err := license.CheckRunningEnterprise(mgr); err != nil {
@@ -206,6 +213,10 @@ func newPgStreamInput(conf *service.ParsedConfig, mgr *service.Resources) (s ser
 		return nil, err
 	}
 
+	if unchangedToastValue, err = conf.FieldAny(fieldUnchangedToastValue); err != nil {
+		return nil, err
+	}
+
 	pgConnConfig, err := pgconn.ParseConfigWithOptions(dsn, pgconn.ParseConfigOptions{
 		// Don't support dynamic reading of password
 		GetSSLPassword: func(context.Context) string { return "" },
@@ -237,6 +248,7 @@ func newPgStreamInput(conf *service.ParsedConfig, mgr *service.Resources) (s ser
 			WalMonitorInterval:         walMonitorInterval,
 			MaxParallelSnapshotTables:  maxParallelSnapshotTables,
 			Logger:                     mgr.Logger(),
+			UnchangedToastValue:        unchangedToastValue,
 		},
 		batching:        batching,
 		checkpointLimit: checkpointLimit,