Class: Exwiw::Adapter::PostgresqlAdapter

Inherits:

Base

• Object
• Base
• Exwiw::Adapter::PostgresqlAdapter

Defined in:

lib/exwiw/adapter/postgresql_adapter.rb

Instance Attribute Summary

Attributes inherited from Base

#connection_config

Instance Method Summary

• #build_query(table, dump_target, table_by_name) ⇒ Object
• #compile_ast(query_ast, select_cast_to: nil) ⇒ Object
• #dump_schema(ordered_tables, output_path) ⇒ Object
• #execute(query_ast) ⇒ Object
• #explain(query_ast) ⇒ Object
• #post_insert_sql(table) ⇒ Object

  Transcribe the FROM-side sequence cursor backing ‘table.primary_key` onto the import target.

• #to_bulk_delete(select_query_ast, table) ⇒ Object
• #to_bulk_insert(results, table) ⇒ Object
• #to_copy_from_stdin(results, table) ⇒ Object

Methods inherited from Base

#commented_sql, #describe_query, #dumpable?, #initialize, #output_extension, #pre_insert_sql, #query_comment_text, #schema_output_extension, #sql_query_comment, #supports_bulk_delete?, table_config_class, #validate_as_dump_target!

Constructor Details

This class inherits a constructor from Exwiw::Adapter::Base

Instance Method Details

#build_query(table, dump_target, table_by_name) ⇒ Object

# File 'lib/exwiw/adapter/postgresql_adapter.rb', line 6

def build_query(table, dump_target, table_by_name)
  Exwiw::QueryAstBuilder.run(table.name, table_by_name, dump_target, @logger)
end

#compile_ast(query_ast, select_cast_to: nil) ⇒ Object

Raises:

• (NotImplementedError)

# File 'lib/exwiw/adapter/postgresql_adapter.rb', line 231

def compile_ast(query_ast, select_cast_to: nil)
  raise NotImplementedError unless query_ast.is_a?(Exwiw::QueryAst::Select)

  sql = "SELECT "
  sql += if query_ast.select_all
           "*"
         else
           cols = query_ast.columns.map { |col| compile_column_name(query_ast, col) }
           cols = cols.map { |c| "#{c}::#{select_cast_to}" } if select_cast_to
           cols.join(', ')
         end
  sql += " FROM #{query_ast.from_table_name}"

  query_ast.join_clauses.each do |join|
    fk_expr = "#{join.base_table_name}.#{join.foreign_key}"
    pk_expr = "#{join.join_table_name}.#{join.primary_key}"
    if types_need_cast?(
      column_pg_type(join.base_table_name, join.foreign_key),
      column_pg_type(join.join_table_name, join.primary_key)
    )
      fk_expr = "#{fk_expr}::text"
      pk_expr = "#{pk_expr}::text"
    end
    sql += " JOIN #{join.join_table_name} ON #{fk_expr} = #{pk_expr}"

    join.where_clauses.each do |where|
      compiled_where_condition = compile_where_condition(where, join.join_table_name)
      sql += " AND #{compiled_where_condition}"
    end

    # base_where_clauses is compiled against the joined-from table
    # (base_table_name), e.g. the type-column filter on a polymorphic
    # source table.
    join.base_where_clauses.each do |where|
      compiled_where_condition = compile_where_condition(where, join.base_table_name)
      sql += " AND #{compiled_where_condition}"
    end
  end

  if query_ast.where_clauses.any?
    sql += " WHERE "
    sql += query_ast.where_clauses.map { |where| compile_where_condition(where, query_ast.from_table_name) }.join(' AND ')
  end

  sql
end

#dump_schema(ordered_tables, output_path) ⇒ Object

# File 'lib/exwiw/adapter/postgresql_adapter.rb', line 24

def dump_schema(ordered_tables, output_path)
  require 'open3'

  table_names = ordered_tables.map(&:name)
  if table_names.empty?
    File.write(output_path, "-- Auto-generated by exwiw. No tables in scope.\n")
    return
  end

  cmd = [
    'pg_dump',
    "--host=#{@connection_config.host}",
    "--port=#{@connection_config.port}",
    "--username=#{@connection_config.user}",
    '--schema-only',
    '--no-owner',
    '--no-acl',
    *table_names.flat_map { |t| ['--table', t] },
    @connection_config.database_name,
  ]
  env = { 'PGPASSWORD' => @connection_config.password.to_s }

  @logger.debug("  Running pg_dump for #{table_names.size} table(s)...")
  stdout, stderr, status = Open3.capture3(env, *cmd)
  unless status.success?
    if stderr.include?('command not found') || stderr.empty?
      raise "Failed to run `pg_dump`. Ensure the postgresql client is installed and on PATH. stderr: #{stderr}"
    end
    raise "pg_dump failed (exit #{status.exitstatus}): #{stderr}"
  end

  # Enums are prepended first, then extensions — so extensions end up at the top of the output.
  enum_types = query_enum_types(table_names)
  unless enum_types.empty?
    enum_ddl = DdlPostprocessor.create_type_enum_statements(enum_types)
    @logger.debug("  Found #{enum_types.size} enum type(s) to prepend.")
    stdout = enum_ddl + stdout
  end

  extensions = query_extensions
  unless extensions.empty?
    ext_ddl = extensions.map do |extname, schema|
      stmt = "CREATE EXTENSION IF NOT EXISTS #{connection.quote_ident(extname)}"
      stmt += " SCHEMA #{connection.quote_ident(schema)}" unless schema == "public"
      # Best-effort prepend: a restore target that genuinely cannot create the
      # extension should not abort the whole restore. Two such cases are caught:
      #   feature_not_supported (0A000) -- the extension's binaries are unavailable
      #   invalid_schema_name   (3F000) -- the extension's required schema is absent
      # insufficient_privilege (42501) is deliberately NOT caught: a restore role
      # lacking CREATE privilege is a misconfiguration to fix, not to skip silently.
      # The skip is re-raised as a WARNING so it surfaces in the restore logs
      # instead of vanishing.
      warning = connection.escape_literal("exwiw: skipped CREATE EXTENSION #{extname} (SQLSTATE %): %")
      "DO $$ BEGIN #{stmt}; " \
        "EXCEPTION WHEN feature_not_supported OR invalid_schema_name THEN " \
        "RAISE WARNING #{warning}, SQLSTATE, SQLERRM; END $$;"
    end.join("\n") + "\n\n"
    @logger.debug("  Found #{extensions.size} extension(s) to prepend.")
    stdout = ext_ddl + stdout
  end

  idempotent = stdout
  idempotent = DdlPostprocessor.add_if_not_exists_to_create_schema(idempotent)
  idempotent = DdlPostprocessor.add_if_not_exists_to_create_sequence(idempotent)
  idempotent = DdlPostprocessor.add_if_not_exists_to_create_table(idempotent)
  idempotent = DdlPostprocessor.add_if_not_exists_to_create_index(idempotent)
  idempotent = DdlPostprocessor.wrap_add_constraint_in_do_block(idempotent)
  idempotent = DdlPostprocessor.strip_triggers(idempotent)

  File.open(output_path, 'w') do |file|
    file.puts("-- Auto-generated by exwiw via pg_dump. Idempotent DDL for postgresql.")
    file.write(idempotent)
  end
  @logger.info("  Wrote schema for #{table_names.size} table(s) to #{output_path}.")
end

#execute(query_ast) ⇒ Object

# File 'lib/exwiw/adapter/postgresql_adapter.rb', line 10

def execute(query_ast)
  sql = commented_sql(query_ast)

  @logger.debug("  Executing SQL: \n#{sql}")
  connection.exec(sql).values
end

#explain(query_ast) ⇒ Object

# File 'lib/exwiw/adapter/postgresql_adapter.rb', line 17

def explain(query_ast)
  sql = commented_sql(query_ast)

  @logger.debug("  Executing EXPLAIN: \n#{sql}")
  connection.exec("EXPLAIN #{sql}").values.map(&:first).join("\n")
end

#post_insert_sql(table) ⇒ Object

Transcribe the FROM-side sequence cursor backing ‘table.primary_key` onto the import target. Without this, importing into a clean DB leaves the sequence at 1 while the inserted rows occupy higher IDs, so the next default-PK INSERT collides. We query FROM’s ‘last_value` / `is_called` directly (matching what pg_dump emits) rather than using MAX(pk), so a subsetted dump still preserves the source’s “next id”. Returns nil for non-auto-increment PKs (pg_get_serial_sequence -> NULL).

Scope: ONLY the sequence attached to the primary key is synced. If a table has additional auto-increment columns (e.g. a non-PK SERIAL), those sequences are NOT transcribed and a subsequent default-value INSERT on them can collide. Rails-managed schemas don’t hit this because only ‘id` is auto-increment, but bare PostgreSQL schemas may.

