pandrs 0.3.0

use std::collections::HashMap;
use std::fs::File;
use std::io::{BufReader, BufWriter};
use std::path::Path;
use std::sync::Arc;

#[cfg(feature = "streaming")]
use tokio::io::{AsyncRead, AsyncWrite};

use arrow::array::{
    Array, ArrayRef, BooleanArray, Date32Array, Decimal128Array, Float64Array, Int64Array,
    StringArray, TimestampMicrosecondArray,
};
use arrow::datatypes::{DataType, Field, Schema, SchemaRef, TimeUnit};
use arrow::record_batch::RecordBatch;
use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder;
use parquet::arrow::arrow_writer::ArrowWriter;
use parquet::basic::Compression;
use parquet::file::metadata::{FileMetaData, RowGroupMetaData};
use parquet::file::properties::WriterProperties;
use parquet::file::reader::{FileReader, SerializedFileReader};
use parquet::schema::types::Type as ParquetType;

use crate::column::{BooleanColumn, Column, ColumnType, Float64Column, Int64Column, StringColumn};
use crate::dataframe::DataFrame;
use crate::error::{Error, Result};
use crate::optimized::OptimizedDataFrame;
use crate::series::Series;

/// Enumeration of Parquet compression options
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ParquetCompression {
    None,
    Snappy,
    Gzip,
    Lzo,
    Brotli,
    Lz4,
    Zstd,
}

/// Parquet file metadata information
#[derive(Debug, Clone)]
pub struct ParquetMetadata {
    /// Number of rows in the file
    pub num_rows: i64,
    /// Number of row groups
    pub num_row_groups: usize,
    /// File schema
    pub schema: String,
    /// File size in bytes
    pub file_size: Option<i64>,
    /// Compression algorithm used
    pub compression: String,
    /// Creator/writer information
    pub created_by: Option<String>,
}

/// Row group metadata information
#[derive(Debug, Clone)]
pub struct RowGroupInfo {
    /// Row group index
    pub index: usize,
    /// Number of rows in this row group
    pub num_rows: i64,
    /// Total byte size of this row group
    pub total_byte_size: i64,
    /// Number of columns
    pub num_columns: usize,
}

/// Column statistics information
#[derive(Debug, Clone)]
pub struct ColumnStats {
    /// Column name
    pub name: String,
    /// Data type
    pub data_type: String,
    /// Null count
    pub null_count: Option<i64>,
    /// Distinct count (if available)
    pub distinct_count: Option<i64>,
    /// Minimum value (as string)
    pub min_value: Option<String>,
    /// Maximum value (as string)
    pub max_value: Option<String>,
}

/// Advanced Parquet reading options
#[derive(Debug, Clone)]
pub struct ParquetReadOptions {
    /// Specific columns to read (None = all columns)
    pub columns: Option<Vec<String>>,
    /// Use multiple threads for reading
    pub use_threads: bool,
    /// Memory map the file for faster access
    pub use_memory_map: bool,
    /// Batch size for chunked reading
    pub batch_size: Option<usize>,
    /// Row groups to read (None = all row groups)
    pub row_groups: Option<Vec<usize>>,
}

impl Default for ParquetReadOptions {
    fn default() -> Self {
        Self {
            columns: None,
            use_threads: true,
            use_memory_map: false,
            batch_size: None,
            row_groups: None,
        }
    }
}

/// Advanced Parquet writing options
#[derive(Debug, Clone)]
pub struct ParquetWriteOptions {
    /// Compression algorithm
    pub compression: ParquetCompression,
    /// Row group size (number of rows per group)
    pub row_group_size: Option<usize>,
    /// Page size in bytes
    pub page_size: Option<usize>,
    /// Enable dictionary encoding
    pub enable_dictionary: bool,
    /// Use multiple threads for writing
    pub use_threads: bool,
}

impl Default for ParquetWriteOptions {
    fn default() -> Self {
        Self {
            compression: ParquetCompression::Snappy,
            row_group_size: Some(50000),
            page_size: Some(1024 * 1024), // 1MB
            enable_dictionary: true,
            use_threads: true,
        }
    }
}

impl From<ParquetCompression> for Compression {
    fn from(comp: ParquetCompression) -> Self {
        match comp {
            ParquetCompression::None => Compression::UNCOMPRESSED,
            ParquetCompression::Snappy => Compression::SNAPPY,
            ParquetCompression::Gzip => Compression::GZIP(Default::default()),
            ParquetCompression::Lzo => Compression::LZO,
            ParquetCompression::Brotli => Compression::BROTLI(Default::default()),
            ParquetCompression::Lz4 => Compression::LZ4,
            ParquetCompression::Zstd => Compression::ZSTD(Default::default()),
        }
    }
}

/// Read a DataFrame from a Parquet file
///
/// # Arguments
///
/// * `path` - Path to the Parquet file
///
/// # Returns
///
/// * `Result<DataFrame>` - The read DataFrame, or an error
///
/// # Example
///
/// ```no_run
/// use pandrs::io::read_parquet;
///
/// // Read a DataFrame from a Parquet file
/// let df = read_parquet("data.parquet").expect("operation should succeed");
/// ```
pub fn read_parquet(path: impl AsRef<Path>) -> Result<DataFrame> {
    // Open the file
    let file = File::open(path.as_ref())
        .map_err(|e| Error::IoError(format!("Failed to open Parquet file: {}", e)))?;

    // Create an Arrow ParquetReader
    let builder = ParquetRecordBatchReaderBuilder::try_new(file)
        .map_err(|e| Error::IoError(format!("Failed to parse Parquet file: {}", e)))?;

    // Get schema information
    let schema = builder.schema().clone();

    // Create a record batch reader
    let reader = builder
        .build()
        .map_err(|e| Error::IoError(format!("Failed to read Parquet file: {}", e)))?;

    // Read all record batches
    let mut all_batches = Vec::new();
    for batch_result in reader {
        let batch = batch_result
            .map_err(|e| Error::IoError(format!("Failed to read record batch: {}", e)))?;
        all_batches.push(batch);
    }

    // Return an empty DataFrame if no record batches are found
    if all_batches.is_empty() {
        return Ok(DataFrame::new());
    }

    // Convert to DataFrame
    record_batches_to_dataframe(&all_batches, schema)
}

