Class: Pgoutput::SourceAdapter::ReplicaIdentityResolver

Inherits:
Object
  • Object
show all
Defined in:
lib/pgoutput/source_adapter/replica_identity_resolver.rb

Overview

Resolves schema-aware replica keys from decoded row values.

Relations may be keyed by pgoutput relation id, [schema, table], or a "schema.table" string. Each value is the ordered list of replica-key columns. Relation-id mappings take precedence because they identify the exact relation metadata seen by the replication stream.

Examples:

Resolve a composite key

resolver = ReplicaIdentityResolver.new(
  ["public", "memberships"] => ["tenant_id", "member_uuid"]
)

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(relations) ⇒ void

Parameters:

  • relations (Hash{Integer, Array<String>, Array<String>=>Array<String>}) —

    relation identity to ordered replica-key column names



44
45
46
47
 
# File 'lib/pgoutput/source_adapter/replica_identity_resolver.rb', line 44

def initialize(relations)
  @relations = relations.to_h.transform_values { |columns| normalize_columns(columns) }.freeze
  freeze
end

Instance Attribute Details

#relations ⇒ Hash (readonly)

Normalized relation keys mapped to ordered replica-identity columns.

Returns:

  • (Hash) 


22
23
24
 
# File 'lib/pgoutput/source_adapter/replica_identity_resolver.rb', line 22

def relations
  @relations
end

Class Method Details

.extract(event, values, columns) ⇒ Hash{String=>Object}

Extract a complete replica key from an ordered column list.

Parameters:

  • event (Object) —

    decoded row event with schema/table metadata

  • values (Hash, nil) —

    decoded row values

  • columns (Array<String>) —

    ordered replica-key column names

Returns:

  • (Hash{String=>Object}) —

    complete replica key

Raises:

  • (Error) —

    when values or any configured column are unavailable



31
32
33
34
35
36
37
38
39
 
# File 'lib/pgoutput/source_adapter/replica_identity_resolver.rb', line 31

def self.extract(event, values, columns)
  unless values.respond_to?(:key?)
    raise Error, "replica identity values are unavailable for #{qualified_name(event)}"
  end

  columns.to_h do |column|
    [column, fetch_column(event, values, column)]
  end
end

Instance Method Details

#call(event, values) ⇒ Hash{String=>Object}?

Extract the configured replica key for a decoded row event.

Parameters:

  • event (Object) —

    decoded row event with relation/schema/table metadata

  • values (Hash, nil) —

    decoded row values

Returns:

  • (Hash{String=>Object}, nil) —

    complete replica key, or nil when the relation is not configured

Raises:

  • (Error) —

    when a configured key column is absent from row values



56
57
58
59
60
61
 
# File 'lib/pgoutput/source_adapter/replica_identity_resolver.rb', line 56

def call(event, values)
  columns = columns_for(event)
  return nil unless columns

  self.class.extract(event, values, columns)
end