Class: ActiveRecord::ConnectionAdapters::ClickHouse::HTTPConnection

Inherits:
Object
  • Object
show all
Defined in:
lib/active_record/connection_adapters/clickhouse/http_connection.rb

Overview

Raw connection to a ClickHouse server over its HTTP interface: one persistent keep-alive Net::HTTP socket per adapter instance (the adapter lock serializes use). Results arrive as RowBinaryWithNamesAndTypes by default — names, type strings, then packed binary rows — so every value comes back with its server type; queries whose types have no binary decoder retry transparently on the JSON wire (select_format: :json in the config forces JSON for everything).

Defined Under Namespace

Classes: ChunkedBody, ExecutionError, RawResult

Constant Summary collapse

BINARY_FORMAT = 
"RowBinaryWithNamesAndTypes"
JSON_FORMAT = 
"JSONCompactEachRowWithNamesAndTypes"
CONNECT_ERRORS =

Failures raised before the request reaches a server: retrying them on another replica can never double a write. Anything mid-flight (read timeout, reset) raises instead — the statement may have executed. ENETUNREACH added blind (approved 2026-07-22): connect-phase by nature, but not reproducible in the test container.

[
  Errno::ECONNREFUSED, Errno::EHOSTUNREACH, Errno::ENETUNREACH, SocketError, Net::OpenTimeout
].freeze

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(config) ⇒ HTTPConnection

Returns a new instance of HTTPConnection.



101
102
103
104
105
106
107
 
# File 'lib/active_record/connection_adapters/clickhouse/http_connection.rb', line 101

def initialize(config)
  @config = config
  @select_format = config[:select_format].to_s == "json" ? JSON_FORMAT : BINARY_FORMAT
  @endpoints = build_endpoints
  @endpoint_index = self.class.claim_start_index(@endpoints, failover_cooldown)
  @http = nil
end

Class Method Details

.claim_start_index(endpoints, cooldown) ⇒ Object 



40
41
42
43
44
45
46
47
48
49
 
# File 'lib/active_record/connection_adapters/clickhouse/http_connection.rb', line 40

def claim_start_index(endpoints, cooldown)
  @ledger_lock.synchronize do
    start = @start_counter % endpoints.size
    @start_counter += 1
    healthy_offset = endpoints.size.times.find do |offset|
      !recently_failed?(endpoints[(start + offset) % endpoints.size], cooldown)
    end
    (start + (healthy_offset || 0)) % endpoints.size
  end
end

.record_connect_failure(endpoint) ⇒ Object 



51
52
53
54
55
 
# File 'lib/active_record/connection_adapters/clickhouse/http_connection.rb', line 51

def record_connect_failure(endpoint)
  @ledger_lock.synchronize do
    @connect_failure_times[endpoint] = Process.clock_gettime(Process::CLOCK_MONOTONIC)
  end
end

Instance Method Details

#close ⇒ Object 



172
173
174
175
176
 
# File 'lib/active_record/connection_adapters/clickhouse/http_connection.rb', line 172

def close
  @http.finish if @http&.started?
rescue IOError
  nil
end

#current_endpoint ⇒ Object 



109
110
111
 
# File 'lib/active_record/connection_adapters/clickhouse/http_connection.rb', line 109

def current_endpoint
  @endpoints[@endpoint_index].join(":")
end

#execute(sql, params: {}) ⇒ Object 



140
141
142
143
144
 
# File 'lib/active_record/connection_adapters/clickhouse/http_connection.rb', line 140

def execute(sql, params: {})
  parse(perform(sql, params, @select_format), @select_format)
rescue RowBinary::Undecodable
  parse(perform(sql, params, JSON_FORMAT), JSON_FORMAT)
end

#execute_stream(sql, lines) ⇒ Object

Streams pre-encoded body lines as one chunked POST; the statement travels in the query string because the request body is the data.



148
149
150
151
152
153
 
# File 'lib/active_record/connection_adapters/clickhouse/http_connection.rb', line 148

def execute_stream(sql, lines)
  request = post_request(query_params({}, JSON_FORMAT).merge(query: sql))
  request["Transfer-Encoding"] = "chunked"
  request.body_stream = ChunkedBody.new(lines)
  parse(raise_unless_success(send_request(request)), JSON_FORMAT)
end

#ping ⇒ Object 



165
166
167
168
169
170
 
# File 'lib/active_record/connection_adapters/clickhouse/http_connection.rb', line 165

def ping
  execute("SELECT 1")
  true
rescue StandardError
  false
end

#with_request_settings(settings) ⇒ Object

Scopes extra server settings to the requests made inside the block — the write-side counterpart of the SETTINGS clause SELECTs carry in-SQL.



157
158
159
160
161
162
163
 
# File 'lib/active_record/connection_adapters/clickhouse/http_connection.rb', line 157

def with_request_settings(settings)
  previous = @request_settings
  @request_settings = (previous || {}).merge(settings)
  yield
ensure
  @request_settings = previous
end