fix(appkit): decode ARROW_STREAM DESCRIBE results in typegen

SDK executeStatement returns ARROW_STREAM by default (rows in
result.attachment, data_array empty), so the metric and query DESCRIBE
fetchers silently degraded on warehouses that don't default to
JSON_ARRAY. Add normalizeResultRows (apache-arrow tableFromIPC) and
request ARROW_STREAM + INLINE in both fetchers; downstream parsers read
the populated data_array unchanged. Verified live against a real
warehouse: real measure/dimension unions, cache no longer degraded.

Co-authored-by: Isaac
Signed-off-by: Atila Fassina <atila@fassina.eu>

Author: atilafassina

packages/appkit/package.json

@@ -77,6 +77,7 @@
     "@opentelemetry/sdk-trace-base": "2.6.0",
     "@opentelemetry/semantic-conventions": "1.38.0",
     "@types/semver": "7.7.1",
+    "apache-arrow": "21.1.0",
     "dotenv": "16.6.1",
     "express": "4.22.0",
     "get-port": "7.2.0",

packages/appkit/src/type-generator/metric-registry.ts

@@ -1,6 +1,7 @@
 import fs from "node:fs/promises";
 import path from "node:path";
 import type { WorkspaceClient } from "@databricks/sdk-experimental";
+import { normalizeResultRows } from "./statement-result";
 import type { DatabricksStatementExecutionResponse } from "./types";
 
 /**
@@ -1136,8 +1137,16 @@ export function createWorkspaceDescribeFetcher(
       statement: `DESCRIBE TABLE EXTENDED ${quotedFqn} AS JSON`,
       warehouse_id: warehouseId,
       wait_timeout: "30s",
+      // INLINE + ARROW_STREAM returns the single DESCRIBE row as a base64
+      // Arrow IPC attachment (the SDK's default disposition would also stream
+      // it as an attachment, but pinning these makes the wire shape explicit).
+      // normalizeResultRows decodes that attachment into `result.data_array`
+      // so `parseDescribeTableExtendedJson` can read the JSON-string cell;
+      // it is a no-op passthrough when the warehouse already populated rows.
+      format: "ARROW_STREAM",
+      disposition: "INLINE",
     })) as DatabricksStatementExecutionResponse;
-    return result;
+    return await normalizeResultRows(result);
   };
 }
 

packages/appkit/src/type-generator/query-registry.ts

@@ -6,6 +6,7 @@ import { createLogger } from "../logging/logger";
 import { CACHE_VERSION, hashSQL, loadCache, saveCache } from "./cache";
 import { decidePreflight, type PreflightMode } from "./preflight";
 import { Spinner } from "./spinner";
+import { normalizeResultRows } from "./statement-result";
 import {
   type DatabricksStatementExecutionResponse,
   type QueryFatalError,
@@ -684,6 +685,16 @@ export async function generateQueriesFromDescribe(
         const result = (await client.statementExecution.executeStatement({
           statement: `DESCRIBE QUERY ${cleanedSql}`,
           warehouse_id: warehouseId,
+          // Wait synchronously for completion (matches the metric fetcher):
+          // without it the call can return PENDING/RUNNING with no rows yet,
+          // which classifies as a no-result degrade and ships `unknown`.
+          wait_timeout: "30s",
+          // INLINE + ARROW_STREAM returns the DESCRIBE rows as a base64 Arrow
+          // IPC attachment; normalizeResultRows (below) decodes it into
+          // `result.data_array` so convertToQueryType can read the columns.
+          // No-op passthrough when the warehouse already populated rows.
+          format: "ARROW_STREAM",
+          disposition: "INLINE",
         })) as DatabricksStatementExecutionResponse;
 
         completed++;
@@ -732,7 +743,16 @@ export async function generateQueriesFromDescribe(
           };
         }
 
-        const { type, hasResults } = convertToQueryType(result, sql, queryName);
+        // Decode an Arrow IPC attachment into `result.data_array` (no-op when
+        // the warehouse already populated rows) so convertToQueryType can read
+        // the described columns. State classification above intentionally stays
+        // on the raw `result` — the normalizer preserves `status` untouched.
+        const normalized = await normalizeResultRows(result);
+        const { type, hasResults } = convertToQueryType(
+          normalized,
+          sql,
+          queryName,
+        );
         if (!hasResults) {
           // Described, but no result columns. Emit `unknown` and retry next run;
           // do not cache (we never persist `result: unknown`).

packages/appkit/src/type-generator/statement-result.ts

@@ -0,0 +1,94 @@
+import type { DatabricksStatementExecutionResponse } from "./types";
+
+/**
+ * Normalize a Statement Execution response so downstream parsers can always
+ * read rows from `result.data_array`, regardless of the wire format the
+ * warehouse chose.
+ *
+ * ## Why this exists
+ *
+ * `@databricks/sdk-experimental`'s `executeStatement` defaults to an
+ * `ARROW_STREAM` disposition. With an `INLINE` disposition the single
+ * DESCRIBE row is returned as a base64-encoded Arrow IPC stream in
+ * `result.attachment` and `result.data_array` is left undefined. The metric
+ * and query type generators only ever read `result.data_array`, so without
+ * this normalization an Arrow response reads as "returned no rows" — the
+ * registry ships empty and the runtime fail-closed gate 503s every affected
+ * metric/query. (A warehouse configured to return `JSON_ARRAY` populates
+ * `data_array` directly and needs no decoding — that path, and every mocked
+ * test, flows through here unchanged.)
+ *
+ * ## Behavior
+ *
+ * - `data_array` already present → return the response unchanged (passthrough;
+ *   keeps JSON_ARRAY warehouses and all `data_array`-based mocked tests working
+ *   with zero decode cost).
+ * - otherwise `attachment` present → lazily import `apache-arrow`, decode the
+ *   IPC stream, and return a response with `result.data_array` populated as
+ *   `(string | null)[][]`. All other fields (`status`, `statement_id`,
+ *   `manifest`, and any other `result` keys) are preserved.
+ * - neither present → return the response unchanged (empty result; the
+ *   downstream "returned no rows" path then degrades correctly).
+ *
+ * ## Failure contract
+ *
+ * Decode is best-effort and never throws: a malformed/empty attachment
+ * resolves to the original response (with `data_array` still absent) rather
+ * than rejecting. This is deliberate — the metric/query sync paths treat a
+ * response without rows as a deterministic "no rows" degrade (warn-and-continue
+ * + sticky cache), so swallowing the decode error here keeps that contract
+ * intact instead of crashing the whole generation pass on one bad payload.
+ *
+ * The `apache-arrow` import is lazy (dynamic `import()`) so the dependency only
+ * loads when an attachment actually needs decoding — JSON_ARRAY warehouses and
+ * unit tests that build `data_array` directly never pull it in.
+ *
+ * @param response - the raw Statement Execution response
+ * @returns a response guaranteed to expose rows via `result.data_array` when
+ *   they were decodable, otherwise the response unchanged
+ */
+export async function normalizeResultRows(
+  response: DatabricksStatementExecutionResponse,
+): Promise<DatabricksStatementExecutionResponse> {
+  // Passthrough: rows already materialized (JSON_ARRAY warehouses + every
+  // mocked test). `data_array` being an empty array still counts as present —
+  // that is a genuine "no rows" answer we must not overwrite with a decode.
+  if (response.result?.data_array !== undefined) {
+    return response;
+  }
+
+  const attachment = response.result?.attachment;
+  if (attachment === undefined) {
+    // No rows, no attachment: nothing to normalize. Let the downstream
+    // "returned no rows" degrade path fire.
+    return response;
+  }
+
+  try {
+    // Lazy import: only pull apache-arrow into the process when an attachment
+    // genuinely needs decoding.
+    const { tableFromIPC } = await import("apache-arrow");
+    const bytes = Buffer.from(attachment, "base64");
+    const table = tableFromIPC(bytes);
+    const dataArray: (string | null)[][] = table
+      .toArray()
+      .map((row) =>
+        Object.values(row as Record<string, unknown>).map((value) =>
+          value == null ? null : String(value),
+        ),
+      );
+
+    return {
+      ...response,
+      result: {
+        ...response.result,
+        data_array: dataArray,
+      },
+    };
+  } catch {
+    // Best-effort: a corrupt/partial Arrow payload must not crash the
+    // generation pass. Returning the response unchanged (data_array still
+    // absent) routes it into the deterministic "no rows" degrade downstream.
+    return response;
+  }
+}

