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

use arrow_schema::{DataType, Field, TimeUnit};
use ducklake_catalog::{
    ColumnEntry, ColumnId, DeleteFormat, FilePhysicalColumn, FilePhysicalSchema, ResolvedFile,
    ResolvedMappingTarget,
};
use parquet::arrow::{PARQUET_FIELD_ID_META_KEY, arrow_reader::ParquetRecordBatchReaderBuilder};
use parquet::basic::{ConvertedType, LogicalType};
use parquet::schema::types::{SchemaDescriptor, Type as ParquetSchemaType};

use crate::{CreateTableColumn, Result, StorageError, local_filesystem_path};

const LEGACY_NAME_FALLBACK_MAPPING_TYPE: &str = "legacy_name_fallback";

pub(crate) fn legacy_name_fallback_mapping(
    columns: &[ColumnEntry],
) -> ducklake_catalog::ResolvedMapping {
    let by_parent_and_name = columns
        .iter()
        .map(|column| {
            (
                (column.parent_column, column.column_name.clone()),
                ResolvedMappingTarget {
                    mapping_column_id: column.column_id,
                    target_field_id: column.column_id,
                },
            )
        })
        .collect();

    ducklake_catalog::ResolvedMapping {
        mapping_type: LEGACY_NAME_FALLBACK_MAPPING_TYPE.into(),
        by_parent_and_name,
    }
}

fn missing_parquet_field_id_error(err: &StorageError) -> bool {
    matches!(err, StorageError::Unsupported(message) if message.contains("has no field id"))
}

pub fn physical_schema_for_file(file: &ResolvedFile) -> Result<FilePhysicalSchema> {
    if !file.data.file_format.eq_ignore_ascii_case("parquet") {
        return Err(StorageError::Unsupported(format!(
            "file format {}",
            file.data.file_format
        )));
    }

    let parquet_file = File::open(local_filesystem_path(&file.resolved_path)?)
        .map_err(|err| StorageError::Backend(err.to_string()))?;
    let builder = ParquetRecordBatchReaderBuilder::try_new(parquet_file)
        .map_err(|err| StorageError::Backend(err.to_string()))?;
    let enum_paths = parquet_enum_paths(builder.parquet_schema());

    physical_schema_from_arrow_schema_with_overrides(builder.schema().as_ref(), file, &enum_paths)
}

pub fn physical_schema_from_arrow_schema(
    schema: &arrow_schema::Schema,
    file: &ResolvedFile,
) -> Result<FilePhysicalSchema> {
    physical_schema_from_arrow_schema_with_overrides(schema, file, &HashSet::new())
}

pub(crate) fn physical_schema_from_arrow_schema_with_overrides(
    schema: &arrow_schema::Schema,
    file: &ResolvedFile,
    enum_paths: &HashSet<Vec<String>>,
) -> Result<FilePhysicalSchema> {
    if let Some(mapping) = &file.mapping {
        if mapping.mapping_type == LEGACY_NAME_FALLBACK_MAPPING_TYPE {
            match collect_physical_schema_by_field_id(schema, enum_paths) {
                Ok(schema) => return Ok(schema),
                Err(err) if missing_parquet_field_id_error(&err) => {}
                Err(err) => return Err(err),
            }
        }
        let mut by_field_id = HashMap::new();
        collect_mapped_physical_schema(
            schema,
            &mapping.by_parent_and_name,
            &mut by_field_id,
            enum_paths,
        )?;
        return Ok(FilePhysicalSchema { by_field_id });
    } else if let Some(mapping_id) = file.data.mapping_id {
        return Err(StorageError::Backend(format!(
            "mapping {} was not resolved for {}",
            mapping_id.0, file.resolved_path
        )));
    }

    collect_physical_schema_by_field_id(schema, enum_paths)
}

