lancedb/nodejs/src/table.rs

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

use std::collections::HashMap;

use chrono::{DateTime, Utc};

use lancedb::ipc::{ipc_file_to_batches, ipc_file_to_schema};
use lancedb::table::{
    AddDataMode, ColumnAlteration as LanceColumnAlteration,
    FieldMetadataUpdate as LanceFieldMetadataUpdate, FtsToken as LanceDbFtsToken,
    FunctionWorkerResourceBudget as LanceFunctionWorkerResourceBudget, NewColumnTransform, Ref,
    Table as LanceDbTable,
};
use napi::Env;
use napi::bindgen_prelude::*;
use napi::threadsafe_function::{ThreadsafeFunction, ThreadsafeFunctionCallMode};
use napi_derive::napi;

use crate::blob::{BlobFile, copy_blob_buffers, parse_row_ids, spawn_abortable};
use crate::error::NapiErrorExt;
use crate::index::Index;
use crate::merge::NativeMergeInsertBuilder;
use crate::query::{Query, TakeQuery, VectorQuery};
use crate::util::schema_to_buffer;

/// Per-invocation file-compaction settings. All numbers must be safe integers.
#[napi(object)]
pub struct CompactionOptions {
    pub target_rows_per_fragment: Option<f64>,
    pub max_source_rows: Option<f64>,
    pub io_buffer_size: Option<f64>,
    pub batch_size: Option<f64>,
}

impl CompactionOptions {
    fn into_overrides(self) -> napi::Result<lancedb::table::optimize::CompactionOverrides> {
        const MAX_SAFE_INTEGER: u64 = 9_007_199_254_740_991;
        fn integer(
            value: Option<f64>,
            name: &str,
            max: u64,
            allow_zero: bool,
        ) -> napi::Result<Option<u64>> {
            let Some(value) = value else {
                return Ok(None);
            };
            let min = if allow_zero { 0 } else { 1 };
            let max = max.min(MAX_SAFE_INTEGER);
            if !value.is_finite()
                || value.fract() != 0.0
                || value < min as f64
                || value > max as f64
            {
                return Err(napi::Error::from_reason(format!(
                    "{name} must be a safe integer between {min} and {max}"
                )));
            }
            Ok(Some(value as u64))
        }
        let mut options = lancedb::table::optimize::CompactionOverrides::default();
        if let Some(target) = integer(
            self.target_rows_per_fragment,
            "targetRowsPerFragment",
            u32::MAX as u64,
            false,
        )? {
            options.target_rows_per_fragment = Some(target as usize);
        }
        options.max_source_rows = integer(
            self.max_source_rows,
            "maxSourceRows",
            usize::MAX as u64,
            false,
        )?
        .map(|v| v as usize);
        options.io_buffer_size = integer(self.io_buffer_size, "ioBufferSize", u64::MAX, true)?;
        options.batch_size =
            integer(self.batch_size, "batchSize", u32::MAX as u64, false)?.map(|v| v as usize);
        Ok(options)
    }
}

#[napi]
pub struct Table {
    // We keep a duplicate of the table name so we can use it for error
    // messages even if the table has been closed
    pub name: String,
    pub(crate) inner: Option<LanceDbTable>,
}

impl Table {
    pub(crate) fn inner_ref(&self) -> napi::Result<&LanceDbTable> {
        self.inner
            .as_ref()
            .ok_or_else(|| napi::Error::from_reason(format!("Table {} is closed", self.name)))
    }
}

#[napi]
impl Table {
    pub(crate) fn new(table: LanceDbTable) -> Self {
        Self {
            name: table.name().to_string(),
            inner: Some(table),
        }
    }

    #[napi]
    pub fn display(&self) -> String {
        match &self.inner {
            None => format!("ClosedTable({})", self.name),
            Some(inner) => inner.to_string(),
        }
    }

    #[napi]
    pub fn is_open(&self) -> bool {
        self.inner.is_some()
    }

    #[napi]
    pub fn close(&mut self) {
        self.inner.take();
    }

    /// Return Schema as empty Arrow IPC file.
    #[napi(catch_unwind)]
    pub async fn schema(&self) -> napi::Result<Buffer> {
        let schema = self.inner_ref()?.schema().await.default_error()?;
        schema_to_buffer(&schema)
    }

