Fix double-distributed query

Use local offsets table in BatchFinder CTE to avoid double-distributed
query error (ClickHouse error 288). The BatchFinder queries
access_logs_federated (distributed table), which executes on each shard.
Each shard should read its local offsets table rather than querying
offsets_federated, which would create a distributed query within a
distributed query.

Both tables are sharded by raftSessionID, so data is co-located and
each shard has the offsets it needs locally. CommitOffset and GetOffset
continue to use offsets_federated for proper cross-shard routing.

Author: dvasilas

pkg/clickhouse/constants.go

@@ -11,6 +11,9 @@ const (
 	// TableAccessLogsFederated is the distributed table across all shards
 	TableAccessLogsFederated = "access_logs_federated"
 
+	// TableOffsets is the local table tracking processing offsets per bucket (MergeTree)
+	TableOffsets = "offsets"
+
 	// TableOffsetsFederated is the distributed table for offsets across all shards
 	TableOffsetsFederated = "offsets_federated"
 )

pkg/logcourier/batchfinder.go

@@ -123,7 +123,7 @@ func (bf *BatchFinder) FindBatches(ctx context.Context) ([]LogBatch, error) {
            OR min_ts <= now() - INTERVAL ? SECOND
         ORDER BY min_ts ASC
         LIMIT ?
-    `, bf.database, clickhouse.TableOffsetsFederated, bf.database, clickhouse.TableAccessLogsFederated)
+    `, bf.database, clickhouse.TableOffsets, bf.database, clickhouse.TableAccessLogsFederated)
 
 	rows, err := bf.client.Query(ctx, query, bf.countThreshold, bf.timeThresholdSec, bf.maxBuckets)
 	if err != nil {