use std::collections::{HashMap, HashSet};

use ducklake_catalog::{
    CatalogError, DataFile, PartitionTransform, ResolvedFilePartitionValue,
    ResolvedInlinedDataTable, ResolvedMappingTarget, SUPPORTED_VERSION, Snapshot, SnapshotSelector,
    TableScanPlan, resolve_files, resolve_snapshot, try_build_schema,
};

use crate::{
    CatalogBackend, CatalogSettings, ListedCatalogPath, ListedFile, ListedOption, ListedSnapshot,
    ListedTag, Result, StorageError, TableInfo, legacy_name_fallback_mapping,
    reject_unsupported_files,
};

pub async fn plan_table_scan<B: CatalogBackend + ?Sized>(
    backend: &B,
    schema_name: &str,
    table_name: &str,
    selector: SnapshotSelector,
) -> Result<TableScanPlan> {
    let meta = backend.metadata().await?;
    if meta.version != SUPPORTED_VERSION {
        return Err(CatalogError::UnsupportedVersion {
            found: meta.version.clone(),
            supported: SUPPORTED_VERSION.into(),
        }
        .into());
    }
    if meta.encrypted {
        return Err(StorageError::Unsupported(
            "Encrypted DuckLake catalogs are not supported".into(),
        ));
    }

    let snapshots = backend.snapshots().await?;
    let snapshot_id = resolve_snapshot(&snapshots, selector)?;
    let snapshot = snapshots
        .iter()
        .find(|s| s.snapshot_id == snapshot_id)
        .ok_or_else(|| CatalogError::NotFound(format!("snapshot {}", snapshot_id.0)))?;

    let schema = resolve_catalog_identifier(
        backend.schemas_at(snapshot_id).await?,
        schema_name,
        "schema",
        |schema| &schema.schema_name,
    )?;

    let table = resolve_catalog_identifier(
        backend.tables_at(schema.schema_id, snapshot_id).await?,
        table_name,
        "table",
        |table| &table.table_name,
    )?;

    let columns = backend.columns_at(table.table_id, snapshot_id).await?;
    let table_schema = try_build_schema(&columns, snapshot_id)?;
    let data_files = backend.data_files_at(table.table_id, snapshot_id).await?;
    let delete_files = backend.delete_files_at(table.table_id, snapshot_id).await?;
    let mut files = resolve_files(
        &data_files,
        &delete_files,
        &meta,
        &table,
        &schema,
        snapshot_id,
    )?;
    reject_unsupported_files(&files)?;
    let mut mapped_hive_partition_values = Vec::new();
    for file in &mut files {
        if let Some(mapping_id) = file.data.mapping_id {
            let (mapping, entries) = backend.column_mapping(mapping_id).await?;
            let mut by_parent_and_name = HashMap::new();
            for entry in entries {
                if entry.is_partition {
                    if let Some(partition_value) =
                        hive_partition_value_from_path(&file.resolved_path, &entry.source_name)
                    {
                        mapped_hive_partition_values.push(ResolvedFilePartitionValue {
                            data_file_id: file.data.data_file_id,
                            column_id: entry.target_field_id,
                            transform: PartitionTransform::Identity,
                            partition_value,
                        });
                    }
                    continue;
                }
                by_parent_and_name.insert(
                    (entry.parent_column, entry.source_name),
                    ResolvedMappingTarget {
                        mapping_column_id: entry.column_id,
                        target_field_id: entry.target_field_id,
                    },
                );
            }
            file.mapping = Some(ducklake_catalog::ResolvedMapping {
                mapping_type: mapping.mapping_type,
                by_parent_and_name,
            });
        }
    }
    let mut legacy_name_fallbacks: HashMap<
        ducklake_catalog::SnapshotId,
        ducklake_catalog::ResolvedMapping,
    > = HashMap::new();
    for file in &mut files {
        if file.data.mapping_id.is_some() || file.mapping.is_some() {
            continue;
        }
        let file_snapshot = file.data.validity.begin;
        let mapping = match legacy_name_fallbacks.get(&file_snapshot) {
            Some(mapping) => mapping.clone(),
            None => {
                let columns = backend.columns_at(table.table_id, file_snapshot).await?;
                let mapping = legacy_name_fallback_mapping(&columns);
                legacy_name_fallbacks.insert(file_snapshot, mapping.clone());
                mapping
            }
        };
        if !mapping.by_parent_and_name.is_empty() {
            file.mapping = Some(mapping);
        }
    }
    let mut inlined_deletions_by_file: HashMap<_, Vec<_>> = HashMap::new();
    for deletion in backend
        .inlined_file_deletions_at(table.table_id, snapshot_id)
        .await?
    {
        inlined_deletions_by_file
            .entry(deletion.data_file_id)
            .or_default()
            .push(deletion);
    }
    for file in &mut files {
        if let Some(deletions) = inlined_deletions_by_file.remove(&file.data.data_file_id) {
            file.inlined_deletions = deletions;
        }
    }
    let stats = backend.file_stats(table.table_id).await?;
    let variant_stats = match backend.file_variant_stats(table.table_id).await {
        Ok(stats) => stats,
        Err(err) if missing_optional_file_variant_stats_table(&err) => Vec::new(),
        Err(err) => return Err(err),
    };
    let latest_snapshot = snapshots.iter().map(|snapshot| snapshot.snapshot_id).max();
    let table_column_stats = if latest_snapshot == Some(snapshot_id) {
        match backend.table_column_stats(table.table_id).await {
            Ok(stats) => stats,
            Err(err) if missing_optional_table_column_stats_table(&err) => Vec::new(),
            Err(err) => return Err(err),
        }
    } else {
        Vec::new()
    };
    let mut partition_values =
        resolved_partition_values(backend, table.table_id, snapshot_id, &data_files).await?;
    partition_values.extend(mapped_hive_partition_values);

    let inlined =
        resolved_inlined_tables(backend, table.table_id, snapshot, &snapshots, &table_schema)
            .await?;

    Ok(TableScanPlan {
        schema_entry: schema,
        table,
        schema: table_schema,
        snapshot: snapshot_id,
        data_path: meta.data_path.clone(),
        files,
        stats,
        variant_stats,
        table_column_stats,
        partition_values,
        inlined,
    })
}

