Batching ClickHouse Inserts: A Queue and Go Worker, Async Inserts and Deduplicated Retries

I still prefer to load ClickHouse through a queue and a Go worker I control. The worker collects rows into batches, so I choose the batch size and when to send. Async inserts suit setups with many writers sending small writes that they cannot batch themselves.

Retries depend on ClickHouse’s insert deduplication, with a limited retention window and other preconditions. Don’t count on exactly-once. For updates to Replicated tables, keep non-deterministic functions such as now() out of the query. A development setup using only plain MergeTree won’t catch that restriction.

I’ve covered parts, background merges, the sparse index and S3 tiering in ClickHouse Basics: MergeTree and Tiered Storage with S3. This post deals with writes and updates. The SQL examples ran on ClickHouse 26.9.2.1 through the engine embedded in chDB 4.4.0. I had no server, so the async insert and replication behaviour described here comes from the official documentation and source code; I didn’t run those cases. The Go example was compiled and unit-tested with Go 1.26.9 and clickhouse-go v2.49.0.

1. Row-by-row inserts leave a pile of parts

Every synchronous INSERT into a MergeTree table produces at least one part. Send one row at a time and you get one part per row (async inserts combine requests on the server first; more on that in section 4). The two tables below are identical, with background merges stopped on the first so the raw part count stays visible. One gets 200 single-row inserts, the other one insert of 200 rows:

CREATE TABLE bets (
  bet_id UInt64, player_id UInt32, amount Decimal(18,2), created_at DateTime
) ENGINE = MergeTree ORDER BY (player_id, created_at)
SETTINGS old_parts_lifetime = 0;
CREATE TABLE bets_batch AS bets;
SYSTEM STOP MERGES bets;

-- run 200 times in a loop, one row each
INSERT INTO bets VALUES (1, 1, 10.00, now());

-- 200 rows in one INSERT
INSERT INTO bets_batch
SELECT number + 1, (number + 1) % 7, 10.00, now() FROM numbers(200);

SELECT table, count() AS active_parts, sum(rows) AS rows
FROM system.parts
WHERE database = currentDatabase() AND active AND table IN ('bets', 'bets_batch')
GROUP BY table ORDER BY table;

   ┌─table──────┬─active_parts─┬─rows─┐
1. │ bets       │          200 │  200 │
2. │ bets_batch │            1 │  200 │
   └────────────┴──────────────┴──────┘

Same 200 rows: 200 parts on one side, one on the other. Merges don’t stop in production, but they cost disk reads and writes, and once inserts outpace merges the parts keep piling up. MergeTree has two thresholds. When the active parts in a single partition reach parts_to_delay_insert (default 1000), ClickHouse deliberately slows INSERTs down; at parts_to_throw_insert (default 3000) the INSERT fails with Too many parts (N). Merges are processing significantly slower than inserts, the error usually referred to as TOO_MANY_PARTS. Both checks apply only while the partition’s average part size stays at or below max_avg_part_size_for_too_many_parts (1 GiB by default).

A batch spanning several partitions produces one part per partition. If the partition key is too fine-grained, increasing the batch size still leaves you with many parts.

A bet-slip pipeline I worked on used this approach. Bet records went into a queue, where a Go worker collected them into batches for bulk inserts into ClickHouse. ClickHouse then split the data by age into hot, warm and cold tiers, with the cold tier on S3.

2. A queue and a Go worker: you choose batch size and timing

The worker controls batch size and send timing, and commits the offset only after a successful write. If batching or retries misbehave, I can go straight to its code and logs.

The worker sends when the batch is full or the timer fires. If the write fails, it resends the same rows with the same token and commits only after success. Here’s the write-and-retry code. The full program, including the queue-reading loop and a demo queue that reads JSON lines from stdin, passes go vet and compiles. This example uses a different bets table from section 1, with columns bet_id, player_id, amount_cents and created_at. Amounts are stored as integer cents:

// flush retries the same rows with the same token, then commits the queue offset.
func (w *Worker) flush(ctx context.Context, batch []Message) error {
	if len(batch) == 0 {
		return nil
	}
	if w.MaxRetries < 1 {
		return fmt.Errorf("MaxRetries must be at least 1, got %d", w.MaxRetries)
	}
	first, last := batch[0].Offset, batch[len(batch)-1].Offset
	token := fmt.Sprintf("bets-%d-%d", first, last)
	rows := make([]Bet, len(batch))
	for i, m := range batch {
		rows[i] = m.Bet
	}

	var err error
	for attempt := 1; attempt <= w.MaxRetries; attempt++ {
		if err = w.Insert(ctx, token, rows); err == nil {
			break
		}
		log.Printf("insert %s attempt %d: %v", token, attempt, err)
		if attempt == w.MaxRetries {
			break
		}
		select {
		case <-time.After(w.Backoff * time.Duration(attempt)):
		case <-ctx.Done():
			return ctx.Err()
		}
	}
	if err != nil {
		return fmt.Errorf("insert %s gave up: %w", token, err)
	}
	if err := w.Queue.Commit(ctx, last); err != nil {
		return fmt.Errorf("commit offset %d: %w", last, err)
	}
	return nil
}

// ClickHouseInsert writes one batch through the native protocol.
func ClickHouseInsert(conn driver.Conn) InsertFunc {
	return func(ctx context.Context, token string, rows []Bet) error {
		ctx = clickhouse.Context(ctx, clickhouse.WithSettings(clickhouse.Settings{
			"async_insert":               0, // the worker already batches
			"insert_deduplication_token": token,
		}))
		batch, err := conn.PrepareBatch(ctx, "INSERT INTO bets (bet_id, player_id, amount_cents, created_at)")
		if err != nil {
			return fmt.Errorf("prepare batch: %w", err)
		}
		defer batch.Close()
		for _, r := range rows {
			if err := batch.Append(r.BetID, r.PlayerID, r.AmountCents, r.CreatedAt); err != nil {
				return fmt.Errorf("append bet %d: %w", r.BetID, err)
			}
		}
		if err := batch.Send(); err != nil {
			return fmt.Errorf("send batch: %w", err)
		}
		return nil
	}
}

There was no ClickHouse server to connect to this time, so the write step is pulled out into an InsertFunc and the unit tests use a fake insert. The first test has five rows in batches of three, with the first two writes failing. The second sets MaxRetries to 0 and checks that the worker doesn’t skip the write and commit anyway (the log timestamps are from this run):

=== RUN   TestRetryReusesTokenAndCommitsAfterSuccess
2026/10/15 22:12:17 insert bets-1-3 attempt 1: connection reset
2026/10/15 22:12:17 insert bets-1-3 attempt 2: connection reset
    worker_test.go:52: insert calls: [bets-1-3 bets-1-3 bets-1-3 bets-4-5]
    worker_test.go:53: committed offsets: [3 5]
--- PASS: TestRetryReusesTokenAndCommitsAfterSuccess (0.00s)
=== RUN   TestZeroRetriesNeverCommits
    worker_test.go:77: Run: EOF
        MaxRetries must be at least 1, got 0
    worker_test.go:78: inserts: 0, committed offsets: []
--- PASS: TestZeroRetriesNeverCommits (0.00s)

All three attempts use bets-1-3, and offset 3 is committed only after the write succeeds. With MaxRetries at 0 there is no write and no commit. Without that check the retry loop never runs, err stays nil and the code goes straight to commit. I also pointed the program at a port with nothing listening: all five attempts failed with connection refused, the worker stopped, and no offset was committed.

async_insert: 0 disables the server-side buffer. Since 26.2, async_insert has been on by default, but this worker already batches. Each Send() is therefore one INSERT and one deduplication unit. On the native protocol, clickhouse-go’s Batch.Flush() sends the rows buffered so far. Using it splits a batch into several blocks, and a retry then deduplicates only if it reproduces the same block boundaries, so this example only uses Send().

3. Proxies like clickhouse-bulk and their dumps directory

An earlier project had no queue and put a bulk-insert proxy in front of ClickHouse; clickhouse-bulk is one such service. It accepts lots of small INSERTs and, once it reaches flush_count (default 10000; TabSeparated counts lines, other formats count each request as one) or flush_interval (default 1000 ms), sends them as one large INSERT. When ClickHouse returns an error, the unsent data is written to dump_dir (default dumps) and retried every dump_check_interval (default 300 seconds).

My concern was where the dumps directory lived. Put it on the same disk as ClickHouse and a full disk probably stops both from writing. Run the proxy in a container without mounting a volume for dumps, and the files live only in the container’s writable layer. Remove and recreate the container and the data waiting to be resent is gone. That’s why I was considering separate storage.