    #[napi(
        catch_unwind,
        ts_args_type = "buf: Buffer, mode: string, progressCallback?: (progress: WriteProgressInfo) => void, allowExternalBlobOutsideBases?: boolean"
    )]
    pub async fn add(
        &self,
        buf: Buffer,
        mode: String,
        progress_callback: Option<ProgressFn>,
        allow_external_blob_outside_bases: Option<bool>,
    ) -> napi::Result<AddResult> {
        let batches = ipc_file_to_batches(buf.to_vec())
            .map_err(|e| napi::Error::from_reason(format!("Failed to read IPC file: {}", e)))?;
        let batches = batches
            .into_iter()
            .map(|batch| {
                batch.map_err(|e| {
                    napi::Error::from_reason(format!(
                        "Failed to read record batch from IPC file: {}",
                        e
                    ))
                })
            })
            .collect::<Result<Vec<_>>>()?;
        let mut op = self.inner_ref()?.add(batches);

        op = if mode == "append" {
            op.mode(AddDataMode::Append)
        } else if mode == "overwrite" {
            op.mode(AddDataMode::Overwrite)
        } else {
            return Err(napi::Error::from_reason(format!("Invalid mode: {}", mode)));
        };

        op = op
            .allow_external_blob_outside_bases(allow_external_blob_outside_bases.unwrap_or(false));

        if let Some(tsfn) = progress_callback {
            op = op.progress(move |p| {
                // NonBlocking: dispatch onto the JS event loop without
                // blocking the writer thread.  With napi-rs's default
                // unbounded queue, events are not dropped — a slow JS
                // callback will just queue them.
                tsfn.call(
                    WriteProgressInfo::from(p),
                    ThreadsafeFunctionCallMode::NonBlocking,
                );
            });
        }

        let res = op.execute().await.default_error()?;
        Ok(res.into())
    }

    #[napi(catch_unwind)]
    pub async fn count_rows(&self, filter: Option<String>) -> napi::Result<i64> {
        self.inner_ref()?
            .count_rows(filter)
            .await
            .map(|val| val as i64)
            .default_error()
    }

    #[napi(catch_unwind)]
    pub async fn delete(&self, predicate: String) -> napi::Result<DeleteResult> {
        let res = self.inner_ref()?.delete(&predicate).await.default_error()?;
        Ok(res.into())
    }

    #[napi(catch_unwind)]
    pub async fn create_index(
        &self,
        index: Option<&Index>,
        column: String,
        replace: Option<bool>,
        wait_timeout_s: Option<i64>,
        name: Option<String>,
        train: Option<bool>,
    ) -> napi::Result<()> {
        let lancedb_index = if let Some(index) = index {
            index.consume()?
        } else {
            lancedb::index::Index::Auto
        };
        let mut builder = self.inner_ref()?.create_index(&[column], lancedb_index);
        if let Some(replace) = replace {
            builder = builder.replace(replace);
        }
        if let Some(timeout) = wait_timeout_s {
            builder =
                builder.wait_timeout(std::time::Duration::from_secs(timeout.try_into().unwrap()));
        }
        if let Some(name) = name {
            builder = builder.name(name);
        }
        if let Some(train) = train {
            builder = builder.train(train);
        }
        builder.execute().await.default_error()
    }

    #[napi(catch_unwind)]
    pub async fn create_index_async(
        &self,
        index: Option<&Index>,
        column: String,
        replace: Option<bool>,
        wait_timeout_s: Option<i64>,
        name: Option<String>,
        train: Option<bool>,
    ) -> napi::Result<crate::job::Job> {
        let lancedb_index = if let Some(index) = index {
            index.consume()?
        } else {
            lancedb::index::Index::Auto
        };
        let mut builder = self.inner_ref()?.create_index(&[column], lancedb_index);
        if let Some(replace) = replace {
            builder = builder.replace(replace);
        }
        if let Some(timeout) = wait_timeout_s {
            builder =
                builder.wait_timeout(std::time::Duration::from_secs(timeout.try_into().unwrap()));
        }
        if let Some(name) = name {
            builder = builder.name(name);
        }
        if let Some(train) = train {
            builder = builder.train(train);
        }
        let job = builder.execute_async().await.default_error()?;
        Ok(crate::job::Job::new(job))
    }

    #[napi(catch_unwind)]
    pub async fn drop_index(&self, index_name: String) -> napi::Result<()> {
        self.inner_ref()?
            .drop_index(&index_name)
            .await
            .default_error()
    }

    #[napi(catch_unwind)]
    pub async fn prewarm_index(&self, index_name: String) -> napi::Result<()> {
        self.inner_ref()?
            .prewarm_index(&index_name)
            .await
            .default_error()
    }

    #[napi(catch_unwind)]
    pub async fn prewarm_data(&self, columns: Option<Vec<String>>) -> napi::Result<()> {
        self.inner_ref()?
            .prewarm_data(columns)
            .await
            .default_error()
    }

    #[napi(catch_unwind)]
    pub async fn wait_for_index(&self, index_names: Vec<String>, timeout_s: i64) -> Result<()> {
        let timeout = std::time::Duration::from_secs(timeout_s.try_into().unwrap());
        let index_names: Vec<&str> = index_names.iter().map(|s| s.as_str()).collect();
        let slice: &[&str] = &index_names;

        self.inner_ref()?
            .wait_for_index(slice, timeout)
            .await
            .default_error()
    }

    #[napi(catch_unwind)]
    pub async fn stats(&self) -> Result<TableStatistics> {
        let stats = self.inner_ref()?.stats().await.default_error()?;
        Ok(stats.into())
    }

    #[napi(catch_unwind)]
    pub async fn initial_storage_options(&self) -> napi::Result<Option<HashMap<String, String>>> {
        Ok(self.inner_ref()?.initial_storage_options().await)
    }

    #[napi(catch_unwind)]
    pub async fn latest_storage_options(&self) -> napi::Result<Option<HashMap<String, String>>> {
        self.inner_ref()?
            .latest_storage_options()
            .await
            .default_error()
    }

    #[napi(catch_unwind)]
    pub async fn update(
        &self,
        only_if: Option<String>,
        columns: Vec<(String, String)>,
    ) -> napi::Result<UpdateResult> {
        let mut op = self.inner_ref()?.update();
        if let Some(only_if) = only_if {
            op = op.only_if(only_if);
        }
        for (column_name, value) in columns {
            op = op.column(column_name, value);
        }
        let res = op.execute().await.default_error()?;
        Ok(res.into())
    }

    #[napi(catch_unwind)]
    pub fn query(&self) -> napi::Result<Query> {
        Ok(Query::new(self.inner_ref()?.query()))
    }

    /// Return a read-only table handle pinned to the current query revision.
    #[napi(catch_unwind)]
    pub async fn query_snapshot(&self) -> napi::Result<Self> {
        let snapshot = self.inner_ref()?.query_snapshot().await.default_error()?;
        Ok(Self::new(snapshot))
    }

    #[napi(catch_unwind)]
    pub fn take_offsets(&self, offsets: Vec<i64>) -> napi::Result<TakeQuery> {
        Ok(TakeQuery::new(
            self.inner_ref()?.take_offsets(
                offsets
                    .into_iter()
                    .map(|o| {
                        u64::try_from(o).map_err(|e| {
                            napi::Error::from_reason(format!(
                                "Failed to convert offset to u64: {}",
                                e
                            ))
                        })
                    })
                    .collect::<Result<Vec<_>>>()?,
            ),
        ))
    }

    #[napi(catch_unwind)]
    pub fn take_row_ids(&self, row_ids: Vec<BigInt>) -> napi::Result<TakeQuery> {
        Ok(TakeQuery::new(
            self.inner_ref()?.take_row_ids(
                row_ids
                    .into_iter()
                    .map(|id| {
                        let (negative, value, lossless) = id.get_u64();
                        if negative {
                            Err(napi::Error::from_reason(
                                "Row id cannot be negative".to_string(),
                            ))
                        } else if !lossless {
                            Err(napi::Error::from_reason(
                                "Row id is too large to fit in u64".to_string(),
                            ))
                        } else {
                            Ok(value)
                        }
                    })
                    .collect::<Result<Vec<_>>>()?,
            ),
        ))
    }

    #[napi(catch_unwind)]
    pub async fn blob_columns(&self) -> napi::Result<Vec<String>> {
        self.inner_ref()?.blob_columns().await.default_error()
    }

    #[napi(catch_unwind)]
    pub fn fetch_blobs<'env>(
        &self,
        env: &'env Env,
        column: String,
        row_ids: Vec<BigInt>,
        signal: Option<AbortSignal>,
    ) -> napi::Result<PromiseRaw<'env, Vec<Option<Buffer>>>> {
        let row_ids = parse_row_ids(row_ids)?;
        let table = self.inner_ref()?.clone();
        spawn_abortable(env, signal, async move {
            let array = table
                .fetch_blobs(column.as_str(), &row_ids)
                .await
                .default_error()?;
            Ok(copy_blob_buffers(array))
        })
    }

    #[napi(catch_unwind)]
    pub fn fetch_blob_files<'env>(
        &self,
        env: &'env Env,
        column: String,
        row_ids: Vec<BigInt>,
        signal: Option<AbortSignal>,
    ) -> napi::Result<PromiseRaw<'env, Vec<Option<BlobFile>>>> {
        let row_ids = parse_row_ids(row_ids)?;
        let table = self.inner_ref()?.clone();
        spawn_abortable(env, signal, async move {
            let files = table
                .fetch_blob_files(column.as_str(), &row_ids)
                .await
                .default_error()?;
            Ok(files
                .into_iter()
                .map(|file| file.map(BlobFile::new))
                .collect())
        })
    }

    #[napi(catch_unwind)]
    pub fn vector_search(&self, vector: Float32Array) -> napi::Result<VectorQuery> {
        self.query()?.nearest_to(vector)
    }

    #[napi(catch_unwind)]
    pub async fn add_columns(
        &self,
        transforms: Vec<AddColumnsSql>,
    ) -> napi::Result<AddColumnsResult> {
        let transforms = transforms
            .into_iter()
            .map(|sql| (sql.name, sql.value_sql))
            .collect::<Vec<_>>();
        let transforms = NewColumnTransform::SqlExpressions(transforms);
        let res = self
            .inner_ref()?
            .add_columns()
            .transform(transforms)
            .execute()
            .await
            .default_error()?;
        Ok(res.into())
    }

    #[napi(catch_unwind)]
    pub async fn add_computed_columns(
        &self,
        columns: Vec<AddColumnsSql>,
    ) -> napi::Result<AddColumnsResult> {
        let table = self.inner_ref()?;
        let mut builder = table.add_columns();
        for column in columns {
            builder = builder.computed(column.name, column.value_sql);
        }
        let res = builder.execute().await.default_error()?;
        Ok(res.into())
    }

    #[napi(catch_unwind)]
    pub async fn refresh_column(
        &self,
        column: String,
        options: Option<RefreshColumnOptions>,
    ) -> napi::Result<RefreshColumnResult> {
        let options = options.unwrap_or_default().try_into()?;
        let res = self
            .inner_ref()?
            .refresh_column_with_options(column, options)
            .await
            .default_error()?;
        Ok(res.into())
    }

    #[napi(catch_unwind)]
    pub async fn refresh_column_async(
        &self,
        column: String,
        options: Option<RefreshColumnOptions>,
    ) -> napi::Result<crate::job::Job> {
        let options = options.unwrap_or_default().try_into()?;
        let job = self
            .inner_ref()?
            .refresh_column_async_with_options(column, options)
            .await
            .default_error()?;
        Ok(crate::job::Job::new(job))
    }

    #[napi(catch_unwind)]
    pub async fn function_errors(
        &self,
        options: Option<FunctionErrorsOptions>,
    ) -> napi::Result<FunctionErrors> {
        let options = options.unwrap_or_default();
        let limit = options
            .limit
            .map(|limit| {
                usize::try_from(limit)
                    .map_err(|_| napi::Error::from_reason("limit must be a non-negative integer"))
            })
            .transpose()?;
        let request = lancedb::function::FunctionErrorsRequest {
            job_id: options.job_id,
            column: options.column,
            limit,
        };
        let errors = self
            .inner_ref()?
            .function_errors(request)
            .await
            .default_error()?;
        Ok(errors.into())
    }

    #[napi(catch_unwind)]
    pub async fn refresh_materialized_view(
        &self,
        source_version: Option<i64>,
        full: Option<bool>,
    ) -> napi::Result<RefreshMaterializedViewResult> {
        let view = lancedb::MaterializedView::from_table(self.inner_ref()?.clone())
            .await
            .default_error()?;
        let mut builder = view.refresh().full(full.unwrap_or(false));
        if let Some(version) = source_version {
            let version = u64::try_from(version).map_err(|_| {
                napi::Error::from_reason("sourceVersion must be a non-negative integer")
            })?;
            builder = builder.source_version(version);
        }
        let result = builder.execute().await.default_error()?;
        Ok(result.into())
    }

    #[napi(catch_unwind)]
    pub async fn materialized_view_definition(&self) -> napi::Result<String> {
        let inner = self.inner_ref()?.clone();
        let view = lancedb::MaterializedView::from_table(inner)
            .await
            .default_error()?;
        Ok(lancedb::materialized_view::definition_metadata_from_sql(
            view.definition_sql(),
        ))
    }

    #[napi(catch_unwind)]
    pub async fn add_columns_with_schema(
        &self,
        schema_buf: Buffer,
    ) -> napi::Result<AddColumnsResult> {
        let schema = ipc_file_to_schema(schema_buf.to_vec())
            .map_err(|e| napi::Error::from_reason(format!("Failed to read IPC schema: {}", e)))?;

        let transforms = NewColumnTransform::AllNulls(schema);
        let res = self
            .inner_ref()?
            .add_columns()
            .transform(transforms)
            .execute()
            .await
            .default_error()?;
        Ok(res.into())
    }

    #[napi(catch_unwind)]
    pub async fn alter_columns(
        &self,
        alterations: Vec<ColumnAlteration>,
    ) -> napi::Result<AlterColumnsResult> {
        for alteration in &alterations {
            if alteration.rename.is_none()
                && alteration.nullable.is_none()
                && alteration.data_type.is_none()
            {
                return Err(napi::Error::from_reason(
                    "Alteration must have a 'rename', 'dataType', or 'nullable' field.",
                ));
            }
        }
        let alterations = alterations
            .into_iter()
            .map(LanceColumnAlteration::try_from)
            .collect::<std::result::Result<Vec<_>, String>>()
            .map_err(napi::Error::from_reason)?;

        let res = self
            .inner_ref()?
            .alter_columns(&alterations)
            .await
            .default_error()?;
        Ok(res.into())
    }

    #[napi(catch_unwind)]
    pub async fn update_field_metadata(
        &self,
        updates: Vec<FieldMetadataUpdate>,
    ) -> napi::Result<UpdateFieldMetadataResult> {
        let updates = updates
            .into_iter()
            .map(LanceFieldMetadataUpdate::from)
            .collect::<Vec<_>>();
        let res = self
            .inner_ref()?
            .update_field_metadata(&updates)
            .await
            .default_error()?;
        Ok(res.into())
    }

    #[napi(catch_unwind)]
    pub async fn drop_columns(&self, columns: Vec<String>) -> napi::Result<DropColumnsResult> {
        let col_refs = columns.iter().map(String::as_str).collect::<Vec<_>>();
        let res = self
            .inner_ref()?
            .drop_columns(&col_refs)
            .await
            .default_error()?;
        Ok(res.into())
    }

    #[napi(catch_unwind)]
    pub async fn set_unenforced_primary_key(&self, columns: Vec<String>) -> napi::Result<()> {
        self.inner_ref()?
            .set_unenforced_primary_key(columns)
            .await
            .default_error()
    }

    #[napi(catch_unwind, js_name = "enableRTE")]
    pub async fn enable_rte(&self, spec: RTEWriteSpec) -> napi::Result<()> {
        let native_spec = lancedb::table::RTEWriteSpec::try_from(spec)?;
        self.inner_ref()?
            .enable_rte(native_spec)
            .await
            .default_error()
    }

    #[napi(catch_unwind, js_name = "disableRTE")]
    pub async fn disable_rte(&self) -> napi::Result<()> {
        self.inner_ref()?.disable_rte().await.default_error()
    }

    #[napi(catch_unwind, js_name = "getRTESpec")]
    pub async fn get_rte_spec(&self) -> napi::Result<Option<RTEWriteSpec>> {
        let spec = self.inner_ref()?.get_rte_spec().await.default_error()?;
        Ok(spec.map(RTEWriteSpec::from))
    }

    #[napi(catch_unwind, js_name = "closeRTEWriters")]
    pub async fn close_rte_writers(&self) -> napi::Result<()> {
        self.inner_ref()?.close_rte_writers().await.default_error()
    }

    #[napi(catch_unwind, js_name = "flushRTE")]
    pub async fn flush_rte(&self) -> napi::Result<()> {
        self.inner_ref()?.flush_rte().await.default_error()
    }

    #[napi(catch_unwind, js_name = "compactRTE")]
    pub async fn compact_rte(&self) -> napi::Result<()> {
        self.inner_ref()?.compact_rte().await.default_error()
    }

    #[napi(catch_unwind, js_name = "checkpointRTE")]
    pub async fn checkpoint_rte(&self) -> napi::Result<()> {
        self.inner_ref()?.checkpoint_rte().await.default_error()
    }

    #[napi(catch_unwind, js_name = "getRTEStats")]
    pub async fn get_rte_stats(
        &self,
        include_sstable_rows: bool,
    ) -> napi::Result<Option<RTEStats>> {
        let stats = self
            .inner_ref()?
            .get_rte_stats(include_sstable_rows)
            .await
            .default_error()?;
        Ok(stats.map(RTEStats::from))
    }

    #[napi(catch_unwind)]
    pub async fn version(&self) -> napi::Result<i64> {
        self.inner_ref()?
            .version()
            .await
            .map(|val| val as i64)
            .default_error()
    }

    #[napi(catch_unwind)]
    pub async fn checkout_current(&self) -> napi::Result<Self> {
        let table = self.inner_ref()?.checkout_current().await.default_error()?;
        Ok(Self::new(table))
    }

    #[napi(catch_unwind)]
    pub async fn checkout(&self, version: i64) -> napi::Result<()> {
        self.inner_ref()?
            .checkout(version as u64)
            .await
            .default_error()
    }

    #[napi(catch_unwind)]
    pub async fn checkout_tag(&self, tag: String) -> napi::Result<()> {
        self.inner_ref()?
            .checkout_tag(tag.as_str())
            .await
            .default_error()
    }

    #[napi(catch_unwind)]
    pub async fn checkout_latest(&self) -> napi::Result<()> {
        self.inner_ref()?.checkout_latest().await.default_error()
    }

    #[napi(catch_unwind)]
    pub async fn list_versions(&self) -> napi::Result<Vec<Version>> {
        self.inner_ref()?
            .list_versions()
            .await
            .map(|versions| {
                versions
                    .iter()
                    .map(|version| Version {
                        version: version.version as i64,
                        timestamp: version.timestamp.timestamp_micros(),
                        metadata: version
                            .metadata
                            .iter()
                            .map(|(k, v)| (k.clone(), v.clone()))
                            .collect(),
                    })
                    .collect()
            })
            .default_error()
    }

    #[napi(catch_unwind)]
    pub async fn restore(&self) -> napi::Result<()> {
        self.inner_ref()?.restore().await.default_error()
    }

    #[napi(catch_unwind)]
    pub async fn tags(&self) -> napi::Result<Tags> {
        Ok(Tags {
            inner: self.inner_ref()?.clone(),
        })
    }

    #[napi(catch_unwind)]
    pub async fn branches(&self) -> napi::Result<Branches> {
        Ok(Branches {
            inner: self.inner_ref()?.clone(),
        })
    }

    /// The branch this handle is scoped to, or `null` for the main branch.
    #[napi]
    pub fn current_branch(&self) -> napi::Result<Option<String>> {
        Ok(self.inner_ref()?.current_branch())
    }

    #[napi(catch_unwind)]
    pub async fn optimize(
        &self,
        before_timestamp_ms: Option<i64>,
        delete_unverified: Option<bool>,
        compaction_options: Option<CompactionOptions>,
    ) -> napi::Result<OptimizeStats> {
        let inner = self.inner_ref()?;

        let before_timestamp = before_timestamp_ms
            .map(|ms| {
                DateTime::from_timestamp_millis(ms).ok_or_else(|| {
                    napi::Error::from_reason(format!(
                        "cleanupOlderThan timestamp is out of range: {ms}"
                    ))
                })
            })
            .transpose()?;

        let stats = inner
            .optimize_full_with_overrides(
                compaction_options
                    .map(CompactionOptions::into_overrides)
                    .transpose()?
                    .unwrap_or_default(),
                None,
                before_timestamp,
                delete_unverified,
            )
            .await
            .default_error()?;
        let compaction_stats = stats.compaction.unwrap();
        let prune_stats = stats.prune.unwrap();
        Ok(OptimizeStats {
            compaction: CompactionStats {
                files_added: compaction_stats.files_added as i64,
                files_removed: compaction_stats.files_removed as i64,
                fragments_added: compaction_stats.fragments_added as i64,
                fragments_removed: compaction_stats.fragments_removed as i64,
            },
            prune: RemovalStats {
                bytes_removed: prune_stats.bytes_removed as i64,
                old_versions_removed: prune_stats.old_versions as i64,
            },
        })
    }

    #[napi(catch_unwind)]
    pub async fn list_indices(&self) -> napi::Result<Vec<IndexConfig>> {
        Ok(self
            .inner_ref()?
            .list_indices()
            .await
            .default_error()?
            .into_iter()
            .map(IndexConfig::from)
            .collect::<Vec<_>>())
    }

    #[napi(catch_unwind)]
    pub async fn tokenize(
        &self,
        query: String,
        column: Option<String>,
        index_name: Option<String>,
    ) -> napi::Result<Vec<FtsToken>> {
        let table = self.inner_ref()?;
        let tokens = match (column.as_deref(), index_name.as_deref()) {
            (Some(_), Some(_)) | (None, None) => {
                return Err(napi::Error::from_reason(
                    "Specify exactly one of 'column' or 'indexName'",
                ));
            }
            (Some(column), None) => table.tokenize_with_column(&query, column).await,
            (None, Some(index_name)) => table.tokenize(&query, index_name).await,
        }
        .default_error()?;
        Ok(tokens.into_iter().map(FtsToken::from).collect())
    }

    #[napi(catch_unwind)]
    pub async fn index_stats(&self, index_name: String) -> napi::Result<Option<IndexStatistics>> {
        let tbl = self.inner_ref()?;
        let stats = tbl.index_stats(&index_name).await.default_error()?;
        Ok(stats.map(IndexStatistics::from))
    }

    #[napi(catch_unwind)]
    pub fn merge_insert(&self, on: Vec<String>) -> napi::Result<NativeMergeInsertBuilder> {
        let on: Vec<_> = on.iter().map(String::as_str).collect();
        Ok(self.inner_ref()?.merge_insert(on.as_slice()).into())
    }

    #[napi(catch_unwind)]
    pub async fn uses_v2_manifest_paths(&self) -> napi::Result<bool> {
        self.inner_ref()?
            .as_native()
            .ok_or_else(|| lancedb::Error::NotSupported {
                message: "uses_v2_manifest_paths is not supported in LanceDB Enterprise.".into(),
            })
            .default_error()?
            .uses_v2_manifest_paths()
            .await
            .default_error()
    }

    #[napi(catch_unwind)]
    pub async fn migrate_manifest_paths_v2(&self) -> napi::Result<()> {
        self.inner_ref()?
            .as_native()
            .ok_or_else(|| lancedb::Error::NotSupported {
                message: "migrate_manifest_paths_v2 is not supported in LanceDB Enterprise.".into(),
            })
            .default_error()?
            .migrate_manifest_paths_v2()
            .await
            .default_error()
    }
}

