Class: Mammoth::Sources::PostgresPublicationInspector

Inherits:
Object
  • Object
show all
Defined in:
lib/mammoth/sources/postgres_publication_inspector.rb

Overview

Read-only PostgreSQL publication and replica-identity catalog inspector.

Constant Summary collapse

QUERY = 
<<~SQL
  SELECT
    relation.oid AS relation_id,
    namespace.nspname AS schema_name,
    relation.relname AS table_name,
    bool_or(publication.pubupdate) AS publishes_updates,
    bool_or(publication.pubdelete) AS publishes_deletes,
    relation.relreplident AS replica_identity,
    CASE relation.relreplident
      WHEN 'd' THEN (
        SELECT COALESCE(jsonb_agg(attribute.attname ORDER BY key.ordinality), '[]'::jsonb)
        FROM pg_index AS identity_columns_index
        JOIN LATERAL unnest(identity_columns_index.indkey)
          WITH ORDINALITY AS key(attribute_number, ordinality) ON true
        JOIN pg_attribute AS attribute
          ON attribute.attrelid = relation.oid
         AND attribute.attnum = key.attribute_number
        WHERE identity_columns_index.indrelid = relation.oid
          AND identity_columns_index.indisprimary
          AND identity_columns_index.indisvalid
          AND identity_columns_index.indisready
          AND identity_columns_index.indislive
      )
      WHEN 'i' THEN (
        SELECT COALESCE(jsonb_agg(attribute.attname ORDER BY key.ordinality), '[]'::jsonb)
        FROM pg_index AS identity_columns_index
        JOIN LATERAL unnest(identity_columns_index.indkey)
          WITH ORDINALITY AS key(attribute_number, ordinality) ON true
        JOIN pg_attribute AS attribute
          ON attribute.attrelid = relation.oid
         AND attribute.attnum = key.attribute_number
        WHERE identity_columns_index.indrelid = relation.oid
          AND identity_columns_index.indisreplident
          AND identity_columns_index.indisvalid
          AND identity_columns_index.indisready
          AND identity_columns_index.indislive
      )
      WHEN 'f' THEN (
        SELECT COALESCE(jsonb_agg(attribute.attname ORDER BY attribute.attnum), '[]'::jsonb)
        FROM pg_attribute AS attribute
        WHERE attribute.attrelid = relation.oid
          AND attribute.attnum > 0
          AND NOT attribute.attisdropped
      )
      ELSE '[]'::jsonb
    END AS replica_identity_columns,
    EXISTS (
      SELECT 1
      FROM pg_index AS primary_index
      WHERE primary_index.indrelid = relation.oid
        AND primary_index.indisprimary
        AND primary_index.indisvalid
        AND primary_index.indisready
        AND primary_index.indislive
    ) AS primary_key_usable,
    EXISTS (
      SELECT 1
      FROM pg_index AS identity_index
      WHERE identity_index.indrelid = relation.oid
        AND identity_index.indisreplident
        AND identity_index.indisvalid
        AND identity_index.indisready
        AND identity_index.indislive
    ) AS replica_identity_index_usable
  FROM pg_publication_tables AS publication_table
  JOIN pg_publication AS publication
    ON publication.pubname = publication_table.pubname
  JOIN pg_namespace AS namespace
    ON namespace.nspname = publication_table.schemaname
  JOIN pg_class AS relation
    ON relation.relnamespace = namespace.oid
   AND relation.relname = publication_table.tablename
  WHERE publication.pubname IN (
    SELECT jsonb_array_elements_text($1::jsonb)
  )
  GROUP BY namespace.nspname, relation.relname, relation.oid, relation.relreplident
  ORDER BY namespace.nspname, relation.relname
SQL

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(database_url:, connection_factory: nil) ⇒ PostgresPublicationInspector

Returns a new instance of PostgresPublicationInspector.

Parameters:

  • database_url (String) —

    PostgreSQL connection URL

  • connection_factory (#call, nil) (defaults to: nil) —

    optional connection factory for tests



156
157
158
159
 
# File 'lib/mammoth/sources/postgres_publication_inspector.rb', line 156

def initialize(database_url:, connection_factory: nil)
  @database_url = database_url
  @connection_factory = connection_factory
end

Instance Attribute Details

#database_url ⇒ Object (readonly)

Returns the value of attribute database_url.



152
153
154
 
# File 'lib/mammoth/sources/postgres_publication_inspector.rb', line 152

def database_url
  @database_url
end

Instance Method Details

#inspect(publication_names) ⇒ Array<PostgresPublicationTable>

Inspect tables included by the configured publications.

Parameters:

  • publication_names (Array<String>) —

    configured publication names

Returns:



165
166
167
168
169
170
171
172
173
 
# File 'lib/mammoth/sources/postgres_publication_inspector.rb', line 165

def inspect(publication_names)
  connection = open_connection
  rows = connection.exec_params(QUERY, [JSON.generate(publication_names)])
  rows.map { |row| build_table(row) }
rescue pg_error_class => e
  raise ReplicationError, "PostgreSQL publication inspection failed: #{e.message}"
ensure
  connection&.close unless connection&.finished?
end