pub(crate) fn reject_unsupported_files(files: &[ResolvedFile]) -> Result<()> {
    if let Some(file) = files
        .iter()
        .find(|file| !file.data.file_format.eq_ignore_ascii_case("parquet"))
    {
        return Err(StorageError::Unsupported(format!(
            "DuckLake data file {} has unsupported file format {}",
            file.data.data_file_id.0, file.data.file_format
        )));
    }

    if let Some(file) = files.iter().find(|file| file.data.encryption_key.is_some()) {
        return Err(StorageError::Unsupported(format!(
            "Encrypted DuckLake data file {} is not supported",
            file.data.data_file_id.0
        )));
    }

    if let Some((delete, format)) = files
        .iter()
        .filter_map(|file| file.delete.as_ref())
        .find_map(|delete| match &delete.format {
            DeleteFormat::Unknown(format) => Some((delete, format.as_str())),
            _ => None,
        })
    {
        return Err(StorageError::Unsupported(format!(
            "DuckLake delete file {} has unsupported format {}",
            delete.delete_file_id.0, format
        )));
    }

    if let Some(delete) = files
        .iter()
        .filter_map(|file| file.delete.as_ref())
        .find(|delete| delete.encryption_key.is_some())
    {
        return Err(StorageError::Unsupported(format!(
            "Encrypted DuckLake delete file {} is not supported",
            delete.delete_file_id.0
        )));
    }

    Ok(())
}

fn collect_physical_schema_by_field_id(
    schema: &arrow_schema::Schema,
    enum_paths: &HashSet<Vec<String>>,
) -> Result<FilePhysicalSchema> {
    let mut by_field_id = HashMap::new();
    for field in schema.fields() {
        collect_physical_field(field, &mut by_field_id, &mut Vec::new(), enum_paths)?;
    }
    Ok(FilePhysicalSchema { by_field_id })
}

fn collect_physical_field(
    field: &Field,
    out: &mut HashMap<ColumnId, FilePhysicalColumn>,
    path: &mut Vec<String>,
    enum_paths: &HashSet<Vec<String>>,
) -> Result<()> {
    path.push(field.name().clone());
    let Some(id) = field
        .metadata()
        .get(PARQUET_FIELD_ID_META_KEY)
        .and_then(|id| id.parse::<i64>().ok())
    else {
        path.pop();
        return Err(StorageError::Unsupported(format!(
            "Parquet field {} has no field id",
            field.name()
        )));
    };

    let field_id = ColumnId(id);
    out.insert(
        field_id,
        FilePhysicalColumn {
            field_id,
            physical_name: field.name().clone(),
            physical_path: path.clone(),
            physical_type: arrow_type_to_ducklake_with_overrides(
                field.data_type(),
                path,
                enum_paths,
            )?,
        },
    );

    collect_physical_children(field.data_type(), out, path, enum_paths)?;

    path.pop();
    Ok(())
}

fn collect_mapped_physical_schema(
    schema: &arrow_schema::Schema,
    mapping: &HashMap<(Option<ColumnId>, String), ResolvedMappingTarget>,
    by_field_id: &mut HashMap<ColumnId, FilePhysicalColumn>,
    enum_paths: &HashSet<Vec<String>>,
) -> Result<()> {
    for field in schema.fields() {
        collect_mapped_physical_field(
            field,
            mapping,
            by_field_id,
            &mut Vec::new(),
            None,
            enum_paths,
        )?;
    }
    Ok(())
}

fn collect_mapped_physical_field(
    field: &Field,
    mapping: &HashMap<(Option<ColumnId>, String), ResolvedMappingTarget>,
    out: &mut HashMap<ColumnId, FilePhysicalColumn>,
    path: &mut Vec<String>,
    parent_column: Option<ColumnId>,
    enum_paths: &HashSet<Vec<String>>,
) -> Result<()> {
    collect_mapped_physical_field_with_source_names(
        field,
        mapping,
        out,
        path,
        parent_column,
        &[],
        enum_paths,
    )
}

