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

import {
  Data,
  SchemaLike,
  TableLike,
  fromTableToStreamBuffer,
  isArrowTable,
  makeArrowTable,
  resolveBlobInputs,
} from "./arrow";
import {
  Table as ArrowTable,
  fromTableToBuffer,
  makeEmptyTable,
} from "./arrow";
import { EmbeddingFunctionConfig, getRegistry } from "./embedding/registry";
import { Job } from "./job";
import { Listing, ListingOptions, validateListingOptions } from "./listing";
import {
  MaterializedView,
  MaterializedViewSelect,
  normalizeSelect,
  validateNonNegativeInteger,
} from "./materialized_view";
import { Connection as LanceDbConnection } from "./native";
import type {
  CreateNamespaceResponse,
  DescribeNamespaceResponse,
  DropNamespaceResponse,
  JobInfo,
} from "./native";
import { ViewDescription, viewDescriptionFromNative } from "./view";
export type {
  CreateNamespaceResponse,
  DescribeNamespaceResponse,
  DropNamespaceResponse,
};
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.
   *
   * If this is set to "exist_ok" then an existing table is opened and any
   * provided data is ignored.
   */
  mode: "create" | "overwrite" | "exist_ok";
  /**
   * 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 ListingOptions} 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;
  /** The maximum number of names to return. Omitted returns all names; zero returns none. */
  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[]>;

  /** Lazily list tables in a namespace, requesting 100 per page by default.
   * @example
   * ```ts
   * for await (const name of conn.listTables([], { pageLimit: 20 })) {
   *   console.log(name);
   * }
   * ```
   */
  abstract listTables(
    namespacePath?: string[],
    options?: ListingOptions,
  ): Listing<string>;

  /**
   * 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`.
   *
   * Local creation and refresh require a source build with the
   * `experimental-functions` Cargo feature, disabled in published packages.
   * Remote Cloud/Enterprise operations do not require this feature.
   *
   * The view is populated before creation returns. Set `withNoData` to create
   * only its definition and empty backing table. The view is a normal table:
   * it can be queried, indexed and searched, and it appears in `tableNames`.
   */
  abstract createMaterializedView(
    name: string,
    source: string,
    options?: {
      select?: MaterializedViewSelect;
      where?: string;
      limit?: number;
      withNoData?: boolean;
    },
  ): 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.
   *
   * Iterate lazily with `for await...of`. Local listings inspect table schemas
   * one page at a time. Options control page size and the starting token.
   * See {@link Listing} for pagination state and error behavior.
   */
  abstract listMaterializedViews(options?: ListingOptions): Listing<string>;

  /**
   * Drop the materialized view named `name`.
   *
   * The view may become unavailable before physical cleanup finishes. Use
   * {@link dropMaterializedViewAsync} to retain and wait for the cleanup job.
   *
   * Rejects a table that exists but is not a materialized view.
   */
  abstract dropMaterializedView(
    name: string,
    namespacePath?: string[],
  ): Promise<void>;

  /**
   * Start dropping the materialized view named `name` and return its cleanup
   * job without waiting for completion.
   *
   * Rejects a table that exists but is not a materialized view.
   */
  abstract dropMaterializedViewAsync(
    name: string,
    namespacePath?: string[],
  ): Promise<Job>;

  /**
   * Create a view: a named query the database plans on every read.
   *
   * The query is planned once, at creation, so one that cannot be planned is
   * rejected now rather than at the first read. A view holds no rows, and its
   * readers see its sources as they are at read time.
   *
   * There is no replace: a name already taken is an error, and changing a
   * view is a drop followed by a create.
   */
  abstract createView(
    name: string,
    query: string,
    namespacePath?: string[],
  ): Promise<ViewDescription>;

  /**
   * What this database records about the view named `name`: its defining
   * query and the schema that query resolved to.
   */
  abstract describeView(
    name: string,
    namespacePath?: string[],
  ): Promise<ViewDescription>;

  /**
   * Drop the view named `name` and wait for its definition to be deleted.
   *
   * The tables it reads are untouched: a view holds no rows of its own. Use
   * {@link dropViewAsync} to retain the cleanup job instead of waiting on it.
   */
  abstract dropView(name: string, namespacePath?: string[]): Promise<void>;

  /**
   * Start dropping the view named `name` and return the job deleting its
   * definition, without waiting for completion.
   *
   * The name is free before this resolves. When nothing was bound to it, the
   * returned job is already finished and has no id.
   */
  abstract dropViewAsync(name: string, namespacePath?: string[]): Promise<Job>;

  /**
   * The names of the views in one namespace.
   *
   * Names only; a definition comes from {@link describeView}.
   * Iterate lazily with `for await...of`. See {@link Listing} for pagination
   * state and error behavior; options control page size and the starting token.
   */
  abstract listViews(
    namespacePath?: string[],
    options?: ListingOptions,
  ): Listing<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>;

  /** Lazily list immediate child namespaces, 100 per page by default. */
  abstract listNamespaces(
    namespacePath?: string[],
    options?: ListingOptions,
  ): Listing<string>;

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

  /**
   * Iterate lazily over server-side jobs across the database's tables.
   * Use `for await...of`. Options control page size and the starting token.
   * See {@link Listing} for pagination state and error behavior.
   */
  abstract listJobs(options?: ListingOptions): Listing<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>;

  /**
   * Pause a server-side job by id.
   *
   * The job's workers drain and it stays parked until resumed. Resolves to
   * "pausing", "already_paused", or "committing" -- a job finalizing its
   * results cannot be parked; retry shortly.
   */
  abstract pauseJob(jobId: string): Promise<string>;

  /**
   * Resume a paused server-side job by id.
   *
   * Its workers pick their work back up from checkpoints. Resolves to
   * "resumed", "still_pausing" -- the pause's worker drain is not confirmed
   * yet; retry shortly -- or "not_paused".
   */
  abstract resumeJob(jobId: string): Promise<string>;
}

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

    return this.inner.tableNames(
      namespacePath ?? [],
      tableNamesOptions?.startAfter,
      normalizeTableListingLimit(tableNamesOptions?.limit),
    );
  }

  async createMaterializedView(
    name: string,
    source: string,
    options?: {
      select?: MaterializedViewSelect;
      where?: string;
      limit?: number;
      withNoData?: boolean;
    },
  ): Promise<MaterializedView> {
    validateNonNegativeInteger(options?.limit, "limit");
    const innerTable = await this.inner.createMaterializedView(
      name,
      source,
      normalizeSelect(options?.select),
      options?.where,
      options?.limit,
      options?.withNoData ?? false,
    );
    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));
  }

  listMaterializedViews(options: ListingOptions = {}): Listing<string> {
    validateListingOptions(options);
    return new Listing(
      this.inner.listMaterializedViews(options.pageToken, options.pageLimit),
    );
  }

  async dropMaterializedView(
    name: string,
    namespacePath?: string[],
  ): Promise<void> {
    return this.inner.dropMaterializedView(name, namespacePath ?? []);
  }

  async dropMaterializedViewAsync(
    name: string,
    namespacePath?: string[],
  ): Promise<Job> {
    return new Job(
      await this.inner.dropMaterializedViewAsync(name, namespacePath ?? []),
    );
  }

  async createView(
    name: string,
    query: string,
    namespacePath?: string[],
  ): Promise<ViewDescription> {
    return viewDescriptionFromNative(
      await this.inner.createView(name, query, namespacePath ?? []),
    );
  }

  async describeView(
    name: string,
    namespacePath?: string[],
  ): Promise<ViewDescription> {
    return viewDescriptionFromNative(
      await this.inner.describeView(name, namespacePath ?? []),
    );
  }

  async dropView(name: string, namespacePath?: string[]): Promise<void> {
    return this.inner.dropView(name, namespacePath ?? []);
  }

  async dropViewAsync(name: string, namespacePath?: string[]): Promise<Job> {
    return new Job(await this.inner.dropViewAsync(name, namespacePath ?? []));
  }

  listViews(
    namespacePath?: string[],
    options: ListingOptions = {},
  ): Listing<string> {
    validateListingOptions(options);
    return new Listing(
      this.inner.listViews(
        namespacePath ?? [],
        options.pageToken,
        options.pageLimit,
      ),
    );
  }

  listTables(
    namespacePath?: string[],
    options: ListingOptions = {},
  ): Listing<string> {
    validateListingOptions(options);
    return new Listing(
      this.inner.listTables(
        namespacePath ?? [],
        options.pageToken,
        options.pageLimit,
      ),
    );
  }

  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 {
    let storageOptions = options?.storageOptions;
    if (options?.dataStorageVersion !== undefined) {
      storageOptions = {
        ...storageOptions,
        newTableDataStorageVersion: options.dataStorageVersion,
      };
    }

    if (options?.enableV2ManifestPaths !== undefined) {
      storageOptions = {
        ...storageOptions,
        newTableEnableV2ManifestPaths: options.enableV2ManifestPaths
          ? "true"
          : "false",
      };
    }

    return cleanseStorageOptions(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 ?? options;
    }

    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 ?? options;
    }

    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: ListingOptions = {},
  ): Listing<string> {
    validateListingOptions(options);
    return new Listing(
      this.inner.listNamespaces(
        namespacePath ?? [],
        options.pageToken,
        options.pageLimit,
      ),
    );
  }

  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));
  }

  listJobs(options: ListingOptions = {}): Listing<JobInfo> {
    validateListingOptions(options);
    return new Listing(
      this.inner.listJobs(options.pageToken, options.pageLimit),
    );
  }

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

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

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

function normalizeTableListingLimit(limit?: number): number | undefined {
  validateNonNegativeInteger(limit, "limit");
  // The native binding takes u32: larger JS numbers otherwise wrap to zero or
  // a smaller page size. A listing limit is an upper bound, so saturate it.
  return limit === undefined ? undefined : Math.min(limit, 0xffffffff);
}

/**
 * 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 {
    const records = await resolveBlobInputs(
      data as Record<string, unknown>[],
      options?.schema,
    );
    table = makeArrowTable(records, 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 };
  }
}