From 19e866850deb7c72035abfdf88cc2e4d277e8c7e Mon Sep 17 00:00:00 2001
From: Yuri Smirnov <tycoooon@gmail.com>
Date: Thu, 13 Aug 2026 15:55:24 +0300
Subject: [PATCH 1/7] Add limit_by dataset method for ClickHouse

---
 .../click_house/backends/base.rb              | 31 +++++++++++++++---
 spec/umbrellio_utils/clickhouse_spec.rb       | 32 +++++++++++++++++++
 2 files changed, 58 insertions(+), 5 deletions(-)

diff --git a/lib/umbrellio_utils/click_house/backends/base.rb b/lib/umbrellio_utils/click_house/backends/base.rb
index cedbbee..17d2a49 100644
--- a/lib/umbrellio_utils/click_house/backends/base.rb
+++ b/lib/umbrellio_utils/click_house/backends/base.rb
@@ -13,13 +13,34 @@ 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, n: 1)
+            raise Sequel::Error, "limit_by requires at least one expression" if exprs.empty?
+            clone(limit_by: { exprs:, n: })
+          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[:n])
+              sql << " BY "
+              expression_list_append(sql, limit_by[:exprs])
+            end
+
+            super
+          end
         end
 
         # Concrete backends implement the low-level ops (execute / query /
@@ -36,7 +57,7 @@ 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)
diff --git a/spec/umbrellio_utils/clickhouse_spec.rb b/spec/umbrellio_utils/clickhouse_spec.rb
index 06f87b1..829741c 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, n: 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)

From 1b3af6949f41e65f89767080397fda1680dedfe0 Mon Sep 17 00:00:00 2001
From: Yuri Smirnov <tycoooon@gmail.com>
Date: Thu, 13 Aug 2026 15:59:11 +0300
Subject: [PATCH 2/7] Introspect ClickHouse table metadata from system.tables

---
 lib/umbrellio_utils/click_house.rb            |   3 +-
 .../click_house/backends/base.rb              |  37 +++++-
 .../click_house/table_metadata.rb             | 102 ++++++++++++++++
 spec/support/clickhouse.rb                    |  17 +++
 .../click_house/table_metadata_spec.rb        | 113 ++++++++++++++++++
 spec/umbrellio_utils/clickhouse_spec.rb       |   2 +-
 6 files changed, 269 insertions(+), 5 deletions(-)
 create mode 100644 lib/umbrellio_utils/click_house/table_metadata.rb
 create mode 100644 spec/umbrellio_utils/click_house/table_metadata_spec.rb

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 17d2a49..cbd2042 100644
--- a/lib/umbrellio_utils/click_house/backends/base.rb
+++ b/lib/umbrellio_utils/click_house/backends/base.rb
@@ -25,16 +25,16 @@ def literal_string_append(sql, str)
 
           # ClickHouse `LIMIT n BY expr, ...` — keeps the first n rows per
           # distinct combination of the expressions, applied after ORDER BY.
-          def limit_by(*exprs, n: 1)
+          def limit_by(*exprs, rows: 1)
             raise Sequel::Error, "limit_by requires at least one expression" if exprs.empty?
-            clone(limit_by: { exprs:, n: })
+            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[:n])
+              literal_append(sql, limit_by[:rows])
               sql << " BY "
               expression_list_append(sql, limit_by[:exprs])
             end
@@ -64,6 +64,18 @@ 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
           config.database.to_sym
         end