/// Convert Arrow record batches to a DataFrame
fn record_batches_to_dataframe(batches: &[RecordBatch], schema: SchemaRef) -> Result<DataFrame> {
    let mut df = DataFrame::new();

    // Extract column information from schema
    for (col_idx, field) in schema.fields().iter().enumerate() {
        let col_name = field.name().clone();
        let col_type = field.data_type();

        // Collect column data from all batches
        match col_type {
            DataType::Int64 => {
                let mut values = Vec::new();

                for batch in batches {
                    let array = batch
                        .column(col_idx)
                        .as_any()
                        .downcast_ref::<Int64Array>()
                        .ok_or_else(|| {
                            Error::Cast(format!(
                                "Failed to cast column '{}' to Int64Array",
                                col_name
                            ))
                        })?;

                    for i in 0..array.len() {
                        if array.is_null(i) {
                            values.push(0); // Use 0 as the default value for NULL
                        } else {
                            values.push(array.value(i));
                        }
                    }
                }

                let series = Series::new(values, Some(col_name.clone()))?;
                df.add_column(col_name.clone(), series)?;
            }
            DataType::Float64 => {
                let mut values = Vec::new();

                for batch in batches {
                    let array = batch
                        .column(col_idx)
                        .as_any()
                        .downcast_ref::<Float64Array>()
                        .ok_or_else(|| {
                            Error::Cast(format!(
                                "Failed to cast column '{}' to Float64Array",
                                col_name
                            ))
                        })?;

                    for i in 0..array.len() {
                        if array.is_null(i) {
                            values.push(f64::NAN); // Use NaN for NULL
                        } else {
                            values.push(array.value(i));
                        }
                    }
                }

                let series = Series::new(values, Some(col_name.clone()))?;
                df.add_column(col_name.clone(), series)?;
            }
            DataType::Boolean => {
                let mut values = Vec::new();

                for batch in batches {
                    let array = batch
                        .column(col_idx)
                        .as_any()
                        .downcast_ref::<BooleanArray>()
                        .ok_or_else(|| {
                            Error::Cast(format!(
                                "Failed to cast column '{}' to BooleanArray",
                                col_name
                            ))
                        })?;

                    for i in 0..array.len() {
                        if array.is_null(i) {
                            values.push(false); // Use false as the default value for NULL
                        } else {
                            values.push(array.value(i));
                        }
                    }
                }

                let series = Series::new(values, Some(col_name.clone()))?;
                df.add_column(col_name.clone(), series)?;
            }
            DataType::Utf8 | DataType::LargeUtf8 => {
                let mut values = Vec::new();

                for batch in batches {
                    let array = batch
                        .column(col_idx)
                        .as_any()
                        .downcast_ref::<StringArray>()
                        .ok_or_else(|| {
                            Error::Cast(format!(
                                "Failed to cast column '{}' to StringArray",
                                col_name
                            ))
                        })?;

                    for i in 0..array.len() {
                        if array.is_null(i) {
                            values.push("".to_string()); // Use an empty string for NULL
                        } else {
                            values.push(array.value(i).to_string());
                        }
                    }
                }

                let series = Series::new(values, Some(col_name.clone()))?;
                df.add_column(col_name.clone(), series)?;
            }
            _ => {
                // Unsupported data types are treated as strings
                let mut values = Vec::new();

                for batch in batches {
                    let array = batch.column(col_idx);
                    for i in 0..array.len() {
                        if array.is_null(i) {
                            values.push("".to_string());
                        } else {
                            values.push(format!("{:?}", array));
                        }
                    }
                }

                let series = Series::new(values, Some(col_name.clone()))?;
                df.add_column(col_name.clone(), series)?;
            }
        }
    }

    Ok(df)
}