fn collect_mapped_physical_field_with_source_names(
    field: &Field,
    mapping: &HashMap<(Option<ColumnId>, String), ResolvedMappingTarget>,
    out: &mut HashMap<ColumnId, FilePhysicalColumn>,
    path: &mut Vec<String>,
    parent_column: Option<ColumnId>,
    source_name_fallbacks: &[&str],
    enum_paths: &HashSet<Vec<String>>,
) -> Result<()> {
    path.push(field.name().clone());
    let mut current = mapping.get(&(parent_column, field.name().clone())).copied();
    for source_name in source_name_fallbacks {
        if current.is_some() {
            break;
        }
        current = mapping
            .get(&(parent_column, (*source_name).into()))
            .copied();
    }
    if let Some(target) = current {
        let field_id = target.target_field_id;
        out.insert(
            field_id,
            FilePhysicalColumn {
                field_id,
                physical_name: field.name().clone(),
                physical_path: path.clone(),
                physical_type: arrow_type_to_ducklake_with_overrides(
                    field.data_type(),
                    path,
                    enum_paths,
                )?,
            },
        );
    }

    collect_mapped_physical_children(
        field.data_type(),
        mapping,
        out,
        path,
        current.map(|target| target.mapping_column_id),
        enum_paths,
    )?;

    path.pop();
    Ok(())
}

fn collect_physical_children(
    data_type: &DataType,
    out: &mut HashMap<ColumnId, FilePhysicalColumn>,
    path: &mut Vec<String>,
    enum_paths: &HashSet<Vec<String>>,
) -> Result<()> {
    match data_type {
        DataType::Struct(fields) => {
            for child in fields {
                collect_physical_field(child, out, path, enum_paths)?;
            }
        }
        DataType::List(field)
        | DataType::LargeList(field)
        | DataType::ListView(field)
        | DataType::LargeListView(field) => {
            collect_physical_field(field, out, path, enum_paths)?;
        }
        DataType::Map(entries, _) => {
            if let DataType::Struct(fields) = entries.data_type() {
                for child in fields {
                    collect_physical_field(child, out, path, enum_paths)?;
                }
            }
        }
        _ => {}
    }
    Ok(())
}

fn collect_mapped_physical_children(
    data_type: &DataType,
    mapping: &HashMap<(Option<ColumnId>, String), ResolvedMappingTarget>,
    out: &mut HashMap<ColumnId, FilePhysicalColumn>,
    path: &mut Vec<String>,
    parent_column: Option<ColumnId>,
    enum_paths: &HashSet<Vec<String>>,
) -> Result<()> {
    match data_type {
        DataType::Struct(fields) => {
            for child in fields {
                collect_mapped_physical_field(
                    child,
                    mapping,
                    out,
                    path,
                    parent_column,
                    enum_paths,
                )?;
            }
        }
        DataType::List(field)
        | DataType::LargeList(field)
        | DataType::ListView(field)
        | DataType::LargeListView(field) => {
            collect_mapped_physical_field_with_source_names(
                field,
                mapping,
                out,
                path,
                parent_column,
                &["list"],
                enum_paths,
            )?;
        }
        DataType::Map(entries, _) => {
            if let DataType::Struct(fields) = entries.data_type() {
                for child in fields {
                    collect_mapped_physical_field(
                        child,
                        mapping,
                        out,
                        path,
                        parent_column,
                        enum_paths,
                    )?;
                }
            }
        }
        _ => {}
    }
    Ok(())
}

fn parquet_enum_paths(schema: &SchemaDescriptor) -> HashSet<Vec<String>> {
    schema
        .columns()
        .iter()
        .filter(|column| parquet_column_is_enum(column.self_type()))
        .map(|column| column.path().parts().to_vec())
        .collect()
}

fn parquet_column_is_enum(column_type: &ParquetSchemaType) -> bool {
    let basic = column_type.get_basic_info();
    matches!(basic.logical_type_ref(), Some(LogicalType::Enum))
        || basic.converted_type() == ConvertedType::ENUM
}

fn arrow_type_to_ducklake_with_overrides(
    data_type: &DataType,
    physical_path: &[String],
    enum_paths: &HashSet<Vec<String>>,
) -> Result<String> {
    if enum_paths.contains(physical_path) {
        Ok("varchar".to_string())
    } else {
        arrow_type_to_ducklake(data_type)
    }
}

