fix: validate lakehouse upload files before streaming the payload

cli/src/commands/lakehouse.ts

@@ -1,4 +1,5 @@
 import type { ArgsDef } from "citty";
+import { statSync } from "node:fs";
 import { CliError } from "@/lib/errors.ts";
 import { booleanArg, enumArg, optionalStringArg, stringArg } from "@/lib/operation-codec.ts";
 import { httpEffect, progressPlan, scopedPlan } from "@/lib/operation-effect.ts";
@@ -32,6 +33,21 @@ import {
 const LAKEHOUSE_FILE_FORMAT_OPTIONS = ["csv", "json", "parquet"] as const;
 const UPLOAD_MODE_OPTIONS = ["create", "append", "overwrite"] as const;
 
+function getUploadFileSizeBytes(filePath: string): number {
+  try {
+    const fileStat = statSync(filePath);
+    if (!fileStat.isFile()) {
+      throw new CliError(`File not found: ${filePath}`);
+    }
+    return fileStat.size;
+  } catch (error) {
+    if (error instanceof CliError) {
+      throw error;
+    }
+    throw new CliError(`File not found: ${filePath}`);
+  }
+}
+
 const queryRunArgs = {
   statement: { type: "string", description: "SQL statement to run", required: true },
   format: {
@@ -294,12 +310,7 @@ export const uploadCommand = defineOperationCommand({
   async parse({ args }) {
     const mode = enumArg(args, "mode", UPLOAD_MODE_OPTIONS);
     const filePath = stringArg(args, "file");
-
-    try {
-      await Bun.file(filePath).arrayBuffer();
-    } catch {
-      throw new CliError(`File not found: ${filePath}`);
-    }
+    const fileSizeBytes = getUploadFileSizeBytes(filePath);
 
     const readTimeoutMs = parseRequestReadTimeoutMs(args);
     return {
@@ -308,6 +319,7 @@ export const uploadCommand = defineOperationCommand({
       table: stringArg(args, "table"),
       mode,
       filePath,
+      fileSizeBytes,
       contentType: parseLakehouseFileContentType(optionalStringArg(args, "format")),
       httpOptions: readTimeoutMs !== undefined ? { readTimeoutMs } : undefined,
     };
@@ -320,6 +332,7 @@ export const uploadCommand = defineOperationCommand({
         table: input.table,
         mode: input.mode,
         filePath: input.filePath,
+        fileSizeBytes: input.fileSizeBytes,
         contentType: input.contentType,
         httpOptions: input.httpOptions,
       });
@@ -372,12 +385,7 @@ export const upsertCommand = defineOperationCommand({
   },
   async parse({ args }) {
     const filePath = stringArg(args, "file");
-
-    try {
-      await Bun.file(filePath).arrayBuffer();
-    } catch {
-      throw new CliError(`File not found: ${filePath}`);
-    }
+    const fileSizeBytes = getUploadFileSizeBytes(filePath);
 
     const readTimeoutMs = parseRequestReadTimeoutMs(args);
     return {
@@ -386,6 +394,7 @@ export const upsertCommand = defineOperationCommand({
       table: stringArg(args, "table"),
       primaryKey: stringArg(args, "primary-key"),
       filePath,
+      fileSizeBytes,
       contentType: parseLakehouseFileContentType(optionalStringArg(args, "format")),
       httpOptions: readTimeoutMs !== undefined ? { readTimeoutMs } : undefined,
     };
@@ -398,6 +407,7 @@ export const upsertCommand = defineOperationCommand({
         table: input.table,
         primaryKey: input.primaryKey,
         filePath: input.filePath,
+        fileSizeBytes: input.fileSizeBytes,
         contentType: input.contentType,
         httpOptions: input.httpOptions,
       });

cli/src/lib/lakehouse-transport.ts

@@ -20,6 +20,7 @@ export type LakehouseUploadRequestInput = {
   table: string;
   mode: string;
   filePath: string;
+  fileSizeBytes: number;
   contentType?: string;
   httpOptions?: Partial<HttpSendOptions>;
 };
@@ -30,6 +31,7 @@ export type LakehouseUpsertRequestInput = {
   table: string;
   primaryKey: string;
   filePath: string;
+  fileSizeBytes: number;
   contentType?: string;
   httpOptions?: Partial<HttpSendOptions>;
 };
@@ -102,14 +104,17 @@ function createLakehouseFileRequest(
   endpoint: string,
 ): LakehouseUploadRequestScope {
   const file = Bun.file(input.filePath);
-  const fileSizeBytes = file.size;
   let body: Blob | ReadableStream = input.contentType ? file : file.stream();
-  let uploadProgress = createUploadProgressReporter(fileSizeBytes);
+  let uploadProgress = createUploadProgressReporter(input.fileSizeBytes);
 
-  if (shouldShowProgress() && fileSizeBytes > 0) {
-    body = wrapStreamWithByteProgress(file.stream(), fileSizeBytes, (sentBytes, totalBytes) => {
-      uploadProgress.report(sentBytes, totalBytes);
-    });
+  if (shouldShowProgress() && input.fileSizeBytes > 0) {
+    body = wrapStreamWithByteProgress(
+      file.stream(),
+      input.fileSizeBytes,
+      (sentBytes, totalBytes) => {
+        uploadProgress.report(sentBytes, totalBytes);
+      },
+    );
   } else {
     uploadProgress = createUploadProgressReporter(0);
   }

cli/tests/api.test.ts

@@ -15,6 +15,7 @@ import { OPENAPI_OPERATIONS } from "@/generated/openapi-operations.ts";
 import { setCliContext } from "@/context.ts";
 import { buildCompletionSpec, flattenTopLevelNames } from "@/lib/completion-spec.ts";
 import { createCliRuntime, getCliRuntime, setCliRuntime } from "@/lib/runtime.ts";
+import { runCommandWithTestRuntime } from "@tests/cli-test-runtime.ts";
 
 function createCaptureSink(json: boolean) {
   const stdout: string[] = [];
@@ -214,9 +215,7 @@ describe("api", () => {
         ]),
       );
 
-      await runCommand(buildMainCommand(), {
-        rawArgs: normalizeApiInvocatorRawArgs(["api", "/whoami"]),
-      });
+      await runCommandWithTestRuntime(normalizeApiInvocatorRawArgs(["api", "/whoami"]));
 
       const logContent = readFileSync(logFile, "utf8");
       expect(logContent).toContain("/rest/v1/whoami");
@@ -235,9 +234,7 @@ describe("api", () => {
         ]),
       );
 
-      await runCommand(buildMainCommand(), {
-        rawArgs: ["api", "POST", "/service_accounts", "-f", "label=CI Bot"],
-      });
+      await runCommandWithTestRuntime(["api", "POST", "/service_accounts", "-f", "label=CI Bot"]);
 
       const logContent = readFileSync(logFile, "utf8");
       const payloadLines = logContent.match(/^PAYLOAD=.*$/gm) ?? [];

cli/tests/cli-test-runtime.ts

@@ -0,0 +1,22 @@
+import { runCommand } from "citty";
+import { buildMainCommand } from "@/cli.ts";
+import type { CliContext } from "@/context.ts";
+import { createCliRuntime, runWithCliRuntime, type CliRuntime } from "@/lib/runtime.ts";
+
+export function createCliTestRuntime(
+  context: CliContext = { debug: false, json: true, agent: false },
+): CliRuntime {
+  const runtime = createCliRuntime(context);
+  runtime.output.writeStderr = () => {};
+  runtime.output.writeJson = () => {};
+  runtime.output.writeRaw = () => {};
+  runtime.output.writeHuman = () => {};
+  runtime.output.writeMetadata = () => {};
+  return runtime;
+}
+
+export async function runCommandWithTestRuntime(rawArgs: string[]): Promise<void> {
+  await runWithCliRuntime(createCliTestRuntime(), () =>
+    runCommand(buildMainCommand(), { rawArgs }),
+  );
+}

cli/tests/lakehouse.test.ts

@@ -1,9 +1,7 @@
 import { describe, expect, test, beforeEach, afterEach } from "bun:test";
-import { runCommand } from "citty";
-import { mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs";
+import { mkdirSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs";
 import { join } from "node:path";
 import { tmpdir } from "node:os";
-import { buildMainCommand } from "@/cli.ts";
 import { setCliContext } from "@/context.ts";
 import { ParseError } from "@/lib/errors.ts";
 import {
@@ -29,12 +27,8 @@ import {
   runOperationPlan,
 } from "@/lib/operation-effect.ts";
 import type { OperationContext } from "@/lib/operation-command.ts";
-import {
-  createCliRuntime,
-  getCliRuntime,
-  refreshCliRuntimeContext,
-  setCliRuntime,
-} from "@/lib/runtime.ts";
+import { getCliRuntime, refreshCliRuntimeContext } from "@/lib/runtime.ts";
+import { runCommandWithTestRuntime } from "@tests/cli-test-runtime.ts";
 import {
   lakehouseAppendOperation,
   lakehouseAppendTaskOperation,
@@ -331,21 +325,8 @@ describe("lakehouse request construction", () => {
       ]),
     );
 
-    const previousRuntime = getCliRuntime();
-    const runtime = createCliRuntime({ debug: false, json: true, agent: false });
-    runtime.output.writeJson = () => {};
-    runtime.output.writeRaw = () => {};
-    runtime.output.writeHuman = () => {};
-    setCliRuntime(runtime);
-
-    try {
-      await runCommand(buildMainCommand(), { rawArgs: ["query", "show", queryId] });
-      await runCommand(buildMainCommand(), {
-        rawArgs: ["query", "cancel", queryId, "--session-id", "session-1"],
-      });
-    } finally {
-      setCliRuntime(previousRuntime);
-    }
+    await runCommandWithTestRuntime(["query", "show", queryId]);
+    await runCommandWithTestRuntime(["query", "cancel", queryId, "--session-id", "session-1"]);
 
     const logContent = readFileSync(logFile, "utf8");
     expect(logContent).toContain("METHOD=GET");
@@ -430,6 +411,7 @@ describe("lakehouse request construction", () => {
       table: "users",
       mode: "overwrite",
       filePath: uploadFile,
+      fileSizeBytes: 15,
       contentType: "text/csv",
     });
     try {
@@ -456,6 +438,7 @@ describe("lakehouse request construction", () => {
       table: "users",
       mode: "overwrite",
       filePath: uploadFile,
+      fileSizeBytes: 15,
       contentType: "text/csv",
     });
     try {
@@ -476,6 +459,7 @@ describe("lakehouse request construction", () => {
       table: "users",
       mode: "overwrite",
       filePath: uploadFile,
+      fileSizeBytes: 15,
     });
     try {
       expect(uploadScope.request.contentType).toBeUndefined();
@@ -505,6 +489,7 @@ describe("lakehouse request construction", () => {
       table: "users",
       primaryKey: "id",
       filePath: uploadFile,
+      fileSizeBytes: 15,
       contentType: "text/csv",
     });
     try {
@@ -520,6 +505,91 @@ describe("lakehouse request construction", () => {
     expect(logContent).toMatch(/PAYLOAD=@(?:blob|stream)/);
   });
 
+  test("upload command validates file metadata without buffering the payload", async () => {
+    const uploadFile = join(testHome, "command-data.csv");
+    writeFileSync(uploadFile, "id,name\n1,Alice", "utf8");
+    writeFileSync(
+      mockFile,
+      JSON.stringify([
+        {
+          urlPattern: "/upload",
+          method: "POST",
+          body: '{"ok":true}',
+        },
+      ]),
+    );
+
+    await runCommandWithTestRuntime([
+      "upload",
+      "--catalog",
+      "memory",
+      "--schema",
+      "main",
+      "--table",
+      "users",
+      "--format",
+      "csv",
+      "--mode",
+      "overwrite",
+      "--file",
+      uploadFile,
+    ]);
+
+    const logContent = readFileSync(logFile, "utf8");
+    expect(logContent).toContain("URL=https://example.com/upload?");
+    expect(logContent).toContain("PAYLOAD=@blob");
+    expect(logContent).not.toContain("id,name");
+  });
+
+  test("upload command rejects missing files and directories", async () => {
+    const directoryPath = join(testHome, "upload-directory");
+    mkdirSync(directoryPath);
+
+    try {
+      await runCommandWithTestRuntime([
+        "upload",
+        "--catalog",
+        "memory",
+        "--schema",
+        "main",
+        "--table",
+        "users",
+        "--format",
+        "csv",
+        "--mode",
+        "overwrite",
+        "--file",
+        join(testHome, "missing.csv"),
+      ]);
+      throw new Error("expected missing file failure");
+    } catch (error) {
+      expect(error).toBeInstanceOf(Error);
+      expect((error as Error).message).toContain("File not found");
+    }
+
+    try {
+      await runCommandWithTestRuntime([
+        "upload",
+        "--catalog",
+        "memory",
+        "--schema",
+        "main",
+        "--table",
+        "users",
+        "--format",
+        "csv",
+        "--mode",
+        "overwrite",
+        "--file",
+        directoryPath,
+      ]);
+      throw new Error("expected directory failure");
+    } catch (error) {
+      expect(error).toBeInstanceOf(Error);
+      expect((error as Error).message).toContain("File not found");
+    }
+  });
+
   test("cancel URL-encodes query id in path", async () => {
     const queryId = "query/id+special";
     const encodedQueryId = encodeURIComponent(queryId);