postgres_cdc: Add signal detection

[josephwoodward]

This pull request adds the initial signalling detection mechanism, first supporting log signal type that can be used to verify integration. This acts as the foundation to support reshapshotting and incremental snapshotting.

[claude]

Startup validation queries information_schema.columns but only checks column names, so a wrong column type is never caught at config/connect time. The consequence is that the signal table is accepted, signals silently never take effect, and the user instead gets a per-row p.logger.Errorf("failed to detect control signal in change event: ...") for every row inserted into the signal table, forever. The new docs make the gap explicit ("startup validation checks column names only, not types, so a wrong column type (e.g. data JSONB instead of TEXT) is only caught at runtime, on the first signal row read"), and signaller_integration_test.go's "logs a per-row error when data column has the wrong type" subtest asserts exactly that behaviour.

This conflicts with CONTRIBUTING.md §1.2.4 — "Strongly lints and validates user-provided configuration, clearly telling users of any problems" — and §1.1.3's "don't make me think": a misconfiguration that is fully detectable from the catalog row already being read is deferred to a repeating runtime error log.

Suggested fix: select data_type alongside column_name in the same query and reject at startup when type/data are not string-ish types (text, character varying, character), with an error naming the offending column and its actual type. That turns the documented runtime failure mode into a startup config error and lets the docs drop the caveat.

Anchor:

connect/internal/impl/postgresql/pglogicalstream/logical_stream.go

Lines 931 to 947 in b0f2773

	}
	sql := "SELECT column_name FROM information_schema.columns WHERE table_schema = $1 AND table_name = $2"
	query, err := sanitize.SQLQuery(sql, wireSchema, wireSignalTable)
	if err != nil {
	return TableFQN{}, err
	}
	res, err := stream.pgConn.Exec(ctx, query).ReadAll()
	if err != nil {
	return TableFQN{}, fmt.Errorf("checking signal table %s columns: %w", signalTable, err)
	}
	if len(res) == 0 || len(res[0].Rows) == 0 {
	return signalTable, fmt.Errorf("signal table %s does not exist", signalTable)
	}
	columns := make(map[string]bool, len(res[0].Rows))
	for _, row := range res[0].Rows {
	columns[string(row[0])] = true