lance_namespace_impls/

dir.rs

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

//! Directory-based Lance Namespace implementation.
//!
//! This module provides a directory-based implementation of the Lance namespace
//! that stores tables as Lance datasets in a filesystem directory structure.

pub mod manifest;

use arrow::record_batch::RecordBatchIterator;
use arrow_ipc::reader::StreamReader;
use async_trait::async_trait;
use bytes::Bytes;
use futures::TryStreamExt;
use lance::dataset::builder::DatasetBuilder;
use lance::dataset::{Dataset, WriteMode, WriteParams};
use lance::session::Session;
use lance_io::object_store::{ObjectStore, ObjectStoreParams, ObjectStoreRegistry};
use lance_table::io::commit::ManifestNamingScheme;
use object_store::path::Path;
use object_store::{Error as ObjectStoreError, ObjectStore as OSObjectStore, PutMode, PutOptions};
use std::collections::HashMap;
use std::io::Cursor;
use std::sync::Arc;

use crate::context::DynamicContextProvider;
use lance_namespace::models::{
    BatchDeleteTableVersionsRequest, BatchDeleteTableVersionsResponse, CreateNamespaceRequest,
    CreateNamespaceResponse, CreateTableRequest, CreateTableResponse, CreateTableVersionRequest,
    CreateTableVersionResponse, DeclareTableRequest, DeclareTableResponse,
    DescribeNamespaceRequest, DescribeNamespaceResponse, DescribeTableRequest,
    DescribeTableResponse, DescribeTableVersionRequest, DescribeTableVersionResponse,
    DropNamespaceRequest, DropNamespaceResponse, DropTableRequest, DropTableResponse, Identity,
    ListNamespacesRequest, ListNamespacesResponse, ListTableVersionsRequest,
    ListTableVersionsResponse, ListTablesRequest, ListTablesResponse, NamespaceExistsRequest,
    TableExistsRequest, TableVersion,
};

use lance_core::{Error, Result, box_error};
use lance_namespace::LanceNamespace;
use lance_namespace::schema::arrow_schema_to_json;

use crate::credentials::{
    CredentialVendor, create_credential_vendor_for_location, has_credential_vendor_config,
};

/// Result of checking table status atomically.
///
/// This struct captures the state of a table directory in a single snapshot,
/// avoiding race conditions between checking existence and other status flags.
pub(crate) struct TableStatus {
    /// Whether the table directory exists (has any files)
    pub(crate) exists: bool,
    /// Whether the table has a `.lance-deregistered` marker file
    pub(crate) is_deregistered: bool,
    /// Whether the table has a `.lance-reserved` marker file (declared but not written)
    pub(crate) has_reserved_file: bool,
}

/// Builder for creating a DirectoryNamespace.
///
/// This builder provides a fluent API for configuring and establishing
/// connections to directory-based Lance namespaces.
///
/// # Examples
///
/// ```no_run
/// # use lance_namespace_impls::DirectoryNamespaceBuilder;
/// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
/// // Create a local directory namespace
/// let namespace = DirectoryNamespaceBuilder::new("/path/to/data")
///     .build()
///     .await?;
/// # Ok(())
/// # }
/// ```
///
/// ```no_run
/// # use lance_namespace_impls::DirectoryNamespaceBuilder;
/// # use lance::session::Session;
/// # use std::sync::Arc;
/// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
/// // Create with custom storage options and session
/// let session = Arc::new(Session::default());
/// let namespace = DirectoryNamespaceBuilder::new("s3://bucket/path")
///     .storage_option("region", "us-west-2")
///     .storage_option("access_key_id", "key")
///     .session(session)
///     .build()
///     .await?;
/// # Ok(())
/// # }
/// ```
#[derive(Clone)]
pub struct DirectoryNamespaceBuilder {
    root: String,
    storage_options: Option<HashMap<String, String>>,
    session: Option<Arc<Session>>,
    manifest_enabled: bool,
    dir_listing_enabled: bool,
    inline_optimization_enabled: bool,
    table_version_tracking_enabled: bool,
    /// When true, table versions are stored in the `__manifest` table instead of
    /// relying on Lance's native version management.
    table_version_storage_enabled: bool,
    credential_vendor_properties: HashMap<String, String>,
    context_provider: Option<Arc<dyn DynamicContextProvider>>,
    commit_retries: Option<u32>,
}

impl std::fmt::Debug for DirectoryNamespaceBuilder {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("DirectoryNamespaceBuilder")
            .field("root", &self.root)
            .field("storage_options", &self.storage_options)
            .field("manifest_enabled", &self.manifest_enabled)
            .field("dir_listing_enabled", &self.dir_listing_enabled)
            .field(
                "inline_optimization_enabled",
                &self.inline_optimization_enabled,
            )
            .field(
                "table_version_tracking_enabled",
                &self.table_version_tracking_enabled,
            )
            .field(
                "table_version_storage_enabled",
                &self.table_version_storage_enabled,
            )
            .field(
                "context_provider",
                &self.context_provider.as_ref().map(|_| "Some(...)"),
            )
            .finish()
    }
}

impl DirectoryNamespaceBuilder {
    /// Create a new DirectoryNamespaceBuilder with the specified root path.
    ///
    /// # Arguments
    ///
    /// * `root` - Root directory path (local path or cloud URI like s3://bucket/path)
    pub fn new(root: impl Into<String>) -> Self {
        Self {
            root: root.into().trim_end_matches('/').to_string(),
            storage_options: None,
            session: None,
            manifest_enabled: true,
            dir_listing_enabled: true, // Default to enabled for backwards compatibility
            inline_optimization_enabled: true,
            table_version_tracking_enabled: false, // Default to disabled
            table_version_storage_enabled: false,  // Default to disabled
            credential_vendor_properties: HashMap::new(),
            context_provider: None,
            commit_retries: None,
        }
    }

    /// Enable or disable manifest-based listing.
    ///
    /// When enabled (default), the namespace uses a `__manifest` table to track tables.
    /// When disabled, relies solely on directory scanning.
    pub fn manifest_enabled(mut self, enabled: bool) -> Self {
        self.manifest_enabled = enabled;
        self
    }

    /// Enable or disable directory-based listing fallback.
    ///
    /// When enabled (default), falls back to directory scanning for tables not in the manifest.
    /// When disabled, only consults the manifest table.
    pub fn dir_listing_enabled(mut self, enabled: bool) -> Self {
        self.dir_listing_enabled = enabled;
        self
    }

    /// Enable or disable inline optimization of the __manifest table.
    ///
    /// When enabled (default), performs compaction and indexing on the __manifest table
    /// after every write operation to maintain optimal performance.
    /// When disabled, manual optimization must be performed separately.
    pub fn inline_optimization_enabled(mut self, enabled: bool) -> Self {
        self.inline_optimization_enabled = enabled;
        self
    }

    /// Enable or disable table version tracking through the namespace.
    ///
    /// When enabled, `describe_table` returns `managed_versioning: true` to indicate
    /// that commits should go through the namespace's table version APIs rather than
    /// direct object store operations.
    ///
    /// When disabled (default), `managed_versioning` is not set.
    pub fn table_version_tracking_enabled(mut self, enabled: bool) -> Self {
        self.table_version_tracking_enabled = enabled;
        self
    }

    /// Enable or disable table version management through the `__manifest` table.
    ///
    /// When enabled, table versions are tracked as `table_version` entries in the
    /// `__manifest` Lance table. This enables:
    /// - Centralized version tracking instead of per-table `_versions/` directories
    ///
    /// Requires `manifest_enabled` to be true.
    /// When disabled (default), version storage uses per-table storage operations.
    pub fn table_version_storage_enabled(mut self, enabled: bool) -> Self {
        self.table_version_storage_enabled = enabled;
        self
    }

    /// Create a DirectoryNamespaceBuilder from properties HashMap.
    ///
    /// This method parses a properties map into builder configuration.
    /// It expects:
    /// - `root`: The root directory path (required)
    /// - `manifest_enabled`: Enable manifest-based table tracking (optional, default: true)
    /// - `dir_listing_enabled`: Enable directory listing for table discovery (optional, default: true)
    /// - `inline_optimization_enabled`: Enable inline optimization of __manifest table (optional, default: true)
    /// - `storage.*`: Storage options (optional, prefix will be stripped)
    ///
    /// Credential vendor properties (prefixed with `credential_vendor.`, prefix is stripped):
    /// - `credential_vendor.enabled`: Set to "true" to enable credential vending (required)
    /// - `credential_vendor.permission`: Permission level: read, write, or admin (default: read)
    ///
    /// AWS-specific properties (for s3:// locations):
    /// - `credential_vendor.aws_role_arn`: AWS IAM role ARN (required for AWS)
    /// - `credential_vendor.aws_external_id`: AWS external ID (optional)
    /// - `credential_vendor.aws_region`: AWS region (optional)
    /// - `credential_vendor.aws_role_session_name`: AWS role session name (optional)
    /// - `credential_vendor.aws_duration_millis`: Credential duration in ms (default: 3600000, range: 15min-12hrs)
    ///
    /// GCP-specific properties (for gs:// locations):
    /// - `credential_vendor.gcp_service_account`: Service account to impersonate (optional)
    ///
    /// Note: GCP uses Application Default Credentials (ADC). To use a service account key file,
    /// set the `GOOGLE_APPLICATION_CREDENTIALS` environment variable before starting.
    /// GCP token duration cannot be configured; it's determined by the STS endpoint (typically 1 hour).
    ///
    /// Azure-specific properties (for az:// locations):
    /// - `credential_vendor.azure_account_name`: Azure storage account name (required for Azure)
    /// - `credential_vendor.azure_tenant_id`: Azure tenant ID (optional)
    /// - `credential_vendor.azure_duration_millis`: Credential duration in ms (default: 3600000, up to 7 days)
    ///
    /// # Arguments
    ///
    /// * `properties` - Configuration properties
    /// * `session` - Optional Lance session to reuse object store registry
    ///
    /// # Returns
    ///
    /// Returns a `DirectoryNamespaceBuilder` instance.
    ///
    /// # Errors
    ///
    /// Returns an error if the `root` property is missing.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// # use lance_namespace_impls::DirectoryNamespaceBuilder;
    /// # use std::collections::HashMap;
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// let mut properties = HashMap::new();
    /// properties.insert("root".to_string(), "/path/to/data".to_string());
    /// properties.insert("manifest_enabled".to_string(), "true".to_string());
    /// properties.insert("dir_listing_enabled".to_string(), "false".to_string());
    /// properties.insert("storage.region".to_string(), "us-west-2".to_string());
    ///
    /// let namespace = DirectoryNamespaceBuilder::from_properties(properties, None)?
    ///     .build()
    ///     .await?;
    /// # Ok(())
    /// # }
    /// ```
    pub fn from_properties(
        properties: HashMap<String, String>,
        session: Option<Arc<Session>>,
    ) -> Result<Self> {
        // Extract root from properties (required)
        let root = properties.get("root").cloned().ok_or_else(|| {
            Error::namespace_source(
                "Missing required property 'root' for directory namespace".into(),
            )
        })?;

        // Extract storage options (properties prefixed with "storage.")
        let storage_options: HashMap<String, String> = properties
            .iter()
            .filter_map(|(k, v)| {
                k.strip_prefix("storage.")
                    .map(|key| (key.to_string(), v.clone()))
            })
            .collect();

        let storage_options = if storage_options.is_empty() {
            None
        } else {
            Some(storage_options)
        };

        // Extract manifest_enabled (default: true)
        let manifest_enabled = properties
            .get("manifest_enabled")
            .and_then(|v| v.parse::<bool>().ok())
            .unwrap_or(true);

        // Extract dir_listing_enabled (default: true)
        let dir_listing_enabled = properties
            .get("dir_listing_enabled")
            .and_then(|v| v.parse::<bool>().ok())
            .unwrap_or(true);

        // Extract inline_optimization_enabled (default: true)
        let inline_optimization_enabled = properties
            .get("inline_optimization_enabled")
            .and_then(|v| v.parse::<bool>().ok())
            .unwrap_or(true);

        // Extract table_version_tracking_enabled (default: false)
        let table_version_tracking_enabled = properties
            .get("table_version_tracking_enabled")
            .and_then(|v| v.parse::<bool>().ok())
            .unwrap_or(false);

        // Extract table_version_storage_enabled (default: false)
        let table_version_storage_enabled = properties
            .get("table_version_storage_enabled")
            .and_then(|v| v.parse::<bool>().ok())
            .unwrap_or(false);

        // Extract credential vendor properties (properties prefixed with "credential_vendor.")
        // The prefix is stripped to get short property names
        // The build() method will check if enabled=true before creating the vendor
        let credential_vendor_properties: HashMap<String, String> = properties
            .iter()
            .filter_map(|(k, v)| {
                k.strip_prefix("credential_vendor.")
                    .map(|key| (key.to_string(), v.clone()))
            })
            .collect();

        let commit_retries = properties
            .get("commit_retries")
            .and_then(|v| v.parse::<u32>().ok());

        Ok(Self {
            root: root.trim_end_matches('/').to_string(),
            storage_options,
            session,
            manifest_enabled,
            dir_listing_enabled,
            inline_optimization_enabled,
            table_version_tracking_enabled,
            table_version_storage_enabled,
            credential_vendor_properties,
            context_provider: None,
            commit_retries,
        })
    }

    /// Add a storage option.
    ///
    /// # Arguments
    ///
    /// * `key` - Storage option key (e.g., "region", "access_key_id")
    /// * `value` - Storage option value
    pub fn storage_option(mut self, key: impl Into<String>, value: impl Into<String>) -> Self {
        self.storage_options
            .get_or_insert_with(HashMap::new)
            .insert(key.into(), value.into());
        self
    }

    /// Add multiple storage options.
    ///
    /// # Arguments
    ///
    /// * `options` - HashMap of storage options to add
    pub fn storage_options(mut self, options: HashMap<String, String>) -> Self {
        self.storage_options
            .get_or_insert_with(HashMap::new)
            .extend(options);
        self
    }

    /// Set the Lance session to use for this namespace.
    ///
    /// When a session is provided, the namespace will reuse the session's
    /// object store registry, allowing multiple namespaces and datasets
    /// to share the same underlying storage connections.
    ///
    /// # Arguments
    ///
    /// * `session` - Arc-wrapped Lance session
    pub fn session(mut self, session: Arc<Session>) -> Self {
        self.session = Some(session);
        self
    }

    /// Set the number of retries for commit operations on the manifest table.
    /// If not set, defaults to [`lance_table::io::commit::CommitConfig`] default (20).
    pub fn commit_retries(mut self, retries: u32) -> Self {
        self.commit_retries = Some(retries);
        self
    }

    /// Add a credential vendor property.
    ///
    /// Use short property names without the `credential_vendor.` prefix.
    /// Common properties: `enabled`, `permission`.
    /// AWS properties: `aws_role_arn`, `aws_external_id`, `aws_region`, `aws_role_session_name`, `aws_duration_millis`.
    /// GCP properties: `gcp_service_account`.
    /// Azure properties: `azure_account_name`, `azure_tenant_id`, `azure_duration_millis`.
    ///
    /// # Arguments
    ///
    /// * `key` - Property key (e.g., "enabled", "aws_role_arn")
    /// * `value` - Property value
    ///
    /// # Example
    ///
    /// ```no_run
    /// # use lance_namespace_impls::DirectoryNamespaceBuilder;
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// let namespace = DirectoryNamespaceBuilder::new("s3://my-bucket/data")
    ///     .credential_vendor_property("enabled", "true")
    ///     .credential_vendor_property("aws_role_arn", "arn:aws:iam::123456789012:role/MyRole")
    ///     .credential_vendor_property("permission", "read")
    ///     .build()
    ///     .await?;
    /// # Ok(())
    /// # }
    /// ```
    pub fn credential_vendor_property(
        mut self,
        key: impl Into<String>,
        value: impl Into<String>,
    ) -> Self {
        self.credential_vendor_properties
            .insert(key.into(), value.into());
        self
    }

    /// Add multiple credential vendor properties.
    ///
    /// Use short property names without the `credential_vendor.` prefix.
    ///
    /// # Arguments
    ///
    /// * `properties` - HashMap of credential vendor properties to add
    pub fn credential_vendor_properties(mut self, properties: HashMap<String, String>) -> Self {
        self.credential_vendor_properties.extend(properties);
        self
    }

    /// Set a dynamic context provider for per-request context.
    ///
    /// The provider can be used to generate additional context for operations.
    /// For DirectoryNamespace, the context is stored but not directly used
    /// in operations (unlike RestNamespace where it's converted to HTTP headers).
    ///
    /// # Arguments
    ///
    /// * `provider` - The context provider implementation
    pub fn context_provider(mut self, provider: Arc<dyn DynamicContextProvider>) -> Self {
        self.context_provider = Some(provider);
        self
    }

    /// Build the DirectoryNamespace.
    ///
    /// # Returns
    ///
    /// Returns a `DirectoryNamespace` instance.
    ///
    /// # Errors
    ///
    /// Returns an error if:
    /// - The root path is invalid
    /// - Connection to the storage backend fails
    /// - Storage options are invalid
    pub async fn build(self) -> Result<DirectoryNamespace> {
        // Validate: table_version_storage_enabled requires manifest_enabled
        if self.table_version_storage_enabled && !self.manifest_enabled {
            return Err(Error::invalid_input(
                "table_version_storage_enabled requires manifest_enabled=true",
            ));
        }

        let (object_store, base_path) =
            Self::initialize_object_store(&self.root, &self.storage_options, &self.session).await?;

        let manifest_ns = if self.manifest_enabled {
            match manifest::ManifestNamespace::from_directory(
                self.root.clone(),
                self.storage_options.clone(),
                self.session.clone(),
                object_store.clone(),
                base_path.clone(),
                self.dir_listing_enabled,
                self.inline_optimization_enabled,
                self.commit_retries,
                self.table_version_storage_enabled,
            )
            .await
            {
                Ok(ns) => Some(Arc::new(ns)),
                Err(e) => {
                    // Failed to initialize manifest namespace, fall back to directory listing only
                    log::warn!(
                        "Failed to initialize manifest namespace, falling back to directory listing only: {}",
                        e
                    );
                    None
                }
            }
        } else {
            None
        };

        // Create credential vendor once during initialization if enabled
        let credential_vendor = if has_credential_vendor_config(&self.credential_vendor_properties)
        {
            create_credential_vendor_for_location(&self.root, &self.credential_vendor_properties)
                .await?
                .map(Arc::from)
        } else {
            None
        };

        Ok(DirectoryNamespace {
            root: self.root,
            storage_options: self.storage_options,
            session: self.session,
            object_store,
            base_path,
            manifest_ns,
            dir_listing_enabled: self.dir_listing_enabled,
            table_version_tracking_enabled: self.table_version_tracking_enabled,
            table_version_storage_enabled: self.table_version_storage_enabled,
            credential_vendor,
            context_provider: self.context_provider,
        })
    }

    /// Initialize the Lance ObjectStore based on the configuration
    async fn initialize_object_store(
        root: &str,
        storage_options: &Option<HashMap<String, String>>,
        session: &Option<Arc<Session>>,
    ) -> Result<(Arc<ObjectStore>, Path)> {
        // Build ObjectStoreParams from storage options
        let accessor = storage_options.clone().map(|opts| {
            Arc::new(lance_io::object_store::StorageOptionsAccessor::with_static_options(opts))
        });
        let params = ObjectStoreParams {
            storage_options_accessor: accessor,
            ..Default::default()
        };

        // Use object store registry from session if provided, otherwise create a new one
        let registry = if let Some(session) = session {
            session.store_registry()
        } else {
            Arc::new(ObjectStoreRegistry::default())
        };

        // Use Lance's object store factory to create from URI
        let (object_store, base_path) = ObjectStore::from_uri_and_params(registry, root, &params)
            .await
            .map_err(|e| {
                Error::namespace_source(format!("Failed to create object store: {}", e).into())
            })?;

        Ok((object_store, base_path))
    }
}