pub async fn list_table_files<B: CatalogBackend + ?Sized>(
    backend: &B,
    schema_name: &str,
    table_name: &str,
    selector: SnapshotSelector,
) -> Result<Vec<ListedFile>> {
    let plan = plan_table_scan(backend, schema_name, table_name, selector).await?;
    let mut files: Vec<_> = plan
        .files
        .into_iter()
        .map(|file| ListedFile {
            data_file: file.resolved_path,
            data_file_size_bytes: file.data.file_size_bytes,
            data_file_footer_size: file.data.footer_size,
            data_file_encryption_key: file.data.encryption_key,
            delete_file: file.resolved_delete_path,
            delete_file_size_bytes: file.delete.as_ref().map(|delete| delete.file_size_bytes),
            delete_file_footer_size: file.delete.as_ref().and_then(|delete| delete.footer_size),
            delete_file_encryption_key: file
                .delete
                .as_ref()
                .and_then(|delete| delete.encryption_key.clone()),
        })
        .collect();
    files.sort_by(|left, right| left.data_file.cmp(&right.data_file));
    Ok(files)
}

pub async fn list_table_info<B: CatalogBackend + ?Sized>(
    backend: &B,
    selector: SnapshotSelector,
) -> Result<Vec<TableInfo>> {
    let meta = backend.metadata().await?;
    if meta.version != SUPPORTED_VERSION {
        return Err(CatalogError::UnsupportedVersion {
            found: meta.version.clone(),
            supported: SUPPORTED_VERSION.into(),
        }
        .into());
    }
    if meta.encrypted {
        return Err(StorageError::Unsupported(
            "Encrypted DuckLake catalogs are not supported".into(),
        ));
    }

    let snapshots = backend.snapshots().await?;
    let snapshot_id = resolve_snapshot(&snapshots, selector)?;
    let mut tables = Vec::new();
    for schema in backend.schemas_at(snapshot_id).await? {
        for table in backend.tables_at(schema.schema_id, snapshot_id).await? {
            let data_files = backend.data_files_at(table.table_id, snapshot_id).await?;
            let delete_files = backend.delete_files_at(table.table_id, snapshot_id).await?;
            let columns = backend.columns_at(table.table_id, snapshot_id).await?;
            let column_count = i64::try_from(
                columns
                    .iter()
                    .filter(|column| column.parent_column.is_none())
                    .count(),
            )
            .map_err(|_| {
                StorageError::Backend("DuckLake table_info column count overflow".into())
            })?;
            let record_count = match backend.table_stats(table.table_id).await? {
                Some(stats) => stats.record_count,
                None => data_files.iter().map(|file| file.record_count).try_fold(
                    0_i64,
                    |acc, value| {
                        acc.checked_add(value).ok_or_else(|| {
                            StorageError::Backend(
                                "DuckLake table_info record count overflow".into(),
                            )
                        })
                    },
                )?,
            };
            let file_count = i64::try_from(data_files.len()).map_err(|_| {
                StorageError::Backend("DuckLake table_info data-file count overflow".into())
            })?;
            let delete_file_count = i64::try_from(delete_files.len()).map_err(|_| {
                StorageError::Backend("DuckLake table_info delete-file count overflow".into())
            })?;
            let file_size_bytes = data_files
                .iter()
                .map(|file| file.file_size_bytes)
                .try_fold(0_i64, |acc, value| {
                    acc.checked_add(value).ok_or_else(|| {
                        StorageError::Backend("DuckLake table_info data-file size overflow".into())
                    })
                })?;
            let delete_file_size_bytes = delete_files
                .iter()
                .map(|file| file.file_size_bytes)
                .try_fold(0_i64, |acc, value| {
                    acc.checked_add(value).ok_or_else(|| {
                        StorageError::Backend(
                            "DuckLake table_info delete-file size overflow".into(),
                        )
                    })
                })?;
            tables.push(TableInfo {
                table_name: table.table_name,
                schema_id: table.schema_id,
                table_id: table.table_id,
                table_uuid: table.table_uuid,
                record_count,
                column_count,
                file_count,
                file_size_bytes,
                delete_file_count,
                delete_file_size_bytes,
            });
        }
    }
    tables.sort_by(|left, right| {
        (left.schema_id.0, left.table_id.0).cmp(&(right.schema_id.0, right.table_id.0))
    });
    Ok(tables)
}

