diff --git a/Gemfile.lock b/Gemfile.lock index a88f88c..a2c44fe 100644 --- a/Gemfile.lock +++ b/Gemfile.lock @@ -11,7 +11,7 @@ GIT PATH remote: . specs: - umbrellio-utils (1.15.0) + umbrellio-utils (1.16.0) memery (~> 1) GEM diff --git a/README.md b/README.md index 84841cd..1bbe7fe 100644 --- a/README.md +++ b/README.md @@ -115,6 +115,37 @@ end Utils::Constants.useful_method #=> "Just string" ``` +### ClickHouse deduplication + +Datasets built with `CH.from` understand ClickHouse's `LIMIT n BY`: + +```ruby +CH.from(:events).order(Sequel.desc(:version)).limit_by(:user_id, rows: 3) +#=> SELECT * FROM "events" ORDER BY "version" DESC LIMIT 3 BY "user_id" +``` + +On a `ReplacingMergeTree` table, `#deduplicate` uses that to collapse row versions by hand rather than relying on the `final` setting, which merges the entire table even for a point lookup: + +```ruby +CH.from(:external_operations_distributed) + .where(order_id: 42) # inside the dedup subquery + .deduplicate # boundary + .order(:created_at) # outside +``` + +The sorting key, version column and `is_deleted` column are read from `system.tables` and cached per process; `Distributed` tables are resolved through to the local table they wrap. Deduplicated datasets are sent with `final: 0` automatically, since running FINAL inside the subquery would be both slow and redundant. An explicit `final:` passed to `query` / `count` always wins. + +`#deduplicate` is a boundary, and which side a filter lands on matters: + +- **Before it — immutable selectors only** (primary keys, foreign keys). Filtering a *mutable* column first can match a superseded row version and resurrect a row that FINAL would have dropped. +- **After it — everything else**, including any filter on a column that changes over a row's lifetime. + +`is_deleted` is applied after the boundary for exactly that reason, and is omitted when the engine declares no such column. + +Two chain methods are handled rather than passed through, because the dedup subquery has to control them: the subquery always projects `SELECT *` (so the outer query can still filter on `is_deleted` and on columns you did not select) and any projection you set is re-applied outside it; and the version ordering leads the subquery's `ORDER BY`, since it decides which row survives, with any ordering you set kept after it as a tiebreaker. + +`#deduplicate` raises on a table it cannot collapse — a non-Replacing engine, a Replacing engine declared without a version argument, or a dataset that is not a single table source (joined, multi-source, or a subquery). + ### Instrumentation The gem ships a set of opt-in files for collecting GVL and allocation stats diff --git a/lib/umbrellio_utils/click_house.rb b/lib/umbrellio_utils/click_house.rb index 133722c..49e995c 100644 --- a/lib/umbrellio_utils/click_house.rb +++ b/lib/umbrellio_utils/click_house.rb @@ -10,12 +10,13 @@ module ClickHouse extend self autoload :Backends, "umbrellio_utils/click_house/backends" + autoload :TableMetadata, "umbrellio_utils/click_house/table_metadata" VALID_BACKENDS = %i[legacy native].freeze DELEGATED = %i[ execute query query_value query_each count insert - from describe_table server_version tables + from describe_table server_version tables table_metadata create_database drop_database db_name config truncate_table! drop_table! optimize_table! on_cluster parse_value pg_table_connection populate_temp_table! with_temp_table diff --git a/lib/umbrellio_utils/click_house/backends/base.rb b/lib/umbrellio_utils/click_house/backends/base.rb index cedbbee..d15bdca 100644 --- a/lib/umbrellio_utils/click_house/backends/base.rb +++ b/lib/umbrellio_utils/click_house/backends/base.rb @@ -13,13 +13,103 @@ module Backends class Base include Singleton - # ClickHouse uses C-style escape sequences in string literals, so - # backslashes must be doubled. Sequel's default (Postgres) escaping - # only escapes single-quotes. - module ClickHouseStringEscaping + # ClickHouse-specific dataset behaviour: string escaping plus grammar + # Sequel's Postgres dataset doesn't know about. + module ClickHouseDatasetMethods + # ClickHouse uses C-style escape sequences in string literals, so + # backslashes must be doubled. Sequel's default (Postgres) escaping + # only escapes single-quotes. def literal_string_append(sql, str) sql << "'" << str.gsub("\\") { "\\\\" }.gsub("'", "''") << "'" end + + # ClickHouse `LIMIT n BY expr, ...` — keeps the first n rows per + # distinct combination of the expressions, applied after ORDER BY. + def limit_by(*exprs, rows: 1) + raise Sequel::Error, "limit_by requires at least one expression" if exprs.empty? + clone(limit_by: { exprs:, rows: }) + end + + # `LIMIT n BY` precedes the regular LIMIT/OFFSET in ClickHouse. + def select_limit_sql(sql) + if (limit_by = @opts[:limit_by]) + sql << " LIMIT " + literal_append(sql, limit_by[:rows]) + sql << " BY " + expression_list_append(sql, limit_by[:exprs]) + end + + super + end + + # Collapse a ReplacingMergeTree's row versions by hand instead of + # relying on the `final` setting, which merges the whole table. + # + # This is a boundary: everything chained BEFORE it goes inside the + # dedup subquery, everything chained AFTER applies to the result. + # Only immutable selectors belong before it — filtering a mutable + # column first can match a superseded version and resurrect a row + # that FINAL would have dropped. `is_deleted` is applied after the + # boundary for the same reason. + def deduplicate + table_name, db_name = deduplication_source + meta = ClickHouse.table_metadata(table_name, **db_name) + + unless meta.replacing? + raise Sequel::Error, + "#{table_name} is a #{meta.engine}; deduplicate needs a ReplacingMergeTree" + end + + unless meta.version + raise Sequel::Error, + "#{table_name} declares no version column; deduplicate needs one" + end + + wrapped = ClickHouse.from(wrap_source(deduplicated_source(meta))) + .clone(ch_dedup: true) + .then { |ds| @opts[:select] ? ds.select(*@opts[:select]) : ds } + + meta.is_deleted ? wrapped.where(meta.is_deleted => 0) : wrapped + end + + private + + # The subquery must project every column, both so the outer query can + # filter on `is_deleted` and so a caller's projection still resolves; + # that projection is re-applied outside instead. + # + # The version ordering leads, since it decides which row survives — + # any ordering the caller set is kept after it as a tiebreaker. + def deduplicated_source(meta) + clone(select: nil) + .order(Sequel.desc(meta.version), *@opts[:order]) + .limit_by(*meta.sorting_key.map { |expr| Sequel.lit(expr) }) + end + + def wrap_source(inner) + source = Array(@opts[:from]).first + source.is_a?(Sequel::SQL::AliasedExpression) ? inner.as(source.alias) : inner + end + + # => [table_name, {} | { db_name: ... }] + def deduplication_source + sources = Array(@opts[:from]) + source = sources.first + source = source.expression if source.is_a?(Sequel::SQL::AliasedExpression) + + if sources.size != 1 || @opts[:join] + raise Sequel::Error, "deduplicate requires a single table source" + end + + case source + when Sequel::SQL::QualifiedIdentifier + [source.column, { db_name: source.table }] + when Sequel::SQL::Identifier + [source.value, {}] + else + raise Sequel::Error, "deduplicate requires a single table source" + end + end end # Concrete backends implement the low-level ops (execute / query / @@ -36,11 +126,23 @@ def from(source, db_name: self.db_name) else DB.from(source) end - ds.clone(ch: true).with_extend(ClickHouseStringEscaping) + ds.clone(ch: true).with_extend(ClickHouseDatasetMethods) end - def count(dataset) - query_value(dataset.select(SQL.ch_count)) + def count(dataset, **) + query_value(dataset.select(SQL.ch_count), **) + end + + # 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. + 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 def db_name @@ -144,6 +246,43 @@ def with_temp_table( protected + # Every read path needs both halves, and pairing them here keeps a new + # one from silently reinstating session-wide FINAL inside a dedup + # subquery — the exact cost `deduplicate` exists to remove. + def prepare_query(dataset, opts) + [sql_for(dataset), settings_for(dataset, opts)] + end + + # `final` is usually a session-wide default, which would make a + # deduplicated query merge the whole table inside its own subquery — + # slow and redundant, since the subquery already collapses versions. + # An explicit `final:` from the caller always wins. + def settings_for(dataset, opts) + return opts if opts.key?(:final) + return opts unless dataset.is_a?(Sequel::Dataset) && dataset.opts[:ch_dedup] + + opts.merge(final: 0) + end + + def build_table_metadata(db_name, table_name) + row = query( + from(:tables, db_name: :system) + .where(database: db_name, name: table_name) + .select(:engine, :engine_full, :sorting_key), + ).first + + unless row + raise ClickHouse::TableMetadata::UnknownTable, "#{db_name}.#{table_name} not found" + end + + if row[:engine] == "Distributed" + database, table = ClickHouse::TableMetadata.distributed_target(row[:engine_full]) + return table_metadata(table, db_name: database) + end + + ClickHouse::TableMetadata.parse(**row) + end + def log_errors(sql) yield rescue self.class::SERVER_ERROR => e diff --git a/lib/umbrellio_utils/click_house/backends/legacy.rb b/lib/umbrellio_utils/click_house/backends/legacy.rb index 68e825f..0012ddb 100644 --- a/lib/umbrellio_utils/click_house/backends/legacy.rb +++ b/lib/umbrellio_utils/click_house/backends/legacy.rb @@ -16,15 +16,15 @@ def execute(sql, host: nil, **opts) end def query(dataset, host: nil, **opts) - sql = sql_for(dataset) + sql, settings = prepare_query(dataset, opts) log_errors(sql) do - select_all(sql, host:, **opts).map { |x| Misc::StrictHash[x.symbolize_keys] } + select_all(sql, host:, **settings).map { |x| Misc::StrictHash[x.symbolize_keys] } end end def query_value(dataset, host: nil, **opts) - sql = sql_for(dataset) - log_errors(sql) { select_value(sql, host:, **opts) } + sql, settings = prepare_query(dataset, opts) + log_errors(sql) { select_value(sql, host:, **settings) } end def query_each(dataset, host: nil, **, &) diff --git a/lib/umbrellio_utils/click_house/backends/native.rb b/lib/umbrellio_utils/click_house/backends/native.rb index 347b36f..051a824 100644 --- a/lib/umbrellio_utils/click_house/backends/native.rb +++ b/lib/umbrellio_utils/click_house/backends/native.rb @@ -29,18 +29,18 @@ def execute(sql, host: nil, **opts) # rubocop:disable Lint/UnusedMethodArgument end def query(dataset, host: nil, **opts) # rubocop:disable Lint/UnusedMethodArgument - sql = sql_for(dataset) - log_errors(sql) { pool.query(sql, settings: opts) } + sql, settings = prepare_query(dataset, opts) + log_errors(sql) { pool.query(sql, settings:) } end def query_value(dataset, host: nil, **opts) # rubocop:disable Lint/UnusedMethodArgument - sql = sql_for(dataset) - log_errors(sql) { pool.query_value(sql, settings: opts) } + sql, settings = prepare_query(dataset, opts) + log_errors(sql) { pool.query_value(sql, settings:) } end def query_each(dataset, host: nil, **opts, &) # rubocop:disable Lint/UnusedMethodArgument - sql = sql_for(dataset) - log_errors(sql) { pool.query_each(sql, settings: opts, &) } + sql, settings = prepare_query(dataset, opts) + log_errors(sql) { pool.query_each(sql, settings:, &) } end def insert(table_name, db_name: self.db_name, rows: []) diff --git a/lib/umbrellio_utils/click_house/table_metadata.rb b/lib/umbrellio_utils/click_house/table_metadata.rb new file mode 100644 index 0000000..516d23a --- /dev/null +++ b/lib/umbrellio_utils/click_house/table_metadata.rb @@ -0,0 +1,110 @@ +# frozen_string_literal: true + +module UmbrellioUtils + module ClickHouse + # Layout of a MergeTree-family table, read out of `system.tables`. + # + # ReplacingMergeTree collapses rows sharing the full sorting key, keeping + # the one with the highest version and dropping it entirely when the + # is_deleted column is set. `Dataset#deduplicate` reproduces that by hand, + # so it needs all three parts; `version` and `is_deleted` are only + # meaningful on a Replacing engine, and nil elsewhere. + TableMetadata = Struct.new(:engine, :sorting_key, :version, :is_deleted) + + # Reopened rather than declared with a block: constants defined inside a + # `Struct.new do ... end` block leak to the enclosing lexical scope. + class TableMetadata + class UnknownTable < StandardError + end + + REPLICATED_ARGS_COUNT = 2 # zookeeper path + replica name + + def replacing? + engine.include?("Replacing") + end + + class << self + # Built first so it can answer #replacing? for itself — the engine + # test lives in one place only. + def parse(engine:, engine_full:, sorting_key:) + meta = new(engine, split_args(sorting_key), nil, nil) + return meta unless meta.replacing? + + args = engine_args(engine_full) + args = args.drop(REPLICATED_ARGS_COUNT) if engine.start_with?("Replicated") + meta.version, meta.is_deleted = args[0]&.to_sym, args[1]&.to_sym + + meta + end + + # Distributed('cluster', 'database', 'table'[, sharding_key]) + def distributed_target(engine_full) + _cluster, database, table = engine_args(engine_full) + [unquote(database), unquote(table)] + end + + # Arguments of the leading engine call, or [] when the engine takes none. + # Only a parenthesis directly after the engine name counts — later + # clauses such as `PARTITION BY toYYYYMM(created_at)` must not be read. + def engine_args(engine_full) + open_index = engine_full.index("(") + return [] unless open_index + return [] unless engine_full[0...open_index].match?(/\A\w+\z/) + + split_args(engine_full[(open_index + 1)...close_index(engine_full, open_index)]) + end + + # Split on top-level commas only: sorting keys hold function calls and + # engine arguments hold quoted paths, both of which may contain commas. + def split_args(source) + args = [] + current = +"" + + scan(source) do |char, depth, in_string| + if char == "," && depth.zero? && !in_string + args << current.strip + current = +"" + else + current << char + end + end + + args << current.strip + args.reject(&:empty?) + end + + private + + # Single lexer for both scans, so they can't disagree about what counts + # as a parenthesis: a ZooKeeper path may legally contain one inside + # quotes, and treating it as structure truncates the argument list. + def scan(source) + depth = 0 + in_string = false + + source.to_s.each_char.with_index do |char, index| + case char + when "'" then in_string = !in_string + when "(" then depth += 1 unless in_string + when ")" then depth -= 1 unless in_string + end + + yield(char, depth, in_string, index) + end + end + + def close_index(source, open_index) + scan(source[open_index..]) do |_char, depth, in_string, index| + return open_index + index if depth.zero? && !in_string + end + + raise ArgumentError, "unbalanced parentheses in engine: #{source}" + end + + def unquote(value) + value.to_s.delete_prefix("'").delete_suffix("'") + end + end + end + end +end diff --git a/lib/umbrellio_utils/version.rb b/lib/umbrellio_utils/version.rb index 8aa0f2f..bf0eec0 100644 --- a/lib/umbrellio_utils/version.rb +++ b/lib/umbrellio_utils/version.rb @@ -1,5 +1,5 @@ # frozen_string_literal: true module UmbrellioUtils - VERSION = "1.15.0" + VERSION = "1.16.0" end diff --git a/spec/support/clickhouse.rb b/spec/support/clickhouse.rb index 404dfb9..bfc5821 100644 --- a/spec/support/clickhouse.rb +++ b/spec/support/clickhouse.rb @@ -16,3 +16,37 @@ ENGINE = MergeTree() ORDER BY id; SQL + +# ReplacingMergeTree with a multi-column sorting key, a version column and an +# is_deleted column — the shape `#deduplicate` has to introspect. +client.execute(<<~SQL) + CREATE TABLE IF NOT EXISTS test_replacing + (group_id Int32, id Int32, payload String, version Int32, is_deleted UInt8) + ENGINE = ReplacingMergeTree(version, is_deleted) + ORDER BY (group_id, id); +SQL + +# Version but no is_deleted — the outer delete filter must then be omitted. +client.execute(<<~SQL) + CREATE TABLE IF NOT EXISTS test_replacing_no_delete + (id Int32, payload String, version Int32) + ENGINE = ReplacingMergeTree(version) + ORDER BY id; +SQL + +# A Replacing engine declared without a version argument is legal, and +# deduplicate cannot pick a winner on one. +client.execute(<<~SQL) + CREATE TABLE IF NOT EXISTS test_replacing_no_version + (id Int32, payload String) + ENGINE = ReplacingMergeTree + ORDER BY id; +SQL + +# Distributed tables carry no sorting key of their own, so metadata lookup has +# to resolve through to the local table. +client.execute(<<~SQL) + CREATE TABLE IF NOT EXISTS test_replacing_distributed + (group_id Int32, id Int32, payload String, version Int32, is_deleted UInt8) + ENGINE = Distributed('click_cluster', 'umbrellio_utils_test', 'test_replacing', id); +SQL diff --git a/spec/umbrellio_utils/click_house/deduplicate_spec.rb b/spec/umbrellio_utils/click_house/deduplicate_spec.rb new file mode 100644 index 0000000..1de718a --- /dev/null +++ b/spec/umbrellio_utils/click_house/deduplicate_spec.rb @@ -0,0 +1,118 @@ +# frozen_string_literal: true + +describe UmbrellioUtils::ClickHouse do + let(:ch) { described_class } + + before do + ch.truncate_table!("test_replacing") + ch.truncate_table!("test_replacing_no_delete") + end + + describe "#deduplicate" do + describe "generated SQL" do + it "wraps the source in a dedup subquery keyed on the full sorting key" do + expect(ch.from(:test_replacing).where(group_id: 1).deduplicate.sql).to eq( + 'SELECT * FROM (SELECT * FROM "test_replacing" WHERE ("group_id" = 1) ' \ + 'ORDER BY "version" DESC LIMIT 1 BY group_id, id) AS "t1" ' \ + 'WHERE ("is_deleted" = 0) ORDER BY rand()', + ) + end + + it "keeps filters chained after the boundary outside the subquery" do + sql = ch.from(:test_replacing).where(group_id: 1).deduplicate.where(payload: "x").sql + expect(sql).to include('WHERE ("group_id" = 1) ORDER BY "version" DESC') + expect(sql).to end_with( + %q{WHERE (("is_deleted" = 0) AND ("payload" = 'x')) ORDER BY rand()}, + ) + end + + it "omits the delete filter when the engine declares no is_deleted column" do + sql = ch.from(:test_replacing_no_delete).deduplicate.sql + expect(sql).to eq( + 'SELECT * FROM (SELECT * FROM "test_replacing_no_delete" ' \ + 'ORDER BY "version" DESC LIMIT 1 BY id) AS "t1" ORDER BY rand()', + ) + end + + it "resolves the sorting key of a Distributed table through the local table" do + sql = ch.from(:test_replacing_distributed).where(id: 1).deduplicate.sql + expect(sql).to include("LIMIT 1 BY group_id, id") + end + + it "preserves the source alias so qualified references keep working" do + ds = ch.from(Sequel[:test_replacing].as(:rows)).deduplicate + expect(ds.sql).to include('AS "rows"') + expect(ds.where(Sequel[:rows][:payload] => "x").sql).to include('"rows"."payload"') + end + + it "keeps a caller's projection outside the subquery" do + sql = ch.from(:test_replacing).select(:id).deduplicate.sql + expect(sql).to start_with('SELECT "id" FROM (SELECT * FROM "test_replacing"') + expect(sql).to include('WHERE ("is_deleted" = 0)') + end + + it "orders by version first and keeps the caller's ordering as a tiebreaker" do + sql = ch.from(:test_replacing).order(:payload).deduplicate.sql + expect(sql).to include('ORDER BY "version" DESC, "payload" LIMIT 1 BY') + end + end + + describe "refusals" do + it "refuses a joined dataset" do + ds = ch.from(:test_replacing).join(Sequel[:test].as(:t), id: :id) + expect { ds.deduplicate }.to raise_error(Sequel::Error, /single table source/) + end + + it "refuses more than one source" do + ds = ch.from(:test_replacing).from(:test_replacing, :test) + expect { ds.deduplicate }.to raise_error(Sequel::Error, /single table source/) + end + + it "refuses a subquery source" do + ds = ch.from(ch.from(:test_replacing)) + expect { ds.deduplicate }.to raise_error(Sequel::Error, /single table source/) + end + + it "refuses a non-replacing engine" do + expect { ch.from(:test).deduplicate } + .to raise_error(Sequel::Error, /MergeTree.*ReplacingMergeTree/) + end + + it "refuses a Replacing table with no version column" do + expect { ch.from(:test_replacing_no_version).deduplicate } + .to raise_error(Sequel::Error, /declares no version column/) + end + end + + describe "results" do + before do + ch.insert("test_replacing", rows:) + ch.optimize_table!("test_replacing") + end + + let(:rows) do + [ + { group_id: 1, id: 1, payload: "old", version: 1, is_deleted: 0 }, + { group_id: 1, id: 1, payload: "new", version: 2, is_deleted: 0 }, + { group_id: 1, id: 2, payload: "kept", version: 1, is_deleted: 0 }, + { group_id: 2, id: 3, payload: "gone", version: 1, is_deleted: 0 }, + { group_id: 2, id: 3, payload: "gone", version: 2, is_deleted: 1 }, + ] + end + + it "keeps the newest version of each key and drops deleted rows" do + query = ch.from(:test_replacing).deduplicate.order(:id).select(:id, :payload) + expect(ch.query(query)).to eq([{ id: 1, payload: "new" }, { id: 2, payload: "kept" }]) + end + + it "matches what FINAL returns" do + deduped = ch.query(ch.from(:test_replacing).deduplicate.order(:id).select(:id, :payload)) + final = ch.query( + ch.from(:test_replacing).order(:id).select(:id, :payload).where(is_deleted: 0), + final: 1, + ) + expect(deduped).to eq(final) + end + end + end +end diff --git a/spec/umbrellio_utils/click_house/final_settings_spec.rb b/spec/umbrellio_utils/click_house/final_settings_spec.rb new file mode 100644 index 0000000..f4fda02 --- /dev/null +++ b/spec/umbrellio_utils/click_house/final_settings_spec.rb @@ -0,0 +1,89 @@ +# frozen_string_literal: true + +describe UmbrellioUtils::ClickHouse, "FINAL settings" do + let(:ch) { described_class } + let(:backend) { ch.backend } + + describe "#settings_for" do + subject(:settings) { backend.send(:settings_for, dataset, opts) } + + let(:opts) { {} } + + context "with a deduplicated dataset" do + let(:dataset) { ch.from(:test_replacing).deduplicate } + + it "disables FINAL" do + expect(settings).to eq(final: 0) + end + + context "when the caller passes final explicitly" do + let(:opts) { { final: 1 } } + + it "leaves the caller's value alone" do + expect(settings).to eq(final: 1) + end + end + + context "with other settings present" do + let(:opts) { { max_threads: 2 } } + + it "keeps them" do + expect(settings).to eq(max_threads: 2, final: 0) + end + end + end + + context "with a plain dataset" do + let(:dataset) { ch.from(:test_replacing) } + + it "adds nothing" do + expect(settings).to eq({}) + end + end + + context "with a raw SQL string" do + let(:dataset) { "SELECT 1" } + + it "adds nothing" do + expect(settings).to eq({}) + end + end + end + + describe "settings reaching the driver" do + # The dataset is built before the stub: #deduplicate itself queries + # system.tables, which would otherwise be the call that gets recorded. + it "sends final=0 for a deduplicated query" do + dataset = ch.from(:test_replacing).deduplicate + allow(backend).to receive(:select_all).and_return([]) + ch.query(dataset) + expect(backend).to have_received(:select_all).with(anything, hash_including(final: 0)) + end + + it "sends no final override for a plain query" do + dataset = ch.from(:test_replacing) + allow(backend).to receive(:select_all).and_return([]) + ch.query(dataset) + expect(backend).to have_received(:select_all).with(anything, hash_not_including(:final)) + end + end + + describe "#count" do + before do + ch.truncate_table!("test_replacing") + ch.insert("test_replacing", rows: [ + { group_id: 1, id: 1, payload: "old", version: 1, is_deleted: 0 }, + { group_id: 1, id: 1, payload: "new", version: 2, is_deleted: 0 }, + ]) + ch.optimize_table!("test_replacing") + end + + it "counts deduplicated rows once" do + expect(ch.count(ch.from(:test_replacing).deduplicate)).to eq(1) + end + + it "forwards settings" do + expect(ch.count(ch.from(:test_replacing), final: 1)).to eq(1) + end + end +end diff --git a/spec/umbrellio_utils/click_house/table_metadata_spec.rb b/spec/umbrellio_utils/click_house/table_metadata_spec.rb new file mode 100644 index 0000000..dd0c2d2 --- /dev/null +++ b/spec/umbrellio_utils/click_house/table_metadata_spec.rb @@ -0,0 +1,126 @@ +# frozen_string_literal: true + +describe UmbrellioUtils::ClickHouse::TableMetadata do + describe ".parse" do + def parse(engine_full, sorting_key: "id") + engine = engine_full[/\A\w+/] + described_class.parse(engine:, engine_full:, sorting_key:) + end + + it "reads version and is_deleted from ReplacingMergeTree" do + meta = parse("ReplacingMergeTree(version, is_deleted) ORDER BY id") + expect(meta.version).to eq(:version) + expect(meta.is_deleted).to eq(:is_deleted) + end + + it "reads a version-only ReplacingMergeTree" do + meta = parse("ReplacingMergeTree(version) ORDER BY id") + expect(meta.version).to eq(:version) + expect(meta.is_deleted).to be_nil + end + + it "handles ReplacingMergeTree without arguments" do + meta = parse("ReplacingMergeTree PARTITION BY toYYYYMM(created_at) ORDER BY id") + expect(meta.version).to be_nil + expect(meta.is_deleted).to be_nil + end + + it "skips the zookeeper path and replica of ReplicatedReplacingMergeTree" do + meta = parse( + "ReplicatedReplacingMergeTree('/clickhouse/tables/{shard}/db/t', '{replica}', " \ + "updated_at, is_deleted) PARTITION BY toYYYYMM(created_at) ORDER BY id", + ) + expect(meta.version).to eq(:updated_at) + expect(meta.is_deleted).to eq(:is_deleted) + end + + it "handles a replicated version-only table" do + meta = parse( + "ReplicatedReplacingMergeTree('/tables/{shard}/t', '{replica}', ver) ORDER BY id", + ) + expect(meta.version).to eq(:ver) + expect(meta.is_deleted).to be_nil + end + + it "handles a replicated table with no replacing arguments" do + meta = parse("ReplicatedReplacingMergeTree('/tables/{shard}/t', '{replica}') ORDER BY id") + expect(meta.version).to be_nil + expect(meta.is_deleted).to be_nil + end + + it "splits a sorting key containing function calls" do + meta = parse( + "ReplacingMergeTree(v) ORDER BY id", + sorting_key: "toYYYYMM(created_at), project_id, player_id, id", + ) + expect(meta.sorting_key).to eq( + ["toYYYYMM(created_at)", "project_id", "player_id", "id"], + ) + end + + it "treats an empty sorting key as no columns" do + expect(parse("ReplacingMergeTree(v) ORDER BY id", sorting_key: "").sorting_key).to eq([]) + end + + it "reports a non-replacing engine without version columns" do + meta = parse("MergeTree PARTITION BY toYYYYMM(created_at) ORDER BY id") + expect(meta).to have_attributes( + engine: "MergeTree", replacing?: false, version: nil, is_deleted: nil, + ) + expect(meta.sorting_key).to eq(%w[id]) + end + + it "does not treat a parenthesis inside a quoted argument as structure" do + meta = parse( + "ReplicatedReplacingMergeTree('/clickhouse/tables/{shard}/db)/t', '{replica}', " \ + "version, is_deleted) ORDER BY id", + ) + expect(meta.version).to eq(:version) + expect(meta.is_deleted).to eq(:is_deleted) + end + end + + describe ".distributed_target" do + it "extracts database and table from a Distributed engine" do + expect( + described_class.distributed_target( + "Distributed('click_cluster', 'unetsafe', 'external_operations', order_id)", + ), + ).to eq(%w[unetsafe external_operations]) + end + + it "handles a Distributed engine without a sharding key" do + expect(described_class.distributed_target("Distributed('cluster', 'db', 'tbl')")) + .to eq(%w[db tbl]) + end + end +end + +describe UmbrellioUtils::ClickHouse do + let(:ch) { described_class } + + describe "#table_metadata" do + it "reads a local ReplacingMergeTree table" do + meta = ch.table_metadata(:test_replacing) + expect(meta.sorting_key).to eq(%w[group_id id]) + expect(meta.version).to eq(:version) + expect(meta.is_deleted).to eq(:is_deleted) + end + + it "resolves a Distributed table to its local table" do + expect(ch.table_metadata(:test_replacing_distributed)) + .to eq(ch.table_metadata(:test_replacing)) + end + + it "raises for an unknown table" do + expect { ch.table_metadata(:no_such_table_here) } + .to raise_error(described_class::TableMetadata::UnknownTable, /no_such_table_here/) + end + + it "answers the sorting key of a non-replacing table" do + meta = ch.table_metadata(:test) + expect(meta).to have_attributes(engine: "MergeTree", replacing?: false) + expect(meta.sorting_key).to eq(%w[id]) + end + end +end diff --git a/spec/umbrellio_utils/clickhouse_spec.rb b/spec/umbrellio_utils/clickhouse_spec.rb index 06f87b1..51e35c4 100644 --- a/spec/umbrellio_utils/clickhouse_spec.rb +++ b/spec/umbrellio_utils/clickhouse_spec.rb @@ -55,6 +55,38 @@ end end + describe "#limit_by" do + it "emits LIMIT 1 BY for a single expression" do + expect(ch.from(:test).order(:id).limit_by(:id).sql).to eq( + 'SELECT * FROM "test" ORDER BY "id" LIMIT 1 BY "id"', + ) + end + + it "emits every expression and honors n" do + ds = ch.from(:test).order(:id).limit_by(:id, :name, rows: 2) + expect(ds.sql).to eq('SELECT * FROM "test" ORDER BY "id" LIMIT 2 BY "id", "name"') + end + + it "places LIMIT n BY before LIMIT/OFFSET" do + ds = ch.from(:test).order(:id).limit_by(:id).limit(2, 1) + expect(ds.sql).to eq('SELECT * FROM "test" ORDER BY "id" LIMIT 1 BY "id" LIMIT 2 OFFSET 1') + end + + it "accepts raw expressions" do + ds = ch.from(:test).order(:id).limit_by(Sequel.lit("id % 2")) + expect(ds.sql).to eq('SELECT * FROM "test" ORDER BY "id" LIMIT 1 BY id % 2') + end + + it "raises without expressions" do + expect { ch.from(:test).limit_by }.to raise_error(Sequel::Error, /at least one expression/) + end + + it "collapses rows by the given expression" do + query = ch.from(:test).order(Sequel.desc(:id)).limit_by(Sequel.lit("id % 2")).select(:id) + expect(ch.query(query)).to eq([{ id: 3 }, { id: 2 }]) + end + end + describe "#query" do specify do query = ch.from(:test).order(:id).select(:id)