#[napi(object)]
/// A description of an index currently configured on a column
pub struct IndexConfig {
    /// The name of the index
    pub name: String,
    /// The type of the index
    pub index_type: String,
    /// The columns in the index
    ///
    /// Currently this is always an array of size 1. In the future there may
    /// be more columns to represent composite indices.
    pub columns: Vec<String>,
    /// The UUID of the first segment of the index.
    ///
    /// `undefined` in LanceDB Enterprise, which does not yet surface this.
    pub index_uuid: Option<String>,
    /// The protobuf type URL, a precise type identifier for the index.
    ///
    /// `undefined` in LanceDB Enterprise.
    pub type_url: Option<String>,
    /// When the index was created.
    ///
    /// `undefined` in LanceDB Enterprise or indices created before timestamps were tracked.
    pub created_at: Option<DateTime<Utc>>,
    /// The number of rows indexed, across all segments.
    ///
    /// May be `undefined` in LanceDB Enterprise if the server does not report it.
    pub num_indexed_rows: Option<i64>,
    /// The number of rows not yet covered by this index.
    ///
    /// May be `undefined` in LanceDB Enterprise if the server does not report it.
    pub num_unindexed_rows: Option<i64>,
    /// The total size in bytes of all index files across all segments.
    ///
    /// `undefined` in LanceDB Enterprise or indices without size tracking.
    pub size_bytes: Option<i64>,
    /// The number of segments that make up the index.
    ///
    /// `undefined` in LanceDB Enterprise.
    pub num_segments: Option<i32>,
    /// The on-disk index format version.
    ///
    /// `undefined` in LanceDB Enterprise.
    pub index_version: Option<i32>,
    /// Index-type-specific details parsed as a JavaScript object.
    ///
    /// Falls back to a raw string if JSON parsing fails. `undefined` in
    /// LanceDB Enterprise or when details are unavailable.
    pub index_details: Option<serde_json::Value>,
}