/// Directory-based implementation of Lance Namespace.
///
/// This implementation stores tables as Lance datasets in a directory structure.
/// It supports local filesystems and cloud storage backends through Lance's object store.
///
/// ## Manifest-based Listing
///
/// When `manifest_enabled=true`, the namespace uses a special `__manifest` Lance table to track tables
/// instead of scanning the filesystem. This provides:
/// - Better performance for listing operations
/// - Ability to track table metadata
/// - Foundation for future features like namespaces and table renaming
///
/// When `dir_listing_enabled=true`, the namespace falls back to directory scanning for tables not
/// found in the manifest, enabling gradual migration.
///
/// ## Credential Vending
///
/// When credential vendor properties are configured, `describe_table` will vend temporary
/// credentials based on the table location URI. The vendor type is auto-selected:
/// - `s3://` locations use AWS STS AssumeRole
/// - `gs://` locations use GCP OAuth2 tokens
/// - `az://` locations use Azure SAS tokens
pub struct DirectoryNamespace {
    root: String,
    storage_options: Option<HashMap<String, String>>,
    #[allow(dead_code)]
    session: Option<Arc<Session>>,
    object_store: Arc<ObjectStore>,
    base_path: Path,
    manifest_ns: Option<Arc<manifest::ManifestNamespace>>,
    dir_listing_enabled: bool,
    /// When true, `describe_table` returns `managed_versioning: true` to indicate
    /// commits should go through namespace table version APIs.
    table_version_tracking_enabled: bool,
    /// When true, table versions are stored in the `__manifest` table.
    table_version_storage_enabled: bool,
    /// Credential vendor created once during initialization.
    /// Used to vend temporary credentials for table access.
    credential_vendor: Option<Arc<dyn CredentialVendor>>,
    /// Dynamic context provider for per-request context.
    /// Stored but not directly used in operations (available for future extensions).
    #[allow(dead_code)]
    context_provider: Option<Arc<dyn DynamicContextProvider>>,
}

impl std::fmt::Debug for DirectoryNamespace {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "{}", self.namespace_id())
    }
}

impl std::fmt::Display for DirectoryNamespace {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "{}", self.namespace_id())
    }
}

/// Describes the version ranges to delete for a single table.
/// Used by `batch_delete_table_versions` and `delete_physical_version_files`.
struct TableDeleteEntry {
    table_id: Option<Vec<String>>,
    ranges: Vec<(i64, i64)>,
}

impl DirectoryNamespace {
    /// Apply pagination to a list of table names
    ///
    /// Sorts the list alphabetically and applies pagination using page_token (start_after) and limit.
    ///
    /// # Arguments
    /// * `names` - The vector of table names to paginate
    /// * `page_token` - Skip items until finding one greater than this value (start_after semantics)
    /// * `limit` - Maximum number of items to keep
    fn apply_pagination(names: &mut Vec<String>, page_token: Option<String>, limit: Option<i32>) {
        // Sort alphabetically for consistent ordering
        names.sort();

        // Apply page_token filtering (start_after semantics)
        if let Some(start_after) = page_token {
            if let Some(index) = names
                .iter()
                .position(|name| name.as_str() > start_after.as_str())
            {
                names.drain(0..index);
            } else {
                names.clear();
            }
        }

        // Apply limit
        if let Some(limit) = limit
            && limit >= 0
        {
            names.truncate(limit as usize);
        }
    }

    /// List tables using directory scanning (fallback method)
    async fn list_directory_tables(&self) -> Result<Vec<String>> {
        let mut tables = Vec::new();
        let entries = self
            .object_store
            .read_dir(self.base_path.clone())
            .await
            .map_err(|e| {
                Error::io_source(box_error(std::io::Error::other(format!(
                    "Failed to list directory: {}",
                    e
                ))))
            })?;

        for entry in entries {
            let path = entry.trim_end_matches('/');
            if !path.ends_with(".lance") {
                continue;
            }

            let table_name = &path[..path.len() - 6];

            // Use atomic check to skip deregistered tables and declared-but-not-written tables
            let status = self.check_table_status(table_name).await;
            if status.is_deregistered || status.has_reserved_file {
                continue;
            }

            tables.push(table_name.to_string());
        }

        Ok(tables)
    }

    /// Validate that the namespace ID represents the root namespace
    fn validate_root_namespace_id(id: &Option<Vec<String>>) -> Result<()> {
        if let Some(id) = id
            && !id.is_empty()
        {
            return Err(Error::namespace_source(format!(
                "Directory namespace only supports root namespace operations, but got namespace ID: {:?}. Expected empty ID.",
                id
            ).into()));
        }
        Ok(())
    }

    /// Extract table name from table ID
    fn table_name_from_id(id: &Option<Vec<String>>) -> Result<String> {
        let id = id.as_ref().ok_or_else(|| {
            Error::namespace_source("Directory namespace table ID cannot be empty".into())
        })?;

        if id.len() != 1 {
            return Err(Error::namespace_source(format!(
                "Multi-level table IDs are only supported when manifest mode is enabled, but got: {:?}",
                id
            )
            .into()));
        }

        Ok(id[0].clone())
    }

    async fn resolve_table_location(&self, id: &Option<Vec<String>>) -> Result<String> {
        let mut describe_req = DescribeTableRequest::new();
        describe_req.id = id.clone();
        describe_req.load_detailed_metadata = Some(false);

        let describe_resp = self.describe_table(describe_req).await?;

        describe_resp.location.ok_or_else(|| {
            Error::namespace_source(format!("Table location not found for: {:?}", id).into())
        })
    }

    fn table_full_uri(&self, table_name: &str) -> String {
        format!("{}/{}.lance", &self.root, table_name)
    }

    fn uri_to_object_store_path(uri: &str) -> Path {
        let path_str = if let Some(rest) = uri.strip_prefix("file://") {
            rest
        } else if let Some(rest) = uri.strip_prefix("s3://") {
            rest.split_once('/').map(|(_, p)| p).unwrap_or(rest)
        } else if let Some(rest) = uri.strip_prefix("gs://") {
            rest.split_once('/').map(|(_, p)| p).unwrap_or(rest)
        } else if let Some(rest) = uri.strip_prefix("az://") {
            rest.split_once('/').map(|(_, p)| p).unwrap_or(rest)
        } else {
            uri
        };
        Path::from(path_str)
    }

    /// Get the object store path for a table (relative to base_path)
    fn table_path(&self, table_name: &str) -> Path {
        self.base_path
            .child(format!("{}.lance", table_name).as_str())
    }

    /// Get the reserved file path for a table
    fn table_reserved_file_path(&self, table_name: &str) -> Path {
        self.base_path
            .child(format!("{}.lance", table_name).as_str())
            .child(".lance-reserved")
    }

    /// Get the deregistered marker file path for a table
    fn table_deregistered_file_path(&self, table_name: &str) -> Path {
        self.base_path
            .child(format!("{}.lance", table_name).as_str())
            .child(".lance-deregistered")
    }

    /// Atomically check table existence and deregistration status.
    ///
    /// This performs a single directory listing to get a consistent snapshot of the
    /// table's state, avoiding race conditions between checking existence and
    /// checking deregistration status.
    pub(crate) async fn check_table_status(&self, table_name: &str) -> TableStatus {
        let table_path = self.table_path(table_name);
        match self.object_store.read_dir(table_path).await {
            Ok(entries) => {
                let exists = !entries.is_empty();
                let is_deregistered = entries.iter().any(|e| e.ends_with(".lance-deregistered"));
                let has_reserved_file = entries.iter().any(|e| e.ends_with(".lance-reserved"));
                TableStatus {
                    exists,
                    is_deregistered,
                    has_reserved_file,
                }
            }
            Err(_) => TableStatus {
                exists: false,
                is_deregistered: false,
                has_reserved_file: false,
            },
        }
    }

    async fn put_marker_file_atomic(
        &self,
        path: &Path,
        file_description: &str,
    ) -> std::result::Result<(), String> {
        let put_opts = PutOptions {
            mode: PutMode::Create,
            ..Default::default()
        };

        match self
            .object_store
            .inner
            .put_opts(path, bytes::Bytes::new().into(), put_opts)
            .await
        {
            Ok(_) => Ok(()),
            Err(ObjectStoreError::AlreadyExists { .. })
            | Err(ObjectStoreError::Precondition { .. }) => {
                Err(format!("{} already exists", file_description))
            }
            Err(e) => Err(format!("Failed to create {}: {}", file_description, e)),
        }
    }

    /// Get storage options for a table, using credential vending if configured.
    ///
    /// If credential vendor properties are configured and the table location matches
    /// a supported cloud provider, this will create an appropriate vendor and vend
    /// temporary credentials scoped to the table location. Otherwise, returns the
    /// static storage options.
    ///
    /// The vendor type is auto-selected based on the table URI:
    /// - `s3://` locations use AWS STS AssumeRole
    /// - `gs://` locations use GCP OAuth2 tokens
    /// - `az://` locations use Azure SAS tokens
    ///
    /// The permission level (Read, Write, Admin) is configured at namespace
    /// initialization time via the `credential_vendor_permission` property.
    ///
    /// # Arguments
    ///
    /// * `table_uri` - The full URI of the table
    /// * `identity` - Optional identity from the request for identity-based credential vending
    async fn get_storage_options_for_table(
        &self,
        table_uri: &str,
        vend_credentials: bool,
        identity: Option<&Identity>,
    ) -> Result<Option<HashMap<String, String>>> {
        if vend_credentials && let Some(ref vendor) = self.credential_vendor {
            let vended = vendor.vend_credentials(table_uri, identity).await?;
            return Ok(Some(vended.storage_options));
        }
        // When no credential vendor is configured, return None to avoid
        // leaking the namespace's own static credentials to clients.
        Ok(None)
    }

    /// Migrate directory-based tables to the manifest.
    ///
    /// This is a one-time migration operation that:
    /// 1. Scans the directory for existing `.lance` tables
    /// 2. Registers any unmigrated tables in the manifest
    /// 3. Returns the count of tables that were migrated
    ///
    /// This method is safe to run multiple times - it will skip tables that are already
    /// registered in the manifest.
    ///
    /// # Usage
    ///
    /// After creating tables in directory-only mode or dual mode, you can migrate them
    /// to the manifest to enable manifest-only mode:
    ///
    /// ```no_run
    /// # use lance_namespace_impls::DirectoryNamespaceBuilder;
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// // Create namespace with dual mode (manifest + directory listing)
    /// let namespace = DirectoryNamespaceBuilder::new("/path/to/data")
    ///     .manifest_enabled(true)
    ///     .dir_listing_enabled(true)
    ///     .build()
    ///     .await?;
    ///
    /// // ... tables are created and used ...
    ///
    /// // Migrate existing directory tables to manifest
    /// let migrated_count = namespace.migrate().await?;
    /// println!("Migrated {} tables", migrated_count);
    ///
    /// // Now you can disable directory listing for better performance:
    /// // (requires rebuilding the namespace)
    /// let namespace = DirectoryNamespaceBuilder::new("/path/to/data")
    ///     .manifest_enabled(true)
    ///     .dir_listing_enabled(false)  // All tables now in manifest
    ///     .build()
    ///     .await?;
    /// # Ok(())
    /// # }
    /// ```
    ///
    /// # Returns
    ///
    /// Returns the number of tables that were migrated to the manifest.
    ///
    /// # Errors
    ///
    /// Returns an error if:
    /// - Manifest is not enabled
    /// - Directory listing fails
    /// - Manifest registration fails
    pub async fn migrate(&self) -> Result<usize> {
        // We only care about tables in the root namespace
        let Some(ref manifest_ns) = self.manifest_ns else {
            return Ok(0); // No manifest, nothing to migrate
        };

        // Get all table locations already in the manifest
        let manifest_locations = manifest_ns.list_manifest_table_locations().await?;

        // Get all tables from directory
        let dir_tables = self.list_directory_tables().await?;

        // Register each directory table that doesn't have an overlapping location
        // If a directory name already exists in the manifest,
        // that means the table must have already been migrated or created
        // in the manifest, so we can skip it.
        let mut migrated_count = 0;
        for table_name in dir_tables {
            // For root namespace tables, the directory name is "table_name.lance"
            let dir_name = format!("{}.lance", table_name);
            if !manifest_locations.contains(&dir_name) {
                manifest_ns.register_table(&table_name, dir_name).await?;
                migrated_count += 1;
            }
        }

        Ok(migrated_count)
    }

    /// Delete physical manifest files for the given table version ranges (best-effort).
    ///
    /// This helper is used by `batch_delete_table_versions` in both the manifest-enabled
    /// and non-manifest paths. It resolves each table's storage location, computes the
    /// version file paths, and attempts to delete them. Errors are logged (best-effort)
    /// when `best_effort` is true, or returned immediately when false.
    ///
    /// Returns the number of files successfully deleted.
    async fn delete_physical_version_files(
        &self,
        table_entries: &[TableDeleteEntry],
        best_effort: bool,
    ) -> Result<i64> {
        let mut deleted_count = 0i64;
        for te in table_entries {
            let table_uri = self.resolve_table_location(&te.table_id).await?;
            let table_path = Self::uri_to_object_store_path(&table_uri);
            let table_path_str = table_path.as_ref();
            let versions_dir_path = Path::from(format!("{}_versions", table_path_str));

            for (start, end) in &te.ranges {
                for version in *start..=*end {
                    let version_path =
                        versions_dir_path.child(format!("{}.manifest", version as u64));
                    match self.object_store.inner.delete(&version_path).await {
                        Ok(_) => {
                            deleted_count += 1;
                        }
                        Err(object_store::Error::NotFound { .. }) => {}
                        Err(e) => {
                            if best_effort {
                                log::warn!(
                                    "Failed to delete manifest file for version {} of table {:?}: {:?}",
                                    version,
                                    te.table_id,
                                    e
                                );
                            } else {
                                return Err(Error::namespace_source(
                                    format!(
                                        "Failed to delete version {} for table at '{}': {}",
                                        version, table_uri, e
                                    )
                                    .into(),
                                ));
                            }
                        }
                    }
                }
            }
        }
        Ok(deleted_count)
    }
}

#[async_trait]
impl LanceNamespace for DirectoryNamespace {
    async fn list_namespaces(
        &self,
        request: ListNamespacesRequest,
    ) -> Result<ListNamespacesResponse> {
        if let Some(ref manifest_ns) = self.manifest_ns {
            return manifest_ns.list_namespaces(request).await;
        }

        Self::validate_root_namespace_id(&request.id)?;
        Ok(ListNamespacesResponse::new(vec![]))
    }

    async fn describe_namespace(
        &self,
        request: DescribeNamespaceRequest,
    ) -> Result<DescribeNamespaceResponse> {
        if let Some(ref manifest_ns) = self.manifest_ns {
            return manifest_ns.describe_namespace(request).await;
        }

        Self::validate_root_namespace_id(&request.id)?;
        #[allow(clippy::needless_update)]
        Ok(DescribeNamespaceResponse {
            properties: Some(HashMap::new()),
            ..Default::default()
        })
    }

    async fn create_namespace(
        &self,
        request: CreateNamespaceRequest,
    ) -> Result<CreateNamespaceResponse> {
        if let Some(ref manifest_ns) = self.manifest_ns {
            return manifest_ns.create_namespace(request).await;
        }

        if request.id.is_none() || request.id.as_ref().unwrap().is_empty() {
            return Err(Error::namespace_source(
                "Root namespace already exists and cannot be created".into(),
            ));
        }

        Err(Error::not_supported_source(
            "Child namespaces are only supported when manifest mode is enabled".into(),
        ))
    }

    async fn drop_namespace(&self, request: DropNamespaceRequest) -> Result<DropNamespaceResponse> {
        if let Some(ref manifest_ns) = self.manifest_ns {
            return manifest_ns.drop_namespace(request).await;
        }

        if request.id.is_none() || request.id.as_ref().unwrap().is_empty() {
            return Err(Error::namespace_source(
                "Root namespace cannot be dropped".into(),
            ));
        }

        Err(Error::not_supported_source(
            "Child namespaces are only supported when manifest mode is enabled".into(),
        ))
    }

    async fn namespace_exists(&self, request: NamespaceExistsRequest) -> Result<()> {
        if let Some(ref manifest_ns) = self.manifest_ns {
            return manifest_ns.namespace_exists(request).await;
        }

        if request.id.is_none() || request.id.as_ref().unwrap().is_empty() {
            return Ok(());
        }

        Err(Error::namespace_source(
            "Child namespaces are only supported when manifest mode is enabled".into(),
        ))
    }

    async fn list_tables(&self, request: ListTablesRequest) -> Result<ListTablesResponse> {
        // Validate that namespace ID is provided
        let namespace_id = request
            .id
            .as_ref()
            .ok_or_else(|| Error::invalid_input_source("Namespace ID is required".into()))?;

        // For child namespaces, always delegate to manifest (if enabled)
        if !namespace_id.is_empty() {
            if let Some(ref manifest_ns) = self.manifest_ns {
                return manifest_ns.list_tables(request).await;
            }
            return Err(Error::not_supported_source(
                "Child namespaces are only supported when manifest mode is enabled".into(),
            ));
        }

        // When only manifest is enabled (no directory listing), delegate directly to manifest
        if let Some(ref manifest_ns) = self.manifest_ns
            && !self.dir_listing_enabled
        {
            return manifest_ns.list_tables(request).await;
        }

        // When both manifest and directory listing are enabled, we need to merge and deduplicate
        let mut tables = if self.manifest_ns.is_some() && self.dir_listing_enabled {
            // Get all manifest table locations (for deduplication)
            let manifest_locations = if let Some(ref manifest_ns) = self.manifest_ns {
                manifest_ns.list_manifest_table_locations().await?
            } else {
                std::collections::HashSet::new()
            };

            // Get all manifest tables (without pagination for merging)
            let mut manifest_request = request.clone();
            manifest_request.limit = None;
            manifest_request.page_token = None;
            let manifest_tables = if let Some(ref manifest_ns) = self.manifest_ns {
                let manifest_response = manifest_ns.list_tables(manifest_request).await?;
                manifest_response.tables
            } else {
                vec![]
            };

            // Start with all manifest table names
            // Add directory tables that aren't already in the manifest (by location)
            let mut all_tables: Vec<String> = manifest_tables;
            let dir_tables = self.list_directory_tables().await?;
            for table_name in dir_tables {
                // Check if this table's location is already in the manifest
                // Manifest stores full URIs, so we need to check both formats
                let full_location = format!("{}/{}.lance", self.root, table_name);
                let relative_location = format!("{}.lance", table_name);
                if !manifest_locations.contains(&full_location)
                    && !manifest_locations.contains(&relative_location)
                {
                    all_tables.push(table_name);
                }
            }

            all_tables
        } else {
            self.list_directory_tables().await?
        };

        // Apply sorting and pagination
        Self::apply_pagination(&mut tables, request.page_token, request.limit);
        let response = ListTablesResponse::new(tables);
        Ok(response)
    }

    async fn describe_table(&self, request: DescribeTableRequest) -> Result<DescribeTableResponse> {
        if let Some(ref manifest_ns) = self.manifest_ns {
            match manifest_ns.describe_table(request.clone()).await {
                Ok(mut response) => {
                    if let Some(ref table_uri) = response.table_uri {
                        // For backwards compatibility, only skip vending credentials when explicitly set to false
                        let vend = request.vend_credentials.unwrap_or(true);
                        let identity = request.identity.as_deref();
                        response.storage_options = self
                            .get_storage_options_for_table(table_uri, vend, identity)
                            .await?;
                    }
                    // Set managed_versioning flag when table_version_tracking_enabled
                    if self.table_version_tracking_enabled {
                        response.managed_versioning = Some(true);
                    }
                    return Ok(response);
                }
                Err(_)
                    if self.dir_listing_enabled
                        && request.id.as_ref().is_some_and(|id| id.len() == 1) =>
                {
                    // Fall through to directory check only for single-level IDs
                }
                Err(e) => return Err(e),
            }
        }

        let table_name = Self::table_name_from_id(&request.id)?;
        let table_uri = self.table_full_uri(&table_name);

        // Atomically check table existence and deregistration status
        let status = self.check_table_status(&table_name).await;

        if !status.exists {
            return Err(Error::namespace_source(
                format!("Table does not exist: {}", table_name).into(),
            ));
        }

        if status.is_deregistered {
            return Err(Error::namespace_source(
                format!("Table is deregistered: {}", table_name).into(),
            ));
        }

        let load_detailed_metadata = request.load_detailed_metadata.unwrap_or(false);
        // For backwards compatibility, only skip vending credentials when explicitly set to false
        let vend_credentials = request.vend_credentials.unwrap_or(true);
        let identity = request.identity.as_deref();

        // If not loading detailed metadata, return minimal response with just location
        if !load_detailed_metadata {
            let storage_options = self
                .get_storage_options_for_table(&table_uri, vend_credentials, identity)
                .await?;
            return Ok(DescribeTableResponse {
                table: Some(table_name),
                namespace: request.id.as_ref().map(|id| {
                    if id.len() > 1 {
                        id[..id.len() - 1].to_vec()
                    } else {
                        vec![]
                    }
                }),
                location: Some(table_uri.clone()),
                table_uri: Some(table_uri),
                storage_options,
                managed_versioning: if self.table_version_tracking_enabled {
                    Some(true)
                } else {
                    None
                },
                ..Default::default()
            });
        }

        // Try to load the dataset to get real information
        // Use DatasetBuilder with storage options to support S3 with custom endpoints
        let mut builder = DatasetBuilder::from_uri(&table_uri);
        if let Some(opts) = &self.storage_options {
            builder = builder.with_storage_options(opts.clone());
        }
        if let Some(sess) = &self.session {
            builder = builder.with_session(sess.clone());
        }
        match builder.load().await {
            Ok(mut dataset) => {
                // If a specific version is requested, checkout that version
                if let Some(requested_version) = request.version {
                    dataset = dataset.checkout_version(requested_version as u64).await?;
                }

                let version_info = dataset.version();
                let lance_schema = dataset.schema();
                let arrow_schema: arrow_schema::Schema = lance_schema.into();
                let json_schema = arrow_schema_to_json(&arrow_schema)?;
                let storage_options = self
                    .get_storage_options_for_table(&table_uri, vend_credentials, identity)
                    .await?;

                // Convert BTreeMap to HashMap for the response
                let metadata: std::collections::HashMap<String, String> =
                    version_info.metadata.into_iter().collect();

                Ok(DescribeTableResponse {
                    table: Some(table_name),
                    namespace: request.id.as_ref().map(|id| {
                        if id.len() > 1 {
                            id[..id.len() - 1].to_vec()
                        } else {
                            vec![]
                        }
                    }),
                    version: Some(version_info.version as i64),
                    location: Some(table_uri.clone()),
                    table_uri: Some(table_uri),
                    schema: Some(Box::new(json_schema)),
                    storage_options,
                    metadata: Some(metadata),
                    managed_versioning: if self.table_version_tracking_enabled {
                        Some(true)
                    } else {
                        None
                    },
                    ..Default::default()
                })
            }
            Err(err) => {
                // Use the reserved file status from the atomic check
                if status.has_reserved_file {
                    let storage_options = self
                        .get_storage_options_for_table(&table_uri, vend_credentials, identity)
                        .await?;
                    Ok(DescribeTableResponse {
                        table: Some(table_name),
                        namespace: request.id.as_ref().map(|id| {
                            if id.len() > 1 {
                                id[..id.len() - 1].to_vec()
                            } else {
                                vec![]
                            }
                        }),
                        location: Some(table_uri.clone()),
                        table_uri: Some(table_uri),
                        storage_options,
                        managed_versioning: if self.table_version_tracking_enabled {
                            Some(true)
                        } else {
                            None
                        },
                        ..Default::default()
                    })
                } else {
                    Err(Error::namespace_source(
                        format!(
                            "Table directory exists but cannot load dataset {}: {:?}",
                            table_name, err
                        )
                        .into(),
                    ))
                }
            }
        }
    }

