Merge pull request 'fix: cast DateTime64 to DateTime in ClickHouse TTL expression' (#98) from feature/clickhouse-phase1 into main

Reviewed-on: cameleer/cameleer3-server#98

.gitea/workflows/ci.yml

@@ -222,12 +222,21 @@ jobs:
             --from-literal=AUTHENTIK_SECRET_KEY="${AUTHENTIK_SECRET_KEY}" \
             --dry-run=client -o yaml | kubectl apply -f -
 
+          kubectl create secret generic clickhouse-credentials \
+            --namespace=cameleer \
+            --from-literal=CLICKHOUSE_USER="${CLICKHOUSE_USER:-default}" \
+            --from-literal=CLICKHOUSE_PASSWORD="$CLICKHOUSE_PASSWORD" \
+            --dry-run=client -o yaml | kubectl apply -f -
+
           kubectl apply -f deploy/postgres.yaml
           kubectl -n cameleer rollout status statefulset/postgres --timeout=120s
 
           kubectl apply -f deploy/opensearch.yaml
           kubectl -n cameleer rollout status statefulset/opensearch --timeout=180s
 
+          kubectl apply -f deploy/clickhouse.yaml
+          kubectl -n cameleer rollout status statefulset/clickhouse --timeout=180s
+
           kubectl apply -f deploy/authentik.yaml
           kubectl -n cameleer rollout status deployment/authentik-server --timeout=180s
 
@@ -253,6 +262,8 @@ jobs:
           AUTHENTIK_PG_USER: ${{ secrets.AUTHENTIK_PG_USER }}
           AUTHENTIK_PG_PASSWORD: ${{ secrets.AUTHENTIK_PG_PASSWORD }}
           AUTHENTIK_SECRET_KEY: ${{ secrets.AUTHENTIK_SECRET_KEY }}
+          CLICKHOUSE_USER: ${{ secrets.CLICKHOUSE_USER }}
+          CLICKHOUSE_PASSWORD: ${{ secrets.CLICKHOUSE_PASSWORD }}
 
   deploy-feature:
     needs: docker
@@ -292,7 +303,7 @@ jobs:
         run: kubectl create namespace "$BRANCH_NS" --dry-run=client -o yaml | kubectl apply -f -
       - name: Copy secrets from cameleer namespace
         run: |
-          for SECRET in gitea-registry postgres-credentials opensearch-credentials cameleer-auth; do
+          for SECRET in gitea-registry postgres-credentials opensearch-credentials clickhouse-credentials cameleer-auth; do
             kubectl get secret "$SECRET" -n cameleer -o json \
               | jq 'del(.metadata.namespace, .metadata.resourceVersion, .metadata.uid, .metadata.creationTimestamp, .metadata.managedFields)' \
               | kubectl apply -n "$BRANCH_NS" -f -

cameleer3-server-app/pom.xml

@@ -57,6 +57,12 @@
             <artifactId>opensearch-rest-client</artifactId>
             <version>2.19.0</version>
         </dependency>
+        <dependency>
+            <groupId>com.clickhouse</groupId>
+            <artifactId>clickhouse-jdbc</artifactId>
+            <version>0.9.7</version>
+            <classifier>all</classifier>
+        </dependency>
         <dependency>
             <groupId>org.springdoc</groupId>
             <artifactId>springdoc-openapi-starter-webmvc-ui</artifactId>
@@ -126,6 +132,11 @@
             <version>2.1.1</version>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.testcontainers</groupId>
+            <artifactId>testcontainers-clickhouse</artifactId>
+            <scope>test</scope>
+        </dependency>
         <dependency>
             <groupId>org.awaitility</groupId>
             <artifactId>awaitility</artifactId>

cameleer3-server-app/src/main/java/com/cameleer3/server/app/config/ClickHouseConfig.java

@@ -0,0 +1,34 @@
+package com.cameleer3.server.app.config;
+
+import com.zaxxer.hikari.HikariDataSource;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.jdbc.core.JdbcTemplate;
+
+import javax.sql.DataSource;
+
+@Configuration
+@EnableConfigurationProperties(ClickHouseProperties.class)
+@ConditionalOnProperty(name = "clickhouse.enabled", havingValue = "true")
+public class ClickHouseConfig {
+
+    @Bean(name = "clickHouseDataSource")
+    public DataSource clickHouseDataSource(ClickHouseProperties props) {
+        HikariDataSource ds = new HikariDataSource();
+        ds.setJdbcUrl(props.getUrl());
+        ds.setUsername(props.getUsername());
+        ds.setPassword(props.getPassword());
+        ds.setMaximumPoolSize(10);
+        ds.setPoolName("clickhouse-pool");
+        return ds;
+    }
+
+    @Bean(name = "clickHouseJdbcTemplate")
+    public JdbcTemplate clickHouseJdbcTemplate(
+            @Qualifier("clickHouseDataSource") DataSource ds) {
+        return new JdbcTemplate(ds);
+    }
+}

cameleer3-server-app/src/main/java/com/cameleer3/server/app/config/ClickHouseProperties.java

@@ -0,0 +1,20 @@
+package com.cameleer3.server.app.config;
+
+import org.springframework.boot.context.properties.ConfigurationProperties;
+
+@ConfigurationProperties(prefix = "clickhouse")
+public class ClickHouseProperties {
+
+    private String url = "jdbc:clickhouse://localhost:8123/cameleer";
+    private String username = "default";
+    private String password = "";
+
+    public String getUrl() { return url; }
+    public void setUrl(String url) { this.url = url; }
+
+    public String getUsername() { return username; }
+    public void setUsername(String username) { this.username = username; }
+
+    public String getPassword() { return password; }
+    public void setPassword(String password) { this.password = password; }
+}

cameleer3-server-app/src/main/java/com/cameleer3/server/app/config/ClickHouseSchemaInitializer.java

@@ -0,0 +1,56 @@
+package com.cameleer3.server.app.config;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.boot.context.event.ApplicationReadyEvent;
+import org.springframework.context.event.EventListener;
+import org.springframework.core.io.Resource;
+import org.springframework.core.io.support.PathMatchingResourcePatternResolver;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.stereotype.Component;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import java.util.Comparator;
+
+@Component
+@ConditionalOnProperty(name = "clickhouse.enabled", havingValue = "true")
+public class ClickHouseSchemaInitializer {
+
+    private static final Logger log = LoggerFactory.getLogger(ClickHouseSchemaInitializer.class);
+
+    private final JdbcTemplate clickHouseJdbc;
+
+    public ClickHouseSchemaInitializer(
+            @Qualifier("clickHouseJdbcTemplate") JdbcTemplate clickHouseJdbc) {
+        this.clickHouseJdbc = clickHouseJdbc;
+    }
+
+    @EventListener(ApplicationReadyEvent.class)
+    public void initializeSchema() {
+        try {
+            PathMatchingResourcePatternResolver resolver = new PathMatchingResourcePatternResolver();
+            Resource[] scripts = resolver.getResources("classpath:clickhouse/*.sql");
+
+            Arrays.sort(scripts, Comparator.comparing(Resource::getFilename));
+
+            for (Resource script : scripts) {
+                String sql = script.getContentAsString(StandardCharsets.UTF_8);
+                log.info("Executing ClickHouse schema script: {}", script.getFilename());
+                for (String statement : sql.split(";")) {
+                    String trimmed = statement.trim();
+                    if (!trimmed.isEmpty()) {
+                        clickHouseJdbc.execute(trimmed);
+                    }
+                }
+            }
+
+            log.info("ClickHouse schema initialization complete ({} scripts)", scripts.length);
+        } catch (Exception e) {
+            log.error("ClickHouse schema initialization failed — server will continue but ClickHouse features may not work", e);
+        }
+    }
+}

cameleer3-server-app/src/main/java/com/cameleer3/server/app/config/StorageBeanConfig.java

@@ -1,5 +1,9 @@
 package com.cameleer3.server.app.config;
 
+import com.cameleer3.server.app.storage.ClickHouseMetricsQueryStore;
+import com.cameleer3.server.app.storage.ClickHouseMetricsStore;
+import com.cameleer3.server.app.storage.PostgresMetricsQueryStore;
+import com.cameleer3.server.app.storage.PostgresMetricsStore;
 import com.cameleer3.server.core.admin.AuditRepository;
 import com.cameleer3.server.core.admin.AuditService;
 import com.cameleer3.server.core.detail.DetailService;
@@ -8,9 +12,12 @@ import com.cameleer3.server.core.ingestion.IngestionService;
 import com.cameleer3.server.core.ingestion.WriteBuffer;
 import com.cameleer3.server.core.storage.*;
 import com.cameleer3.server.core.storage.model.MetricsSnapshot;
+import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.jdbc.core.JdbcTemplate;
 
 @Configuration
 public class StorageBeanConfig {
@@ -41,4 +48,30 @@ public class StorageBeanConfig {
         return new IngestionService(executionStore, diagramStore, metricsBuffer,
                 searchIndexer::onExecutionUpdated, bodySizeLimit);
     }
+
+    @Bean
+    @ConditionalOnProperty(name = "cameleer.storage.metrics", havingValue = "clickhouse")
+    public MetricsStore clickHouseMetricsStore(
+            @Qualifier("clickHouseJdbcTemplate") JdbcTemplate clickHouseJdbc) {
+        return new ClickHouseMetricsStore(clickHouseJdbc);
+    }
+
+    @Bean
+    @ConditionalOnProperty(name = "cameleer.storage.metrics", havingValue = "postgres", matchIfMissing = true)
+    public MetricsStore postgresMetricsStore(JdbcTemplate jdbc) {
+        return new PostgresMetricsStore(jdbc);
+    }
+
+    @Bean
+    @ConditionalOnProperty(name = "cameleer.storage.metrics", havingValue = "clickhouse")
+    public MetricsQueryStore clickHouseMetricsQueryStore(
+            @Qualifier("clickHouseJdbcTemplate") JdbcTemplate clickHouseJdbc) {
+        return new ClickHouseMetricsQueryStore(clickHouseJdbc);
+    }
+
+    @Bean
+    @ConditionalOnProperty(name = "cameleer.storage.metrics", havingValue = "postgres", matchIfMissing = true)
+    public MetricsQueryStore postgresMetricsQueryStore(JdbcTemplate jdbc) {
+        return new PostgresMetricsQueryStore(jdbc);
+    }
 }

cameleer3-server-app/src/main/java/com/cameleer3/server/app/controller/AgentMetricsController.java

@@ -2,22 +2,23 @@ package com.cameleer3.server.app.controller;
 
 import com.cameleer3.server.app.dto.AgentMetricsResponse;
 import com.cameleer3.server.app.dto.MetricBucket;
-import org.springframework.jdbc.core.JdbcTemplate;
+import com.cameleer3.server.core.storage.MetricsQueryStore;
+import com.cameleer3.server.core.storage.model.MetricTimeSeries;
 import org.springframework.web.bind.annotation.*;
 
-import java.sql.Timestamp;
 import java.time.Instant;
 import java.time.temporal.ChronoUnit;
 import java.util.*;
+import java.util.stream.Collectors;
 
 @RestController
 @RequestMapping("/api/v1/agents/{agentId}/metrics")
 public class AgentMetricsController {
 
-    private final JdbcTemplate jdbc;
+    private final MetricsQueryStore metricsQueryStore;
 
-    public AgentMetricsController(JdbcTemplate jdbc) {
-        this.jdbc = jdbc;
+    public AgentMetricsController(MetricsQueryStore metricsQueryStore) {
+        this.metricsQueryStore = metricsQueryStore;
     }
 
     @GetMapping
@@ -32,34 +33,18 @@ public class AgentMetricsController {
         if (to == null) to = Instant.now();
 
         List<String> metricNames = Arrays.asList(names.split(","));
-        long intervalMs = (to.toEpochMilli() - from.toEpochMilli()) / Math.max(buckets, 1);
-        String intervalStr = intervalMs + " milliseconds";
 
-        Map<String, List<MetricBucket>> result = new LinkedHashMap<>();
-        for (String name : metricNames) {
-            result.put(name.trim(), new ArrayList<>());
-        }
+        Map<String, List<MetricTimeSeries.Bucket>> raw =
+                metricsQueryStore.queryTimeSeries(agentId, metricNames, from, to, buckets);
 
-        String sql = """
-            SELECT time_bucket(CAST(? AS interval), collected_at) AS bucket,
-                   metric_name,
-                   AVG(metric_value) AS avg_value
-            FROM agent_metrics
-            WHERE agent_id = ?
-              AND collected_at >= ? AND collected_at < ?
-              AND metric_name = ANY(?)
-            GROUP BY bucket, metric_name
-            ORDER BY bucket
-            """;
-
-        String[] namesArray = metricNames.stream().map(String::trim).toArray(String[]::new);
-        jdbc.query(sql, rs -> {
-            String metricName = rs.getString("metric_name");
-            Instant bucket = rs.getTimestamp("bucket").toInstant();
-            double value = rs.getDouble("avg_value");
-            result.computeIfAbsent(metricName, k -> new ArrayList<>())
-                  .add(new MetricBucket(bucket, value));
-        }, intervalStr, agentId, Timestamp.from(from), Timestamp.from(to), namesArray);
+        Map<String, List<MetricBucket>> result = raw.entrySet().stream()
+                .collect(Collectors.toMap(
+                        Map.Entry::getKey,
+                        e -> e.getValue().stream()
+                                .map(b -> new MetricBucket(b.time(), b.value()))
+                                .toList(),
+                        (a, b) -> a,
+                        LinkedHashMap::new));
 
         return new AgentMetricsResponse(result);
     }

cameleer3-server-app/src/main/java/com/cameleer3/server/app/storage/ClickHouseMetricsQueryStore.java

@@ -0,0 +1,66 @@
+package com.cameleer3.server.app.storage;
+
+import com.cameleer3.server.core.storage.MetricsQueryStore;
+import com.cameleer3.server.core.storage.model.MetricTimeSeries;
+import org.springframework.jdbc.core.JdbcTemplate;
+
+import java.time.Instant;
+import java.util.*;
+
+public class ClickHouseMetricsQueryStore implements MetricsQueryStore {
+
+    private final JdbcTemplate jdbc;
+
+    public ClickHouseMetricsQueryStore(JdbcTemplate jdbc) {
+        this.jdbc = jdbc;
+    }
+
+    @Override
+    public Map<String, List<MetricTimeSeries.Bucket>> queryTimeSeries(
+            String agentId, List<String> metricNames,
+            Instant from, Instant to, int buckets) {
+
+        long intervalSeconds = Math.max(60,
+                (to.getEpochSecond() - from.getEpochSecond()) / Math.max(buckets, 1));
+
+        Map<String, List<MetricTimeSeries.Bucket>> result = new LinkedHashMap<>();
+        for (String name : metricNames) {
+            result.put(name.trim(), new ArrayList<>());
+        }
+
+        String[] namesArray = metricNames.stream().map(String::trim).toArray(String[]::new);
+
+        // ClickHouse JDBC doesn't support array params with IN (?).
+        // Build the IN clause with properly escaped values.
+        StringBuilder inClause = new StringBuilder();
+        for (int i = 0; i < namesArray.length; i++) {
+            if (i > 0) inClause.append(", ");
+            inClause.append("'").append(namesArray[i].replace("'", "\\'")).append("'");
+        }
+
+        String finalSql = """
+            SELECT toStartOfInterval(collected_at, INTERVAL %d SECOND) AS bucket,
+                   metric_name,
+                   avg(metric_value) AS avg_value
+            FROM agent_metrics
+            WHERE agent_id = ?
+              AND collected_at >= ?
+              AND collected_at < ?
+              AND metric_name IN (%s)
+            GROUP BY bucket, metric_name
+            ORDER BY bucket
+            """.formatted(intervalSeconds, inClause);
+
+        jdbc.query(finalSql, rs -> {
+            String metricName = rs.getString("metric_name");
+            Instant bucket = rs.getTimestamp("bucket").toInstant();
+            double value = rs.getDouble("avg_value");
+            result.computeIfAbsent(metricName, k -> new ArrayList<>())
+                    .add(new MetricTimeSeries.Bucket(bucket, value));
+        }, agentId,
+           java.sql.Timestamp.from(from),
+           java.sql.Timestamp.from(to));
+
+        return result;
+    }
+}

cameleer3-server-app/src/main/java/com/cameleer3/server/app/storage/ClickHouseMetricsStore.java

@@ -0,0 +1,41 @@
+package com.cameleer3.server.app.storage;
+
+import com.cameleer3.server.core.storage.MetricsStore;
+import com.cameleer3.server.core.storage.model.MetricsSnapshot;
+import org.springframework.jdbc.core.JdbcTemplate;
+
+import java.sql.Timestamp;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+public class ClickHouseMetricsStore implements MetricsStore {
+
+    private final JdbcTemplate jdbc;
+
+    public ClickHouseMetricsStore(JdbcTemplate jdbc) {
+        this.jdbc = jdbc;
+    }
+
+    @Override
+    public void insertBatch(List<MetricsSnapshot> snapshots) {
+        if (snapshots.isEmpty()) return;
+
+        jdbc.batchUpdate("""
+                INSERT INTO agent_metrics (agent_id, metric_name, metric_value, tags, collected_at)
+                VALUES (?, ?, ?, ?, ?)
+                """,
+                snapshots.stream().map(s -> new Object[]{
+                        s.agentId(),
+                        s.metricName(),
+                        s.metricValue(),
+                        tagsToClickHouseMap(s.tags()),
+                        Timestamp.from(s.collectedAt())
+                }).toList());
+    }
+
+    private Map<String, String> tagsToClickHouseMap(Map<String, String> tags) {
+        if (tags == null || tags.isEmpty()) return new HashMap<>();
+        return new HashMap<>(tags);
+    }
+}

cameleer3-server-app/src/main/java/com/cameleer3/server/app/storage/PostgresMetricsQueryStore.java

@@ -0,0 +1,55 @@
+package com.cameleer3.server.app.storage;
+
+import com.cameleer3.server.core.storage.MetricsQueryStore;
+import com.cameleer3.server.core.storage.model.MetricTimeSeries;
+import org.springframework.jdbc.core.JdbcTemplate;
+
+import java.sql.Timestamp;
+import java.time.Instant;
+import java.util.*;
+
+public class PostgresMetricsQueryStore implements MetricsQueryStore {
+
+    private final JdbcTemplate jdbc;
+
+    public PostgresMetricsQueryStore(JdbcTemplate jdbc) {
+        this.jdbc = jdbc;
+    }
+
+    @Override
+    public Map<String, List<MetricTimeSeries.Bucket>> queryTimeSeries(
+            String agentId, List<String> metricNames,
+            Instant from, Instant to, int buckets) {
+
+        long intervalMs = (to.toEpochMilli() - from.toEpochMilli()) / Math.max(buckets, 1);
+        String intervalStr = intervalMs + " milliseconds";
+
+        Map<String, List<MetricTimeSeries.Bucket>> result = new LinkedHashMap<>();
+        for (String name : metricNames) {
+            result.put(name.trim(), new ArrayList<>());
+        }
+
+        String sql = """
+            SELECT time_bucket(CAST(? AS interval), collected_at) AS bucket,
+                   metric_name,
+                   AVG(metric_value) AS avg_value
+            FROM agent_metrics
+            WHERE agent_id = ?
+              AND collected_at >= ? AND collected_at < ?
+              AND metric_name = ANY(?)
+            GROUP BY bucket, metric_name
+            ORDER BY bucket
+            """;
+
+        String[] namesArray = metricNames.stream().map(String::trim).toArray(String[]::new);
+        jdbc.query(sql, rs -> {
+            String metricName = rs.getString("metric_name");
+            Instant bucket = rs.getTimestamp("bucket").toInstant();
+            double value = rs.getDouble("avg_value");
+            result.computeIfAbsent(metricName, k -> new ArrayList<>())
+                    .add(new MetricTimeSeries.Bucket(bucket, value));
+        }, intervalStr, agentId, Timestamp.from(from), Timestamp.from(to), namesArray);
+
+        return result;
+    }
+}

cameleer3-server-app/src/main/java/com/cameleer3/server/app/storage/PostgresMetricsStore.java

@@ -5,12 +5,10 @@ import com.cameleer3.server.core.storage.model.MetricsSnapshot;
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import org.springframework.jdbc.core.JdbcTemplate;
-import org.springframework.stereotype.Repository;
 
 import java.sql.Timestamp;
 import java.util.List;
 
-@Repository
 public class PostgresMetricsStore implements MetricsStore {
 
     private static final ObjectMapper MAPPER = new ObjectMapper();

cameleer3-server-app/src/main/resources/application.yml

@@ -48,6 +48,8 @@ opensearch:
 cameleer:
   body-size-limit: ${CAMELEER_BODY_SIZE_LIMIT:16384}
   retention-days: ${CAMELEER_RETENTION_DAYS:30}
+  storage:
+    metrics: ${CAMELEER_STORAGE_METRICS:postgres}
 
 security:
   access-token-expiry-ms: 3600000
@@ -66,6 +68,12 @@ springdoc:
   swagger-ui:
     path: /api/v1/swagger-ui
 
+clickhouse:
+  enabled: ${CLICKHOUSE_ENABLED:false}
+  url: ${CLICKHOUSE_URL:jdbc:clickhouse://localhost:8123/cameleer}
+  username: ${CLICKHOUSE_USERNAME:default}
+  password: ${CLICKHOUSE_PASSWORD:}
+
 management:
   endpoints:
     web:

cameleer3-server-app/src/main/resources/clickhouse/V1__agent_metrics.sql

@@ -0,0 +1,14 @@
+CREATE TABLE IF NOT EXISTS agent_metrics (
+    tenant_id          LowCardinality(String) DEFAULT 'default',
+    collected_at       DateTime64(3),
+    agent_id           LowCardinality(String),
+    metric_name        LowCardinality(String),
+    metric_value       Float64,
+    tags               Map(String, String) DEFAULT map(),
+    server_received_at DateTime64(3) DEFAULT now64(3)
+)
+ENGINE = MergeTree()
+PARTITION BY (tenant_id, toYYYYMM(collected_at))
+ORDER BY (tenant_id, agent_id, metric_name, collected_at)
+TTL toDateTime(collected_at) + INTERVAL 365 DAY DELETE
+SETTINGS index_granularity = 8192;

cameleer3-server-app/src/test/java/com/cameleer3/server/app/AbstractPostgresIT.java

@@ -7,6 +7,7 @@ import org.springframework.jdbc.core.JdbcTemplate;
 import org.springframework.test.context.ActiveProfiles;
 import org.springframework.test.context.DynamicPropertyRegistry;
 import org.springframework.test.context.DynamicPropertySource;
+import org.testcontainers.clickhouse.ClickHouseContainer;
 import org.testcontainers.containers.PostgreSQLContainer;
 import org.testcontainers.utility.DockerImageName;
 
@@ -20,6 +21,7 @@ public abstract class AbstractPostgresIT {
 
     static final PostgreSQLContainer<?> postgres;
     static final OpensearchContainer<?> opensearch;
+    static final ClickHouseContainer clickhouse;
 
     static {
         postgres = new PostgreSQLContainer<>(TIMESCALEDB_IMAGE)
@@ -30,6 +32,9 @@ public abstract class AbstractPostgresIT {
 
         opensearch = new OpensearchContainer<>("opensearchproject/opensearch:2.19.0");
         opensearch.start();
+
+        clickhouse = new ClickHouseContainer("clickhouse/clickhouse-server:24.12");
+        clickhouse.start();
     }
 
     @Autowired
@@ -46,5 +51,9 @@ public abstract class AbstractPostgresIT {
         registry.add("spring.flyway.user", postgres::getUsername);
         registry.add("spring.flyway.password", postgres::getPassword);
         registry.add("opensearch.url", opensearch::getHttpHostAddress);
+        registry.add("clickhouse.enabled", () -> "true");
+        registry.add("clickhouse.url", clickhouse::getJdbcUrl);
+        registry.add("clickhouse.username", clickhouse::getUsername);
+        registry.add("clickhouse.password", clickhouse::getPassword);
     }
 }

cameleer3-server-app/src/test/java/com/cameleer3/server/app/storage/ClickHouseMetricsQueryStoreIT.java

@@ -0,0 +1,114 @@
+package com.cameleer3.server.app.storage;
+
+import com.cameleer3.server.core.storage.model.MetricTimeSeries;
+import com.zaxxer.hikari.HikariDataSource;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.testcontainers.clickhouse.ClickHouseContainer;
+import org.testcontainers.junit.jupiter.Container;
+import org.testcontainers.junit.jupiter.Testcontainers;
+
+import java.time.Instant;
+import java.util.List;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+@Testcontainers
+class ClickHouseMetricsQueryStoreIT {
+
+    @Container
+    static final ClickHouseContainer clickhouse =
+            new ClickHouseContainer("clickhouse/clickhouse-server:24.12");
+
+    private JdbcTemplate jdbc;
+    private ClickHouseMetricsQueryStore queryStore;
+
+    @BeforeEach
+    void setUp() {
+        HikariDataSource ds = new HikariDataSource();
+        ds.setJdbcUrl(clickhouse.getJdbcUrl());
+        ds.setUsername(clickhouse.getUsername());
+        ds.setPassword(clickhouse.getPassword());
+
+        jdbc = new JdbcTemplate(ds);
+
+        jdbc.execute("""
+            CREATE TABLE IF NOT EXISTS agent_metrics (
+                tenant_id          LowCardinality(String) DEFAULT 'default',
+                collected_at       DateTime64(3),
+                agent_id           LowCardinality(String),
+                metric_name        LowCardinality(String),
+                metric_value       Float64,
+                tags               Map(String, String) DEFAULT map(),
+                server_received_at DateTime64(3) DEFAULT now64(3)
+            )
+            ENGINE = MergeTree()
+            ORDER BY (tenant_id, agent_id, metric_name, collected_at)
+            """);
+
+        jdbc.execute("TRUNCATE TABLE agent_metrics");
+
+        // Seed test data: 6 data points across 1 hour for two metrics
+        Instant base = Instant.parse("2026-03-31T10:00:00Z");
+        for (int i = 0; i < 6; i++) {
+            Instant ts = base.plusSeconds(i * 600); // every 10 minutes
+            jdbc.update("INSERT INTO agent_metrics (agent_id, metric_name, metric_value, collected_at) VALUES (?, ?, ?, ?)",
+                    "agent-1", "cpu.usage", 50.0 + i * 5, java.sql.Timestamp.from(ts));
+            jdbc.update("INSERT INTO agent_metrics (agent_id, metric_name, metric_value, collected_at) VALUES (?, ?, ?, ?)",
+                    "agent-1", "memory.free", 1000.0 - i * 100, java.sql.Timestamp.from(ts));
+        }
+
+        queryStore = new ClickHouseMetricsQueryStore(jdbc);
+    }
+
+    @Test
+    void queryTimeSeries_returnsDataGroupedByMetric() {
+        Instant from = Instant.parse("2026-03-31T10:00:00Z");
+        Instant to = Instant.parse("2026-03-31T11:00:00Z");
+
+        Map<String, List<MetricTimeSeries.Bucket>> result =
+                queryStore.queryTimeSeries("agent-1", List.of("cpu.usage", "memory.free"), from, to, 6);
+
+        assertThat(result).containsKeys("cpu.usage", "memory.free");
+        assertThat(result.get("cpu.usage")).isNotEmpty();
+        assertThat(result.get("memory.free")).isNotEmpty();
+    }
+
+    @Test
+    void queryTimeSeries_bucketsAverageCorrectly() {
+        Instant from = Instant.parse("2026-03-31T10:00:00Z");
+        Instant to = Instant.parse("2026-03-31T11:00:00Z");
+
+        // 1 bucket for the entire hour = average of all 6 values
+        Map<String, List<MetricTimeSeries.Bucket>> result =
+                queryStore.queryTimeSeries("agent-1", List.of("cpu.usage"), from, to, 1);
+
+        assertThat(result.get("cpu.usage")).hasSize(1);
+        // Values: 50, 55, 60, 65, 70, 75 → avg = 62.5
+        assertThat(result.get("cpu.usage").get(0).value()).isCloseTo(62.5, org.assertj.core.data.Offset.offset(0.1));
+    }
+
+    @Test
+    void queryTimeSeries_noData_returnsEmptyLists() {
+        Instant from = Instant.parse("2025-01-01T00:00:00Z");
+        Instant to = Instant.parse("2025-01-01T01:00:00Z");
+
+        Map<String, List<MetricTimeSeries.Bucket>> result =
+                queryStore.queryTimeSeries("agent-1", List.of("cpu.usage"), from, to, 6);
+
+        assertThat(result.get("cpu.usage")).isEmpty();
+    }
+
+    @Test
+    void queryTimeSeries_unknownAgent_returnsEmpty() {
+        Instant from = Instant.parse("2026-03-31T10:00:00Z");
+        Instant to = Instant.parse("2026-03-31T11:00:00Z");
+
+        Map<String, List<MetricTimeSeries.Bucket>> result =
+                queryStore.queryTimeSeries("nonexistent", List.of("cpu.usage"), from, to, 6);
+
+        assertThat(result.get("cpu.usage")).isEmpty();
+    }
+}

cameleer3-server-app/src/test/java/com/cameleer3/server/app/storage/ClickHouseMetricsStoreIT.java

@@ -0,0 +1,108 @@
+package com.cameleer3.server.app.storage;
+
+import com.cameleer3.server.core.storage.model.MetricsSnapshot;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.testcontainers.clickhouse.ClickHouseContainer;
+import org.testcontainers.junit.jupiter.Container;
+import org.testcontainers.junit.jupiter.Testcontainers;
+
+import com.zaxxer.hikari.HikariDataSource;
+
+import java.time.Instant;
+import java.util.List;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+@Testcontainers
+class ClickHouseMetricsStoreIT {
+
+    @Container
+    static final ClickHouseContainer clickhouse =
+            new ClickHouseContainer("clickhouse/clickhouse-server:24.12");
+
+    private JdbcTemplate jdbc;
+    private ClickHouseMetricsStore store;
+
+    @BeforeEach
+    void setUp() {
+        HikariDataSource ds = new HikariDataSource();
+        ds.setJdbcUrl(clickhouse.getJdbcUrl());
+        ds.setUsername(clickhouse.getUsername());
+        ds.setPassword(clickhouse.getPassword());
+
+        jdbc = new JdbcTemplate(ds);
+
+        jdbc.execute("""
+            CREATE TABLE IF NOT EXISTS agent_metrics (
+                tenant_id          LowCardinality(String) DEFAULT 'default',
+                collected_at       DateTime64(3),
+                agent_id           LowCardinality(String),
+                metric_name        LowCardinality(String),
+                metric_value       Float64,
+                tags               Map(String, String) DEFAULT map(),
+                server_received_at DateTime64(3) DEFAULT now64(3)
+            )
+            ENGINE = MergeTree()
+            ORDER BY (tenant_id, agent_id, metric_name, collected_at)
+            """);
+
+        jdbc.execute("TRUNCATE TABLE agent_metrics");
+
+        store = new ClickHouseMetricsStore(jdbc);
+    }
+
+    @Test
+    void insertBatch_writesMetricsToClickHouse() {
+        List<MetricsSnapshot> batch = List.of(
+                new MetricsSnapshot("agent-1", Instant.parse("2026-03-31T10:00:00Z"),
+                        "cpu.usage", 75.5, Map.of("host", "server-1")),
+                new MetricsSnapshot("agent-1", Instant.parse("2026-03-31T10:00:01Z"),
+                        "memory.free", 1024.0, null)
+        );
+
+        store.insertBatch(batch);
+
+        Integer count = jdbc.queryForObject(
+                "SELECT count() FROM agent_metrics WHERE agent_id = 'agent-1'",
+                Integer.class);
+        assertThat(count).isEqualTo(2);
+    }
+
+    @Test
+    void insertBatch_storesTags() {
+        store.insertBatch(List.of(
+                new MetricsSnapshot("agent-2", Instant.parse("2026-03-31T10:00:00Z"),
+                        "disk.used", 500.0, Map.of("mount", "/data", "fs", "ext4"))
+        ));
+
+        // Just verify we can read back the row with tags
+        Integer count = jdbc.queryForObject(
+                "SELECT count() FROM agent_metrics WHERE agent_id = 'agent-2'",
+                Integer.class);
+        assertThat(count).isEqualTo(1);
+    }
+
+    @Test
+    void insertBatch_emptyList_doesNothing() {
+        store.insertBatch(List.of());
+
+        Integer count = jdbc.queryForObject("SELECT count() FROM agent_metrics", Integer.class);
+        assertThat(count).isEqualTo(0);
+    }
+
+    @Test
+    void insertBatch_nullTags_defaultsToEmptyMap() {
+        store.insertBatch(List.of(
+                new MetricsSnapshot("agent-3", Instant.parse("2026-03-31T10:00:00Z"),
+                        "cpu.usage", 50.0, null)
+        ));
+
+        Integer count = jdbc.queryForObject(
+                "SELECT count() FROM agent_metrics WHERE agent_id = 'agent-3'",
+                Integer.class);
+        assertThat(count).isEqualTo(1);
+    }
+}

cameleer3-server-core/src/main/java/com/cameleer3/server/core/storage/MetricsQueryStore.java

@@ -0,0 +1,14 @@
+package com.cameleer3.server.core.storage;
+
+import com.cameleer3.server.core.storage.model.MetricTimeSeries;
+
+import java.time.Instant;
+import java.util.List;
+import java.util.Map;
+
+public interface MetricsQueryStore {
+
+    Map<String, List<MetricTimeSeries.Bucket>> queryTimeSeries(
+            String agentId, List<String> metricNames,
+            Instant from, Instant to, int buckets);
+}

cameleer3-server-core/src/main/java/com/cameleer3/server/core/storage/model/MetricTimeSeries.java

@@ -0,0 +1,9 @@
+package com.cameleer3.server.core.storage.model;
+
+import java.time.Instant;
+import java.util.List;
+
+public record MetricTimeSeries(String metricName, List<Bucket> buckets) {
+
+    public record Bucket(Instant time, double value) {}
+}

deploy/base/server.yaml

@@ -75,6 +75,22 @@ spec:
                   name: cameleer-auth
                   key: CAMELEER_JWT_SECRET
                   optional: true
+            - name: CLICKHOUSE_ENABLED
+              value: "true"
+            - name: CLICKHOUSE_URL
+              value: "jdbc:clickhouse://clickhouse.cameleer.svc.cluster.local:8123/cameleer"
+            - name: CLICKHOUSE_USERNAME
+              valueFrom:
+                secretKeyRef:
+                  name: clickhouse-credentials
+                  key: CLICKHOUSE_USER
+            - name: CLICKHOUSE_PASSWORD
+              valueFrom:
+                secretKeyRef:
+                  name: clickhouse-credentials
+                  key: CLICKHOUSE_PASSWORD
+            - name: CAMELEER_STORAGE_METRICS
+              value: "postgres"
 
           resources:
             requests:

deploy/clickhouse.yaml

@@ -0,0 +1,103 @@
+apiVersion: apps/v1
+kind: StatefulSet
+metadata:
+  name: clickhouse
+  namespace: cameleer
+spec:
+  serviceName: clickhouse
+  replicas: 1
+  selector:
+    matchLabels:
+      app: clickhouse
+  template:
+    metadata:
+      labels:
+        app: clickhouse
+    spec:
+      containers:
+        - name: clickhouse
+          image: clickhouse/clickhouse-server:24.12
+          env:
+            - name: CLICKHOUSE_USER
+              valueFrom:
+                secretKeyRef:
+                  name: clickhouse-credentials
+                  key: CLICKHOUSE_USER
+            - name: CLICKHOUSE_PASSWORD
+              valueFrom:
+                secretKeyRef:
+                  name: clickhouse-credentials
+                  key: CLICKHOUSE_PASSWORD
+            - name: CLICKHOUSE_DEFAULT_ACCESS_MANAGEMENT
+              value: "1"
+          ports:
+            - containerPort: 8123
+              name: http
+            - containerPort: 9000
+              name: native
+          volumeMounts:
+            - name: data
+              mountPath: /var/lib/clickhouse
+            - name: initdb
+              mountPath: /docker-entrypoint-initdb.d
+          resources:
+            requests:
+              memory: "2Gi"
+              cpu: "500m"
+            limits:
+              memory: "4Gi"
+              cpu: "2000m"
+          livenessProbe:
+            httpGet:
+              path: /ping
+              port: 8123
+            initialDelaySeconds: 10
+            periodSeconds: 10
+            timeoutSeconds: 3
+            failureThreshold: 3
+          readinessProbe:
+            httpGet:
+              path: /ping
+              port: 8123
+            initialDelaySeconds: 5
+            periodSeconds: 5
+            timeoutSeconds: 3
+            failureThreshold: 3
+      volumes:
+        - name: initdb
+          configMap:
+            name: clickhouse-initdb
+  volumeClaimTemplates:
+    - metadata:
+        name: data
+      spec:
+        accessModes: ["ReadWriteOnce"]
+        resources:
+          requests:
+            storage: 50Gi
+---
+apiVersion: v1
+kind: Service
+metadata:
+  name: clickhouse
+  namespace: cameleer
+spec:
+  clusterIP: None
+  selector:
+    app: clickhouse
+  ports:
+    - port: 8123
+      targetPort: 8123
+      name: http
+    - port: 9000
+      targetPort: 9000
+      name: native
+---
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: clickhouse-initdb
+  namespace: cameleer
+data:
+  01-create-database.sql: |
+    CREATE DATABASE IF NOT EXISTS cameleer;

docs/superpowers/specs/2026-03-31-append-only-execution-protocol.md

@@ -0,0 +1,146 @@
+# Append-Only Execution Data Protocol
+
+A reference document for redesigning the Cameleer agent's data reporting to be append-only,
+eliminating the need for upserts in the storage layer.
+
+## Problem
+
+The current protocol sends execution data in two phases:
+
+1. **RUNNING phase**: Agent sends a partial record when a route starts executing (execution_id, route_id, start_time, status=RUNNING). No bodies, no duration, no error info.
+2. **COMPLETED/FAILED phase**: Agent sends an enriched record when execution finishes (duration, output body, headers, errors, processor tree).
+
+The server uses `INSERT ... ON CONFLICT DO UPDATE SET COALESCE(...)` to merge these into a single row. This works in PostgreSQL but creates problems for append-only stores like ClickHouse, Kafka topics, or any event-sourced architecture.
+
+### Why This Matters
+
+- **ClickHouse**: No native upsert. Must use ReplacingMergeTree (eventual consistency, FINAL overhead) or application-side buffering.
+- **Event streaming**: Kafka/Pulsar topics are append-only. Two-phase lifecycle requires a stateful stream processor to merge.
+- **Data lakes**: Parquet files are immutable. Updates require read-modify-write of entire files.
+- **Materialized views**: Insert-triggered aggregations (ClickHouse MVs, Kafka Streams, Flink) double-count if they see both RUNNING and COMPLETED inserts for the same execution.
+
+## Proposed Protocol Change
+
+### Option A: Single-Phase Reporting (Recommended)
+
+The agent buffers the execution locally and sends a **single, complete record** only when the execution reaches a terminal state (COMPLETED or FAILED).
+
+```
+Current:  Agent -> [RUNNING] -> Server -> [COMPLETED] -> Server (upsert)
+Proposed: Agent -> [buffer locally] -> [COMPLETED with all fields] -> Server (append)
+```
+
+**What changes in the agent:**
+- `RouteExecutionTracker` holds in-flight executions in a local `ConcurrentHashMap`
+- On route start: create tracker entry with start_time, route_id, etc.
+- On route complete: enrich tracker entry with duration, bodies, errors, processor tree
+- On report: send the complete record in one HTTP POST
+- On timeout (configurable, e.g., 5 minutes): flush as RUNNING (for visibility of stuck routes)
+
+**What changes in the server:**
+- Storage becomes pure append: `INSERT INTO executions VALUES (...)` — no upsert, no COALESCE
+- No `SearchIndexer` / `ExecutionAccumulator` needed — the server just writes what it receives
+- Materialized views count correctly (one insert = one execution)
+- Works with any append-only store (ClickHouse, Kafka, S3/Parquet)
+
+**Trade-offs:**
+- RUNNING executions are not visible on the server until they complete (or timeout-flush)
+- "Active execution count" must come from agent heartbeat/registry data, not from stored RUNNING rows
+- If the agent crashes, in-flight executions are lost (same as current behavior — RUNNING rows become orphans anyway)
+
+### Option B: Event Log with Reconstruction
+
+Send both phases as separate **events** (not records), and let the server reconstruct the current state.
+
+```
+Event 1: {type: "EXECUTION_STARTED", executionId: "abc", startTime: ..., routeId: ...}
+Event 2: {type: "EXECUTION_COMPLETED", executionId: "abc", duration: 250, outputBody: ..., processors: [...]}
+```
+
+**Server-side:**
+- Store raw events in an append-only log table
+- Reconstruct current state via `SELECT argMax(field, event_time) FROM events WHERE execution_id = ? GROUP BY execution_id`
+- Or: use a materialized view with `AggregatingMergeTree` + `argMaxState` to maintain a "latest state" table
+
+**Trade-offs:**
+- More complex server-side reconstruction
+- Higher storage (two rows per execution instead of one)
+- More flexible: supports any number of state transitions (RUNNING -> PAUSED -> RUNNING -> COMPLETED)
+- Natural fit for event sourcing architectures
+
+### Option C: Hybrid (Current Cameleer3-Server Approach)
+
+Keep the two-phase protocol but handle merging at the server application layer. This is what cameleer3-server implements today with the `ExecutionAccumulator`:
+
+- RUNNING POST -> hold in `ConcurrentHashMap` (no DB write)
+- COMPLETED POST -> merge with RUNNING in-memory -> single INSERT to DB
+- Timeout sweep -> flush stale RUNNING entries for visibility
+
+**Trade-offs:**
+- No agent changes required
+- Server must be stateful (in-memory accumulator)
+- Crash window: active executions lost if server restarts
+- Adds complexity to the server that wouldn't exist with Option A
+
+## Recommendation
+
+**Option A (single-phase reporting)** is the strongest choice for a new protocol version:
+
+1. **Simplest server implementation**: Pure append, no state, no merging
+2. **Works everywhere**: ClickHouse, Kafka, S3, any append-only store
+3. **Correct by construction**: MVs, aggregations, and stream processing all see one event per execution
+4. **Agent is the natural place to buffer**: The agent already tracks in-flight executions for instrumentation — it just needs to hold the report until completion
+5. **Minimal data loss risk**: Agent crash loses in-flight data regardless of protocol — this doesn't make it worse
+
+### Migration Strategy
+
+1. Add `protocol_version` field to agent registration
+2. v1 agents: server uses `ExecutionAccumulator` (current behavior)
+3. v2 agents: server does pure append (no accumulator needed for v2 data)
+4. Both can coexist — the server checks protocol version per agent
+
+### Fields for Single-Phase Record
+
+The complete record sent by a v2 agent:
+
+```json
+{
+  "executionId": "uuid",
+  "routeId": "myRoute",
+  "agentId": "agent-1",
+  "applicationName": "my-app",
+  "correlationId": "corr-123",
+  "exchangeId": "exchange-456",
+  "status": "COMPLETED",
+  "startTime": "2026-03-31T10:00:00.000Z",
+  "endTime": "2026-03-31T10:00:00.250Z",
+  "durationMs": 250,
+  "errorMessage": null,
+  "errorStackTrace": null,
+  "errorType": null,
+  "errorCategory": null,
+  "rootCauseType": null,
+  "rootCauseMessage": null,
+  "inputSnapshot": {"body": "...", "headers": {"Content-Type": "application/json"}},
+  "outputSnapshot": {"body": "...", "headers": {"Content-Type": "application/xml"}},
+  "attributes": {"key": "value"},
+  "traceId": "otel-trace-id",
+  "spanId": "otel-span-id",
+  "replayExchangeId": null,
+  "processors": [
+    {
+      "processorId": "proc-1",
+      "processorType": "to",
+      "status": "COMPLETED",
+      "startTime": "...",
+      "endTime": "...",
+      "durationMs": 120,
+      "inputBody": "...",
+      "outputBody": "...",
+      "children": []
+    }
+  ]
+}
+```
+
+All fields populated. No second POST needed. Server does a single INSERT.

docs/superpowers/specs/2026-03-31-clickhouse-migration-design.md

@@ -0,0 +1,916 @@
+# ClickHouse Migration Design
+
+Replace PostgreSQL/TimescaleDB + OpenSearch with ClickHouse OSS for all observability data.
+PostgreSQL retained only for RBAC, config, and audit log.
+
+## Context
+
+Cameleer3-server currently uses three storage systems:
+
+- **PostgreSQL/TimescaleDB**: executions, processor_executions, agent_metrics (hypertables), agent_events, route_diagrams, plus RBAC/config/audit tables. Continuous aggregates for dashboard statistics.
+- **OpenSearch**: executions-YYYY-MM-DD indices (full-text search on bodies/headers/errors), logs-YYYY-MM-DD indices (application log storage with 7-day retention).
+- **Dual-write pattern**: PG is source of truth, OpenSearch is async-indexed via debounced `SearchIndexer`.
+
+This architecture has scaling limits: three systems to operate, data duplication between PG and OpenSearch, TimescaleDB continuous aggregates with limited flexibility, and no multitenancy support.
+
+**Goal**: Consolidate to ClickHouse OSS (self-hosted) for all observability data. Add multitenancy with custom per-tenant, per-document-type retention. Support billions of documents, terabytes of data, sub-second wildcard search.
+
+## Decisions
+
+| Decision | Choice | Rationale |
+|----------|--------|-----------|
+| Deployment | Self-hosted ClickHouse OSS on k3s | All needed features available in OSS. Fits existing infra. |
+| Execution lifecycle | Approach B: Application-side accumulator | Merges RUNNING+COMPLETED in-memory, writes one row. Avoids upsert problem. |
+| Table engine (executions) | ReplacingMergeTree | Handles rare late corrections via version column. Normal flow writes once. |
+| Table engine (all others) | MergeTree | Append-only data, no dedup needed. |
+| Client | JDBC + JdbcTemplate | Familiar pattern, matches current PG code. Async inserts via JDBC URL settings. |
+| Multitenancy | Shared tables + tenant_id column | Row policies for defense-in-depth. Application-layer WHERE for primary enforcement. |
+| Retention | Application-driven scheduler | Per-tenant, per-document-type. Config in PG, execution via ALTER TABLE DELETE. |
+| Search | Ngram bloom filter indexes | Sub-second wildcard search. Materialized `_search_text` column for cross-field search. |
+| Highlighting | Application-side in Java | Extract 120-char fragment around match from returned fields. |
+| Storage tiering | Local SSD only (initially) | S3/MinIO tiering can be added later via TTL MOVE rules. |
+
+## ClickHouse OSS Constraints
+
+These are features NOT available in the open-source version:
+
+| Constraint | Impact on Cameleer3 |
+|------------|---------------------|
+| No SharedMergeTree | No elastic compute scaling; must size nodes up-front. Acceptable for self-hosted. |
+| No BM25 relevance scoring | Search returns matches without ranking. Acceptable for observability (want all matches, not ranked). |
+| No search highlighting | Replaced by application-side highlighting in Java. |
+| No fuzzy/typo-tolerant search | Must match exact tokens or use ngram index for substring match. Acceptable. |
+| No ClickPipes | Must build own ingestion pipeline. Already exists (agents push via HTTP POST). |
+| No managed backups | Must configure `clickhouse-backup` (Altinity, open-source) or built-in BACKUP SQL. |
+| No auto-scaling | Manual capacity planning. Single node handles 14+ TiB, sufficient for initial scale. |
+
+General ClickHouse constraints (apply to both OSS and Cloud):
+
+| Constraint | Mitigation |
+|------------|------------|
+| ORDER BY is immutable | Careful upfront schema design. Documented below. |
+| No transactions | Single-table INSERT atomic per block. No cross-table atomicity needed. |
+| Mutations are expensive | Avoid ALTER UPDATE/DELETE. Use ReplacingMergeTree for corrections, append-only for everything else. |
+| Row policies skip mutations | Application-layer WHERE on mutations. Mutations are rare (retention scheduler only). |
+| No JPA/Hibernate | Use JdbcTemplate (already the pattern for PG). |
+| JSON max_dynamic_paths | Store attributes as flattened String, not JSON type. Use ngram index for search. |
+| Text indexes can't index JSON subcolumns | Extract searchable text into materialized String columns. |
+| MVs only process new inserts | Historical data backfill writes through MV pipeline. |
+| MV errors block source inserts | Careful MV design. Test thoroughly before production. |
+| ReplacingMergeTree eventual consistency | Use FINAL on queries that need latest version. |
+
+## What Stays in PostgreSQL
+
+| Table | Reason |
+|-------|--------|
+| `users`, `roles`, `groups`, `user_groups`, `user_roles`, `group_roles` | RBAC with relational joins, foreign keys, transactions |
+| `server_config` | Global config, low volume, needs transactions |
+| `application_config` | Per-app observability settings |
+| `app_settings` | Per-app SLA thresholds |
+| `audit_log` | Security compliance, needs transactions, joins with RBAC tables |
+| OIDC config | Auth provider config |
+| `tenant_retention_config` (new) | Per-tenant retention settings, referenced by scheduler |
+
+## What Moves to ClickHouse
+
+| Data | Current Location | ClickHouse Table |
+|------|-----------------|------------------|
+| Route executions | PG `executions` hypertable + OpenSearch `executions-*` | `executions` |
+| Processor executions | PG `processor_executions` hypertable | `processor_executions` |
+| Agent metrics | PG `agent_metrics` hypertable | `agent_metrics` |
+| Agent events | PG `agent_events` | `agent_events` |
+| Route diagrams | PG `route_diagrams` | `route_diagrams` |
+| Application logs | OpenSearch `logs-*` | `logs` |
+| Dashboard statistics | PG continuous aggregates (`stats_1m_*`) | ClickHouse materialized views (`stats_1m_*`) |
+
+## Table Schemas
+
+### executions
+
+```sql
+CREATE TABLE executions (
+    tenant_id            LowCardinality(String),
+    execution_id         String,
+    start_time           DateTime64(3),
+    _version             UInt64 DEFAULT 1,
+    route_id             LowCardinality(String),
+    agent_id             LowCardinality(String),
+    application_name     LowCardinality(String),
+    status               LowCardinality(String),
+    correlation_id       String DEFAULT '',
+    exchange_id          String DEFAULT '',
+    end_time             Nullable(DateTime64(3)),
+    duration_ms          Nullable(Int64),
+    error_message        String DEFAULT '',
+    error_stacktrace     String DEFAULT '',
+    error_type           LowCardinality(String) DEFAULT '',
+    error_category       LowCardinality(String) DEFAULT '',
+    root_cause_type      String DEFAULT '',
+    root_cause_message   String DEFAULT '',
+    diagram_content_hash String DEFAULT '',
+    engine_level         LowCardinality(String) DEFAULT '',
+    input_body           String DEFAULT '',
+    output_body          String DEFAULT '',
+    input_headers        String DEFAULT '',
+    output_headers       String DEFAULT '',
+    attributes           String DEFAULT '',
+    trace_id             String DEFAULT '',
+    span_id              String DEFAULT '',
+    processors_json      String DEFAULT '',
+    has_trace_data       Bool DEFAULT false,
+    is_replay            Bool DEFAULT false,
+
+    _search_text         String MATERIALIZED
+        concat(error_message, ' ', error_stacktrace, ' ', attributes,
+               ' ', input_body, ' ', output_body, ' ', input_headers,
+               ' ', output_headers, ' ', root_cause_message),
+
+    INDEX idx_search   _search_text TYPE ngrambf_v1(3, 256, 2, 0) GRANULARITY 4,
+    INDEX idx_error    error_message TYPE ngrambf_v1(3, 256, 2, 0) GRANULARITY 4,
+    INDEX idx_bodies   concat(input_body, ' ', output_body) TYPE ngrambf_v1(3, 256, 2, 0) GRANULARITY 4,
+    INDEX idx_headers  concat(input_headers, ' ', output_headers) TYPE ngrambf_v1(3, 256, 2, 0) GRANULARITY 4,
+    INDEX idx_status   status TYPE set(10) GRANULARITY 1,
+    INDEX idx_corr     correlation_id TYPE bloom_filter(0.01) GRANULARITY 4
+)
+ENGINE = ReplacingMergeTree(_version)
+PARTITION BY (tenant_id, toYYYYMM(start_time))
+ORDER BY (tenant_id, start_time, application_name, route_id, execution_id)
+TTL start_time + INTERVAL 365 DAY DELETE
+SETTINGS index_granularity = 8192;
+```
+
+Design rationale:
+- **ORDER BY** `(tenant_id, start_time, application_name, route_id, execution_id)`: Matches UI query pattern (tenant -> time range -> app -> route). Time before application because observability queries almost always include a time range.
+- **PARTITION BY** `(tenant_id, toYYYYMM(start_time))`: Enables per-tenant partition drops for retention. Monthly granularity balances partition count vs drop efficiency.
+- **ReplacingMergeTree(_version)**: Normal flow writes once (version 1). Late corrections write version 2+. Background merges keep latest version.
+- **`_search_text` materialized column**: Computed at insert time. Concatenates all searchable fields for cross-field wildcard search.
+- **`ngrambf_v1(3, 256, 2, 0)`**: 3-char ngrams in a 256-byte bloom filter with 2 hash functions. Prunes most granules for `LIKE '%term%'` queries. The bloom filter size (256 bytes) is a starting point — increase to 4096-8192 if false positive rates are too high for long text fields. Tune after benchmarking with real data.
+- **`LowCardinality(String)`**: Dictionary encoding for columns with few distinct values. Major compression improvement.
+- **TTL 365 days**: Safety net. Application-driven scheduler handles per-tenant retention at finer granularity.
+
+### processor_executions
+
+```sql
+CREATE TABLE processor_executions (
+    tenant_id             LowCardinality(String),
+    execution_id          String,
+    processor_id          String,
+    start_time            DateTime64(3),
+    route_id              LowCardinality(String),
+    application_name      LowCardinality(String),
+    processor_type        LowCardinality(String),
+    parent_processor_id   String DEFAULT '',
+    depth                 UInt16 DEFAULT 0,
+    status                LowCardinality(String),
+    end_time              Nullable(DateTime64(3)),
+    duration_ms           Nullable(Int64),
+    error_message         String DEFAULT '',
+    error_stacktrace      String DEFAULT '',
+    error_type            LowCardinality(String) DEFAULT '',
+    error_category        LowCardinality(String) DEFAULT '',
+    root_cause_type       String DEFAULT '',
+    root_cause_message    String DEFAULT '',
+    input_body            String DEFAULT '',
+    output_body           String DEFAULT '',
+    input_headers         String DEFAULT '',
+    output_headers        String DEFAULT '',
+    attributes            String DEFAULT '',
+    loop_index            Nullable(Int32),
+    loop_size             Nullable(Int32),
+    split_index           Nullable(Int32),
+    split_size            Nullable(Int32),
+    multicast_index       Nullable(Int32),
+    resolved_endpoint_uri String DEFAULT '',
+    error_handler_type    LowCardinality(String) DEFAULT '',
+    circuit_breaker_state LowCardinality(String) DEFAULT '',
+    fallback_triggered    Bool DEFAULT false,
+
+    _search_text          String MATERIALIZED
+        concat(error_message, ' ', error_stacktrace, ' ', attributes,
+               ' ', input_body, ' ', output_body, ' ', input_headers, ' ', output_headers),
+
+    INDEX idx_search  _search_text TYPE ngrambf_v1(3, 256, 2, 0) GRANULARITY 4,
+    INDEX idx_exec_id execution_id TYPE bloom_filter(0.01) GRANULARITY 4
+)
+ENGINE = MergeTree()
+PARTITION BY (tenant_id, toYYYYMM(start_time))
+ORDER BY (tenant_id, start_time, application_name, route_id, execution_id, processor_id)
+TTL start_time + INTERVAL 365 DAY DELETE
+SETTINGS index_granularity = 8192;
+```
+
+### logs
+
+```sql
+CREATE TABLE logs (
+    tenant_id    LowCardinality(String),
+    timestamp    DateTime64(3),
+    application  LowCardinality(String),
+    agent_id     LowCardinality(String),
+    level        LowCardinality(String),
+    logger_name  LowCardinality(String) DEFAULT '',
+    message      String,
+    thread_name  LowCardinality(String) DEFAULT '',
+    stack_trace  String DEFAULT '',
+    exchange_id  String DEFAULT '',
+    mdc          Map(String, String) DEFAULT map(),
+
+    INDEX idx_msg   message TYPE ngrambf_v1(3, 256, 2, 0) GRANULARITY 4,
+    INDEX idx_stack stack_trace TYPE ngrambf_v1(3, 256, 2, 0) GRANULARITY 4,
+    INDEX idx_level level TYPE set(10) GRANULARITY 1
+)
+ENGINE = MergeTree()
+PARTITION BY (tenant_id, toYYYYMM(timestamp))
+ORDER BY (tenant_id, application, timestamp)
+TTL timestamp + INTERVAL 365 DAY DELETE
+SETTINGS index_granularity = 8192;
+```
+
+### agent_metrics
+
+```sql
+CREATE TABLE agent_metrics (
+    tenant_id          LowCardinality(String),
+    collected_at       DateTime64(3),
+    agent_id           LowCardinality(String),
+    metric_name        LowCardinality(String),
+    metric_value       Float64,
+    tags               Map(String, String) DEFAULT map(),
+    server_received_at DateTime64(3) DEFAULT now64(3)
+)
+ENGINE = MergeTree()
+PARTITION BY (tenant_id, toYYYYMM(collected_at))
+ORDER BY (tenant_id, agent_id, metric_name, collected_at)
+TTL collected_at + INTERVAL 365 DAY DELETE
+SETTINGS index_granularity = 8192;
+```
+
+### agent_events
+
+```sql
+CREATE TABLE agent_events (
+    tenant_id   LowCardinality(String),
+    timestamp   DateTime64(3) DEFAULT now64(3),
+    agent_id    LowCardinality(String),
+    app_id      LowCardinality(String),
+    event_type  LowCardinality(String),
+    detail      String DEFAULT ''
+)
+ENGINE = MergeTree()
+PARTITION BY (tenant_id, toYYYYMM(timestamp))
+ORDER BY (tenant_id, app_id, agent_id, timestamp)
+TTL timestamp + INTERVAL 365 DAY DELETE;
+```
+
+### route_diagrams
+
+```sql
+CREATE TABLE route_diagrams (
+    tenant_id        LowCardinality(String),
+    content_hash     String,
+    route_id         LowCardinality(String),
+    agent_id         LowCardinality(String),
+    application_name LowCardinality(String),
+    definition       String,
+    created_at       DateTime64(3) DEFAULT now64(3)
+)
+ENGINE = ReplacingMergeTree(created_at)
+ORDER BY (tenant_id, content_hash)
+SETTINGS index_granularity = 8192;
+```
+
+## Materialized Views (Stats)
+
+Replace TimescaleDB continuous aggregates. ClickHouse MVs trigger on INSERT and store aggregate states in target tables.
+
+### stats_1m_all (global)
+
+```sql
+CREATE TABLE stats_1m_all (
+    tenant_id    LowCardinality(String),
+    bucket       DateTime,
+    total_count  AggregateFunction(count, UInt64),
+    failed_count AggregateFunction(countIf, UInt64, UInt8),
+    running_count AggregateFunction(countIf, UInt64, UInt8),
+    duration_sum AggregateFunction(sum, Nullable(Int64)),
+    duration_max AggregateFunction(max, Nullable(Int64)),
+    p99_duration AggregateFunction(quantile(0.99), Nullable(Int64))
+)
+ENGINE = AggregatingMergeTree()
+PARTITION BY (tenant_id, toYYYYMM(bucket))
+ORDER BY (tenant_id, bucket)
+TTL bucket + INTERVAL 365 DAY DELETE;
+
+CREATE MATERIALIZED VIEW stats_1m_all_mv TO stats_1m_all AS
+SELECT
+    tenant_id,
+    toStartOfMinute(start_time) AS bucket,
+    countState()                        AS total_count,
+    countIfState(status = 'FAILED')     AS failed_count,
+    countIfState(status = 'RUNNING')    AS running_count,
+    sumState(duration_ms)               AS duration_sum,
+    maxState(duration_ms)               AS duration_max,
+    quantileState(0.99)(duration_ms)    AS p99_duration
+FROM executions
+GROUP BY tenant_id, bucket;
+```
+
+### stats_1m_app (per-application)
+
+```sql
+CREATE TABLE stats_1m_app (
+    tenant_id        LowCardinality(String),
+    application_name LowCardinality(String),
+    bucket           DateTime,
+    total_count      AggregateFunction(count, UInt64),
+    failed_count     AggregateFunction(countIf, UInt64, UInt8),
+    running_count    AggregateFunction(countIf, UInt64, UInt8),
+    duration_sum     AggregateFunction(sum, Nullable(Int64)),
+    duration_max     AggregateFunction(max, Nullable(Int64)),
+    p99_duration     AggregateFunction(quantile(0.99), Nullable(Int64))
+)
+ENGINE = AggregatingMergeTree()
+PARTITION BY (tenant_id, toYYYYMM(bucket))
+ORDER BY (tenant_id, application_name, bucket)
+TTL bucket + INTERVAL 365 DAY DELETE;
+
+CREATE MATERIALIZED VIEW stats_1m_app_mv TO stats_1m_app AS
+SELECT
+    tenant_id,
+    application_name,
+    toStartOfMinute(start_time) AS bucket,
+    countState()                        AS total_count,
+    countIfState(status = 'FAILED')     AS failed_count,
+    countIfState(status = 'RUNNING')    AS running_count,
+    sumState(duration_ms)               AS duration_sum,
+    maxState(duration_ms)               AS duration_max,
+    quantileState(0.99)(duration_ms)    AS p99_duration
+FROM executions
+GROUP BY tenant_id, application_name, bucket;
+```
+
+### stats_1m_route (per-route)
+
+```sql
+CREATE TABLE stats_1m_route (
+    tenant_id        LowCardinality(String),
+    application_name LowCardinality(String),
+    route_id         LowCardinality(String),
+    bucket           DateTime,
+    total_count      AggregateFunction(count, UInt64),
+    failed_count     AggregateFunction(countIf, UInt64, UInt8),
+    running_count    AggregateFunction(countIf, UInt64, UInt8),
+    duration_sum     AggregateFunction(sum, Nullable(Int64)),
+    duration_max     AggregateFunction(max, Nullable(Int64)),
+    p99_duration     AggregateFunction(quantile(0.99), Nullable(Int64))
+)
+ENGINE = AggregatingMergeTree()
+PARTITION BY (tenant_id, toYYYYMM(bucket))
+ORDER BY (tenant_id, application_name, route_id, bucket)
+TTL bucket + INTERVAL 365 DAY DELETE;
+
+CREATE MATERIALIZED VIEW stats_1m_route_mv TO stats_1m_route AS
+SELECT
+    tenant_id,
+    application_name,
+    route_id,
+    toStartOfMinute(start_time) AS bucket,
+    countState()                        AS total_count,
+    countIfState(status = 'FAILED')     AS failed_count,
+    countIfState(status = 'RUNNING')    AS running_count,
+    sumState(duration_ms)               AS duration_sum,
+    maxState(duration_ms)               AS duration_max,
+    quantileState(0.99)(duration_ms)    AS p99_duration
+FROM executions
+GROUP BY tenant_id, application_name, route_id, bucket;
+```
+
+### stats_1m_processor (per-processor-type)
+
+```sql
+CREATE TABLE stats_1m_processor (
+    tenant_id        LowCardinality(String),
+    application_name LowCardinality(String),
+    processor_type   LowCardinality(String),
+    bucket           DateTime,
+    total_count      AggregateFunction(count, UInt64),
+    failed_count     AggregateFunction(countIf, UInt64, UInt8),
+    duration_sum     AggregateFunction(sum, Nullable(Int64)),
+    duration_max     AggregateFunction(max, Nullable(Int64)),
+    p99_duration     AggregateFunction(quantile(0.99), Nullable(Int64))
+)
+ENGINE = AggregatingMergeTree()
+PARTITION BY (tenant_id, toYYYYMM(bucket))
+ORDER BY (tenant_id, application_name, processor_type, bucket)
+TTL bucket + INTERVAL 365 DAY DELETE;
+
+CREATE MATERIALIZED VIEW stats_1m_processor_mv TO stats_1m_processor AS
+SELECT
+    tenant_id,
+    application_name,
+    processor_type,
+    toStartOfMinute(start_time) AS bucket,
+    countState()                        AS total_count,
+    countIfState(status = 'FAILED')     AS failed_count,
+    sumState(duration_ms)               AS duration_sum,
+    maxState(duration_ms)               AS duration_max,
+    quantileState(0.99)(duration_ms)    AS p99_duration
+FROM processor_executions
+GROUP BY tenant_id, application_name, processor_type, bucket;
+```
+
+### stats_1m_processor_detail (per-processor-id)
+
+```sql
+CREATE TABLE stats_1m_processor_detail (
+    tenant_id        LowCardinality(String),
+    application_name LowCardinality(String),
+    route_id         LowCardinality(String),
+    processor_id     String,
+    bucket           DateTime,
+    total_count      AggregateFunction(count, UInt64),
+    failed_count     AggregateFunction(countIf, UInt64, UInt8),
+    duration_sum     AggregateFunction(sum, Nullable(Int64)),
+    duration_max     AggregateFunction(max, Nullable(Int64)),
+    p99_duration     AggregateFunction(quantile(0.99), Nullable(Int64))
+)
+ENGINE = AggregatingMergeTree()
+PARTITION BY (tenant_id, toYYYYMM(bucket))
+ORDER BY (tenant_id, application_name, route_id, processor_id, bucket)
+TTL bucket + INTERVAL 365 DAY DELETE;
+
+CREATE MATERIALIZED VIEW stats_1m_processor_detail_mv TO stats_1m_processor_detail AS
+SELECT
+    tenant_id,
+    application_name,
+    route_id,
+    processor_id,
+    toStartOfMinute(start_time) AS bucket,
+    countState()                        AS total_count,
+    countIfState(status = 'FAILED')     AS failed_count,
+    sumState(duration_ms)               AS duration_sum,
+    maxState(duration_ms)               AS duration_max,
+    quantileState(0.99)(duration_ms)    AS p99_duration
+FROM processor_executions
+GROUP BY tenant_id, application_name, route_id, processor_id, bucket;
+```
+
+## Ingestion Pipeline
+
+### Current Flow (replaced)
+
+```
+Agent POST -> IngestionService -> PostgresExecutionStore.upsert() -> PG
+                               -> SearchIndexer (debounced 2s) -> reads from PG -> OpenSearch
+```
+
+### New Flow
+
+```
+Agent POST -> IngestionService -> ExecutionAccumulator
+                                     |-- RUNNING: ConcurrentHashMap (no DB write)
+                                     |-- COMPLETED/FAILED: merge with pending -> WriteBuffer
+                                     '-- Timeout sweep (60s): flush stale -> WriteBuffer
+                                                                    |
+                                                    ClickHouseExecutionStore.insertBatch()
+                                                    ClickHouseProcessorStore.insertBatch()
+```
+
+### ExecutionAccumulator
+
+New component replacing `SearchIndexer`. Core responsibilities:
+
+1. **On RUNNING POST**: Store `PendingExecution` in `ConcurrentHashMap<String, PendingExecution>` keyed by `execution_id`. Return 200 OK immediately. No database write.
+
+2. **On COMPLETED/FAILED POST**: Look up pending RUNNING by `execution_id`. If found, merge fields using the same COALESCE logic currently in `PostgresExecutionStore.upsert()`. Produce a complete `MergedExecution` and push to `WriteBuffer`. If not found (race condition or RUNNING already flushed by timeout), write COMPLETED directly with `_version=2`.
+
+3. **Timeout sweep** (scheduled every 60s): Scan for RUNNING entries older than 5 minutes. Flush them to ClickHouse as-is with status=RUNNING, making them visible in the UI. When COMPLETED eventually arrives, it writes with `_version=2` (ReplacingMergeTree deduplicates).
+
+4. **Late corrections**: If a correction arrives for an already-written execution, insert with `_version` incremented. ReplacingMergeTree handles deduplication.
+
+### WriteBuffer
+
+Reuse the existing `WriteBuffer` pattern (bounded queue, configurable batch size, scheduled drain):
+
+- Buffer capacity: 50,000 items
+- Batch size: 5,000 per flush
+- Flush interval: 1 second
+- Separate buffers for executions and processor_executions (independent batch inserts)
+- Drain calls `ClickHouseExecutionStore.insertBatch()` using JDBC batch update
+
+### Logs Ingestion
+
+Direct batch INSERT, bypasses accumulator (logs are single-phase):
+
+```
+Agent POST /api/v1/data/logs -> LogIngestionController -> ClickHouseLogStore.insertBatch()
+```
+
+### Metrics Ingestion
+
+Existing `MetricsWriteBuffer` targets ClickHouse instead of PG:
+
+```
+Agent POST /api/v1/data/metrics -> MetricsController -> WriteBuffer -> ClickHouseMetricsStore.insertBatch()
+```
+
+### JDBC Batch Insert Pattern
+
+```java
+jdbcTemplate.batchUpdate(
+    "INSERT INTO executions (tenant_id, execution_id, start_time, ...) VALUES (?, ?, ?, ...)",
+    batchArgs
+);
+```
+
+JDBC URL includes `async_insert=1&wait_for_async_insert=0` for server-side buffering, preventing "too many parts" errors under high load.
+
+## Search Implementation
+
+### Query Translation
+
+Current OpenSearch bool queries map to ClickHouse SQL:
+
+```sql
+-- Full-text wildcard search with time range, status filter, and pagination
+SELECT *
+FROM executions FINAL
+WHERE tenant_id = {tenant_id:String}
+  AND start_time >= {time_from:DateTime64(3)}
+  AND start_time < {time_to:DateTime64(3)}
+  AND status IN ({statuses:Array(String)})
+  AND (
+    _search_text LIKE '%{search_term}%'
+    OR execution_id IN (
+      SELECT DISTINCT execution_id
+      FROM processor_executions
+      WHERE tenant_id = {tenant_id:String}
+        AND start_time >= {time_from:DateTime64(3)}
+        AND start_time < {time_to:DateTime64(3)}
+        AND _search_text LIKE '%{search_term}%'
+    )
+  )
+ORDER BY start_time DESC
+LIMIT {limit:UInt32} OFFSET {offset:UInt32}
+```
+
+### Scoped Searches
+
+| Scope | ClickHouse WHERE clause |
+|-------|------------------------|
+| textInBody | `input_body LIKE '%term%' OR output_body LIKE '%term%'` |
+| textInHeaders | `input_headers LIKE '%term%' OR output_headers LIKE '%term%'` |
+| textInErrors | `error_message LIKE '%term%' OR error_stacktrace LIKE '%term%'` |
+| global text | `_search_text LIKE '%term%'` (covers all fields) |
+
+All accelerated by `ngrambf_v1` indexes which prune 95%+ of data granules before scanning.
+
+### Application-Side Highlighting
+
+```java
+public String extractHighlight(String text, String searchTerm, int contextChars) {
+    int idx = text.toLowerCase().indexOf(searchTerm.toLowerCase());
+    if (idx < 0) return null;
+    int start = Math.max(0, idx - contextChars / 2);
+    int end = Math.min(text.length(), idx + searchTerm.length() + contextChars / 2);
+    return (start > 0 ? "..." : "")
+         + text.substring(start, end)
+         + (end < text.length() ? "..." : "");
+}
+```
+
+Returns the same `highlight` map structure the UI currently expects.
+
+### Nested Processor Search
+
+OpenSearch nested queries become a subquery on the `processor_executions` table:
+
+```sql
+execution_id IN (
+    SELECT DISTINCT execution_id
+    FROM processor_executions
+    WHERE tenant_id = ? AND start_time >= ? AND start_time < ?
+      AND _search_text LIKE '%term%'
+)
+```
+
+This is evaluated once with ngram index acceleration, then joined via IN.
+
+## Stats Query Translation
+
+### TimescaleDB -> ClickHouse Query Patterns
+
+| TimescaleDB | ClickHouse |
+|-------------|------------|
+| `time_bucket('1 minute', bucket)` | `toStartOfInterval(bucket, INTERVAL 1 MINUTE)` |
+| `SUM(total_count)` | `countMerge(total_count)` |
+| `SUM(failed_count)` | `countIfMerge(failed_count)` |
+| `approx_percentile(0.99, rollup(p99_duration))` | `quantileMerge(0.99)(p99_duration)` |
+| `SUM(duration_sum) / SUM(total_count)` | `sumMerge(duration_sum) / countMerge(total_count)` |
+| `MAX(duration_max)` | `maxMerge(duration_max)` |
+
+### Example: Timeseries Query
+
+```sql
+SELECT
+    toStartOfInterval(bucket, INTERVAL {interval:UInt32} SECOND) AS period,
+    countMerge(total_count)    AS total_count,
+    countIfMerge(failed_count) AS failed_count,
+    sumMerge(duration_sum) / countMerge(total_count) AS avg_duration,
+    quantileMerge(0.99)(p99_duration) AS p99_duration
+FROM stats_1m_app
+WHERE tenant_id = {tenant_id:String}
+  AND application_name = {app:String}
+  AND bucket >= {from:DateTime}
+  AND bucket < {to:DateTime}
+GROUP BY period
+ORDER BY period
+```
+
+### SLA and Top Errors
+
+SLA queries hit the raw `executions` table (need per-row duration filtering):
+
+```sql
+SELECT
+    countIf(duration_ms <= {threshold:Int64} AND status != 'RUNNING') * 100.0 / count() AS sla_pct
+FROM executions FINAL
+WHERE tenant_id = ? AND application_name = ? AND start_time >= ? AND start_time < ?
+```
+
+Top errors query:
+
+```sql
+SELECT
+    error_message,
+    count() AS error_count,
+    max(start_time) AS last_seen
+FROM executions FINAL
+WHERE tenant_id = ? AND status = 'FAILED'
+  AND start_time >= now() - INTERVAL 1 HOUR
+GROUP BY error_message
+ORDER BY error_count DESC
+LIMIT 10
+```
+
+## Multitenancy
+
+### Data Isolation
+
+**Primary**: Application-layer WHERE clause injection. Every ClickHouse query gets `WHERE tenant_id = ?` from the authenticated user's JWT claims.
+
+**Defense-in-depth**: ClickHouse row policies:
+
+```sql
+-- Create a ClickHouse user per tenant
+CREATE USER tenant_acme IDENTIFIED BY '...';
+
+-- Row policy ensures tenant can only see their data
+CREATE ROW POLICY tenant_acme_executions ON executions
+    FOR SELECT USING tenant_id = 'acme';
+
+-- Repeat for all tables
+```
+
+### Tenant ID in Schema
+
+`tenant_id` is the first column in every table's ORDER BY and PARTITION BY. This ensures:
+- Data for the same tenant is physically co-located on disk
+- Queries filtering by tenant_id use the sparse index efficiently
+- Partition drops for retention are scoped to individual tenants
+
+### Resource Quotas
+
+```sql
+CREATE SETTINGS PROFILE tenant_limits
+    SETTINGS max_execution_time = 30,
+             max_rows_to_read = 100000000,
+             max_memory_usage = '4G';
+
+ALTER USER tenant_acme SETTINGS PROFILE tenant_limits;
+```
+
+Prevents noisy neighbor problems where one tenant's expensive query affects others.
+
+## Retention
+
+### Strategy: Application-Driven Scheduler
+
+Per-tenant, per-document-type retention is too dynamic for static ClickHouse TTL rules. Instead:
+
+1. **Config table** in PostgreSQL:
+
+```sql
+CREATE TABLE tenant_retention_config (
+    tenant_id       VARCHAR(255) NOT NULL,
+    document_type   VARCHAR(50) NOT NULL,   -- executions, logs, metrics, etc.
+    retention_days  INT NOT NULL,
+    PRIMARY KEY (tenant_id, document_type)
+);
+```
+
+2. **RetentionScheduler** (Spring `@Scheduled`, runs daily at 03:00 UTC):
+
+```java
+@Scheduled(cron = "0 0 3 * * *")
+public void enforceRetention() {
+    List<TenantRetention> configs = retentionConfigRepo.findAll();
+    for (TenantRetention config : configs) {
+        String table = config.documentType();  // executions, logs, metrics, etc.
+        clickHouseJdbc.execute(
+            "ALTER TABLE " + table + " DELETE WHERE tenant_id = ? AND start_time < now() - INTERVAL ? DAY",
+            config.tenantId(), config.retentionDays()
+        );
+    }
+}
+```
+
+3. **Safety-net TTL**: Each table has a generous default TTL (365 days) as a backstop in case the scheduler fails. The scheduler handles the per-tenant granularity.
+
+4. **Partition-aligned drops**: Since `PARTITION BY (tenant_id, toYYYYMM(start_time))`, when all rows in a partition match the DELETE condition, ClickHouse drops the entire partition (fast, no rewrite). Enable `ttl_only_drop_parts=1` on tables.
+
+## Java/Spring Integration
+
+### Dependencies
+
+```xml
+<dependency>
+    <groupId>com.clickhouse</groupId>
+    <artifactId>clickhouse-jdbc</artifactId>
+    <version>0.7.x</version>  <!-- latest stable -->
+    <classifier>all</classifier>
+</dependency>
+```
+
+### Configuration
+
+```yaml
+clickhouse:
+  url: jdbc:clickhouse://clickhouse:8123/cameleer?async_insert=1&wait_for_async_insert=0
+  username: cameleer_app
+  password: ${CLICKHOUSE_PASSWORD}
+```
+
+### DataSource Bean
+
+```java
+@Configuration
+public class ClickHouseConfig {
+    @Bean
+    public DataSource clickHouseDataSource(ClickHouseProperties props) {
+        HikariDataSource ds = new HikariDataSource();
+        ds.setJdbcUrl(props.getUrl());
+        ds.setUsername(props.getUsername());
+        ds.setPassword(props.getPassword());
+        ds.setMaximumPoolSize(10);
+        return ds;
+    }
+
+    @Bean
+    public JdbcTemplate clickHouseJdbcTemplate(
+            @Qualifier("clickHouseDataSource") DataSource ds) {
+        return new JdbcTemplate(ds);
+    }
+}
+```
+
+### Interface Implementations
+
+Existing interfaces remain unchanged. New implementations:
+
+| Interface | Current Impl | New Impl |
+|-----------|-------------|----------|
+| `ExecutionStore` | `PostgresExecutionStore` | `ClickHouseExecutionStore` |
+| `SearchIndex` | `OpenSearchIndex` | `ClickHouseSearchIndex` |
+| `StatsStore` | `PostgresStatsStore` | `ClickHouseStatsStore` |
+| `DiagramStore` | `PostgresDiagramStore` | `ClickHouseDiagramStore` |
+| `MetricsStore` | `PostgresMetricsStore` | `ClickHouseMetricsStore` |
+| (log search) | `OpenSearchLogIndex` | `ClickHouseLogStore` |
+| (new) | `SearchIndexer` | `ExecutionAccumulator` |
+
+## Kubernetes Deployment
+
+### ClickHouse StatefulSet
+
+```yaml
+apiVersion: apps/v1
+kind: StatefulSet
+metadata:
+  name: clickhouse
+spec:
+  serviceName: clickhouse
+  replicas: 1  # single node initially
+  template:
+    spec:
+      containers:
+      - name: clickhouse
+        image: clickhouse/clickhouse-server:26.2
+        ports:
+        - containerPort: 8123  # HTTP
+        - containerPort: 9000  # Native
+        volumeMounts:
+        - name: data
+          mountPath: /var/lib/clickhouse
+        - name: config
+          mountPath: /etc/clickhouse-server/config.d
+        resources:
+          requests:
+            memory: "4Gi"
+            cpu: "2"
+          limits:
+            memory: "8Gi"
+            cpu: "4"
+  volumeClaimTemplates:
+  - metadata:
+      name: data
+    spec:
+      accessModes: ["ReadWriteOnce"]
+      resources:
+        requests:
+          storage: 100Gi  # NVMe/SSD
+```
+
+### Health Check
+
+```yaml
+livenessProbe:
+  httpGet:
+    path: /ping
+    port: 8123
+readinessProbe:
+  httpGet:
+    path: /ping
+    port: 8123
+```
+
+## Migration Path
+
+### Phase 1: Foundation
+
+- Add `clickhouse-jdbc` dependency
+- Create `ClickHouseConfig` (DataSource, JdbcTemplate)
+- Schema initialization (idempotent DDL scripts, not Flyway -- ClickHouse DDL is different enough)
+- Implement `ClickHouseMetricsStore` (simplest table, validates pipeline)
+- Deploy ClickHouse to k8s alongside existing PG+OpenSearch
+
+### Phase 2: Executions + Search
+
+- Build `ExecutionAccumulator` (replaces SearchIndexer)
+- Implement `ClickHouseExecutionStore` and `ClickHouseProcessorStore`
+- Implement `ClickHouseSearchIndex` (ngram-based SQL queries)
+- Feature flag: dual-write to both PG and CH, read from PG
+
+### Phase 3: Stats & Analytics
+
+- Create MV definitions (all 5 stats views)
+- Implement `ClickHouseStatsStore`
+- Validate stats accuracy: compare CH vs PG continuous aggregates
+
+### Phase 4: Remaining Tables
+
+- `ClickHouseDiagramStore` (ReplacingMergeTree)
+- `ClickHouseAgentEventStore`
+- `ClickHouseLogStore` (replaces OpenSearchLogIndex)
+- Application-side highlighting
+
+### Phase 5: Multitenancy
+
+- Tables already include `tenant_id` from Phase 1 (schema is forward-looking). This phase activates multitenancy.
+- Wire `tenant_id` from JWT claims into all ClickHouse queries (application-layer WHERE injection)
+- Add `tenant_id` to PostgreSQL RBAC/config tables
+- Create ClickHouse row policies per tenant (defense-in-depth)
+- Create `tenant_retention_config` table in PG and `RetentionScheduler` component
+- Tenant user management and resource quotas in ClickHouse
+
+### Phase 6: Cutover
+
+- Backfill historical data from PG/OpenSearch to ClickHouse
+- Switch read path to ClickHouse (feature flag)
+- Validate end-to-end
+- Remove OpenSearch dependency (POM, config, k8s manifests)
+- Remove TimescaleDB extensions and hypertable-specific code
+- Keep PostgreSQL for RBAC/config/audit only
+
+## Verification
+
+### Functional Verification
+
+1. **Ingestion**: Send executions via agent, verify they appear in ClickHouse with correct fields
+2. **Two-phase lifecycle**: Send RUNNING, then COMPLETED. Verify single merged row in CH
+3. **Search**: Wildcard search across bodies, headers, errors. Verify sub-second response
+4. **Stats**: Dashboard statistics match expected values. Compare with PG aggregates during dual-write
+5. **Logs**: Ingest log batches, query by app/level/time/text. Verify correctness
+6. **Retention**: Configure per-tenant retention, run scheduler, verify expired data is deleted
+7. **Multitenancy**: Two tenants, verify data isolation (one tenant cannot see another's data)
+
+### Performance Verification
+
+1. **Insert throughput**: 5K executions/batch at 1 flush/sec sustained
+2. **Search latency**: Sub-second for `LIKE '%term%'` across 1M+ rows
+3. **Stats query latency**: Dashboard stats in <100ms from materialized views
+4. **Log search**: <1s for text search across 7 days of logs
+
+### Data Integrity
+
+1. During dual-write phase: compare row counts between PG and CH
+2. After cutover: spot-check execution details, processor trees, search results