Class: UmbrellioUtils::ClickHouse::Backends::Base

Inherits:
Object
  • Object
show all
Includes:
Singleton
Defined in:
lib/umbrellio_utils/click_house/backends/base.rb

Overview

Abstract backend. Each concrete backend (Legacy for the click_house gem, Native for the clickhouse-native gem) implements the low-level ops (execute / query / insert / describe_table / server_version / tables / create_database / drop_database / config / logger) and a SERVER_ERROR constant used by log_errors.

Direct Known Subclasses

Legacy, Native

Defined Under Namespace

Modules: ClickHouseDatasetMethods

Instance Method Summary collapse

Instance Method Details

#count(dataset) ⇒ Object 



132
133
134
 
# File 'lib/umbrellio_utils/click_house/backends/base.rb', line 132

def count(dataset, **)
  query_value(dataset.select(SQL.ch_count), **)
end

#create_database(name, if_not_exists: false, cluster: nil, engine: nil) ⇒ Object 



152
153
154
155
156
157
158
159
160
161
162
 
# File 'lib/umbrellio_utils/click_house/backends/base.rb', line 152

def create_database(name, if_not_exists: false, cluster: nil, engine: nil)
  admin_execute(
    format(
      "CREATE DATABASE %<exists>s %<name>s %<cluster>s %<engine>s",
      exists: if_not_exists ? "IF NOT EXISTS" : "",
      name:,
      cluster: cluster ? "ON CLUSTER #{cluster}" : "",
      engine: engine ? "ENGINE = #{engine}" : "",
    ),
  )
end

#db_name ⇒ Object 



148
149
150
 
# File 'lib/umbrellio_utils/click_house/backends/base.rb', line 148

def db_name
  config.database.to_sym
end

#drop_database(name, if_exists: false, cluster: nil) ⇒ Object 



164
165
166
167
168
169
170
171
172
173
 
# File 'lib/umbrellio_utils/click_house/backends/base.rb', line 164

def drop_database(name, if_exists: false, cluster: nil)
  admin_execute(
    format(
      "DROP DATABASE %<exists>s %<name>s %<cluster>s",
      exists: if_exists ? "IF EXISTS" : "",
      name:,
      cluster: cluster ? "ON CLUSTER #{cluster}" : "",
    ),
  )
end

#drop_table!(table_name, db_name: self.db_name) ⇒ Object 



193
194
195
 
# File 'lib/umbrellio_utils/click_house/backends/base.rb', line 193

def drop_table!(table_name, db_name: self.db_name)
  execute("DROP TABLE #{db_name}.#{table_name} #{on_cluster(sync: true)}")
end

#from(source, db_name: self.db_name) ⇒ Object

Concrete backends implement the low-level ops (execute / query / insert / describe_table / server_version / tables / admin_execute / config / logger) and define SERVER_ERROR.



119
120
121
122
123
124
125
126
127
128
129
130
 
# File 'lib/umbrellio_utils/click_house/backends/base.rb', line 119

def from(source, db_name: self.db_name)
  ds =
    case source
    when Symbol
      DB.from(db_name == self.db_name ? SQL[source] : SQL[db_name][source])
    when nil
      DB.dataset
    else
      DB.from(source)
    end
  ds.clone(ch: true).with_extend(ClickHouseDatasetMethods)
end

#on_cluster(sync: false) ⇒ Object

Returns the ON CLUSTER <name> [SYNC] clause for DDL, or "" if UmbrellioUtils.config.clickhouse_cluster is blank or we're in a Rails test env. Test-env suppression saves hundreds of ms per DDL on a single-node CH (each ON CLUSTER op blocks waiting for replicas that don't exist). The cluster name is still used by callers like Distributed engine declarations, regardless of this clause.



182
183
184
185
186
187
 
# File 'lib/umbrellio_utils/click_house/backends/base.rb', line 182

def on_cluster(sync: false)
  name = UmbrellioUtils.config.clickhouse_cluster
  return "" if name.blank?
  return "" if defined?(Rails) && Rails.env.test?
  sync ? "ON CLUSTER #{name} SYNC" : "ON CLUSTER #{name}"