impl From<lancedb::index::IndexConfig> for IndexConfig {
    fn from(value: lancedb::index::IndexConfig) -> Self {
        let index_type = format!("{:?}", value.index_type);
        Self {
            index_type,
            columns: value.columns,
            name: value.name,
            index_uuid: value.index_uuid,
            type_url: value.type_url,
            created_at: value.created_at,
            num_indexed_rows: value.num_indexed_rows.map(|n| n as i64),
            num_unindexed_rows: value.num_unindexed_rows.map(|n| n as i64),
            size_bytes: value.size_bytes.map(|n| n as i64),
            num_segments: value.num_segments.map(|n| n as i32),
            index_version: value.index_version,
            index_details: value
                .index_details
                .and_then(|s| serde_json::from_str(&s).ok()),
        }
    }
}

#[napi(object)]
/// A token produced by the tokenizer configured on a full-text search index.
pub struct FtsToken {
    /// The token text after the index tokenizer has applied its filters.
    pub text: String,
    /// The token position used by full-text query matching.
    pub position: u32,
}

impl From<LanceDbFtsToken> for FtsToken {
    fn from(token: LanceDbFtsToken) -> Self {
        Self {
            text: token.text,
            position: token.position,
        }
    }
}

/// Specification enabling the realtime engine (RTE), which routes
/// `mergeInsert` through Lance's MemWAL LSM-style write path.
///
/// `specType` must be `"bucket"`, `"identity"`, or `"unsharded"`. For
/// `"bucket"`, `column` and `numBuckets` are required; for `"identity"`,
/// `column` is required.
#[napi(object, js_name = "RTEWriteSpec")]
#[derive(Clone, Debug)]
pub struct RTEWriteSpec {
    /// One of `"bucket"`, `"identity"`, or `"unsharded"`.
    pub spec_type: String,
    /// Bucket and identity variants: the sharding column.
    pub column: Option<String>,
    /// Bucket variant: the number of buckets, in `[1, 1024]`.
    pub num_buckets: Option<u32>,
    /// Indexes the RTE keeps up to date. Omitted resolves every
    /// maintainable index on install; an empty array means none.
    pub maintained_indexes: Option<Vec<String>>,
    /// Default `ShardWriter` configuration recorded in the MemWAL index.
    pub writer_config_defaults: Option<HashMap<String, String>>,
}

impl TryFrom<RTEWriteSpec> for lancedb::table::RTEWriteSpec {
    type Error = napi::Error;