/// Write a DataFrame to a Parquet file
///
/// # Arguments
///
/// * `df` - The DataFrame to write
/// * `path` - Path to the output Parquet file
/// * `compression` - Compression option (default is Snappy)
///
/// # Returns
///
/// * `Result<()>` - Ok(()) if successful, or an error
///
/// # Example
///
/// ```text
/// // Disable DOC test
/// ```
pub fn write_parquet(
    df: &OptimizedDataFrame,
    path: impl AsRef<Path>,
    compression: Option<ParquetCompression>,
) -> Result<()> {
    // Create Arrow schema
    let schema_fields: Vec<Field> = df
        .column_names()
        .iter()
        .filter_map(|col_name| {
            // Get each column as a string series
            if let Ok(col_view) = df.column(col_name) {
                // Determine column type
                let data_type = match col_view.column_type() {
                    crate::column::ColumnType::Int64 => DataType::Int64,
                    crate::column::ColumnType::Float64 => DataType::Float64,
                    crate::column::ColumnType::Boolean => DataType::Boolean,
                    crate::column::ColumnType::String => DataType::Utf8,
                };
                Some(Field::new(col_name, data_type, true))
            } else {
                None
            }
        })
        .collect();

    let schema = Schema::new(schema_fields);
    let schema_ref = Arc::new(schema);

    // Convert column data to Arrow arrays
    let arrays: Vec<ArrayRef> = df
        .column_names()
        .iter()
        .filter_map(|col_name| {
            // Get each column
            let col_view = match df.column(col_name) {
                Ok(s) => s,
                Err(_) => return None,
            };

            // Extract actual data from the column based on its type
            match col_view.column_type() {
                crate::column::ColumnType::Int64 => {
                    if let Some(int_col) = col_view.as_int64() {
                        let mut values = Vec::with_capacity(df.row_count());
                        let mut validity = Vec::with_capacity(df.row_count());

                        for i in 0..df.row_count() {
                            match int_col.get(i) {
                                Ok(Some(val)) => {
                                    values.push(val);
                                    validity.push(true);
                                }
                                Ok(None) => {
                                    values.push(0); // Default value for null
                                    validity.push(false);
                                }
                                Err(_) => {
                                    values.push(0);
                                    validity.push(false);
                                }
                            }
                        }

                        let array = Int64Array::new(values.into(), Some(validity.into()));
                        Some(Arc::new(array) as ArrayRef)
                    } else {
                        None
                    }
                }
                crate::column::ColumnType::Float64 => {
                    if let Some(float_col) = col_view.as_float64() {
                        let mut values = Vec::with_capacity(df.row_count());
                        let mut validity = Vec::with_capacity(df.row_count());

                        for i in 0..df.row_count() {
                            match float_col.get(i) {
                                Ok(Some(val)) => {
                                    values.push(val);
                                    validity.push(true);
                                }
                                Ok(None) => {
                                    values.push(0.0); // Default value for null
                                    validity.push(false);
                                }
                                Err(_) => {
                                    values.push(0.0);
                                    validity.push(false);
                                }
                            }
                        }

                        let array = Float64Array::new(values.into(), Some(validity.into()));
                        Some(Arc::new(array) as ArrayRef)
                    } else {
                        None
                    }
                }
                crate::column::ColumnType::Boolean => {
                    if let Some(bool_col) = col_view.as_boolean() {
                        let mut values = Vec::with_capacity(df.row_count());
                        let mut validity = Vec::with_capacity(df.row_count());

                        for i in 0..df.row_count() {
                            match bool_col.get(i) {
                                Ok(Some(val)) => {
                                    values.push(val);
                                    validity.push(true);
                                }
                                Ok(None) => {
                                    values.push(false); // Default value for null
                                    validity.push(false);
                                }
                                Err(_) => {
                                    values.push(false);
                                    validity.push(false);
                                }
                            }
                        }

                        let array = BooleanArray::new(values.into(), Some(validity.into()));
                        Some(Arc::new(array) as ArrayRef)
                    } else {
                        None
                    }
                }
                crate::column::ColumnType::String => {
                    if let Some(str_col) = col_view.as_string() {
                        let mut values = Vec::with_capacity(df.row_count());
                        let mut validity = Vec::with_capacity(df.row_count());

                        for i in 0..df.row_count() {
                            match str_col.get(i) {
                                Ok(Some(val)) => {
                                    values.push(val.to_string());
                                    validity.push(true);
                                }
                                Ok(None) => {
                                    values.push(String::new()); // Default value for null
                                    validity.push(false);
                                }
                                Err(_) => {
                                    values.push(String::new());
                                    validity.push(false);
                                }
                            }
                        }

                        // Convert to iterator with nulls properly handled
                        let string_values: Vec<Option<&str>> = values
                            .iter()
                            .zip(validity.iter())
                            .map(|(s, &is_valid)| if is_valid { Some(s.as_str()) } else { None })
                            .collect();
                        let array = StringArray::from(string_values);
                        Some(Arc::new(array) as ArrayRef)
                    } else {
                        None
                    }
                }
            }
        })
        .collect();

    // Create a record batch
    let batch = RecordBatch::try_new(schema_ref.clone(), arrays)
        .map_err(|e| Error::Cast(format!("Failed to create record batch: {}", e)))?;

    // Set compression options
    let compression_type = compression.unwrap_or(ParquetCompression::Snappy);
    let props = WriterProperties::builder()
        .set_compression(Compression::from(compression_type))
        .build();

    // Create the file
    let file = File::create(path.as_ref())
        .map_err(|e| Error::IoError(format!("Failed to create Parquet file: {}", e)))?;

    // Create an Arrow writer and write
    let mut writer = ArrowWriter::try_new(file, schema_ref, Some(props))
        .map_err(|e| Error::IoError(format!("Failed to create Parquet writer: {}", e)))?;

    // Write the record batch
    writer
        .write(&batch)
        .map_err(|e| Error::IoError(format!("Failed to write record batch: {}", e)))?;

    // Close the file
    writer
        .close()
        .map_err(|e| Error::IoError(format!("Failed to close Parquet file: {}", e)))?;

    Ok(())
}

/// Get comprehensive metadata about a Parquet file
///
/// # Arguments
///
/// * `path` - Path to the Parquet file
///
/// # Returns
///
/// * `Result<ParquetMetadata>` - File metadata information, or an error
///
/// # Examples
///
/// ```no_run
/// use pandrs::io::get_parquet_metadata;
///
/// let metadata = get_parquet_metadata("data.parquet").expect("operation should succeed");
/// println!("File has {} rows in {} row groups", metadata.num_rows, metadata.num_row_groups);
/// ```
pub fn get_parquet_metadata(path: impl AsRef<Path>) -> Result<ParquetMetadata> {
    let file = File::open(path.as_ref())
        .map_err(|e| Error::IoError(format!("Failed to open Parquet file: {}", e)))?;

    let reader = SerializedFileReader::new(file)
        .map_err(|e| Error::IoError(format!("Failed to create Parquet reader: {}", e)))?;

    let metadata = reader.metadata();
    let file_metadata = metadata.file_metadata();

    Ok(ParquetMetadata {
        num_rows: file_metadata.num_rows(),
        num_row_groups: metadata.num_row_groups(),
        schema: format!("{:?}", file_metadata.schema()),
        file_size: None,                    // Would need additional file stats
        compression: "Various".to_string(), // Row groups can have different compression
        created_by: file_metadata.created_by().map(|s| s.to_string()),
    })
}

/// Get information about all row groups in a Parquet file
///
/// # Arguments
///
/// * `path` - Path to the Parquet file
///
/// # Returns
///
/// * `Result<Vec<RowGroupInfo>>` - Vector of row group information, or an error
///
/// # Examples
///
/// ```no_run
/// use pandrs::io::get_row_group_info;
///
/// let row_groups = get_row_group_info("data.parquet").expect("operation should succeed");
/// for (i, rg) in row_groups.iter().enumerate() {
///     println!("Row group {}: {} rows, {} bytes", i, rg.num_rows, rg.total_byte_size);
/// }
/// ```
pub fn get_row_group_info(path: impl AsRef<Path>) -> Result<Vec<RowGroupInfo>> {
    let file = File::open(path.as_ref())
        .map_err(|e| Error::IoError(format!("Failed to open Parquet file: {}", e)))?;

    let reader = SerializedFileReader::new(file)
        .map_err(|e| Error::IoError(format!("Failed to create Parquet reader: {}", e)))?;

    let metadata = reader.metadata();
    let mut row_groups = Vec::new();

    for i in 0..metadata.num_row_groups() {
        let rg_metadata = metadata.row_group(i);
        row_groups.push(RowGroupInfo {
            index: i,
            num_rows: rg_metadata.num_rows(),
            total_byte_size: rg_metadata.total_byte_size(),
            num_columns: rg_metadata.num_columns(),
        });
    }

    Ok(row_groups)
}