    async fn table_exists(&self, request: TableExistsRequest) -> Result<()> {
        if let Some(ref manifest_ns) = self.manifest_ns {
            match manifest_ns.table_exists(request.clone()).await {
                Ok(()) => return Ok(()),
                Err(_) if self.dir_listing_enabled => {
                    // Fall through to directory check
                }
                Err(e) => return Err(e),
            }
        }

        let table_name = Self::table_name_from_id(&request.id)?;

        // Atomically check table existence and deregistration status
        let status = self.check_table_status(&table_name).await;

        if !status.exists {
            return Err(Error::namespace_source(
                format!("Table does not exist: {}", table_name).into(),
            ));
        }

        if status.is_deregistered {
            return Err(Error::namespace_source(
                format!("Table is deregistered: {}", table_name).into(),
            ));
        }

        Ok(())
    }

    async fn drop_table(&self, request: DropTableRequest) -> Result<DropTableResponse> {
        if let Some(ref manifest_ns) = self.manifest_ns {
            return manifest_ns.drop_table(request).await;
        }

        let table_name = Self::table_name_from_id(&request.id)?;
        let table_uri = self.table_full_uri(&table_name);
        let table_path = self.table_path(&table_name);

        self.object_store
            .remove_dir_all(table_path)
            .await
            .map_err(|e| {
                Error::namespace_source(
                    format!("Failed to drop table {}: {}", table_name, e).into(),
                )
            })?;

        Ok(DropTableResponse {
            id: request.id,
            location: Some(table_uri),
            ..Default::default()
        })
    }

    async fn create_table(
        &self,
        request: CreateTableRequest,
        request_data: Bytes,
    ) -> Result<CreateTableResponse> {
        if let Some(ref manifest_ns) = self.manifest_ns {
            return manifest_ns.create_table(request, request_data).await;
        }

        let table_name = Self::table_name_from_id(&request.id)?;
        let table_uri = self.table_full_uri(&table_name);
        if request_data.is_empty() {
            return Err(Error::namespace_source(
                "Request data (Arrow IPC stream) is required for create_table".into(),
            ));
        }

        // Parse the Arrow IPC stream from request_data
        let cursor = Cursor::new(request_data.to_vec());
        let stream_reader = StreamReader::try_new(cursor, None).map_err(|e| {
            Error::namespace_source(format!("Invalid Arrow IPC stream: {}", e).into())
        })?;
        let arrow_schema = stream_reader.schema();

        // Collect all batches from the stream
        let mut batches = Vec::new();
        for batch_result in stream_reader {
            batches.push(batch_result.map_err(|e| {
                Error::namespace_source(
                    format!("Failed to read batch from IPC stream: {}", e).into(),
                )
            })?);
        }

        // Create RecordBatchReader from the batches
        let reader = if batches.is_empty() {
            let batch = arrow::record_batch::RecordBatch::new_empty(arrow_schema.clone());
            let batches = vec![Ok(batch)];
            RecordBatchIterator::new(batches, arrow_schema.clone())
        } else {
            let batch_results: Vec<_> = batches.into_iter().map(Ok).collect();
            RecordBatchIterator::new(batch_results, arrow_schema)
        };

        let store_params = self.storage_options.as_ref().map(|opts| ObjectStoreParams {
            storage_options_accessor: Some(Arc::new(
                lance_io::object_store::StorageOptionsAccessor::with_static_options(opts.clone()),
            )),
            ..Default::default()
        });

        let write_params = WriteParams {
            mode: WriteMode::Create,
            store_params,
            ..Default::default()
        };

        // Create the Lance dataset using the actual Lance API
        Dataset::write(reader, &table_uri, Some(write_params))
            .await
            .map_err(|e| {
                Error::namespace_source(format!("Failed to create Lance dataset: {}", e).into())
            })?;

        Ok(CreateTableResponse {
            version: Some(1),
            location: Some(table_uri),
            storage_options: self.storage_options.clone(),
            ..Default::default()
        })
    }

    async fn declare_table(&self, request: DeclareTableRequest) -> Result<DeclareTableResponse> {
        if let Some(ref manifest_ns) = self.manifest_ns {
            let mut response = manifest_ns.declare_table(request.clone()).await?;
            if let Some(ref location) = response.location {
                // For backwards compatibility, only skip vending credentials when explicitly set to false
                let vend = request.vend_credentials.unwrap_or(true);
                let identity = request.identity.as_deref();
                response.storage_options = self
                    .get_storage_options_for_table(location, vend, identity)
                    .await?;
            }
            // Set managed_versioning when table_version_tracking_enabled
            if self.table_version_tracking_enabled {
                response.managed_versioning = Some(true);
            }
            return Ok(response);
        }

        let table_name = Self::table_name_from_id(&request.id)?;
        let table_uri = self.table_full_uri(&table_name);

        // Validate location if provided
        if let Some(location) = &request.location {
            let location = location.trim_end_matches('/');
            if location != table_uri {
                return Err(Error::namespace_source(
                    format!(
                        "Cannot declare table {} at location {}, must be at location {}",
                        table_name, location, table_uri
                    )
                    .into(),
                ));
            }
        }

        // Check if table already has data (created via create_table).
        // The atomic put only prevents races between concurrent declare_table calls,
        // not between declare_table and existing data.
        let status = self.check_table_status(&table_name).await;
        if status.exists && !status.has_reserved_file {
            // Table has data but no reserved file - it was created with data
            return Err(Error::namespace_source(
                format!("Table already exists: {}", table_name).into(),
            ));
        }

        // Atomically create the .lance-reserved file to mark the table as declared.
        // This uses put_if_not_exists semantics to avoid race conditions between
        // concurrent declare_table calls.
        let reserved_file_path = self.table_reserved_file_path(&table_name);

        self.put_marker_file_atomic(&reserved_file_path, &format!("table {}", table_name))
            .await
            .map_err(|e| Error::namespace_source(e.into()))?;

        // For backwards compatibility, only skip vending credentials when explicitly set to false
        let vend_credentials = request.vend_credentials.unwrap_or(true);
        let identity = request.identity.as_deref();
        let storage_options = self
            .get_storage_options_for_table(&table_uri, vend_credentials, identity)
            .await?;

        Ok(DeclareTableResponse {
            location: Some(table_uri),
            storage_options,
            managed_versioning: if self.table_version_tracking_enabled {
                Some(true)
            } else {
                None
            },
            ..Default::default()
        })
    }

    async fn register_table(
        &self,
        request: lance_namespace::models::RegisterTableRequest,
    ) -> Result<lance_namespace::models::RegisterTableResponse> {
        // If manifest is enabled, delegate to manifest namespace
        if let Some(ref manifest_ns) = self.manifest_ns {
            return LanceNamespace::register_table(manifest_ns.as_ref(), request).await;
        }

        // Without manifest, register_table is not supported
        Err(Error::not_supported_source(
            "register_table is only supported when manifest mode is enabled".into(),
        ))
    }

    async fn deregister_table(
        &self,
        request: lance_namespace::models::DeregisterTableRequest,
    ) -> Result<lance_namespace::models::DeregisterTableResponse> {
        // If manifest is enabled, delegate to manifest namespace
        if let Some(ref manifest_ns) = self.manifest_ns {
            return LanceNamespace::deregister_table(manifest_ns.as_ref(), request).await;
        }

        // V1 mode: create a .lance-deregistered marker file in the table directory
        let table_name = Self::table_name_from_id(&request.id)?;
        let table_uri = self.table_full_uri(&table_name);

        // Check table existence and deregistration status.
        // This provides better error messages for common cases.
        let status = self.check_table_status(&table_name).await;

        if !status.exists {
            return Err(Error::namespace_source(
                format!("Table does not exist: {}", table_name).into(),
            ));
        }

        if status.is_deregistered {
            return Err(Error::namespace_source(
                format!("Table is already deregistered: {}", table_name).into(),
            ));
        }

        // Atomically create the .lance-deregistered marker file.
        // This uses put_if_not_exists semantics to prevent race conditions
        // when multiple processes try to deregister the same table concurrently.
        // If a race occurs and another process already created the file,
        // we'll get an AlreadyExists error which we convert to a proper message.
        let deregistered_path = self.table_deregistered_file_path(&table_name);
        self.put_marker_file_atomic(
            &deregistered_path,
            &format!("deregistration marker for table {}", table_name),
        )
        .await
        .map_err(|e| {
            // Convert "already exists" to "already deregistered" for better UX
            let message = if e.contains("already exists") {
                format!("Table is already deregistered: {}", table_name)
            } else {
                e
            };
            Error::namespace_source(message.into())
        })?;

        Ok(lance_namespace::models::DeregisterTableResponse {
            id: request.id,
            location: Some(table_uri),
            ..Default::default()
        })
    }

    async fn list_table_versions(
        &self,
        request: ListTableVersionsRequest,
    ) -> Result<ListTableVersionsResponse> {
        // When table_version_storage_enabled, query from __manifest
        if self.table_version_storage_enabled
            && let Some(ref manifest_ns) = self.manifest_ns
        {
            let table_id = request.id.clone().unwrap_or_default();
            let want_descending = request.descending == Some(true);
            return manifest_ns
                .list_table_versions(&table_id, want_descending, request.limit)
                .await;
        }

        // Fallback when table_version_storage is not enabled: list from _versions/ directory
        let table_uri = self.resolve_table_location(&request.id).await?;

        let table_path = Self::uri_to_object_store_path(&table_uri);
        let versions_dir = table_path.child("_versions");
        let manifest_metas: Vec<_> = self
            .object_store
            .read_dir_all(&versions_dir, None)
            .try_collect()
            .await
            .map_err(|e| {
                Error::namespace_source(
                    format!(
                        "Failed to list manifest files for table at '{}': {}",
                        table_uri, e
                    )
                    .into(),
                )
            })?;

        let is_v2_naming = manifest_metas
            .first()
            .is_some_and(|meta| meta.location.filename().is_some_and(|f| f.len() == 29));

        let mut table_versions: Vec<TableVersion> = manifest_metas
            .into_iter()
            .filter_map(|meta| {
                let filename = meta.location.filename()?;
                let version_str = filename.strip_suffix(".manifest")?;
                if version_str.starts_with('d') {
                    return None;
                }
                let file_version: u64 = version_str.parse().ok()?;

                let actual_version = if file_version > u64::MAX / 2 {
                    u64::MAX - file_version
                } else {
                    file_version
                };

                // Use full path from object_store (relative to object store root)
                Some(TableVersion {
                    version: actual_version as i64,
                    manifest_path: meta.location.to_string(),
                    manifest_size: Some(meta.size as i64),
                    e_tag: meta.e_tag,
                    timestamp_millis: Some(meta.last_modified.timestamp_millis()),
                    metadata: None,
                })
            })
            .collect();

        let list_is_ordered = self.object_store.list_is_lexically_ordered;
        let want_descending = request.descending == Some(true);

        let needs_sort = if list_is_ordered {
            if is_v2_naming {
                !want_descending
            } else {
                want_descending
            }
        } else {
            true
        };

        if needs_sort {
            if want_descending {
                table_versions.sort_by(|a, b| b.version.cmp(&a.version));
            } else {
                table_versions.sort_by(|a, b| a.version.cmp(&b.version));
            }
        }

        if let Some(limit) = request.limit {
            table_versions.truncate(limit as usize);
        }

        Ok(ListTableVersionsResponse {
            versions: table_versions,
            page_token: None,
        })
    }

    async fn create_table_version(
        &self,
        request: CreateTableVersionRequest,
    ) -> Result<CreateTableVersionResponse> {
        let table_uri = self.resolve_table_location(&request.id).await?;

        let staging_manifest_path = &request.manifest_path;
        let version = request.version as u64;

        let table_path = Self::uri_to_object_store_path(&table_uri);

        // Determine naming scheme from request, default to V2
        let naming_scheme = match request.naming_scheme.as_deref() {
            Some("V1") => ManifestNamingScheme::V1,
            _ => ManifestNamingScheme::V2,
        };

        // Compute final path using the naming scheme
        let final_path = naming_scheme.manifest_path(&table_path, version);

        let staging_path = Self::uri_to_object_store_path(staging_manifest_path);
        let manifest_data = self
            .object_store
            .inner
            .get(&staging_path)
            .await
            .map_err(|e| {
                Error::namespace_source(
                    format!(
                        "Failed to read staging manifest at '{}': {}",
                        staging_manifest_path, e
                    )
                    .into(),
                )
            })?
            .bytes()
            .await
            .map_err(|e| {
                Error::namespace_source(
                    format!(
                        "Failed to read staging manifest bytes at '{}': {}",
                        staging_manifest_path, e
                    )
                    .into(),
                )
            })?;

        let manifest_size = manifest_data.len() as i64;

        let put_result = self
            .object_store
            .inner
            .put_opts(
                &final_path,
                manifest_data.into(),
                PutOptions {
                    mode: PutMode::Create,
                    ..Default::default()
                },
            )
            .await
            .map_err(|e| match e {
                object_store::Error::AlreadyExists { .. }
                | object_store::Error::Precondition { .. } => Error::namespace_source(
                    format!(
                        "Version {} already exists for table at '{}'",
                        version, table_uri
                    )
                    .into(),
                ),
                _ => Error::namespace_source(
                    format!(
                        "Failed to create version {} for table at '{}': {}",
                        version, table_uri, e
                    )
                    .into(),
                ),
            })?;

        // Delete the staging manifest after successful copy
        if let Err(e) = self.object_store.inner.delete(&staging_path).await {
            log::warn!(
                "Failed to delete staging manifest at '{}': {:?}",
                staging_path,
                e
            );
        }

        // If table_version_storage_enabled is enabled, also record in __manifest (best-effort)
        if self.table_version_storage_enabled
            && let Some(ref manifest_ns) = self.manifest_ns
        {
            let table_id_str =
                manifest::ManifestNamespace::str_object_id(&request.id.clone().unwrap_or_default());
            let object_id =
                manifest::ManifestNamespace::build_version_object_id(&table_id_str, version as i64);
            let metadata_json = serde_json::json!({
                "manifest_path": final_path.to_string(),
                "manifest_size": manifest_size,
                "e_tag": put_result.e_tag,
                "naming_scheme": request.naming_scheme.as_deref().unwrap_or("V2"),
            })
            .to_string();

            if let Err(e) = manifest_ns
                .insert_into_manifest_with_metadata(
                    vec![manifest::ManifestEntry {
                        object_id,
                        object_type: manifest::ObjectType::TableVersion,
                        location: None,
                        metadata: Some(metadata_json),
                    }],
                    None,
                )
                .await
            {
                log::warn!(
                    "Failed to record table version in __manifest (best-effort): {:?}",
                    e
                );
            }
        }

        Ok(CreateTableVersionResponse {
            transaction_id: None,
            version: Some(Box::new(TableVersion {
                version: version as i64,
                manifest_path: final_path.to_string(),
                manifest_size: Some(manifest_size),
                e_tag: put_result.e_tag,
                timestamp_millis: None,
                metadata: None,
            })),
        })
    }

    async fn describe_table_version(
        &self,
        request: DescribeTableVersionRequest,
    ) -> Result<DescribeTableVersionResponse> {
        // When table_version_storage_enabled and a specific version is requested,
        // query from __manifest to avoid opening the entire dataset
        if self.table_version_storage_enabled
            && let (Some(manifest_ns), Some(version)) = (&self.manifest_ns, request.version)
        {
            let table_id = request.id.clone().unwrap_or_default();
            return manifest_ns.describe_table_version(&table_id, version).await;
        }

        // Fallback when table_version_storage is not enabled: open the dataset to describe the version
        let table_uri = self.resolve_table_location(&request.id).await?;

        // Use DatasetBuilder with storage options to support S3 with custom endpoints
        let mut builder = DatasetBuilder::from_uri(&table_uri);
        if let Some(opts) = &self.storage_options {
            builder = builder.with_storage_options(opts.clone());
        }
        if let Some(sess) = &self.session {
            builder = builder.with_session(sess.clone());
        }
        let mut dataset = builder.load().await.map_err(|e| {
            Error::namespace_source(
                format!("Failed to open table at '{}': {}", table_uri, e).into(),
            )
        })?;

        if let Some(version) = request.version {
            dataset = dataset
                .checkout_version(version as u64)
                .await
                .map_err(|e| {
                    Error::namespace_source(
                        format!(
                            "Failed to checkout version {} for table at '{}': {}",
                            version, table_uri, e
                        )
                        .into(),
                    )
                })?;
        }

        let version_info = dataset.version();
        let manifest_location = dataset.manifest_location();
        let metadata: std::collections::HashMap<String, String> =
            version_info.metadata.into_iter().collect();

        let table_version = TableVersion {
            version: version_info.version as i64,
            manifest_path: manifest_location.path.to_string(),
            manifest_size: manifest_location.size.map(|s| s as i64),
            e_tag: manifest_location.e_tag.clone(),
            timestamp_millis: Some(version_info.timestamp.timestamp_millis()),
            metadata: if metadata.is_empty() {
                None
            } else {
                Some(metadata)
            },
        };

        Ok(DescribeTableVersionResponse {
            version: Box::new(table_version),
        })
    }

    async fn batch_delete_table_versions(
        &self,
        request: BatchDeleteTableVersionsRequest,
    ) -> Result<BatchDeleteTableVersionsResponse> {
        // Single-table mode: use `id` (from path parameter) + `ranges` to delete
        // versions from one table.
        let ranges: Vec<(i64, i64)> = request
            .ranges
            .iter()
            .map(|r| {
                let start = r.start_version;
                let end = if r.end_version > 0 {
                    r.end_version
                } else {
                    start
                };
                (start, end)
            })
            .collect();
        let table_entries = vec![TableDeleteEntry {
            table_id: request.id.clone(),
            ranges,
        }];

        let mut total_deleted_count = 0i64;

        if self.table_version_storage_enabled
            && let Some(ref manifest_ns) = self.manifest_ns
        {
            // Phase 1 (atomic commit point): Delete version records from __manifest
            // for ALL tables in a single atomic operation. This is the authoritative
            // source of truth — once __manifest entries are removed, the versions
            // are logically deleted across all tables atomically.

            // Collect all (table_id_str, ranges) for batch deletion
            let mut all_object_ids: Vec<String> = Vec::new();
            for te in &table_entries {
                let table_id_str = manifest::ManifestNamespace::str_object_id(
                    &te.table_id.clone().unwrap_or_default(),
                );
                for (start, end) in &te.ranges {
                    for version in *start..=*end {
                        let object_id = manifest::ManifestNamespace::build_version_object_id(
                            &table_id_str,
                            version,
                        );
                        all_object_ids.push(object_id);
                    }
                }
            }

            if !all_object_ids.is_empty() {
                total_deleted_count = manifest_ns
                    .batch_delete_table_versions_by_object_ids(&all_object_ids)
                    .await?;
            }

            // Phase 2: Delete physical manifest files (best-effort).
            // Even if some file deletions fail, the versions are already removed from
            // __manifest, so they won't be visible to readers. Leftover files are
            // orphaned but harmless and can be cleaned up later.
            let _ = self
                .delete_physical_version_files(&table_entries, true)
                .await;

            return Ok(BatchDeleteTableVersionsResponse {
                deleted_count: Some(total_deleted_count),
                transaction_id: None,
            });
        }

        // Fallback when table_version_storage is not enabled: delete physical files directly (no __manifest)
        total_deleted_count = self
            .delete_physical_version_files(&table_entries, false)
            .await?;

        Ok(BatchDeleteTableVersionsResponse {
            deleted_count: Some(total_deleted_count),
            transaction_id: None,
        })
    }

