Add Flight retrieval: flights (GetFlightInfo) and get (DoGet)

README.md

@@ -32,6 +32,13 @@ import { tableFromArrays } from 'apache-arrow';
 const table = tableFromArrays({ id: [1, 2], name: ['Alice', 'Bob'] });
 await client.put(table).toPath(['my', 'dataset']).execute();
 
+// Retrieve: describe a dataset, then fetch an endpoint's ticket
+const info = await client.flights({ path: ['my', 'dataset'] }).execute();
+const result = await client.get(info.endpoints()[0].ticket).execute();
+
+result.table(); // apache-arrow Table
+result.rows();  // [{ id: 1, name: 'Alice' }, { id: 2, name: 'Bob' }]
+
 // Execute a server action
 const results = await client
   .action('myAction')
@@ -43,9 +50,14 @@ await client.close();
 
 ## API
 
-### `new FlightClient(address, options?)`
+### `new FlightClient(location, options?)`
 
-Creates a client connected to a Flight server.
+Creates a client connected to a Flight server. `location` is a `host:port` address,
+a `grpc://` / `grpc+tls://` URI, or a Flight `Location` object (as found on an
+endpoint). The scheme sets TLS (`grpc+tls://` → on); a bare address falls back to
+the `tls` option. Only bare addresses and `grpc://` / `grpc+tls://` are accepted
+(case-insensitive); every other URI scheme — reuse-connection, `http(s)://`,
+`grpc+unix://`, … — throws.
 
 | Option       | Type                  | Description                        |
 | ------------ | --------------------- | ---------------------------------- |
@@ -66,6 +78,59 @@ client.put(rows)
   .execute();                  // → Promise<PutResult[]>
 ```
 
+### `client.flights(descriptor)`
+
+Starts a `GetFlightInfo` operation. `descriptor` is `{ path: string[] }` or
+`{ cmd: Buffer }`.
+
+Returns a `FlightsOperation` builder whose `execute()` resolves to a
+`FlightInfoResult`:
+
+```ts
+const info = await client.flights({ path: ['bucket', 'key'] })  // or { cmd: Buffer.from('SELECT ...') }
+  .withHeaders({ authorization: 'Bearer tok' })
+  .execute();                  // → Promise<FlightInfoResult>
+
+info.raw();        // the underlying FlightInfo (schema, totals, metadata)
+info.endpoints();  // FlightEndpoint[] — each carries a ticket and locations
+```
+
+A flight may be partitioned across several endpoints; consume each one to read the
+whole flight. An endpoint's `location` says where its ticket can be redeemed: an
+empty list (or `arrow-flight-reuse-connection`) means the current client; a
+`grpc://` / `grpc+tls://` URI means another server, reached by opening a new client:
+
+```ts
+for (const endpoint of info.endpoints()) {
+  const reuse = endpoint.location.length === 0;
+  const target = reuse ? client : new FlightClient(endpoint.location[0]);
+  const result = await target.get(endpoint.ticket).execute();
+  // ... use result.table() / result.rows()
+  if (!reuse) await target.close();
+}
+```
+
+### `client.get(ticket)`
+
+Starts a `DoGet` operation. `ticket` is a `Ticket` (as found in
+`FlightInfo.endpoint[].ticket`) or raw ticket bytes (`Buffer` / `Uint8Array`).
+
+Returns a `GetOperation` builder whose `execute()` resolves to a `FlightResult`:
+
+```ts
+const result = await client.get(ticket)
+  .withHeaders({ ... })
+  .execute();                  // → Promise<FlightResult>
+
+result.table();  // apache-arrow Table
+result.rows();   // Record<string, unknown>[]
+result.raw();    // FlightData[] (the raw stream)
+```
+
+The whole stream is drained by `execute()`. `table()` and `rows()` are derived
+lazily and cached; reading an empty stream leaves `raw()` empty and makes
+`table()`/`rows()` throw.
+
 ### `client.action(type)`
 
 Starts a `DoAction` operation.
@@ -78,14 +143,19 @@ client.action('compact')
   .execute();                  // → Promise<Buffer[]>
 ```
 
-### `rowsToTable(rows)`
+### `rowsToTable(rows)` / `tableToRows(table)`
 
-Converts an array of plain objects to a columnar Arrow `Table`.
+Convert between an array of plain objects and a columnar Arrow `Table`.
 
 ```ts
 const table = rowsToTable([{ x: 1 }, { x: 2 }]);
+const rows = tableToRows(table);   // [{ x: 1 }, { x: 2 }]
 ```
 
+`tableToRows` recurses into nested structs and lists, so rows are plain objects.
+Arrow types map to their JS equivalents — notably 64-bit integers come back as
+`bigint` (not JSON-serializable) and timestamps as `Date`.
+
 ## Development
 
 ```bash

src/client.ts

