diff --git a/apps/alerting/alchemy.run.ts b/apps/alerting/alchemy.run.ts index f9983d57..06a0cc97 100644 --- a/apps/alerting/alchemy.run.ts +++ b/apps/alerting/alchemy.run.ts @@ -69,6 +69,12 @@ export const createAlertingWorker = ({ stage, mapleDb }: CreateAlertingWorkerOpt }), TINYBIRD_HOST: requireEnv("TINYBIRD_HOST"), TINYBIRD_TOKEN: Redacted.make(requireEnv("TINYBIRD_TOKEN")), + // Alert-rule evaluation runs Tinybird-scoped raw SQL through + // TinybirdOrgTokenService, which requires both of these — without them + // every tick fails with "TINYBIRD_SIGNING_KEY is required for + // Tinybird-scoped raw SQL" (same bindings as the api worker). + ...optionalSecret("TINYBIRD_SIGNING_KEY"), + ...optionalPlain("TINYBIRD_WORKSPACE_ID"), MAPLE_AUTH_MODE: process.env.MAPLE_AUTH_MODE?.trim() || "self_hosted", MAPLE_DEFAULT_ORG_ID: process.env.MAPLE_DEFAULT_ORG_ID?.trim() || "default", MAPLE_INGEST_KEY_ENCRYPTION_KEY: Redacted.make(requireEnv("MAPLE_INGEST_KEY_ENCRYPTION_KEY")), diff --git a/apps/api/src/lib/WarehouseQueryService.test.ts b/apps/api/src/lib/WarehouseQueryService.test.ts index 281244ee..a1931c88 100644 --- a/apps/api/src/lib/WarehouseQueryService.test.ts +++ b/apps/api/src/lib/WarehouseQueryService.test.ts @@ -726,6 +726,15 @@ describe("createTinybirdSdkSqlClient.sql FORMAT normalization", () => { assert.match(sent, /FORMAT JSON$/) }) + it("does not double-append when profile SETTINGS precede FORMAT JSON", async () => { + // Canonical order emitted by appendSettings — Tinybird rejects the inverse. + const sent = await captureSql( + "SELECT 1 SETTINGS max_execution_time=15, max_memory_usage=1500000000\nFORMAT JSON", + ) + assert.strictEqual(countFormats(sent), 1) + assert.match(sent, /SETTINGS max_execution_time=15, max_memory_usage=1500000000\nFORMAT JSON$/) + }) + it("does not double-append when FORMAT JSON is followed by profile SETTINGS", async () => { const sent = await captureSql( "SELECT 1\nFORMAT JSON SETTINGS max_execution_time=15, max_memory_usage=1500000000", diff --git a/apps/api/src/lib/WarehouseQueryService.ts b/apps/api/src/lib/WarehouseQueryService.ts index 27c8614f..c5b16ac2 100644 --- a/apps/api/src/lib/WarehouseQueryService.ts +++ b/apps/api/src/lib/WarehouseQueryService.ts @@ -166,9 +166,11 @@ const createTinybirdSdkSqlClient = ( // Tinybird Cloud currently defaults /v0/sql to JSON, while Tinybird Local // defaults to tab-separated output. The SDK always calls response.json(), so // make the expected wire format explicit for both environments. DSL-compiled - // queries already end with `FORMAT JSON` (optionally followed by profile - // SETTINGS) — appending a second FORMAT clause is a ClickHouse syntax error, - // so only add one when the query doesn't carry its own. + // queries already end with `FORMAT JSON` (profile SETTINGS are inserted + // before it by appendSettings) — appending a second FORMAT clause is a + // ClickHouse syntax error, so only add one when the query doesn't carry + // its own. The trailing-SETTINGS alternative covers SQL from callers that + // still emit the legacy `FORMAT JSON SETTINGS …` order. const trimmed = sql.trimEnd().replace(/;$/, "") const hasFormat = /\bFORMAT\s+\w+(\s+SETTINGS\s[^\n]*)?$/i.test(trimmed) const jsonSql = hasFormat ? trimmed : `${trimmed}\nFORMAT JSON` diff --git a/apps/cli/src/server/inserts.ts b/apps/cli/src/server/inserts.ts index 6a956cc3..701c09d3 100644 --- a/apps/cli/src/server/inserts.ts +++ b/apps/cli/src/server/inserts.ts @@ -53,3 +53,48 @@ export function buildInsertSql(datasource: string, ndjson: string): string { if (!template) throw new Error(`no insert mapping for datasource '${datasource}'`) return template.prefix + escapeSqlLiteral(ndjson) + template.suffix } + +/** + * chDB parses the entire statement — inlined data literal included — against + * its default `max_query_size` (~256KB), so a large batch in one statement + * fails with "Code: 62 … Max query size exceeded". Budget for the escaped + * payload per statement, leaving headroom for the template prefix/suffix. + */ +const MAX_ESCAPED_PAYLOAD_BYTES = 200_000 + +export interface InsertStatement { + readonly sql: string + readonly rowCount: number +} + +/** + * Like {@link buildInsertSql}, but splits the NDJSON batch on line boundaries + * into as many statements as needed so each stays under chDB's query-size + * limit. A single line larger than the budget is emitted as its own statement + * (never split mid-line). + */ +export function buildInsertStatements(datasource: string, ndjson: string): InsertStatement[] { + const template = templates.get(datasource) + if (!template) throw new Error(`no insert mapping for datasource '${datasource}'`) + const out: InsertStatement[] = [] + let chunk: string[] = [] + let chunkBytes = 0 + const flush = () => { + if (chunk.length === 0) return + out.push({ sql: template.prefix + chunk.join("\n") + template.suffix, rowCount: chunk.length }) + chunk = [] + chunkBytes = 0 + } + for (const line of ndjson.split("\n")) { + if (line.length === 0) continue + // Escaping is per-character, so escaping line-by-line and joining with + // "\n" is identical to escaping the whole batch at once. + const escaped = escapeSqlLiteral(line) + const bytes = Buffer.byteLength(escaped, "utf8") + 1 + if (chunkBytes > 0 && chunkBytes + bytes > MAX_ESCAPED_PAYLOAD_BYTES) flush() + chunk.push(escaped) + chunkBytes += bytes + } + flush() + return out +} diff --git a/apps/cli/src/server/serve.ts b/apps/cli/src/server/serve.ts index 5ee6b1f0..fca8539a 100644 --- a/apps/cli/src/server/serve.ts +++ b/apps/cli/src/server/serve.ts @@ -8,7 +8,7 @@ import { gunzipSync } from "node:zlib" import { TelemetryLayer } from "../core/telemetry" import { isLoopbackHostname } from "../lib/local-address" import { acquireChdb, type Chdb, type ChdbError } from "./chdb" -import { buildInsertSql } from "./inserts" +import { buildInsertStatements } from "./inserts" import { encodeLogs, encodeMetrics, encodeTraces, type EncodedBatch } from "./otlp/encode" import { decodeLogsRequest, decodeMetricsRequest, decodeTraceRequest } from "./otlp/proto" import schemaSql from "./schema/local-schema.sql" with { type: "text" } @@ -177,16 +177,18 @@ async function ingest(db: Chdb, signal: Signal, req: Request): Promise JSON.stringify({ body: `row-${i}${"x".repeat(pad)}` }) + +describe("buildInsertStatements", () => { + it("emits a single statement for a small batch, identical to buildInsertSql", () => { + const ndjson = [line(1), line(2), line(3)].join("\n") + const statements = buildInsertStatements("logs", ndjson) + expect(statements).toHaveLength(1) + expect(statements[0]?.rowCount).toBe(3) + expect(statements[0]?.sql).toBe(buildInsertSql("logs", ndjson)) + }) + + it("ignores empty lines when counting rows", () => { + const statements = buildInsertStatements("logs", `${line(1)}\n\n${line(2)}\n`) + expect(statements).toHaveLength(1) + expect(statements[0]?.rowCount).toBe(2) + }) + + // Regression: a large OTLP logs batch inlined into one statement exceeded + // chDB's default max_query_size (~256KB) → "Code: 62 … Max query size + // exceeded". Batches must be split on line boundaries under the budget. + it("splits a large batch into multiple statements under the size budget", () => { + const rows = Array.from({ length: 50 }, (_, i) => line(i, 10_000)) + const statements = buildInsertStatements("logs", rows.join("\n")) + expect(statements.length).toBeGreaterThan(1) + expect(statements.reduce((n, s) => n + s.rowCount, 0)).toBe(50) + for (const statement of statements) { + expect(Buffer.byteLength(statement.sql, "utf8")).toBeLessThan(256 * 1024) + } + }) + + it("splits only on line boundaries — concatenated payloads round-trip", () => { + const rows = Array.from({ length: 50 }, (_, i) => line(i, 10_000)) + const statements = buildInsertStatements("logs", rows.join("\n")) + const payloads = statements.map((s) => { + const start = s.sql.lastIndexOf(", '") + 3 + return s.sql.slice(start, -2) + }) + expect(payloads.join("\n")).toBe(rows.join("\n")) + }) + + it("passes a single oversized line through as its own statement", () => { + const huge = line(0, 300_000) + const statements = buildInsertStatements("logs", [line(1), huge, line(2)].join("\n")) + expect(statements.map((s) => s.rowCount)).toEqual([1, 1, 1]) + expect(statements[1]?.sql).toContain("x".repeat(300_000)) + }) + + it("budgets on escaped bytes, not raw bytes", () => { + // Each backslash escapes to two characters; 60 rows of ~3.4KB raw escape + // to ~6.7KB each (~400KB total), forcing at least a three-way split. + const rows = Array.from({ length: 60 }, () => JSON.stringify({ body: "\\".repeat(3_333) })) + const statements = buildInsertStatements("logs", rows.join("\n")) + expect(statements.length).toBeGreaterThan(2) + for (const statement of statements) { + expect(Buffer.byteLength(statement.sql, "utf8")).toBeLessThan(256 * 1024) + } + }) + + it("throws on an unknown datasource", () => { + expect(() => buildInsertStatements("nope", line(1))).toThrow("no insert mapping") + }) +}) diff --git a/packages/query-engine/src/profiles/query-profile.test.ts b/packages/query-engine/src/profiles/query-profile.test.ts index 79ba5096..c1197682 100644 --- a/packages/query-engine/src/profiles/query-profile.test.ts +++ b/packages/query-engine/src/profiles/query-profile.test.ts @@ -52,6 +52,31 @@ describe("appendSettings", () => { it("appends max_block_size", () => { expect(appendSettings("SELECT 1", { maxBlockSize: 512 })).toBe("SELECT 1 SETTINGS max_block_size=512") }) + + // Regression: Tinybird rejects `FORMAT JSON SETTINGS …` with + // "Syntax error: failed at position … (FORMAT)" — SETTINGS must precede + // a trailing FORMAT clause. + it("inserts SETTINGS before a trailing FORMAT clause", () => { + expect(appendSettings("SELECT 1 FORMAT JSON", { maxExecutionTime: 15 })).toBe( + "SELECT 1 SETTINGS max_execution_time=15 FORMAT JSON", + ) + }) + + it("inserts SETTINGS before FORMAT with a trailing semicolon and newlines", () => { + expect(appendSettings("SELECT 1\nFORMAT JSONEachRow;", { maxThreads: 2 })).toBe( + "SELECT 1 SETTINGS max_threads=2\nFORMAT JSONEachRow", + ) + }) + + it("leaves a trailing FORMAT untouched when settings are empty", () => { + expect(appendSettings("SELECT 1 FORMAT JSON", {})).toBe("SELECT 1 FORMAT JSON") + }) + + it("only treats a FORMAT at the end of the query as the format clause", () => { + expect(appendSettings("SELECT formatDateTime(now(), '%F') AS format_col", { maxThreads: 2 })).toBe( + "SELECT formatDateTime(now(), '%F') AS format_col SETTINGS max_threads=2", + ) + }) }) describe("stripTinybirdRestrictedSettings", () => { diff --git a/packages/query-engine/src/profiles/query-profile.ts b/packages/query-engine/src/profiles/query-profile.ts index 127fb16f..21239582 100644 --- a/packages/query-engine/src/profiles/query-profile.ts +++ b/packages/query-engine/src/profiles/query-profile.ts @@ -126,8 +126,16 @@ export const stripTinybirdRestrictedSettings = ( } /** - * Append a ClickHouse `SETTINGS` clause to a SQL string. Returns the - * input unchanged when no settings are provided. + * Matches a trailing `FORMAT ` clause (the DSL compiler terminates every + * query with `FORMAT JSON`). `SETTINGS` must precede `FORMAT` — Tinybird's + * ClickHouse rejects `FORMAT JSON SETTINGS …` with a syntax error. + */ +const trailingFormatRe = /\s+FORMAT\s+\w+\s*$/i + +/** + * Add a ClickHouse `SETTINGS` clause to a SQL string, inserting it before a + * trailing `FORMAT ` clause when present. Returns the input unchanged + * when no settings are provided. * * Caller must guarantee the SQL doesn't already contain a SETTINGS * clause — none of maple's DSL queries do today. @@ -142,7 +150,14 @@ export const appendSettings = (sql: string, settings: WarehouseQuerySettings | u } } if (parts.length === 0) return sql - return `${sql.replace(/;\s*$/, "")} SETTINGS ${parts.join(", ")}` + const clause = `SETTINGS ${parts.join(", ")}` + const trimmed = sql.replace(/;\s*$/, "") + const formatMatch = trimmed.match(trailingFormatRe) + if (formatMatch) { + const body = trimmed.slice(0, formatMatch.index) + return `${body} ${clause}${formatMatch[0]}` + } + return `${trimmed} ${clause}` } /**