Class: QueryExecution

Inherits:
Object
  • Object
show all
Defined in:
app/models/query_execution.rb

Constant Summary collapse

NUM_SAMPLE_ROWS = 
100

Class Method Summary collapse

Class Method Details

.perform(result_id, role) ⇒ Object 



5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
 
# File 'app/models/query_execution.rb', line 5

def self.perform(result_id, role)
  result = Result.find(result_id)

  unless Role.configured_connections.include?(role)
    raise "Role '#{role}' does not have connection credentials configured."
  end

  body = result.compiled_body
  result.mark_running!
  sample_callback = ->(sample) { result.mark_processing_from_sample(sample) }

  connection = RedshiftConnectionPool.instance.get(role)

  connection.reconnect_on_failure do
    query_stream = PgStream::Stream.new(connection.pg_connection, body)
    result.headers = query_stream.headers
    result.save!

    rrrc = result.redis_result_row_count

    stream_processor = PgStream::Processor.new(query_stream)
    stream_processor.register(ResultCsvGenerator.new(result_id, result.headers).callbacks)
    stream_processor.register(SampleSkimmer.new(NUM_SAMPLE_ROWS, &sample_callback).callbacks)
    stream_processor.register(CountPublisher.new(rrrc).callbacks)

    row_count = stream_processor.execute
    result.mark_complete_with_count(row_count)
  end
rescue *RedshiftPG::USER_ERROR_CLASSES => e
  result.mark_failed!(e.message)
rescue => e
  result.mark_failed!(e.message) if result
  raise
end