    fn try_from(value: RTEWriteSpec) -> napi::Result<Self> {
        let writer_config_defaults = value.writer_config_defaults.unwrap_or_default();
        let spec = match value.spec_type.as_str() {
            "bucket" => {
                let column = value.column.ok_or_else(|| {
                    napi::Error::from_reason("RTEWriteSpec bucket requires `column`")
                })?;
                let num_buckets = value.num_buckets.ok_or_else(|| {
                    napi::Error::from_reason("RTEWriteSpec bucket requires `numBuckets`")
                })?;
                Self::bucket(column, num_buckets)
            }
            "identity" => {
                let column = value.column.ok_or_else(|| {
                    napi::Error::from_reason("RTEWriteSpec identity requires `column`")
                })?;
                Self::identity(column)
            }
            "unsharded" => Self::unsharded(),
            other => {
                return Err(napi::Error::from_reason(format!(
                    "RTEWriteSpec `specType` must be 'bucket', 'identity', or 'unsharded', got '{}'",
                    other
                )));
            }
        };
        Ok(spec
            .with_maintained_indexes(value.maintained_indexes)
            .with_writer_config_defaults(writer_config_defaults))
    }
}

impl From<lancedb::table::RTEWriteSpec> for RTEWriteSpec {
    fn from(spec: lancedb::table::RTEWriteSpec) -> Self {
        use lancedb::table::RTEWriteSpec as Native;
        match spec {
            Native::Bucket {
                column,
                num_buckets,
                maintained_indexes,
                writer_config_defaults,
            } => Self {
                spec_type: "bucket".to_string(),
                column: Some(column),
                num_buckets: Some(num_buckets),
                maintained_indexes,
                writer_config_defaults: Some(writer_config_defaults),
            },
            Native::Identity {
                column,
                maintained_indexes,
                writer_config_defaults,
            } => Self {
                spec_type: "identity".to_string(),
                column: Some(column),
                num_buckets: None,
                maintained_indexes,
                writer_config_defaults: Some(writer_config_defaults),
            },
            Native::Unsharded {
                maintained_indexes,
                writer_config_defaults,
            } => Self {
                spec_type: "unsharded".to_string(),
                column: None,
                num_buckets: None,
                maintained_indexes,
                writer_config_defaults: Some(writer_config_defaults),
            },
        }
    }
}

/// One SSTable.
#[napi(object)]
#[derive(Clone, Debug)]
pub struct SsTableStats {
    /// The generation number. Increases as memtables are frozen into SSTables.
    pub generation: i64,
    /// On-disk size of the SSTable.
    pub bytes: i64,
    /// Present only when `includeSstableRows` was requested. Off by default
    /// because each count opens an uncached Lance dataset.
    pub rows: Option<i64>,
}

impl From<lancedb::table::SsTableStats> for SsTableStats {
    fn from(g: lancedb::table::SsTableStats) -> Self {
        Self {
            generation: g.generation as i64,
            bytes: g.bytes as i64,
            rows: g.rows.map(|r| r as i64),
        }
    }
}

/// One in-memory memtable.
#[napi(object)]
#[derive(Clone, Debug)]
pub struct MemtableStats {
    /// The generation this memtable will become once frozen.
    pub generation: i64,
    /// Rows currently buffered.
    pub rows: i64,
    /// Estimated in-memory size.
    pub bytes: i64,
    /// Record batches currently buffered.
    pub batches: i64,
    /// Names of the indexes this memtable carries. An absent name is the whole
    /// answer to "why is my fresh-tier search on that column brute-force".
    pub indexes: Vec<String>,
}

impl From<lancedb::table::MemtableStats> for MemtableStats {
    fn from(m: lancedb::table::MemtableStats) -> Self {
        Self {
            generation: m.generation as i64,
            rows: m.rows as i64,
            bytes: m.bytes as i64,
            batches: m.batches as i64,
            indexes: m.indexes,
        }
    }
}

/// Live state of one table shard. A table is N table shards on one node; flattening to a
/// single number hides the one hot table shard that is usually why someone opened
/// this endpoint.
#[napi(object)]
#[derive(Clone, Debug)]
pub struct TableShardStats {
    /// The id of this table shard.
    pub shard_id: String,
    /// `"Active"` or `"Sealed"` (drop-table 2PC in flight).
    pub status: String,
    /// Epoch of the writer that currently owns the shard.
    pub writer_epoch: i64,
    /// Version of the shard manifest these numbers were read from.
    pub manifest_version: i64,
    /// The generation the active memtable will become.
    pub current_generation: i64,
    /// WAL position replay resumes from.
    pub replay_after_wal_entry_position: i64,
    /// Highest WAL position the writer has seen. The difference against
    /// `replayAfterWalEntryPosition` is the WAL lag.
    pub wal_entry_position_last_seen: i64,
    /// SSTables not yet merged into the base table.
    pub sstables: Vec<SsTableStats>,
    /// Whether a pass owns this table shard's compaction latch right now. Says *a*
    /// driver is running, not *whose*, and the latch is held from dispatch —
    /// including while the pass queues for a pod-wide compactor permit. Read it
    /// as "do not pile on", never as "mine is progressing".
    pub compacting: bool,
    /// Oldest first, active last. Absent for a `"Sealed"` table shard, whose
    /// in-memory state is torn down.
    pub memtables: Option<Vec<MemtableStats>>,
}

impl From<lancedb::table::TableShardStats> for TableShardStats {
    fn from(b: lancedb::table::TableShardStats) -> Self {
        Self {
            shard_id: b.shard_id,
            status: b.status,
            writer_epoch: b.writer_epoch as i64,
            manifest_version: b.manifest_version as i64,
            current_generation: b.current_generation as i64,
            replay_after_wal_entry_position: b.replay_after_wal_entry_position as i64,
            wal_entry_position_last_seen: b.wal_entry_position_last_seen as i64,
            sstables: b.sstables.into_iter().map(Into::into).collect(),
            compacting: b.compacting,
            memtables: b
                .memtables
                .map(|ms| ms.into_iter().map(Into::into).collect()),
        }
    }
}

/// Live per-table-shard RTE state, as returned by `Table#getRTEStats`.
///
/// Nothing here is derived: sums and differences (total SSTable bytes, WAL lag) are
/// the caller's to compute.
#[napi(object, js_name = "RTEStats")]
#[derive(Clone, Debug)]
pub struct RTEStats {
    /// One entry per table shard backing this table.
    pub table_shards: Vec<TableShardStats>,
}

impl From<lancedb::table::RTEStats> for RTEStats {
    fn from(stats: lancedb::table::RTEStats) -> Self {
        Self {
            table_shards: stats.table_shards.into_iter().map(Into::into).collect(),
        }
    }
}

/// Statistics about a compaction operation.
#[napi(object)]
#[derive(Clone, Debug)]
pub struct CompactionStats {
    /// The number of fragments removed
    pub fragments_removed: i64,
    /// The number of new, compacted fragments added
    pub fragments_added: i64,
    /// The number of data files removed
    pub files_removed: i64,
    /// The number of new, compacted data files added
    pub files_added: i64,
}

/// Statistics about a cleanup operation
#[napi(object)]
#[derive(Clone, Debug)]
pub struct RemovalStats {
    /// The number of bytes removed
    pub bytes_removed: i64,
    /// The number of old versions removed
    pub old_versions_removed: i64,
}

/// Statistics about an optimize operation
#[napi(object)]
#[derive(Clone, Debug)]
pub struct OptimizeStats {
    /// Statistics about the compaction operation
    pub compaction: CompactionStats,
    /// Statistics about the removal operation
    pub prune: RemovalStats,
}

/// Progress snapshot for a write operation, delivered to the JS callback
/// passed to `Table.add`.
#[napi(object)]
#[derive(Clone, Debug)]
pub struct WriteProgressInfo {
    /// Number of rows written so far.
    pub output_rows: i64,
    /// Number of bytes written so far.
    pub output_bytes: i64,
    /// Total rows expected, if the input source reports it.
    /// Always set on the final callback (where `done` is `true`).
    pub total_rows: Option<i64>,
    /// Wall-clock seconds since monitoring started.
    pub elapsed_seconds: f64,
    /// Number of parallel write tasks currently in flight.
    pub active_tasks: i64,
    /// Total number of parallel write tasks (the write parallelism).
    pub total_tasks: i64,
    /// `true` for the final callback; `false` otherwise.
    pub done: bool,
}

impl From<&lancedb::table::write_progress::WriteProgress> for WriteProgressInfo {
    fn from(p: &lancedb::table::write_progress::WriteProgress) -> Self {
        Self {
            output_rows: p.output_rows() as i64,
            output_bytes: p.output_bytes() as i64,
            total_rows: p.total_rows().map(|n| n as i64),
            elapsed_seconds: p.elapsed().as_secs_f64(),
            active_tasks: p.active_tasks() as i64,
            total_tasks: p.total_tasks() as i64,
            done: p.done(),
        }
    }
}