pub async fn list_catalog_paths<B: CatalogBackend + ?Sized>(
    backend: &B,
    selector: SnapshotSelector,
) -> Result<Vec<ListedCatalogPath>> {
    let meta = backend.metadata().await?;
    if meta.version != SUPPORTED_VERSION {
        return Err(CatalogError::UnsupportedVersion {
            found: meta.version.clone(),
            supported: SUPPORTED_VERSION.into(),
        }
        .into());
    }
    if meta.encrypted {
        return Err(StorageError::Unsupported(
            "Encrypted DuckLake catalogs are not supported".into(),
        ));
    }

    let snapshots = backend.snapshots().await?;
    let snapshot_id = resolve_snapshot(&snapshots, selector)?;
    let mut rows = Vec::new();
    for schema in backend.schemas_at(snapshot_id).await? {
        rows.push(ListedCatalogPath {
            object_kind: "schema".to_string(),
            schema_name: Some(schema.schema_name.clone()),
            table_name: None,
            data_file_id: None,
            path: schema.path.clone(),
            path_is_relative: schema.path_is_relative,
        });

        for table in backend.tables_at(schema.schema_id, snapshot_id).await? {
            rows.push(ListedCatalogPath {
                object_kind: "table".to_string(),
                schema_name: Some(schema.schema_name.clone()),
                table_name: Some(table.table_name.clone()),
                data_file_id: None,
                path: table.path.clone(),
                path_is_relative: table.path_is_relative,
            });

            for file in backend.data_files_at(table.table_id, snapshot_id).await? {
                rows.push(ListedCatalogPath {
                    object_kind: "data_file".to_string(),
                    schema_name: Some(schema.schema_name.clone()),
                    table_name: Some(table.table_name.clone()),
                    data_file_id: Some(file.data_file_id),
                    path: Some(file.path),
                    path_is_relative: file.path_is_relative,
                });
            }
        }
    }
    rows.sort_by(|left, right| {
        (
            left.object_kind.as_str(),
            left.schema_name.as_deref(),
            left.table_name.as_deref(),
            left.data_file_id.map(|id| id.0),
            left.path.as_deref(),
        )
            .cmp(&(
                right.object_kind.as_str(),
                right.schema_name.as_deref(),
                right.table_name.as_deref(),
                right.data_file_id.map(|id| id.0),
                right.path.as_deref(),
            ))
    });
    Ok(rows)
}