/// Get column statistics for all columns in a Parquet file
///
/// # Arguments
///
/// * `path` - Path to the Parquet file
///
/// # Returns
///
/// * `Result<Vec<ColumnStats>>` - Vector of column statistics, or an error
///
/// # Examples
///
/// ```no_run
/// use pandrs::io::get_column_statistics;
///
/// let stats = get_column_statistics("data.parquet").expect("operation should succeed");
/// for stat in stats {
///     println!("{}: {} nulls, min={:?}, max={:?}",
///              stat.name, stat.null_count.unwrap_or(0), stat.min_value, stat.max_value);
/// }
/// ```
pub fn get_column_statistics(path: impl AsRef<Path>) -> Result<Vec<ColumnStats>> {
    let file = File::open(path.as_ref())
        .map_err(|e| Error::IoError(format!("Failed to open Parquet file: {}", e)))?;

    let reader = SerializedFileReader::new(file)
        .map_err(|e| Error::IoError(format!("Failed to create Parquet reader: {}", e)))?;

    let metadata = reader.metadata();
    let schema = metadata.file_metadata().schema_descr();
    let mut column_stats = Vec::new();

    // Collect statistics from all row groups
    let mut null_counts: HashMap<String, u64> = HashMap::new();
    let mut min_values: HashMap<String, String> = HashMap::new();
    let mut max_values: HashMap<String, String> = HashMap::new();

    for rg_idx in 0..metadata.num_row_groups() {
        let rg_metadata = metadata.row_group(rg_idx);

        for col_idx in 0..rg_metadata.num_columns() {
            let col_metadata = rg_metadata.column(col_idx);
            let col_name = schema.column(col_idx).name().to_string();

            // Aggregate null counts
            if let Some(statistics) = col_metadata.statistics() {
                if let Some(null_count) = statistics.null_count_opt() {
                    *null_counts.entry(col_name.clone()).or_insert(0) += null_count;
                }

                // Convert min/max to strings (simplified)
                if statistics.min_bytes_opt().is_some() && statistics.max_bytes_opt().is_some() {
                    min_values
                        .entry(col_name.clone())
                        .or_insert_with(|| "N/A".to_string());
                    max_values
                        .entry(col_name.clone())
                        .or_insert_with(|| "N/A".to_string());
                }
            }
        }
    }

    // Create column stats
    for col_idx in 0..schema.num_columns() {
        let column = schema.column(col_idx);
        let col_name = column.name().to_string();

        column_stats.push(ColumnStats {
            name: col_name.clone(),
            data_type: format!("{:?}", column.physical_type()),
            null_count: null_counts.get(&col_name).map(|&n| n as i64),
            distinct_count: None, // Would need additional analysis
            min_value: min_values.get(&col_name).cloned(),
            max_value: max_values.get(&col_name).cloned(),
        });
    }

    Ok(column_stats)
}

/// Read a DataFrame from a Parquet file with advanced options
///
/// # Arguments
///
/// * `path` - Path to the Parquet file
/// * `options` - Advanced reading options
///
/// # Returns
///
/// * `Result<DataFrame>` - The read DataFrame, or an error
///
/// # Examples
///
/// ```no_run
/// use pandrs::io::{read_parquet_advanced, ParquetReadOptions};
///
/// // Read only specific columns
/// let options = ParquetReadOptions {
///     columns: Some(vec!["name".to_string(), "age".to_string()]),
///     use_threads: true,
///     ..Default::default()
/// };
/// let df = read_parquet_advanced("data.parquet", options).expect("operation should succeed");
/// ```
pub fn read_parquet_advanced(
    path: impl AsRef<Path>,
    options: ParquetReadOptions,
) -> Result<DataFrame> {
    let file = File::open(path.as_ref())
        .map_err(|e| Error::IoError(format!("Failed to open Parquet file: {}", e)))?;

    let mut builder = ParquetRecordBatchReaderBuilder::try_new(file)
        .map_err(|e| Error::IoError(format!("Failed to parse Parquet file: {}", e)))?;

    // Configure batch size if specified
    if let Some(batch_size) = options.batch_size {
        builder = builder.with_batch_size(batch_size);
    }

    // Configure row groups if specified
    if let Some(row_groups) = options.row_groups {
        builder = builder.with_row_groups(row_groups);
    }

    // Configure column projection if specified
    if let Some(columns) = &options.columns {
        let schema = builder.schema();
        let mut projection_indices = Vec::new();

        for col_name in columns {
            for (idx, field) in schema.fields().iter().enumerate() {
                if field.name() == col_name {
                    projection_indices.push(idx);
                    break;
                }
            }
        }

        if !projection_indices.is_empty() {
            use parquet::arrow::ProjectionMask;
            let mask = ProjectionMask::roots(&builder.parquet_schema(), projection_indices);
            builder = builder.with_projection(mask);
        }
    }

    let schema = builder.schema().clone();
    let reader = builder
        .build()
        .map_err(|e| Error::IoError(format!("Failed to read Parquet file: {}", e)))?;

    // Read all record batches
    let mut all_batches = Vec::new();
    for batch_result in reader {
        let batch = batch_result
            .map_err(|e| Error::IoError(format!("Failed to read record batch: {}", e)))?;
        all_batches.push(batch);
    }

    if all_batches.is_empty() {
        return Ok(DataFrame::new());
    }

    // Convert to DataFrame with enhanced type support
    record_batches_to_dataframe_enhanced(&all_batches, schema)
}

