Class: Presto::Client::StatementClient

Inherits:
Object
  • Object
show all
Defined in:
lib/presto/client/statement_client.rb

Constant Summary collapse

HEADERS = 
{
  "User-Agent" => "presto-ruby/#{VERSION}",
}

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(faraday, query, options) ⇒ StatementClient 



41
42
43
44
45
46
47
48
49
50
 
# File 'lib/presto/client/statement_client.rb', line 41

def initialize(faraday, query, options)
  @faraday = faraday
  @faraday.headers.merge!(HEADERS)

  @options = options
  @query = query
  @closed = false
  @exception = nil
  post_query_request!
end

Instance Attribute Details

#exception ⇒ Object (readonly)

Returns the value of attribute exception.



107
108
109
 
# File 'lib/presto/client/statement_client.rb', line 107

def exception
  @exception
end

#query ⇒ Object (readonly)

Returns the value of attribute query.



97
98
99
 
# File 'lib/presto/client/statement_client.rb', line 97

def query
  @query
end

Instance Method Details

#advance ⇒ Object 



129
130
131
132
133
134
135
136
137
138
139
 
# File 'lib/presto/client/statement_client.rb', line 129

def advance
  if closed? || !has_next?
    return false
  end
  uri = @results.next_uri

  body = faraday_get_with_retry(uri)
  @results = Models::QueryResults.decode(MultiJson.load(body))

  return true
end

#cancel_leaf_stage ⇒ Object 



176
177
178
179
180
181
182
183
184
 
# File 'lib/presto/client/statement_client.rb', line 176

def cancel_leaf_stage
  if uri = @results.next_uri
    response = @faraday.delete do |req|
      req.url uri
    end
    return response.status / 100 == 2
  end
  return false
end

#close ⇒ Object 



186
187
188
189
190
191
192
193
194
195
 
# File 'lib/presto/client/statement_client.rb', line 186

def close
  return if @closed

  # cancel running statement
  # TODO make async reqeust and ignore response?
  cancel_leaf_stage

  @closed = true
  nil
end

#closed? ⇒ Boolean 



103
104
105
 
# File 'lib/presto/client/statement_client.rb', line 103

def closed?
  @closed
end

#current_results ⇒ Object 



121
122
123
 
# File 'lib/presto/client/statement_client.rb', line 121

def current_results
  @results
end

#debug? ⇒ Boolean 



99
100
101
 
# File 'lib/presto/client/statement_client.rb', line 99

def debug?
  !!@options[:debug]
end

#exception? ⇒ Boolean 



109
110
111
 
# File 'lib/presto/client/statement_client.rb', line 109

def exception?
  @exception
end

#faraday_get_with_retry(uri, &block) ⇒ Object

Raises:



146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
 
# File 'lib/presto/client/statement_client.rb', line 146

def faraday_get_with_retry(uri, &block)
  start = Time.now
  attempts = 0

  begin
    begin
      response = @faraday.get(uri)
    rescue => e
      @exception = e
      raise @exception
    end

    if response.status == 200 && !response.body.to_s.empty?
      return response.body
    end

    if response.status != 503  # retry only if 503 Service Unavailable
      # deterministic error
      @exception = PrestoHttpError.new(response.status, "Presto API error at #{uri} returned #{response.status}: #{response.body}")
      raise @exception
    end

    attempts += 1
    sleep attempts * 0.1
  end while (Time.now - start) < 2*60*60 && !@closed

  @exception = PrestoHttpError.new(408, "Presto API error due to timeout")
  raise @exception
end

#has_next? ⇒ Boolean 



125
126
127
 
# File 'lib/presto/client/statement_client.rb', line 125

def has_next?
  !!@results.next_uri
end

#query_failed? ⇒ Boolean 



113
114
115
 
# File 'lib/presto/client/statement_client.rb', line 113

def query_failed?
  @results.error != nil
end

#query_info ⇒ Object 



141
142
143
144
 
# File 'lib/presto/client/statement_client.rb', line 141

def query_info
  body = faraday_get_with_retry("/v1/query/#{@results.id}")
  Models::QueryInfo.decode(MultiJson.load(body))
end

#query_succeeded? ⇒ Boolean 



117
118
119
 
# File 'lib/presto/client/statement_client.rb', line 117

def query_succeeded?
  @results.error == nil && !@exception && !@closed
end