type ProgressFn = ThreadsafeFunction<WriteProgressInfo, (), WriteProgressInfo, Status, false>;

///  A definition of a column alteration. The alteration changes the column at
/// `path` to have the new name `name`, to be nullable if `nullable` is true,
/// and to have the data type `data_type`. At least one of `rename` or `nullable`
/// must be provided.
#[napi(object)]
pub struct ColumnAlteration {
    /// The path to the column to alter. This is a dot-separated path to the column.
    /// If it is a top-level column then it is just the name of the column. If it is
    /// a nested column then it is the path to the column, e.g. "a.b.c" for a column
    /// `c` nested inside a column `b` nested inside a column `a`.
    pub path: String,
    /// The new name of the column. If not provided then the name will not be changed.
    /// This must be distinct from the names of all other columns in the table.
    pub rename: Option<String>,
    /// A new data type for the column. If not provided then the data type will not be changed.
    /// Changing data types is limited to casting to the same general type. For example, these
    /// changes are valid:
    /// * `int32` -> `int64` (integers)
    /// * `double` -> `float` (floats)
    /// * `string` -> `large_string` (strings)
    /// But these changes are not:
    /// * `int32` -> `double` (mix integers and floats)
    /// * `string` -> `int32` (mix strings and integers)
    pub data_type: Option<String>,
    /// Set the new nullability. Note that a nullable column cannot be made non-nullable.
    pub nullable: Option<bool>,
}

/// A per-field metadata update, addressed by dot-path. Merges into the field's
/// existing metadata by default; a `null` value deletes a key, and `replace`
/// swaps the field's entire metadata map.
#[napi(object)]
pub struct FieldMetadataUpdate {
    /// Dot-separated path to the field (e.g. "embedding" or "a.b.c").
    pub path: String,
    /// Metadata keys to set; a `null` value deletes that key.
    pub metadata: HashMap<String, Option<String>>,
    /// If true, replace the field's entire metadata map instead of merging.
    pub replace: Option<bool>,
}

impl From<FieldMetadataUpdate> for LanceFieldMetadataUpdate {
    fn from(js: FieldMetadataUpdate) -> Self {
        Self {
            path: js.path,
            metadata: js.metadata,
            replace: js.replace.unwrap_or(false),
        }
    }
}

impl TryFrom<ColumnAlteration> for LanceColumnAlteration {
    type Error = String;
    fn try_from(js: ColumnAlteration) -> std::result::Result<Self, Self::Error> {
        let ColumnAlteration {
            path,
            rename,
            nullable,
            data_type,
        } = js;
        let data_type = if let Some(data_type) = data_type {
            Some(
                lancedb::utils::string_to_datatype(&data_type)
                    .ok_or_else(|| format!("Invalid data type: {}", data_type))?,
            )
        } else {
            None
        };
        Ok(Self {
            path,
            rename,
            nullable,
            data_type,
        })
    }
}

/// A definition of a new column to add to a table.
#[napi(object)]
pub struct AddColumnsSql {
    /// The name of the new column.
    pub name: String,
    /// The values to populate the new column with, as a SQL expression.
    /// The expression can reference other columns in the table.
    pub value_sql: String,
}

#[napi(object)]
pub struct IndexStatistics {
    /// The number of rows indexed by the index
    pub num_indexed_rows: f64,
    /// The number of rows not indexed
    pub num_unindexed_rows: f64,
    /// The type of the index
    pub index_type: String,
    /// The type of the distance function used by the index. This is only
    /// present for vector indices. Scalar and full text search indices do
    /// not have a distance function.
    pub distance_type: Option<String>,
    /// The number of parts this index is split into.
    pub num_indices: Option<u32>,
}
impl From<lancedb::index::IndexStatistics> for IndexStatistics {
    fn from(value: lancedb::index::IndexStatistics) -> Self {
        Self {
            num_indexed_rows: value.num_indexed_rows as f64,
            num_unindexed_rows: value.num_unindexed_rows as f64,
            index_type: value.index_type.to_string(),
            distance_type: value.distance_type.map(|d| d.to_string()),
            num_indices: value.num_indices,
        }
    }
}

#[napi(object)]
pub struct TableStatistics {
    /// The total size, in bytes, of the table's data files, index files, and
    /// overlay files
    ///
    /// Read from the manifest, so this excludes deletion files and manifests.
    pub total_bytes: i64,

    /// The number of rows in the table
    pub num_rows: i64,

    /// The number of rows marked as deleted across all fragments of the table
    ///
    /// These rows are not included in `numRows`, but still occupy space on disk
    /// until the table is compacted, so a large value here indicates that the
    /// table should be optimized. Fragments in which every row was deleted are
    /// dropped outright, so their rows are not counted here.
    ///
    /// Absent (`undefined`) when the backend does not report deletion counts.
    pub num_deleted_rows: Option<i64>,

    /// The number of indices in the table
    pub num_indices: i64,

    /// Statistics on table fragments
    pub fragment_stats: FragmentStatistics,
}

#[napi(object)]
pub struct FragmentStatistics {
    /// The number of fragments in the table
    pub num_fragments: i64,

    /// The number of uncompacted fragments in the table
    pub num_small_fragments: i64,

    /// Statistics on the number of rows in the table fragments
    pub lengths: FragmentSummaryStats,
}

#[napi(object)]
pub struct FragmentSummaryStats {
    /// The number of rows in the fragment with the fewest rows
    pub min: i64,

    /// The number of rows in the fragment with the most rows
    pub max: i64,

    /// The mean number of rows in the fragments
    pub mean: i64,

    /// The 25th percentile of number of rows in the fragments
    pub p25: i64,

    /// The 50th percentile of number of rows in the fragments
    pub p50: i64,

    /// The 75th percentile of number of rows in the fragments
    pub p75: i64,

    /// The 99th percentile of number of rows in the fragments
    pub p99: i64,
}

impl From<lancedb::table::TableStatistics> for TableStatistics {
    fn from(v: lancedb::table::TableStatistics) -> Self {
        Self {
            total_bytes: v.total_bytes as i64,
            num_rows: v.num_rows as i64,
            num_deleted_rows: v.num_deleted_rows.map(|n| n as i64),
            num_indices: v.num_indices as i64,
            fragment_stats: FragmentStatistics {
                num_fragments: v.fragment_stats.num_fragments as i64,
                num_small_fragments: v.fragment_stats.num_small_fragments as i64,
                lengths: FragmentSummaryStats {
                    min: v.fragment_stats.lengths.min as i64,
                    max: v.fragment_stats.lengths.max as i64,
                    mean: v.fragment_stats.lengths.mean as i64,
                    p25: v.fragment_stats.lengths.p25 as i64,
                    p50: v.fragment_stats.lengths.p50 as i64,
                    p75: v.fragment_stats.lengths.p75 as i64,
                    p99: v.fragment_stats.lengths.p99 as i64,
                },
            },
        }
    }
}

#[napi(object)]
pub struct Version {
    pub version: i64,
    pub timestamp: i64,
    pub metadata: HashMap<String, String>,
}

#[napi(object)]
pub struct UpdateResult {
    pub rows_updated: i64,
    pub version: i64,
}

impl From<lancedb::table::UpdateResult> for UpdateResult {
    fn from(value: lancedb::table::UpdateResult) -> Self {
        Self {
            rows_updated: value.rows_updated as i64,
            version: value.version as i64,
        }
    }
}

#[napi(object)]
pub struct AddResult {
    pub version: i64,
}

impl From<lancedb::table::AddResult> for AddResult {
    fn from(value: lancedb::table::AddResult) -> Self {
        Self {
            version: value.version as i64,
        }
    }
}

#[napi(object)]
pub struct DeleteResult {
    pub num_deleted_rows: i64,
    pub version: i64,
}

impl From<lancedb::table::DeleteResult> for DeleteResult {
    fn from(value: lancedb::table::DeleteResult) -> Self {
        Self {
            num_deleted_rows: value.num_deleted_rows as i64,
            version: value.version as i64,
        }
    }
}

#[napi(object)]
pub struct MergeResult {
    pub version: i64,
    pub num_inserted_rows: i64,
    pub num_updated_rows: i64,
    pub num_deleted_rows: i64,
    pub num_attempts: i64,
    pub num_rows: i64,
}

