pgoutput-source-adapter

pgoutput-source-adapter adapts decoded pgoutput events into downstream change-event platform primitives.

The first supported target is the CDC Ecosystem:

Pgoutput::SourceAdapter::Cdc

It normalizes Pgoutput::Decoder::Events into CDC::Core::ChangeEvent and CDC::Core::TransactionEnvelope objects.

Boundary

The pgoutput family remains standalone:

pgoutput-client   -> PostgreSQL logical replication transport
pgoutput-parser   -> pgoutput protocol messages
pgoutput-decoder  -> typed Ruby row-change events

This gem is the adapter layer:

Pgoutput::Decoder::Events
        |
        v
Pgoutput::SourceAdapter::Cdc
        |
        v
CDC::Core::ChangeEvent / TransactionEnvelope

That keeps the lower-level pgoutput gems usable outside the CDC Ecosystem while still providing a clean bridge into cdc-core for users building CDC platforms.

Installation

gem "pgoutput-source-adapter"

require "pgoutput/source_adapter"

The generated bundle gem require path also works:

require "pgoutput/source/adapter"

Usage

Normalize a decoded insert event:

adapter = Pgoutput::SourceAdapter::Cdc.new
change_event = adapter.normalize(decoded_insert)

change_event.operation
# => :insert

change_event.schema
change_event.table
change_event.new_values

Normalize a transaction-shaped batch:

results = adapter.normalize_many([
  decoded_begin,
  decoded_insert,
  decoded_update,
  decoded_commit
])

envelope = results.first
# => CDC::Core::TransactionEnvelope

Normalize an unbounded replication stream incrementally:

inputs = Enumerator.new do |stream|
  runner.start do |payload, metadata|
    decoded = decoder.decode(parser.process(payload))
    stream << adapter.stream_event(decoded, source_position: metadata.wal_end_lsn)
  end
end

adapter.each_normalized(inputs) do |work|
  processor.process(work)
end

Transaction buffering belongs to the source adapter. Changes outside a transaction are yielded immediately; changes between Begin and Commit are yielded as a CDC::Core::TransactionEnvelope. normalize_many uses the same normalization path for finite batches.

Each change inside a transaction receives a stable, zero-based sequence_number. The ordinal follows decoded row-event order, resets at each Begin, and is reproduced when the same transaction input is replayed. Changes normalized outside transaction boundaries retain a nil sequence number. Downstream consumers can include this discriminator in deterministic event IDs so otherwise identical changes in the same transaction remain distinct.

Primary keys

For update and delete events, pgoutput may provide an old-key tuple. When it does, the complete tuple is used as CDC::Core::ChangeEvent#primary_key, including composite and non-id replica keys.

For inserts and events without an old-key tuple, the default resolver uses relation key-column metadata when the decoded event exposes replica_identity_columns, key_columns, or a relation whose column flags mark replica-key columns. The adapter extracts every declared column and raises Pgoutput::SourceAdapter::Error rather than returning a partial key.

When decoded events do not carry relation key metadata, configure the reusable schema-aware resolver. Relations may be keyed by relation id, [schema, table], or a qualified name:

resolver = Pgoutput::SourceAdapter::ReplicaIdentityResolver.new(
  ["public", "memberships"] => ["tenant_id", "member_uuid"],
  "public.accounts" => ["account_uuid"]
)

adapter = Pgoutput::SourceAdapter::Cdc.new(
  primary_key_resolver: resolver
)

The legacy id / "id" inference remains as a compatibility fallback only when neither relation metadata nor an explicit resolver is available. Custom callables remain supported through primary_key_resolver:.

Metadata

Each normalized event includes pgoutput metadata:

{
  "source" => "pgoutput",
  "relation_id" => 123,
  "pgoutput_event" => "Insert"
}

Additional metadata can be injected:

adapter = Pgoutput::SourceAdapter::Cdc.new(
  metadata_builder: ->(_event) { { pipeline: "default" } }
)

Public namespace

Pgoutput::SourceAdapter
Pgoutput::SourceAdapter::Cdc
Pgoutput::SourceAdapter::ReplicaIdentityResolver

A compatibility alias is also provided for the generated gem path:

Pgoutput::Source::Adapter

Non-goals

This gem does not:

• connect to PostgreSQL
• parse pgoutput protocol messages
• decode PostgreSQL values
• run processors
• manage replication slots
• persist sink data

Those responsibilities belong to pgoutput-client, pgoutput-parser, pgoutput-decoder, runtime gems, or application code.

Development

bundle exec rake

The default quality task enforces 100% line and branch coverage and 100% YARD API documentation coverage. Run bundle exec yard stats --list-undoc to inspect the documented public API objects.

License

MIT.