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

[josephwoodward]

Adds support for signalling a re-snapshot, but fundamentally changes the way the PostgreSQL connector works as the snapshot/streaming modes are quite tightly coupled.

[claude]

Signal rows are emitted with operation=control_signal, contradicting the documented contract and this PR's own integration tests.

This line overwrites the message's operation metadata with control_signal, and the signal row is then flushed downstream via handleControlSignal → flushBatch/msgChan. But the field documentation states signal rows are published as regular messages with operation=insert:

• spec description:

  connect/internal/impl/postgresql/input_pg_stream.go

  Lines 215 to 217 in d824669

  	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

• generated docs:

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

  Lines 583 to 585 in d824669

  	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

And every actionable-signal assertion in the new integration test expects operation: "insert" for the rpcn_signal_table row, e.g.

connect/internal/impl/postgresql/control_signal_integration_test.go

Lines 341 to 364 in d824669

	require.ElementsMatch(t, received, []any{
	map[string]any{"operation": "insert", "table": "rpcn_signal_table", "lsn": "XXX/XXX"},
	map[string]any{"operation": "read", "table": "events"},
	map[string]any{"operation": "read", "table": "events"},
	})
	mu.Unlock()
	db.MustExec(`INSERT INTO dbo.rpcn_signal_table (type, data) VALUES ('execute-snapshot', '{"data-collections": ["dbo.events"]}')`)
	// Wait for the second re-snapshot: another signal row plus two snapshot reads,
	// accumulated on top of the previous three.
	assert.EventuallyWithT(t, func(c *assert.CollectT) {
	mu.Lock()
	defer mu.Unlock()
	assert.Len(c, received, 6)
	}, 25*time.Second, 100*time.Millisecond)
	mu.Lock()
	require.ElementsMatch(t, received, []any{
	map[string]any{"operation": "insert", "table": "rpcn_signal_table", "lsn": "XXX/XXX"},
	map[string]any{"operation": "read", "table": "events"},
	map[string]any{"operation": "read", "table": "events"},
	map[string]any{"operation": "insert", "table": "rpcn_signal_table", "lsn": "XXX/XXX"},
	map[string]any{"operation": "read", "table": "events"},

. As written those rows arrive as control_signal, so the documented contract is broken and the enabled-signalling integration tests would not match their expected sets.

Since handleControlSignal is dispatched on sig != nil (not on this metadata), the control_signal operation serves no internal purpose and only leaks to the output. Consider dropping this override so signal rows forward as ordinary inserts, consistent with the docs and tests.

[claude]

Unrelated benchmark output change looks like a committed local debugging tweak. Alongside the intended signal_table_name addition, this file also switches the benchmark output from drop: {} + the benchmark processor (which measures throughput) to writing rows into ./benchmark_results.json, and drops the count: 1000 batching bound (diff). These changes are unrelated to the signalling feature and disable the throughput measurement the bench harness exists for (§1.3.4). This looks like a local iteration left in by accident — worth confirming it's intended before merging.

[josephwoodward]

This was an experimental branch to understand how a "no restart" version would look and it carries too much complexity versus restarting the connector on a signal.