pub async fn catalog_database_size_bytes<B: CatalogBackend + ?Sized>(
    backend: &B,
    selector: SnapshotSelector,
) -> Result<i64> {
    list_table_info(backend, selector)
        .await?
        .into_iter()
        .try_fold(0_i64, |acc, table| {
            acc.checked_add(table.file_size_bytes)
                .and_then(|value| value.checked_add(table.delete_file_size_bytes))
                .ok_or_else(|| {
                    StorageError::Backend("DuckLake catalog database size overflow".into())
                })
        })
}

pub async fn list_snapshots<B: CatalogBackend + ?Sized>(
    backend: &B,
) -> Result<Vec<ListedSnapshot>> {
    let meta = backend.metadata().await?;
    if meta.version != SUPPORTED_VERSION {
        return Err(CatalogError::UnsupportedVersion {
            found: meta.version.clone(),
            supported: SUPPORTED_VERSION.into(),
        }
        .into());
    }
    if meta.encrypted {
        return Err(StorageError::Unsupported(
            "Encrypted DuckLake catalogs are not supported".into(),
        ));
    }

    let mut changes_by_snapshot = backend
        .snapshot_changes()
        .await?
        .into_iter()
        .map(|change| (change.snapshot_id, change))
        .collect::<HashMap<_, _>>();
    let mut snapshots = backend
        .snapshots()
        .await?
        .into_iter()
        .map(|snapshot| {
            let change = changes_by_snapshot.remove(&snapshot.snapshot_id);
            ListedSnapshot {
                snapshot_id: snapshot.snapshot_id,
                snapshot_time: snapshot.snapshot_time,
                schema_version: snapshot.schema_version,
                next_catalog_id: snapshot.next_catalog_id,
                next_file_id: snapshot.next_file_id,
                changes_made: change
                    .as_ref()
                    .and_then(|change| change.changes_made.clone()),
                author: change.as_ref().and_then(|change| change.author.clone()),
                commit_message: change
                    .as_ref()
                    .and_then(|change| change.commit_message.clone()),
                commit_extra_info: change.and_then(|change| change.commit_extra_info),
            }
        })
        .collect::<Vec<_>>();
    snapshots.sort_by_key(|snapshot| snapshot.snapshot_id);
    Ok(snapshots)
}

pub async fn current_snapshot_id<B: CatalogBackend + ?Sized>(
    backend: &B,
) -> Result<ducklake_catalog::SnapshotId> {
    let meta = backend.metadata().await?;
    if meta.version != SUPPORTED_VERSION {
        return Err(CatalogError::UnsupportedVersion {
            found: meta.version.clone(),
            supported: SUPPORTED_VERSION.into(),
        }
        .into());
    }
    if meta.encrypted {
        return Err(StorageError::Unsupported(
            "Encrypted DuckLake catalogs are not supported".into(),
        ));
    }

    let snapshots = backend.snapshots().await?;
    Ok(resolve_snapshot(&snapshots, SnapshotSelector::Latest)?)
}

pub async fn catalog_settings<B: CatalogBackend + ?Sized>(backend: &B) -> Result<CatalogSettings> {
    let meta = backend.metadata().await?;
    if meta.version != SUPPORTED_VERSION {
        return Err(CatalogError::UnsupportedVersion {
            found: meta.version.clone(),
            supported: SUPPORTED_VERSION.into(),
        }
        .into());
    }
    if meta.encrypted {
        return Err(StorageError::Unsupported(
            "Encrypted DuckLake catalogs are not supported".into(),
        ));
    }

    Ok(CatalogSettings {
        catalog_type: backend.catalog_type().to_string(),
        // DuckDB's helper returns EXT_VERSION_DUCKLAKE when compiled in, or an
        // empty non-NULL string otherwise. We expose the same non-NULL shape.
        extension_version: String::new(),
        data_path: meta.data_path,
    })
}