packages/appkit/src/type-generator/tests/fixtures/describe-arrow-attachment.b64

@@ -0,0 +1 @@
+/////3gAAAAQAAAAAAAKAAwACgAJAAQACgAAABAAAAAAAQQACAAIAAAABAAIAAAABAAAAAEAAAAUAAAAEAAUABAAAAAPAAQAAAAIABAAAAAYAAAADAAAAAAAAAUQAAAAAAAAAAQABAAEAAAADQAAAGpzb25fbWV0YWRhdGEAAAD/////uAAAABAAAAAMABoAGAAXAAQACAAMAAAAIAAAAEAFAAAAAAAAAAAAAAAAAAMEAAoAGAAMAAgABAAKAAAALAAAABAAAAABAAAAAAAAAAAAAAABAAAAAQAAAAAAAAAAAAAAAAAAAAAAAAADAAAAAAAAAAAAAAABAAAAAAAAAEAAAAAAAAAACAAAAAAAAACAAAAAAAAAAJEEAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAD/AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAJEEAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAHsiY2F0YWxvZ19uYW1lIjoiYXBwa2l0X2RlbW8iLCJzY2hlbWFfbmFtZSI6InB1YmxpYyIsIm5hbWVzcGFjZSI6WyJwdWJsaWMiXSwidGFibGVfbmFtZSI6InJldmVudWVfbWV0cmljcyIsImNvbHVtbnMiOlt7Im5hbWUiOiJyZWdpb24iLCJudWxsYWJsZSI6dHJ1ZSwidHlwZSI6eyJjb2xsYXRpb24iOiJVVEY4X0JJTkFSWSIsIm5hbWUiOiJzdHJpbmcifX0seyJuYW1lIjoic2VnbWVudCIsIm51bGxhYmxlIjp0cnVlLCJ0eXBlIjp7ImNvbGxhdGlvbiI6IlVURjhfQklOQVJZIiwibmFtZSI6InN0cmluZyJ9fSx7Im5hbWUiOiJjcmVhdGVkX2F0IiwibnVsbGFibGUiOmZhbHNlLCJ0eXBlIjp7Im5hbWUiOiJ0aW1lc3RhbXBfbHR6In19LHsiaXNfbWVhc3VyZSI6dHJ1ZSwibmFtZSI6Im1yciIsIm51bGxhYmxlIjp0cnVlLCJ0eXBlIjp7Im5hbWUiOiJkb3VibGUifX0seyJjb21tZW50IjoiQW5udWFsaXplZCBjb250cmFjdCB2YWx1ZSBhY3Jvc3MgYWxsIGFjdGl2ZSBzdWJzY3JpcHRpb25zIiwiaXNfbWVhc3VyZSI6dHJ1ZSwibmFtZSI6ImFyciIsIm51bGxhYmxlIjp0cnVlLCJ0eXBlIjp7Im5hbWUiOiJkb3VibGUifX0seyJpc19tZWFzdXJlIjp0cnVlLCJuYW1lIjoibmV3X2FyciIsIm51bGxhYmxlIjp0cnVlLCJ0eXBlIjp7Im5hbWUiOiJkb3VibGUifX0seyJpc19tZWFzdXJlIjp0cnVlLCJuYW1lIjoiY2h1cm5lZF9hcnIiLCJudWxsYWJsZSI6dHJ1ZSwidHlwZSI6eyJuYW1lIjoiZG91YmxlIn19XSwidHlwZSI6Ik1FVFJJQ19WSUVXIiwicHJvdmlkZXIiOiJ1bmtub3duIiwibG9jYXRpb24iOiIvIiwiaXNfbWFuYWdlZF9sb2NhdGlvbiI6ZmFsc2UsInRhYmxlX3Byb3BlcnRpZXMiOnsibWV0cmljX3ZpZXcuZnJvbS5uYW1lIjoiYXBwa2l0X2RlbW8ucHVibGljLmZhY3Rfc3Vic2NyaXB0aW9uIiwibWV0cmljX3ZpZXcuZnJvbS50eXBlIjoiQVNTRVQiLCJtZXRyaWNfdmlldy5qb2lucyI6Ilt7XCJuYW1lXCI6XCJjdXN0b21lclwifV0iLCJtZXRyaWNfdmlldy5tYXRlcmlhbGl6YXRpb24uZW5hYmxlZCI6ImZhbHNlIiwibWV0cmljX3ZpZXcud2hlcmUiOiJzb3VyY2Uuc3RhdHVzID0gJ2FjdGl2ZScifSwiY29sbGF0aW9uIjoiVVRGOF9CSU5BUlkiLCJjcmVhdGVkX3RpbWUiOiIyMDI2LTA0LTMwVDA3OjE3OjMyWiIsImxhc3RfYWNjZXNzIjoiVU5LTk9XTiJ9AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAD/////AAAAAA==

