comi-logic/cfw-autumn
8
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add Tinybird dual-write infrastructure

Browse Source 
Add parallel event ingestion to Tinybird alongside existing Postgres/ClickHouse flow.
Events are sent to both systems simultaneously via EventBatchingManager.

- Add Tinybird client setup and ingest endpoint (zod-bird)
- Add event mapping and send utilities with retry/error handling
- Add complete Tinybird datafiles (datasources, pipes, materializations, copies, scripts)
- Modify EventBatchingManager to dual-write to SQS + Tinybird
- Route track events through batching manager

Tinybird will silently skip if not configured (TINYBIRD_URL/TINYBIRD_TOKEN).
This commit is contained in:
amianthus
2026-01-29 17:36:11 +00:00
parent a60b5fca56
commit 6d52409d6c
9 changed files with 1341 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

20
server/tinybird/copies/events_backfill.pipe Normal file
View File

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

66
server/tinybird/pipes/aggregate.pipe Normal file
View File

@@ -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 %}

65
server/tinybird/pipes/aggregate_groupable.pipe Normal file
View File

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

36
server/tinybird/pipes/aggregate_simple.pipe Normal file
View File

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

40
server/tinybird/pipes/list_events.pipe Normal file
View File

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

235
server/tinybird/scripts/backfill_events.sh Executable file
View File

@@ -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"

405
server/tinybird/scripts/backfill_events.ts Normal file
View File

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

310
server/tinybird/scripts/benchmark_groupable.sh Normal file
View File

@@ -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 ==="

164
server/tinybird/scripts/benchmark_mv.sh Normal file
View File

@@ -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 ==="