pub async fn list_options<B: CatalogBackend + ?Sized>(
    backend: &B,
    selector: SnapshotSelector,
) -> Result<Vec<ListedOption>> {
    let meta = backend.metadata().await?;
    if meta.version != SUPPORTED_VERSION {
        return Err(CatalogError::UnsupportedVersion {
            found: meta.version.clone(),
            supported: SUPPORTED_VERSION.into(),
        }
        .into());
    }
    if meta.encrypted {
        return Err(StorageError::Unsupported(
            "Encrypted DuckLake catalogs are not supported".into(),
        ));
    }

    let snapshots = backend.snapshots().await?;
    let snapshot_id = resolve_snapshot(&snapshots, selector)?;
    let schemas = backend.schemas_at(snapshot_id).await?;
    let schema_names = schemas
        .iter()
        .map(|schema| (schema.schema_id.0, schema.schema_name.clone()))
        .collect::<HashMap<_, _>>();
    let mut table_names = HashMap::new();
    for schema in &schemas {
        for table in backend.tables_at(schema.schema_id, snapshot_id).await? {
            table_names.insert(
                table.table_id.0,
                format!("{}.{}", schema.schema_name, table.table_name),
            );
        }
    }

    let mut options = Vec::new();
    for entry in meta.raw {
        let (scope, scope_entry) = match entry.scope.as_deref() {
            None => ("GLOBAL".to_string(), None),
            Some("schema") => (
                "SCHEMA".to_string(),
                entry
                    .scope_id
                    .and_then(|scope_id| schema_names.get(&scope_id).cloned()),
            ),
            Some("table") => (
                "TABLE".to_string(),
                entry
                    .scope_id
                    .and_then(|scope_id| table_names.get(&scope_id).cloned()),
            ),
            Some(scope) => {
                return Err(StorageError::Backend(format!(
                    "unsupported DuckLake setting scope {scope:?}; expected schema/table"
                )));
            }
        };
        let description = ducklake_option_description(&entry.key).map(str::to_string);
        options.push(ListedOption {
            option_name: entry.key,
            description,
            value: entry.value,
            scope,
            scope_entry,
        });
    }
    options.sort_by(|left, right| {
        left.option_name
            .cmp(&right.option_name)
            .then_with(|| left.scope.cmp(&right.scope))
            .then_with(|| left.scope_entry.cmp(&right.scope_entry))
            .then_with(|| left.value.cmp(&right.value))
    });
    Ok(options)
}

pub async fn list_tags<B: CatalogBackend + ?Sized>(
    backend: &B,
    selector: SnapshotSelector,
) -> Result<Vec<ListedTag>> {
    let meta = backend.metadata().await?;
    if meta.version != SUPPORTED_VERSION {
        return Err(CatalogError::UnsupportedVersion {
            found: meta.version.clone(),
            supported: SUPPORTED_VERSION.into(),
        }
        .into());
    }
    if meta.encrypted {
        return Err(StorageError::Unsupported(
            "Encrypted DuckLake catalogs are not supported".into(),
        ));
    }

    let snapshots = backend.snapshots().await?;
    let snapshot_id = resolve_snapshot(&snapshots, selector)?;
    let schemas = backend.schemas_at(snapshot_id).await?;
    let mut objects = HashMap::new();
    let mut visible_tables = Vec::new();

    for schema in schemas {
        objects.insert(
            schema.schema_id.0,
            ("schema".to_string(), Some(schema.schema_name.clone()), None),
        );

        for table in backend.tables_at(schema.schema_id, snapshot_id).await? {
            objects.insert(
                table.table_id.0,
                (
                    "table".to_string(),
                    Some(schema.schema_name.clone()),
                    Some(table.table_name.clone()),
                ),
            );
            visible_tables.push((table.table_id, schema.schema_name.clone(), table.table_name));
        }

        for view in backend.views_at(schema.schema_id, snapshot_id).await? {
            objects.insert(
                view.view_id.0,
                (
                    "view".to_string(),
                    Some(schema.schema_name.clone()),
                    Some(view.view_name),
                ),
            );
        }
    }

    let mut tags = Vec::new();
    for tag in backend.tags_at(snapshot_id).await? {
        let (object_kind, schema_name, object_name) = objects
            .get(&tag.object_id)
            .cloned()
            .unwrap_or(("unknown".to_string(), None, None));
        tags.push(ListedTag {
            object_kind,
            object_id: tag.object_id,
            schema_name,
            object_name,
            column_id: None,
            column_name: None,
            key: tag.key,
            value: tag.value,
        });
    }

    for (table_id, schema_name, table_name) in visible_tables {
        let column_names = backend
            .columns_at(table_id, snapshot_id)
            .await?
            .into_iter()
            .map(|column| (column.column_id, column.column_name))
            .collect::<HashMap<_, _>>();

        for tag in backend.column_tags_at(table_id, snapshot_id).await? {
            tags.push(ListedTag {
                object_kind: "column".to_string(),
                object_id: table_id.0,
                schema_name: Some(schema_name.clone()),
                object_name: Some(table_name.clone()),
                column_id: Some(tag.column_id),
                column_name: column_names.get(&tag.column_id).cloned(),
                key: tag.key,
                value: tag.value,
            });
        }
    }

    tags.sort_by(|left, right| {
        left.object_kind
            .cmp(&right.object_kind)
            .then_with(|| left.schema_name.cmp(&right.schema_name))
            .then_with(|| left.object_name.cmp(&right.object_name))
            .then_with(|| left.object_id.cmp(&right.object_id))
            .then_with(|| left.column_id.cmp(&right.column_id))
            .then_with(|| left.column_name.cmp(&right.column_name))
            .then_with(|| left.key.cmp(&right.key))
            .then_with(|| left.value.cmp(&right.value))
    });
    Ok(tags)
}