@@ -165,6 +177,25 @@ def with_temp_table(
 
         protected
 
+        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/table_metadata.rb b/lib/umbrellio_utils/click_house/table_metadata.rb
new file mode 100644
index 0000000..b8fe204
--- /dev/null
+++ b/lib/umbrellio_utils/click_house/table_metadata.rb
@@ -0,0 +1,102 @@
+# frozen_string_literal: true
+
+module UmbrellioUtils
+  module ClickHouse
+    # Layout of a ReplacingMergeTree 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.
+    TableMetadata = Struct.new(: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 UnsupportedEngine < StandardError
+      end
+
+      class UnknownTable < StandardError
+      end
+
+      REPLICATED_ARGS_COUNT = 2 # zookeeper path + replica name
+
+      class << self
+        def parse(engine:, engine_full:, sorting_key:)
+          unless engine.include?("Replacing")
+            raise UnsupportedEngine,
+                  "#{engine} is not a ReplacingMergeTree; deduplicate is not applicable"
+          end
+
+          args = engine_args(engine_full)
+          args = args.drop(REPLICATED_ARGS_COUNT) if engine.start_with?("Replicated")
+
+          new(split_args(sorting_key), args[0]&.to_sym, args[1]&.to_sym)
+        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 = +""
+          depth = 0
+          in_string = false
+
+          source.to_s.each_char do |char|
+            case char
+            when "'" then in_string = !in_string
+            when "(" then depth += 1 unless in_string
+            when ")" then depth -= 1 unless in_string
+            when ","
+              if depth.zero? && !in_string
+                args << current.strip
+                current = +""
+                next
+              end
+            end
+
+            current << char
+          end
+
+          args << current.strip
+          args.reject(&:empty?)
+        end
+
+        private
+
+        def close_index(source, open_index)
+          depth = 0
+
+          source[open_index..].each_char.with_index(open_index) do |char, index|
+            depth += 1 if char == "("
+            depth -= 1 if char == ")"
+            return index if depth.zero?
+          end
+
+          raise UnsupportedEngine, "unbalanced parentheses in engine: #{source}"
+        end
+
+        def unquote(value)
+          value.to_s.delete_prefix("'").delete_suffix("'")
+        end
+      end
+    end
+  end
+end
diff --git a/spec/support/clickhouse.rb b/spec/support/clickhouse.rb
index 404dfb9..50b21d5 100644
--- a/spec/support/clickhouse.rb
+++ b/spec/support/clickhouse.rb
@@ -16,3 +16,20 @@
   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
+
+# 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/table_metadata_spec.rb b/spec/umbrellio_utils/click_house/table_metadata_spec.rb
new file mode 100644
index 0000000..8f50a52
--- /dev/null
+++ b/spec/umbrellio_utils/click_house/table_metadata_spec.rb
@@ -0,0 +1,113 @@
+# 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 "rejects a non-replacing engine" do
+      expect { parse("MergeTree PARTITION BY toYYYYMM(created_at) ORDER BY id") }
+        .to raise_error(described_class::UnsupportedEngine, /MergeTree/)
+    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 "raises for a non-replacing engine" do
+      expect { ch.table_metadata(:test) }
+        .to raise_error(described_class::TableMetadata::UnsupportedEngine)
+    end
+  end
+end
diff --git a/spec/umbrellio_utils/clickhouse_spec.rb b/spec/umbrellio_utils/clickhouse_spec.rb
index 829741c..51e35c4 100644
--- a/spec/umbrellio_utils/clickhouse_spec.rb
+++ b/spec/umbrellio_utils/clickhouse_spec.rb
@@ -63,7 +63,7 @@
     end
 
     it "emits every expression and honors n" do
-      ds = ch.from(:test).order(:id).limit_by(:id, :name, n: 2)
+      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
 

From e304a1816c227118644aca8d66ca90e984c8b485 Mon Sep 17 00:00:00 2001
From: Yuri Smirnov <tycoooon@gmail.com>
Date: Thu, 13 Aug 2026 16:01:10 +0300
Subject: [PATCH 3/7] Add deduplicate dataset method replacing table-wide FINAL

---
 .../click_house/backends/base.rb              | 56 ++++++++++++
 spec/support/clickhouse.rb                    |  8 ++
 .../click_house/deduplicate_spec.rb           | 85 +++++++++++++++++++
 3 files changed, 149 insertions(+)
 create mode 100644 spec/umbrellio_utils/click_house/deduplicate_spec.rb

diff --git a/lib/umbrellio_utils/click_house/backends/base.rb b/lib/umbrellio_utils/click_house/backends/base.rb
index cbd2042..9f34ec2 100644
--- a/lib/umbrellio_utils/click_house/backends/base.rb
+++ b/lib/umbrellio_utils/click_house/backends/base.rb
@@ -41,6 +41,62 @@ def select_limit_sql(sql)
 
             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.version
+              raise Sequel::Error,
+                    "#{table_name} declares no version column; deduplicate needs one"
+            end
+
+            inner = order(Sequel.desc(meta.version))
+              .limit_by(*meta.sorting_key.map { |expr| Sequel.lit(expr) })
+
+            wrapped = ClickHouse.from(alias_for_source ? inner.as(alias_for_source) : inner)
+              .clone(ch_dedup: true)
+
+            meta.is_deleted ? wrapped.where(meta.is_deleted => 0) : wrapped
+          end
+
+          private
+
+          def alias_for_source
+            source = Array(@opts[:from]).first
+            source.alias if source.is_a?(Sequel::SQL::AliasedExpression)
+          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, {}]
+            when Symbol
+              [source, {}]
+            else
+              raise Sequel::Error, "deduplicate requires a single table source"
+            end
+          end
         end
 
         # Concrete backends implement the low-level ops (execute / query /
diff --git a/spec/support/clickhouse.rb b/spec/support/clickhouse.rb
index 50b21d5..a04f572 100644
--- a/spec/support/clickhouse.rb
+++ b/spec/support/clickhouse.rb
@@ -26,6 +26,14 @@
   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
+
 # Distributed tables carry no sorting key of their own, so metadata lookup has
 # to resolve through to the local table.
 client.execute(<<~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..48a096f
--- /dev/null
+++ b/spec/umbrellio_utils/click_house/deduplicate_spec.rb
@@ -0,0 +1,85 @@
+# 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 "refuses a dataset without exactly one table source" 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
+    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

From b54bc229a7c293a9f9c8c5b1c48e066b12c7ff71 Mon Sep 17 00:00:00 2001
From: Yuri Smirnov <tycoooon@gmail.com>
Date: Thu, 13 Aug 2026 16:04:04 +0300
Subject: [PATCH 4/7] Auto-disable FINAL for deduplicated datasets and forward
 count settings

---
 .../click_house/backends/base.rb              | 15 +++-
 .../click_house/backends/legacy.rb            |  2 +
 .../click_house/backends/native.rb            |  9 +-
 .../click_house/final_settings_spec.rb        | 89 +++++++++++++++++++
 4 files changed, 110 insertions(+), 5 deletions(-)
 create mode 100644 spec/umbrellio_utils/click_house/final_settings_spec.rb

diff --git a/lib/umbrellio_utils/click_house/backends/base.rb b/lib/umbrellio_utils/click_house/backends/base.rb
index 9f34ec2..97ce3cc 100644
--- a/lib/umbrellio_utils/click_house/backends/base.rb
+++ b/lib/umbrellio_utils/click_house/backends/base.rb
@@ -116,8 +116,8 @@ def from(source, db_name: self.db_name)
           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.
@@ -233,6 +233,17 @@ def with_temp_table(
 
         protected
 
+        # `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)
diff --git a/lib/umbrellio_utils/click_house/backends/legacy.rb b/lib/umbrellio_utils/click_house/backends/legacy.rb
index 68e825f..2c1c834 100644
--- a/lib/umbrellio_utils/click_house/backends/legacy.rb
+++ b/lib/umbrellio_utils/click_house/backends/legacy.rb
@@ -17,6 +17,7 @@ def execute(sql, host: nil, **opts)
 
         def query(dataset, host: nil, **opts)
           sql = sql_for(dataset)
+          opts = settings_for(dataset, opts)
           log_errors(sql) do
             select_all(sql, host:, **opts).map { |x| Misc::StrictHash[x.symbolize_keys] }
           end
@@ -24,6 +25,7 @@ def query(dataset, host: nil, **opts)
 
         def query_value(dataset, host: nil, **opts)
           sql = sql_for(dataset)
+          opts = settings_for(dataset, opts)
           log_errors(sql) { select_value(sql, host:, **opts) }
         end
 
diff --git a/lib/umbrellio_utils/click_house/backends/native.rb b/lib/umbrellio_utils/click_house/backends/native.rb
index 347b36f..5aaa62f 100644
--- a/lib/umbrellio_utils/click_house/backends/native.rb
+++ b/lib/umbrellio_utils/click_house/backends/native.rb
@@ -30,17 +30,20 @@ def execute(sql, host: nil, **opts) # rubocop:disable Lint/UnusedMethodArgument
 
         def query(dataset, host: nil, **opts) # rubocop:disable Lint/UnusedMethodArgument
           sql = sql_for(dataset)
-          log_errors(sql) { pool.query(sql, settings: opts) }
+          settings = settings_for(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) }
+          settings = settings_for(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, &) }
+          settings = settings_for(dataset, opts)
+          log_errors(sql) { pool.query_each(sql, settings:, &) }
         end
 
         def insert(table_name, db_name: self.db_name, rows: [])
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

From 2d92322bc15613c35976a7d16f6a96b665086809 Mon Sep 17 00:00:00 2001
From: Yuri Smirnov <tycoooon@gmail.com>
Date: Thu, 13 Aug 2026 16:04:58 +0300
Subject: [PATCH 5/7] Document dedup helpers and bump to 1.16.0

---
 Gemfile.lock                   |  2 +-
 README.md                      | 27 +++++++++++++++++++++++++++
 lib/umbrellio_utils/version.rb |  2 +-
 3 files changed, 29 insertions(+), 2 deletions(-)

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..7b974bf 100644
--- a/README.md
+++ b/README.md
@@ -115,6 +115,33 @@ 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.
+
 ### Instrumentation
 
 The gem ships a set of opt-in files for collecting GVL and allocation stats
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

From e5e1852a6d43581bd4ef881a9f33b22934107bdc Mon Sep 17 00:00:00 2001
From: Yuri Smirnov <tycoooon@gmail.com>
Date: Thu, 13 Aug 2026 17:36:56 +0300
Subject: [PATCH 6/7] Address review: projection and ordering across the dedup
 boundary

---
 README.md                                     |  4 ++
 .../click_house/backends/base.rb              | 36 +++++++---
 .../click_house/backends/legacy.rb            | 10 ++-
 .../click_house/backends/native.rb            |  9 +--
 .../click_house/table_metadata.rb             | 67 ++++++++++---------
 spec/support/clickhouse.rb                    |  9 +++
 .../click_house/deduplicate_spec.rb           | 35 +++++++++-
 .../click_house/table_metadata_spec.rb        | 25 +++++--
 8 files changed, 137 insertions(+), 58 deletions(-)

diff --git a/README.md b/README.md
index 7b974bf..1bbe7fe 100644
--- a/README.md
+++ b/README.md
@@ -142,6 +142,10 @@ The sorting key, version column and `is_deleted` column are read from `system.ta
 
 `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/backends/base.rb b/lib/umbrellio_utils/click_house/backends/base.rb
index 97ce3cc..d15bdca 100644
--- a/lib/umbrellio_utils/click_house/backends/base.rb
+++ b/lib/umbrellio_utils/click_house/backends/base.rb
@@ -55,25 +55,40 @@ 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
 
-            inner = order(Sequel.desc(meta.version))
-              .limit_by(*meta.sorting_key.map { |expr| Sequel.lit(expr) })
-
-            wrapped = ClickHouse.from(alias_for_source ? inner.as(alias_for_source) : inner)
+            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
 
-          def alias_for_source
+          # 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.alias if source.is_a?(Sequel::SQL::AliasedExpression)
+            source.is_a?(Sequel::SQL::AliasedExpression) ? inner.as(source.alias) : inner
           end
 
           # => [table_name, {} | { db_name: ... }]
@@ -91,8 +106,6 @@ def deduplication_source
               [source.column, { db_name: source.table }]
             when Sequel::SQL::Identifier
               [source.value, {}]
-            when Symbol
-              [source, {}]
             else
               raise Sequel::Error, "deduplicate requires a single table source"
             end
@@ -233,6 +246,13 @@ 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.
diff --git a/lib/umbrellio_utils/click_house/backends/legacy.rb b/lib/umbrellio_utils/click_house/backends/legacy.rb
index 2c1c834..0012ddb 100644
--- a/lib/umbrellio_utils/click_house/backends/legacy.rb
+++ b/lib/umbrellio_utils/click_house/backends/legacy.rb
@@ -16,17 +16,15 @@ def execute(sql, host: nil, **opts)
         end
 
         def query(dataset, host: nil, **opts)
-          sql = sql_for(dataset)
-          opts = settings_for(dataset, opts)
+          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)
-          opts = settings_for(dataset, opts)
-          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 5aaa62f..051a824 100644
--- a/lib/umbrellio_utils/click_house/backends/native.rb
+++ b/lib/umbrellio_utils/click_house/backends/native.rb
@@ -29,20 +29,17 @@ 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)
-          settings = settings_for(dataset, 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)
-          settings = settings_for(dataset, 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)
-          settings = settings_for(dataset, opts)
+          sql, settings = prepare_query(dataset, opts)
           log_errors(sql) { pool.query_each(sql, settings:, &) }
         end
 
