// SPDX-License-Identifier: Apache-2.0
// SPDX-FileCopyrightText: Copyright The LanceDB Authors

import { tableFromIPC } from "apache-arrow";
import {
  Data,
  SchemaLike,
  TableLike,
  fromTableToStreamBuffer,
  isArrowTable,
  makeArrowTable,
} from "./arrow";
import {
  Table as ArrowTable,
  fromTableToBuffer,
  makeEmptyTable,
} from "./arrow";
import { EmbeddingFunctionConfig, getRegistry } from "./embedding/registry";
import { Connection as LanceDbConnection } from "./native";
import type {
  CreateNamespaceResponse,
  DescribeNamespaceResponse,
  DropNamespaceResponse,
  Job,
  JobDescription,
  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 for pagination.
   *
   * 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 continue
   * 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 tables in this database.
   *
   * Results may be paginated. To retrieve subsequent pages, pass the
   * `pageToken` returned by a previous call. A page may be shorter than
   * `limit` without being the last one, so walk until the 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>} Table names and an optional token
   *   for fetching the next page.
   */
  abstract listTables(
    options?: Partial<ListTablesOptions>,
  ): Promise<ListTablesResponse>;
  /**
   * List a page of 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>} Table names and an optional token
   *   for fetching the next page.
   */
  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
   */
  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>;

  /**
   * 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>;

  /**
   * A {@link Job} handle for a server-side job by id.
   *
   * The handle is constructed without a server round trip; an unknown id
   * surfaces when the handle is used. Dropping the handle has no effect on
   * the job itself.
   */
  abstract job(jobId: string): Job;

  /** List server-side jobs across the database's tables. */
  abstract listJobs(): Promise<JobInfo[]>;

  /**
   * Describe a single server-side job by id.
   *
   * Resolves to `null` when the server has no such job.
   */
  abstract getJob(jobId: string): Promise<JobDescription | null>;

  /**
   * 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>;

  /**
   * The lifecycle event history of a server-side job, as an Arrow table.
   *
   * Lists history across all jobs when `jobId` is omitted.
   */
  abstract jobHistory(jobId?: string): Promise<ArrowTable>;
}

/** @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 listTables(
    namespacePathOrOptions?: string[] | Partial<ListTablesOptions>,
    options?: Partial<ListTablesOptions>,
  ): Promise<ListTablesResponse> {
    // Detect if first argument is namespacePath array or options object
    let namespacePath: string[] | undefined;
    let listTablesOptions: Partial<ListTablesOptions> | undefined;

    if (Array.isArray(namespacePathOrOptions)) {
      namespacePath = namespacePathOrOptions;
      listTablesOptions = options;
    } else {
      namespacePath = undefined;
      listTablesOptions = 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 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,
    );
  }

  job(jobId: string): Job {
    return this.inner.job(jobId);
  }

  async listJobs(): Promise<JobInfo[]> {
    return this.inner.listJobs();
  }

  async getJob(jobId: string): Promise<JobDescription | null> {
    return this.inner.getJob(jobId);
  }

  async cancelJob(jobId: string): Promise<boolean> {
    return this.inner.cancelJob(jobId);
  }

  async jobHistory(jobId?: string): Promise<ArrowTable> {
    const buf = await this.inner.jobHistory(jobId);
    if (buf.length === 0) {
      return new ArrowTable();
    }
    return tableFromIPC(buf);
  }
}

/**
 * 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 };
  }
}