refactor: proposal — stash inline Arrow IPC, serve via /arrow-result

DRAFT proposal. Replaces the `arrow_inline` SSE message type from #256
with an out-of-band delivery mechanism. Motivation: remove the SSE
event-size cap as a constraint on inline Arrow result size.

Architecture
------------
ARROW_STREAM responses are now delivered uniformly:
1. The connector returns the base64 Arrow IPC attachment in
   `result.attachment` for INLINE responses (unchanged from #256).
2. The analytics route base64-decodes it once, stashes the resulting
   buffer in `InlineArrowStash` (bounded LRU + TTL), and emits the
   same `{ type: "arrow", statement_id }` SSE message that
   EXTERNAL_LINKS already uses — with a synthetic id prefixed
   `inline-`.
3. The client (unchanged from main) calls `/arrow-result/:jobId` for
   any `type: "arrow"` message.
4. The route handler checks the stash first; if the id has the
   `inline-` prefix and is still present, it serves the bytes from
   memory. Otherwise it falls through to the existing warehouse
   fetch path for real EXTERNAL_LINKS statement_ids.

The `arrow_inline` SSE message type, the client-side base64 decode
helper, and the SSE buffer-size bumps from #256 all go away.

Reasoning
---------
SSE was designed for streams of small control messages. Pushing
multi-MB Arrow IPC bytes through it has two unavoidable costs:
- single-event memory ceiling on both server and client,
- proxy/load-balancer compatibility issues for large events.

Bumping the SSE buffer (5b in the design discussion) raises both
caps but doesn't fix the architectural mismatch. This PR moves bulk
bytes to HTTP, where they belong:
- HTTP/2 streaming, gzip, and browser background-fetch all work
  naturally.
- SSE buffer can stay at the conservative 1 MiB default.
- The wire protocol unifies INLINE and EXTERNAL_LINKS — the client
  has a single code path for ARROW_STREAM data.
- Inline result size cap goes back up to the Databricks API limit
  (25 MiB) instead of being constrained by SSE.

Tradeoffs
---------
- Server holds IPC buffers in memory until the client fetches them
  (one-shot `take()` removes on read; passive TTL evicts otherwise).
  For 100 entries × 25 MiB worst case, that's 2.5 GiB peak — but
  steady state is much smaller because entries are removed as soon
  as the client fetches.
- Single-process only. A multi-server deployment would need a
  shared store (Redis or equivalent) keyed on the synthetic id. The
  stash interface is small enough to swap implementations.
- One-shot reads mean the client cannot retry a failed fetch.
  Acceptable for our use case (warehouse query returned data, the
  hook either succeeds in fetching or surfaces a generic error).

Status
------
DRAFT: opened to discuss the architectural direction before landing.
#256 ships with the simpler 5b SSE-buffer-bump approach (8 MiB) which
is sufficient for the realistic analytics range. This proposal is the
cleaner long-term solution if the team wants to invest in the
server-side state.

Co-authored-by: Isaac
Signed-off-by: James Broadhead <jamesbroadhead@gmail.com>

Author: jamesbroadhead

packages/appkit-ui/src/js/sse/connect-sse.ts

@@ -18,10 +18,7 @@ export async function connectSSE<Payload = unknown>(
     lastEventId: initialLastEventId = null,
     retryDelay = 2000,
     maxRetries = 3,
-    // 8 MiB — sized to receive inline Arrow IPC attachments from
-    // ARROW_STREAM analytics responses; matches the server's stream
-    // `maxEventSize`. Most events are well under 1 MiB in practice.
-    maxBufferSize = 8 * 1024 * 1024,
+    maxBufferSize = 1024 * 1024, // 1MB
     timeout = 300000, // 5 minutes
     onError,
   } = options;

packages/appkit-ui/src/react/hooks/__tests__/use-analytics-query.test.ts

@@ -30,93 +30,65 @@ describe("useAnalyticsQuery", () => {
     lastConnectArgs = null;
   });
 
-  test("decodes arrow_inline base64 attachment via ArrowClient.processArrowBuffer", async () => {
-    const fakeTable = { numRows: 1, schema: { fields: [] } };
+  test("fetches Arrow IPC via /arrow-result for type:arrow (covers both inline-stash and external-link paths)", async () => {
+    const fakeTable = { numRows: 0, schema: { fields: [] } };
+    mockFetchArrow.mockResolvedValueOnce(new Uint8Array([1, 2, 3]));
     mockProcessArrowBuffer.mockResolvedValueOnce(fakeTable);
 
-    // 'AQID' decodes to bytes [1, 2, 3].
-    const base64 = "AQID";
-
     const { result } = renderHook(() =>
       useAnalyticsQuery("q", null, { format: "ARROW_STREAM" }),
     );
 
-    // Drive the SSE onMessage handler with an arrow_inline payload.
+    // Server emits the same {type:"arrow", statement_id} shape regardless of
+    // whether the bytes came from the warehouse (EXTERNAL_LINKS) or were
+    // stashed locally (INLINE).
     await lastConnectArgs.onMessage({
-      data: JSON.stringify({ type: "arrow_inline", attachment: base64 }),
+      data: JSON.stringify({ type: "arrow", statement_id: "inline-abc" }),
     });
 
     await waitFor(() => {
       expect(result.current.data).toBe(fakeTable);
     });
-
-    expect(mockProcessArrowBuffer).toHaveBeenCalledTimes(1);
-    const passedBuffer = mockProcessArrowBuffer.mock.calls[0][0] as Uint8Array;
-    expect(passedBuffer).toBeInstanceOf(Uint8Array);
-    expect(Array.from(passedBuffer)).toEqual([1, 2, 3]);
-    // Inline path must NOT trigger a network fetch.
-    expect(mockFetchArrow).not.toHaveBeenCalled();
+    expect(mockFetchArrow).toHaveBeenCalledTimes(1);
+    expect(mockFetchArrow.mock.calls[0][0]).toBe(
+      "/api/analytics/arrow-result/inline-abc",
+    );
   });
 
-  test("surfaces an error when arrow_inline decode fails", async () => {
-    mockProcessArrowBuffer.mockRejectedValueOnce(new Error("bad ipc"));
-
+  test("still handles type:result rows for JSON_ARRAY", async () => {
     const { result } = renderHook(() =>
-      useAnalyticsQuery("q", null, { format: "ARROW_STREAM" }),
+      useAnalyticsQuery("q", null, { format: "JSON_ARRAY" }),
     );
 
     await lastConnectArgs.onMessage({
-      data: JSON.stringify({ type: "arrow_inline", attachment: "AQID" }),
+      data: JSON.stringify({
+        type: "result",
+        data: [{ id: 1 }, { id: 2 }],
+      }),
     });
 
     await waitFor(() => {
-      expect(result.current.error).toBe(
-        "Unable to load data, please try again",
-      );
+      expect(result.current.data).toEqual([{ id: 1 }, { id: 2 }]);
     });
-    expect(result.current.loading).toBe(false);
+    expect(mockProcessArrowBuffer).not.toHaveBeenCalled();
   });
 
-  test("rejects arrow_inline with missing/empty/non-string attachment without crashing atob", async () => {
-    const cases: Array<unknown> = [undefined, null, "", 123, { foo: "bar" }];
-
-    for (const attachment of cases) {
-      mockProcessArrowBuffer.mockClear();
-      const { result, unmount } = renderHook(() =>
-        useAnalyticsQuery("q", null, { format: "ARROW_STREAM" }),
-      );
-
-      await lastConnectArgs.onMessage({
-        data: JSON.stringify({ type: "arrow_inline", attachment }),
-      });
-
-      await waitFor(() => {
-        expect(result.current.error).toBe(
-          "Unable to load data, please try again",
-        );
-      });
-      // Critically: must NOT call processArrowBuffer (or atob) on the bad input.
-      expect(mockProcessArrowBuffer).not.toHaveBeenCalled();
-
-      unmount();
-    }
-  });
+  test("surfaces an error when /arrow-result fetch fails", async () => {
+    mockFetchArrow.mockRejectedValueOnce(new Error("HTTP 404"));
 
-  test("still handles type:result rows for JSON_ARRAY", async () => {
     const { result } = renderHook(() =>
-      useAnalyticsQuery("q", null, { format: "JSON_ARRAY" }),
+      useAnalyticsQuery("q", null, { format: "ARROW_STREAM" }),
     );
 
     await lastConnectArgs.onMessage({
-      data: JSON.stringify({
-        type: "result",
-        data: [{ id: 1 }, { id: 2 }],
-      }),
+      data: JSON.stringify({ type: "arrow", statement_id: "inline-stale" }),
     });
 
     await waitFor(() => {
-      expect(result.current.data).toEqual([{ id: 1 }, { id: 2 }]);
+      expect(result.current.error).toBe(
+        "Unable to load data, please try again",
+      );
     });
-    expect(mockProcessArrowBuffer).not.toHaveBeenCalled();
+    expect(result.current.loading).toBe(false);
   });
 });

packages/appkit-ui/src/react/hooks/use-analytics-query.ts

@@ -22,16 +22,6 @@ function getArrowStreamUrl(id: string) {
   return `/api/analytics/arrow-result/${id}`;
 }
 
-/** Decode a base64 string into a Uint8Array suitable for Arrow IPC parsing. */
-function decodeBase64(b64: string): Uint8Array {
-  const binary = atob(b64);
-  const bytes = new Uint8Array(binary.length);
-  for (let i = 0; i < binary.length; i++) {
-    bytes[i] = binary.charCodeAt(i);
-  }
-  return bytes;
-}
-
 /**
  * Subscribe to an analytics query over SSE and returns its latest result.
  * Integration hook between client and analytics plugin.
@@ -139,7 +129,9 @@ export function useAnalyticsQuery<
             return;
           }
 
-          // success - Arrow format (external links: fetch from server)
+          // success - Arrow format. The server delivers Arrow IPC bytes via
+          // /arrow-result/:jobId for both INLINE (stashed server-side) and
+          // EXTERNAL_LINKS (forwarded from the warehouse) responses.
           if (parsed.type === "arrow") {
             try {
               const arrowData = await ArrowClient.fetchArrow(
@@ -161,36 +153,6 @@ export function useAnalyticsQuery<
             }
           }
 
-          // success - Arrow format (inline: decode base64 IPC payload locally)
-          if (parsed.type === "arrow_inline") {
-            if (
-              typeof parsed.attachment !== "string" ||
-              parsed.attachment.length === 0
-            ) {
-              console.error(
-                "[useAnalyticsQuery] arrow_inline message missing attachment",
-              );
-              setLoading(false);
-              setError("Unable to load data, please try again");
-              return;
-            }
-            try {
-              const buffer = decodeBase64(parsed.attachment);
-              const table = await ArrowClient.processArrowBuffer(buffer);
-              setLoading(false);
-              setData(table as ResultType);
-              return;
-            } catch (error) {
-              console.error(
-                "[useAnalyticsQuery] Failed to decode inline Arrow data",
-                error,
-              );
-              setLoading(false);
-              setError("Unable to load data, please try again");
-              return;
-            }
-          }
-
           // error
           if (parsed.type === "error" || parsed.error || parsed.code) {
             const errorMsg =

packages/appkit/src/connectors/sql-warehouse/client.ts

@@ -27,13 +27,13 @@ import { executeStatementDefaults } from "./defaults";
 const logger = createLogger("connectors:sql-warehouse");
 
 /**
- * Maximum size for inline Arrow IPC attachments (8 MiB decoded).
- * Aligned with `streamDefaults.maxEventSize` so anything that would exceed
- * the SSE event cap fails here with a clear error rather than a confusing
- * "Buffer size exceeded" downstream. Larger results should use
- * `disposition: "EXTERNAL_LINKS"`, which the analytics fallback handles.
+ * Maximum size for inline Arrow IPC attachments (25 MiB decoded).
+ * Matches the Databricks Statement Execution API hard cap on INLINE
+ * disposition. The bytes are stashed server-side (see InlineArrowStash) and
+ * served out of band via /arrow-result/:jobId, so the SSE event-size cap
+ * does not apply here.
  */
-const MAX_INLINE_ATTACHMENT_BYTES = 8 * 1024 * 1024;
+const MAX_INLINE_ATTACHMENT_BYTES = 25 * 1024 * 1024;
 
 interface SQLWarehouseConfig {
   timeout?: number;

packages/appkit/src/connectors/sql-warehouse/tests/client.test.ts

@@ -293,8 +293,8 @@ describe("SQLWarehouseConnector._transformDataArray", () => {
 
     test("rejects oversized attachments to bound memory", () => {
       const connector = createConnector();
-      // 8 MiB decoded cap → ~12 MiB of base64 chars decodes to >8 MiB.
-      const oversized = "A".repeat(12 * 1024 * 1024);
+      // 25 MiB decoded cap → ~36 MiB of base64 chars decodes to >25 MiB.
+      const oversized = "A".repeat(36 * 1024 * 1024);
       const response = {
         statement_id: "stmt-oversized",
         status: { state: "SUCCEEDED" },

packages/appkit/src/plugins/analytics/analytics.ts

@@ -12,6 +12,7 @@ import { createLogger } from "../../logging/logger";
 import { Plugin, toPlugin } from "../../plugin";
 import type { PluginManifest } from "../../registry";
 import { queryDefaults } from "./defaults";
+import { InlineArrowStash } from "./inline-arrow-stash";
 import manifest from "./manifest.json";
 import { QueryProcessor } from "./query";
 import type {
@@ -33,11 +34,13 @@ export class AnalyticsPlugin extends Plugin {
   // analytics services
   private SQLClient: SQLWarehouseConnector;
   private queryProcessor: QueryProcessor;
+  private inlineStash: InlineArrowStash;
 
   constructor(config: IAnalyticsConfig) {
     super(config);
     this.config = config;
     this.queryProcessor = new QueryProcessor();
+    this.inlineStash = new InlineArrowStash();
 
     this.SQLClient = new SQLWarehouseConnector({
       timeout: config.timeout,
@@ -76,6 +79,24 @@ export class AnalyticsPlugin extends Plugin {
   ): Promise<void> {
     try {
       const { jobId } = req.params;
+
+      // Inline path: ARROW_STREAM + INLINE responses are stashed by the query
+      // route and served from memory rather than fetched from the warehouse.
+      const stashed = this.inlineStash.take(jobId);
+      if (stashed) {
+        res.setHeader("Content-Type", "application/octet-stream");
+        res.setHeader("Content-Length", stashed.length.toString());
+        // Don't cache — stash entries are one-shot.
+        res.setHeader("Cache-Control", "no-store");
+        logger.debug(
+          "Serving inline Arrow buffer from stash: %d bytes for jobId=%s",
+          stashed.length,
+          jobId,
+        );
+        res.send(stashed);
+        return;
+      }
+
       const workspaceClient = getWorkspaceClient();
 
       logger.debug("Processing Arrow job request for jobId=%s", jobId);
@@ -247,12 +268,15 @@ export class AnalyticsPlugin extends Plugin {
         { disposition: "INLINE", format: "ARROW_STREAM" },
         signal,
       );
-      // INLINE responses with an Arrow IPC attachment are forwarded as base64
-      // for the client to decode into an Arrow Table. Anything else (rare:
-      // data_array under ARROW_STREAM, or an empty result) falls back to the
-      // generic "result" payload.
+      // INLINE responses carry the Arrow IPC bytes as a base64 `attachment`.
+      // Stash them server-side and emit the same `{type:"arrow", statement_id}`
+      // shape that EXTERNAL_LINKS uses, so the client fetches the bytes via
+      // the existing /arrow-result/:jobId path. This keeps SSE messages
+      // small and unifies the wire protocol.
       if (result?.attachment) {
-        return { type: "arrow_inline", attachment: result.attachment };
+        const buffer = Buffer.from(result.attachment, "base64");
+        const statement_id = this.inlineStash.put(buffer);
+        return { type: "arrow", statement_id };
       }
       return { type: "result", ...result };
     } catch (err: unknown) {

packages/appkit/src/plugins/analytics/inline-arrow-stash.ts

@@ -0,0 +1,101 @@
+import { randomUUID } from "node:crypto";
+
+/**
+ * Bounded TTL stash of inline Arrow IPC payloads, keyed by a synthetic
+ * statement ID prefixed with `inline-`.
+ *
+ * The analytics route puts each ARROW_STREAM + INLINE response into the stash
+ * and emits `{ type: "arrow", statement_id: <inline-...> }` over SSE. The
+ * client fetches the bytes via the existing `/arrow-result/:jobId` endpoint,
+ * which checks this stash first and only delegates to the warehouse fetch
+ * when the ID is not stashed (i.e., a real EXTERNAL_LINKS statement).
+ *
+ * Decoupling bulk bytes from the SSE channel keeps `streamDefaults.maxEventSize`
+ * small (control messages stay small), at the cost of in-process memory for
+ * the duration of the round trip.
+ *
+ * Bounds:
+ * - per-entry size: enforced upstream by the connector (`MAX_INLINE_ATTACHMENT_BYTES`).
+ * - max entries: LRU-evict when full.
+ * - TTL: time after which an unread entry is dropped.
+ *
+ * Reads are one-shot — `take()` removes the entry — because each query has
+ * exactly one consumer. This bounds peak memory in steady state to roughly
+ * one entry per active analytics query, not `maxEntries × maxBytes`.
+ *
+ * Single-process only. A multi-server deployment would need a shared store
+ * (e.g. Redis) — see PR description for the limitation.
+ */
+interface InlineArrowStashOptions {
+  /** Maximum number of pending entries (LRU eviction beyond this). Default 100. */
+  maxEntries?: number;
+  /** Time in ms before an unread entry is auto-evicted. Default 60_000 (60s). */
+  ttlMs?: number;
+}
+
+interface StashEntry {
+  buffer: Buffer;
+  expiresAt: number;
+}
+
+export class InlineArrowStash {
+  private readonly entries = new Map<string, StashEntry>();
+  private readonly maxEntries: number;
+  private readonly ttlMs: number;
+
+  constructor(options: InlineArrowStashOptions = {}) {
+    this.maxEntries = options.maxEntries ?? 100;
+    this.ttlMs = options.ttlMs ?? 60_000;
+  }
+
+  /** Stash an Arrow IPC buffer and return the synthetic statement_id. */
+  put(buffer: Buffer): string {
+    this._evictExpired();
+    while (this.entries.size >= this.maxEntries) {
+      // LRU: oldest insertion order — Map iterates in insertion order.
+      const oldestKey = this.entries.keys().next().value;
+      if (oldestKey === undefined) break;
+      this.entries.delete(oldestKey);
+    }
+    const id = `inline-${randomUUID()}`;
+    this.entries.set(id, {
+      buffer,
+      expiresAt: Date.now() + this.ttlMs,
+    });
+    return id;
+  }
+
+  /**
+   * Retrieve and remove a stashed buffer. Returns `null` if the id is not in
+   * the stash, expired, or not prefixed `inline-` (in which case the caller
+   * should treat it as a real warehouse statement_id).
+   */
+  take(id: string): Buffer | null {
+    if (!id.startsWith("inline-")) return null;
+    const entry = this.entries.get(id);
+    if (!entry) return null;
+    this.entries.delete(id);
+    if (entry.expiresAt < Date.now()) return null;
+    return entry.buffer;
+  }
+
+  /** Drop expired entries without consuming them. */
+  private _evictExpired(): void {
+    const now = Date.now();
+    for (const [id, entry] of this.entries) {
+      if (entry.expiresAt < now) {
+        this.entries.delete(id);
+      }
+    }
+  }
+
+  /** For tests/observability. */
+  size(): number {
+    return this.entries.size;
+  }
+
+  /** For tests. */
+  clear(): void {
+    this.entries.clear();
+  }
+}