Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion README.md
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down
134 changes: 134 additions & 0 deletions packages/server/workers/lib/__tests__/arrow.test.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,134 @@
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<string, unknown>;
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("round-trips null/undefined values as null", () => {
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);
const rows = decoded.toArray().map((row: unknown) => {
const values = row as Record<string, unknown>;
return { a: values.a, b: values.b };
});

expect(rows).toEqual([
{ a: "x", b: 1 },
{ a: null, b: null },
{ a: null, b: null },
]);
});

test("only allocates validity bitmaps for columns with nulls", () => {
const table = recordsToTable([
{ nullable: null, complete: "x" },
{ nullable: "value", complete: "y" },
]);
const batch = table.batches[0];
const nullable = batch.getChild("nullable")?.data[0];
const complete = batch.getChild("complete")?.data[0];

expect(nullable?.nullCount).toBe(1);
expect(nullable?.nullBitmap.byteLength).toBe(8);
expect(complete?.nullCount).toBe(0);
expect(complete?.nullBitmap.byteLength).toBe(0);
});

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);
});
});
126 changes: 121 additions & 5 deletions packages/server/workers/lib/arrow.ts
Original file line number Diff line number Diff line change
@@ -1,8 +1,124 @@
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 };

function buildValidityBitmap(values: RecordValue[]): {
nullCount: number;
nullBitmap?: Uint8Array;
} {
let nullCount = 0;
for (const value of values) {
if (value == null) {
nullCount++;
}
}
if (nullCount === 0) {
return { nullCount };
}

const nullBitmap = new Uint8Array(Math.ceil(values.length / 64) * 8);
for (let i = 0; i < values.length; i++) {
if (values[i] != null) {
nullBitmap[i >> 3] |= 1 << (i & 7);
}
}
return { nullCount, nullBitmap };
}

// 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<Utf8> {
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;
const { nullCount, nullBitmap } = buildValidityBitmap(values);
return makeData({
type: new Utf8(),
length: values.length,
nullCount,
nullBitmap,
valueOffsets,
data,
});
}

function buildFloat64Data(values: (number | null | undefined)[]): Data<Float64> {
const buf = new Float64Array(values.length);
for (let i = 0; i < values.length; i++) {
buf[i] = values[i] ?? 0;
}
const { nullCount, nullBitmap } = buildValidityBitmap(values);
return makeData({
type: new Float64(),
length: values.length,
nullCount,
nullBitmap,
data: buf,
});
}

/**
* Converts homogeneous records to an Arrow Table without Builder codegen and
* preserves null or undefined values as Arrow nulls. Column type is inferred
* from the first non-null sample: `number` becomes 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,
Expand All @@ -27,10 +143,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,
Expand All @@ -46,8 +162,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"));
Expand Down