Class: Exwiw::Adapter::PostgresqlAdapter::StreamingResult

Inherits:
Object
  • Object
show all
Includes:
Enumerable
Defined in:
lib/exwiw/adapter/postgresql_adapter.rb

Overview

A lazy, streaming stand-in for the materialized rows #execute used to return (‘connection.exec(sql).values`). It pulls rows off the wire one at a time via libpq’s single-row mode instead of buffering the whole result set, so the dump’s dominant memory cost — a Ruby array as large as the table — never materializes. The Runner drives it exactly like the old Array: #size to skip empty tables and log the count, then a single streaming pass (SqlBulkInsert#write_inserts -> each_slice) to write the INSERT.

Mirrors MongodbAdapter::StreamingResult; two SQL-specific differences:

- #size cannot be answered cheaply from the cursor, so it runs a
  separate `SELECT COUNT(*)` of the same query. (MongoDB uses
  count_documents, an index-only walk; the SQL COUNT re-runs the query
  plan but transfers no row data — Postgres prunes the unused
  projection of the wrapped subquery.) This keeps the Runner contract
  unchanged, so MongoDB and the other SQL adapters are untouched.
- the streaming pass ties up the connection until fully drained. The
  Runner always drains it (write_inserts) before issuing any further
  query (post_insert_sql / DELETE) on the same connection, so the
  ordering invariant holds.

Instance Method Summary collapse

Constructor Details

#initialize(connection:, data_sql:, count_sql:) ⇒ StreamingResult

Returns a new instance of StreamingResult.



31
32
33
34
35
 
# File 'lib/exwiw/adapter/postgresql_adapter.rb', line 31

def initialize(connection:, data_sql:, count_sql:)
  @connection = connection
  @data_sql = data_sql
  @count_sql = count_sql
end

Instance Method Details

#each ⇒ Object

Stream the result set row by row. Each row is an Array of String|nil in libpq’s text format — byte-identical to what ‘#exec(sql).values` produced, so the generated INSERT is unchanged.



45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
 
# File 'lib/exwiw/adapter/postgresql_adapter.rb', line 45

def each
  return enum_for(:each) { size } unless block_given?

  @connection.send_query(@data_sql)
  @connection.set_single_row_mode
  begin
    while (result = @connection.get_result)
      begin
        result.check
        result.each_row { |row| yield row }
      ensure
        result.clear
      end
    end
  rescue StandardError
    # If iteration is abandoned mid-stream (a SQL error surfaced by
    # #check, or the consumer raised), drain any results still queued so
    # a later query on this same connection does not fail with "another
    # command is already in progress".
    drain
    raise
  end
  self
end

#size ⇒ Object Also known as: length 



37
38
39
 
# File 'lib/exwiw/adapter/postgresql_adapter.rb', line 37

def size
  @size ||= @connection.exec(@count_sql).getvalue(0, 0).to_i
end