Class: ClickhouseNative::Pool
- Inherits:
-
Object
- Object
- ClickhouseNative::Pool
- Defined in:
- lib/clickhouse_native/pool.rb
Instance Attribute Summary collapse
-
#database ⇒ Object
readonly
Returns the value of attribute database.
-
#host ⇒ Object
readonly
Returns the value of attribute host.
-
#port ⇒ Object
readonly
Returns the value of attribute port.
Instance Method Summary collapse
- #describe_table(table, db_name: nil) ⇒ Object
- #execute(sql, **opts) ⇒ Object
-
#initialize(host:, port:, database: "default", user: "default", password: "", compression: :none, logger: nil, settings: {}, pool_size: 5, pool_timeout: 5, ping_before_query: true, tcp_keepalive: true, retry_timeout: 1) ⇒ Pool
constructor
A new instance of Pool.
- #insert(table, rows, **opts) ⇒ Object
- #ping ⇒ Object
- #query(sql, **opts) ⇒ Object
- #query_each(sql, **opts, &block) ⇒ Object
- #query_value(sql, **opts) ⇒ Object
- #server_version ⇒ Object
-
#with ⇒ Object
On an aborted operation, discard the client rather than reuse it: the socket is left in an unknown state.
Constructor Details
#initialize(host:, port:, database: "default", user: "default", password: "", compression: :none, logger: nil, settings: {}, pool_size: 5, pool_timeout: 5, ping_before_query: true, tcp_keepalive: true, retry_timeout: 1) ⇒ Pool
Returns a new instance of Pool.
9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 |
# File 'lib/clickhouse_native/pool.rb', line 9 def initialize(host:, port:, database: "default", user: "default", password: "", compression: :none, logger: nil, settings: {}, pool_size: 5, pool_timeout: 5, ping_before_query: true, tcp_keepalive: true, retry_timeout: 1) @host = host @port = port @database = database client_kwargs = { host:, port:, database:, user:, password:, compression:, logger:, settings:, ping_before_query:, tcp_keepalive:, retry_timeout: } @pool = ConnectionPool.new(size: pool_size, timeout: pool_timeout) do Client.new(**client_kwargs) end end |
Instance Attribute Details
#database ⇒ Object (readonly)
Returns the value of attribute database.
7 8 9 |
# File 'lib/clickhouse_native/pool.rb', line 7 def database @database end |
#host ⇒ Object (readonly)
Returns the value of attribute host.
7 8 9 |
# File 'lib/clickhouse_native/pool.rb', line 7 def host @host end |
#port ⇒ Object (readonly)
Returns the value of attribute port.
7 8 9 |
# File 'lib/clickhouse_native/pool.rb', line 7 def port @port end |
Instance Method Details
#describe_table(table, db_name: nil) ⇒ Object
94 95 96 |
# File 'lib/clickhouse_native/pool.rb', line 94 def describe_table(table, db_name: nil) with { |c| c.describe_table(table, db_name:) } end |
#execute(sql, **opts) ⇒ Object
66 67 68 |
# File 'lib/clickhouse_native/pool.rb', line 66 def execute(sql, **opts) with { |c| c.execute(sql, **opts) } end |
#insert(table, rows, **opts) ⇒ Object
82 83 84 |
# File 'lib/clickhouse_native/pool.rb', line 82 def insert(table, rows, **opts) with { |c| c.insert(table, rows, **opts) } end |
#ping ⇒ Object
86 87 88 |
# File 'lib/clickhouse_native/pool.rb', line 86 def ping with(&:ping) end |
#query(sql, **opts) ⇒ Object
70 71 72 |
# File 'lib/clickhouse_native/pool.rb', line 70 def query(sql, **opts) with { |c| c.query(sql, **opts) } end |
#query_each(sql, **opts, &block) ⇒ Object
74 75 76 |
# File 'lib/clickhouse_native/pool.rb', line 74 def query_each(sql, **opts, &block) with { |c| c.query_each(sql, **opts, &block) } end |
#query_value(sql, **opts) ⇒ Object
78 79 80 |
# File 'lib/clickhouse_native/pool.rb', line 78 def query_value(sql, **opts) with { |c| c.query_value(sql, **opts) } end |
#server_version ⇒ Object
90 91 92 |
# File 'lib/clickhouse_native/pool.rb', line 90 def server_version with(&:server_version) end |
#with ⇒ Object
On an aborted operation, discard the client rather than reuse it: the socket is left in an unknown state. The C++ binding issues ResetConnection, but a subsequent send can still surface buffered protocol errors from the prior aborted operation — those get attributed to whatever SQL we tried next, producing misleading log lines and re-raises in unrelated code. A fresh socket + handshake is cheap relative to debugging that.
ConnectionError gets one automatic retry: pooled connections that have been idle long enough for the server / an LB to FIN them surface as "closed" on the very next recv (errno is whatever stale value was left in the thread — "closed: Success", "closed: Operation now in progress", etc. all mean the same recv()==0). Discarding and re-checking out lands a fresh socket and the operation succeeds. The retry only triggers when the dead-connection error fired before any data was sent, so write operations don't risk double-execution.
This is the backstop: with ping_before_query on (the default) the driver already pings and transparently reconnects a dead socket before running the query, so most stale connections never surface as a ConnectionError here at all. This retry still covers the residual race (socket dies between the ping and the query).
A bare rescue is not enough to spot an abandoned query. Timeout and
Sidekiq shutdown raise off Exception rather than StandardError, and
Thread#kill raises nothing at all — Parallel.in_threads kills every
sibling worker the moment one of them fails. Miss those and the
connection goes back in with the server still streaming a response at
it; the next checkout reads that leftover as its own, which surfaces
as a bogus packet type far from here.
55 56 57 58 59 60 61 62 63 64 |
# File 'lib/clickhouse_native/pool.rb', line 55 def with attempts = 0 begin @pool.with { |client| discard_unless_clean { yield client } } rescue ConnectionError attempts += 1 retry if attempts == 1 raise end end |