postgres_cdc: add signalling support (without the need to restart)

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

@@ -99,6 +99,7 @@ input:
       role: "" # No default (optional)
       role_external_id: "" # No default (optional)
       roles: [] # No default (optional)
+    signal_table_name: ""
     auto_replay_nacks: true
     batching:
       count: 0
@@ -561,6 +562,61 @@ Optional external ID for the role assumption.
 
 *Default*: `""`
 
+=== `signal_table_name`
+
+The name of the table used to send control signals to the connector, excluding the schema. Leave empty (the default) to disable signalling entirely. The table must
+exist in the schema configured via the `schema` field and must have exactly these columns:
+
+- **id** — any type representable as a string (e.g. `SERIAL`, `BIGSERIAL`, `UUID`, `VARCHAR`)
+- **type** — `VARCHAR` — the signal type (see supported signals below)
+- **data** — `TEXT` — a JSON object containing signal parameters
+
+Create the table with:
+
+```sql
+CREATE TABLE <schema>.<signal_table_name> (
+    id   SERIAL PRIMARY KEY,
+    type VARCHAR(32),
+    data TEXT
+);
+```
+
+Signal rows are published as regular output messages (`operation=insert`, `table=<signal_table_name>`).
+To exclude them from downstream processing, filter on the `table` metadata field using a
+`mapping` processor:
+
+```yaml
+pipeline:
+  processors:
+    - mapping: |
+        root = if @table == "rpcn_signal_table" { deleted() } else { this }
+```
+
+**Supported signals**
+
+**`execute-snapshot`** — triggers a re-snapshot of one or more tables without dropping the
+replication slot. The `data` column must contain a JSON object with a
+`data-collections` key listing the fully-qualified tables (`schema.table`) to snapshot.
+`data-collections` must be non-empty; a signal with an empty or absent `data-collections`
+is ignored and streaming continues uninterrupted.
+
+```sql
+INSERT INTO dbo.rpcn_signal_table (type, data)
+VALUES ('execute-snapshot', '{"data-collections": ["dbo.events", "dbo.products"]}');
+```
+
+
+*Type*: `string`
+
+*Default*: `""`
+Requires version 4.102.0 or newer
+
+```yml
+# Examples
+
+signal_table_name: rpcn_signal_table
+```
+
 === `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/bench/Taskfile.yaml

@@ -64,6 +64,18 @@ tasks:
     cmds:
       - psql {{.PG_DSN}} -c "SELECT pg_drop_replication_slot('bench_slot') FROM pg_replication_slots WHERE slot_name = 'bench_slot';"
 
+  psql:create:signal-table:
+    desc: Create rpcn_signal_table in the public schema used for snapshot signalling
+    cmds:
+      - psql {{.PG_DSN}} -c "CREATE TABLE IF NOT EXISTS public.rpcn_signal_table (id SERIAL PRIMARY KEY, type VARCHAR(32), data TEXT);"
+
+  psql:signal:snapshot:
+    desc: Insert a snapshot signal into public.rpcn_signal_table to trigger a re-snapshot
+    deps: [psql:create:signal-table]
+    cmds:
+      - |
+        psql {{.PG_DSN}} -c "INSERT INTO public.rpcn_signal_table (type, data) VALUES ('execute-snapshot', '{\"data-collections\": [\"public.Cart\"]}');"
+
   # ── CDC / Snapshot benchmarks ────────────────────────────────────────────────
 
   bench:run:

internal/impl/postgresql/bench/benchmark_config.yaml

@@ -11,8 +11,8 @@ input:
       - products
       - cart
     slot_name: bench_slot
+    signal_table_name: rpcn_signal_table
     batching:
-      count: 1000
       period: 1s
 
 # pipeline:
@@ -22,14 +22,14 @@ input:
 #         root.metadata = @
 
 output:
-  processors:
-    - benchmark:
-        interval: 1s
-        count_bytes: true
-  # file:
-  #   path: "./benchmark_results.json"
-  #   codec: lines
-  drop: {}
+  # processors:
+  #   - benchmark:
+  #       interval: 1s
+  #       count_bytes: true
+  file:
+    path: "./benchmark_results.json"
+    codec: lines
+  # drop: {}
 
 logger:
   level: INFO

internal/impl/postgresql/bench/create.sql

@@ -1,4 +1,9 @@
 -- PostgreSQL Benchmark Setup Script
+CREATE TABLE IF NOT EXISTS public.rpcn_signal_table (
+    id SERIAL PRIMARY KEY,
+    type VARCHAR(32),
+    data TEXT
+);
 
 CREATE TABLE IF NOT EXISTS public.users (
     id            SERIAL PRIMARY KEY,

internal/impl/postgresql/control_signal.go

@@ -0,0 +1,150 @@
+// Copyright 2026 Redpanda Data, Inc.
+//
+// Licensed as a Redpanda Enterprise file under the Redpanda Community
+// License (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// https://github.com/redpanda-data/connect/blob/main/licenses/rcl.md
+
+package pgstream
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"strings"
+	"time"
+
+	"github.com/Jeffail/checkpoint"
+
+	"github.com/redpanda-data/benthos/v4/public/service"
+	"github.com/redpanda-data/connect/v4/internal/impl/postgresql/pglogicalstream"
+	"github.com/redpanda-data/connect/v4/internal/replication"
+)
+
+var _ replication.Signaller = (*pgSignaller)(nil)
+
+type pgSignaller struct {
+	*replication.ControlSignaller
+
+	schema          string
+	signalTableName string
+}
+
+// newPGSignaller creates a replication.Signaller that detects signal INSERTs on the given schema.tableName.
+func newPGSignaller(schema, tableName string, log *service.Logger) *pgSignaller {
+	s := replication.NewControlSignaller(log)
+	return &pgSignaller{ControlSignaller: s, schema: schema, signalTableName: tableName}
+}
+
+// Listen returns any actionable signal found. Signal rows are always
+// forwarded downstream as normal messages regardless of the outcome here.
+//
+// An empty tableName means signalling is disabled: every message is ignored.
+//
+// Only validated execute-snapshot signals return non-nil - everything else
+// (an unsupported type, or a validated no-op) returns (nil, nil) and is
+// acked immediately like any other message.
+func (s *pgSignaller) Listen(_ context.Context, signal any) (*replication.ControlSignal, error) {
+	if s.signalTableName == "" {
+		return nil, nil
+	}
+	msg, ok := signal.(pglogicalstream.StreamMessage)
+	if !ok {
+		return nil, nil
+	}
+	if msg.Schema != s.schema || msg.Table != s.signalTableName {
+		return nil, nil
+	}
+	if msg.Operation != pglogicalstream.InsertOpType {
+		return nil, nil
+	}
+
+	// deserialise control signal
+	row, ok := msg.Data.(map[string]any)
+	if !ok {
+		return nil, fmt.Errorf("expected map for %s message data, got %T", s.signalTableName, msg.Data)
+	}
+	dataStr, ok := row["data"].(string)
+	if !ok {
+		return nil, fmt.Errorf("expected string for %s.data column, got %T", s.signalTableName, row["data"])
+	}
+	var sig replication.ControlSignal
+	if err := json.Unmarshal([]byte(dataStr), &sig); err != nil {
+		return nil, fmt.Errorf("unmarshaling signal %s.data: %w", s.signalTableName, err)
+	}
+	sig.ID = fmt.Sprintf("%v", row["id"])
+	if sig.Type, ok = row["type"].(string); !ok {
+		return nil, errors.New("parsing 'type' data")
+	}
+
+	log := s.Log.With("id", sig.ID, "type", sig.Type)
+
+	if !sig.IsSnapshot() {
+		log.Infof("Signal %q received but not a recognized action, forwarding as a regular message", sig.Type)
+		return nil, nil
+	}
+
+	// Invalid or no-op signals are not returned as actionable.
+	if len(sig.DataCollections) == 0 {
+		log.Warnf("Signal %q received but data-collections is empty — ignoring, streaming continues uninterrupted", sig.Type)
+		return nil, nil
+	}
+	if len(tableNamesFromSchema(sig.DataCollections, s.schema)) == 0 {
+		log.Warnf("Signal %q received but data-collections %v matched no tables for schema %q — ignoring, streaming continues uninterrupted", sig.Type, sig.DataCollections, s.schema)
+		return nil, nil
+	}
+
+	log.Infof("Signal %q received: operation=%s lsn=%v", sig.Type, msg.Operation, msg.LSN)
+
+	if msg.LSN != nil {
+		sig.LSN = []byte(*msg.LSN)
+	}
+	return &sig, nil
+}
+
+// awaitCheckpointLSN blocks until checkpointer's highest resolved offset has reached or passed target,
+// meaning every message up to and including it has been acknowledged downstream. An empty target is a no-op.
+func awaitCheckpointLSN(ctx context.Context, cp *checkpoint.Capped[*string], target []byte, waitInterval time.Duration) error {
+	if len(target) == 0 {
+		return nil
+	}
+	targetLSN, err := pglogicalstream.ParseLSN(string(target))
+	if err != nil {
+		return fmt.Errorf("unable to parse target LSN: %w", err)
+	}
+
+	ticker := time.NewTicker(waitInterval)
+	defer ticker.Stop()
+	for {
+		if highest := cp.Highest(); highest != nil && *highest != nil {
+			if gotLSN, err := pglogicalstream.ParseLSN(**highest); err == nil && gotLSN >= targetLSN {
+				return nil
+			}
+		}
+		select {
+		case <-ticker.C:
+		case <-ctx.Done():
+			return ctx.Err()
+		}
+	}
+}
+
+func tableNamesFromSchema(collections []string, schema string) []string {
+	if len(collections) == 0 {
+		return nil
+	}
+	tables := make([]string, 0, len(collections))
+	for _, dc := range collections {
+		table := dc
+		if idx := strings.LastIndex(dc, "."); idx >= 0 {
+			if !strings.EqualFold(dc[:idx], schema) {
+				continue
+			}
+			table = dc[idx+1:]
+		}
+		tables = append(tables, table)
+	}
+	return tables
+}