From 3f3d3aa670696d0993ef1f5739fd24e0c7aa5e56 Mon Sep 17 00:00:00 2001
From: KeKs0r <marc.hoeffl@laserfocus.io>
Date: Mon, 30 Mar 2026 10:52:58 +0200
Subject: [PATCH] fix: add user = currentUser() filter to system table queries

Some ClickHouse Cloud instances enforce row-level security on
system.query_log, causing queries without a user filter to fail with
NOT_FOUND_COLUMN_IN_BLOCK. Adding the filter is a no-op on instances
without row policies and required on those with them.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
---
 .changeset/system-table-user-filter.md                   | 7 +++++++
 packages/clickhouse/src/index.ts                         | 5 +++--
 packages/plugin-backfill/src/async-backfill.ts           | 5 +++--
 packages/plugin-obsessiondb/src/query/remote-executor.ts | 5 +++--
 4 files changed, 16 insertions(+), 6 deletions(-)
 create mode 100644 .changeset/system-table-user-filter.md

diff --git a/.changeset/system-table-user-filter.md b/.changeset/system-table-user-filter.md
new file mode 100644
index 0000000..b08406c
--- /dev/null
+++ b/.changeset/system-table-user-filter.md
@@ -0,0 +1,7 @@
+---
+"@chkit/clickhouse": patch
+"@chkit/plugin-backfill": patch
+"@chkit/plugin-obsessiondb": patch
+---
+
+Add `user = currentUser()` filter to all system.processes and system.query_log queries to satisfy ClickHouse row-level security policies.
diff --git a/packages/clickhouse/src/index.ts b/packages/clickhouse/src/index.ts
index 6081976..a62dea6 100644
--- a/packages/clickhouse/src/index.ts
+++ b/packages/clickhouse/src/index.ts
@@ -327,7 +327,7 @@ export function createClickHouseExecutor(config: NonNullable<ChxConfig['clickhou
     async queryStatus(queryId: string, options?: { afterTime?: string }): Promise<QueryStatus> {
       try {
         const running = await client.query({
-          query: `SELECT read_rows, read_bytes, written_rows, written_bytes, elapsed FROM clusterAllReplicas('parallel_replicas', system.processes) WHERE query_id = {qid:String} SETTINGS skip_unavailable_shards = 1`,
+          query: `SELECT read_rows, read_bytes, written_rows, written_bytes, elapsed FROM clusterAllReplicas('parallel_replicas', system.processes) WHERE user = currentUser() AND query_id = {qid:String} SETTINGS skip_unavailable_shards = 1`,
           query_params: { qid: queryId },
           format: 'JSONEachRow',
         })
@@ -354,7 +354,8 @@ export function createClickHouseExecutor(config: NonNullable<ChxConfig['clickhou
         const log = await client.query({
           query: `SELECT type, written_rows, written_bytes, query_duration_ms, exception
 FROM clusterAllReplicas('parallel_replicas', system.query_log)
-WHERE query_id = {qid:String}
+WHERE user = currentUser()
+  AND query_id = {qid:String}
   AND type IN ('QueryFinish', 'ExceptionWhileProcessing')
   AND is_initial_query = 1
   AND query_start_time >= parseDateTimeBestEffort({after:String})
diff --git a/packages/plugin-backfill/src/async-backfill.ts b/packages/plugin-backfill/src/async-backfill.ts
index d86fa44..f393499 100644
--- a/packages/plugin-backfill/src/async-backfill.ts
+++ b/packages/plugin-backfill/src/async-backfill.ts
@@ -149,7 +149,7 @@ export async function syncProgress(
   const safePrefix = prefix.replace(/'/g, "''").replace(/%/g, '\\%').replace(/_/g, '\\_')
 
   const runningRows = await executor.query<{ query_id: string }>(
-    `SELECT query_id FROM clusterAllReplicas('parallel_replicas', system.processes) WHERE query_id LIKE '${safePrefix}%' SETTINGS skip_unavailable_shards = 1`
+    `SELECT query_id FROM clusterAllReplicas('parallel_replicas', system.processes) WHERE user = currentUser() AND query_id LIKE '${safePrefix}%' SETTINGS skip_unavailable_shards = 1`
   )
   const runningSet = new Set(runningRows.map((r) => r.query_id))
 
@@ -163,7 +163,8 @@ export async function syncProgress(
   }>(
     `SELECT query_id, type, written_rows, written_bytes, query_duration_ms, exception
 FROM clusterAllReplicas('parallel_replicas', system.query_log)
-WHERE query_id LIKE '${safePrefix}%'
+WHERE user = currentUser()
+  AND query_id LIKE '${safePrefix}%'
   AND type IN ('QueryFinish', 'ExceptionWhileProcessing')
   AND is_initial_query = 1
 ORDER BY event_time DESC
diff --git a/packages/plugin-obsessiondb/src/query/remote-executor.ts b/packages/plugin-obsessiondb/src/query/remote-executor.ts
index 4ec54e4..fbf4021 100644
--- a/packages/plugin-obsessiondb/src/query/remote-executor.ts
+++ b/packages/plugin-obsessiondb/src/query/remote-executor.ts
@@ -69,7 +69,7 @@ export function createRemoteExecutor(deps: {
         : ''
 
       const running = await executor.query<{ query_id: string }>(
-        `SELECT query_id FROM system.processes WHERE query_id = '${queryId}' LIMIT 1`,
+        `SELECT query_id FROM system.processes WHERE user = currentUser() AND query_id = '${queryId}' LIMIT 1`,
       )
       if (running.length > 0) return { status: 'running' as const }
 
@@ -82,7 +82,8 @@ export function createRemoteExecutor(deps: {
       }>(
         `SELECT type, written_rows, written_bytes, query_duration_ms, exception
 FROM system.query_log
-WHERE query_id = '${queryId}'
+WHERE user = currentUser()
+  AND query_id = '${queryId}'
   AND type IN ('QueryFinish', 'ExceptionWhileProcessing')
   ${afterFilter}
 ORDER BY event_time DESC