diff --git a/lib/umbrellio_utils/click_house/table_metadata.rb b/lib/umbrellio_utils/click_house/table_metadata.rb
index b8fe204..2fd0f44 100644
--- a/lib/umbrellio_utils/click_house/table_metadata.rb
+++ b/lib/umbrellio_utils/click_house/table_metadata.rb
@@ -2,36 +2,36 @@
 
 module UmbrellioUtils
   module ClickHouse
-    # Layout of a ReplacingMergeTree table, read out of `system.tables`.
+    # 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.
-    TableMetadata = Struct.new(:sorting_key, :version, :is_deleted)
+    # 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 UnsupportedEngine < StandardError
-      end
-
       class UnknownTable < StandardError
       end
 
       REPLICATED_ARGS_COUNT = 2 # zookeeper path + replica name
 
+      def replacing?
+        engine.include?("Replacing")
+      end
+
       class << self
         def parse(engine:, engine_full:, sorting_key:)
-          unless engine.include?("Replacing")
-            raise UnsupportedEngine,
-                  "#{engine} is not a ReplacingMergeTree; deduplicate is not applicable"
-          end
+          key = split_args(sorting_key)
+          return new(engine, key, nil, nil) unless engine.include?("Replacing")
 
           args = engine_args(engine_full)
           args = args.drop(REPLICATED_ARGS_COUNT) if engine.start_with?("Replicated")
 