impl From<lancedb::table::MergeResult> for MergeResult {
    fn from(value: lancedb::table::MergeResult) -> Self {
        Self {
            version: value.version as i64,
            num_inserted_rows: value.num_inserted_rows as i64,
            num_updated_rows: value.num_updated_rows as i64,
            num_deleted_rows: value.num_deleted_rows as i64,
            num_attempts: value.num_attempts as i64,
            num_rows: value.num_rows as i64,
        }
    }
}

#[napi(object)]
pub struct AddColumnsResult {
    pub version: i64,
}

#[napi(object)]
pub struct RefreshColumnResult {
    pub rows_filled: i64,
    pub version: i64,
}

/// Which per-row Function errors to list; every filter is optional.
#[napi(object)]
#[derive(Clone, Debug, Default)]
pub struct FunctionErrorsOptions {
    /// Only errors recorded by this job.
    pub job_id: Option<String>,
    /// Only errors on this column.
    pub column: Option<String>,
    /// At most this many records (server default 10000, cap 100000).
    pub limit: Option<i64>,
}

/// Aggregate worker resources for one Enterprise Function refresh.
#[napi(object)]
#[derive(Clone, Debug)]
pub struct FunctionWorkerResourceBudget {
    /// Aggregate CPU budget in millicores; 1000 is one CPU.
    pub cpu_millis: f64,
    /// Aggregate memory budget in bytes.
    pub memory_bytes: f64,
    /// Aggregate GPU budget. Omitted means zero, a CPU-only refresh.
    pub gpu_count: Option<f64>,
}

fn resource_budget_integer(value: f64, name: &str) -> napi::Result<u64> {
    const MAX_SAFE_INTEGER: f64 = 9_007_199_254_740_991.0;
    if !value.is_finite() || value.fract() != 0.0 || !(1.0..=MAX_SAFE_INTEGER).contains(&value) {
        return Err(napi::Error::from_reason(format!(
            "{name} must be a positive safe integer"
        )));
    }
    Ok(value as u64)
}

fn resource_budget_gpu_count(value: Option<f64>) -> napi::Result<u32> {
    let Some(value) = value else {
        return Ok(0);
    };
    if !value.is_finite() || value.fract() != 0.0 || !(0.0..=u32::MAX as f64).contains(&value) {
        return Err(napi::Error::from_reason(
            "resourceBudget.gpuCount must be an integer between 0 and 4294967295",
        ));
    }
    Ok(value as u32)
}

fn positive_u32(value: Option<f64>, name: &str) -> napi::Result<Option<u32>> {
    value
        .map(|value| {
            if !value.is_finite()
                || value.fract() != 0.0
                || !(1.0..=u32::MAX as f64).contains(&value)
            {
                return Err(napi::Error::from_reason(format!(
                    "{name} must be an integer between 1 and 4294967295"
                )));
            }
            Ok(value as u32)
        })
        .transpose()
}

/// Options for `Table.refreshColumn` and `Table.refreshColumnAsync`. Every
/// option is LanceDB Enterprise only; a LanceDB Embedded refresh ignores
/// `batchSize`, `gpuClass` and `gpuCount` and rejects the rest, including
/// `resourceBudget`.
#[napi(object)]
#[derive(Clone, Debug, Default)]
pub struct RefreshColumnOptions {
    /// Only refresh the rows matching this SQL filter.
    #[napi(js_name = "where")]
    pub where_clause: Option<String>,
    /// Maximum rows per Function call; must be greater than zero.
    pub batch_size: Option<u32>,
    /// Override the registered Function's maximum concurrent user-code calls
    /// for this logical execution.
    pub max_concurrency: Option<f64>,
    /// GPU class for Function workers.
    pub gpu_class: Option<String>,
    /// Complete aggregate CPU and memory budget for a registered Function
    /// refresh. `gpuCount` inside the budget defaults to zero.
    pub resource_budget: Option<FunctionWorkerResourceBudget>,
    /// Positive aggregate GPU budget for Function workers.
    pub gpu_count: Option<u32>,
    /// `"fail"` (default) stops on a row error; `"skip"` writes null and
    /// records the error, subject to the failure limits.
    pub error_policy: Option<String>,
    /// Under a skip policy, fail the job once more than this share of rows
    /// (0 to 1) has failed.
    pub max_failed_fraction: Option<f64>,
    /// Under a skip policy, fail the job once more than this many rows have
    /// failed.
    pub max_failed_rows: Option<i64>,
    /// Deadline for one Function call, in seconds.
    pub row_timeout_seconds: Option<f64>,
    /// Deadline for Function initialization, in seconds.
    pub init_timeout_seconds: Option<f64>,
    /// Worker count for an expression column; Function columns reject it.
    pub num_workers: Option<u32>,
    /// Cap on executor processes per Function worker.
    pub max_executor_processes_per_worker: Option<u32>,
}

impl TryFrom<RefreshColumnOptions> for lancedb::table::RefreshColumnOptions {
    type Error = napi::Error;

    fn try_from(value: RefreshColumnOptions) -> napi::Result<Self> {
        let mut options = Self::default();
        options.where_clause = value.where_clause;
        options.batch_size = value.batch_size.map(|size| size as usize);
        options.max_concurrency = positive_u32(value.max_concurrency, "maxConcurrency")?;
        options.gpu_class = value.gpu_class;
        options.resource_budget = value
            .resource_budget
            .map(|budget| {
                Ok::<_, napi::Error>(
                    LanceFunctionWorkerResourceBudget::new(
                        resource_budget_integer(budget.cpu_millis, "resourceBudget.cpuMillis")?,
                        resource_budget_integer(budget.memory_bytes, "resourceBudget.memoryBytes")?,
                    )
                    .with_gpu_count(resource_budget_gpu_count(budget.gpu_count)?),
                )
            })
            .transpose()?;
        options.gpu_count = value.gpu_count;
        options.error_policy = match value.error_policy.as_deref() {
            None => None,
            Some("fail") => Some(lancedb::function::FunctionErrorPolicy::Fail),
            Some("skip") => Some(lancedb::function::FunctionErrorPolicy::Skip),
            Some(other) => {
                return Err(napi::Error::from_reason(format!(
                    "errorPolicy must be \"fail\" or \"skip\", got \"{other}\""
                )));
            }
        };
        options.max_failed_fraction = value.max_failed_fraction;
        options.max_failed_rows = value
            .max_failed_rows
            .map(|rows| {
                u64::try_from(rows).map_err(|_| {
                    napi::Error::from_reason("maxFailedRows must be a non-negative integer")
                })
            })
            .transpose()?;
        options.row_timeout_seconds = value.row_timeout_seconds;
        options.init_timeout_seconds = value.init_timeout_seconds;
        options.num_workers = value.num_workers.map(|workers| workers as usize);
        options.max_executor_processes_per_worker = value.max_executor_processes_per_worker;
        Ok(options)
    }
}

/// One row a Function refresh skipped, as the server recorded it.
#[napi(object)]
#[derive(Clone, Debug)]
pub struct FunctionErrorRecord {
    pub job_id: String,
    pub fragment_id: i64,
    /// The row's offset within the fragment; absent when the fragment's
    /// detail was capped.
    pub row_offset: Option<i64>,
    pub column: String,
    pub function: String,
    pub function_version: String,
    pub table_version: i64,
    pub error_type: String,
    pub error_message: String,
    /// User-code traceback, possibly truncated by the server.
    pub traceback: Option<String>,
    pub created_at_millis: i64,
}

impl From<lancedb::function::FunctionErrorRecord> for FunctionErrorRecord {
    fn from(record: lancedb::function::FunctionErrorRecord) -> Self {
        Self {
            job_id: record.job_id,
            fragment_id: record.fragment_id as i64,
            row_offset: record.row_offset.map(i64::from),
            column: record.column,
            function: record.function,
            function_version: record.function_version,
            table_version: record.table_version as i64,
            error_type: record.error_type,
            error_message: record.error_message,
            traceback: record.traceback,
            created_at_millis: record.created_at_millis,
        }
    }
}

/// A fragment whose per-row error detail was capped: `rowsSkipped` rows
/// failed, of which only `rowsRecorded` have a record of their own.
#[napi(object)]
#[derive(Clone, Debug)]
pub struct FunctionErrorFragment {
    pub job_id: String,
    pub fragment_id: i64,
    pub rows_skipped: i64,
    pub rows_recorded: i64,
}