pub fn create_table_columns_from_arrow_schema(
    schema: &arrow_schema::Schema,
) -> Result<Vec<CreateTableColumn>> {
    let mut columns = Vec::new();
    let mut next_column_id = 1_i64;
    for field in schema.fields() {
        collect_create_table_column(field, None, &mut next_column_id, &mut columns)?;
    }
    Ok(columns)
}

fn collect_create_table_column(
    field: &Field,
    parent_column: Option<ducklake_catalog::ColumnId>,
    next_column_id: &mut i64,
    columns: &mut Vec<CreateTableColumn>,
) -> Result<()> {
    let column_id = ducklake_catalog::ColumnId(*next_column_id);
    *next_column_id = next_column_id
        .checked_add(1)
        .ok_or_else(|| StorageError::Backend("DuckLake column id overflow".into()))?;
    columns.push(CreateTableColumn {
        column_id,
        column_order: column_id.0,
        column_name: field.name().clone(),
        column_type: arrow_type_to_ducklake(field.data_type())?,
        nulls_allowed: field.is_nullable(),
        parent_column,
    });

    match field.data_type() {
        DataType::Struct(fields) => {
            for child in fields {
                collect_create_table_column(child, Some(column_id), next_column_id, columns)?;
            }
        }
        DataType::List(child)
        | DataType::LargeList(child)
        | DataType::ListView(child)
        | DataType::LargeListView(child) => {
            collect_create_table_column(child, Some(column_id), next_column_id, columns)?;
        }
        DataType::Map(entries, _) => {
            let DataType::Struct(fields) = entries.data_type() else {
                return Err(StorageError::Unsupported(
                    "Arrow map without struct entries".into(),
                ));
            };
            for child in fields {
                collect_create_table_column(child, Some(column_id), next_column_id, columns)?;
            }
        }
        _ => {}
    }

    Ok(())
}

fn arrow_type_to_ducklake(data_type: &DataType) -> Result<String> {
    let ty = match data_type {
        DataType::Boolean => "boolean",
        DataType::Int8 => "int8",
        DataType::Int16 => "int16",
        DataType::Int32 => "int32",
        DataType::Int64 => "int64",
        DataType::UInt8 => "uint8",
        DataType::UInt16 => "uint16",
        DataType::UInt32 => "uint32",
        DataType::UInt64 => "uint64",
        DataType::Float32 => "float32",
        DataType::Float64 => "float64",
        DataType::Decimal128(precision, scale) | DataType::Decimal256(precision, scale) => {
            return Ok(format!("decimal({precision},{scale})"));
        }
        DataType::Date32 => "date",
        DataType::Time32(_) | DataType::Time64(TimeUnit::Microsecond) => "time",
        DataType::Time64(TimeUnit::Nanosecond) => "time_ns",
        DataType::Timestamp(TimeUnit::Second, None) => "timestamp_s",
        DataType::Timestamp(TimeUnit::Millisecond, None) => "timestamp_ms",
        DataType::Timestamp(TimeUnit::Microsecond, None) => "timestamp",
        DataType::Timestamp(TimeUnit::Nanosecond, None) => "timestamp_ns",
        DataType::Timestamp(_, Some(_)) => "timestamptz",
        DataType::Interval(_) => "interval",
        DataType::Utf8 | DataType::LargeUtf8 | DataType::Utf8View => "varchar",
        DataType::Binary | DataType::LargeBinary | DataType::BinaryView => "blob",
        DataType::FixedSizeBinary(16) => "uuid",
        DataType::Struct(_) => "struct",
        DataType::List(_)
        | DataType::LargeList(_)
        | DataType::ListView(_)
        | DataType::LargeListView(_) => "list",
        DataType::Map(_, _) => "map",
        other => {
            return Err(StorageError::Unsupported(format!(
                "Arrow physical type {other:?}"
            )));
        }
    };
    Ok(ty.to_string())
}
