fix(clickhouse): request LZ4 responses from ClickHouse 26.9+

.changelog/fix-clickhouse-lz4-responses.md

@@ -0,0 +1,5 @@
+---
+tidx: patch
+---
+
+Fixed the ClickHouse sink failing against ClickHouse 26.9+ with `decompression error: incorrect magic number` by requesting LZ4-framed responses via `network_compression_method=lz4`.

src/sync/ch_sink.rs

@@ -160,7 +160,11 @@ impl ClickHouseSink {
         }
 
         let url = url.trim_end_matches('/');
-        let mut base_client = clickhouse::Client::default().with_url(url);
+        // ClickHouse 26.9 frames `compress=1` responses with ZSTD by default,
+        // but the clickhouse crate only decodes LZ4 native frames.
+        let mut base_client = clickhouse::Client::default()
+            .with_url(url)
+            .with_option("network_compression_method", "lz4");
         if let Some(user) = user {
             base_client = base_client.with_user(user);
         }

tests/clickhouse_test.rs

@@ -3094,7 +3094,8 @@ async fn test_backfill_replays_blocks_with_stale_child_rows() {
     assert_eq!(receipt_result["data"][0]["gas_used"].as_i64(), Some(21_000));
 }
 
-/// A canonical block can coexist with an older distinct version until a merge.
+/// A canonical block can coexist with an older distinct version that
+/// ReplacingMergeTree has not collapsed (an unmerged part or another partition).
 /// Backfill must replace the entire block instead of treating the canonical row
 /// as sufficient.
 #[tokio::test]
@@ -3107,10 +3108,15 @@ async fn test_backfill_replays_blocks_with_extra_block_versions() {
     let blocks = vec![make_block(1)];
     writer::write_blocks(&pool, &blocks).await.unwrap();
 
-    let mut stale_blocks = blocks.clone();
-    stale_blocks[0].hash = vec![0xee; 32];
-    stale_blocks.extend(blocks.clone());
-    ch_sink.write_blocks(&stale_blocks).await.unwrap();
+    // Versions sharing a part are collapsed on insert, and stopping merges
+    // would also block the repair's DELETE mutation, so write the stale
+    // version separately into another monthly partition.
+    let mut stale_block = blocks[0].clone();
+    stale_block.hash = vec![0xee; 32];
+    stale_block.timestamp -= chrono::Duration::days(31);
+    stale_block.timestamp_ms = stale_block.timestamp.timestamp_millis();
+    ch_sink.write_blocks(&[stale_block]).await.unwrap();
+    ch_sink.write_blocks(&blocks).await.unwrap();
     assert_eq!(ch.table_count("blocks").await.unwrap(), 2);
 
     sinks.backfill_clickhouse(TEST_CHAIN_ID).await.unwrap();