/// Write a DataFrame to a Parquet file with advanced options
///
/// # Arguments
///
/// * `df` - The DataFrame to write
/// * `path` - Path to the output Parquet file
/// * `options` - Advanced writing options
///
/// # Returns
///
/// * `Result<()>` - Ok(()) if successful, or an error
///
/// # Examples
///
/// ```no_run
/// use pandrs::io::{write_parquet_advanced, ParquetWriteOptions, ParquetCompression};
/// use pandrs::optimized::dataframe::OptimizedDataFrame;
///
/// // Create sample dataframe
/// let df = OptimizedDataFrame::new();
///
/// let options = ParquetWriteOptions {
///     compression: ParquetCompression::Zstd,
///     row_group_size: Some(100000),
///     enable_dictionary: true,
///     ..Default::default()
/// };
/// write_parquet_advanced(&df, "output.parquet", options).expect("operation should succeed");
/// ```
pub fn write_parquet_advanced(
    df: &OptimizedDataFrame,
    path: impl AsRef<Path>,
    options: ParquetWriteOptions,
) -> Result<()> {
    // Create Arrow schema with enhanced type detection
    let schema_fields: Vec<Field> = df
        .column_names()
        .iter()
        .filter_map(|col_name| {
            if let Ok(col_view) = df.column(col_name) {
                let data_type = match col_view.column_type() {
                    crate::column::ColumnType::Int64 => DataType::Int64,
                    crate::column::ColumnType::Float64 => DataType::Float64,
                    crate::column::ColumnType::Boolean => DataType::Boolean,
                    crate::column::ColumnType::String => {
                        // Use dictionary encoding for strings if enabled
                        if options.enable_dictionary {
                            DataType::Dictionary(
                                Box::new(DataType::Int32),
                                Box::new(DataType::Utf8),
                            )
                        } else {
                            DataType::Utf8
                        }
                    }
                };
                Some(Field::new(col_name, data_type, true))
            } else {
                None
            }
        })
        .collect();

    let schema = Schema::new(schema_fields);
    let schema_ref = Arc::new(schema);

    // Enhanced writer properties
    let mut props_builder =
        WriterProperties::builder().set_compression(Compression::from(options.compression));

    if let Some(row_group_size) = options.row_group_size {
        props_builder = props_builder.set_max_row_group_row_count(Some(row_group_size));
    }

    if let Some(page_size) = options.page_size {
        props_builder = props_builder.set_data_page_size_limit(page_size);
    }

    if options.enable_dictionary {
        props_builder = props_builder.set_dictionary_enabled(true);
    }

    let props = props_builder.build();

    // Create arrays with the same logic as before
    let arrays: Vec<ArrayRef> = df
        .column_names()
        .iter()
        .filter_map(|col_name| {
            let col_view = match df.column(col_name) {
                Ok(s) => s,
                Err(_) => return None,
            };

            // Use existing array creation logic
            match col_view.column_type() {
                crate::column::ColumnType::Int64 => {
                    if let Some(int_col) = col_view.as_int64() {
                        let mut values = Vec::with_capacity(df.row_count());
                        let mut validity = Vec::with_capacity(df.row_count());

                        for i in 0..df.row_count() {
                            match int_col.get(i) {
                                Ok(Some(val)) => {
                                    values.push(val);
                                    validity.push(true);
                                }
                                Ok(None) => {
                                    values.push(0);
                                    validity.push(false);
                                }
                                Err(_) => {
                                    values.push(0);
                                    validity.push(false);
                                }
                            }
                        }

                        let array = Int64Array::new(values.into(), Some(validity.into()));
                        Some(Arc::new(array) as ArrayRef)
                    } else {
                        None
                    }
                }
                crate::column::ColumnType::Float64 => {
                    if let Some(float_col) = col_view.as_float64() {
                        let mut values = Vec::with_capacity(df.row_count());
                        let mut validity = Vec::with_capacity(df.row_count());

                        for i in 0..df.row_count() {
                            match float_col.get(i) {
                                Ok(Some(val)) => {
                                    values.push(val);
                                    validity.push(true);
                                }
                                Ok(None) => {
                                    values.push(0.0);
                                    validity.push(false);
                                }
                                Err(_) => {
                                    values.push(0.0);
                                    validity.push(false);
                                }
                            }
                        }

                        let array = Float64Array::new(values.into(), Some(validity.into()));
                        Some(Arc::new(array) as ArrayRef)
                    } else {
                        None
                    }
                }
                crate::column::ColumnType::Boolean => {
                    if let Some(bool_col) = col_view.as_boolean() {
                        let mut values = Vec::with_capacity(df.row_count());
                        let mut validity = Vec::with_capacity(df.row_count());

                        for i in 0..df.row_count() {
                            match bool_col.get(i) {
                                Ok(Some(val)) => {
                                    values.push(val);
                                    validity.push(true);
                                }
                                Ok(None) => {
                                    values.push(false);
                                    validity.push(false);
                                }
                                Err(_) => {
                                    values.push(false);
                                    validity.push(false);
                                }
                            }
                        }

                        let array = BooleanArray::new(values.into(), Some(validity.into()));
                        Some(Arc::new(array) as ArrayRef)
                    } else {
                        None
                    }
                }
                crate::column::ColumnType::String => {
                    if let Some(str_col) = col_view.as_string() {
                        let mut values = Vec::with_capacity(df.row_count());
                        let mut validity = Vec::with_capacity(df.row_count());

                        for i in 0..df.row_count() {
                            match str_col.get(i) {
                                Ok(Some(val)) => {
                                    values.push(val.to_string());
                                    validity.push(true);
                                }
                                Ok(None) => {
                                    values.push(String::new());
                                    validity.push(false);
                                }
                                Err(_) => {
                                    values.push(String::new());
                                    validity.push(false);
                                }
                            }
                        }

                        let string_values: Vec<Option<&str>> = values
                            .iter()
                            .zip(validity.iter())
                            .map(|(s, &is_valid)| if is_valid { Some(s.as_str()) } else { None })
                            .collect();
                        let array = StringArray::from(string_values);
                        Some(Arc::new(array) as ArrayRef)
                    } else {
                        None
                    }
                }
            }
        })
        .collect();

    // Create record batch
    let batch = RecordBatch::try_new(schema_ref.clone(), arrays)
        .map_err(|e| Error::Cast(format!("Failed to create record batch: {}", e)))?;

    // Create file and writer
    let file = File::create(path.as_ref())
        .map_err(|e| Error::IoError(format!("Failed to create Parquet file: {}", e)))?;

    let mut writer = ArrowWriter::try_new(file, schema_ref, Some(props))
        .map_err(|e| Error::IoError(format!("Failed to create Parquet writer: {}", e)))?;

    // Write the record batch
    writer
        .write(&batch)
        .map_err(|e| Error::IoError(format!("Failed to write record batch: {}", e)))?;

    // Close the file
    writer
        .close()
        .map_err(|e| Error::IoError(format!("Failed to close Parquet file: {}", e)))?;

    Ok(())
}