-          new(split_args(sorting_key), args[0]&.to_sym, args[1]&.to_sym)
+          new(engine, key, args[0]&.to_sym, args[1]&.to_sym)
         end
 
         # Distributed('cluster', 'database', 'table'[, sharding_key])
@@ -56,41 +56,46 @@ def engine_args(engine_full)
         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 do |char|
+          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
-            when ","
-              if depth.zero? && !in_string
-                args << current.strip
-                current = +""
-                next
-              end
             end
 
-            current << char
+            yield(char, depth, in_string, index)
           end
-
-          args << current.strip
-          args.reject(&:empty?)
         end
 
-        private
-
         def close_index(source, open_index)
-          depth = 0
-
-          source[open_index..].each_char.with_index(open_index) do |char, index|
-            depth += 1 if char == "("
-            depth -= 1 if char == ")"
-            return index if depth.zero?
+          scan(source[open_index..]) do |_char, depth, in_string, index|
+            return open_index + index if depth.zero? && !in_string
           end
 
-          raise UnsupportedEngine, "unbalanced parentheses in engine: #{source}"
+          raise ArgumentError, "unbalanced parentheses in engine: #{source}"
         end
 
         def unquote(value)
diff --git a/spec/support/clickhouse.rb b/spec/support/clickhouse.rb
index a04f572..bfc5821 100644
--- a/spec/support/clickhouse.rb
+++ b/spec/support/clickhouse.rb
@@ -34,6 +34,15 @@
   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)
