fix: add user = currentUser() filter to system table queries

.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.

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})

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

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