fn ducklake_option_description(option_name: &str) -> Option<&'static str> {
    match option_name.to_ascii_lowercase().as_str() {
        "data_inlining_row_limit" => {
            Some("Maximum amount of rows to inline in a single append operation")
        }
        "parquet_compression" => Some(
            "Compression algorithm for Parquet files (uncompressed, snappy, gzip, zstd, brotli, lz4, lz4_raw)",
        ),
        "parquet_version" => Some("Parquet format version (1 or 2)"),
        "parquet_compression_level" => Some("Compression level for Parquet files"),
        "parquet_row_group_size" => Some("Number of rows per row group in Parquet files"),
        "parquet_row_group_size_bytes" => Some("Number of bytes per row group in Parquet files"),
        "hive_file_pattern" => {
            Some("If partitioned data should be written in a hive-like folder structure")
        }
        "target_file_size" => {
            Some("The target data file size for insertion and compaction operations")
        }
        "version" => Some("DuckLake format version"),
        "created_by" => Some("Tool used to write the DuckLake"),
        "data_path" => Some("Path to data files"),
        "require_commit_message" => {
            Some("If an explicit commit message is required for a snapshot commit.")
        }
        "rewrite_delete_threshold" => Some(
            "A threshold that determines the minimum amount of data that must be removed from a file before a rewrite is warranted. From 0 - 1.",
        ),
        "delete_older_than" => Some(
            "How old unused files must be to be removed by the 'ducklake_delete_orphaned_files' and 'ducklake_cleanup_old_files' cleanup functions.",
        ),
        "expire_older_than" => Some(
            "How old snapshots must be, by default, to be expired by: 'ducklake_expire_snapshots'",
        ),
        "auto_compact" => Some(
            "Pre-defined schema used as a default value for the following compaction functions 'ducklake_flush_inlined_data','ducklake_merge_adjacent_files', 'ducklake_rewrite_data_files', 'ducklake_delete_orphaned_files'",
        ),
        "encrypted" => Some("Whether or not to encrypt Parquet files written to the data path"),
        "per_thread_output" => {
            Some("Whether to create separate output files per thread during parallel insertion")
        }
        "write_deletion_vectors" => Some(
            "[EXPERIMENTAL - do not use outside testing] Whether to write Iceberg V3 deletion vectors (puffin) instead of positional delete files (parquet)",
        ),
        "sort_on_insert" => {
            Some("Whether to sort data on INSERT according to SET SORTED BY (default: true)")
        }
        _ => None,
    }
}