    fn namespace_id(&self) -> String {
        format!("DirectoryNamespace {{ root: {:?} }}", self.root)
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use arrow_ipc::reader::StreamReader;
    use lance::dataset::Dataset;
    use lance_core::utils::tempfile::{TempStdDir, TempStrDir};
    use lance_namespace::models::{
        CreateTableRequest, JsonArrowDataType, JsonArrowField, JsonArrowSchema, ListTablesRequest,
    };
    use lance_namespace::schema::convert_json_arrow_schema;
    use std::io::Cursor;
    use std::sync::Arc;

    /// Helper to create a test DirectoryNamespace with a temporary directory
    async fn create_test_namespace() -> (DirectoryNamespace, TempStdDir) {
        let temp_dir = TempStdDir::default();

        let namespace = DirectoryNamespaceBuilder::new(temp_dir.to_str().unwrap())
            .build()
            .await
            .unwrap();
        (namespace, temp_dir)
    }

    /// Helper to create test IPC data from a schema
    fn create_test_ipc_data(schema: &JsonArrowSchema) -> Vec<u8> {
        use arrow::ipc::writer::StreamWriter;

        let arrow_schema = convert_json_arrow_schema(schema).unwrap();
        let arrow_schema = Arc::new(arrow_schema);
        let batch = arrow::record_batch::RecordBatch::new_empty(arrow_schema.clone());
        let mut buffer = Vec::new();
        {
            let mut writer = StreamWriter::try_new(&mut buffer, &arrow_schema).unwrap();
            writer.write(&batch).unwrap();
            writer.finish().unwrap();
        }
        buffer
    }

    /// Helper to create a simple test schema
    fn create_test_schema() -> JsonArrowSchema {
        let int_type = JsonArrowDataType::new("int32".to_string());
        let string_type = JsonArrowDataType::new("utf8".to_string());

        let id_field = JsonArrowField {
            name: "id".to_string(),
            r#type: Box::new(int_type),
            nullable: false,
            metadata: None,
        };

        let name_field = JsonArrowField {
            name: "name".to_string(),
            r#type: Box::new(string_type),
            nullable: true,
            metadata: None,
        };

        JsonArrowSchema {
            fields: vec![id_field, name_field],
            metadata: None,
        }
    }

    #[tokio::test]
    async fn test_create_table() {
        let (namespace, _temp_dir) = create_test_namespace().await;

        // Create test IPC data
        let schema = create_test_schema();
        let ipc_data = create_test_ipc_data(&schema);

        let mut request = CreateTableRequest::new();
        request.id = Some(vec!["test_table".to_string()]);

        let response = namespace
            .create_table(request, bytes::Bytes::from(ipc_data))
            .await
            .unwrap();

        assert!(response.location.is_some());
        assert!(response.location.unwrap().ends_with("test_table.lance"));
        assert_eq!(response.version, Some(1));
    }

    #[tokio::test]
    async fn test_create_table_without_data() {
        let (namespace, _temp_dir) = create_test_namespace().await;

        let mut request = CreateTableRequest::new();
        request.id = Some(vec!["test_table".to_string()]);

        let result = namespace.create_table(request, bytes::Bytes::new()).await;
        assert!(result.is_err());
        assert!(
            result
                .unwrap_err()
                .to_string()
                .contains("Arrow IPC stream) is required")
        );
    }

    #[tokio::test]
    async fn test_create_table_with_invalid_id() {
        let (namespace, _temp_dir) = create_test_namespace().await;

        // Create test IPC data
        let schema = create_test_schema();
        let ipc_data = create_test_ipc_data(&schema);

        // Test with empty ID
        let mut request = CreateTableRequest::new();
        request.id = Some(vec![]);

        let result = namespace
            .create_table(request, bytes::Bytes::from(ipc_data.clone()))
            .await;
        assert!(result.is_err());

        // Test with multi-level ID - should now work with manifest enabled
        // First create the parent namespace
        let mut create_ns_req = CreateNamespaceRequest::new();
        create_ns_req.id = Some(vec!["test_namespace".to_string()]);
        namespace.create_namespace(create_ns_req).await.unwrap();

        // Now create table in the namespace
        let mut request = CreateTableRequest::new();
        request.id = Some(vec!["test_namespace".to_string(), "table".to_string()]);

        let result = namespace
            .create_table(request, bytes::Bytes::from(ipc_data))
            .await;
        // Should succeed with manifest enabled
        assert!(
            result.is_ok(),
            "Multi-level table IDs should work with manifest enabled"
        );
    }

    #[tokio::test]
    async fn test_list_tables() {
        let (namespace, _temp_dir) = create_test_namespace().await;

        // Initially, no tables
        let mut request = ListTablesRequest::new();
        request.id = Some(vec![]);
        let response = namespace.list_tables(request).await.unwrap();
        assert_eq!(response.tables.len(), 0);

        // Create test IPC data
        let schema = create_test_schema();
        let ipc_data = create_test_ipc_data(&schema);

        // Create a table
        let mut create_request = CreateTableRequest::new();
        create_request.id = Some(vec!["table1".to_string()]);
        namespace
            .create_table(create_request, bytes::Bytes::from(ipc_data.clone()))
            .await
            .unwrap();

        // Create another table
        let mut create_request = CreateTableRequest::new();
        create_request.id = Some(vec!["table2".to_string()]);
        namespace
            .create_table(create_request, bytes::Bytes::from(ipc_data))
            .await
            .unwrap();

        // List tables should return both
        let mut request = ListTablesRequest::new();
        request.id = Some(vec![]);
        let response = namespace.list_tables(request).await.unwrap();
        let tables = response.tables;
        assert_eq!(tables.len(), 2);
        assert!(tables.contains(&"table1".to_string()));
        assert!(tables.contains(&"table2".to_string()));
    }

    #[tokio::test]
    async fn test_list_tables_with_namespace_id() {
        let (namespace, _temp_dir) = create_test_namespace().await;

        // First create a child namespace
        let mut create_ns_req = CreateNamespaceRequest::new();
        create_ns_req.id = Some(vec!["test_namespace".to_string()]);
        namespace.create_namespace(create_ns_req).await.unwrap();

        // Now list tables in the child namespace
        let mut request = ListTablesRequest::new();
        request.id = Some(vec!["test_namespace".to_string()]);

        let result = namespace.list_tables(request).await;
        // Should succeed (with manifest enabled) and return empty list (no tables yet)
        assert!(
            result.is_ok(),
            "list_tables should work with child namespace when manifest is enabled"
        );
        let response = result.unwrap();
        assert_eq!(
            response.tables.len(),
            0,
            "Namespace should have no tables yet"
        );
    }

    #[tokio::test]
    async fn test_describe_table() {
        let (namespace, _temp_dir) = create_test_namespace().await;

        // Create a table first
        let schema = create_test_schema();
        let ipc_data = create_test_ipc_data(&schema);

        let mut create_request = CreateTableRequest::new();
        create_request.id = Some(vec!["test_table".to_string()]);
        namespace
            .create_table(create_request, bytes::Bytes::from(ipc_data))
            .await
            .unwrap();

        // Describe the table
        let mut request = DescribeTableRequest::new();
        request.id = Some(vec!["test_table".to_string()]);
        let response = namespace.describe_table(request).await.unwrap();

        assert!(response.location.is_some());
        assert!(response.location.unwrap().ends_with("test_table.lance"));
    }

    #[tokio::test]
    async fn test_describe_nonexistent_table() {
        let (namespace, _temp_dir) = create_test_namespace().await;

        let mut request = DescribeTableRequest::new();
        request.id = Some(vec!["nonexistent".to_string()]);

        let result = namespace.describe_table(request).await;
        assert!(result.is_err());
        assert!(
            result
                .unwrap_err()
                .to_string()
                .contains("Table does not exist")
        );
    }

    #[tokio::test]
    async fn test_table_exists() {
        let (namespace, _temp_dir) = create_test_namespace().await;

        // Create a table
        let schema = create_test_schema();
        let ipc_data = create_test_ipc_data(&schema);

        let mut create_request = CreateTableRequest::new();
        create_request.id = Some(vec!["existing_table".to_string()]);
        namespace
            .create_table(create_request, bytes::Bytes::from(ipc_data))
            .await
            .unwrap();

        // Check existing table
        let mut request = TableExistsRequest::new();
        request.id = Some(vec!["existing_table".to_string()]);
        let result = namespace.table_exists(request).await;
        assert!(result.is_ok());

        // Check non-existent table
        let mut request = TableExistsRequest::new();
        request.id = Some(vec!["nonexistent".to_string()]);
        let result = namespace.table_exists(request).await;
        assert!(result.is_err());
        assert!(
            result
                .unwrap_err()
                .to_string()
                .contains("Table does not exist")
        );
    }

    #[tokio::test]
    async fn test_drop_table() {
        let (namespace, _temp_dir) = create_test_namespace().await;

        // Create a table
        let schema = create_test_schema();
        let ipc_data = create_test_ipc_data(&schema);

        let mut create_request = CreateTableRequest::new();
        create_request.id = Some(vec!["table_to_drop".to_string()]);
        namespace
            .create_table(create_request, bytes::Bytes::from(ipc_data))
            .await
            .unwrap();

        // Verify it exists
        let mut exists_request = TableExistsRequest::new();
        exists_request.id = Some(vec!["table_to_drop".to_string()]);
        assert!(namespace.table_exists(exists_request.clone()).await.is_ok());

        // Drop the table
        let mut drop_request = DropTableRequest::new();
        drop_request.id = Some(vec!["table_to_drop".to_string()]);
        let response = namespace.drop_table(drop_request).await.unwrap();
        assert!(response.location.is_some());

        // Verify it no longer exists
        assert!(namespace.table_exists(exists_request).await.is_err());
    }

    #[tokio::test]
    async fn test_drop_nonexistent_table() {
        let (namespace, _temp_dir) = create_test_namespace().await;

        let mut request = DropTableRequest::new();
        request.id = Some(vec!["nonexistent".to_string()]);

        // Should not fail when dropping non-existent table (idempotent)
        let result = namespace.drop_table(request).await;
        // The operation might succeed or fail depending on implementation
        // But it should not panic
        let _ = result;
    }

    #[tokio::test]
    async fn test_root_namespace_operations() {
        let (namespace, _temp_dir) = create_test_namespace().await;

        // Test list_namespaces - should return empty list for root
        let mut request = ListNamespacesRequest::new();
        request.id = Some(vec![]);
        let result = namespace.list_namespaces(request).await;
        assert!(result.is_ok());
        assert_eq!(result.unwrap().namespaces.len(), 0);

        // Test describe_namespace - should succeed for root
        let mut request = DescribeNamespaceRequest::new();
        request.id = Some(vec![]);
        let result = namespace.describe_namespace(request).await;
        assert!(result.is_ok());

        // Test namespace_exists - root always exists
        let mut request = NamespaceExistsRequest::new();
        request.id = Some(vec![]);
        let result = namespace.namespace_exists(request).await;
        assert!(result.is_ok());

        // Test create_namespace - root cannot be created
        let mut request = CreateNamespaceRequest::new();
        request.id = Some(vec![]);
        let result = namespace.create_namespace(request).await;
        assert!(result.is_err());
        assert!(result.unwrap_err().to_string().contains("already exists"));

        // Test drop_namespace - root cannot be dropped
        let mut request = DropNamespaceRequest::new();
        request.id = Some(vec![]);
        let result = namespace.drop_namespace(request).await;
        assert!(result.is_err());
        assert!(
            result
                .unwrap_err()
                .to_string()
                .contains("cannot be dropped")
        );
    }

    #[tokio::test]
    async fn test_non_root_namespace_operations() {
        let (namespace, _temp_dir) = create_test_namespace().await;

        // With manifest enabled (default), child namespaces are now supported
        // Test create_namespace for non-root - should succeed with manifest
        let mut request = CreateNamespaceRequest::new();
        request.id = Some(vec!["child".to_string()]);
        let result = namespace.create_namespace(request).await;
        assert!(
            result.is_ok(),
            "Child namespace creation should succeed with manifest enabled"
        );

        // Test namespace_exists for non-root - should exist after creation
        let mut request = NamespaceExistsRequest::new();
        request.id = Some(vec!["child".to_string()]);
        let result = namespace.namespace_exists(request).await;
        assert!(
            result.is_ok(),
            "Child namespace should exist after creation"
        );

        // Test drop_namespace for non-root - should succeed
        let mut request = DropNamespaceRequest::new();
        request.id = Some(vec!["child".to_string()]);
        let result = namespace.drop_namespace(request).await;
        assert!(
            result.is_ok(),
            "Child namespace drop should succeed with manifest enabled"
        );

        // Verify namespace no longer exists
        let mut request = NamespaceExistsRequest::new();
        request.id = Some(vec!["child".to_string()]);
        let result = namespace.namespace_exists(request).await;
        assert!(
            result.is_err(),
            "Child namespace should not exist after drop"
        );
    }

    #[tokio::test]
    async fn test_config_custom_root() {
        let temp_dir = TempStdDir::default();
        let custom_path = temp_dir.join("custom");
        std::fs::create_dir(&custom_path).unwrap();

        let namespace = DirectoryNamespaceBuilder::new(custom_path.to_string_lossy().to_string())
            .build()
            .await
            .unwrap();

        // Create test IPC data
        let schema = create_test_schema();
        let ipc_data = create_test_ipc_data(&schema);

        // Create a table and verify location
        let mut request = CreateTableRequest::new();
        request.id = Some(vec!["test_table".to_string()]);

        let response = namespace
            .create_table(request, bytes::Bytes::from(ipc_data))
            .await
            .unwrap();

        assert!(response.location.unwrap().contains("custom"));
    }

    #[tokio::test]
    async fn test_config_storage_options() {
        let temp_dir = TempStdDir::default();

        let namespace = DirectoryNamespaceBuilder::new(temp_dir.to_str().unwrap())
            .storage_option("option1", "value1")
            .storage_option("option2", "value2")
            .build()
            .await
            .unwrap();

        // Create test IPC data
        let schema = create_test_schema();
        let ipc_data = create_test_ipc_data(&schema);

        // Create a table and check storage options are included
        let mut request = CreateTableRequest::new();
        request.id = Some(vec!["test_table".to_string()]);

        let response = namespace
            .create_table(request, bytes::Bytes::from(ipc_data))
            .await
            .unwrap();

        let storage_options = response.storage_options.unwrap();
        assert_eq!(storage_options.get("option1"), Some(&"value1".to_string()));
        assert_eq!(storage_options.get("option2"), Some(&"value2".to_string()));
    }

    /// When no credential vendor is configured, `describe_table` and
    /// `declare_table` must strip credential keys from storage options
    /// while preserving non-credential config (region, endpoint, etc.).
    #[tokio::test]
    async fn test_no_storage_options_without_vendor() {
        use lance_namespace::models::DeclareTableRequest;

        let temp_dir = TempStdDir::default();

        // No manifest, no credential vendor, but storage options with credentials
        let namespace = DirectoryNamespaceBuilder::new(temp_dir.to_str().unwrap())
            .manifest_enabled(false)
            .storage_option("aws_access_key_id", "AKID")
            .storage_option("aws_secret_access_key", "SECRET")
            .storage_option("region", "us-east-1")
            .build()
            .await
            .unwrap();

        let schema = create_test_schema();
        let ipc_data = create_test_ipc_data(&schema);

        // create_table
        let mut create_req = CreateTableRequest::new();
        create_req.id = Some(vec!["t1".to_string()]);
        namespace
            .create_table(create_req, bytes::Bytes::from(ipc_data))
            .await
            .unwrap();

        // describe_table should not return storage options without a vendor
        let mut desc_req = DescribeTableRequest::new();
        desc_req.id = Some(vec!["t1".to_string()]);
        let resp = namespace.describe_table(desc_req).await.unwrap();
        assert!(resp.storage_options.is_none());

        // declare_table should not return storage options without a vendor
        let mut decl_req = DeclareTableRequest::new();
        decl_req.id = Some(vec!["t2".to_string()]);
        let resp = namespace.declare_table(decl_req).await.unwrap();
        assert!(resp.storage_options.is_none());
    }

    /// Same test with manifest mode enabled.
    #[tokio::test]
    async fn test_no_storage_options_without_vendor_manifest() {
        let temp_dir = TempStdDir::default();

        let namespace = DirectoryNamespaceBuilder::new(temp_dir.to_str().unwrap())
            .storage_option("aws_access_key_id", "AKID")
            .storage_option("aws_secret_access_key", "SECRET")
            .storage_option("region", "us-east-1")
            .build()
            .await
            .unwrap();

        let schema = create_test_schema();
        let ipc_data = create_test_ipc_data(&schema);

        let mut create_req = CreateTableRequest::new();
        create_req.id = Some(vec!["t1".to_string()]);
        namespace
            .create_table(create_req, bytes::Bytes::from(ipc_data))
            .await
            .unwrap();

        // describe_table through manifest should not return storage options without a vendor
        let mut desc_req = DescribeTableRequest::new();
        desc_req.id = Some(vec!["t1".to_string()]);
        let resp = namespace.describe_table(desc_req).await.unwrap();
        assert!(resp.storage_options.is_none());
    }

    #[tokio::test]
    async fn test_from_properties_manifest_enabled() {
        let temp_dir = TempStdDir::default();

        let mut properties = HashMap::new();
        properties.insert("root".to_string(), temp_dir.to_str().unwrap().to_string());
        properties.insert("manifest_enabled".to_string(), "true".to_string());
        properties.insert("dir_listing_enabled".to_string(), "false".to_string());

        let builder = DirectoryNamespaceBuilder::from_properties(properties, None).unwrap();
        assert!(builder.manifest_enabled);
        assert!(!builder.dir_listing_enabled);

        let namespace = builder.build().await.unwrap();

        // Create test IPC data
        let schema = create_test_schema();
        let ipc_data = create_test_ipc_data(&schema);

        // Create a table
        let mut request = CreateTableRequest::new();
        request.id = Some(vec!["test_table".to_string()]);

        let response = namespace
            .create_table(request, bytes::Bytes::from(ipc_data))
            .await
            .unwrap();

        assert!(response.location.is_some());
    }

    #[tokio::test]
    async fn test_from_properties_dir_listing_enabled() {
        let temp_dir = TempStdDir::default();

        let mut properties = HashMap::new();
        properties.insert("root".to_string(), temp_dir.to_str().unwrap().to_string());
        properties.insert("manifest_enabled".to_string(), "false".to_string());
        properties.insert("dir_listing_enabled".to_string(), "true".to_string());

        let builder = DirectoryNamespaceBuilder::from_properties(properties, None).unwrap();
        assert!(!builder.manifest_enabled);
        assert!(builder.dir_listing_enabled);

        let namespace = builder.build().await.unwrap();

        // Create test IPC data
        let schema = create_test_schema();
        let ipc_data = create_test_ipc_data(&schema);

        // Create a table
        let mut request = CreateTableRequest::new();
        request.id = Some(vec!["test_table".to_string()]);

        let response = namespace
            .create_table(request, bytes::Bytes::from(ipc_data))
            .await
            .unwrap();

        assert!(response.location.is_some());
    }

    #[tokio::test]
    async fn test_from_properties_defaults() {
        let temp_dir = TempStdDir::default();

        let mut properties = HashMap::new();
        properties.insert("root".to_string(), temp_dir.to_str().unwrap().to_string());

        let builder = DirectoryNamespaceBuilder::from_properties(properties, None).unwrap();
        // Both should default to true
        assert!(builder.manifest_enabled);
        assert!(builder.dir_listing_enabled);
    }

    #[tokio::test]
    async fn test_from_properties_with_storage_options() {
        let temp_dir = TempStdDir::default();

        let mut properties = HashMap::new();
        properties.insert("root".to_string(), temp_dir.to_str().unwrap().to_string());
        properties.insert("manifest_enabled".to_string(), "true".to_string());
        properties.insert("storage.region".to_string(), "us-west-2".to_string());
        properties.insert("storage.bucket".to_string(), "my-bucket".to_string());

        let builder = DirectoryNamespaceBuilder::from_properties(properties, None).unwrap();
        assert!(builder.manifest_enabled);
        assert!(builder.storage_options.is_some());

        let storage_options = builder.storage_options.unwrap();
        assert_eq!(
            storage_options.get("region"),
            Some(&"us-west-2".to_string())
        );
        assert_eq!(
            storage_options.get("bucket"),
            Some(&"my-bucket".to_string())
        );
    }

    #[tokio::test]
    async fn test_various_arrow_types() {
        let (namespace, _temp_dir) = create_test_namespace().await;

        // Create schema with various types
        let fields = vec![
            JsonArrowField {
                name: "bool_col".to_string(),
                r#type: Box::new(JsonArrowDataType::new("bool".to_string())),
                nullable: true,
                metadata: None,
            },
            JsonArrowField {
                name: "int8_col".to_string(),
                r#type: Box::new(JsonArrowDataType::new("int8".to_string())),
                nullable: true,
                metadata: None,
            },
            JsonArrowField {
                name: "float64_col".to_string(),
                r#type: Box::new(JsonArrowDataType::new("float64".to_string())),
                nullable: true,
                metadata: None,
            },
            JsonArrowField {
                name: "binary_col".to_string(),
                r#type: Box::new(JsonArrowDataType::new("binary".to_string())),
                nullable: true,
                metadata: None,
            },
        ];

        let schema = JsonArrowSchema {
            fields,
            metadata: None,
        };

        // Create IPC data
        let ipc_data = create_test_ipc_data(&schema);

        let mut request = CreateTableRequest::new();
        request.id = Some(vec!["complex_table".to_string()]);

        let response = namespace
            .create_table(request, bytes::Bytes::from(ipc_data))
            .await
            .unwrap();

        assert!(response.location.is_some());
    }

    #[tokio::test]
    async fn test_connect_dir() {
        let temp_dir = TempStdDir::default();

        let namespace = DirectoryNamespaceBuilder::new(temp_dir.to_str().unwrap())
            .build()
            .await
            .unwrap();

        // Test basic operation through the concrete type
        let mut request = ListTablesRequest::new();
        request.id = Some(vec![]);
        let response = namespace.list_tables(request).await.unwrap();
        assert_eq!(response.tables.len(), 0);
    }

    #[tokio::test]
    async fn test_create_table_with_ipc_data() {
        use arrow::array::{Int32Array, StringArray};
        use arrow::ipc::writer::StreamWriter;

        let (namespace, _temp_dir) = create_test_namespace().await;

        // Create a schema with some fields
        let schema = create_test_schema();

        // Create some test data that matches the schema
        let arrow_schema = convert_json_arrow_schema(&schema).unwrap();
        let arrow_schema = Arc::new(arrow_schema);

        // Create a RecordBatch with actual data
        let id_array = Int32Array::from(vec![1, 2, 3]);
        let name_array = StringArray::from(vec!["Alice", "Bob", "Charlie"]);
        let batch = arrow::record_batch::RecordBatch::try_new(
            arrow_schema.clone(),
            vec![Arc::new(id_array), Arc::new(name_array)],
        )
        .unwrap();

        // Write the batch to an IPC stream
        let mut buffer = Vec::new();
        {
            let mut writer = StreamWriter::try_new(&mut buffer, &arrow_schema).unwrap();
            writer.write(&batch).unwrap();
            writer.finish().unwrap();
        }

        // Create table with the IPC data
        let mut request = CreateTableRequest::new();
        request.id = Some(vec!["test_table_with_data".to_string()]);

        let response = namespace
            .create_table(request, Bytes::from(buffer))
            .await
            .unwrap();

        assert_eq!(response.version, Some(1));
        assert!(
            response
                .location
                .unwrap()
                .contains("test_table_with_data.lance")
        );

        // Verify table exists
        let mut exists_request = TableExistsRequest::new();
        exists_request.id = Some(vec!["test_table_with_data".to_string()]);
        namespace.table_exists(exists_request).await.unwrap();
    }

    #[tokio::test]
    async fn test_child_namespace_create_and_list() {
        let (namespace, _temp_dir) = create_test_namespace().await;

        // Create multiple child namespaces
        for i in 1..=3 {
            let mut create_req = CreateNamespaceRequest::new();
            create_req.id = Some(vec![format!("ns{}", i)]);
            let result = namespace.create_namespace(create_req).await;
            assert!(result.is_ok(), "Failed to create child namespace ns{}", i);
        }

        // List child namespaces
        let list_req = ListNamespacesRequest {
            id: Some(vec![]),
            ..Default::default()
        };
        let result = namespace.list_namespaces(list_req).await;
        assert!(result.is_ok());
        let namespaces = result.unwrap().namespaces;
        assert_eq!(namespaces.len(), 3);
        assert!(namespaces.contains(&"ns1".to_string()));
        assert!(namespaces.contains(&"ns2".to_string()));
        assert!(namespaces.contains(&"ns3".to_string()));
    }

    #[tokio::test]
    async fn test_nested_namespace_hierarchy() {
        let (namespace, _temp_dir) = create_test_namespace().await;

        // Create parent namespace
        let mut create_req = CreateNamespaceRequest::new();
        create_req.id = Some(vec!["parent".to_string()]);
        namespace.create_namespace(create_req).await.unwrap();

        // Create nested children
        let mut create_req = CreateNamespaceRequest::new();
        create_req.id = Some(vec!["parent".to_string(), "child1".to_string()]);
        namespace.create_namespace(create_req).await.unwrap();

        let mut create_req = CreateNamespaceRequest::new();
        create_req.id = Some(vec!["parent".to_string(), "child2".to_string()]);
        namespace.create_namespace(create_req).await.unwrap();

        // List children of parent
        let list_req = ListNamespacesRequest {
            id: Some(vec!["parent".to_string()]),
            ..Default::default()
        };
        let result = namespace.list_namespaces(list_req).await;
        assert!(result.is_ok());
        let children = result.unwrap().namespaces;
        assert_eq!(children.len(), 2);
        assert!(children.contains(&"child1".to_string()));
        assert!(children.contains(&"child2".to_string()));

        // List root should only show parent
        let list_req = ListNamespacesRequest {
            id: Some(vec![]),
            ..Default::default()
        };
        let result = namespace.list_namespaces(list_req).await;
        assert!(result.is_ok());
        let root_namespaces = result.unwrap().namespaces;
        assert_eq!(root_namespaces.len(), 1);
        assert_eq!(root_namespaces[0], "parent");
    }

    #[tokio::test]
    async fn test_table_in_child_namespace() {
        let (namespace, _temp_dir) = create_test_namespace().await;

        // Create child namespace
        let mut create_ns_req = CreateNamespaceRequest::new();
        create_ns_req.id = Some(vec!["test_ns".to_string()]);
        namespace.create_namespace(create_ns_req).await.unwrap();

        // Create table in child namespace
        let schema = create_test_schema();
        let ipc_data = create_test_ipc_data(&schema);
        let mut create_table_req = CreateTableRequest::new();
        create_table_req.id = Some(vec!["test_ns".to_string(), "table1".to_string()]);
        let result = namespace
            .create_table(create_table_req, bytes::Bytes::from(ipc_data))
            .await;
        assert!(result.is_ok(), "Failed to create table in child namespace");

        // List tables in child namespace
        let list_req = ListTablesRequest {
            id: Some(vec!["test_ns".to_string()]),
            ..Default::default()
        };
        let result = namespace.list_tables(list_req).await;
        assert!(result.is_ok());
        let tables = result.unwrap().tables;
        assert_eq!(tables.len(), 1);
        assert_eq!(tables[0], "table1");

        // Verify table exists
        let mut exists_req = TableExistsRequest::new();
        exists_req.id = Some(vec!["test_ns".to_string(), "table1".to_string()]);
        let result = namespace.table_exists(exists_req).await;
        assert!(result.is_ok());

        // Describe table in child namespace
        let mut describe_req = DescribeTableRequest::new();
        describe_req.id = Some(vec!["test_ns".to_string(), "table1".to_string()]);
        let result = namespace.describe_table(describe_req).await;
        assert!(result.is_ok());
        let response = result.unwrap();
        assert!(response.location.is_some());
    }

    #[tokio::test]
    async fn test_multiple_tables_in_child_namespace() {
        let (namespace, _temp_dir) = create_test_namespace().await;

        // Create child namespace
        let mut create_ns_req = CreateNamespaceRequest::new();
        create_ns_req.id = Some(vec!["test_ns".to_string()]);
        namespace.create_namespace(create_ns_req).await.unwrap();

        // Create multiple tables
        let schema = create_test_schema();
        let ipc_data = create_test_ipc_data(&schema);
        for i in 1..=3 {
            let mut create_table_req = CreateTableRequest::new();
            create_table_req.id = Some(vec!["test_ns".to_string(), format!("table{}", i)]);
            namespace
                .create_table(create_table_req, bytes::Bytes::from(ipc_data.clone()))
                .await
                .unwrap();
        }

        // List tables
        let list_req = ListTablesRequest {
            id: Some(vec!["test_ns".to_string()]),
            ..Default::default()
        };
        let result = namespace.list_tables(list_req).await;
        assert!(result.is_ok());
        let tables = result.unwrap().tables;
        assert_eq!(tables.len(), 3);
        assert!(tables.contains(&"table1".to_string()));
        assert!(tables.contains(&"table2".to_string()));
        assert!(tables.contains(&"table3".to_string()));
    }

    #[tokio::test]
    async fn test_drop_table_in_child_namespace() {
        let (namespace, _temp_dir) = create_test_namespace().await;

        // Create child namespace
        let mut create_ns_req = CreateNamespaceRequest::new();
        create_ns_req.id = Some(vec!["test_ns".to_string()]);
        namespace.create_namespace(create_ns_req).await.unwrap();

        // Create table
        let schema = create_test_schema();
        let ipc_data = create_test_ipc_data(&schema);
        let mut create_table_req = CreateTableRequest::new();
        create_table_req.id = Some(vec!["test_ns".to_string(), "table1".to_string()]);
        namespace
            .create_table(create_table_req, bytes::Bytes::from(ipc_data))
            .await
            .unwrap();

        // Drop table
        let mut drop_req = DropTableRequest::new();
        drop_req.id = Some(vec!["test_ns".to_string(), "table1".to_string()]);
        let result = namespace.drop_table(drop_req).await;
        assert!(result.is_ok(), "Failed to drop table in child namespace");

        // Verify table no longer exists
        let mut exists_req = TableExistsRequest::new();
        exists_req.id = Some(vec!["test_ns".to_string(), "table1".to_string()]);
        let result = namespace.table_exists(exists_req).await;
        assert!(result.is_err());
    }

    #[tokio::test]
    async fn test_deeply_nested_namespace() {
        let (namespace, _temp_dir) = create_test_namespace().await;

        // Create deeply nested namespace hierarchy
        let mut create_req = CreateNamespaceRequest::new();
        create_req.id = Some(vec!["level1".to_string()]);
        namespace.create_namespace(create_req).await.unwrap();

        let mut create_req = CreateNamespaceRequest::new();
        create_req.id = Some(vec!["level1".to_string(), "level2".to_string()]);
        namespace.create_namespace(create_req).await.unwrap();

        let mut create_req = CreateNamespaceRequest::new();
        create_req.id = Some(vec![
            "level1".to_string(),
            "level2".to_string(),
            "level3".to_string(),
        ]);
        namespace.create_namespace(create_req).await.unwrap();

        // Create table in deeply nested namespace
        let schema = create_test_schema();
        let ipc_data = create_test_ipc_data(&schema);
        let mut create_table_req = CreateTableRequest::new();
        create_table_req.id = Some(vec![
            "level1".to_string(),
            "level2".to_string(),
            "level3".to_string(),
            "table1".to_string(),
        ]);
        let result = namespace
            .create_table(create_table_req, bytes::Bytes::from(ipc_data))
            .await;
        assert!(
            result.is_ok(),
            "Failed to create table in deeply nested namespace"
        );

        // Verify table exists
        let mut exists_req = TableExistsRequest::new();
        exists_req.id = Some(vec![
            "level1".to_string(),
            "level2".to_string(),
            "level3".to_string(),
            "table1".to_string(),
        ]);
        let result = namespace.table_exists(exists_req).await;
        assert!(result.is_ok());
    }

    #[tokio::test]
    async fn test_namespace_with_properties() {
        let (namespace, _temp_dir) = create_test_namespace().await;

        // Create namespace with properties
        let mut properties = HashMap::new();
        properties.insert("owner".to_string(), "test_user".to_string());
        properties.insert("description".to_string(), "Test namespace".to_string());

        let mut create_req = CreateNamespaceRequest::new();
        create_req.id = Some(vec!["test_ns".to_string()]);
        create_req.properties = Some(properties.clone());
        namespace.create_namespace(create_req).await.unwrap();

        // Describe namespace and verify properties
        let describe_req = DescribeNamespaceRequest {
            id: Some(vec!["test_ns".to_string()]),
            ..Default::default()
        };
        let result = namespace.describe_namespace(describe_req).await;
        assert!(result.is_ok());
        let response = result.unwrap();
        assert!(response.properties.is_some());
        let props = response.properties.unwrap();
        assert_eq!(props.get("owner"), Some(&"test_user".to_string()));
        assert_eq!(
            props.get("description"),
            Some(&"Test namespace".to_string())
        );
    }

    #[tokio::test]
    async fn test_cannot_drop_namespace_with_tables() {
        let (namespace, _temp_dir) = create_test_namespace().await;

        // Create namespace
        let mut create_ns_req = CreateNamespaceRequest::new();
        create_ns_req.id = Some(vec!["test_ns".to_string()]);
        namespace.create_namespace(create_ns_req).await.unwrap();

        // Create table in namespace
        let schema = create_test_schema();
        let ipc_data = create_test_ipc_data(&schema);
        let mut create_table_req = CreateTableRequest::new();
        create_table_req.id = Some(vec!["test_ns".to_string(), "table1".to_string()]);
        namespace
            .create_table(create_table_req, bytes::Bytes::from(ipc_data))
            .await
            .unwrap();

        // Try to drop namespace - should fail
        let mut drop_req = DropNamespaceRequest::new();
        drop_req.id = Some(vec!["test_ns".to_string()]);
        let result = namespace.drop_namespace(drop_req).await;
        assert!(
            result.is_err(),
            "Should not be able to drop namespace with tables"
        );
    }

    #[tokio::test]
    async fn test_isolation_between_namespaces() {
        let (namespace, _temp_dir) = create_test_namespace().await;

        // Create two namespaces
        let mut create_req = CreateNamespaceRequest::new();
        create_req.id = Some(vec!["ns1".to_string()]);
        namespace.create_namespace(create_req).await.unwrap();

        let mut create_req = CreateNamespaceRequest::new();
        create_req.id = Some(vec!["ns2".to_string()]);
        namespace.create_namespace(create_req).await.unwrap();

        // Create table with same name in both namespaces
        let schema = create_test_schema();
        let ipc_data = create_test_ipc_data(&schema);

        let mut create_table_req = CreateTableRequest::new();
        create_table_req.id = Some(vec!["ns1".to_string(), "table1".to_string()]);
        namespace
            .create_table(create_table_req, bytes::Bytes::from(ipc_data.clone()))
            .await
            .unwrap();

        let mut create_table_req = CreateTableRequest::new();
        create_table_req.id = Some(vec!["ns2".to_string(), "table1".to_string()]);
        namespace
            .create_table(create_table_req, bytes::Bytes::from(ipc_data))
            .await
            .unwrap();

        // List tables in each namespace
        let list_req = ListTablesRequest {
            id: Some(vec!["ns1".to_string()]),
            page_token: None,
            limit: None,
            ..Default::default()
        };
        let result = namespace.list_tables(list_req).await.unwrap();
        assert_eq!(result.tables.len(), 1);
        assert_eq!(result.tables[0], "table1");

        let list_req = ListTablesRequest {
            id: Some(vec!["ns2".to_string()]),
            page_token: None,
            limit: None,
            ..Default::default()
        };
        let result = namespace.list_tables(list_req).await.unwrap();
        assert_eq!(result.tables.len(), 1);
        assert_eq!(result.tables[0], "table1");

        // Drop table in ns1 shouldn't affect ns2
        let mut drop_req = DropTableRequest::new();
        drop_req.id = Some(vec!["ns1".to_string(), "table1".to_string()]);
        namespace.drop_table(drop_req).await.unwrap();

        // Verify ns1 table is gone but ns2 table still exists
        let mut exists_req = TableExistsRequest::new();
        exists_req.id = Some(vec!["ns1".to_string(), "table1".to_string()]);
        assert!(namespace.table_exists(exists_req).await.is_err());

        let mut exists_req = TableExistsRequest::new();
        exists_req.id = Some(vec!["ns2".to_string(), "table1".to_string()]);
        assert!(namespace.table_exists(exists_req).await.is_ok());
    }

    #[tokio::test]
    async fn test_migrate_directory_tables() {
        let temp_dir = TempStdDir::default();
        let temp_path = temp_dir.to_str().unwrap();

        // Step 1: Create tables in directory-only mode
        let dir_only_ns = DirectoryNamespaceBuilder::new(temp_path)
            .manifest_enabled(false)
            .dir_listing_enabled(true)
            .build()
            .await
            .unwrap();

        // Create some tables
        let schema = create_test_schema();
        let ipc_data = create_test_ipc_data(&schema);

        for i in 1..=3 {
            let mut create_req = CreateTableRequest::new();
            create_req.id = Some(vec![format!("table{}", i)]);
            dir_only_ns
                .create_table(create_req, bytes::Bytes::from(ipc_data.clone()))
                .await
                .unwrap();
        }

        drop(dir_only_ns);

        // Step 2: Create namespace with dual mode (manifest + directory listing)
        let dual_mode_ns = DirectoryNamespaceBuilder::new(temp_path)
            .manifest_enabled(true)
            .dir_listing_enabled(true)
            .build()
            .await
            .unwrap();

        // Before migration, tables should be visible (via directory listing fallback)
        let mut list_req = ListTablesRequest::new();
        list_req.id = Some(vec![]);
        let tables = dual_mode_ns.list_tables(list_req).await.unwrap().tables;
        assert_eq!(tables.len(), 3);

        // Run migration
        let migrated_count = dual_mode_ns.migrate().await.unwrap();
        assert_eq!(migrated_count, 3, "Should migrate all 3 tables");

        // Verify tables are now in manifest
        let mut list_req = ListTablesRequest::new();
        list_req.id = Some(vec![]);
        let tables = dual_mode_ns.list_tables(list_req).await.unwrap().tables;
        assert_eq!(tables.len(), 3);

        // Run migration again - should be idempotent
        let migrated_count = dual_mode_ns.migrate().await.unwrap();
        assert_eq!(
            migrated_count, 0,
            "Should not migrate already-migrated tables"
        );

        drop(dual_mode_ns);

        // Step 3: Create namespace with manifest-only mode
        let manifest_only_ns = DirectoryNamespaceBuilder::new(temp_path)
            .manifest_enabled(true)
            .dir_listing_enabled(false)
            .build()
            .await
            .unwrap();

        // Tables should still be accessible (now from manifest only)
        let mut list_req = ListTablesRequest::new();
        list_req.id = Some(vec![]);
        let tables = manifest_only_ns.list_tables(list_req).await.unwrap().tables;
        assert_eq!(tables.len(), 3);
        assert!(tables.contains(&"table1".to_string()));
        assert!(tables.contains(&"table2".to_string()));
        assert!(tables.contains(&"table3".to_string()));
    }

    #[tokio::test]
    async fn test_migrate_without_manifest() {
        let temp_dir = TempStdDir::default();
        let temp_path = temp_dir.to_str().unwrap();

        // Create namespace without manifest
        let namespace = DirectoryNamespaceBuilder::new(temp_path)
            .manifest_enabled(false)
            .dir_listing_enabled(true)
            .build()
            .await
            .unwrap();

        // migrate() should return 0 when manifest is not enabled
        let migrated_count = namespace.migrate().await.unwrap();
        assert_eq!(migrated_count, 0);
    }

    #[tokio::test]
    async fn test_register_table() {
        use lance_namespace::models::{RegisterTableRequest, TableExistsRequest};

        let temp_dir = TempStdDir::default();
        let temp_path = temp_dir.to_str().unwrap();

        let namespace = DirectoryNamespaceBuilder::new(temp_path)
            .build()
            .await
            .unwrap();

        // Create a physical table first using lance directly
        let schema = create_test_schema();
        let ipc_data = create_test_ipc_data(&schema);

        let table_uri = format!("{}/external_table.lance", temp_path);
        let cursor = Cursor::new(ipc_data);
        let stream_reader = StreamReader::try_new(cursor, None).unwrap();
        let batches: Vec<_> = stream_reader
            .collect::<std::result::Result<Vec<_>, _>>()
            .unwrap();
        let schema = batches[0].schema();
        let batch_results: Vec<_> = batches.into_iter().map(Ok).collect();
        let reader = RecordBatchIterator::new(batch_results, schema);
        Dataset::write(Box::new(reader), &table_uri, None)
            .await
            .unwrap();

        // Register the table
        let mut register_req = RegisterTableRequest::new("external_table.lance".to_string());
        register_req.id = Some(vec!["registered_table".to_string()]);

        let response = namespace.register_table(register_req).await.unwrap();
        assert_eq!(response.location, Some("external_table.lance".to_string()));

        // Verify table exists in namespace
        let mut exists_req = TableExistsRequest::new();
        exists_req.id = Some(vec!["registered_table".to_string()]);
        assert!(namespace.table_exists(exists_req).await.is_ok());

        // Verify we can list the table
        let mut list_req = ListTablesRequest::new();
        list_req.id = Some(vec![]);
        let tables = namespace.list_tables(list_req).await.unwrap();
        assert!(tables.tables.contains(&"registered_table".to_string()));
    }

    #[tokio::test]
    async fn test_register_table_duplicate_fails() {
        use lance_namespace::models::RegisterTableRequest;

        let temp_dir = TempStdDir::default();
        let temp_path = temp_dir.to_str().unwrap();

        let namespace = DirectoryNamespaceBuilder::new(temp_path)
            .build()
            .await
            .unwrap();

        // Register a table
        let mut register_req = RegisterTableRequest::new("test_table.lance".to_string());
        register_req.id = Some(vec!["test_table".to_string()]);

        namespace
            .register_table(register_req.clone())
            .await
            .unwrap();

        // Try to register again - should fail
        let result = namespace.register_table(register_req).await;
        assert!(result.is_err());
        assert!(result.unwrap_err().to_string().contains("already exists"));
    }

    #[tokio::test]
    async fn test_deregister_table() {
        use lance_namespace::models::{DeregisterTableRequest, TableExistsRequest};

        let temp_dir = TempStdDir::default();
        let temp_path = temp_dir.to_str().unwrap();

        // Create namespace with manifest-only mode (no directory listing fallback)
        // This ensures deregistered tables are truly invisible
        let namespace = DirectoryNamespaceBuilder::new(temp_path)
            .manifest_enabled(true)
            .dir_listing_enabled(false)
            .build()
            .await
            .unwrap();

        // Create a table
        let schema = create_test_schema();
        let ipc_data = create_test_ipc_data(&schema);

        let mut create_req = CreateTableRequest::new();
        create_req.id = Some(vec!["test_table".to_string()]);
        namespace
            .create_table(create_req, bytes::Bytes::from(ipc_data))
            .await
            .unwrap();

        // Verify table exists
        let mut exists_req = TableExistsRequest::new();
        exists_req.id = Some(vec!["test_table".to_string()]);
        assert!(namespace.table_exists(exists_req.clone()).await.is_ok());

        // Deregister the table
        let mut deregister_req = DeregisterTableRequest::new();
        deregister_req.id = Some(vec!["test_table".to_string()]);
        let response = namespace.deregister_table(deregister_req).await.unwrap();

        // Should return location and id
        assert!(
            response.location.is_some(),
            "Deregister should return location"
        );
        let location = response.location.as_ref().unwrap();
        // Location should be a proper file:// URI with the temp path
        // Use uri_to_url to normalize the temp path to a URL for comparison
        let expected_url = lance_io::object_store::uri_to_url(temp_path)
            .expect("Failed to convert temp path to URL");
        let expected_prefix = expected_url.to_string();
        assert!(
            location.starts_with(&expected_prefix),
            "Location should start with '{}', got: {}",
            expected_prefix,
            location
        );
        assert!(
            location.contains("test_table"),
            "Location should contain table name: {}",
            location
        );
        assert_eq!(response.id, Some(vec!["test_table".to_string()]));

        // Verify table no longer exists in namespace (removed from manifest)
        assert!(namespace.table_exists(exists_req).await.is_err());

        // Verify physical data still exists at the returned location
        let dataset = Dataset::open(location).await;
        assert!(
            dataset.is_ok(),
            "Physical table data should still exist at {}",
            location
        );
    }

    #[tokio::test]
    async fn test_deregister_table_in_child_namespace() {
        use lance_namespace::models::{
            CreateNamespaceRequest, DeregisterTableRequest, TableExistsRequest,
        };

        let temp_dir = TempStdDir::default();
        let temp_path = temp_dir.to_str().unwrap();

        let namespace = DirectoryNamespaceBuilder::new(temp_path)
            .build()
            .await
            .unwrap();

        // Create child namespace
        let mut create_ns_req = CreateNamespaceRequest::new();
        create_ns_req.id = Some(vec!["test_ns".to_string()]);
        namespace.create_namespace(create_ns_req).await.unwrap();

        // Create a table in the child namespace
        let schema = create_test_schema();
        let ipc_data = create_test_ipc_data(&schema);

        let mut create_req = CreateTableRequest::new();
        create_req.id = Some(vec!["test_ns".to_string(), "test_table".to_string()]);
        namespace
            .create_table(create_req, bytes::Bytes::from(ipc_data))
            .await
            .unwrap();

        // Deregister the table
        let mut deregister_req = DeregisterTableRequest::new();
        deregister_req.id = Some(vec!["test_ns".to_string(), "test_table".to_string()]);
        let response = namespace.deregister_table(deregister_req).await.unwrap();

        // Should return location and id in child namespace
        assert!(
            response.location.is_some(),
            "Deregister should return location"
        );
        let location = response.location.as_ref().unwrap();
        // Location should be a proper file:// URI with the temp path
        // Use uri_to_url to normalize the temp path to a URL for comparison
        let expected_url = lance_io::object_store::uri_to_url(temp_path)
            .expect("Failed to convert temp path to URL");
        let expected_prefix = expected_url.to_string();
        assert!(
            location.starts_with(&expected_prefix),
            "Location should start with '{}', got: {}",
            expected_prefix,
            location
        );
        assert!(
            location.contains("test_ns") && location.contains("test_table"),
            "Location should contain namespace and table name: {}",
            location
        );
        assert_eq!(
            response.id,
            Some(vec!["test_ns".to_string(), "test_table".to_string()])
        );

        // Verify table no longer exists
        let mut exists_req = TableExistsRequest::new();
        exists_req.id = Some(vec!["test_ns".to_string(), "test_table".to_string()]);
        assert!(namespace.table_exists(exists_req).await.is_err());
    }

    #[tokio::test]
    async fn test_register_without_manifest_fails() {
        use lance_namespace::models::RegisterTableRequest;

        let temp_dir = TempStdDir::default();
        let temp_path = temp_dir.to_str().unwrap();

        // Create namespace without manifest
        let namespace = DirectoryNamespaceBuilder::new(temp_path)
            .manifest_enabled(false)
            .build()
            .await
            .unwrap();

        // Try to register - should fail (register requires manifest)
        let mut register_req = RegisterTableRequest::new("test_table.lance".to_string());
        register_req.id = Some(vec!["test_table".to_string()]);
        let result = namespace.register_table(register_req).await;
        assert!(result.is_err());
        assert!(
            result
                .unwrap_err()
                .to_string()
                .contains("manifest mode is enabled")
        );

        // Note: deregister_table now works in V1 mode via .lance-deregistered marker files
        // See test_deregister_table_v1_mode for that test case
    }

    #[tokio::test]
    async fn test_register_table_rejects_absolute_uri() {
        use lance_namespace::models::RegisterTableRequest;

        let temp_dir = TempStdDir::default();
        let temp_path = temp_dir.to_str().unwrap();

        let namespace = DirectoryNamespaceBuilder::new(temp_path)
            .build()
            .await
            .unwrap();

        // Try to register with absolute URI - should fail
        let mut register_req = RegisterTableRequest::new("s3://bucket/table.lance".to_string());
        register_req.id = Some(vec!["test_table".to_string()]);
        let result = namespace.register_table(register_req).await;
        assert!(result.is_err());
        let err_msg = result.unwrap_err().to_string();
        assert!(err_msg.contains("Absolute URIs are not allowed"));
    }

    #[tokio::test]
    async fn test_register_table_rejects_absolute_path() {
        use lance_namespace::models::RegisterTableRequest;

        let temp_dir = TempStdDir::default();
        let temp_path = temp_dir.to_str().unwrap();

        let namespace = DirectoryNamespaceBuilder::new(temp_path)
            .build()
            .await
            .unwrap();

        // Try to register with absolute path - should fail
        let mut register_req = RegisterTableRequest::new("/tmp/table.lance".to_string());
        register_req.id = Some(vec!["test_table".to_string()]);
        let result = namespace.register_table(register_req).await;
        assert!(result.is_err());
        let err_msg = result.unwrap_err().to_string();
        assert!(err_msg.contains("Absolute paths are not allowed"));
    }

    #[tokio::test]
    async fn test_register_table_rejects_path_traversal() {
        use lance_namespace::models::RegisterTableRequest;

        let temp_dir = TempStdDir::default();
        let temp_path = temp_dir.to_str().unwrap();

        let namespace = DirectoryNamespaceBuilder::new(temp_path)
            .build()
            .await
            .unwrap();

        // Try to register with path traversal - should fail
        let mut register_req = RegisterTableRequest::new("../outside/table.lance".to_string());
        register_req.id = Some(vec!["test_table".to_string()]);
        let result = namespace.register_table(register_req).await;
        assert!(result.is_err());
        let err_msg = result.unwrap_err().to_string();
        assert!(err_msg.contains("Path traversal is not allowed"));
    }

    #[tokio::test]
    async fn test_namespace_write() {
        use arrow::array::Int32Array;
        use arrow::datatypes::{DataType, Field as ArrowField, Schema as ArrowSchema};
        use arrow::record_batch::{RecordBatch, RecordBatchIterator};
        use lance::dataset::{Dataset, WriteMode, WriteParams};
        use lance_namespace::LanceNamespace;

        let (namespace, _temp_dir) = create_test_namespace().await;
        let namespace = Arc::new(namespace) as Arc<dyn LanceNamespace>;

        // Use child namespace instead of root
        let table_id = vec!["test_ns".to_string(), "test_table".to_string()];
        let schema = Arc::new(ArrowSchema::new(vec![
            ArrowField::new("a", DataType::Int32, false),
            ArrowField::new("b", DataType::Int32, false),
        ]));

        // Test 1: CREATE mode
        let data1 = RecordBatch::try_new(
            schema.clone(),
            vec![
                Arc::new(Int32Array::from(vec![1, 2, 3])),
                Arc::new(Int32Array::from(vec![10, 20, 30])),
            ],
        )
        .unwrap();

        let reader1 = RecordBatchIterator::new(vec![data1].into_iter().map(Ok), schema.clone());
        let dataset =
            Dataset::write_into_namespace(reader1, namespace.clone(), table_id.clone(), None)
                .await
                .unwrap();

        assert_eq!(dataset.count_rows(None).await.unwrap(), 3);
        assert_eq!(dataset.version().version, 1);

        // Test 2: APPEND mode
        let data2 = RecordBatch::try_new(
            schema.clone(),
            vec![
                Arc::new(Int32Array::from(vec![4, 5])),
                Arc::new(Int32Array::from(vec![40, 50])),
            ],
        )
        .unwrap();

        let params_append = WriteParams {
            mode: WriteMode::Append,
            ..Default::default()
        };

        let reader2 = RecordBatchIterator::new(vec![data2].into_iter().map(Ok), schema.clone());
        let dataset = Dataset::write_into_namespace(
            reader2,
            namespace.clone(),
            table_id.clone(),
            Some(params_append),
        )
        .await
        .unwrap();

        assert_eq!(dataset.count_rows(None).await.unwrap(), 5);
        assert_eq!(dataset.version().version, 2);

        // Test 3: OVERWRITE mode
        let data3 = RecordBatch::try_new(
            schema.clone(),
            vec![
                Arc::new(Int32Array::from(vec![100, 200])),
                Arc::new(Int32Array::from(vec![1000, 2000])),
            ],
        )
        .unwrap();

        let params_overwrite = WriteParams {
            mode: WriteMode::Overwrite,
            ..Default::default()
        };

        let reader3 = RecordBatchIterator::new(vec![data3].into_iter().map(Ok), schema.clone());
        let dataset = Dataset::write_into_namespace(
            reader3,
            namespace.clone(),
            table_id.clone(),
            Some(params_overwrite),
        )
        .await
        .unwrap();

        assert_eq!(dataset.count_rows(None).await.unwrap(), 2);
        assert_eq!(dataset.version().version, 3);

        // Verify old data was replaced
        let result = dataset.scan().try_into_batch().await.unwrap();
        let a_col = result
            .column_by_name("a")
            .unwrap()
            .as_any()
            .downcast_ref::<Int32Array>()
            .unwrap();
        assert_eq!(a_col.values(), &[100, 200]);
    }

    // ============================================================
    // Tests for declare_table
    // ============================================================

    #[tokio::test]
    async fn test_declare_table_v1_mode() {
        use lance_namespace::models::{
            DeclareTableRequest, DescribeTableRequest, TableExistsRequest,
        };

        let temp_dir = TempStdDir::default();
        let temp_path = temp_dir.to_str().unwrap();

        // Create namespace in V1 mode (no manifest)
        let namespace = DirectoryNamespaceBuilder::new(temp_path)
            .manifest_enabled(false)
            .build()
            .await
            .unwrap();

        // Declare a table
        let mut declare_req = DeclareTableRequest::new();
        declare_req.id = Some(vec!["test_table".to_string()]);
        let response = namespace.declare_table(declare_req).await.unwrap();

        // Should return location
        assert!(response.location.is_some());
        let location = response.location.as_ref().unwrap();
        assert!(location.ends_with("test_table.lance"));

        // Table should exist (via reserved file)
        let mut exists_req = TableExistsRequest::new();
        exists_req.id = Some(vec!["test_table".to_string()]);
        assert!(namespace.table_exists(exists_req).await.is_ok());

        // Describe should work but return no version/schema (not written yet)
        let mut describe_req = DescribeTableRequest::new();
        describe_req.id = Some(vec!["test_table".to_string()]);
        let describe_response = namespace.describe_table(describe_req).await.unwrap();
        assert!(describe_response.location.is_some());
        assert!(describe_response.version.is_none()); // Not written yet
        assert!(describe_response.schema.is_none()); // Not written yet
    }

    #[tokio::test]
    async fn test_declare_table_with_manifest() {
        use lance_namespace::models::{DeclareTableRequest, TableExistsRequest};

        let temp_dir = TempStdDir::default();
        let temp_path = temp_dir.to_str().unwrap();

        // Create namespace with manifest
        let namespace = DirectoryNamespaceBuilder::new(temp_path)
            .manifest_enabled(true)
            .dir_listing_enabled(false)
            .build()
            .await
            .unwrap();

        // Declare a table
        let mut declare_req = DeclareTableRequest::new();
        declare_req.id = Some(vec!["test_table".to_string()]);
        let response = namespace.declare_table(declare_req).await.unwrap();

        // Should return location
        assert!(response.location.is_some());

        // Table should exist in manifest
        let mut exists_req = TableExistsRequest::new();
        exists_req.id = Some(vec!["test_table".to_string()]);
        assert!(namespace.table_exists(exists_req).await.is_ok());
    }

    #[tokio::test]
    async fn test_declare_table_when_table_exists() {
        use lance_namespace::models::DeclareTableRequest;

        let temp_dir = TempStdDir::default();
        let temp_path = temp_dir.to_str().unwrap();

        let namespace = DirectoryNamespaceBuilder::new(temp_path)
            .manifest_enabled(false)
            .build()
            .await
            .unwrap();

        // First create a table with actual data
        let schema = create_test_schema();
        let ipc_data = create_test_ipc_data(&schema);
        let mut create_req = CreateTableRequest::new();
        create_req.id = Some(vec!["test_table".to_string()]);
        namespace
            .create_table(create_req, bytes::Bytes::from(ipc_data))
            .await
            .unwrap();

        // Try to declare the same table - should fail because it already has data
        let mut declare_req = DeclareTableRequest::new();
        declare_req.id = Some(vec!["test_table".to_string()]);
        let result = namespace.declare_table(declare_req).await;
        assert!(result.is_err());
    }

    // ============================================================
    // Tests for deregister_table in V1 mode
    // ============================================================

    #[tokio::test]
    async fn test_deregister_table_v1_mode() {
        use lance_namespace::models::{DeregisterTableRequest, TableExistsRequest};

        let temp_dir = TempStdDir::default();
        let temp_path = temp_dir.to_str().unwrap();

        // Create namespace in V1 mode (no manifest, with dir listing)
        let namespace = DirectoryNamespaceBuilder::new(temp_path)
            .manifest_enabled(false)
            .dir_listing_enabled(true)
            .build()
            .await
            .unwrap();

        // Create a table with data
        let schema = create_test_schema();
        let ipc_data = create_test_ipc_data(&schema);
        let mut create_req = CreateTableRequest::new();
        create_req.id = Some(vec!["test_table".to_string()]);
        namespace
            .create_table(create_req, bytes::Bytes::from(ipc_data))
            .await
            .unwrap();

        // Verify table exists
        let mut exists_req = TableExistsRequest::new();
        exists_req.id = Some(vec!["test_table".to_string()]);
        assert!(namespace.table_exists(exists_req.clone()).await.is_ok());

        // Deregister the table
        let mut deregister_req = DeregisterTableRequest::new();
        deregister_req.id = Some(vec!["test_table".to_string()]);
        let response = namespace.deregister_table(deregister_req).await.unwrap();

        // Should return location
        assert!(response.location.is_some());
        let location = response.location.as_ref().unwrap();
        assert!(location.contains("test_table"));

        // Table should no longer exist (deregistered)
        let result = namespace.table_exists(exists_req).await;
        assert!(result.is_err());
        assert!(result.unwrap_err().to_string().contains("deregistered"));

        // Physical data should still exist
        let dataset = Dataset::open(location).await;
        assert!(dataset.is_ok(), "Physical table data should still exist");
    }

    #[tokio::test]
    async fn test_deregister_table_v1_already_deregistered() {
        use lance_namespace::models::DeregisterTableRequest;

        let temp_dir = TempStdDir::default();
        let temp_path = temp_dir.to_str().unwrap();

        let namespace = DirectoryNamespaceBuilder::new(temp_path)
            .manifest_enabled(false)
            .dir_listing_enabled(true)
            .build()
            .await
            .unwrap();

        // Create a table
        let schema = create_test_schema();
        let ipc_data = create_test_ipc_data(&schema);
        let mut create_req = CreateTableRequest::new();
        create_req.id = Some(vec!["test_table".to_string()]);
        namespace
            .create_table(create_req, bytes::Bytes::from(ipc_data))
            .await
            .unwrap();

        // Deregister once
        let mut deregister_req = DeregisterTableRequest::new();
        deregister_req.id = Some(vec!["test_table".to_string()]);
        namespace
            .deregister_table(deregister_req.clone())
            .await
            .unwrap();

        // Try to deregister again - should fail
        let result = namespace.deregister_table(deregister_req).await;
        assert!(result.is_err());
        assert!(
            result
                .unwrap_err()
                .to_string()
                .contains("already deregistered")
        );
    }

    // ============================================================
    // Tests for list_tables skipping deregistered tables
    // ============================================================

    #[tokio::test]
    async fn test_list_tables_skips_deregistered_v1() {
        use lance_namespace::models::DeregisterTableRequest;

        let temp_dir = TempStdDir::default();
        let temp_path = temp_dir.to_str().unwrap();

        let namespace = DirectoryNamespaceBuilder::new(temp_path)
            .manifest_enabled(false)
            .dir_listing_enabled(true)
            .build()
            .await
            .unwrap();

        // Create two tables
        let schema = create_test_schema();
        let ipc_data = create_test_ipc_data(&schema);

        let mut create_req1 = CreateTableRequest::new();
        create_req1.id = Some(vec!["table1".to_string()]);
        namespace
            .create_table(create_req1, bytes::Bytes::from(ipc_data.clone()))
            .await
            .unwrap();

        let mut create_req2 = CreateTableRequest::new();
        create_req2.id = Some(vec!["table2".to_string()]);
        namespace
            .create_table(create_req2, bytes::Bytes::from(ipc_data))
            .await
            .unwrap();

        // List tables - should see both (root namespace = empty vec)
        let mut list_req = ListTablesRequest::new();
        list_req.id = Some(vec![]);
        let list_response = namespace.list_tables(list_req.clone()).await.unwrap();
        assert_eq!(list_response.tables.len(), 2);

        // Deregister table1
        let mut deregister_req = DeregisterTableRequest::new();
        deregister_req.id = Some(vec!["table1".to_string()]);
        namespace.deregister_table(deregister_req).await.unwrap();

        // List tables - should only see table2
        let list_response = namespace.list_tables(list_req).await.unwrap();
        assert_eq!(list_response.tables.len(), 1);
        assert!(list_response.tables.contains(&"table2".to_string()));
        assert!(!list_response.tables.contains(&"table1".to_string()));
    }

    // ============================================================
    // Tests for describe_table and table_exists with deregistered tables
    // ============================================================

    #[tokio::test]
    async fn test_describe_table_fails_for_deregistered_v1() {
        use lance_namespace::models::{DeregisterTableRequest, DescribeTableRequest};

        let temp_dir = TempStdDir::default();
        let temp_path = temp_dir.to_str().unwrap();

        let namespace = DirectoryNamespaceBuilder::new(temp_path)
            .manifest_enabled(false)
            .dir_listing_enabled(true)
            .build()
            .await
            .unwrap();

        // Create a table
        let schema = create_test_schema();
        let ipc_data = create_test_ipc_data(&schema);
        let mut create_req = CreateTableRequest::new();
        create_req.id = Some(vec!["test_table".to_string()]);
        namespace
            .create_table(create_req, bytes::Bytes::from(ipc_data))
            .await
            .unwrap();

        // Describe should work before deregistration
        let mut describe_req = DescribeTableRequest::new();
        describe_req.id = Some(vec!["test_table".to_string()]);
        assert!(namespace.describe_table(describe_req.clone()).await.is_ok());

        // Deregister
        let mut deregister_req = DeregisterTableRequest::new();
        deregister_req.id = Some(vec!["test_table".to_string()]);
        namespace.deregister_table(deregister_req).await.unwrap();

        // Describe should fail after deregistration
        let result = namespace.describe_table(describe_req).await;
        assert!(result.is_err());
        assert!(result.unwrap_err().to_string().contains("deregistered"));
    }

    #[tokio::test]
    async fn test_table_exists_fails_for_deregistered_v1() {
        use lance_namespace::models::{DeregisterTableRequest, TableExistsRequest};

        let temp_dir = TempStdDir::default();
        let temp_path = temp_dir.to_str().unwrap();

        let namespace = DirectoryNamespaceBuilder::new(temp_path)
            .manifest_enabled(false)
            .dir_listing_enabled(true)
            .build()
            .await
            .unwrap();

        // Create a table
        let schema = create_test_schema();
        let ipc_data = create_test_ipc_data(&schema);
        let mut create_req = CreateTableRequest::new();
        create_req.id = Some(vec!["test_table".to_string()]);
        namespace
            .create_table(create_req, bytes::Bytes::from(ipc_data))
            .await
            .unwrap();

        // Table exists should work before deregistration
        let mut exists_req = TableExistsRequest::new();
        exists_req.id = Some(vec!["test_table".to_string()]);
        assert!(namespace.table_exists(exists_req.clone()).await.is_ok());

        // Deregister
        let mut deregister_req = DeregisterTableRequest::new();
        deregister_req.id = Some(vec!["test_table".to_string()]);
        namespace.deregister_table(deregister_req).await.unwrap();

        // Table exists should fail after deregistration
        let result = namespace.table_exists(exists_req).await;
        assert!(result.is_err());
        assert!(result.unwrap_err().to_string().contains("deregistered"));
    }

    #[tokio::test]
    async fn test_atomic_table_status_check() {
        // This test verifies that the TableStatus check is atomic
        // by ensuring a single directory listing is used

        let temp_dir = TempStdDir::default();
        let temp_path = temp_dir.to_str().unwrap();

        let namespace = DirectoryNamespaceBuilder::new(temp_path)
            .manifest_enabled(false)
            .dir_listing_enabled(true)
            .build()
            .await
            .unwrap();

        // Create a table
        let schema = create_test_schema();
        let ipc_data = create_test_ipc_data(&schema);
        let mut create_req = CreateTableRequest::new();
        create_req.id = Some(vec!["test_table".to_string()]);
        namespace
            .create_table(create_req, bytes::Bytes::from(ipc_data))
            .await
            .unwrap();

        // Table status should show exists=true, is_deregistered=false
        let status = namespace.check_table_status("test_table").await;
        assert!(status.exists);
        assert!(!status.is_deregistered);
        assert!(!status.has_reserved_file);
    }

    #[tokio::test]
    async fn test_table_version_tracking_enabled_managed_versioning() {
        use lance_namespace::models::DescribeTableRequest;

        let temp_dir = TempStdDir::default();
        let temp_path = temp_dir.to_str().unwrap();

        // Create namespace with table_version_tracking_enabled=true
        let namespace = DirectoryNamespaceBuilder::new(temp_path)
            .table_version_tracking_enabled(true)
            .build()
            .await
            .unwrap();

        // Create a table
        let schema = create_test_schema();
        let ipc_data = create_test_ipc_data(&schema);
        let mut create_req = CreateTableRequest::new();
        create_req.id = Some(vec!["test_table".to_string()]);
        namespace
            .create_table(create_req, bytes::Bytes::from(ipc_data))
            .await
            .unwrap();

        // Describe table should return managed_versioning=true
        let mut describe_req = DescribeTableRequest::new();
        describe_req.id = Some(vec!["test_table".to_string()]);
        let describe_resp = namespace.describe_table(describe_req).await.unwrap();

        // managed_versioning should be true
        assert_eq!(
            describe_resp.managed_versioning,
            Some(true),
            "managed_versioning should be true when table_version_tracking_enabled=true"
        );
    }

    #[tokio::test]
    async fn test_table_version_tracking_disabled_no_managed_versioning() {
        use lance_namespace::models::DescribeTableRequest;

        let temp_dir = TempStdDir::default();
        let temp_path = temp_dir.to_str().unwrap();

        // Create namespace with table_version_tracking_enabled=false (default)
        let namespace = DirectoryNamespaceBuilder::new(temp_path)
            .table_version_tracking_enabled(false)
            .build()
            .await
            .unwrap();

        // Create a table
        let schema = create_test_schema();
        let ipc_data = create_test_ipc_data(&schema);
        let mut create_req = CreateTableRequest::new();
        create_req.id = Some(vec!["test_table".to_string()]);
        namespace
            .create_table(create_req, bytes::Bytes::from(ipc_data))
            .await
            .unwrap();

        // Describe table should not have managed_versioning set
        let mut describe_req = DescribeTableRequest::new();
        describe_req.id = Some(vec!["test_table".to_string()]);
        let describe_resp = namespace.describe_table(describe_req).await.unwrap();

        // managed_versioning should be None when table_version_tracking_enabled=false
        assert!(
            describe_resp.managed_versioning.is_none(),
            "managed_versioning should be None when table_version_tracking_enabled=false, got: {:?}",
            describe_resp.managed_versioning
        );
    }

    #[tokio::test]
    #[cfg(not(windows))]
    async fn test_list_table_versions() {
        use arrow::array::{Int32Array, RecordBatchIterator};
        use arrow::datatypes::{DataType, Field, Schema as ArrowSchema};
        use arrow::record_batch::RecordBatch;
        use lance::dataset::{Dataset, WriteMode, WriteParams};
        use lance_namespace::models::{CreateNamespaceRequest, ListTableVersionsRequest};

        let temp_dir = TempStrDir::default();
        let temp_path: &str = &temp_dir;

        let namespace: Arc<dyn LanceNamespace> = Arc::new(
            DirectoryNamespaceBuilder::new(temp_path)
                .table_version_tracking_enabled(true)
                .build()
                .await
                .unwrap(),
        );

        // Create parent namespace first
        let mut create_ns_req = CreateNamespaceRequest::new();
        create_ns_req.id = Some(vec!["workspace".to_string()]);
        namespace.create_namespace(create_ns_req).await.unwrap();

        // Create a table using write_into_namespace (version 1)
        let table_id = vec!["workspace".to_string(), "test_table".to_string()];
        let arrow_schema = Arc::new(ArrowSchema::new(vec![Field::new(
            "id",
            DataType::Int32,
            false,
        )]));
        let batch = RecordBatch::try_new(
            arrow_schema.clone(),
            vec![Arc::new(Int32Array::from(vec![1, 2, 3]))],
        )
        .unwrap();
        let batches = RecordBatchIterator::new(vec![Ok(batch.clone())], arrow_schema.clone());
        let write_params = WriteParams {
            mode: WriteMode::Create,
            ..Default::default()
        };
        let mut dataset = Dataset::write_into_namespace(
            batches,
            namespace.clone(),
            table_id.clone(),
            Some(write_params),
        )
        .await
        .unwrap();

        // Append to create version 2
        let batch2 = RecordBatch::try_new(
            arrow_schema.clone(),
            vec![Arc::new(Int32Array::from(vec![100, 200]))],
        )
        .unwrap();
        let batches = RecordBatchIterator::new(vec![Ok(batch2)], arrow_schema.clone());
        dataset.append(batches, None).await.unwrap();

        // Append to create version 3
        let batch3 = RecordBatch::try_new(
            arrow_schema.clone(),
            vec![Arc::new(Int32Array::from(vec![300, 400]))],
        )
        .unwrap();
        let batches = RecordBatchIterator::new(vec![Ok(batch3)], arrow_schema);
        dataset.append(batches, None).await.unwrap();

        // List versions - should have versions 1, 2, and 3
        let mut list_req = ListTableVersionsRequest::new();
        list_req.id = Some(table_id.clone());
        let list_resp = namespace.list_table_versions(list_req).await.unwrap();

        assert_eq!(
            list_resp.versions.len(),
            3,
            "Should have 3 versions, got: {:?}",
            list_resp.versions
        );

        // Verify each version
        for expected_version in 1..=3 {
            let version = list_resp
                .versions
                .iter()
                .find(|v| v.version == expected_version)
                .unwrap_or_else(|| panic!("Expected version {}", expected_version));

            assert!(
                !version.manifest_path.is_empty(),
                "manifest_path should be set for version {}",
                expected_version
            );
            assert!(
                version.manifest_path.contains(".manifest"),
                "manifest_path should contain .manifest for version {}",
                expected_version
            );
            assert!(
                version.manifest_size.is_some(),
                "manifest_size should be set for version {}",
                expected_version
            );
            assert!(
                version.manifest_size.unwrap() > 0,
                "manifest_size should be > 0 for version {}",
                expected_version
            );
            assert!(
                version.timestamp_millis.is_some(),
                "timestamp_millis should be set for version {}",
                expected_version
            );
        }
    }

    #[tokio::test]
    #[cfg(not(windows))]
    async fn test_describe_table_version() {
        use arrow::array::{Int32Array, RecordBatchIterator};
        use arrow::datatypes::{DataType, Field, Schema as ArrowSchema};
        use arrow::record_batch::RecordBatch;
        use lance::dataset::{Dataset, WriteMode, WriteParams};
        use lance_namespace::models::{CreateNamespaceRequest, DescribeTableVersionRequest};

        let temp_dir = TempStrDir::default();
        let temp_path: &str = &temp_dir;

        let namespace: Arc<dyn LanceNamespace> = Arc::new(
            DirectoryNamespaceBuilder::new(temp_path)
                .table_version_tracking_enabled(true)
                .build()
                .await
                .unwrap(),
        );

        // Create parent namespace first
        let mut create_ns_req = CreateNamespaceRequest::new();
        create_ns_req.id = Some(vec!["workspace".to_string()]);
        namespace.create_namespace(create_ns_req).await.unwrap();

        // Create a table using write_into_namespace (version 1)
        let table_id = vec!["workspace".to_string(), "test_table".to_string()];
        let arrow_schema = Arc::new(ArrowSchema::new(vec![Field::new(
            "id",
            DataType::Int32,
            false,
        )]));
        let batch = RecordBatch::try_new(
            arrow_schema.clone(),
            vec![Arc::new(Int32Array::from(vec![1, 2, 3]))],
        )
        .unwrap();
        let batches = RecordBatchIterator::new(vec![Ok(batch)], arrow_schema.clone());
        let write_params = WriteParams {
            mode: WriteMode::Create,
            ..Default::default()
        };
        let mut dataset = Dataset::write_into_namespace(
            batches,
            namespace.clone(),
            table_id.clone(),
            Some(write_params),
        )
        .await
        .unwrap();

        // Append data to create version 2
        let batch2 = RecordBatch::try_new(
            arrow_schema.clone(),
            vec![Arc::new(Int32Array::from(vec![100, 200]))],
        )
        .unwrap();
        let batches = RecordBatchIterator::new(vec![Ok(batch2)], arrow_schema);
        dataset.append(batches, None).await.unwrap();

        // Describe version 1
        let mut describe_req = DescribeTableVersionRequest::new();
        describe_req.id = Some(table_id.clone());
        describe_req.version = Some(1);
        let describe_resp = namespace
            .describe_table_version(describe_req)
            .await
            .unwrap();

        let version = &describe_resp.version;
        assert_eq!(version.version, 1);
        assert!(version.timestamp_millis.is_some());
        assert!(
            !version.manifest_path.is_empty(),
            "manifest_path should be set"
        );
        assert!(
            version.manifest_path.contains(".manifest"),
            "manifest_path should contain .manifest"
        );
        assert!(
            version.manifest_size.is_some(),
            "manifest_size should be set"
        );
        assert!(
            version.manifest_size.unwrap() > 0,
            "manifest_size should be > 0"
        );

        // Describe version 2
        let mut describe_req = DescribeTableVersionRequest::new();
        describe_req.id = Some(table_id.clone());
        describe_req.version = Some(2);
        let describe_resp = namespace
            .describe_table_version(describe_req)
            .await
            .unwrap();

        let version = &describe_resp.version;
        assert_eq!(version.version, 2);
        assert!(version.timestamp_millis.is_some());
        assert!(
            !version.manifest_path.is_empty(),
            "manifest_path should be set"
        );
        assert!(
            version.manifest_size.is_some(),
            "manifest_size should be set"
        );
        assert!(
            version.manifest_size.unwrap() > 0,
            "manifest_size should be > 0"
        );
    }

    #[tokio::test]
    #[cfg(not(windows))]
    async fn test_describe_table_version_latest() {
        use arrow::array::{Int32Array, RecordBatchIterator};
        use arrow::datatypes::{DataType, Field, Schema as ArrowSchema};
        use arrow::record_batch::RecordBatch;
        use lance::dataset::{Dataset, WriteMode, WriteParams};
        use lance_namespace::models::{CreateNamespaceRequest, DescribeTableVersionRequest};

        let temp_dir = TempStrDir::default();
        let temp_path: &str = &temp_dir;

        let namespace: Arc<dyn LanceNamespace> = Arc::new(
            DirectoryNamespaceBuilder::new(temp_path)
                .table_version_tracking_enabled(true)
                .build()
                .await
                .unwrap(),
        );

        // Create parent namespace first
        let mut create_ns_req = CreateNamespaceRequest::new();
        create_ns_req.id = Some(vec!["workspace".to_string()]);
        namespace.create_namespace(create_ns_req).await.unwrap();

        // Create a table using write_into_namespace (version 1)
        let table_id = vec!["workspace".to_string(), "test_table".to_string()];
        let arrow_schema = Arc::new(ArrowSchema::new(vec![Field::new(
            "id",
            DataType::Int32,
            false,
        )]));
        let batch = RecordBatch::try_new(
            arrow_schema.clone(),
            vec![Arc::new(Int32Array::from(vec![1, 2, 3]))],
        )
        .unwrap();
        let batches = RecordBatchIterator::new(vec![Ok(batch)], arrow_schema.clone());
        let write_params = WriteParams {
            mode: WriteMode::Create,
            ..Default::default()
        };
        let mut dataset = Dataset::write_into_namespace(
            batches,
            namespace.clone(),
            table_id.clone(),
            Some(write_params),
        )
        .await
        .unwrap();

        // Append to create version 2
        let batch2 = RecordBatch::try_new(
            arrow_schema.clone(),
            vec![Arc::new(Int32Array::from(vec![100, 200]))],
        )
        .unwrap();
        let batches = RecordBatchIterator::new(vec![Ok(batch2)], arrow_schema.clone());
        dataset.append(batches, None).await.unwrap();

        // Append to create version 3
        let batch3 = RecordBatch::try_new(
            arrow_schema.clone(),
            vec![Arc::new(Int32Array::from(vec![300, 400]))],
        )
        .unwrap();
        let batches = RecordBatchIterator::new(vec![Ok(batch3)], arrow_schema);
        dataset.append(batches, None).await.unwrap();

        // Describe latest version (no version specified)
        let mut describe_req = DescribeTableVersionRequest::new();
        describe_req.id = Some(table_id.clone());
        describe_req.version = None;
        let describe_resp = namespace
            .describe_table_version(describe_req)
            .await
            .unwrap();

        // Should return version 3 as it's the latest
        assert_eq!(describe_resp.version.version, 3);
    }

    #[tokio::test]
    #[cfg(not(windows))]
    async fn test_create_table_version() {
        use futures::TryStreamExt;
        use lance::dataset::builder::DatasetBuilder;
        use lance_namespace::models::CreateTableVersionRequest;

        let temp_dir = TempStrDir::default();
        let temp_path: &str = &temp_dir;

        let namespace: Arc<dyn LanceNamespace> = Arc::new(
            DirectoryNamespaceBuilder::new(temp_path)
                .table_version_tracking_enabled(true)
                .build()
                .await
                .unwrap(),
        );

        // Create a table
        let schema = create_test_schema();
        let ipc_data = create_test_ipc_data(&schema);
        let mut create_req = CreateTableRequest::new();
        create_req.id = Some(vec!["test_table".to_string()]);
        namespace
            .create_table(create_req, bytes::Bytes::from(ipc_data))
            .await
            .unwrap();

        // Open the dataset using from_namespace to get proper object_store and paths
        let table_id = vec!["test_table".to_string()];
        let dataset = DatasetBuilder::from_namespace(namespace.clone(), table_id.clone())
            .await
            .unwrap()
            .load()
            .await
            .unwrap();

        // Use dataset's object_store to find and copy the manifest
        let versions_path = dataset.versions_dir();
        let manifest_metas: Vec<_> = dataset
            .object_store()
            .inner
            .list(Some(&versions_path))
            .try_collect()
            .await
            .unwrap();

        let manifest_meta = manifest_metas
            .iter()
            .find(|m| {
                m.location
                    .filename()
                    .map(|f| f.ends_with(".manifest"))
                    .unwrap_or(false)
            })
            .expect("No manifest file found");

        // Read the existing manifest data
        let manifest_data = dataset
            .object_store()
            .inner
            .get(&manifest_meta.location)
            .await
            .unwrap()
            .bytes()
            .await
            .unwrap();

        // Write to a staging location using the dataset's object_store
        let staging_path = dataset.versions_dir().child("staging_manifest");
        dataset
            .object_store()
            .inner
            .put(&staging_path, manifest_data.into())
            .await
            .unwrap();

        // Create version 2 from staging manifest
        // Use the same naming scheme as the existing dataset (V2)
        let mut create_version_req = CreateTableVersionRequest::new(2, staging_path.to_string());
        create_version_req.id = Some(table_id.clone());
        create_version_req.naming_scheme = Some("V2".to_string());

        let result = namespace.create_table_version(create_version_req).await;
        assert!(
            result.is_ok(),
            "create_table_version should succeed: {:?}",
            result
        );

        // Verify version 2 was created at the path returned in the response
        let response = result.unwrap();
        let version_info = response
            .version
            .expect("response should contain version info");
        let version_2_path = Path::from(version_info.manifest_path);
        let head_result = dataset.object_store().inner.head(&version_2_path).await;
        assert!(
            head_result.is_ok(),
            "Version 2 manifest should exist at {}",
            version_2_path
        );

        // Verify the staging file has been deleted
        let staging_head_result = dataset.object_store().inner.head(&staging_path).await;
        assert!(
            staging_head_result.is_err(),
            "Staging manifest should have been deleted after create_table_version"
        );
    }

    #[tokio::test]
    #[cfg(not(windows))]
    async fn test_create_table_version_conflict() {
        // create_table_version should fail if the version already exists.
        // Each version always writes to a new file location.
        use futures::TryStreamExt;
        use lance::dataset::builder::DatasetBuilder;
        use lance_namespace::models::CreateTableVersionRequest;

        let temp_dir = TempStrDir::default();
        let temp_path: &str = &temp_dir;

        let namespace: Arc<dyn LanceNamespace> = Arc::new(
            DirectoryNamespaceBuilder::new(temp_path)
                .table_version_tracking_enabled(true)
                .build()
                .await
                .unwrap(),
        );

        // Create a table
        let schema = create_test_schema();
        let ipc_data = create_test_ipc_data(&schema);
        let mut create_req = CreateTableRequest::new();
        create_req.id = Some(vec!["test_table".to_string()]);
        namespace
            .create_table(create_req, bytes::Bytes::from(ipc_data))
            .await
            .unwrap();

        // Open the dataset using from_namespace to get proper object_store and paths
        let table_id = vec!["test_table".to_string()];
        let dataset = DatasetBuilder::from_namespace(namespace.clone(), table_id.clone())
            .await
            .unwrap()
            .load()
            .await
            .unwrap();

        // Use dataset's object_store to find and copy the manifest
        let versions_path = dataset.versions_dir();
        let manifest_metas: Vec<_> = dataset
            .object_store()
            .inner
            .list(Some(&versions_path))
            .try_collect()
            .await
            .unwrap();

        let manifest_meta = manifest_metas
            .iter()
            .find(|m| {
                m.location
                    .filename()
                    .map(|f| f.ends_with(".manifest"))
                    .unwrap_or(false)
            })
            .expect("No manifest file found");

        // Read the existing manifest data
        let manifest_data = dataset
            .object_store()
            .inner
            .get(&manifest_meta.location)
            .await
            .unwrap()
            .bytes()
            .await
            .unwrap();

        // Write to a staging location using the dataset's object_store
        let staging_path = dataset.versions_dir().child("staging_manifest");
        dataset
            .object_store()
            .inner
            .put(&staging_path, manifest_data.into())
            .await
            .unwrap();

        // First create version 2 (should succeed)
        let mut create_version_req = CreateTableVersionRequest::new(2, staging_path.to_string());
        create_version_req.id = Some(table_id.clone());
        create_version_req.naming_scheme = Some("V2".to_string());
        let first_result = namespace.create_table_version(create_version_req).await;
        assert!(
            first_result.is_ok(),
            "First create_table_version for version 2 should succeed: {:?}",
            first_result
        );

        // Get the path from the response for verification
        let version_2_path = Path::from(
            first_result
                .unwrap()
                .version
                .expect("response should contain version info")
                .manifest_path,
        );

        // Create version 2 again (should fail - conflict)
        let mut create_version_req = CreateTableVersionRequest::new(2, staging_path.to_string());
        create_version_req.id = Some(table_id.clone());
        create_version_req.naming_scheme = Some("V2".to_string());

        let result = namespace.create_table_version(create_version_req).await;
        assert!(
            result.is_err(),
            "create_table_version should fail for existing version"
        );

        // Verify version 2 still exists using the dataset's object_store
        let head_result = dataset.object_store().inner.head(&version_2_path).await;
        assert!(
            head_result.is_ok(),
            "Version 2 manifest should still exist at {}",
            version_2_path
        );
    }

    #[tokio::test]
    async fn test_create_table_version_table_not_found() {
        use lance_namespace::models::CreateTableVersionRequest;

        let temp_dir = TempStdDir::default();
        let temp_path = temp_dir.to_str().unwrap();

        let namespace = DirectoryNamespaceBuilder::new(temp_path)
            .table_version_tracking_enabled(true)
            .build()
            .await
            .unwrap();

        // Try to create version for non-existent table
        let mut create_version_req =
            CreateTableVersionRequest::new(1, "/some/staging/path".to_string());
        create_version_req.id = Some(vec!["non_existent_table".to_string()]);

        let result = namespace.create_table_version(create_version_req).await;
        assert!(
            result.is_err(),
            "create_table_version should fail for non-existent table"
        );
        let err_msg = result.unwrap_err().to_string();
        assert!(
            err_msg.contains("does not exist"),
            "Error should mention table does not exist, got: {}",
            err_msg
        );
    }

    /// End-to-end integration test module for table version tracking.
    mod e2e_table_version_tracking {
        use super::*;
        use std::sync::atomic::{AtomicUsize, Ordering};

        /// Tracking wrapper around a namespace that counts method invocations.
        struct TrackingNamespace {
            inner: DirectoryNamespace,
            create_table_version_count: AtomicUsize,
            describe_table_version_count: AtomicUsize,
            list_table_versions_count: AtomicUsize,
        }

        impl TrackingNamespace {
            fn new(inner: DirectoryNamespace) -> Self {
                Self {
                    inner,
                    create_table_version_count: AtomicUsize::new(0),
                    describe_table_version_count: AtomicUsize::new(0),
                    list_table_versions_count: AtomicUsize::new(0),
                }
            }

            fn create_table_version_calls(&self) -> usize {
                self.create_table_version_count.load(Ordering::SeqCst)
            }

            fn describe_table_version_calls(&self) -> usize {
                self.describe_table_version_count.load(Ordering::SeqCst)
            }

            fn list_table_versions_calls(&self) -> usize {
                self.list_table_versions_count.load(Ordering::SeqCst)
            }
        }

        impl std::fmt::Debug for TrackingNamespace {
            fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
                f.debug_struct("TrackingNamespace")
                    .field(
                        "create_table_version_calls",
                        &self.create_table_version_calls(),
                    )
                    .finish()
            }
        }

        #[async_trait]
        impl LanceNamespace for TrackingNamespace {
            async fn create_namespace(
                &self,
                request: CreateNamespaceRequest,
            ) -> Result<CreateNamespaceResponse> {
                self.inner.create_namespace(request).await
            }

            async fn describe_namespace(
                &self,
                request: DescribeNamespaceRequest,
            ) -> Result<DescribeNamespaceResponse> {
                self.inner.describe_namespace(request).await
            }

            async fn namespace_exists(&self, request: NamespaceExistsRequest) -> Result<()> {
                self.inner.namespace_exists(request).await
            }

            async fn list_namespaces(
                &self,
                request: ListNamespacesRequest,
            ) -> Result<ListNamespacesResponse> {
                self.inner.list_namespaces(request).await
            }

            async fn drop_namespace(
                &self,
                request: DropNamespaceRequest,
            ) -> Result<DropNamespaceResponse> {
                self.inner.drop_namespace(request).await
            }

            async fn list_tables(&self, request: ListTablesRequest) -> Result<ListTablesResponse> {
                self.inner.list_tables(request).await
            }

            async fn describe_table(
                &self,
                request: DescribeTableRequest,
            ) -> Result<DescribeTableResponse> {
                self.inner.describe_table(request).await
            }

            async fn table_exists(&self, request: TableExistsRequest) -> Result<()> {
                self.inner.table_exists(request).await
            }

            async fn drop_table(&self, request: DropTableRequest) -> Result<DropTableResponse> {
                self.inner.drop_table(request).await
            }

            async fn create_table(
                &self,
                request: CreateTableRequest,
                request_data: Bytes,
            ) -> Result<CreateTableResponse> {
                self.inner.create_table(request, request_data).await
            }

            async fn declare_table(
                &self,
                request: DeclareTableRequest,
            ) -> Result<DeclareTableResponse> {
                self.inner.declare_table(request).await
            }

            async fn list_table_versions(
                &self,
                request: ListTableVersionsRequest,
            ) -> Result<ListTableVersionsResponse> {
                self.list_table_versions_count
                    .fetch_add(1, Ordering::SeqCst);
                self.inner.list_table_versions(request).await
            }

            async fn create_table_version(
                &self,
                request: CreateTableVersionRequest,
            ) -> Result<CreateTableVersionResponse> {
                self.create_table_version_count
                    .fetch_add(1, Ordering::SeqCst);
                self.inner.create_table_version(request).await
            }

            async fn describe_table_version(
                &self,
                request: DescribeTableVersionRequest,
            ) -> Result<DescribeTableVersionResponse> {
                self.describe_table_version_count
                    .fetch_add(1, Ordering::SeqCst);
                self.inner.describe_table_version(request).await
            }

            async fn batch_delete_table_versions(
                &self,
                request: BatchDeleteTableVersionsRequest,
            ) -> Result<BatchDeleteTableVersionsResponse> {
                self.inner.batch_delete_table_versions(request).await
            }

            fn namespace_id(&self) -> String {
                self.inner.namespace_id()
            }
        }

        #[tokio::test]
        async fn test_describe_table_returns_managed_versioning() {
            use lance_namespace::models::{CreateNamespaceRequest, DescribeTableRequest};

            let temp_dir = TempStdDir::default();
            let temp_path = temp_dir.to_str().unwrap();

            // Create namespace with table_version_tracking_enabled and manifest_enabled
            let ns = DirectoryNamespaceBuilder::new(temp_path)
                .table_version_tracking_enabled(true)
                .manifest_enabled(true)
                .build()
                .await
                .unwrap();

            // Create parent namespace
            let mut create_ns_req = CreateNamespaceRequest::new();
            create_ns_req.id = Some(vec!["workspace".to_string()]);
            ns.create_namespace(create_ns_req).await.unwrap();

            // Create a table with multi-level ID (namespace + table)
            let schema = create_test_schema();
            let ipc_data = create_test_ipc_data(&schema);
            let mut create_req = CreateTableRequest::new();
            create_req.id = Some(vec!["workspace".to_string(), "test_table".to_string()]);
            ns.create_table(create_req, bytes::Bytes::from(ipc_data))
                .await
                .unwrap();

            // Describe table should return managed_versioning=true
            let mut describe_req = DescribeTableRequest::new();
            describe_req.id = Some(vec!["workspace".to_string(), "test_table".to_string()]);
            let describe_resp = ns.describe_table(describe_req).await.unwrap();

            // managed_versioning should be true
            assert_eq!(
                describe_resp.managed_versioning,
                Some(true),
                "managed_versioning should be true when table_version_tracking_enabled=true"
            );
        }

        #[tokio::test]
        #[cfg(not(windows))]
        async fn test_external_manifest_store_invokes_namespace_apis() {
            use arrow::array::{Int32Array, StringArray};
            use arrow::datatypes::{DataType, Field, Schema as ArrowSchema};
            use arrow::record_batch::RecordBatch;
            use lance::Dataset;
            use lance::dataset::builder::DatasetBuilder;
            use lance::dataset::{WriteMode, WriteParams};
            use lance_namespace::models::CreateNamespaceRequest;

            let temp_dir = TempStdDir::default();
            let temp_path = temp_dir.to_str().unwrap();

            // Create namespace with table_version_tracking_enabled and manifest_enabled
            let inner_ns = DirectoryNamespaceBuilder::new(temp_path)
                .table_version_tracking_enabled(true)
                .manifest_enabled(true)
                .build()
                .await
                .unwrap();

            let tracking_ns = Arc::new(TrackingNamespace::new(inner_ns));
            let ns: Arc<dyn LanceNamespace> = tracking_ns.clone();

            // Create parent namespace
            let mut create_ns_req = CreateNamespaceRequest::new();
            create_ns_req.id = Some(vec!["workspace".to_string()]);
            ns.create_namespace(create_ns_req).await.unwrap();

            // Create a table with multi-level ID (namespace + table)
            let table_id = vec!["workspace".to_string(), "test_table".to_string()];

            // Create some initial data
            let arrow_schema = Arc::new(ArrowSchema::new(vec![
                Field::new("id", DataType::Int32, false),
                Field::new("name", DataType::Utf8, true),
            ]));
            let batch = RecordBatch::try_new(
                arrow_schema.clone(),
                vec![
                    Arc::new(Int32Array::from(vec![1, 2, 3])),
                    Arc::new(StringArray::from(vec!["a", "b", "c"])),
                ],
            )
            .unwrap();

            // Create a table using write_into_namespace
            let batches = RecordBatchIterator::new(vec![Ok(batch.clone())], arrow_schema.clone());
            let write_params = WriteParams {
                mode: WriteMode::Create,
                ..Default::default()
            };
            let mut dataset = Dataset::write_into_namespace(
                batches,
                ns.clone(),
                table_id.clone(),
                Some(write_params),
            )
            .await
            .unwrap();
            assert_eq!(dataset.version().version, 1);

            // Verify create_table_version was called once during initial write_into_namespace
            assert_eq!(
                tracking_ns.create_table_version_calls(),
                1,
                "create_table_version should have been called once during initial write_into_namespace"
            );

            // Append data - this should call create_table_version again
            let append_batch = RecordBatch::try_new(
                arrow_schema.clone(),
                vec![
                    Arc::new(Int32Array::from(vec![4, 5, 6])),
                    Arc::new(StringArray::from(vec!["d", "e", "f"])),
                ],
            )
            .unwrap();
            let append_batches = RecordBatchIterator::new(vec![Ok(append_batch)], arrow_schema);
            dataset.append(append_batches, None).await.unwrap();

            assert_eq!(
                tracking_ns.create_table_version_calls(),
                2,
                "create_table_version should have been called twice (once for create, once for append)"
            );

            // checkout_latest should call list_table_versions exactly once
            let initial_list_calls = tracking_ns.list_table_versions_calls();
            let latest_dataset = DatasetBuilder::from_namespace(ns.clone(), table_id.clone())
                .await
                .unwrap()
                .load()
                .await
                .unwrap();
            assert_eq!(latest_dataset.version().version, 2);
            assert_eq!(
                tracking_ns.list_table_versions_calls(),
                initial_list_calls + 1,
                "list_table_versions should have been called exactly once during checkout_latest"
            );

            // checkout to specific version should call describe_table_version exactly once
            let initial_describe_calls = tracking_ns.describe_table_version_calls();
            let v1_dataset = DatasetBuilder::from_namespace(ns.clone(), table_id.clone())
                .await
                .unwrap()
                .with_version(1)
                .load()
                .await
                .unwrap();
            assert_eq!(v1_dataset.version().version, 1);
            assert_eq!(
                tracking_ns.describe_table_version_calls(),
                initial_describe_calls + 1,
                "describe_table_version should have been called exactly once during checkout to version 1"
            );
        }

        #[tokio::test]
        #[cfg(not(windows))]
        async fn test_dataset_commit_with_external_manifest_store() {
            use arrow::array::{Int32Array, StringArray};
            use arrow::datatypes::{DataType, Field, Schema as ArrowSchema};
            use arrow::record_batch::RecordBatch;
            use futures::TryStreamExt;
            use lance::dataset::{Dataset, WriteMode, WriteParams};
            use lance_namespace::models::CreateNamespaceRequest;
            use lance_table::io::commit::ManifestNamingScheme;

            let temp_dir = TempStdDir::default();
            let temp_path = temp_dir.to_str().unwrap();

            // Create namespace with table_version_tracking_enabled and manifest_enabled
            let inner_ns = DirectoryNamespaceBuilder::new(temp_path)
                .table_version_tracking_enabled(true)
                .manifest_enabled(true)
                .build()
                .await
                .unwrap();

            let tracking_ns: Arc<dyn LanceNamespace> = Arc::new(TrackingNamespace::new(inner_ns));

            // Create parent namespace
            let mut create_ns_req = CreateNamespaceRequest::new();
            create_ns_req.id = Some(vec!["workspace".to_string()]);
            tracking_ns.create_namespace(create_ns_req).await.unwrap();

            // Create a table using write_into_namespace
            let table_id = vec!["workspace".to_string(), "test_table".to_string()];
            let arrow_schema = Arc::new(ArrowSchema::new(vec![
                Field::new("id", DataType::Int32, false),
                Field::new("name", DataType::Utf8, true),
            ]));
            let batch = RecordBatch::try_new(
                arrow_schema.clone(),
                vec![
                    Arc::new(Int32Array::from(vec![1, 2, 3])),
                    Arc::new(StringArray::from(vec!["a", "b", "c"])),
                ],
            )
            .unwrap();
            let batches = RecordBatchIterator::new(vec![Ok(batch)], arrow_schema.clone());
            let write_params = WriteParams {
                mode: WriteMode::Create,
                ..Default::default()
            };
            let dataset = Dataset::write_into_namespace(
                batches,
                tracking_ns.clone(),
                table_id.clone(),
                Some(write_params),
            )
            .await
            .unwrap();
            assert_eq!(dataset.version().version, 1);

            // Append data using write_into_namespace (APPEND mode)
            let batch2 = RecordBatch::try_new(
                arrow_schema.clone(),
                vec![
                    Arc::new(Int32Array::from(vec![4, 5, 6])),
                    Arc::new(StringArray::from(vec!["d", "e", "f"])),
                ],
            )
            .unwrap();
            let batches = RecordBatchIterator::new(vec![Ok(batch2)], arrow_schema);
            let write_params = WriteParams {
                mode: WriteMode::Append,
                ..Default::default()
            };
            Dataset::write_into_namespace(
                batches,
                tracking_ns.clone(),
                table_id.clone(),
                Some(write_params),
            )
            .await
            .unwrap();

            // Verify version 2 was created using the dataset's object_store
            // List manifests in the versions directory to find the V2 named manifest
            let manifest_metas: Vec<_> = dataset
                .object_store()
                .inner
                .list(Some(&dataset.versions_dir()))
                .try_collect()
                .await
                .unwrap();
            let version_2_found = manifest_metas.iter().any(|m| {
                m.location
                    .filename()
                    .map(|f| {
                        f.ends_with(".manifest")
                            && ManifestNamingScheme::V2.parse_version(f) == Some(2)
                    })
                    .unwrap_or(false)
            });
            assert!(
                version_2_found,
                "Version 2 manifest should exist in versions directory"
            );
        }
    }