diff --git a/spec/umbrellio_utils/click_house/deduplicate_spec.rb b/spec/umbrellio_utils/click_house/deduplicate_spec.rb
index 48a096f..1de718a 100644
--- a/spec/umbrellio_utils/click_house/deduplicate_spec.rb
+++ b/spec/umbrellio_utils/click_house/deduplicate_spec.rb
@@ -45,10 +45,43 @@
         expect(ds.where(Sequel[:rows][:payload] => "x").sql).to include('"rows"."payload"')
       end
 
-      it "refuses a dataset without exactly one table source" do
+      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
diff --git a/spec/umbrellio_utils/click_house/table_metadata_spec.rb b/spec/umbrellio_utils/click_house/table_metadata_spec.rb
index 8f50a52..dd0c2d2 100644
--- a/spec/umbrellio_utils/click_house/table_metadata_spec.rb
+++ b/spec/umbrellio_utils/click_house/table_metadata_spec.rb
@@ -62,9 +62,21 @@ def parse(engine_full, sorting_key: "id")
       expect(parse("ReplacingMergeTree(v) ORDER BY id", sorting_key: "").sorting_key).to eq([])
     end
 
-    it "rejects a non-replacing engine" do
-      expect { parse("MergeTree PARTITION BY toYYYYMM(created_at) ORDER BY id") }
-        .to raise_error(described_class::UnsupportedEngine, /MergeTree/)
+    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
 