The README says nothing about rows still in memory, waiting to be flushed when the process stops. I would treat those as potentially lost.

With a queue, this directory is unnecessary. The data is already in the queue when the worker reads it; if ClickHouse can’t accept writes, the worker doesn’t commit. It can resend later, provided the queue persists data, supports re-reading and still retains the records.

4. Async inserts: for writers that can’t batch

With async inserts, the server does the batching. Clients send small INSERTs as before; the server holds them in memory and writes a part once a threshold is reached. These are the defaults in 26.9:

SELECT name, value FROM system.settings
WHERE name IN ('async_insert', 'wait_for_async_insert', 'async_insert_max_data_size',
  'async_insert_busy_timeout_max_ms', 'async_insert_max_query_number',
  'async_insert_use_adaptive_busy_timeout', 'deduplicate_insert')
ORDER BY name;

   ┌─name───────────────────────────────────┬─value────┐
1. │ async_insert                           │ 1        │
2. │ async_insert_busy_timeout_max_ms       │ 200      │
3. │ async_insert_max_data_size             │ 10485760 │
4. │ async_insert_max_query_number          │ 450      │
5. │ async_insert_use_adaptive_busy_timeout │ 1        │
6. │ deduplicate_insert                     │ enable   │
7. │ wait_for_async_insert                  │ 1        │
   └────────────────────────────────────────┴──────────┘

The buffer flushes when it reaches 10 MiB or 450 queries, or when 200 ms have passed since the first row arrived, whichever happens first. Since 24.2, the timeout has adapted to the incoming rate, between 50 ms and that maximum. ClickHouse Cloud uses a 100 MiB data limit and a one-second timeout by default.

Each node has its own buffers, kept separately per INSERT column structure and settings. You can’t query the data until it has been flushed.

wait_for_async_insert decides when the client gets its answer:

1 (the default): the server responds after the data has been flushed into the table, and flush errors are returned. This is what the documentation recommends for production.
0: the server responds as soon as the data is buffered, without guaranteeing persistence. The client never sees an error, and there is no dead-letter queue. system.asynchronous_inserts only lists requests still waiting to be flushed. For flush results and errors, check system.asynchronous_insert_log; recording requires the asynchronous_insert_log section to be enabled in the server configuration. The documentation calls this mode “very risky”. It bypasses backpressure and can overload the server.

The documentation suggests async inserts for clients that can’t batch themselves, such as observability setups where many agents continuously send small payloads. A worker you control can already manage its batch size, so it doesn’t have that problem. Hundreds of services you can’t change, all sending single rows, are a better fit: using async inserts is less work than adding batching to each service. They’re also more reliable than a Buffer table, which loses its buffered data on a crash.

Async inserts don’t make the parts problem go away. Each flush still produces at least one part per partition, and an overly fine partition key still runs into TOO_MANY_PARTS.

5. Retries rely on deduplication, which has preconditions

After a timeout, you don’t know whether ClickHouse wrote the data. All you can do is resend and rely on deduplication to make that safe. Payment APIs leave you in the same position, as I discussed in The Same Payment Sent Twice. ClickHouse deduplicates inserts per block, remembering the hashes of recently written blocks and skipping those it has already seen.

ReplicatedMergeTree remembers the last 10000 blocks by default, dropping any record more than 3600 seconds older than the newest one (replicated_deduplication_window and replicated_deduplication_window_seconds). For plain MergeTree, non_replicated_deduplication_window defaults to 0, which means no deduplication at all. Here is one table with a retention of 0 and one that keeps 1000 blocks, each receiving the same batch twice:

CREATE TABLE bets_nodedup (bet_id UInt64, player_id UInt32, amount Decimal(18,2))
ENGINE = MergeTree ORDER BY bet_id
SETTINGS non_replicated_deduplication_window = 0;
CREATE TABLE bets_dedup (bet_id UInt64, player_id UInt32, amount Decimal(18,2))
ENGINE = MergeTree ORDER BY bet_id
SETTINGS non_replicated_deduplication_window = 1000;

-- send the same batch twice to each table
INSERT INTO bets_nodedup VALUES (1, 7, 10.00), (2, 7, 25.00), (3, 9, 5.00);
INSERT INTO bets_nodedup VALUES (1, 7, 10.00), (2, 7, 25.00), (3, 9, 5.00);
INSERT INTO bets_dedup VALUES (1, 7, 10.00), (2, 7, 25.00), (3, 9, 5.00);
INSERT INTO bets_dedup VALUES (1, 7, 10.00), (2, 7, 25.00), (3, 9, 5.00);