pub(crate) fn resolve_catalog_identifier<T>(
    mut entries: Vec<T>,
    requested: &str,
    kind: &str,
    identifier: impl Fn(&T) -> &str,
) -> Result<T> {
    if let Some(index) = entries
        .iter()
        .position(|entry| identifier(entry) == requested)
    {
        return Ok(entries.swap_remove(index));
    }

    let matches = entries
        .iter()
        .enumerate()
        .filter_map(|(index, entry)| {
            identifier(entry)
                .eq_ignore_ascii_case(requested)
                .then_some(index)
        })
        .collect::<Vec<_>>();
    match matches.as_slice() {
        [index] => Ok(entries.swap_remove(*index)),
        [] => Err(CatalogError::NotFound(format!("{kind} {requested}")).into()),
        _ => Err(CatalogError::InvalidCatalog(format!(
            "ambiguous {kind} identifier {requested:?}: multiple entries differ only by ASCII case"
        ))
        .into()),
    }
}

async fn resolved_inlined_tables<B: CatalogBackend + ?Sized>(
    backend: &B,
    table: ducklake_catalog::TableId,
    snapshot: &Snapshot,
    snapshots: &[Snapshot],
    current_schema: &ducklake_catalog::TableSchema,
) -> Result<Vec<ResolvedInlinedDataTable>> {
    let mut version_snapshots = match backend.schema_versions(table).await {
        Ok(versions) => versions
            .into_iter()
            .filter(|version| version.begin_snapshot.0 <= snapshot.snapshot_id.0)
            .map(|version| (version.schema_version, version.begin_snapshot))
            .collect::<Vec<_>>(),
        Err(err) if missing_optional_schema_versions_table(&err) => snapshots
            .iter()
            .filter(|candidate| candidate.snapshot_id.0 <= snapshot.snapshot_id.0)
            .map(|candidate| (candidate.schema_version, candidate.snapshot_id))
            .collect::<Vec<_>>(),
        Err(err) => return Err(err),
    };
    if !version_snapshots
        .iter()
        .any(|(schema_version, _)| *schema_version == snapshot.schema_version)
    {
        version_snapshots.push((snapshot.schema_version, snapshot.snapshot_id));
    }
    version_snapshots
        .sort_by_key(|(schema_version, begin_snapshot)| (*schema_version, begin_snapshot.0));
    version_snapshots.dedup_by_key(|(schema_version, _)| *schema_version);

    let mut resolved = Vec::new();
    for (schema_version, begin_snapshot) in version_snapshots {
        let tables = backend.inlined_tables(table, schema_version).await?;
        if tables.is_empty() {
            continue;
        }
        let source_schema = if schema_version == snapshot.schema_version {
            current_schema.clone()
        } else {
            let columns = backend.columns_at(table, begin_snapshot).await?;
            try_build_schema(&columns, begin_snapshot)?
        };
        resolved.extend(tables.into_iter().map(|table| ResolvedInlinedDataTable {
            table,
            schema: source_schema.clone(),
        }));
    }

    Ok(resolved)
}

async fn resolved_partition_values<B: CatalogBackend + ?Sized>(
    backend: &B,
    table: ducklake_catalog::TableId,
    snapshot: ducklake_catalog::SnapshotId,
    data_files: &[DataFile],
) -> Result<Vec<ResolvedFilePartitionValue>> {
    let file_partition_id_by_file: HashMap<_, _> = data_files
        .iter()
        .filter_map(|file| {
            file.partition_id
                .map(|partition| (file.data_file_id, partition))
        })
        .collect();
    if file_partition_id_by_file.is_empty() {
        return Ok(Vec::new());
    }

    let partition_infos = match backend.partition_info_at(table, snapshot).await {
        Ok(partition_infos) => partition_infos,
        Err(err) if missing_optional_partition_table(&err) => return Ok(Vec::new()),
        Err(err) => return Err(err),
    };

    let visible_partition_ids: HashSet<_> = partition_infos
        .into_iter()
        .map(|partition| partition.partition_id)
        .collect();
    if visible_partition_ids.is_empty() {
        return Ok(Vec::new());
    }

    let mut columns_by_partition_key = HashMap::new();
    for partition_id in &visible_partition_ids {
        let columns = match backend.partition_columns(*partition_id).await {
            Ok(columns) => columns,
            Err(err) if missing_optional_partition_table(&err) => return Ok(Vec::new()),
            Err(err) => return Err(err),
        };
        for column in columns {
            columns_by_partition_key
                .insert((column.partition_id, column.partition_key_index), column);
        }
    }
    if columns_by_partition_key.is_empty() {
        return Ok(Vec::new());
    }

    let raw_values = match backend.file_partition_values(table).await {
        Ok(values) => values,
        Err(err) if missing_optional_partition_table(&err) => return Ok(Vec::new()),
        Err(err) => return Err(err),
    };

    let mut resolved = Vec::new();
    for value in raw_values {
        let Some(partition_id) = file_partition_id_by_file.get(&value.data_file_id) else {
            continue;
        };
        if !visible_partition_ids.contains(partition_id) {
            continue;
        }
        let Some(column) =
            columns_by_partition_key.get(&(*partition_id, value.partition_key_index))
        else {
            continue;
        };
        resolved.push(ResolvedFilePartitionValue {
            data_file_id: value.data_file_id,
            column_id: column.column_id,
            transform: column.transform.clone(),
            partition_value: value.partition_value,
        });
    }

    Ok(resolved)
}