    /// Tests for multi-table transaction support via table_version_storage_enabled.
    mod multi_table_transactions {
        use super::*;
        use futures::TryStreamExt;
        use lance::dataset::builder::DatasetBuilder;
        use lance_namespace::models::CreateTableVersionRequest;

        /// Helper to create a namespace with table_version_storage_enabled enabled
        async fn create_managed_namespace(temp_path: &str) -> Arc<DirectoryNamespace> {
            Arc::new(
                DirectoryNamespaceBuilder::new(temp_path)
                    .table_version_tracking_enabled(true)
                    .table_version_storage_enabled(true)
                    .manifest_enabled(true)
                    .build()
                    .await
                    .unwrap(),
            )
        }

        /// Helper to create a table and get its staging manifest path
        async fn create_table_and_get_staging(
            namespace: Arc<dyn LanceNamespace>,
            table_name: &str,
        ) -> (Vec<String>, object_store::path::Path) {
            let schema = create_test_schema();
            let ipc_data = create_test_ipc_data(&schema);
            let mut create_req = CreateTableRequest::new();
            create_req.id = Some(vec![table_name.to_string()]);
            namespace
                .create_table(create_req, bytes::Bytes::from(ipc_data))
                .await
                .unwrap();

            let table_id = vec![table_name.to_string()];
            let dataset = DatasetBuilder::from_namespace(namespace.clone(), table_id.clone())
                .await
                .unwrap()
                .load()
                .await
                .unwrap();

            // Find existing manifest and create a staging copy
            let versions_path = dataset.versions_dir();
            let manifest_metas: Vec<_> = dataset
                .object_store()
                .inner
                .list(Some(&versions_path))
                .try_collect()
                .await
                .unwrap();

            let manifest_meta = manifest_metas
                .iter()
                .find(|m| {
                    m.location
                        .filename()
                        .map(|f| f.ends_with(".manifest"))
                        .unwrap_or(false)
                })
                .expect("No manifest file found");

            let manifest_data = dataset
                .object_store()
                .inner
                .get(&manifest_meta.location)
                .await
                .unwrap()
                .bytes()
                .await
                .unwrap();

            let staging_path = dataset
                .versions_dir()
                .child(format!("staging_{}", table_name));
            dataset
                .object_store()
                .inner
                .put(&staging_path, manifest_data.into())
                .await
                .unwrap();

            (table_id, staging_path)
        }