@@ -15,6 +15,12 @@ import {
 import { PutOperation } from './operations/put.js';
 import { rowsToTable } from './convert.js';
 import { ActionOperation } from './operations/action.js';
+import { GetOperation } from './operations/get.js';
+import type { TicketInput } from './operations/get.js';
+import { FlightsOperation } from './operations/flights.js';
+import type { DescriptorInput } from './operations/flights.js';
+import { parseLocation } from './location.js';
+import type { Location } from './generated/Flight.js';
 
 export interface FlightClientOptions {
   tls?: boolean;
@@ -41,8 +47,10 @@ export class FlightClient {
   private channel: Channel;
   private grpcClient: FlightServiceClient;
 
-  constructor(address: string, options: FlightClientOptions = {}) {
-    const creds = options.tls
+  constructor(location: string | Location, options: FlightClientOptions = {}) {
+    const { address, tls } = parseLocation(location);
+    const useTls = tls ?? options.tls ?? false;
+    const creds = useTls
       ? ChannelCredentials.createSsl()
       : ChannelCredentials.createInsecure();
 
@@ -74,6 +82,14 @@ export class FlightClient {
     return new PutOperation(this.grpcClient, table);
   }
 
+  flights(descriptor: DescriptorInput): FlightsOperation {
+    return new FlightsOperation(this.grpcClient, descriptor);
+  }
+
+  get(ticket: TicketInput | undefined): GetOperation {
+    return new GetOperation(this.grpcClient, ticket);
+  }
+
   action(type: string): ActionOperation {
     return new ActionOperation(this.grpcClient, type);
   }

src/convert.test.ts

@@ -1,6 +1,6 @@
 import { describe, it } from 'node:test';
 import assert from 'node:assert/strict';
-import { rowsToTable } from './convert.js';
+import { rowsToTable, tableToRows } from './convert.js';
 
 describe('rowsToTable', () => {
   it('converts row objects to a columnar Arrow Table', () => {
@@ -75,3 +75,36 @@ describe('rowsToTable', () => {
     assert.equal(table.getChild('bool')!.get(0), true);
   });
 });
+
+describe('tableToRows', () => {
+  it('round-trips flat rows', () => {
+    const rows = [
+      { id: 1, name: 'Alice' },
+      { id: 2, name: 'Bob' },
+    ];
+
+    assert.deepEqual(tableToRows(rowsToTable(rows)), rows);
+  });
+
+  it('deep-converts nested structs to plain objects', () => {
+    const rows = [
+      { id: 1, meta: { a: 1, nested: { x: 10 } } },
+      { id: 2, meta: { a: 2, nested: { x: 20 } } },
+    ];
+
+    const out = tableToRows(rowsToTable(rows));
+
+    assert.deepEqual(out, rows);
+    // Nested value must be a plain object, not an Arrow StructRow proxy.
+    assert.equal((out[0].meta as { constructor: unknown }).constructor, Object);
+  });
+
+  it('keeps null struct cells null', () => {
+    const rows = [
+      { id: 1, meta: { a: 1 } },
+      { id: 2, meta: null },
+    ];
+
+    assert.deepEqual(tableToRows(rowsToTable(rows)), rows);
+  });
+});

src/convert.ts

@@ -63,3 +63,29 @@ export function rowsToTable(rows: Record<string, unknown>[]): Table {
 
   return tableFromJSON(normalised);
 }
+
+/**
+ * Recursively turn an Arrow row value into plain JS.
+ * StructRow.toJSON() is shallow — nested structs come back as StructRow proxies and
+ * lists as Vectors — so recurse until every value is a plain object / array / scalar.
+ */
+function toPlain(value: unknown): unknown {
+  if (value == null || typeof value !== 'object') return value; // scalars, bigint, null
+  if (value instanceof Date || value instanceof Uint8Array) return value;
+  if (Array.isArray(value)) return value.map(toPlain);
+
+  const json = (value as { toJSON?: () => unknown }).toJSON?.();
+  if (Array.isArray(json)) return json.map(toPlain); // Vector (list)
+  if (json != null && typeof json === 'object') {
+    // StructRow / MapRow
+    return Object.fromEntries(
+      Object.entries(json).map(([k, v]) => [k, toPlain(v)]),
+    );
+  }
+  return value;
+}
+
+/** Pivot a columnar Arrow Table back into an array of plain row objects. */
+export function tableToRows(table: Table): Record<string, unknown>[] {
+  return table.toArray().map((row) => toPlain(row) as Record<string, unknown>);
+}

src/index.ts

@@ -1,13 +1,23 @@
 export { FlightClient } from './client.js';
 export type { FlightClientOptions } from './client.js';
 export { PutOperation } from './operations/put.js';
-export { rowsToTable } from './convert.js';
+export { GetOperation, FlightResult } from './operations/get.js';
+export type { TicketInput } from './operations/get.js';
+export { FlightsOperation, FlightInfoResult } from './operations/flights.js';
+export type { DescriptorInput } from './operations/flights.js';
+export { parseLocation } from './location.js';
+export type { ResolvedLocation } from './location.js';
+export { rowsToTable, tableToRows } from './convert.js';
 export { ActionOperation } from './operations/action.js';
 export type {
   Action,
   Result,
   FlightData,
   FlightDescriptor,
+  FlightInfo,
+  FlightEndpoint,
+  Ticket,
+  Location,
   PutResult,
 } from './generated/Flight.js';
 export { FlightDescriptor_DescriptorType } from './generated/Flight.js';

src/ipc.test.ts

@@ -0,0 +1,99 @@
+import { describe, it } from 'node:test';
+import assert from 'node:assert/strict';
+import { tableToIPC } from 'apache-arrow';
+import type { Table } from 'apache-arrow';
+import { parseIPCMessages, flightDataToTable } from './ipc.js';
+import { rowsToTable, tableToRows } from './convert.js';
+import type { FlightData } from './generated/Flight.js';
+
+/** Split a Table into FlightData messages, mirroring what a DoGet server sends. */
+function tableToFlightData(table: Table): FlightData[] {
+  return [...parseIPCMessages(tableToIPC(table))].map(({ header, body }) => ({
+    flightDescriptor: undefined,
+    dataHeader: header,
+    appMetadata: new Uint8Array(),
+    dataBody: body,
+  }));
+}
+
+describe('flightDataToTable', () => {
+  it('round-trips rows through the FlightData framing', () => {
+    const rows = [
+      { id: 1, name: 'Alice' },
+      { id: 2, name: 'Bob' },
+      { id: 3, name: 'Carol' },
+    ];
+
+    const messages = tableToFlightData(rowsToTable(rows));
+    const table = flightDataToTable(messages);
+
+    assert.deepEqual(tableToRows(table), rows);
+  });
+
+  it('round-trips a multi-column table with its schema and values', () => {
+    const rows = [{ int: 1, float: 1.5, str: 'a', bool: true }];
+
+    const table = flightDataToTable(tableToFlightData(rowsToTable(rows)));
+
+    assert.equal(table.numRows, 1);
+    assert.equal(table.numCols, 4);
+    assert.deepEqual(tableToRows(table), rows);
+  });
+
+  it('pads data_header whose length is not a multiple of 8', () => {
+    const rows = [{ id: 1, name: 'Alice' }, { id: 2, name: 'Bob' }];
+    // Simulate a server that sends an unpadded data_header. Appending bytes makes the
+    // length non-8-aligned; trailing bytes are ignored by the flatbuffer reader, so
+    // this exercises the header padding without risk of corrupting the message.
+    const messages = tableToFlightData(rowsToTable(rows)).map((m) => ({
+      ...m,
+      dataHeader: Buffer.concat([Buffer.from(m.dataHeader), Buffer.alloc(5)]),
+    }));
+
+    assert.deepEqual(tableToRows(flightDataToTable(messages)), rows);
+  });
+
+  it('reassembles a multi-record-batch stream', () => {
+    const batchA = tableToFlightData(rowsToTable([{ id: 1 }, { id: 2 }]));
+    const batchB = tableToFlightData(rowsToTable([{ id: 3 }, { id: 4 }]));
+    // Same schema, so reuse the first schema message and append both batch messages.
+    const combined = [...batchA, batchB[batchB.length - 1]];
+
+    const table = flightDataToTable(combined);
+
+    assert.deepEqual(tableToRows(table), [{ id: 1 }, { id: 2 }, { id: 3 }, { id: 4 }]);
+  });
+
+  it('round-trips nested structs and null cells', () => {
+    const rows = [
+      { id: 1, meta: { a: 1, nested: { x: 10 } } },
+      { id: 2, meta: { a: 2, nested: { x: 20 } } },
+      { id: 3, meta: null },
+    ];
+
+    const table = flightDataToTable(tableToFlightData(rowsToTable(rows)));
+
+    assert.deepEqual(tableToRows(table), rows);
+  });
+
+  it('skips app-metadata-only frames (empty dataHeader)', () => {
+    const rows = [{ id: 1 }, { id: 2 }];
+    const messages = tableToFlightData(rowsToTable(rows));
+
+    const withMetaFrame: FlightData[] = [
+      { flightDescriptor: undefined, dataHeader: new Uint8Array(), appMetadata: Buffer.from('meta'), dataBody: new Uint8Array() },
+      ...messages,
+    ];
+
+    assert.deepEqual(tableToRows(flightDataToTable(withMetaFrame)), rows);
+  });
+
+  it('throws when no message carries Arrow data', () => {
+    const metaOnly: FlightData[] = [
+      { flightDescriptor: undefined, dataHeader: new Uint8Array(), appMetadata: Buffer.from('meta'), dataBody: new Uint8Array() },
+    ];
+
+    assert.throws(() => flightDataToTable(metaOnly), /no data received/i);
+    assert.throws(() => flightDataToTable([]), /no data received/i);
+  });
+});