Files
lancedb/nodejs/lancedb/connection.ts
T
Jack Ye 21f11b4463 feat!: replace get_job/job_history with describe_job/query_job_events (#4130)
A 1M-row column refresh over 200 fragments produced no visible result,
and the client could only ever say `"running"`. Everything needed to
diagnose it already existed server-side — the job registry records a
`claim`/`claim_complete` pair per fragment carrying `rows_processed` —
but none of it was reachable.

## Before

Four ways to ask about a job, none of which told you much.

```python
job = table.refresh_column_async("embedding")
job.status()               # "running". That was the entire debug surface.
db.get_job(job_id)         # state, and a spec. No result, no progress.
db.job_history(job_id)     # raw record batches, no limit, no filter
db.job(job_id)             # a handle that knew nothing
```

## After

Open a job the way you open a table; the handle answers everything.

```python
job = db.open_job(job_id)      # raises JobNotFoundError if there is no such job
```

```python
>>> print(job)
Job(
    id='job-1',
    state='failed',
    job_type='refresh_column',
    creation_ms=1757000000000,
    spec={
        "column": "embedding",
        "num_workers": 4
    },
    failure=JobFailureInfo(phase='execute', message='worker died', retryable=True),
)
```

Individual fields are there too — `job.state`, `job.job_type`,
`job.creation_ms`, `job.spec`, `job.result`, `job.failure` — and
`job.result` carries `rows_assigned` / `rows_failed` as soon as the job
succeeds, with no `wait()` required.

Per-fragment progress *while it is still running*:

```python
done = job.events(filter="state = 'claim_complete'", limit=10_000)
done.column("rows_processed").to_pylist()      # [5000, 5000, ...]
```

The handle an async action returns is the same object, one `refresh()`
away:

```python
job = table.refresh_column_async("embedding")
job.refresh()
job.state, job.result
```

TypeScript is the same experience, down to `console.log`:

```ts
const job = await db.openJob(jobId);   // rejects if there is no such job
console.log(job);                      // same multi-line layout
job.state; job.jobType; job.spec; job.result; job.failure;
const done = await job.events({ filter: "state = 'claim_complete'", limit: 10_000 });
```

## Why each piece matters

- **A result without waiting.** `rows_assigned` / `rows_failed` used to
live only on the terminal result, so a job that never terminated
reported nothing at all.
- **`limit`.** The server caps event rows at 1000 and truncates without
saying so, which silently hid most of a 200-fragment job's history.
- **`filter`.** `claim_complete` rows carry per-claim `rows_processed` —
the only progress signal that exists mid-flight.
- **Events outlive the worker.** They live in the job registry, not in
pod logs that vanish with the pod.
- **One place to ask.** `open_job` replaces `describe_job`,
`query_job_events` and `job`, so a question about a job has one answer
instead of one per calling location.
- **A missing job is an error, not a `None`.** The common case is a job
id copied out of a log, where absence is the surprise worth raising —
and it matches `open_table`.
- **Printing is the debug surface.** Every field on its own line, JSON
payloads keeping their structure. An unrefreshed handle stays on one
line, because there is nothing to lay out.
- **In-process jobs say so.** A local refresh reports `state` and leaves
the rest null rather than inventing fields it has no record for.

`list_jobs` and `cancel_job` stay as they were: one lists, the other is
a one-shot action that should not need a describe first.

## Breaking

All shipped in 0.38.0. No deprecated aliases.

| Was | Now |
| --- | --- |
| `Connection.get_job` → `describe_job` | `Connection.open_job` returns
a populated `Job`, or raises |
| `Connection.job_history` → `query_job_events` | `job.events(...)` |
| `Connection.job` | `Connection.open_job` |
| Python events → `List[pa.RecordBatch]` | `pa.Table` |
| `JobDescription.spec_json` / `.result_json` | internal; use `job.spec`
/ `job.result` |

Node's `Job` is now a TypeScript class wrapping the native handle, so it
returns an Arrow table and parsed values like Python does. New
`Error::JobNotFound` / `JobNotFoundError`; the three job exceptions are
now in the Python API reference.
2026-09-05 16:24:23 -07:00

1004 lines
31 KiB
TypeScript

// SPDX-License-Identifier: Apache-2.0
// SPDX-FileCopyrightText: Copyright The LanceDB Authors
import {
Data,
SchemaLike,
TableLike,
fromTableToStreamBuffer,
isArrowTable,
makeArrowTable,
} from "./arrow";
import {
Table as ArrowTable,
fromTableToBuffer,
makeEmptyTable,
} from "./arrow";
import { EmbeddingFunctionConfig, getRegistry } from "./embedding/registry";
import { Job } from "./job";
import {
MaterializedView,
MaterializedViewSelect,
normalizeSelect,
validateNonNegativeInteger,
} from "./materialized_view";
import { Connection as LanceDbConnection } from "./native";
import type {
CreateNamespaceResponse,
DescribeNamespaceResponse,
DropNamespaceResponse,
JobInfo,
ListNamespacesResponse,
ListTablesResponse,
} from "./native";
export type {
CreateNamespaceResponse,
DescribeNamespaceResponse,
DropNamespaceResponse,
ListNamespacesResponse,
ListTablesResponse,
};
import { sanitizeTable } from "./sanitize";
import { LocalTable, Table } from "./table";
export interface CreateTableOptions {
/**
* The mode to use when creating the table.
*
* If this is set to "create" and the table already exists then either
* an error will be thrown or, if existOk is true, then nothing will
* happen. Any provided data will be ignored.
*
* If this is set to "overwrite" then any existing table will be replaced.
*/
mode: "create" | "overwrite";
/**
* If this is true and the table already exists and the mode is "create"
* then no error will be raised.
*/
existOk: boolean;
/**
* Configuration for object storage.
*
* Options already set on the connection will be inherited by the table,
* but can be overridden here.
*
* The available options are described at https://docs.lancedb.com/storage/
*/
storageOptions?: Record<string, string>;
/**
* The version of the data storage format to use.
*
* The default is `stable`.
* Set to "legacy" to use the old format.
*
* @deprecated Pass `new_table_data_storage_version` to storageOptions instead.
*/
dataStorageVersion?: string;
/**
* Use the new V2 manifest paths. These paths provide more efficient
* opening of datasets with many versions on object stores. WARNING:
* turning this on will make the dataset unreadable for older versions
* of LanceDB (prior to 0.10.0). To migrate an existing dataset, instead
* use the {@link LocalTable#migrateManifestPathsV2} method.
*
* @deprecated Pass `new_table_enable_v2_manifest_paths` to storageOptions instead.
*/
enableV2ManifestPaths?: boolean;
schema?: SchemaLike;
embeddingFunction?: EmbeddingFunctionConfig;
}
export interface OpenTableOptions {
/**
* Open the table scoped to this branch instead of the default branch.
*
* Reads and writes on the returned table operate in the branch's context.
*/
branch?: string;
/**
* Open the table pinned to this version, producing a read-only view.
*
* Composes with {@link OpenTableOptions.branch}: when both are set, opens
* that branch at the version; otherwise opens `main` at the version. Call
* `checkoutLatest` to return to a writable state.
*/
version?: number;
/**
* Configuration for object storage.
*
* Options already set on the connection will be inherited by the table,
* but can be overridden here.
*
* The available options are described at https://docs.lancedb.com/storage/
*/
storageOptions?: Record<string, string>;
/**
* Set the size of the index cache, specified as a number of entries
*
* @deprecated Use session-level cache configuration instead.
* Create a Session with custom cache sizes and pass it to the connect() function.
*
* The exact meaning of an "entry" will depend on the type of index:
* - IVF: there is one entry for each IVF partition
* - BTREE: there is one entry for the entire index
*
* This cache applies to the entire opened table, across all indices.
* Setting this value higher will increase performance on larger datasets
* at the expense of more RAM
*/
indexCacheSize?: number;
}
/**
* @deprecated Use {@link ListTablesOptions} with {@link Connection.listTables}
* instead.
*/
export interface TableNamesOptions {
/**
* If present, only return names that come lexicographically after the
* supplied value.
*
* This can be combined with limit to implement pagination by setting this to
* the last table name from the previous page.
*/
startAfter?: string;
/** An optional limit to the number of results to return. */
limit?: number;
}
export interface ListTablesOptions {
/**
* Token from a previous response, to resume listing where it left off.
*
* The token is opaque: it carries whatever the database needs to resume, and
* callers should not construct or interpret one.
*/
pageToken?: string;
/**
* An upper bound on how many tables to return.
*
* A page may hold fewer than this and still not be the last one, so keep
* going while the response carries a page token rather than while pages are
* full.
*/
limit?: number;
}
export interface ListNamespacesOptions {
/** Token from a previous response for pagination. */
pageToken?: string;
/** An optional limit to the number of results to return. */
limit?: number;
}
export interface CreateNamespaceOptions {
/** Creation mode. */
mode?: "create" | "exist_ok" | "overwrite";
/** Properties to set on the new namespace. */
properties?: Record<string, string>;
}
export interface DropNamespaceOptions {
/** Whether to skip if the namespace doesn't exist, or fail. */
mode?: "skip" | "fail";
/** Refuse to drop if non-empty (restrict) or drop recursively (cascade). */
behavior?: "restrict" | "cascade";
}
export interface RenameTableOptions {
/**
* The namespace path of the table being renamed. Defaults to the root
* namespace (`[]`) when omitted.
*/
namespacePath?: string[];
/**
* The namespace path to move the table to as part of the rename. When
* omitted the table stays in `namespacePath`.
*/
newNamespacePath?: string[];
}
/**
* A LanceDB Connection that allows you to open tables and create new ones.
*
* Connection could be local against filesystem or remote against a server.
*
* A Connection is intended to be a long lived object and may hold open
* resources such as HTTP connection pools. This is generally fine and
* a single connection should be shared if it is going to be used many
* times. However, if you are finished with a connection, you may call
* close to eagerly free these resources. Any call to a Connection
* method after it has been closed will result in an error.
*
* Closing a connection is optional. Connections will automatically
* be closed when they are garbage collected.
*
* Any created tables are independent and will continue to work even if
* the underlying connection has been closed.
* @hideconstructor
*/
export abstract class Connection {
[Symbol.for("nodejs.util.inspect.custom")](): string {
return this.display();
}
/**
* Return true if the connection has not been closed
*/
abstract isOpen(): boolean;
/**
* Close the connection, releasing any underlying resources.
*
* It is safe to call this method multiple times.
*
* Any attempt to use the connection after it is closed will result in an error.
*/
abstract close(): void;
/**
* Return a brief description of the connection
*/
abstract display(): string;
/**
* List all the table names in this database.
*
* Tables will be returned in lexicographical order.
* @param {Partial<TableNamesOptions>} options - options to control the
* paging / start point (backwards compatibility)
*
* @deprecated Use {@link Connection.listTables} instead.
*/
abstract tableNames(options?: Partial<TableNamesOptions>): Promise<string[]>;
/**
* List all the table names in this database.
*
* Tables will be returned in lexicographical order.
* @param {string[]} namespacePath - The namespace path to list tables from (defaults to root namespace)
* @param {Partial<TableNamesOptions>} options - options to control the
* paging / start point
*
* @deprecated Use {@link Connection.listTables} instead.
*/
abstract tableNames(
namespacePath?: string[],
options?: Partial<TableNamesOptions>,
): Promise<string[]>;
/**
* List a page of the tables in this database.
*
* To retrieve the tables after the page, pass the `pageToken` the response
* carries back in. A page can be shorter than `limit` without being the last
* one, so walk until a response carries no page token:
*
* ```ts
* const names = [];
* let pageToken = undefined;
* do {
* const page = await conn.listTables({ pageToken, limit: 100 });
* names.push(...page.tables);
* pageToken = page.pageToken;
* } while (pageToken);
* ```
*
* @param {Partial<ListTablesOptions>} options - Pagination options
* (`pageToken`, `limit`).
* @returns {Promise<ListTablesResponse>} A page of table names and an
* optional token for the tables after it.
*/
abstract listTables(
options?: Partial<ListTablesOptions>,
): Promise<ListTablesResponse>;
/**
* List a page of the tables in this database.
*
* @param {string[]} namespacePath - The namespace path to list tables from
* (defaults to root namespace)
* @param {Partial<ListTablesOptions>} options - Pagination options
* (`pageToken`, `limit`).
* @returns {Promise<ListTablesResponse>} A page of table names and an
* optional token for the tables after it.
*/
abstract listTables(
namespacePath?: string[],
options?: Partial<ListTablesOptions>,
): Promise<ListTablesResponse>;
/**
* Open a table in the database.
* @param {string} name - The name of the table
* @param {string[]} namespacePath - The namespace path of the table (defaults to root namespace)
* @param {Partial<OpenTableOptions>} options - Additional options
*/
/**
* Define a materialized view named `name` over the table `source`.
*
* The view is created empty, with the query recorded in its schema
* metadata; `view.refresh()` computes the rows. The view is a normal
* table: it can be queried, indexed and searched, and it appears in
* `tableNames`. The source table must have stable row ids (create it with
* the `newTableEnableStableRowIds` storage option); they keep the view's
* provenance valid across source compactions and cannot be enabled after
* a table exists. Local databases only.
*/
abstract createMaterializedView(
name: string,
source: string,
options?: {
select?: MaterializedViewSelect;
where?: string;
limit?: number;
},
): Promise<MaterializedView>;
/**
* Open the materialized view named `name`.
*
* Rejects a table that exists but is not a materialized view.
*/
abstract openMaterializedView(name: string): Promise<MaterializedView>;
/**
* The names of the materialized views in this database.
*
* Found by reading every table's schema, so this costs an open per table.
*/
abstract listMaterializedViews(): Promise<string[]>;
abstract openTable(
name: string,
namespacePath?: string[],
options?: Partial<OpenTableOptions>,
): Promise<Table>;
/**
* Creates a new Table and initialize it with new data.
* @param {object} options - The options object.
* @param {string} options.name - The name of the table.
* @param {Data} options.data - Non-empty Array of Records to be inserted into the table
* @param {string[]} namespacePath - The namespace path to create the table in (defaults to root namespace)
*
*/
abstract createTable(
options: {
name: string;
data: Data;
} & Partial<CreateTableOptions>,
namespacePath?: string[],
): Promise<Table>;
/**
* Creates a new Table and initialize it with new data.
* @param {string} name - The name of the table.
* @param {Record<string, unknown>[] | TableLike} data - Non-empty Array of Records
* to be inserted into the table
* @param {Partial<CreateTableOptions>} options - Additional options (backwards compatibility)
*/
abstract createTable(
name: string,
data: Record<string, unknown>[] | TableLike,
options?: Partial<CreateTableOptions>,
): Promise<Table>;
/**
* Creates a new Table and initialize it with new data.
* @param {string} name - The name of the table.
* @param {Record<string, unknown>[] | TableLike} data - Non-empty Array of Records
* to be inserted into the table
* @param {string[]} namespacePath - The namespace path to create the table in (defaults to root namespace)
* @param {Partial<CreateTableOptions>} options - Additional options
*/
abstract createTable(
name: string,
data: Record<string, unknown>[] | TableLike,
namespacePath?: string[],
options?: Partial<CreateTableOptions>,
): Promise<Table>;
/**
* Creates a new empty Table
* @param {string} name - The name of the table.
* @param {Schema} schema - The schema of the table
* @param {Partial<CreateTableOptions>} options - Additional options (backwards compatibility)
*/
abstract createEmptyTable(
name: string,
schema: import("./arrow").SchemaLike,
options?: Partial<CreateTableOptions>,
): Promise<Table>;
/**
* Creates a new empty Table
* @param {string} name - The name of the table.
* @param {Schema} schema - The schema of the table
* @param {string[]} namespacePath - The namespace path to create the table in (defaults to root namespace)
* @param {Partial<CreateTableOptions>} options - Additional options
*/
abstract createEmptyTable(
name: string,
schema: import("./arrow").SchemaLike,
namespacePath?: string[],
options?: Partial<CreateTableOptions>,
): Promise<Table>;
/**
* Drop an existing table.
* @param {string} name The name of the table to drop.
* @param {string[]} namespacePath The namespace path of the table (defaults to root namespace).
*/
abstract dropTable(name: string, namespacePath?: string[]): Promise<void>;
/**
* Start dropping a table and return its cleanup job.
*
* The table may become unavailable before its data files are removed. Wait
* on the returned job to know when cleanup has finished.
*/
abstract dropTableAsync(name: string, namespacePath?: string[]): Promise<Job>;
/**
* Drop all tables in the database.
* @param {string[]} namespacePath The namespace path to drop tables from (defaults to root namespace).
*/
abstract dropAllTables(namespacePath?: string[]): Promise<void>;
/**
* Describe a namespace, returning its properties.
*
* @param {string[]} namespacePath - The namespace path to describe, in
* parent → child order, e.g. `["analytics", "sales"]`.
* @returns {Promise<DescribeNamespaceResponse>} The namespace's properties
* (may be undefined if the namespace has none).
*/
abstract describeNamespace(
namespacePath: string[],
): Promise<DescribeNamespaceResponse>;
/**
* List the immediate child namespaces under the given parent.
*
* Results may be paginated. To retrieve subsequent pages, pass the
* `pageToken` returned by a previous call.
*
* @param {string[]} namespacePath - The parent namespace path. Defaults
* to the root namespace if omitted.
* @param {Partial<ListNamespacesOptions>} options - Pagination options
* (`pageToken`, `limit`).
* @returns {Promise<ListNamespacesResponse>} Child namespace names and
* an optional token for fetching the next page.
*/
abstract listNamespaces(
namespacePath?: string[],
options?: Partial<ListNamespacesOptions>,
): Promise<ListNamespacesResponse>;
/**
* Create a new namespace at the given path.
*
* @param {string[]} namespacePath - The namespace path to create.
* @param {Partial<CreateNamespaceOptions>} options - Creation `mode`
* ("create" | "exist_ok" | "overwrite") and optional `properties`
* to attach to the namespace.
* @returns {Promise<CreateNamespaceResponse>} The properties of the
* created namespace and an optional transaction id.
*/
abstract createNamespace(
namespacePath: string[],
options?: Partial<CreateNamespaceOptions>,
): Promise<CreateNamespaceResponse>;
/**
* Drop a namespace.
*
* Use `behavior: "cascade"` to also drop everything contained in the
* namespace (sub-namespaces and tables). The default `"restrict"`
* behavior refuses to drop a non-empty namespace.
*
* @param {string[]} namespacePath - The namespace path to drop.
* @param {Partial<DropNamespaceOptions>} options - `mode` ("skip" | "fail"
* for missing-namespace handling) and `behavior` ("restrict" | "cascade").
* @returns {Promise<DropNamespaceResponse>} Any properties returned by
* the server and an optional transaction id.
*/
abstract dropNamespace(
namespacePath: string[],
options?: Partial<DropNamespaceOptions>,
): Promise<DropNamespaceResponse>;
/**
* Clone a table from a source table.
*
* A shallow clone creates a new table that shares the underlying data files
* with the source table but has its own independent manifest. This allows
* both the source and cloned tables to evolve independently while initially
* sharing the same data, deletion, and index files.
*
* @param {string} targetTableName - The name of the target table to create.
* @param {string} sourceUri - The URI of the source table to clone from.
* @param {object} options - Clone options.
* @param {string[]} options.targetNamespacePath - The namespace path for the target table (defaults to root namespace).
* @param {number} options.sourceVersion - The version of the source table to clone.
* @param {string} options.sourceTag - The tag of the source table to clone.
* @param {boolean} options.isShallow - Whether to perform a shallow clone (defaults to true).
*/
abstract cloneTable(
targetTableName: string,
sourceUri: string,
options?: {
targetNamespacePath?: string[];
sourceVersion?: number;
sourceTag?: string;
isShallow?: boolean;
},
): Promise<Table>;
/**
* Rename a table.
*
* Currently only supported by LanceDB Cloud. Local OSS connections and
* namespace-backed connections (via {@link connectNamespace}) reject with
* a "not supported" error.
*
* @param {string} currentName - The current name of the table.
* @param {string} newName - The new name for the table.
* @param {RenameTableOptions} options - Optional namespace paths. When
* `newNamespacePath` is omitted the table stays in `namespacePath`.
*/
abstract renameTable(
currentName: string,
newName: string,
options?: RenameTableOptions,
): Promise<void>;
/**
* Open a server-side job by id, returning a handle with its record already
* populated. Rejects when the server has no such job, the way
* {@link Connection.openTable} does for a missing table.
*
* The returned {@link Job} answers for its own state, specification,
* result, failure and event history, so there is no separate
* connection-level call for any of them.
*/
abstract openJob(jobId: string): Promise<Job>;
/** List server-side jobs across the database's tables. */
abstract listJobs(): Promise<JobInfo[]>;
/**
* Request cancellation of a server-side job by id.
*
* Resolves to true if the server accepted the cancellation, false if no
* such job exists. Cancelling an already-terminal job is a no-op success.
*/
abstract cancelJob(jobId: string): Promise<boolean>;
}
/** @hideconstructor */
export class LocalConnection extends Connection {
readonly inner: LanceDbConnection;
/** @hidden */
constructor(inner: LanceDbConnection) {
super();
this.inner = inner;
}
isOpen(): boolean {
return this.inner.isOpen();
}
close(): void {
this.inner.close();
}
display(): string {
return this.inner.display();
}
async tableNames(
namespacePathOrOptions?: string[] | Partial<TableNamesOptions>,
options?: Partial<TableNamesOptions>,
): Promise<string[]> {
// Detect if first argument is namespacePath array or options object
let namespacePath: string[] | undefined;
let tableNamesOptions: Partial<TableNamesOptions> | undefined;
if (Array.isArray(namespacePathOrOptions)) {
// First argument is namespacePath array
namespacePath = namespacePathOrOptions;
tableNamesOptions = options;
} else {
// First argument is options object (backwards compatibility)
namespacePath = undefined;
tableNamesOptions = namespacePathOrOptions;
}
return this.inner.tableNames(
namespacePath ?? [],
tableNamesOptions?.startAfter,
tableNamesOptions?.limit,
);
}
async createMaterializedView(
name: string,
source: string,
options?: {
select?: MaterializedViewSelect;
where?: string;
limit?: number;
},
): Promise<MaterializedView> {
validateNonNegativeInteger(options?.limit, "limit");
const innerTable = await this.inner.createMaterializedView(
name,
source,
normalizeSelect(options?.select),
options?.where,
options?.limit,
);
return new MaterializedView(new LocalTable(innerTable));
}
async openMaterializedView(name: string): Promise<MaterializedView> {
const innerTable = await this.inner.openMaterializedView(name);
return new MaterializedView(new LocalTable(innerTable));
}
async listMaterializedViews(): Promise<string[]> {
return await this.inner.listMaterializedViews();
}
async listTables(
namespacePathOrOptions?: string[] | Partial<ListTablesOptions>,
options?: Partial<ListTablesOptions>,
): Promise<ListTablesResponse> {
// Detect if first argument is namespacePath array or options object
const namespacePath = Array.isArray(namespacePathOrOptions)
? namespacePathOrOptions
: undefined;
const listTablesOptions = Array.isArray(namespacePathOrOptions)
? options
: namespacePathOrOptions;
return this.inner.listTables(
namespacePath ?? [],
listTablesOptions?.pageToken,
listTablesOptions?.limit,
);
}
async openTable(
name: string,
namespacePath?: string[],
options?: Partial<OpenTableOptions>,
): Promise<Table> {
const innerTable = await this.inner.openTable(
name,
namespacePath ?? [],
cleanseStorageOptions(options?.storageOptions),
options?.indexCacheSize,
);
let table: Table = new LocalTable(innerTable);
// "main" is the default branch, so treat it as no branch. On a real branch,
// scope and pin in one step (yielding "version V of branch B"); otherwise
// pin the version, if any, against main.
const branch =
options?.branch != null && options.branch !== "main"
? options.branch
: undefined;
if (branch != null) {
table = await (await table.branches()).checkout(branch, options?.version);
} else if (options?.version != null) {
await table.checkout(options.version);
}
return table;
}
async cloneTable(
targetTableName: string,
sourceUri: string,
options?: {
targetNamespacePath?: string[];
sourceVersion?: number;
sourceTag?: string;
isShallow?: boolean;
},
): Promise<Table> {
const innerTable = await this.inner.cloneTable(
targetTableName,
sourceUri,
options?.targetNamespacePath ?? [],
options?.sourceVersion ?? null,
options?.sourceTag ?? null,
options?.isShallow ?? true,
);
return new LocalTable(innerTable);
}
private getStorageOptions(
options?: Partial<CreateTableOptions>,
): Record<string, string> | undefined {
if (options?.dataStorageVersion !== undefined) {
if (options.storageOptions === undefined) {
options.storageOptions = {};
}
options.storageOptions["newTableDataStorageVersion"] =
options.dataStorageVersion;
}
if (options?.enableV2ManifestPaths !== undefined) {
if (options.storageOptions === undefined) {
options.storageOptions = {};
}
options.storageOptions["newTableEnableV2ManifestPaths"] =
options.enableV2ManifestPaths ? "true" : "false";
}
return cleanseStorageOptions(options?.storageOptions);
}
async createTable(
nameOrOptions:
| string
| ({ name: string; data: Data } & Partial<CreateTableOptions>),
dataOrNamespacePath?: Record<string, unknown>[] | TableLike | string[],
namespacePathOrOptions?: string[] | Partial<CreateTableOptions>,
options?: Partial<CreateTableOptions>,
): Promise<Table> {
if (typeof nameOrOptions !== "string" && "name" in nameOrOptions) {
// First overload: createTable(options, namespacePath?)
const { name, data, ...createOptions } = nameOrOptions;
const namespacePath = dataOrNamespacePath as string[] | undefined;
return this._createTableImpl(name, data, namespacePath, createOptions);
}
// Second overload: createTable(name, data, namespacePath?, options?)
const name = nameOrOptions;
const data = dataOrNamespacePath as Record<string, unknown>[] | TableLike;
// Detect if third argument is namespacePath array or options object
let namespacePath: string[] | undefined;
let createOptions: Partial<CreateTableOptions> | undefined;
if (Array.isArray(namespacePathOrOptions)) {
// Third argument is namespacePath array
namespacePath = namespacePathOrOptions;
createOptions = options;
} else {
// Third argument is options object (backwards compatibility)
namespacePath = undefined;
createOptions = namespacePathOrOptions;
}
return this._createTableImpl(name, data, namespacePath, createOptions);
}
private async _createTableImpl(
name: string,
data: Data,
namespacePath?: string[],
options?: Partial<CreateTableOptions>,
): Promise<Table> {
if (data === undefined) {
throw new Error("data is required");
}
const { buf, mode } = await parseTableData(data, options);
const storageOptions = this.getStorageOptions(options);
const innerTable = await this.inner.createTable(
name,
buf,
mode,
namespacePath ?? [],
storageOptions,
);
return new LocalTable(innerTable);
}
async createEmptyTable(
name: string,
schema: import("./arrow").SchemaLike,
namespacePathOrOptions?: string[] | Partial<CreateTableOptions>,
options?: Partial<CreateTableOptions>,
): Promise<Table> {
// Detect if third argument is namespacePath array or options object
let namespacePath: string[] | undefined;
let createOptions: Partial<CreateTableOptions> | undefined;
if (Array.isArray(namespacePathOrOptions)) {
// Third argument is namespacePath array
namespacePath = namespacePathOrOptions;
createOptions = options;
} else {
// Third argument is options object (backwards compatibility)
namespacePath = undefined;
createOptions = namespacePathOrOptions;
}
let mode: string = createOptions?.mode ?? "create";
const existOk = createOptions?.existOk ?? false;
if (mode === "create" && existOk) {
mode = "exist_ok";
}
let metadata: Map<string, string> | undefined = undefined;
if (createOptions?.embeddingFunction !== undefined) {
const embeddingFunction = createOptions.embeddingFunction;
const registry = getRegistry();
metadata = registry.getTableMetadata([embeddingFunction]);
}
const storageOptions = this.getStorageOptions(createOptions);
const table = makeEmptyTable(schema, metadata);
const buf = await fromTableToBuffer(table);
const innerTable = await this.inner.createEmptyTable(
name,
buf,
mode,
namespacePath ?? [],
storageOptions,
);
return new LocalTable(innerTable);
}
async dropTable(name: string, namespacePath?: string[]): Promise<void> {
return this.inner.dropTable(name, namespacePath ?? []);
}
async dropTableAsync(name: string, namespacePath?: string[]): Promise<Job> {
return new Job(await this.inner.dropTableAsync(name, namespacePath ?? []));
}
async dropAllTables(namespacePath?: string[]): Promise<void> {
return this.inner.dropAllTables(namespacePath ?? []);
}
describeNamespace(
namespacePath: string[],
): Promise<DescribeNamespaceResponse> {
return this.inner.describeNamespace(namespacePath);
}
listNamespaces(
namespacePath?: string[],
options?: Partial<ListNamespacesOptions>,
): Promise<ListNamespacesResponse> {
return this.inner.listNamespaces(
namespacePath ?? [],
options?.pageToken,
options?.limit,
);
}
createNamespace(
namespacePath: string[],
options?: Partial<CreateNamespaceOptions>,
): Promise<CreateNamespaceResponse> {
return this.inner.createNamespace(
namespacePath,
options?.mode,
options?.properties,
);
}
dropNamespace(
namespacePath: string[],
options?: Partial<DropNamespaceOptions>,
): Promise<DropNamespaceResponse> {
return this.inner.dropNamespace(
namespacePath,
options?.mode,
options?.behavior,
);
}
async renameTable(
currentName: string,
newName: string,
options?: RenameTableOptions,
): Promise<void> {
return this.inner.renameTable(
currentName,
newName,
options?.namespacePath ?? [],
options?.newNamespacePath,
);
}
async openJob(jobId: string): Promise<Job> {
return new Job(await this.inner.openJob(jobId));
}
async listJobs(): Promise<JobInfo[]> {
return this.inner.listJobs();
}
async cancelJob(jobId: string): Promise<boolean> {
return this.inner.cancelJob(jobId);
}
}
/**
* Takes storage options and makes all the keys snake case.
*/
export function cleanseStorageOptions(
options?: Record<string, string>,
): Record<string, string> | undefined {
if (options === undefined) {
return undefined;
}
const result: Record<string, string> = {};
for (const [key, value] of Object.entries(options)) {
if (value !== undefined) {
const newKey = camelToSnakeCase(key);
result[newKey] = value;
}
}
return result;
}
/**
* Convert a string to snake case. It might already be snake case, in which case it is
* returned unchanged.
*/
function camelToSnakeCase(camel: string): string {
if (camel.includes("_")) {
// Assume if there is at least one underscore, it is already snake case
return camel;
}
if (camel.toLocaleUpperCase() === camel) {
// Assume if the string is all uppercase, it is already snake case
return camel;
}
let result = camel.replace(/[A-Z]/g, (letter) => `_${letter.toLowerCase()}`);
if (result.startsWith("_")) {
result = result.slice(1);
}
return result;
}
async function parseTableData(
data: Record<string, unknown>[] | TableLike,
options?: Partial<CreateTableOptions>,
streaming = false,
) {
let mode: string = options?.mode ?? "create";
const existOk = options?.existOk ?? false;
if (mode === "create" && existOk) {
mode = "exist_ok";
}
let table: ArrowTable;
if (isArrowTable(data)) {
table = sanitizeTable(data);
} else {
table = makeArrowTable(data as Record<string, unknown>[], options);
}
if (streaming) {
const buf = await fromTableToStreamBuffer(
table,
options?.embeddingFunction,
options?.schema,
);
return { buf, mode };
} else {
const buf = await fromTableToBuffer(
table,
options?.embeddingFunction,
options?.schema,
);
return { buf, mode };
}
}