SELECT 'bets_nodedup' AS t, count() FROM bets_nodedup
UNION ALL SELECT 'bets_dedup', count() FROM bets_dedup;

   ┌─t────────────┬─count()─┐
1. │ bets_nodedup │       6 │
2. │ bets_dedup   │       3 │
   └──────────────┴─────────┘

The table with no retention has an extra copy. Deduplication compares the data itself: the documentation specifies “same rows in the same order”. Change the row order on a retry and it is no longer recognised. The next insert sends the same three rows to bets_dedup in a different order. A third table, bets_token, uses the same settings and receives two INSERTs with the same token, again with the second insert reordered:

INSERT INTO bets_dedup VALUES (2, 7, 25.00), (1, 7, 10.00), (3, 9, 5.00);

CREATE TABLE bets_token AS bets_dedup;
INSERT INTO bets_token SETTINGS insert_deduplication_token = 'batch-000042'
VALUES (1, 7, 10.00), (2, 7, 25.00);
INSERT INTO bets_token SETTINGS insert_deduplication_token = 'batch-000042'
VALUES (2, 7, 25.00), (1, 7, 10.00);

SELECT 'bets_token' AS t, count() FROM bets_token
UNION ALL SELECT 'bets_dedup (reordered retry)', count() FROM bets_dedup;

   ┌─t────────────────────────────┬─count()─┐
1. │ bets_token                   │       2 │
2. │ bets_dedup (reordered retry) │       6 │
   └──────────────────────────────┴─────────┘

Once the order changes, bets_dedup accepts the same three rows again. With a token, ClickHouse uses that alone and ignores the data hash, so it catches the retry despite the different order. The worker uses the offset range as its token because re-reading from the queue may serialise the rows differently, while the offset range stays the same. That assumes each batch comes from a single queue partition in increasing offset order, so the first and last offsets describe the whole batch.

This example has a single source. With several queue partitions writing to the same table, include the partition name in the token. Otherwise, matching offset ranges from different partitions collide and distinct data gets discarded as a duplicate.

This still falls short of exactly-once. Duplicates can get through in several ways:

A worker that stalls for over an hour while other writes keep arriving may retry after its deduplication record has expired; the 3600 seconds are measured from the newest record. High write volume can do the same: 10000 blocks may quickly push the old entries out of the retention window.
After a restart, the worker re-reads from the last committed offset. Time-based batching can then produce different offset ranges, giving the same data different tokens. Batch boundaries have to be reproducible for tokens to help, for example by always using fixed offset ranges.
Using Flush() splits a batch into several blocks. So can sending CSV or JSONEachRow over HTTP: while parsing, the server splits blocks according to max_insert_block_size (default 1048449 rows in 26.9) and related settings. If a retry splits the blocks differently, deduplication no longer matches.

The table must also handle duplicates that insert deduplication misses. For bet records, a ReplacingMergeTree with bet_id as the sorting key leaves one row after a merge or when queried with FINAL, even if a duplicate was written. That still leaves conditions to account for: the queue usually guarantees only at-least-once delivery, insert deduplication retains a limited log, and table-level deduplication waits for a merge or FINAL. Until then, queries have to cope with the duplicates.

Two more related defaults changed in 26.2. deduplicate_insert defaults to enable, and async inserts share the same deduplication log as synchronous inserts; within an async buffer, each client query is still deduplicated on its own. Target tables of materialised views are deduplicated as well (deduplicate_blocks_in_dependent_materialized_views is on by default), but for async inserts each input block passing through a view may produce only one output block; a second one throws NOT_IMPLEMENTED. Before 26.2, deduplication of async inserts was off by default.

6. Updating Replicated tables: keep now() out

If you run ALTER TABLE ... UPDATE on a multi-node ClickHouse with now() in the SET clause, it can work fine in development and then fail once it reaches a replicated production setup.

An ALTER ... UPDATE on a Replicated table writes the command to ZooKeeper, then each replica rewrites its own parts. Evaluating now() at different times would give different results. ClickHouse rejects this in MutationsInterpreter.cpp for any table whose engine name starts with Replicated, with this error:

The source storage is replicated so ALTER UPDATE/ALTER DELETE statements must use only deterministic functions. Function '{}' is non-deterministic

The check depends on the table engine and settings; node count makes no difference. It lets non-Replicated tables through. On a plain MergeTree in chDB (a separate bets table with status and settled_at columns), ALTER TABLE bets UPDATE status = 'settled', settled_at = now() WHERE bet_id = 1 succeeds. According to the source, a Replicated table would be rejected even with a single replica. I couldn’t run that case because chDB has no Keeper and can’t create Replicated tables. Use only plain MergeTree in development and you’ll never see this error.

There are two settings to get around this. mutations_execute_nondeterministic_on_initiator (off by default) evaluates now() on the initiating node and sends the result to each replica as a constant. allow_nondeterministic_mutations (also off by default) simply permits it; the documentation leaves you responsible for keeping the nodes consistent. The first option is safer, though ALTER ... UPDATE is still a mutation that rewrites the affected data parts. That is expensive, and the documentation advises against frequent use.

When you need to set values like this, you can write a new version of the rows with INSERT INTO ... SELECT ... instead of running an UPDATE. An INSERT computes the data on the server running the query, then copies it to the replicas without evaluating it again on each one. Using the version column of ReplacingMergeTree, that looks like this:

CREATE TABLE bet_status (
  bet_id UInt64, status LowCardinality(String),
  settled_at Nullable(DateTime64(3)), version UInt64
) ENGINE = ReplacingMergeTree(version) ORDER BY bet_id;

INSERT INTO bet_status VALUES (1, 'open', NULL, 1), (2, 'open', NULL, 1), (3, 'open', NULL, 1);

INSERT INTO bet_status
SELECT bet_id, 'settled', toDateTime64('2026-10-15 21:30:00.000', 3), version + 1
FROM bet_status FINAL
WHERE bet_id IN (1, 2) AND status = 'open';

SELECT * FROM bet_status FINAL ORDER BY bet_id;
SELECT count() AS physical_rows FROM bet_status;

   ┌─bet_id─┬─status──┬──────────────settled_at─┬─version─┐
1. │      1 │ settled │ 2026-10-15 21:30:00.000 │       2 │
2. │      2 │ settled │ 2026-10-15 21:30:00.000 │       2 │
3. │      3 │ open    │                    ᴺᵁᴸᴸ │       1 │
   └────────┴─────────┴─────────────────────────┴─────────┘
   ┌─physical_rows─┐
1. │             5 │
   └───────────────┘

This run left five physical rows. Background merges were still running on this table, so a rerun may show fewer. FINAL returns the latest version of each bet_id; background merges clean up the old versions. The timestamp is a constant computed by the application, giving every replica the same value.

Concurrent updates remain unhandled. If two processes read version 1 and each write version 2, both rows get the same version. Version numbers therefore need a single writer, or a monotonically increasing value generated by the application.

Since 25.7, ClickHouse has supported standard SQL UPDATE through lightweight updates, which record changes in patch parts. The documentation currently marks this as beta. Tables need enable_block_number_column and enable_block_offset_column enabled first; without them, 26.9 returns Lightweight updates are not supported. The feature is intended for changes to about 10% of the table or less. Frequent small updates can still leave you with too many parts. I found no explanation of how it handles non-deterministic functions in the documentation, so that needs verification before I can draw a conclusion.

Use the same table engines and mutation settings in development as in production. If production uses Replicated tables with Keeper, development needs Replicated tables too, even if all you run is a single Keeper and a single replica. That gives you a chance to catch these restrictions before release.

7. The query side: keep players off the big table

In the bet-slip pipeline, querying the detail table directly could drag the whole system down. The plan was to sync aggregated results back to PostgreSQL so that player queries would stay stable. Players would look up their records there, leaving ClickHouse for reporting and analysis.

The reports needed today’s figures in real time. My first idea was to connect PostgreSQL to ClickHouse through an FDW, but that PostgreSQL was a managed cloud service and couldn’t install the FDW extension, so the back office ended up querying a ClickHouse view directly for the current day. I also considered the opposite direction, having ClickHouse write results into PostgreSQL through its PostgreSQL table engine, which supports both SELECT and INSERT against a remote table.