pub(crate) fn hive_partition_value_from_path(
    path: &str,
    source_name: &str,
) -> Option<Option<String>> {
    for segment in path.split('/') {
        let Some((key, raw_value)) = segment.split_once('=') else {
            continue;
        };
        if !key.eq_ignore_ascii_case(source_name) {
            continue;
        }
        let value = percent_decode_path_component(raw_value);
        if value == "__HIVE_DEFAULT_PARTITION__" {
            return Some(None);
        }
        return Some(Some(value));
    }
    None
}

fn percent_decode_path_component(value: &str) -> String {
    let bytes = value.as_bytes();
    let mut decoded = Vec::with_capacity(bytes.len());
    let mut idx = 0;
    while idx < bytes.len() {
        if bytes[idx] == b'%'
            && idx + 2 < bytes.len()
            && let (Some(high), Some(low)) = (hex_value(bytes[idx + 1]), hex_value(bytes[idx + 2]))
        {
            decoded.push((high << 4) | low);
            idx += 3;
            continue;
        }
        decoded.push(bytes[idx]);
        idx += 1;
    }
    String::from_utf8(decoded).unwrap_or_else(|_| value.to_string())
}

fn hex_value(byte: u8) -> Option<u8> {
    match byte {
        b'0'..=b'9' => Some(byte - b'0'),
        b'a'..=b'f' => Some(byte - b'a' + 10),
        b'A'..=b'F' => Some(byte - b'A' + 10),
        _ => None,
    }
}

fn missing_optional_partition_table(err: &StorageError) -> bool {
    let StorageError::Backend(message) = err else {
        return false;
    };
    let lower = message.to_ascii_lowercase();
    let mentions_partition_table = [
        "ducklake_partition_info",
        "ducklake_partition_column",
        "ducklake_file_partition_value",
    ]
    .iter()
    .any(|table| lower.contains(table));
    mentions_partition_table
        && (lower.contains("no such table")
            || lower.contains("does not exist")
            || lower.contains("not found"))
}

fn missing_optional_table_column_stats_table(err: &StorageError) -> bool {
    let StorageError::Backend(message) = err else {
        return false;
    };
    let lower = message.to_ascii_lowercase();
    lower.contains("ducklake_table_column_stats")
        && (lower.contains("no such table")
            || lower.contains("does not exist")
            || lower.contains("not found"))
}

fn missing_optional_file_variant_stats_table(err: &StorageError) -> bool {
    let StorageError::Backend(message) = err else {
        return false;
    };
    let lower = message.to_ascii_lowercase();
    lower.contains("ducklake_file_variant_stats")
        && (lower.contains("no such table")
            || lower.contains("does not exist")
            || lower.contains("not found"))
}

fn missing_optional_schema_versions_table(err: &StorageError) -> bool {
    let StorageError::Backend(message) = err else {
        return false;
    };
    let lower = message.to_ascii_lowercase();
    lower.contains("ducklake_schema_versions")
        && (lower.contains("no such table")
            || lower.contains("does not exist")
            || lower.contains("not found"))
}
