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.

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.