packages/appkit/src/type-generator/tests/generate-queries.test.ts

@@ -83,6 +83,35 @@ function succeededResult(columns: [string, string, string | null][]) {
   };
 }
 
+/**
+ * Build a SUCCEEDED DESCRIBE QUERY response whose rows arrive only as a base64
+ * Arrow IPC `attachment` (no `data_array`) — the ARROW_STREAM/INLINE wire shape
+ * the fetcher now requests. The describeOne path pipes this through
+ * normalizeResultRows, which decodes the attachment so convertToQueryType can
+ * read the columns. Each [name, type, comment] triple becomes one DESCRIBE row.
+ */
+async function succeededArrowAttachmentResult(
+  columns: [string, string, string | null][],
+) {
+  const arrow = await import("apache-arrow");
+  const table = arrow.tableFromArrays({
+    col_name: columns.map((c) => c[0]),
+    data_type: columns.map((c) => c[1]),
+    comment: columns.map((c) => c[2]),
+  });
+  const attachment = Buffer.from(arrow.tableToIPC(table, "stream")).toString(
+    "base64",
+  );
+  return {
+    statement_id: "stmt-arrow",
+    status: { state: "SUCCEEDED" },
+    manifest: { format: "ARROW_STREAM" },
+    // No data_array — rows live in the attachment, like a real INLINE Arrow
+    // response. This is the condition the silent-degrade bug left unread.
+    result: { attachment },
+  };
+}
+
 describe("generateQueriesFromDescribe", () => {
   beforeEach(() => {
     vi.clearAllMocks();
@@ -120,6 +149,60 @@ describe("generateQueriesFromDescribe", () => {
     expect(lastSavedQueries()?.users.type).toContain("id: number");
   });
 
+  test("ARROW attachment path — decodes Arrow rows into a real query schema", async () => {
+    // The warehouse answers ARROW_STREAM/INLINE: columns arrive only as a
+    // base64 Arrow IPC attachment with data_array undefined. describeOne pipes
+    // this through normalizeResultRows before convertToQueryType, so the schema
+    // resolves to real columns instead of the degraded `result: unknown`.
+    mocks.readdir.mockResolvedValue(["users.sql"]);
+    mocks.readFile.mockResolvedValue(
+      "SELECT id, name FROM users WHERE status = :status",
+    );
+    mocks.executeStatement.mockResolvedValue(
+      await succeededArrowAttachmentResult([
+        ["id", "INT", null],
+        ["name", "STRING", "display name"],
+      ]),
+    );
+
+    const { schemas, syntaxErrors, fatalErrors } = await describeQueries(
+      "/queries",
+      "wh-123",
+    );
+
+    expect(schemas).toHaveLength(1);
+    expect(schemas[0].name).toBe("users");
+    // Real columns recovered from the Arrow attachment — not `result: unknown`.
+    expect(schemas[0].type).toContain("id: number");
+    expect(schemas[0].type).toContain("name: string");
+    expect(schemas[0].type).not.toContain("result: unknown");
+    expect(syntaxErrors).toEqual([]);
+    expect(fatalErrors).toEqual([]);
+    // A resolved schema is cached (we only persist non-unknown results).
+    expect(lastSavedQueries()?.users.type).toContain("id: number");
+  });
+
+  test("ARROW attachment request — pins ARROW_STREAM/INLINE and a 30s wait", async () => {
+    // Regression guard for the executeStatement call shape: without the pinned
+    // wait_timeout the call could return PENDING and degrade; without
+    // ARROW_STREAM/INLINE the normalizer would have no attachment to decode.
+    mocks.readdir.mockResolvedValue(["q.sql"]);
+    mocks.readFile.mockResolvedValue("SELECT 1 AS one");
+    mocks.executeStatement.mockResolvedValue(
+      succeededResult([["one", "INT", null]]),
+    );
+
+    await describeQueries("/queries", "wh-123");
+
+    expect(mocks.executeStatement).toHaveBeenCalledTimes(1);
+    expect(mocks.executeStatement.mock.calls[0][0]).toMatchObject({
+      warehouse_id: "wh-123",
+      wait_timeout: "30s",
+      format: "ARROW_STREAM",
+      disposition: "INLINE",
+    });
+  });
+
   test("FAILED status with error message — reports SQL error and produces unknown result type", async () => {
     mocks.readdir.mockResolvedValue(["bad_table.sql"]);
     mocks.readFile.mockResolvedValue("SELECT * FROM bad_table");

packages/appkit/src/type-generator/tests/metric-registry.test.ts

@@ -1,3 +1,4 @@
+import { readFileSync } from "node:fs";
 import fs from "node:fs/promises";
 import os from "node:os";
 import path from "node:path";
@@ -46,6 +47,20 @@ function mockDescribeResponse(
   };
 }
 
+/**
+ * Real Arrow IPC attachment captured live from dogfood:
+ *   DESCRIBE TABLE EXTENDED `appkit_demo`.`public`.`revenue_metrics` AS JSON
+ * with `format: "ARROW_STREAM", disposition: "INLINE"`. The single DESCRIBE
+ * row (one JSON-string cell) is base64 Arrow IPC — the wire shape this fetcher
+ * now requests and the normalizer decodes. Used to prove the fetcher →
+ * normalizer → parser chain extracts real columns from an attachment-only
+ * response (the silent-degrade bug left this unread).
+ */
+const ARROW_ATTACHMENT_B64 = readFileSync(
+  path.join(__dirname, "fixtures", "describe-arrow-attachment.b64"),
+  "utf-8",
+);
+
 /**
  * Cast helper for fixtures that intentionally violate the config type
  * (invalid executors, unknown fields, legacy shapes, ...).
@@ -439,9 +454,48 @@ describe("createWorkspaceDescribeFetcher", () => {
       statement: "DESCRIBE TABLE EXTENDED `demo`.`sales`.`revenue` AS JSON",
       warehouse_id: "wh-1",
       wait_timeout: "30s",
+      // Pinned wire format: the warehouse returns the single DESCRIBE row as a
+      // base64 Arrow IPC attachment, which the normalizer decodes into rows.
+      format: "ARROW_STREAM",
+      disposition: "INLINE",
     });
   });
 
+  test("decodes an Arrow attachment-only response into parseable columns (fetcher → normalizer → parser)", async () => {
+    // The warehouse answers ARROW_STREAM/INLINE: rows arrive as a base64 Arrow
+    // IPC attachment with `data_array` undefined. Before the normalizer was
+    // wired in, parseDescribeTableExtendedJson read this as "no rows" and the
+    // metric shipped degraded. Now the fetcher pipes the response through
+    // normalizeResultRows, so the real describe doc is recovered end-to-end.
+    const statements: Array<Record<string, unknown>> = [];
+    const client = {
+      statementExecution: {
+        executeStatement: async (req: Record<string, unknown>) => {
+          statements.push(req);
+          return {
+            statement_id: "stmt-arrow",
+            status: { state: "SUCCEEDED" },
+            manifest: { format: "ARROW_STREAM" },
+            // Only an attachment — no data_array (the bug's trigger condition).
+            result: { attachment: ARROW_ATTACHMENT_B64 },
+          } as DatabricksStatementExecutionResponse;
+        },
+      },
+    } as unknown as Parameters<typeof createWorkspaceDescribeFetcher>[0];
+
+    const fetcher = createWorkspaceDescribeFetcher(client, "wh-1");
+    const response = await fetcher("appkit_demo.public.revenue_metrics");
+
+    // The fetcher decoded the attachment: rows are now readable.
+    expect(response.result?.data_array).toBeDefined();
+    const parsed = parseDescribeTableExtendedJson(response);
+    const cols = extractMetricColumns(parsed);
+    // The real revenue_metrics describe doc carries measures and dimensions.
+    expect(cols.length).toBeGreaterThan(0);
+    expect(cols.some((c) => c.isMeasure)).toBe(true);
+    expect(cols.some((c) => !c.isMeasure)).toBe(true);
+  });
+
   test("a hyphenated FQN round-trips: validated by resolveMetricConfig, quoted in the statement, response parsed", async () => {
     // Hyphenated catalogs are valid per the shared source regex; unquoted
     // they would be a SQL syntax error against a real warehouse.