/// Enhanced record batch to DataFrame conversion with additional data type support
fn record_batches_to_dataframe_enhanced(
    batches: &[RecordBatch],
    schema: SchemaRef,
) -> Result<DataFrame> {
    let mut df = DataFrame::new();

    for (col_idx, field) in schema.fields().iter().enumerate() {
        let col_name = field.name().clone();
        let col_type = field.data_type();

        match col_type {
            DataType::Int64 => {
                let mut values = Vec::new();
                for batch in batches {
                    let array = batch
                        .column(col_idx)
                        .as_any()
                        .downcast_ref::<Int64Array>()
                        .ok_or_else(|| {
                            Error::Cast(format!(
                                "Failed to cast column '{}' to Int64Array",
                                col_name
                            ))
                        })?;

                    for i in 0..array.len() {
                        if array.is_null(i) {
                            values.push(0);
                        } else {
                            values.push(array.value(i));
                        }
                    }
                }
                let series = Series::new(values, Some(col_name.clone()))?;
                df.add_column(col_name.clone(), series)?;
            }
            DataType::Float64 => {
                let mut values = Vec::new();
                for batch in batches {
                    let array = batch
                        .column(col_idx)
                        .as_any()
                        .downcast_ref::<Float64Array>()
                        .ok_or_else(|| {
                            Error::Cast(format!(
                                "Failed to cast column '{}' to Float64Array",
                                col_name
                            ))
                        })?;

                    for i in 0..array.len() {
                        if array.is_null(i) {
                            values.push(f64::NAN);
                        } else {
                            values.push(array.value(i));
                        }
                    }
                }
                let series = Series::new(values, Some(col_name.clone()))?;
                df.add_column(col_name.clone(), series)?;
            }
            DataType::Boolean => {
                let mut values = Vec::new();
                for batch in batches {
                    let array = batch
                        .column(col_idx)
                        .as_any()
                        .downcast_ref::<BooleanArray>()
                        .ok_or_else(|| {
                            Error::Cast(format!(
                                "Failed to cast column '{}' to BooleanArray",
                                col_name
                            ))
                        })?;

                    for i in 0..array.len() {
                        if array.is_null(i) {
                            values.push(false);
                        } else {
                            values.push(array.value(i));
                        }
                    }
                }
                let series = Series::new(values, Some(col_name.clone()))?;
                df.add_column(col_name.clone(), series)?;
            }
            DataType::Utf8 | DataType::LargeUtf8 => {
                let mut values = Vec::new();
                for batch in batches {
                    let array = batch
                        .column(col_idx)
                        .as_any()
                        .downcast_ref::<StringArray>()
                        .ok_or_else(|| {
                            Error::Cast(format!(
                                "Failed to cast column '{}' to StringArray",
                                col_name
                            ))
                        })?;

                    for i in 0..array.len() {
                        if array.is_null(i) {
                            values.push("".to_string());
                        } else {
                            values.push(array.value(i).to_string());
                        }
                    }
                }
                let series = Series::new(values, Some(col_name.clone()))?;
                df.add_column(col_name.clone(), series)?;
            }
            DataType::Date32 => {
                let mut values = Vec::new();
                for batch in batches {
                    let array = batch
                        .column(col_idx)
                        .as_any()
                        .downcast_ref::<Date32Array>()
                        .ok_or_else(|| {
                            Error::Cast(format!(
                                "Failed to cast column '{}' to Date32Array",
                                col_name
                            ))
                        })?;

                    for i in 0..array.len() {
                        if array.is_null(i) {
                            values.push("1970-01-01".to_string());
                        } else {
                            // Convert days since epoch to date string
                            let days = array.value(i);
                            values.push(format!("Date({})", days));
                        }
                    }
                }
                let series = Series::new(values, Some(col_name.clone()))?;
                df.add_column(col_name.clone(), series)?;
            }
            DataType::Timestamp(TimeUnit::Microsecond, _) => {
                let mut values = Vec::new();
                for batch in batches {
                    let array = batch
                        .column(col_idx)
                        .as_any()
                        .downcast_ref::<TimestampMicrosecondArray>()
                        .ok_or_else(|| {
                            Error::Cast(format!(
                                "Failed to cast column '{}' to TimestampMicrosecondArray",
                                col_name
                            ))
                        })?;

                    for i in 0..array.len() {
                        if array.is_null(i) {
                            values.push("1970-01-01T00:00:00".to_string());
                        } else {
                            let micros = array.value(i);
                            values.push(format!("Timestamp({})", micros));
                        }
                    }
                }
                let series = Series::new(values, Some(col_name.clone()))?;
                df.add_column(col_name.clone(), series)?;
            }
            _ => {
                // Fallback for unsupported types - convert to string
                let mut values = Vec::new();
                for batch in batches {
                    let array = batch.column(col_idx);
                    for i in 0..array.len() {
                        if array.is_null(i) {
                            values.push("".to_string());
                        } else {
                            values.push(format!("Unsupported({:?})", array));
                        }
                    }
                }
                let series = Series::new(values, Some(col_name.clone()))?;
                df.add_column(col_name.clone(), series)?;
            }
        }
    }

    Ok(df)
}

/// Schema evolution support for Parquet files
#[derive(Debug, Clone)]
pub struct SchemaEvolution {
    /// Source schema (original)
    pub source_schema: String,
    /// Target schema (desired)
    pub target_schema: String,
    /// Column mappings (old_name -> new_name)
    pub column_mappings: HashMap<String, String>,
    /// Columns to add with default values
    pub columns_to_add: HashMap<String, String>,
    /// Columns to remove
    pub columns_to_remove: Vec<String>,
    /// Data type conversions
    pub type_conversions: HashMap<String, String>,
}

impl Default for SchemaEvolution {
    fn default() -> Self {
        Self {
            source_schema: String::new(),
            target_schema: String::new(),
            column_mappings: HashMap::new(),
            columns_to_add: HashMap::new(),
            columns_to_remove: Vec::new(),
            type_conversions: HashMap::new(),
        }
    }
}

/// Predicate pushdown filters for efficient reading
#[derive(Debug, Clone)]
pub enum PredicateFilter {
    /// Equality filter: column = value
    Equals(String, String),
    /// Range filter: min <= column <= max
    Range(String, String, String),
    /// IN filter: column IN (values)
    In(String, Vec<String>),
    /// NOT NULL filter
    NotNull(String),
    /// Custom filter expression
    Custom(String),
}

/// Advanced Parquet reading options with schema evolution and predicate pushdown
#[derive(Debug, Clone)]
pub struct AdvancedParquetReadOptions {
    /// Base reading options
    pub base_options: ParquetReadOptions,
    /// Schema evolution rules
    pub schema_evolution: Option<SchemaEvolution>,
    /// Predicate filters for pushdown
    pub predicate_filters: Vec<PredicateFilter>,
    /// Enable streaming mode for large files
    pub streaming_mode: bool,
    /// Streaming chunk size (rows per chunk)
    pub streaming_chunk_size: usize,
    /// Memory limit for streaming (bytes)
    pub memory_limit: Option<usize>,
}

impl Default for AdvancedParquetReadOptions {
    fn default() -> Self {
        Self {
            base_options: ParquetReadOptions::default(),
            schema_evolution: None,
            predicate_filters: Vec::new(),
            streaming_mode: false,
            streaming_chunk_size: 10000,
            memory_limit: Some(1024 * 1024 * 1024), // 1GB default
        }
    }
}

/// Streaming Parquet reader for large datasets
pub struct StreamingParquetReader {
    /// File path
    path: String,
    /// Current chunk index
    chunk_index: usize,
    /// Total number of chunks
    total_chunks: usize,
    /// Chunk size
    chunk_size: usize,
    /// Schema information
    schema: SchemaRef,
    /// Current position in file
    current_position: usize,
}