impl From<lancedb::function::FunctionErrorFragment> for FunctionErrorFragment {
    fn from(fragment: lancedb::function::FunctionErrorFragment) -> Self {
        Self {
            job_id: fragment.job_id,
            fragment_id: fragment.fragment_id as i64,
            rows_skipped: fragment.rows_skipped as i64,
            rows_recorded: fragment.rows_recorded as i64,
        }
    }
}

/// A table's per-row Function errors.
#[napi(object)]
#[derive(Clone, Debug)]
pub struct FunctionErrors {
    /// The recorded rows, newest job first.
    pub records: Vec<FunctionErrorRecord>,
    /// Fragments whose detail was capped.
    pub fragments: Vec<FunctionErrorFragment>,
    /// Whether the listing stopped at its limit.
    pub truncated: bool,
}

impl From<lancedb::function::FunctionErrors> for FunctionErrors {
    fn from(errors: lancedb::function::FunctionErrors) -> Self {
        Self {
            records: errors.records.into_iter().map(Into::into).collect(),
            fragments: errors.fragments.into_iter().map(Into::into).collect(),
            truncated: errors.truncated,
        }
    }
}

#[napi(object)]
pub struct RefreshMaterializedViewResult {
    /// How the view was brought up to date: "rebuild", or absent for LanceDB Enterprise SQL refreshes.
    pub mode: Option<String>,
    /// Rows written by this refresh, or absent for LanceDB Enterprise SQL refreshes.
    pub rows_written: Option<i64>,
    /// Source version refreshed to, or absent for LanceDB Enterprise SQL refreshes.
    pub source_version: Option<i64>,
    pub version: i64,
}

impl From<lancedb::RefreshMaterializedViewResult> for RefreshMaterializedViewResult {
    fn from(value: lancedb::RefreshMaterializedViewResult) -> Self {
        let mode = value.mode.map(|mode| match mode {
            lancedb::RefreshMode::Rebuild => "rebuild",
        });
        Self {
            mode: mode.map(str::to_string),
            rows_written: value.rows_written.map(|rows| rows as i64),
            source_version: value.source_version.map(|version| version as i64),
            version: value.version as i64,
        }
    }
}

impl From<lancedb::table::RefreshColumnResult> for RefreshColumnResult {
    fn from(value: lancedb::table::RefreshColumnResult) -> Self {
        Self {
            rows_filled: value.rows_filled as i64,
            version: value.version as i64,
        }
    }
}

impl From<lancedb::table::AddColumnsResult> for AddColumnsResult {
    fn from(value: lancedb::table::AddColumnsResult) -> Self {
        Self {
            version: value.version as i64,
        }
    }
}

#[napi(object)]
pub struct AlterColumnsResult {
    pub version: i64,
}

impl From<lancedb::table::AlterColumnsResult> for AlterColumnsResult {
    fn from(value: lancedb::table::AlterColumnsResult) -> Self {
        Self {
            version: value.version as i64,
        }
    }
}

#[napi(object)]
pub struct UpdateFieldMetadataResult {
    pub version: i64,
}

impl From<lancedb::table::UpdateFieldMetadataResult> for UpdateFieldMetadataResult {
    fn from(value: lancedb::table::UpdateFieldMetadataResult) -> Self {
        Self {
            version: value.version as i64,
        }
    }
}

#[napi(object)]
pub struct DropColumnsResult {
    pub version: i64,
}

impl From<lancedb::table::DropColumnsResult> for DropColumnsResult {
    fn from(value: lancedb::table::DropColumnsResult) -> Self {
        Self {
            version: value.version as i64,
        }
    }
}

#[napi]
pub struct TagContents {
    pub version: i64,
    pub manifest_size: i64,
}

#[napi]
pub struct BranchContents {
    pub parent_branch: Option<String>,
    pub parent_version: i64,
    pub manifest_size: i64,
}

#[napi]
pub struct Tags {
    inner: LanceDbTable,
}

#[napi]
impl Tags {
    #[napi]
    pub async fn list(&self) -> napi::Result<HashMap<String, TagContents>> {
        let rust_tags = self.inner.tags().await.default_error()?;
        let tag_list = rust_tags.as_ref().list().await.default_error()?;
        let tag_contents = tag_list
            .into_iter()
            .map(|(k, v)| {
                (
                    k,
                    TagContents {
                        version: v.version as i64,
                        manifest_size: v.manifest_size as i64,
                    },
                )
            })
            .collect();

        Ok(tag_contents)
    }

    #[napi]
    pub async fn get_version(&self, tag: String) -> napi::Result<i64> {
        let rust_tags = self.inner.tags().await.default_error()?;
        rust_tags
            .as_ref()
            .get_version(tag.as_str())
            .await
            .map(|v| v as i64)
            .default_error()
    }

    #[napi]
    pub async unsafe fn create(&mut self, tag: String, version: i64) -> napi::Result<()> {
        let mut rust_tags = self.inner.tags().await.default_error()?;
        rust_tags
            .as_mut()
            .create(tag.as_str(), version as u64)
            .await
            .default_error()
    }

    #[napi]
    pub async unsafe fn delete(&mut self, tag: String) -> napi::Result<()> {
        let mut rust_tags = self.inner.tags().await.default_error()?;
        rust_tags
            .as_mut()
            .delete(tag.as_str())
            .await
            .default_error()
    }

    #[napi]
    pub async unsafe fn update(&mut self, tag: String, version: i64) -> napi::Result<()> {
        let mut rust_tags = self.inner.tags().await.default_error()?;
        rust_tags
            .as_mut()
            .update(tag.as_str(), version as u64)
            .await
            .default_error()
    }
}

#[napi]
pub struct Branches {
    inner: LanceDbTable,
}

#[napi]
impl Branches {
    #[napi]
    pub async fn list(&self) -> napi::Result<HashMap<String, BranchContents>> {
        let branches = self.inner.list_branches().await.default_error()?;
        let result = branches
            .into_iter()
            .map(|(k, v)| {
                (
                    k,
                    BranchContents {
                        parent_branch: v.parent_branch,
                        parent_version: v.parent_version as i64,
                        manifest_size: v.manifest_size as i64,
                    },
                )
            })
            .collect();
        Ok(result)
    }

    #[napi]
    pub async fn create(
        &self,
        name: String,
        from_ref: Option<String>,
        from_version: Option<i64>,
    ) -> napi::Result<Table> {
        let from_ref = from_ref.filter(|b| b != "main");
        let from_version = from_version
            .map(|v| {
                u64::try_from(v).map_err(|_| {
                    napi::Error::from_reason("from_version must be a non-negative integer")
                })
            })
            .transpose()?;
        let from = Ref::Version(from_ref, from_version);
        let table = self
            .inner
            .create_branch(&name, from)
            .await
            .default_error()?;
        Ok(Table::new(table))
    }

    #[napi]
    pub async fn checkout(&self, name: String, version: Option<i64>) -> napi::Result<Table> {
        let version = version
            .map(|v| {
                u64::try_from(v)
                    .map_err(|_| napi::Error::from_reason("version must be a non-negative integer"))
            })
            .transpose()?;
        let table = self
            .inner
            .checkout_branch(&name, version)
            .await
            .default_error()?;
        Ok(Table::new(table))
    }

    #[napi]
    pub async fn delete(&self, name: String) -> napi::Result<()> {
        self.inner.delete_branch(&name).await.default_error()
    }

    #[napi(ts_return_type = "Promise<Record<string, unknown>>")]
    pub async fn diff(&self, from_branch: String) -> napi::Result<serde_json::Value> {
        let diff = self.inner.diff_branch(&from_branch).await.default_error()?;
        serde_json::to_value(diff).map_err(|err| {
            napi::Error::from_reason(format!("failed to serialize branch diff: {err}"))
        })
    }

    #[napi(ts_return_type = "Promise<Record<string, unknown>>")]
    pub async fn cherry_pick(
        &self,
        from_branch: String,
        dry_run: Option<bool>,
    ) -> napi::Result<serde_json::Value> {
        let result = self
            .inner
            .cherry_pick(&from_branch, dry_run.unwrap_or(false))
            .await
            .default_error()?;
        serde_json::to_value(result).map_err(|err| {
            napi::Error::from_reason(format!("failed to serialize cherry-pick result: {err}"))
        })
    }
}