        #[tokio::test]
        async fn test_table_version_storage_enabled_requires_manifest() {
            // table_version_storage_enabled=true requires manifest_enabled=true
            let temp_dir = TempStdDir::default();
            let temp_path = temp_dir.to_str().unwrap();

            let result = DirectoryNamespaceBuilder::new(temp_path)
                .table_version_storage_enabled(true)
                .manifest_enabled(false)
                .build()
                .await;

            assert!(
                result.is_err(),
                "Should fail when table_version_storage_enabled=true but manifest_enabled=false"
            );
        }

        #[tokio::test]
        #[cfg(not(windows))]
        async fn test_create_table_version_records_in_manifest() {
            // When table_version_storage_enabled is enabled, single create_table_version
            // should also record the version in __manifest
            let temp_dir = TempStrDir::default();
            let temp_path: &str = &temp_dir;

            let namespace = create_managed_namespace(temp_path).await;
            let ns: Arc<dyn LanceNamespace> = namespace.clone();

            let (table_id, staging_path) =
                create_table_and_get_staging(ns.clone(), "table_managed").await;

            // Create version 2
            let mut create_req = CreateTableVersionRequest::new(2, staging_path.to_string());
            create_req.id = Some(table_id.clone());
            create_req.naming_scheme = Some("V2".to_string());
            let response = namespace.create_table_version(create_req).await.unwrap();

            assert!(response.version.is_some());
            let version = response.version.unwrap();
            assert_eq!(version.version, 2);

            // Verify the version is recorded in __manifest by querying it
            let manifest_ns = namespace.manifest_ns.as_ref().unwrap();
            let table_id_str = manifest::ManifestNamespace::str_object_id(&table_id);
            let versions = manifest_ns
                .query_table_versions(&table_id_str, false, None)
                .await
                .unwrap();

            assert!(
                !versions.is_empty(),
                "Version should be recorded in __manifest"
            );
            let (ver, _path) = &versions[0];
            assert_eq!(*ver, 2, "Recorded version should be 2");
        }
    }
}