refactor: return Result from parquet merge and read instead of throwing

src/benchmarks/terminal-bench/ori-parquet.test.ts

@@ -21,6 +21,7 @@ import {
 import type { Sample } from "../../harness/core";
 import { initialTaskState } from "../../harness/core";
 import { Solver } from "../../harness/solver";
+import { assertSuccess } from "../../internal/testing";
 import {
   asyncBufferFromBytes,
   readResultRows,
@@ -175,7 +176,8 @@ async function runOriSampleToParquetRow() {
   const rows = await readResultRows(
     asyncBufferFromBytes(new Uint8Array(buffer))
   );
-  const row = rows[0];
+  assertSuccess(rows);
+  const row = rows.success[0];
   if (row === undefined) {
     throw new Error("parquet contained no rows");
   }

src/benchmarks/toolcall-schema-fuzz/cli/report.ts

@@ -1,6 +1,12 @@
 import { readFile } from "node:fs/promises";
 
-import { asyncBufferFromBytes, readResultRows } from "../../../results/parquet";
+import * as Result from "effect/Result";
+
+import {
+  asyncBufferFromBytes,
+  parquetErrorMessage,
+  readResultRows,
+} from "../../../results/parquet";
 import { stabilityReport } from "../stability";
 
 const paths = process.argv.slice(2);
@@ -9,11 +15,17 @@ if (paths.length === 0) {
   process.exit(1);
 }
 
-const rowLists = await Promise.all(
-  paths.map(async (path) =>
-    readResultRows(asyncBufferFromBytes(await readFile(path)))
+const rowLists = Result.all(
+  await Promise.all(
+    paths.map(async (path) =>
+      readResultRows(asyncBufferFromBytes(await readFile(path)))
+    )
   )
 );
+if (Result.isFailure(rowLists)) {
+  process.stderr.write(`${parquetErrorMessage(rowLists.failure)}\n`);
+  process.exit(1);
+}
 process.stdout.write(
-  `${JSON.stringify(stabilityReport(rowLists.flat()), null, 2)}\n`
+  `${JSON.stringify(stabilityReport(rowLists.success.flat()), null, 2)}\n`
 );

src/results/parquet-stream-writer.test.ts

@@ -3,12 +3,14 @@ import { describe, expect, it } from "bun:test";
 import { ByteWriter } from "hyparquet-writer";
 
 import { ScoreValue } from "../harness/core";
+import { assertSuccess } from "../internal/testing";
 import {
   asyncBufferFromBytes,
   mergeResultFilesToParquet,
   readResultRows,
   runResultToParquet,
 } from "./parquet";
+import type { BenchmarkResultRow } from "./parquet-schema";
 import { parquetStreamWriter } from "./parquet-stream-writer";
 
 const META = {
@@ -18,7 +20,9 @@ const META = {
   createdAt: "2026-01-01T00:00:00.000Z",
 } as const;
 
-function resultRows(sampleCount: number): ReturnType<typeof readResultRows> {
+async function resultRows(
+  sampleCount: number
+): Promise<readonly BenchmarkResultRow[]> {
   const sampleScores = Array.from({ length: sampleCount }, (_, index) => ({
     sampleId: `s${index}`,
     epoch: 0,
@@ -48,7 +52,9 @@ function resultRows(sampleCount: number): ReturnType<typeof readResultRows> {
     },
     meta: META,
   });
-  return readResultRows(asyncBufferFromBytes(bytes));
+  const rows = await readResultRows(asyncBufferFromBytes(bytes));
+  assertSuccess(rows);
+  return rows.success;
 }
 
 describe("parquetStreamWriter", () => {

src/results/parquet.test.ts

@@ -3,7 +3,7 @@ import assert from "node:assert";
 
 import type { AsyncBuffer } from "hyparquet";
 import { parquetMetadata } from "hyparquet";
-import { ByteWriter } from "hyparquet-writer";
+import { ByteWriter, parquetWriteBuffer } from "hyparquet-writer";
 
 import type { ModelMessage, ResponseItem } from "../harness/core";
 import { MessageRole, ScoreValue } from "../harness/core";
@@ -15,6 +15,8 @@ import { AtifTrajectorySchema } from "./atif-schema";
 import type { MergeResultFilesInput, MergedResultFiles } from "./parquet";
 import {
   mergeResultFilesToParquet,
+  asyncBufferFromBytes,
+  parquetErrorMessage,
   readResultRows,
   runResultToParquet,
   rowScoreToNumber,
@@ -27,6 +29,7 @@ import {
   BenchmarkResultRowSchema,
   ScorerTrajectorySchema,
 } from "./parquet-schema";
+import { parquetStreamWriter } from "./parquet-stream-writer";
 
 const USAGE = {
   inputTokens: 100,
@@ -103,9 +106,11 @@ function bufferToArrayBuffer(buf: Buffer): ArrayBuffer {
   return ab;
 }
 
-function readRows(buf: Buffer): Promise<readonly BenchmarkResultRow[]> {
+async function readRows(buf: Buffer): Promise<readonly BenchmarkResultRow[]> {
   const file = asyncBufferFromArrayBuffer(bufferToArrayBuffer(buf));
-  return readResultRows(file);
+  const rows = await readResultRows(file);
+  assertSuccess(rows);
+  return rows.success;
 }
 
 async function mergeToBytes(
@@ -119,7 +124,8 @@ async function mergeToBytes(
     writer,
     ...(runLevelScores !== undefined && { runLevelScores }),
   });
-  return { bytes: Buffer.from(writer.getBuffer()), merged };
+  assertSuccess(merged);
+  return { bytes: Buffer.from(writer.getBuffer()), merged: merged.success };
 }
 
 async function mergedRowsOf(
@@ -1203,18 +1209,18 @@ describe("mergeResultFilesToParquet", () => {
     );
     let calls = 0;
 
-    await expect(
-      mergeResultFilesToParquet({
-        files: [
-          () => {
-            calls += 1;
-            return Promise.resolve(calls === 1 ? rows : rows.slice(1));
-          },
-        ],
-        meta: { task: META.task, model: META.model, createdAt: META.createdAt },
-        writer: new ByteWriter(),
-      })
-    ).rejects.toThrow("Result file 0 changed between merge passes");
+    const merged = await mergeResultFilesToParquet({
+      files: [
+        () => {
+          calls += 1;
+          return Promise.resolve(calls === 1 ? rows : rows.slice(1));
+        },
+      ],
+      meta: { task: META.task, model: META.model, createdAt: META.createdAt },
+      writer: new ByteWriter(),
+    });
+    assertFailure(merged);
+    expect(merged.failure).toEqual({ kind: "result-file-changed", index: 0 });
   });
   it("rejects a file whose scores change between passes without changing length", async () => {
     const rows = await readRows(
@@ -1229,18 +1235,18 @@ describe("mergeResultFilesToParquet", () => {
     }));
     let calls = 0;
 
-    await expect(
-      mergeResultFilesToParquet({
-        files: [
-          () => {
-            calls += 1;
-            return Promise.resolve(calls === 1 ? rows : flipped);
-          },
-        ],
-        meta: { task: META.task, model: META.model, createdAt: META.createdAt },
-        writer: new ByteWriter(),
-      })
-    ).rejects.toThrow("Result file 0 changed between merge passes");
+    const merged = await mergeResultFilesToParquet({
+      files: [
+        () => {
+          calls += 1;
+          return Promise.resolve(calls === 1 ? rows : flipped);
+        },
+      ],
+      meta: { task: META.task, model: META.model, createdAt: META.createdAt },
+      writer: new ByteWriter(),
+    });
+    assertFailure(merged);
+    expect(merged.failure).toEqual({ kind: "result-file-changed", index: 0 });
   });
   it("rejects a file whose run-level usage changes between passes", async () => {
     const rows = await readRows(
@@ -1252,18 +1258,128 @@ describe("mergeResultFilesToParquet", () => {
     }));
     let calls = 0;
 
-    await expect(
-      mergeResultFilesToParquet({
-        files: [
-          () => {
-            calls += 1;
-            return Promise.resolve(calls === 1 ? rows : changed);
-          },
-        ],
-        meta: { task: META.task, model: META.model, createdAt: META.createdAt },
-        writer: new ByteWriter(),
-      })
-    ).rejects.toThrow("Result file 0 changed between merge passes");
+    const merged = await mergeResultFilesToParquet({
+      files: [
+        () => {
+          calls += 1;
+          return Promise.resolve(calls === 1 ? rows : changed);
+        },
+      ],
+      meta: { task: META.task, model: META.model, createdAt: META.createdAt },
+      writer: new ByteWriter(),
+    });
+    assertFailure(merged);
+    expect(merged.failure).toEqual({ kind: "result-file-changed", index: 0 });
+  });
+});
+
+describe("ParquetError", () => {
+  const meta = {
+    task: META.task,
+    model: META.model,
+    createdAt: META.createdAt,
+  };
+  const scored: MergeResultFilesInput["runLevelScores"] = () => [];
+
+  it("returns result-file-read-failed when a reader rejects", async () => {
+    const cause = new Error("disk gone");
+    const merged = await mergeResultFilesToParquet({
+      files: [() => Promise.reject(cause)],
+      meta,
+      writer: new ByteWriter(),
+    });
+    assertFailure(merged);
+    expect(merged.failure).toEqual({
+      kind: "result-file-read-failed",
+      index: 0,
+      cause,
+    });
+    expect(parquetErrorMessage(merged.failure)).toBe("disk gone");
+  });
+  it("returns parquet-write-failed when the writer sink rejects", async () => {
+    const rows = await readRows(
+      runResultToParquet({ result: RESULT, meta: META })
+    );
+    const cause = new Error("upload closed");
+    const merged = await mergeResultFilesToParquet({
+      files: [() => Promise.resolve(rows)],
+      meta,
+      writer: parquetStreamWriter(() => Promise.reject(cause), 16),
+    });
+    assertFailure(merged);
+    expect(merged.failure).toEqual({ kind: "parquet-write-failed", cause });
+    expect(parquetErrorMessage(merged.failure)).toBe("upload closed");
+  });
+  it("returns parquet-decode-failed for bytes that are not parquet", async () => {
+    const read = await readResultRows(
+      asyncBufferFromBytes(new TextEncoder().encode('{"trial_results":[]}'))
+    );
+    assertFailure(read);
+    expect(read.failure.kind).toBe("parquet-decode-failed");
+    expect(parquetErrorMessage(read.failure)).toBe(
+      "parquet file invalid (footer != PAR1)"
+    );
+  });
+  it("returns invalid-result-row for parquet without the result columns", async () => {
+    const bytes = parquetWriteBuffer({
+      columnData: [{ name: "sample_id", data: ["s0"] }],
+    });
+    const read = await readResultRows(
+      asyncBufferFromBytes(new Uint8Array(bytes))
+    );
+    assertFailure(read);
+    expect(read.failure.kind).toBe("invalid-result-row");
+    expect(parquetErrorMessage(read.failure)).toStartWith(
+      "Invalid benchmark result row: "
+    );
+  });
+  it("returns invalid-result-column-json when a scored JSON column is not JSON", async () => {
+    const rows = await readRows(
+      runResultToParquet({ result: RESULT, meta: META })
+    );
+    const broken = rows.map((row) => ({ ...row, metadata: "nope" }));
+    const merged = await mergeResultFilesToParquet({
+      files: [() => Promise.resolve(broken)],
+      meta,
+      writer: new ByteWriter(),
+      runLevelScores: scored,
+    });
+    assertFailure(merged);
+    expect(merged.failure.kind).toBe("invalid-result-column-json");
+    expect(parquetErrorMessage(merged.failure)).toStartWith("JSON Parse error");
+  });
+  it("returns invalid-result-column when a scored JSON column fails its schema", async () => {
+    const rows = await readRows(
+      runResultToParquet({ result: RESULT, meta: META })
+    );
+    const broken = rows.map((row) => ({
+      ...row,
+      scorer_trajectory: '{"kind":"unknown"}',
+    }));
+    const merged = await mergeResultFilesToParquet({
+      files: [() => Promise.resolve(broken)],
+      meta,
+      writer: new ByteWriter(),
+      runLevelScores: scored,
+    });
+    assertFailure(merged);
+    expect(merged.failure.kind).toBe("invalid-result-column");
+    expect(parquetErrorMessage(merged.failure)).toStartWith(
+      "Invalid result column: "
+    );
+  });
+  it("formats result-file-changed with the file index", () => {
+    expect(parquetErrorMessage({ kind: "result-file-changed", index: 3 })).toBe(
+      "Result file 3 changed between merge passes"
+    );
+  });
+  it("formats a non-Error cause the way unknownErrorToString does", () => {
+    expect(
+      parquetErrorMessage({ kind: "parquet-decode-failed", cause: "boom" })
+    ).toBe("boom");
+    expect(
+      parquetErrorMessage({ kind: "parquet-write-failed", cause: { code: 7 } })
+    ).toBe('{"code":7}');
   });
 });