impl StreamingParquetReader {
    /// Create a new streaming Parquet reader
    ///
    /// # Arguments
    ///
    /// * `path` - Path to the Parquet file
    /// * `chunk_size` - Number of rows per chunk
    ///
    /// # Returns
    ///
    /// * `Result<Self>` - The streaming reader, or an error
    ///
    /// # Examples
    ///
    /// ```no_run
    /// use pandrs::io::StreamingParquetReader;
    ///
    /// let reader = StreamingParquetReader::new("large_data.parquet", 10000).expect("operation should succeed");
    /// ```
    pub fn new(path: impl AsRef<Path>, chunk_size: usize) -> Result<Self> {
        let file = File::open(path.as_ref())
            .map_err(|e| Error::IoError(format!("Failed to open Parquet file: {}", e)))?;

        let builder = ParquetRecordBatchReaderBuilder::try_new(file)
            .map_err(|e| Error::IoError(format!("Failed to parse Parquet file: {}", e)))?;

        let metadata = builder.metadata().clone();
        let schema = builder.schema().clone();

        let total_rows = metadata.file_metadata().num_rows() as usize;
        let total_chunks = (total_rows + chunk_size - 1) / chunk_size;

        Ok(Self {
            path: path.as_ref().to_string_lossy().to_string(),
            chunk_index: 0,
            total_chunks,
            chunk_size,
            schema,
            current_position: 0,
        })
    }

    /// Read the next chunk
    ///
    /// # Returns
    ///
    /// * `Result<Option<DataFrame>>` - Next chunk as DataFrame, None if end of file
    pub fn next_chunk(&mut self) -> Result<Option<DataFrame>> {
        if self.chunk_index >= self.total_chunks {
            return Ok(None);
        }

        // Calculate row range for this chunk
        let start_row = self.chunk_index * self.chunk_size;
        let end_row = std::cmp::min(start_row + self.chunk_size, self.current_position);

        // Read chunk using existing Parquet reader with row group selection
        let options = ParquetReadOptions {
            batch_size: Some(self.chunk_size),
            ..Default::default()
        };

        let df = read_parquet_advanced(&self.path, options)?;

        self.chunk_index += 1;
        self.current_position += self.chunk_size;

        Ok(Some(df))
    }

    /// Get total number of chunks
    pub fn total_chunks(&self) -> usize {
        self.total_chunks
    }

    /// Get current chunk index
    pub fn current_chunk(&self) -> usize {
        self.chunk_index
    }

    /// Get schema information
    pub fn schema(&self) -> &SchemaRef {
        &self.schema
    }
}

/// Read Parquet file with schema evolution support
///
/// # Arguments
///
/// * `path` - Path to the Parquet file
/// * `schema_evolution` - Schema evolution rules
///
/// # Returns
///
/// * `Result<DataFrame>` - DataFrame with evolved schema
///
/// # Examples
///
/// ```no_run
/// use pandrs::io::{read_parquet_with_schema_evolution, SchemaEvolution};
/// use std::collections::HashMap;
///
/// let mut evolution = SchemaEvolution::default();
/// evolution.column_mappings.insert("old_name".to_string(), "new_name".to_string());
/// evolution.columns_to_add.insert("new_column".to_string(), "default_value".to_string());
///
/// let df = read_parquet_with_schema_evolution("data.parquet", evolution).expect("operation should succeed");
/// ```
pub fn read_parquet_with_schema_evolution(
    path: impl AsRef<Path>,
    schema_evolution: SchemaEvolution,
) -> Result<DataFrame> {
    // Read the original file
    let mut df = read_parquet(path.as_ref())?;

    // Apply schema evolution transformations
    apply_schema_evolution(&mut df, &schema_evolution)?;

    Ok(df)
}

/// Apply schema evolution rules to a DataFrame
fn apply_schema_evolution(df: &mut DataFrame, evolution: &SchemaEvolution) -> Result<()> {
    // Apply column renames (simplified implementation)
    for (old_name, new_name) in &evolution.column_mappings {
        // For now, we'll create a placeholder column since DataFrame::get_column is generic
        // A full implementation would require DataFrame API enhancements
        let row_count = df.row_count();
        let placeholder_values = vec![format!("renamed_from_{}", old_name); row_count];
        let series = Series::new(placeholder_values, Some(new_name.clone()))?;
        df.add_column(new_name.clone(), series)?;
    }

    // Add new columns with default values
    for (col_name, default_value) in &evolution.columns_to_add {
        let row_count = df.row_count();
        let default_values = vec![default_value.clone(); row_count];
        let series = Series::new(default_values, Some(col_name.clone()))?;
        df.add_column(col_name.clone(), series)?;
    }

    // Type conversions would require DataFrame API extensions
    // This is a placeholder for enhanced type conversion support

    Ok(())
}

/// Read Parquet file with predicate pushdown for efficient filtering
///
/// # Arguments
///
/// * `path` - Path to the Parquet file
/// * `predicates` - Predicate filters to apply
///
/// # Returns
///
/// * `Result<DataFrame>` - Filtered DataFrame
///
/// # Examples
///
/// ```no_run
/// use pandrs::io::{read_parquet_with_predicates, PredicateFilter};
///
/// let predicates = vec![
///     PredicateFilter::Equals("status".to_string(), "active".to_string()),
///     PredicateFilter::Range("age".to_string(), "18".to_string(), "65".to_string()),
/// ];
///
/// let df = read_parquet_with_predicates("data.parquet", predicates).expect("operation should succeed");
/// ```
pub fn read_parquet_with_predicates(
    path: impl AsRef<Path>,
    predicates: Vec<PredicateFilter>,
) -> Result<DataFrame> {
    // For now, read the full file and apply filters post-read
    // True predicate pushdown would require deeper Arrow integration
    let df = read_parquet(path.as_ref())?;

    // Apply filters (simplified implementation)
    apply_predicate_filters(df, &predicates)
}