end

#optimize_table!(table_name, db_name: self.db_name) ⇒ Object 



197
198
199
200
201
 
# File 'lib/umbrellio_utils/click_house/backends/base.rb', line 197

def optimize_table!(table_name, db_name: self.db_name)
  Timeout.timeout(UmbrellioUtils.config.ch_optimize_timeout) do
    execute("OPTIMIZE TABLE #{db_name}.#{table_name} #{on_cluster} FINAL")
  end
end

#parse_value(value, type:) ⇒ Object 



203
204
205
206
207
208
209
210
211
212
213
214
 
# File 'lib/umbrellio_utils/click_house/backends/base.rb', line 203

def parse_value(value, type:)
  case type
  when /Array/ then Array.wrap(value)
  when /DateTime/
    case value
    when String then value.present? ? Time.zone.parse(value) : nil
    else value
    end
  when /String/ then value&.to_s
  else value
  end
end

#pg_table_connection(table, schema: "public") ⇒ Object 



216
217
218
219
220
221
222
223
224
225
226
 
# File 'lib/umbrellio_utils/click_house/backends/base.rb', line 216

def pg_table_connection(table, schema: "public")
  host = ENV["PGHOST"] || DB.opts[:host].presence || "localhost"
  port = DB.opts[:port] || 5432
  # Etc.getlogin returns "root" under non-TTY shells (e.g. rake from
  # a CI runner), which is almost never a real PG role. Prefer $USER.
   = ENV["USER"].presence || Etc.getlogin
  database = DB.opts[:database].presence || 
  username = DB.opts[:user].presence || 
  password = DB.opts[:password]
  SQL.func(:postgresql, "#{host}:#{port}", database, table, username, password, schema)
end

#populate_temp_table!(temp_table_name, dataset, schema: "public") ⇒ Object 



228
229
230
231
232
233
 
# File 'lib/umbrellio_utils/click_house/backends/base.rb', line 228

def populate_temp_table!(temp_table_name, dataset, schema: "public")
  execute(<<~SQL.squish)
    INSERT INTO TABLE FUNCTION #{DB.literal(pg_table_connection(temp_table_name, schema:))}
    #{dataset.sql}
  SQL
end

#table_metadata(table_name, db_name: self.db_name) ⇒ Object

Sorting key / version / is_deleted of a ReplacingMergeTree table. Distributed tables carry none of these, so they are resolved through to the local table they wrap. Memoized per process, like the layout it describes: a table's engine does not change under a running app.



140
141
142
143
144
145
146
 
# File 'lib/umbrellio_utils/click_house/backends/base.rb', line 140

def table_metadata(table_name, db_name: self.db_name)
  key = [db_name.to_s, table_name.to_s]
  @table_metadata_cache ||= {}
  return @table_metadata_cache[key] if @table_metadata_cache.key?(key)

  @table_metadata_cache[key] = build_table_metadata(*key)
end

#truncate_table!(table_name, db_name: self.db_name) ⇒ Object 



189
190
191
 
# File 'lib/umbrellio_utils/click_house/backends/base.rb', line 189

def truncate_table!(table_name, db_name: self.db_name)
  execute("TRUNCATE TABLE #{db_name}.#{table_name} #{on_cluster(sync: true)}")
end

#with_temp_table(dataset, temp_table_name:, primary_key: [:id], primary_key_types: [:integer]) ⇒ Object 



235
236
237
238
239
240
241
242
243
244
245
 
# File 'lib/umbrellio_utils/click_house/backends/base.rb', line 235

def with_temp_table(
  dataset, temp_table_name:, primary_key: [:id], primary_key_types: [:integer], **, &
)
  unless DB.table_exists?(temp_table_name)
    UmbrellioUtils::Database.create_temp_table(
      nil, primary_key:, primary_key_types:, temp_table_name:, &
    )
    populate_temp_table!(temp_table_name, dataset)
  end
  UmbrellioUtils::Database.with_temp_table(nil, primary_key:, temp_table_name:, **, &)
end