The daily summary for the same report used AggregatingMergeTree with a materialised view. At the time, new data seemed not to merge: the same player and day came back as several rows, and I triggered merges by hand with OPTIMIZE TABLE. Reproducing it later in chDB made the cause clear, and it’s normal AggregatingMergeTree behaviour. Every INSERT that passes through the view lands as a new part, and rows with the same sorting key are only combined by a background merge, which happens whenever it happens. The example below creates a bets_raw detail table with the same columns as section 2, stops merges, sends two INSERTs and reads the result two ways:

CREATE TABLE bets_raw (
  bet_id UInt64, player_id UInt32, amount_cents Int64, created_at DateTime
) ENGINE = MergeTree ORDER BY (player_id, created_at);

CREATE TABLE daily_turnover (
  day Date, player_id UInt32,
  bets AggregateFunction(count), turnover AggregateFunction(sum, Int64)
) ENGINE = AggregatingMergeTree ORDER BY (day, player_id);

CREATE MATERIALIZED VIEW daily_turnover_mv TO daily_turnover AS
SELECT toDate(created_at) AS day, player_id,
       countState() AS bets, sumState(amount_cents) AS turnover
FROM bets_raw GROUP BY day, player_id;

SYSTEM STOP MERGES daily_turnover;
INSERT INTO bets_raw VALUES (1, 7, 1000, '2026-10-15 10:00:00'), (2, 7, 2500, '2026-10-15 11:00:00');
INSERT INTO bets_raw VALUES (3, 7, 500, '2026-10-15 12:00:00');

SELECT day, player_id, finalizeAggregation(bets) AS bets, finalizeAggregation(turnover) AS turnover
FROM daily_turnover ORDER BY bets;

SELECT day, player_id, countMerge(bets) AS bets, sumMerge(turnover) AS turnover
FROM daily_turnover GROUP BY day, player_id;

   ┌────────day─┬─player_id─┬─bets─┬─turnover─┐
1. │ 2026-10-15 │         7 │    1 │      500 │
2. │ 2026-10-15 │         7 │    2 │     3500 │
   └────────────┴───────────┴──────┴──────────┘
   ┌────────day─┬─player_id─┬─bets─┬─turnover─┐
1. │ 2026-10-15 │         7 │    3 │     4000 │
   └────────────┴───────────┴──────┴──────────┘

Reading the table directly shows two partially aggregated rows. Following the documentation, querying with the same aggregate functions plus the -Merge suffix (here countMerge and sumMerge) and a GROUP BY gives the correct 3 bets and 4000 whether or not the parts have merged. After OPTIMIZE TABLE ... FINAL, a direct read also returns a single row, but that forces parts to be rewritten, and running it before every query isn’t practical. It’s a stopgap; the query side should always use -Merge with GROUP BY.

I also looked at sorting and partition keys, sharding, read/write separation and replica backups. Choosing the sorting and partition keys is covered in the basics post.

Not every project needs ClickHouse, either. Some systems deliberately go without it, keeping data tiered in Aurora PostgreSQL and exporting cold data to S3 as Parquet.

Notes

With synchronous inserts, each INSERT produces at least one part. Reaching 1000 active parts in a partition slows INSERTs down, and reaching 3000 fails with TOO_MANY_PARTS. With an overly fine partition key, bigger batches still produce a part per partition.
A queue and a Go worker let you control batch size and wait time. Commit the offset only after a successful write, retry with the same token, and validate MaxRetries or you may commit without writing. With a proxy like clickhouse-bulk instead, the dumps directory holding failed INSERTs should live apart from ClickHouse’s storage and survive the container being recreated.
async_insert has been on by default since 26.2 and suits writers that cannot batch. wait_for_async_insert=0 does not guarantee persistence; the documentation calls it very risky.
Deduplication compares a block’s data and row order. Plain MergeTree doesn’t deduplicate by default; Replicated tables keep 10000 blocks within 3600 seconds of the newest. insert_deduplication_token catches reordered resends, but expired records and changed batch boundaries still let duplicates through, so the table has to tolerate them.
AggregatingMergeTree rows are only combined by background merges, so query with -Merge combinators and GROUP BY; OPTIMIZE TABLE ... FINAL is a stopgap.
By default, ALTER ... UPDATE on a Replicated table cannot use non-deterministic functions such as now(), regardless of node count. A development setup using plain MergeTree won’t catch this engine restriction. Write a new version with INSERT INTO ... SELECT and pass the timestamp in from the application.