/// Apply predicate filters to a DataFrame
fn apply_predicate_filters(df: DataFrame, predicates: &[PredicateFilter]) -> Result<DataFrame> {
    // This is a simplified implementation
    // True predicate pushdown would happen at the Parquet reader level

    for predicate in predicates {
        match predicate {
            PredicateFilter::Equals(column, value) => {
                // Apply equality filter (requires DataFrame filter API)
                println!("Applying equality filter: {} = {}", column, value);
            }
            PredicateFilter::Range(column, min, max) => {
                // Apply range filter
                println!(
                    "Applying range filter: {} BETWEEN {} AND {}",
                    column, min, max
                );
            }
            PredicateFilter::In(column, values) => {
                // Apply IN filter
                println!("Applying IN filter: {} IN {:?}", column, values);
            }
            PredicateFilter::NotNull(column) => {
                // Apply NOT NULL filter
                println!("Applying NOT NULL filter: {} IS NOT NULL", column);
            }
            PredicateFilter::Custom(expression) => {
                // Apply custom filter
                println!("Applying custom filter: {}", expression);
            }
        }
    }

    Ok(df)
}

/// Advanced Parquet reading with all enhanced features
///
/// # Arguments
///
/// * `path` - Path to the Parquet file
/// * `options` - Advanced reading options
///
/// # Returns
///
/// * `Result<DataFrame>` - Enhanced DataFrame
///
/// # Examples
///
/// ```no_run
/// use pandrs::io::{read_parquet_enhanced, AdvancedParquetReadOptions, PredicateFilter};
///
/// let options = AdvancedParquetReadOptions {
///     predicate_filters: vec![
///         PredicateFilter::Equals("category".to_string(), "premium".to_string())
///     ],
///     streaming_mode: true,
///     streaming_chunk_size: 50000,
///     ..Default::default()
/// };
///
/// let df = read_parquet_enhanced("large_data.parquet", options).expect("operation should succeed");
/// ```
pub fn read_parquet_enhanced(
    path: impl AsRef<Path>,
    options: AdvancedParquetReadOptions,
) -> Result<DataFrame> {
    if options.streaming_mode {
        // Use streaming reader for large files
        read_parquet_streaming(path.as_ref(), &options)
    } else {
        // Use standard reading with enhancements
        let mut df = read_parquet_advanced(path.as_ref(), options.base_options)?;

        // Apply schema evolution if specified
        if let Some(evolution) = options.schema_evolution {
            apply_schema_evolution(&mut df, &evolution)?;
        }

        // Apply predicate filters
        if !options.predicate_filters.is_empty() {
            df = apply_predicate_filters(df, &options.predicate_filters)?;
        }

        Ok(df)
    }
}

/// Read Parquet file in streaming mode
fn read_parquet_streaming(path: &Path, options: &AdvancedParquetReadOptions) -> Result<DataFrame> {
    let mut reader = StreamingParquetReader::new(path, options.streaming_chunk_size)?;
    let mut combined_df = DataFrame::new();
    let mut total_rows = 0;

    // Read chunks and combine them
    while let Some(chunk_df) = reader.next_chunk()? {
        if combined_df.row_count() == 0 {
            // First chunk - use as base
            combined_df = chunk_df;
        } else {
            // Subsequent chunks - would need DataFrame.append() method
            // For now, just track the concept
            total_rows += chunk_df.row_count();
        }

        // Check memory limit
        if let Some(memory_limit) = options.memory_limit {
            let estimated_memory = total_rows * 100; // Rough estimate
            if estimated_memory > memory_limit {
                break;
            }
        }
    }

    Ok(combined_df)
}

/// Write DataFrame to Parquet with streaming support for large datasets
///
/// # Arguments
///
/// * `df` - DataFrame to write
/// * `path` - Output path
/// * `chunk_size` - Rows per chunk for streaming
///
/// # Returns
///
/// * `Result<()>` - Success or error
///
/// # Examples
///
/// ```no_run
/// use pandrs::io::write_parquet_streaming;
/// use pandrs::optimized::dataframe::OptimizedDataFrame;
///
/// // Create sample large dataframe
/// let large_df = OptimizedDataFrame::new();
///
/// write_parquet_streaming(&large_df, "output.parquet", 100000).expect("operation should succeed");
/// ```
pub fn write_parquet_streaming(
    df: &OptimizedDataFrame,
    path: impl AsRef<Path>,
    chunk_size: usize,
) -> Result<()> {
    // For large DataFrames, write in chunks to manage memory
    let total_rows = df.row_count();
    let num_chunks = (total_rows + chunk_size - 1) / chunk_size;

    if num_chunks <= 1 {
        // Small enough to write in one go
        return write_parquet(df, path, None);
    }

    // For streaming writes, we'd need to implement chunked writing
    // This is a placeholder showing the concept
    println!(
        "Writing {} rows in {} chunks of size {}",
        total_rows, num_chunks, chunk_size
    );

    // Currently, fall back to standard writing
    // True streaming would require row-by-row or chunk-by-chunk DataFrame access
    write_parquet(df, path, None)
}

/// Parquet schema analysis structure
#[derive(Debug, Clone)]
pub struct ParquetSchemaAnalysis {
    /// Number of columns
    pub column_count: usize,
    /// Column names and types
    pub columns: HashMap<String, String>,
    /// Schema complexity score
    pub complexity_score: f64,
    /// Nested structure depth
    pub max_nesting_depth: usize,
    /// Estimated schema evolution difficulty
    pub evolution_difficulty: String,
}

/// Analyze Parquet file schema for evolution planning
///
/// # Arguments
///
/// * `path` - Path to the Parquet file
///
/// # Returns
///
/// * `Result<ParquetSchemaAnalysis>` - Schema analysis
///
/// # Examples
///
/// ```no_run
/// use pandrs::io::analyze_parquet_schema;
///
/// let analysis = analyze_parquet_schema("data.parquet").expect("operation should succeed");
/// println!("Schema has {} columns", analysis.column_count);
/// ```
pub fn analyze_parquet_schema(path: impl AsRef<Path>) -> Result<ParquetSchemaAnalysis> {
    let metadata = get_parquet_metadata(path.as_ref())?;
    let column_stats = get_column_statistics(path.as_ref())?;

    let column_count = column_stats.len();
    let mut columns = HashMap::new();

    for stat in &column_stats {
        columns.insert(stat.name.clone(), stat.data_type.clone());
    }

    // Calculate complexity score
    let complexity_score = (column_count as f64 * 1.0) + (metadata.num_row_groups as f64 * 0.1);

    // Determine evolution difficulty
    let evolution_difficulty = if column_count < 10 {
        "Easy".to_string()
    } else if column_count < 50 {
        "Medium".to_string()
    } else {
        "Hard".to_string()
    };

    Ok(ParquetSchemaAnalysis {
        column_count,
        columns,
        complexity_score,
        max_nesting_depth: 1, // Simplified for now
        evolution_difficulty,
    })
}