diff --git a/server/tinybird/copies/events_backfill.pipe b/server/tinybird/copies/events_backfill.pipe
new file mode 100644
index 000000000..7d5221b37
--- /dev/null
+++ b/server/tinybird/copies/events_backfill.pipe
@@ -0,0 +1,20 @@
+DESCRIPTION >
+    Chunked backfill of events from Postgres
+
+NODE migrate
+SQL >
+    %
+    SELECT *
+    FROM postgresql(
+        'us-west-3.pg.psdb.cloud:5432',
+        'postgres',
+        'events',
+        {{tb_secret('PG_USERNAME')}},
+        {{tb_secret('PG_PASSWORD')}},
+        'public'
+    )
+    WHERE timestamp > {{DateTime(start_date, '2025-01-01 00:00:00')}}
+      AND timestamp <= {{DateTime(end_date, '2025-01-01 00:00:01')}}
+
+TYPE COPY
+TARGET_DATASOURCE events
diff --git a/server/tinybird/pipes/aggregate.pipe b/server/tinybird/pipes/aggregate.pipe
new file mode 100644
index 000000000..c7a4b24c9
--- /dev/null
+++ b/server/tinybird/pipes/aggregate.pipe
@@ -0,0 +1,66 @@
+DESCRIPTION >
+    Aggregates events into time-bucketed timeseries data.
+    Returns unpivoted data: (period, event_name, group_value, total_value)
+    When group_by is provided, extracts that property from JSON and groups by it.
+
+TOKEN "aggregate_read" READ
+
+NODE filter_events
+DESCRIPTION >
+    Filter events by org, env, customer (optional), event names, and date range.
+
+SQL >
+    %
+    SELECT
+        timestamp,
+        event_name,
+        customer_id,
+        coalesce(value, 1) as value,
+        properties
+    FROM events
+    WHERE
+        org_id = {{ String(org_id, '') }}
+        AND env = {{ String(env, 'test') }}
+        AND event_name IN {{ Array(event_names, 'String', default='[]') }}
+        AND timestamp >= toDateTime({{ String(start_date, '2024-01-01 00:00:00') }})
+        AND timestamp <= toDateTime({{ String(end_date, '2024-12-31 23:59:59') }})
+        {% if defined(customer_id) %}
+        AND customer_id = {{ String(customer_id) }}
+        {% end %}
+
+NODE aggregate_by_period
+TYPE endpoint
+DESCRIPTION >
+    Aggregate events by time period and event name.
+    Optionally groups by a property extracted from the properties JSON.
+    Gap-filling is handled in the application layer.
+
+SQL >
+    %
+    SELECT
+        {% if String(bin_size, 'day') == 'hour' %}
+        toStartOfHour(timestamp, {{ String(timezone, 'UTC') }}) as period,
+        {% elif String(bin_size, 'day') == 'month' %}
+        toStartOfMonth(timestamp, {{ String(timezone, 'UTC') }}) as period,
+        {% else %}
+        toStartOfDay(timestamp, {{ String(timezone, 'UTC') }}) as period,
+        {% end %}
+        event_name,
+        {% if defined(group_by) %}
+        coalesce(
+            nullIf(JSONExtractString(assumeNotNull(properties), {{ String(group_by) }}), ''),
+            'unknown'
+        ) as group_value,
+        {% else %}
+        '' as group_value,
+        {% end %}
+        sum(value) as total_value
+    FROM filter_events
+    {% if defined(group_by) %}
+    WHERE properties IS NOT NULL AND properties != ''
+    GROUP BY period, event_name, group_value
+    ORDER BY period, event_name, group_value
+    {% else %}
+    GROUP BY period, event_name, group_value
+    ORDER BY period, event_name
+    {% end %}
diff --git a/server/tinybird/pipes/aggregate_groupable.pipe b/server/tinybird/pipes/aggregate_groupable.pipe
new file mode 100644
index 000000000..79664bd18
--- /dev/null
+++ b/server/tinybird/pipes/aggregate_groupable.pipe
@@ -0,0 +1,65 @@
+DESCRIPTION >
+    Aggregate queries with grouping by a property key or customer_id.
+    Uses PER-BIN TOP 9 groups + "AUTUMN_RESERVED" bucket (10 max total per bin).
+    Each time bin shows its own top 9 groups, not global top 9.
+    Returns unpivoted data: (period, event_name, group_value, total_value, _truncated)
+    _truncated is true only if at least one bin has more than 9 unique values.
+    Frontend maps "AUTUMN_RESERVED" to "Other values" for display.
+    Parameters:
+      - group_column: 'customer_id' to group by customer, 'property' (default) to group by property_key
+
+TOKEN "aggregate_groupable_read" READ
+
+NODE base
+DESCRIPTION >
+    Aggregate raw data by period, event_name, and group_value.
+SQL >
+    %
+    SELECT
+        {% if String(bin_size, 'day') == 'hour' %}
+        formatDateTime(hour, '%F %T') as period,
+        {% elif String(bin_size, 'day') == 'month' %}
+        formatDateTime(toStartOfMonth(hour, {{ String(timezone, 'UTC') }}), '%F %T') as period,
+        {% else %}
+        formatDateTime(toStartOfDay(hour, {{ String(timezone, 'UTC') }}), '%F %T') as period,
+        {% end %}
+        event_name,
+        {% if String(group_column, 'property') == 'customer_id' %}
+        customer_id as group_value,
+        {% else %}
+        {{ column('properties.' + String(property_key, '')) }}::String as group_value,
+        {% end %}
+        sum(total_value) as total_value
+    FROM events_hourly_mv
+    WHERE
+        org_id = {{ String(org_id, '') }}
+        AND env = {{ String(env, 'test') }}
+        AND event_name IN {{ Array(event_names, 'String', default='[]') }}
+        AND hour >= toDateTime({{ String(start_date, '2024-01-01 00:00:00') }})
+        AND hour <= toDateTime({{ String(end_date, '2024-12-31 23:59:59') }})
+        {% if defined(customer_id) and String(customer_id, '') != '' %}
+        AND customer_id = {{ String(customer_id) }}
+        {% end %}
+    GROUP BY period, event_name, group_value
+
+NODE ranked
+DESCRIPTION >
+    Rank groups within each bin by total_value descending.
+SQL >
+    SELECT
+        *,
+        row_number() OVER (PARTITION BY period, event_name ORDER BY total_value DESC) as rn
+    FROM base
+
+NODE endpoint
+TYPE endpoint
+SQL >
+    SELECT
+        period,
+        event_name,
+        if(rn <= 9, group_value, 'AUTUMN_RESERVED') as group_value,
+        sum(total_value) as total_value,
+        max(rn) > 9 as _truncated
+    FROM ranked
+    GROUP BY period, event_name, group_value
+    ORDER BY period, event_name, group_value
diff --git a/server/tinybird/pipes/aggregate_simple.pipe b/server/tinybird/pipes/aggregate_simple.pipe
new file mode 100644
index 000000000..96edabca2
--- /dev/null
+++ b/server/tinybird/pipes/aggregate_simple.pipe
@@ -0,0 +1,36 @@
+DESCRIPTION >
+    Simple timeseries aggregation pipe with NO grouping and NO limiting.
+    Fastest possible query for basic event aggregation.
+
+TOKEN "aggregate_simple_read" READ
+
+NODE endpoint
+TYPE endpoint
+SQL >
+    %
+    SELECT
+        {% if String(bin_size, 'day') == 'hour' %}
+        formatDateTime(hour, '%F %T') as period,
+        {% elif String(bin_size, 'day') == 'month' %}
+        formatDateTime(toStartOfMonth(hour, {{ String(timezone, 'UTC') }}), '%F %T') as period,
+        {% else %}
+        formatDateTime(toStartOfDay(hour, {{ String(timezone, 'UTC') }}), '%F %T') as period,
+        {% end %}
+        event_name,
+        sum(total_value) as total_value
+    FROM events_hourly_mv
+    WHERE
+        org_id = {{ String(org_id, '') }}
+        AND env = {{ String(env, 'test') }}
+        AND event_name IN {{ Array(event_names, 'String', default='[]') }}
+        AND hour >= toDateTime({{ String(start_date, '2024-01-01 00:00:00') }})
+        AND hour <= toDateTime({{ String(end_date, '2024-12-31 23:59:59') }})
+        {% if defined(customer_id) and String(customer_id, '') != '' %}
+        AND customer_id = {{ String(customer_id) }}
+        {% end %}
+    GROUP BY
+        period,
+        event_name
+    ORDER BY
+        period,
+        event_name
diff --git a/server/tinybird/pipes/list_events.pipe b/server/tinybird/pipes/list_events.pipe
new file mode 100644
index 000000000..dc1903b43
--- /dev/null
+++ b/server/tinybird/pipes/list_events.pipe
@@ -0,0 +1,40 @@
+DESCRIPTION >
+    Lists raw events with filtering by org, env, customer, and date range.
+    Optimized for the raw logs viewer UI. Supports pagination via cursor.
+    Queries events_by_timestamp_mv which is sorted by (org_id, env, timestamp) for fast time-range queries.
+
+TOKEN "list_events_read" READ
+
+NODE endpoint
+TYPE endpoint
+SQL >
+    %
+    SELECT
+        id,
+        org_id,
+        env,
+        customer_id,
+        event_name,
+        timestamp,
+        value,
+        properties,
+        idempotency_key,
+        entity_id
+    FROM events_by_timestamp_mv
+    WHERE
+        org_id = {{ String(org_id, '') }}
+        AND env = {{ String(env, 'test') }}
+        AND timestamp >= toDateTime64({{ String(start_date, '2024-01-01 00:00:00') }}, 6)
+        AND timestamp <= toDateTime64({{ String(end_date, '2024-12-31 23:59:59') }}, 6)
+        {% if defined(customer_id) %}
+        AND customer_id = {{ String(customer_id) }}
+        {% end %}
+        {% if defined(event_name) %}
+        AND event_name = {{ String(event_name) }}
+        {% end %}
+        {% if defined(cursor_timestamp) and defined(cursor_id) %}
+        AND (timestamp < toDateTime64({{ String(cursor_timestamp) }}, 6)
+             OR (timestamp = toDateTime64({{ String(cursor_timestamp) }}, 6) AND id < {{ String(cursor_id) }}))
+        {% end %}
+    ORDER BY timestamp DESC, id DESC
+    LIMIT {{ Int32(limit, 1000) }}
diff --git a/server/tinybird/scripts/backfill_events.sh b/server/tinybird/scripts/backfill_events.sh
new file mode 100755
index 000000000..c7b0a42de
--- /dev/null
+++ b/server/tinybird/scripts/backfill_events.sh
@@ -0,0 +1,235 @@
+#!/bin/bash
+# Backfill events from Postgres to Tinybird
+# Usage: ./backfill_events.sh [--start-date "YYYY-MM-DD HH:MM:SS"]
+#
+# Developer plan limit: 30s execution time per Copy Pipe
+# Chunks are generated dynamically based on boundaries below.
+#
+# Examples:
+#   ./backfill_events.sh                                    # Start from beginning (will prompt to truncate)
+#   ./backfill_events.sh --start-date "2025-12-12 00:00:00" # Resume from Dec 12 midnight (skips truncate)
+
+set -e  # Exit on first failure
+
+cd "$(dirname "$0")/.."  # Navigate to server/tinybird
+
+# Parse arguments
+START_DATE_ARG=""
+while [[ $# -gt 0 ]]; do
+    case $1 in
+        --start-date)
+            START_DATE_ARG="$2"
+            shift 2
+            ;;
+        *)
+            echo "Unknown option: $1"
+            echo "Usage: ./backfill_events.sh [--start-date \"YYYY-MM-DD HH:MM:SS\"]"
+            exit 1
+            ;;
+    esac
+done
+
+echo "=== Events Backfill Script ==="
+echo ""
+
+# Only truncate if no start date specified
+if [[ -z "$START_DATE_ARG" ]]; then
+    read -p "This will TRUNCATE the events datasource first. Continue? (y/n) " -n 1 -r
+    echo ""
+    if [[ ! $REPLY =~ ^[Yy]$ ]]; then
+        echo "Aborted."
+        exit 1
+    fi
+
+    echo ""
+    echo "Truncating events datasource..."
+    tb --cloud datasource truncate events --yes
+    echo "Truncated."
+    echo ""
+else
+    echo "Resuming from $START_DATE_ARG (skipping truncate)"
+    echo ""
+fi
+
+# Boundaries define when chunk size changes
+# Format: "boundary_date,hours_per_chunk"
+# From each boundary until the next, use that chunk size
+BOUNDARIES=(
+    "2026-01-21 00:00:00,24"   # Jan 21: 1-day (24 hour) chunks - last 7 days
+)
+END_DATE="2026-01-28 00:00:00"
+
+# Function to add hours to a date (macOS compatible)
+add_hours() {
+    local dt="$1"
+    local hours="$2"
+    date -j -v+"${hours}H" -f "%Y-%m-%d %H:%M:%S" "$dt" "+%Y-%m-%d %H:%M:%S"
+}
+
+# Function to compare dates (returns 0 if d1 < d2)
+date_lt() {
+    local d1="$1"
+    local d2="$2"
+    local ts1=$(date -j -f "%Y-%m-%d %H:%M:%S" "$d1" "+%s")
+    local ts2=$(date -j -f "%Y-%m-%d %H:%M:%S" "$d2" "+%s")
+    [[ $ts1 -lt $ts2 ]]
+}
+
+# Function to compare dates (returns 0 if d1 <= d2)
+date_le() {
+    local d1="$1"
+    local d2="$2"
+    local ts1=$(date -j -f "%Y-%m-%d %H:%M:%S" "$d1" "+%s")
+    local ts2=$(date -j -f "%Y-%m-%d %H:%M:%S" "$d2" "+%s")
+    [[ $ts1 -le $ts2 ]]
+}
+
+# Function to compare dates (returns 0 if d1 >= d2)
+date_ge() {
+    local d1="$1"
+    local d2="$2"
+    local ts1=$(date -j -f "%Y-%m-%d %H:%M:%S" "$d1" "+%s")
+    local ts2=$(date -j -f "%Y-%m-%d %H:%M:%S" "$d2" "+%s")
+    [[ $ts1 -ge $ts2 ]]
+}
+
+# Get chunk size for a given date
+get_chunk_hours() {
+    local dt="$1"
+    local chunk_hours=72  # default
+    
+    for boundary in "${BOUNDARIES[@]}"; do
+        IFS=',' read -r boundary_date hours <<< "$boundary"
+        if date_ge "$dt" "$boundary_date"; then
+            chunk_hours=$hours
+        fi
+    done
+    
+    echo $chunk_hours
+}
+
+# Wait for any running copy jobs to complete before starting a new one
+wait_for_copy_jobs() {
+    local max_attempts=60  # 5 minutes max wait (60 * 5s)
+    local attempt=0
+    
+    while [[ $attempt -lt $max_attempts ]]; do
+        # Check for waiting/working copy jobs using the status filter
+        local waiting_jobs
+        local working_jobs
+        waiting_jobs=$(tb --cloud job ls --status waiting --kind copy 2>/dev/null | grep -c "^id:" 2>/dev/null) || waiting_jobs=0
+        working_jobs=$(tb --cloud job ls --status working --kind copy 2>/dev/null | grep -c "^id:" 2>/dev/null) || working_jobs=0
+        local total_active=$((waiting_jobs + working_jobs))
+        
+        if [[ "$total_active" -eq 0 ]]; then
+            return 0
+        fi
+        
+        if [[ $attempt -eq 0 ]]; then
+            echo "  Waiting for $total_active existing copy job(s) to complete..."
+        fi
+        
+        sleep 5
+        attempt=$((attempt + 1))
+    done
+    
+    echo "Error: Timed out waiting for existing copy jobs to complete"
+    exit 1
+}
+
+# Generate all chunks
+echo "Generating chunks..."
+CHUNKS=()
+CURRENT="${BOUNDARIES[0]%%,*}"  # Start from first boundary
+
+while date_lt "$CURRENT" "$END_DATE"; do
+    CHUNK_HOURS=$(get_chunk_hours "$CURRENT")
+    NEXT=$(add_hours "$CURRENT" "$CHUNK_HOURS")
+    
+    # Don't go past end date
+    if date_lt "$END_DATE" "$NEXT"; then
+        NEXT="$END_DATE"
+    fi
+    
+    CHUNKS+=("$CURRENT,$NEXT")
+    CURRENT="$NEXT"
+done
+
+TOTAL_CHUNKS=${#CHUNKS[@]}
+echo "Generated $TOTAL_CHUNKS chunks"
+echo ""
+
+# Find starting chunk based on --start-date
+START_CHUNK=1
+if [[ -n "$START_DATE_ARG" ]]; then
+    FOUND=false
+    for i in "${!CHUNKS[@]}"; do
+        IFS=',' read -r CHUNK_START CHUNK_END <<< "${CHUNKS[$i]}"
+        if [[ "$CHUNK_START" == "$START_DATE_ARG" ]]; then
+            START_CHUNK=$((i + 1))
+            FOUND=true
+            break
+        fi
+    done
+    if [[ "$FOUND" == false ]]; then
+        echo "Error: No chunk starts on '$START_DATE_ARG'"
+        echo ""
+        echo "Hint: Chunks start at these times based on boundaries:"
+        for boundary in "${BOUNDARIES[@]}"; do
+            IFS=',' read -r boundary_date hours <<< "$boundary"
+            echo "  From $boundary_date: ${hours}-hour chunks"
+        done
+        exit 1
+    fi
+fi
+
+echo "Starting backfill: chunks $START_CHUNK-$TOTAL_CHUNKS (of $TOTAL_CHUNKS total)"
+echo ""
+
+for i in "${!CHUNKS[@]}"; do
+    CHUNK_NUM=$((i + 1))
+    
+    # Skip chunks before start chunk
+    if [[ $CHUNK_NUM -lt $START_CHUNK ]]; then
+        continue
+    fi
+    
+    IFS=',' read -r CHUNK_START CHUNK_END <<< "${CHUNKS[$i]}"
+    
+    echo "=== Chunk $CHUNK_NUM/$TOTAL_CHUNKS: $CHUNK_START -> $CHUNK_END ==="
+    
+    # Wait for any existing copy jobs to complete first
+    wait_for_copy_jobs
+    
+    START_TIME=$(date +%s)
+    
+    tb --cloud copy run events_backfill \
+        --param start_date="$CHUNK_START" \
+        --param end_date="$CHUNK_END" \
+        --wait
+    
+    END_TIME=$(date +%s)
+    DURATION=$((END_TIME - START_TIME))
+    
+    # Get current row count
+    ROW_COUNT=$(tb --cloud sql "SELECT count() FROM events" --format csv 2>/dev/null | tail -1)
+    
+    echo ""
+    echo "✓ Chunk $CHUNK_NUM COMPLETE ($CHUNK_START -> $CHUNK_END) in ${DURATION}s. Total rows: $ROW_COUNT"
+    echo ""
+    
+    # Show next chunk info
+    if [[ $CHUNK_NUM -lt $TOTAL_CHUNKS ]]; then
+        NEXT_IDX=$CHUNK_NUM
+        IFS=',' read -r NEXT_START NEXT_END <<< "${CHUNKS[$NEXT_IDX]}"
+        echo "  To resume: ./scripts/backfill_events.sh --start-date '$NEXT_START'"
+        echo "  Waiting 3s..."
+        sleep 3
+    fi
+    echo ""
+done
+
+echo "=========================================="
+echo "=== Backfill Complete ==="
+FINAL_COUNT=$(tb --cloud sql "SELECT count() FROM events" --format csv 2>/dev/null | tail -1)
+echo "Final row count: $FINAL_COUNT"
diff --git a/server/tinybird/scripts/backfill_events.ts b/server/tinybird/scripts/backfill_events.ts
new file mode 100644
index 000000000..c6af712ad
--- /dev/null
+++ b/server/tinybird/scripts/backfill_events.ts
@@ -0,0 +1,405 @@
+#!/usr/bin/env bun
+
+/**
+ * Idempotent backfill script for events from Postgres to Tinybird
+ * 
+ * Features:
+ * - Auto-detects date range (first event in Postgres to today 00:00)
+ * - Resumes from where it left off (no truncation needed)
+ * - Configurable chunk size to avoid timeouts
+ * - Retries failed chunks
+ * - Optional truncation with --truncate flag
+ * 
+ * Usage:
+ *   npx tsx scripts/backfill_events.ts
+ *   npx tsx scripts/backfill_events.ts --chunk-hours 12
+ *   npx tsx scripts/backfill_events.ts --start-date "2025-10-01 00:00:00"
+ *   npx tsx scripts/backfill_events.ts --end-date "2026-01-28 00:00:00"
+ *   npx tsx scripts/backfill_events.ts --truncate --start-date "2025-01-30 00:00:00"
+ */
+
+import { execSync } from "child_process";
+
+// Configuration
+const DEFAULT_CHUNK_HOURS = 24; // ~1 month (30 * 24)
+const MAX_RETRIES = 3;
+const RETRY_DELAY_MS = 5000;
+const DELAY_BETWEEN_CHUNKS_MS = 3000;
+
+interface Args {
+  chunkHours: number;
+  startDate?: string;
+  endDate?: string;
+  dryRun: boolean;
+  truncate: boolean;
+}
+
+function parseArgs(): Args {
+  const args: Args = {
+    chunkHours: DEFAULT_CHUNK_HOURS,
+    dryRun: false,
+    truncate: false,
+  };
+
+  for (let i = 2; i < process.argv.length; i++) {
+    const arg = process.argv[i];
+    if (arg === "--chunk-hours" && process.argv[i + 1]) {
+      args.chunkHours = parseInt(process.argv[++i], 10);
+    } else if (arg === "--start-date" && process.argv[i + 1]) {
+      args.startDate = process.argv[++i];
+    } else if (arg === "--end-date" && process.argv[i + 1]) {
+      args.endDate = process.argv[++i];
+    } else if (arg === "--dry-run") {
+      args.dryRun = true;
+    } else if (arg === "--truncate") {
+      args.truncate = true;
+    } else if (arg === "--help" || arg === "-h") {
+      console.log(`
+Usage: npx tsx scripts/backfill_events.ts [options]
+
+Options:
+  --chunk-hours <n>     Hours per chunk (default: ${DEFAULT_CHUNK_HOURS})
+  --start-date <date>   Override start date (default: auto-detect from Tinybird)
+  --end-date <date>     Override end date (default: today 00:00)
+  --dry-run             Show what would be done without executing
+  --truncate            Truncate the events datasource before backfilling (requires --start-date)
+  --help, -h            Show this help message
+
+Examples:
+  npx tsx scripts/backfill_events.ts
+  npx tsx scripts/backfill_events.ts --chunk-hours 12
+  npx tsx scripts/backfill_events.ts --start-date "2025-10-01 00:00:00"
+  npx tsx scripts/backfill_events.ts --truncate --start-date "2025-01-30 00:00:00"
+`);
+      process.exit(0);
+    }
+  }
+
+  return args;
+}
+
+function exec(cmd: string, silent = false): string {
+  try {
+    const result = execSync(cmd, { encoding: "utf-8", stdio: silent ? "pipe" : "inherit" });
+    return result?.trim() ?? "";
+  } catch (error: any) {
+    if (silent) {
+      return error.stdout?.trim() ?? "";
+    }
+    throw error;
+  }
+}
+
+function execCapture(cmd: string): string {
+  try {
+    return execSync(cmd, { encoding: "utf-8", stderr: "pipe" }).trim();
+  } catch (error: any) {
+    return error.stdout?.trim() ?? "";
+  }
+}
+
+function tbSql(query: string): string {
+  const escaped = query.replace(/"/g, '\\"');
+  try {
+    const result = execSync(`tb --cloud sql "${escaped}"`, { 
+      encoding: "utf-8", 
+      stdio: ["pipe", "pipe", "pipe"] 
+    });
+    return result.trim();
+  } catch (error: any) {
+    return error.stdout?.trim() ?? "";
+  }
+}
+
+function getLatestEventInTinybird(): string | null {
+  console.log("Checking latest event in Tinybird...");
+  const result = tbSql("SELECT max(timestamp) FROM events");
+  // Table format - look for timestamp pattern YYYY-MM-DD HH:MM:SS
+  const lines = result.split("\n");
+  for (const line of lines) {
+    const cleaned = line.trim();
+    // Match timestamp format: 2026-01-28 12:34:56 or 2026-01-28 12:34:56.000000
+    if (/^\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}:\d{2}/.test(cleaned)) {
+      if (cleaned === "1970-01-01 00:00:00.000000" || cleaned === "1970-01-01 00:00:00") {
+        return null;
+      }
+      return cleaned;
+    }
+  }
+  return null;
+}
+
+function getEventCount(): number {
+  const result = tbSql("SELECT count() FROM events");
+  // Table format output:
+  // Running against Tinybird Cloud: Workspace autumn_us_west_dev
+  //   count()
+  //    UInt64
+  // ───────────
+  //   3045881
+  const lines = result.split("\n");
+  for (const line of lines) {
+    const cleaned = line.trim();
+    // Look for a line that's just a number
+    const num = parseInt(cleaned, 10);
+    if (!isNaN(num) && String(num) === cleaned) {
+      return num;
+    }
+  }
+  return 0;
+}
+
+function truncateEvents(): void {
+  console.log("Truncating events datasource...");
+  exec("tb --cloud datasource truncate events --yes", false);
+  console.log("Truncated.\n");
+}
+
+function promptConfirmation(message: string): boolean {
+  const readline = require("readline");
+  const rl = readline.createInterface({
+    input: process.stdin,
+    output: process.stdout,
+  });
+
+  return new Promise<boolean>((resolve) => {
+    rl.question(`${message} (y/n) `, (answer: string) => {
+      rl.close();
+      resolve(answer.toLowerCase() === "y" || answer.toLowerCase() === "yes");
+    });
+  }) as unknown as boolean;
+}
+
+async function promptConfirmationAsync(message: string): Promise<boolean> {
+  const readline = require("readline");
+  const rl = readline.createInterface({
+    input: process.stdin,
+    output: process.stdout,
+  });
+
+  return new Promise<boolean>((resolve) => {
+    rl.question(`${message} (y/n) `, (answer: string) => {
+      rl.close();
+      resolve(answer.toLowerCase() === "y" || answer.toLowerCase() === "yes");
+    });
+  });
+}
+
+function getTodayMidnight(): string {
+  const now = new Date();
+  const midnight = new Date(now.getFullYear(), now.getMonth(), now.getDate());
+  return formatDate(midnight);
+}
+
+function formatDate(date: Date): string {
+  return date.toISOString().replace("T", " ").replace("Z", "").split(".")[0];
+}
+
+function parseDate(dateStr: string): Date {
+  // Handle format: "2026-01-21 00:00:00" or "2026-01-21 00:00:00.000000"
+  const cleaned = dateStr.split(".")[0].replace(" ", "T") + "Z";
+  return new Date(cleaned);
+}
+
+function addHours(dateStr: string, hours: number): string {
+  const date = parseDate(dateStr);
+  date.setTime(date.getTime() + hours * 60 * 60 * 1000);
+  return formatDate(date);
+}
+
+function floorToHour(dateStr: string): string {
+  const date = parseDate(dateStr);
+  date.setMinutes(0, 0, 0);
+  return formatDate(date);
+}
+
+async function sleep(ms: number): Promise<void> {
+  return new Promise(resolve => setTimeout(resolve, ms));
+}
+
+async function waitForCopyJobs(): Promise<void> {
+  const maxAttempts = 60;
+  let attempt = 0;
+
+  while (attempt < maxAttempts) {
+    const waitingJobs = execCapture("tb --cloud job ls --status waiting --kind copy 2>/dev/null | grep -c '^id:' || echo 0");
+    const workingJobs = execCapture("tb --cloud job ls --status working --kind copy 2>/dev/null | grep -c '^id:' || echo 0");
+    const total = parseInt(waitingJobs, 10) + parseInt(workingJobs, 10);
+
+    if (total === 0) {
+      return;
+    }
+
+    if (attempt === 0) {
+      console.log(`  Waiting for ${total} existing copy job(s) to complete...`);
+    }
+
+    await sleep(5000);
+    attempt++;
+  }
+
+  throw new Error("Timed out waiting for existing copy jobs to complete");
+}
+
+async function runCopyJob(startDate: string, endDate: string, retries = MAX_RETRIES): Promise<boolean> {
+  for (let attempt = 1; attempt <= retries; attempt++) {
+    try {
+      await waitForCopyJobs();
+      
+      exec(
+        `tb --cloud copy run events_backfill --param start_date="${startDate}" --param end_date="${endDate}" --wait`,
+        false
+      );
+      return true;
+    } catch (error: any) {
+      const errorMsg = error.message || error.toString();
+      
+      if (attempt < retries) {
+        console.log(`  Attempt ${attempt}/${retries} failed. Retrying in ${RETRY_DELAY_MS / 1000}s...`);
+        console.log(`  Error: ${errorMsg.substring(0, 200)}`);
+        await sleep(RETRY_DELAY_MS);
+      } else {
+        console.error(`  All ${retries} attempts failed for chunk ${startDate} -> ${endDate}`);
+        console.error(`  Error: ${errorMsg}`);
+        return false;
+      }
+    }
+  }
+  return false;
+}
+
+interface Chunk {
+  start: string;
+  end: string;
+}
+
+function generateChunks(startDate: string, endDate: string, chunkHours: number): Chunk[] {
+  const chunks: Chunk[] = [];
+  let current = startDate;
+
+  while (parseDate(current) < parseDate(endDate)) {
+    let next = addHours(current, chunkHours);
+    
+    if (parseDate(next) > parseDate(endDate)) {
+      next = endDate;
+    }
+
+    chunks.push({ start: current, end: next });
+    current = next;
+  }
+
+  return chunks;
+}
+
+async function main() {
+  console.log("=== Events Backfill Script (TypeScript) ===\n");
+
+  const args = parseArgs();
+
+  // Handle truncation
+  if (args.truncate) {
+    if (!args.startDate) {
+      console.error("Error: --truncate requires --start-date to be specified.");
+      console.error("This prevents accidentally truncating without knowing where to start.");
+      process.exit(1);
+    }
+
+    const currentCount = getEventCount();
+    console.log(`Current event count: ${currentCount}`);
+    
+    const confirmed = await promptConfirmationAsync(
+      `This will TRUNCATE all ${currentCount} events. Are you sure?`
+    );
+    
+    if (!confirmed) {
+      console.log("Aborted.");
+      process.exit(0);
+    }
+
+    truncateEvents();
+  }
+
+  // Determine end date (today 00:00 UTC)
+  const endDate = args.endDate ?? getTodayMidnight();
+  console.log(`End date: ${endDate}`);
+
+  // Determine start date (from latest event in Tinybird, or from args)
+  let startDate: string;
+  
+  if (args.startDate) {
+    startDate = args.startDate;
+    console.log(`Start date (from args): ${startDate}`);
+  } else {
+    const latestInTinybird = getLatestEventInTinybird();
+    
+    if (latestInTinybird) {
+      // Resume from after the latest event (floor to hour boundary)
+      startDate = floorToHour(latestInTinybird);
+      console.log(`Resuming from latest event in Tinybird: ${latestInTinybird}`);
+      console.log(`Start date (floored to hour): ${startDate}`);
+    } else {
+      console.error("No events in Tinybird and no --start-date provided.");
+      console.error("Please provide --start-date to specify where to start backfilling from.");
+      console.error("\nTo find the first event in your source database, run:");
+      console.error("  SELECT min(timestamp) FROM events");
+      process.exit(1);
+    }
+  }
+
+  // Validate dates
+  if (parseDate(startDate) >= parseDate(endDate)) {
+    console.log("\nNothing to backfill - start date is >= end date.");
+    console.log(`Current event count: ${getEventCount()}`);
+    process.exit(0);
+  }
+
+  // Generate chunks
+  const chunks = generateChunks(startDate, endDate, args.chunkHours);
+  console.log(`\nGenerated ${chunks.length} chunks (${args.chunkHours}h each)`);
+  console.log(`Range: ${startDate} -> ${endDate}\n`);
+
+  if (args.dryRun) {
+    console.log("DRY RUN - Would process these chunks:\n");
+    chunks.forEach((chunk, i) => {
+      console.log(`  ${i + 1}. ${chunk.start} -> ${chunk.end}`);
+    });
+    process.exit(0);
+  }
+
+  // Process chunks
+  for (let i = 0; i < chunks.length; i++) {
+    const chunk = chunks[i];
+    const chunkNum = i + 1;
+
+    console.log(`=== Chunk ${chunkNum}/${chunks.length}: ${chunk.start} -> ${chunk.end} ===`);
+
+    const startTime = Date.now();
+    const success = await runCopyJob(chunk.start, chunk.end);
+    const duration = Math.round((Date.now() - startTime) / 1000);
+
+    if (success) {
+      const rowCount = getEventCount();
+      console.log(`✓ Chunk ${chunkNum} COMPLETE in ${duration}s. Total rows: ${rowCount}\n`);
+    } else {
+      console.log(`✗ Chunk ${chunkNum} FAILED after ${duration}s\n`);
+      console.log(`To resume, run:`);
+      console.log(`  bun scripts/backfill_events.ts\n`);
+      process.exit(1);  // Stop immediately - don't skip chunks
+    }
+
+    // Delay between chunks (unless it's the last one)
+    if (i < chunks.length - 1) {
+      await sleep(DELAY_BETWEEN_CHUNKS_MS);
+    }
+  }
+
+  // Summary
+  console.log("==========================================");
+  console.log("=== Backfill Complete ===");
+  console.log(`Final row count: ${getEventCount()}`);
+}
+
+main().catch((error) => {
+  console.error("Fatal error:", error);
+  process.exit(1);
+});
diff --git a/server/tinybird/scripts/benchmark_groupable.sh b/server/tinybird/scripts/benchmark_groupable.sh
new file mode 100644
index 000000000..21236abaf
--- /dev/null
+++ b/server/tinybird/scripts/benchmark_groupable.sh
@@ -0,0 +1,310 @@
+#!/bin/bash
+# Benchmark: Global Top 9 vs Per-Bin Top 9 grouping approaches
+# Tests with both low cardinality (billing_source) and high cardinality (session_id)
+
+set -e
+cd "$(dirname "$0")/.."
+
+ORG_ID="0pCIbS4AMAFDB1iBMNhARWZt2gDtVwQx"
+ENV="live"
+START_DATE="2024-01-01 00:00:00"
+END_DATE="2025-12-31 23:59:59"
+RUNS=10
+
+echo "=== Benchmark: Global Top 9 vs Per-Bin Top 9 ==="
+echo "Org: $ORG_ID"
+echo "Runs: $RUNS"
+echo ""
+
+run_benchmark() {
+  local name="$1"
+  local query_file="$2"
+  local times=()
+  
+  echo "--- $name ---"
+  
+  for i in $(seq 1 $RUNS); do
+    result=$(tb --cloud sql --stats "$(cat $query_file)" 2>&1)
+    elapsed=$(echo "$result" | grep "Query took" | awk '{print $4}')
+    if [ -z "$elapsed" ]; then
+      echo "  Run $i: ERROR"
+      echo "$result" | head -5
+      continue
+    fi
+    times+=("$elapsed")
+    printf "  Run %2d: %ss\n" "$i" "$elapsed"
+  done
+  
+  if [ ${#times[@]} -eq 0 ]; then
+    echo "  All runs failed!"
+    echo "0" > "/tmp/benchmark_${name}.txt"
+    return
+  fi
+  
+  avg=$(printf '%s\n' "${times[@]}" | awk '{sum+=$1} END {printf "%.6f", sum/NR}')
+  min=$(printf '%s\n' "${times[@]}" | sort -n | head -1)
+  max=$(printf '%s\n' "${times[@]}" | sort -n | tail -1)
+  avg_ms=$(awk "BEGIN {printf \"%.1f\", $avg * 1000}")
+  
+  echo ""
+  echo "  Avg: ${avg}s (${avg_ms}ms) | Min: ${min}s | Max: ${max}s"
+  echo ""
+  
+  echo "$avg" > "/tmp/benchmark_${name}.txt"
+}
+
+# Warmup
+echo "Warming up..."
+tb --cloud sql "SELECT 1" > /dev/null 2>&1
+echo ""
+
+# Create temp SQL files
+mkdir -p /tmp/benchmark_sql
+
+# Global Top 9 - billing_source
+cat > /tmp/benchmark_sql/global_billing.sql << 'EOSQL'
+WITH top_groups AS (
+    SELECT groupArray(9)(group_value) as top_groups
+    FROM (
+        SELECT 
+            properties.billing_source::String as group_value,
+            sum(total_value) as total
+        FROM events_hourly_mv
+        WHERE
+            org_id = '0pCIbS4AMAFDB1iBMNhARWZt2gDtVwQx'
+            AND env = 'live'
+            AND event_name = 'usd_api_credits'
+            AND hour >= toDateTime('2024-01-01 00:00:00')
+            AND hour <= toDateTime('2025-12-31 23:59:59')
+        GROUP BY group_value
+        ORDER BY total DESC
+    )
+),
+max_per_bin AS (
+    SELECT max(bin_unique_count) > 9 as any_bin_truncated
+    FROM (
+        SELECT 
+            formatDateTime(toStartOfDay(hour, 'UTC'), '%F %T') as period,
+            count(DISTINCT properties.billing_source::String) as bin_unique_count
+        FROM events_hourly_mv
+        WHERE
+            org_id = '0pCIbS4AMAFDB1iBMNhARWZt2gDtVwQx'
+            AND env = 'live'
+            AND event_name = 'usd_api_credits'
+            AND hour >= toDateTime('2024-01-01 00:00:00')
+            AND hour <= toDateTime('2025-12-31 23:59:59')
+        GROUP BY period
+    )
+)
+SELECT
+    formatDateTime(toStartOfDay(hour, 'UTC'), '%F %T') as period,
+    event_name,
+    if(
+        has((SELECT top_groups FROM top_groups), properties.billing_source::String),
+        properties.billing_source::String,
+        'Other'
+    ) as group_value,
+    sum(total_value) as total_value,
+    (SELECT any_bin_truncated FROM max_per_bin) as _truncated
+FROM events_hourly_mv
+WHERE
+    org_id = '0pCIbS4AMAFDB1iBMNhARWZt2gDtVwQx'
+    AND env = 'live'
+    AND event_name = 'usd_api_credits'
+    AND hour >= toDateTime('2024-01-01 00:00:00')
+    AND hour <= toDateTime('2025-12-31 23:59:59')
+GROUP BY period, event_name, group_value
+ORDER BY period, event_name, group_value
+EOSQL
+
+# Per-Bin Top 9 - billing_source
+cat > /tmp/benchmark_sql/perbin_billing.sql << 'EOSQL'
+WITH base AS (
+    SELECT
+        formatDateTime(toStartOfDay(hour, 'UTC'), '%F %T') as period,
+        event_name,
+        properties.billing_source::String as group_value,
+        sum(total_value) as total_value
+    FROM events_hourly_mv
+    WHERE
+        org_id = '0pCIbS4AMAFDB1iBMNhARWZt2gDtVwQx'
+        AND env = 'live'
+        AND event_name = 'usd_api_credits'
+        AND hour >= toDateTime('2024-01-01 00:00:00')
+        AND hour <= toDateTime('2025-12-31 23:59:59')
+    GROUP BY period, event_name, group_value
+),
+ranked AS (
+    SELECT
+        *,
+        row_number() OVER (PARTITION BY period, event_name ORDER BY total_value DESC) as rn
+    FROM base
+)
+SELECT
+    period,
+    event_name,
+    if(rn <= 9, group_value, 'Other') as group_value,
+    sum(total_value) as total_value,
+    max(rn) > 9 as _truncated
+FROM ranked
+GROUP BY period, event_name, group_value
+ORDER BY period, event_name, group_value
+EOSQL
+
+# Global Top 9 - session_id
+cat > /tmp/benchmark_sql/global_session.sql << 'EOSQL'
+WITH top_groups AS (
+    SELECT groupArray(9)(group_value) as top_groups
+    FROM (
+        SELECT 
+            properties.session_id::String as group_value,
+            sum(total_value) as total
+        FROM events_hourly_mv
+        WHERE
+            org_id = '0pCIbS4AMAFDB1iBMNhARWZt2gDtVwQx'
+            AND env = 'live'
+            AND event_name = 'usd_api_credits'
+            AND hour >= toDateTime('2024-01-01 00:00:00')
+            AND hour <= toDateTime('2025-12-31 23:59:59')
+        GROUP BY group_value
+        ORDER BY total DESC
+    )
+),
+max_per_bin AS (
+    SELECT max(bin_unique_count) > 9 as any_bin_truncated
+    FROM (
+        SELECT 
+            formatDateTime(toStartOfDay(hour, 'UTC'), '%F %T') as period,
+            count(DISTINCT properties.session_id::String) as bin_unique_count
+        FROM events_hourly_mv
+        WHERE
+            org_id = '0pCIbS4AMAFDB1iBMNhARWZt2gDtVwQx'
+            AND env = 'live'
+            AND event_name = 'usd_api_credits'
+            AND hour >= toDateTime('2024-01-01 00:00:00')
+            AND hour <= toDateTime('2025-12-31 23:59:59')
+        GROUP BY period
+    )
+)
+SELECT
+    formatDateTime(toStartOfDay(hour, 'UTC'), '%F %T') as period,
+    event_name,
+    if(
+        has((SELECT top_groups FROM top_groups), properties.session_id::String),
+        properties.session_id::String,
+        'Other'
+    ) as group_value,
+    sum(total_value) as total_value,
+    (SELECT any_bin_truncated FROM max_per_bin) as _truncated
+FROM events_hourly_mv
+WHERE
+    org_id = '0pCIbS4AMAFDB1iBMNhARWZt2gDtVwQx'
+    AND env = 'live'
+    AND event_name = 'usd_api_credits'
+    AND hour >= toDateTime('2024-01-01 00:00:00')
+    AND hour <= toDateTime('2025-12-31 23:59:59')
+GROUP BY period, event_name, group_value
+ORDER BY period, event_name, group_value
+EOSQL
+
+# Per-Bin Top 9 - session_id
+cat > /tmp/benchmark_sql/perbin_session.sql << 'EOSQL'
+WITH base AS (
+    SELECT
+        formatDateTime(toStartOfDay(hour, 'UTC'), '%F %T') as period,
+        event_name,
+        properties.session_id::String as group_value,
+        sum(total_value) as total_value
+    FROM events_hourly_mv
+    WHERE
+        org_id = '0pCIbS4AMAFDB1iBMNhARWZt2gDtVwQx'
+        AND env = 'live'
+        AND event_name = 'usd_api_credits'
+        AND hour >= toDateTime('2024-01-01 00:00:00')
+        AND hour <= toDateTime('2025-12-31 23:59:59')
+    GROUP BY period, event_name, group_value
+),
+ranked AS (
+    SELECT
+        *,
+        row_number() OVER (PARTITION BY period, event_name ORDER BY total_value DESC) as rn
+    FROM base
+)
+SELECT
+    period,
+    event_name,
+    if(rn <= 9, group_value, 'Other') as group_value,
+    sum(total_value) as total_value,
+    max(rn) > 9 as _truncated
+FROM ranked
+GROUP BY period, event_name, group_value
+ORDER BY period, event_name, group_value
+EOSQL
+
+echo "==========================================="
+echo "=== LOW CARDINALITY: billing_source ==="
+echo "==========================================="
+echo ""
+
+run_benchmark "global_billing" /tmp/benchmark_sql/global_billing.sql
+run_benchmark "perbin_billing" /tmp/benchmark_sql/perbin_billing.sql
+
+echo "==========================================="
+echo "=== HIGH CARDINALITY: session_id ==="
+echo "==========================================="
+echo ""
+
+run_benchmark "global_session" /tmp/benchmark_sql/global_session.sql
+run_benchmark "perbin_session" /tmp/benchmark_sql/perbin_session.sql
+
+echo "==========================================="
+echo "=== SUMMARY ==="
+echo "==========================================="
+echo ""
+
+# Read results
+t_global_billing=$(cat /tmp/benchmark_global_billing.txt 2>/dev/null || echo "0")
+t_perbin_billing=$(cat /tmp/benchmark_perbin_billing.txt 2>/dev/null || echo "0")
+t_global_session=$(cat /tmp/benchmark_global_session.txt 2>/dev/null || echo "0")
+t_perbin_session=$(cat /tmp/benchmark_perbin_session.txt 2>/dev/null || echo "0")
+
+ms_global_billing=$(awk "BEGIN {printf \"%.1f\", $t_global_billing * 1000}")
+ms_perbin_billing=$(awk "BEGIN {printf \"%.1f\", $t_perbin_billing * 1000}")
+ms_global_session=$(awk "BEGIN {printf \"%.1f\", $t_global_session * 1000}")
+ms_perbin_session=$(awk "BEGIN {printf \"%.1f\", $t_perbin_session * 1000}")
+
+echo "LOW CARDINALITY (billing_source):"
+echo "  Global Top 9:   ${ms_global_billing}ms"
+echo "  Per-Bin Top 9:  ${ms_perbin_billing}ms"
+if awk "BEGIN {exit !($t_global_billing > 0 && $t_perbin_billing > 0)}"; then
+  ratio_billing=$(awk "BEGIN {printf \"%.2f\", $t_global_billing / $t_perbin_billing}")
+  pct_faster=$(awk "BEGIN {printf \"%.0f\", ($t_global_billing - $t_perbin_billing) / $t_global_billing * 100}")
+  if [ "$pct_faster" -gt 0 ]; then
+    echo "  Winner:         Per-Bin is ${ratio_billing}x faster (${pct_faster}% improvement)"
+  else
+    pct_slower=$(awk "BEGIN {printf \"%.0f\", ($t_perbin_billing - $t_global_billing) / $t_global_billing * 100}")
+    echo "  Winner:         Global is faster (per-bin ${pct_slower}% slower)"
+  fi
+fi
+echo ""
+
+echo "HIGH CARDINALITY (session_id):"
+echo "  Global Top 9:   ${ms_global_session}ms"
+echo "  Per-Bin Top 9:  ${ms_perbin_session}ms"
+if awk "BEGIN {exit !($t_global_session > 0 && $t_perbin_session > 0)}"; then
+  ratio_session=$(awk "BEGIN {printf \"%.2f\", $t_global_session / $t_perbin_session}")
+  pct_faster=$(awk "BEGIN {printf \"%.0f\", ($t_global_session - $t_perbin_session) / $t_global_session * 100}")
+  if [ "$pct_faster" -gt 0 ]; then
+    echo "  Winner:         Per-Bin is ${ratio_session}x faster (${pct_faster}% improvement)"
+  else
+    pct_slower=$(awk "BEGIN {printf \"%.0f\", ($t_perbin_session - $t_global_session) / $t_global_session * 100}")
+    echo "  Winner:         Global is faster (per-bin ${pct_slower}% slower)"
+  fi
+fi
+echo ""
+
+# Cleanup
+rm -f /tmp/benchmark_*.txt
+rm -rf /tmp/benchmark_sql
+
+echo "=== Done ==="
diff --git a/server/tinybird/scripts/benchmark_mv.sh b/server/tinybird/scripts/benchmark_mv.sh
new file mode 100644
index 000000000..02257b744
--- /dev/null
+++ b/server/tinybird/scripts/benchmark_mv.sh
@@ -0,0 +1,164 @@
+#!/bin/bash
+# Benchmark: JSON String MV vs JSON Type MV
+# Runs each query multiple times and extracts server-side execution time
+
+ORG_ID="0pCIbS4AMAFDB1iBMNhARWZt2gDtVwQx"
+EVENT_NAME="usd_api_credits"
+RUNS=10
+
+echo "=== Benchmark: JSON String vs JSON Type MV ==="
+echo "Org: $ORG_ID"
+echo "Event: $EVENT_NAME"
+echo "Runs: $RUNS"
+echo ""
+
+run_benchmark() {
+  local name="$1"
+  local query="$2"
+  local times=()
+  
+  echo "--- $name ---"
+  
+  for i in $(seq 1 $RUNS); do
+    elapsed=$(tb --cloud sql --stats "$query" 2>&1 | grep "Query took" | awk '{print $4}')
+    times+=("$elapsed")
+    printf "  Run %2d: %ss\n" "$i" "$elapsed"
+  done
+  
+  avg=$(printf '%s\n' "${times[@]}" | awk '{sum+=$1} END {printf "%.6f", sum/NR}')
+  min=$(printf '%s\n' "${times[@]}" | sort -n | head -1)
+  max=$(printf '%s\n' "${times[@]}" | sort -n | tail -1)
+  avg_ms=$(awk "BEGIN {printf \"%.1f\", $avg * 1000}")
+  
+  echo ""
+  echo "  Avg: ${avg}s (${avg_ms}ms) | Min: ${min}s | Max: ${max}s"
+  echo ""
+  
+  # Return avg for comparison
+  echo "$avg" > "/tmp/benchmark_${name// /_}.txt"
+}
+
+# Warmup
+echo "Warming up..."
+tb --cloud sql "SELECT 1" > /dev/null 2>&1
+echo ""
+
+# JSON String MV
+run_benchmark "json_string" "
+  SELECT 
+    JSONExtractString(properties, 'billing_source') as billing_source,
+    JSONExtractString(properties, 'project_id') as project_id,
+    sum(total_value) as total_value,
+    sum(event_count) as event_count
+  FROM events_hourly_exp_json_string_mv
+  WHERE org_id = '$ORG_ID'
+    AND event_name = '$EVENT_NAME'
+  GROUP BY billing_source, project_id
+  ORDER BY total_value DESC
+"
+
+# JSON Type MV (::String cast)
+run_benchmark "json_cast" "
+  SELECT 
+    properties.billing_source::String as billing_source,
+    properties.project_id::String as project_id,
+    sum(total_value) as total_value,
+    sum(event_count) as event_count
+  FROM events_hourly_exp_json_mv
+  WHERE org_id = '$ORG_ID'
+    AND event_name = '$EVENT_NAME'
+  GROUP BY billing_source, project_id
+  ORDER BY total_value DESC
+"
+
+# JSON Type MV (.:String subcolumn)
+run_benchmark "json_subcolumn" "
+  SELECT 
+    properties.billing_source.:String as billing_source,
+    properties.project_id.:String as project_id,
+    sum(total_value) as total_value,
+    sum(event_count) as event_count
+  FROM events_hourly_exp_json_mv
+  WHERE org_id = '$ORG_ID'
+    AND event_name = '$EVENT_NAME'
+  GROUP BY billing_source, project_id
+  ORDER BY total_value DESC
+"
+
+# JSON Type MV (toString - handles all types)
+run_benchmark "json_tostring" "
+  SELECT 
+    toString(properties.billing_source) as billing_source,
+    toString(properties.project_id) as project_id,
+    sum(total_value) as total_value,
+    sum(event_count) as event_count
+  FROM events_hourly_exp_json_mv
+  WHERE org_id = '$ORG_ID'
+    AND event_name = '$EVENT_NAME'
+  GROUP BY billing_source, project_id
+  ORDER BY total_value DESC
+"
+
+echo "==========================================="
+echo "=== Summary ==="
+echo ""
+
+# Read results
+t_string=$(cat /tmp/benchmark_json_string.txt)
+t_cast=$(cat /tmp/benchmark_json_cast.txt)
+t_subcolumn=$(cat /tmp/benchmark_json_subcolumn.txt)
+t_tostring=$(cat /tmp/benchmark_json_tostring.txt)
+
+ms_string=$(awk "BEGIN {printf \"%.1f\", $t_string * 1000}")
+ms_cast=$(awk "BEGIN {printf \"%.1f\", $t_cast * 1000}")
+ms_subcolumn=$(awk "BEGIN {printf \"%.1f\", $t_subcolumn * 1000}")
+ms_tostring=$(awk "BEGIN {printf \"%.1f\", $t_tostring * 1000}")
+
+echo "Results (avg query time):"
+echo "  1. JSON String (JSONExtractString):  ${ms_string}ms"
+echo "  2. JSON Type (::String cast):        ${ms_cast}ms"
+echo "  3. JSON Type (.:String subcolumn):   ${ms_subcolumn}ms"
+echo "  4. JSON Type (toString):             ${ms_tostring}ms"
+echo ""
+
+# Find fastest
+fastest_time=$t_cast
+fastest_name="JSON Type (::String cast)"
+fastest_ms=$ms_cast
+
+if awk "BEGIN {exit !($t_subcolumn < $fastest_time)}"; then
+  fastest_time=$t_subcolumn
+  fastest_name="JSON Type (.:String subcolumn)"
+  fastest_ms=$ms_subcolumn
+fi
+if awk "BEGIN {exit !($t_tostring < $fastest_time)}"; then
+  fastest_time=$t_tostring
+  fastest_name="JSON Type (toString)"
+  fastest_ms=$ms_tostring
+fi
+if awk "BEGIN {exit !($t_string < $fastest_time)}"; then
+  fastest_time=$t_string
+  fastest_name="JSON String (JSONExtractString)"
+  fastest_ms=$ms_string
+fi
+
+echo "Winner: $fastest_name (${fastest_ms}ms)"
+echo ""
+echo "Comparisons:"
+
+ratio_string=$(awk "BEGIN {printf \"%.1f\", $t_string / $fastest_time}")
+ratio_cast=$(awk "BEGIN {printf \"%.1f\", $t_cast / $fastest_time}")
+ratio_subcolumn=$(awk "BEGIN {printf \"%.1f\", $t_subcolumn / $fastest_time}")
+ratio_tostring=$(awk "BEGIN {printf \"%.1f\", $t_tostring / $fastest_time}")
+
+echo "  $fastest_name is:"
+echo "    ${ratio_string}x faster than JSON String (JSONExtractString)"
+echo "    ${ratio_cast}x faster than JSON Type (::String cast)"
+echo "    ${ratio_subcolumn}x faster than JSON Type (.:String subcolumn)"
+echo "    ${ratio_tostring}x faster than JSON Type (toString)"
+
+# Cleanup
+rm -f /tmp/benchmark_json_*.txt
+
+echo ""
+echo "=== Done ==="