diff --git a/README.md b/README.md index dedd7bdf..22f3a180 100644 --- a/README.md +++ b/README.md @@ -294,7 +294,7 @@ There is only one "database": the Cloudflare Analytics Engine dataset, which is Right now there is no local "test" database. This means in local development: - Writes will no-op (no hits will be recorded) -- Reads will be read from the production Analaytics Engine dataset (local development shows production data) +- Reads will be read from the production Analytics Engine dataset (local development shows production data) ### Sampling diff --git a/packages/server/workers/lib/__tests__/arrow.test.ts b/packages/server/workers/lib/__tests__/arrow.test.ts new file mode 100644 index 00000000..e7fbd6ea --- /dev/null +++ b/packages/server/workers/lib/__tests__/arrow.test.ts @@ -0,0 +1,110 @@ +import { describe, expect, test } from "vitest"; +import { tableFromIPC, tableToIPC } from "apache-arrow"; + +import { recordsToTable } from "../arrow"; + +describe("recordsToTable", () => { + test("round-trips mixed string/number columns through Arrow IPC", () => { + const records = [ + { + date: "2026-05-11", + siteId: "site-a", + views: 12, + visitors: 7, + bounces: 3, + path: "/", + country: "US", + }, + { + date: "2026-05-11", + siteId: "site-a", + views: 5, + visitors: 4, + bounces: 0, + path: "/docs", + country: "TW", + }, + ]; + + const table = recordsToTable(records); + const buf = new Uint8Array(tableToIPC(table, "file")); + const decoded = tableFromIPC(buf); + + expect(decoded.numRows).toBe(2); + expect(decoded.schema.fields.map((f) => f.name)).toEqual([ + "date", + "siteId", + "views", + "visitors", + "bounces", + "path", + "country", + ]); + + const rows = decoded.toArray().map((r: unknown) => { + const row = r as Record; + return { + date: String(row.date), + siteId: String(row.siteId), + views: Number(row.views), + visitors: Number(row.visitors), + bounces: Number(row.bounces), + path: String(row.path), + country: String(row.country), + }; + }); + expect(rows).toEqual(records); + }); + + test("handles null/undefined values without crashing", () => { + const records = [ + { a: "x", b: 1 }, + { a: null, b: null }, + { a: undefined, b: undefined }, + ]; + const table = recordsToTable(records); + const buf = new Uint8Array(tableToIPC(table, "file")); + const decoded = tableFromIPC(buf); + expect(decoded.numRows).toBe(3); + }); + + test("returns empty table for empty input", () => { + const table = recordsToTable([]); + expect(table.numRows).toBe(0); + }); + + test("does not invoke `new Function()` (CF Workers codegen ban)", () => { + // Cloudflare Workers' runtime forbids `new Function(...)` and `eval`. + // apache-arrow's Builder path (used by tableFromJSON / vectorFromArray) + // triggers `new Function()` for validity-check codegen — this is the + // exact regression we are guarding against. + const realFunction = globalThis.Function; + let callCount = 0; + const FunctionProxy = new Proxy(realFunction, { + construct(target, args) { + callCount++; + return Reflect.construct(target, args); + }, + apply(target, thisArg, args) { + callCount++; + return Reflect.apply(target, thisArg, args); + }, + }); + + const records = [ + { date: "2026-05-11", siteId: "s", views: 1, path: "/" }, + ]; + + try { + // eslint-disable-next-line @typescript-eslint/no-explicit-any + (globalThis as any).Function = FunctionProxy; + const table = recordsToTable(records); + tableToIPC(table, "file"); + } finally { + // eslint-disable-next-line @typescript-eslint/no-explicit-any + (globalThis as any).Function = realFunction; + } + + expect(callCount).toBe(0); + }); +}); diff --git a/packages/server/workers/lib/arrow.ts b/packages/server/workers/lib/arrow.ts index 746ca262..5407981d 100644 --- a/packages/server/workers/lib/arrow.ts +++ b/packages/server/workers/lib/arrow.ts @@ -1,8 +1,95 @@ import { AnalyticsEngineAPI } from "../../app/analytics/query"; import { ColumnMappings } from "../../app/analytics/schema"; -import { tableFromJSON, tableToIPC } from "apache-arrow"; +import { + type Data, + Float64, + RecordBatch, + Schema, + Table, + Utf8, + makeData, + tableToIPC, +} from "apache-arrow"; import dayjs from "dayjs"; +type RecordValue = string | number | null | undefined; +type Record = { [key: string]: RecordValue }; + +// Build a Utf8 Data buffer directly from an array of strings. +// Bypasses apache-arrow's Builder path, which uses `new Function()` for +// validity-checking codegen and is forbidden by the Cloudflare Workers +// runtime ("Code generation from strings disallowed for this context"). +function buildUtf8Data(values: (string | null | undefined)[]): Data { + const encoder = new TextEncoder(); + const encoded: Uint8Array[] = new Array(values.length); + let totalBytes = 0; + for (let i = 0; i < values.length; i++) { + const v = values[i]; + const bytes = v == null ? new Uint8Array(0) : encoder.encode(String(v)); + encoded[i] = bytes; + totalBytes += bytes.length; + } + const valueOffsets = new Int32Array(values.length + 1); + const data = new Uint8Array(totalBytes); + let pos = 0; + for (let i = 0; i < values.length; i++) { + valueOffsets[i] = pos; + data.set(encoded[i], pos); + pos += encoded[i].length; + } + valueOffsets[values.length] = pos; + return makeData({ + type: new Utf8(), + length: values.length, + nullCount: 0, + valueOffsets, + data, + }); +} + +function buildFloat64Data(values: (number | null | undefined)[]): Data { + const buf = new Float64Array(values.length); + for (let i = 0; i < values.length; i++) { + buf[i] = values[i] ?? 0; + } + return makeData({ + type: new Float64(), + length: values.length, + nullCount: 0, + data: buf, + }); +} + +// Convert an array of homogeneous records to an Arrow Table without invoking +// the Builder/`new Function()` codegen path. Column type is inferred from the +// first non-null sample per column: `number` → Float64, otherwise → Utf8. +export function recordsToTable(records: Record[]): Table { + if (records.length === 0) { + return new Table(new Schema([])); + } + const columnNames = Object.keys(records[0]); + const children: { [name: string]: Data } = {}; + for (const name of columnNames) { + let sample: RecordValue = undefined; + for (const r of records) { + const v = r[name]; + if (v != null) { + sample = v; + break; + } + } + const values = records.map((r) => r[name]); + if (typeof sample === "number") { + children[name] = buildFloat64Data(values as (number | null)[]); + } else { + children[name] = buildUtf8Data( + values.map((v) => (v == null ? null : String(v))), + ); + } + } + return new Table(new RecordBatch(children)); +} + export async function extractAsArrow( { accountId, bearerToken }: { accountId: string; bearerToken: string }, bucket: R2Bucket, @@ -27,10 +114,10 @@ export async function extractAsArrow( ); // Convert Map to array of records for Arrow table creation - const records: any[] = []; + const records: Record[] = []; data.forEach((counts, key) => { const [date, siteId, ...columnValues] = key; - const record: any = { + const record: Record = { date, siteId, views: counts.views, @@ -46,8 +133,8 @@ export async function extractAsArrow( records.push(record); }); - // Create Arrow table from JSON records - const table = tableFromJSON(records); + // Build Arrow table without invoking codegen-based builders. + const table = recordsToTable(records); // Convert to Arrow IPC buffer const arrowBuffer = new Uint8Array(tableToIPC(table, "file"));