@@ -105,9 +117,10 @@ def parse(engine_full, sorting_key: "id")
         .to raise_error(described_class::TableMetadata::UnknownTable, /no_such_table_here/)
     end
 
-    it "raises for a non-replacing engine" do
-      expect { ch.table_metadata(:test) }
-        .to raise_error(described_class::TableMetadata::UnsupportedEngine)
+    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

From 188b0e0a4000d2dd047d35fb0e4d540e99f74220 Mon Sep 17 00:00:00 2001
From: Yuri Smirnov <tycoooon@gmail.com>
Date: Fri, 14 Aug 2026 10:22:04 +0300
Subject: [PATCH 7/7] Ask the struct itself whether the engine is Replacing

---
 lib/umbrellio_utils/click_house/table_metadata.rb | 9 ++++++---
 1 file changed, 6 insertions(+), 3 deletions(-)

diff --git a/lib/umbrellio_utils/click_house/table_metadata.rb b/lib/umbrellio_utils/click_house/table_metadata.rb
index 2fd0f44..516d23a 100644
--- a/lib/umbrellio_utils/click_house/table_metadata.rb
+++ b/lib/umbrellio_utils/click_house/table_metadata.rb
@@ -24,14 +24,17 @@ def 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:)
-          key = split_args(sorting_key)
-          return new(engine, key, nil, nil) unless engine.include?("Replacing")
+          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
 
-          new(engine, key, args[0]&.to_sym, args[1]&.to_sym)
+          meta
         end
 
         # Distributed('cluster', 'database', 'table'[, sharding_key])