# File 'lib/exwiw/adapter/postgresql_adapter.rb', line 147

def post_insert_sql(table)
  pk = table.primary_key
  return nil if pk.nil? || pk.empty?

  seq_name = connection
    .exec_params("SELECT pg_get_serial_sequence($1, $2)", [table.name, pk])
    .values.dig(0, 0)
  return nil if seq_name.nil?

  last_value, is_called = connection
    .exec("SELECT last_value, is_called FROM #{seq_name}")
    .values.first
  is_called_sql = (is_called == 't' || is_called == true) ? 'true' : 'false'

  "SELECT pg_catalog.setval('#{escape_single_quote(seq_name)}', #{last_value}, #{is_called_sql});"
end

#to_bulk_delete(select_query_ast, table) ⇒ Object

Raises:

• (NotImplementedError)

# File 'lib/exwiw/adapter/postgresql_adapter.rb', line 164

def to_bulk_delete(select_query_ast, table)
  raise NotImplementedError unless select_query_ast.is_a?(Exwiw::QueryAst::Select)

  sql = "DELETE FROM #{select_query_ast.from_table_name}"

  if select_query_ast.join_clauses.empty?
    # Ignore filter option, because bulk delete is for cleaning before import,
    # so it should delete all records to avoid foreign key violation & data consistancy.
    compiled_where_conditions = select_query_ast.
      where_clauses.
      select { |where| where.is_a?(Exwiw::QueryAst::WhereClause) }.
      map do |where|
      compile_where_condition(where, select_query_ast.from_table_name)
    end

    if compiled_where_conditions.size > 0
      sql += "\nWHERE "
      sql += compiled_where_conditions.join(' AND ')
    end
    sql += ";"

    return sql
  end

  subquery_ast = Exwiw::QueryAst::Select.new
  first_join = select_query_ast.join_clauses.first.clone

  subquery_ast.from(first_join.join_table_name)
  primay_key_col = table.columns.find { |col| col.name == table.primary_key }
  subquery_ast.select([primay_key_col])
  select_query_ast.join_clauses[1..].each do |join|
    subquery_ast.join(join)
  end
  first_join.where_clauses.each do |where|
    # Ignore filter option, because bulk delete is for cleaning before import,
    # so it should delete all records to avoid foreign key violation & data consistancy.
    subquery_ast.where(where) if where.is_a?(Exwiw::QueryAst::WhereClause)
  end

  foreign_key = first_join.foreign_key
  outer_table = select_query_ast.from_table_name
  inner_table = first_join.join_table_name
  inner_column = first_join.primary_key
  cast_to = types_need_cast?(
    column_pg_type(outer_table, foreign_key),
    column_pg_type(inner_table, inner_column)
  ) ? 'text' : nil
  subquery_sql = compile_ast(subquery_ast, select_cast_to: cast_to)
  outer_expr = "#{outer_table}.#{foreign_key}"
  outer_expr = "#{outer_expr}::text" if cast_to
  sql += "\nWHERE #{outer_expr} IN (#{subquery_sql})"

  # first_join.base_where_clauses holds conditions on the outer
  # delete-target table (from_table_name), such as a polymorphic type
  # column. They are not part of the subquery, so add them to the outer
  # WHERE. This prevents deleting rows that belong to a different
  # polymorphic type.
  first_join.base_where_clauses.each do |where|
    next unless where.is_a?(Exwiw::QueryAst::WhereClause)

    sql += " AND #{compile_where_condition(where, select_query_ast.from_table_name)}"
  end
  sql += ";"

  sql
end

#to_bulk_insert(results, table) ⇒ Object

# File 'lib/exwiw/adapter/postgresql_adapter.rb', line 100

def to_bulk_insert(results, table)
  table_name = table.name

  value_list = results.map do |row|
    quoted_values = row.map do |value|
      escape_value(value)
    end
    "(" + quoted_values.join(', ') + ")"
  end
  values = value_list.join(",\n")

  if table.rails_managed?
    "INSERT INTO #{table_name} VALUES\n#{values};"
  else
    column_names = table.columns.map(&:name).join(', ')
    "INSERT INTO #{table_name} (#{column_names}) VALUES\n#{values};"
  end
end

#to_copy_from_stdin(results, table) ⇒ Object

# File 'lib/exwiw/adapter/postgresql_adapter.rb', line 119

def to_copy_from_stdin(results, table)
  header = if table.rails_managed?
             "COPY #{table.name} FROM stdin;"
           else
             column_names = table.columns.map(&:name).join(', ')
             "COPY #{table.name} (#{column_names}) FROM stdin;"
           end
  lines = [header]
  results.each do |row|
    lines << row.map { |v| escape_copy_value(v) }.join("\t")
  end
  lines << '\\.'
  lines.join("\n")
end