foxhunt/data/src/parquet_persistence.rs

//! # Parquet Market Data Persistence for Replay
//!
//! High-performance Parquet-based market data persistence system for backtesting
//! and trade replay capabilities in the Foxhunt HFT system.

use anyhow::{Context, Result};
use arrow::array::{Array, Float64Array, StringArray, TimestampNanosecondArray, UInt64Array};
use arrow::datatypes::{DataType, Field, Schema, TimeUnit};
use arrow::record_batch::RecordBatch;
use parquet::arrow::{arrow_reader::ParquetRecordBatchReaderBuilder, ArrowWriter};
use parquet::file::properties::{EnabledStatistics, WriterProperties};
use serde::{Deserialize, Serialize};
use std::fs::File;
use std::path::Path;
use std::sync::Arc;
use tokio::sync::{mpsc, RwLock};
use tokio::time::{Duration, Instant};
use tracing::{debug, error, info};

// Import the Parquet-specific market data event from trading_engine
pub use trading_engine::types::metrics::ParquetMarketDataEvent as MarketDataEvent;

/// `Parquet` writer configuration
#[derive(Debug, Clone)]
pub struct ParquetConfig {
    pub base_path: String,
    pub batch_size: usize,
    pub flush_interval_ms: u64,
    pub compression: parquet::basic::Compression,
    pub enable_dictionary: bool,
    pub enable_statistics: EnabledStatistics,
}

impl Default for ParquetConfig {
    fn default() -> Self {
        Self {
            base_path: "./market_data".to_string(),
            batch_size: 10000,
            flush_interval_ms: 5000,
            compression: parquet::basic::Compression::SNAPPY,
            enable_dictionary: true,
            enable_statistics: EnabledStatistics::Page,
        }
    }
}

/// High-performance `Parquet` writer for market data
pub struct ParquetMarketDataWriter {
    config: ParquetConfig,
    buffer: Arc<RwLock<Vec<MarketDataEvent>>>,
    sender: mpsc::UnboundedSender<MarketDataEvent>,
    _writer_handle: tokio::task::JoinHandle<()>,
}

impl ParquetMarketDataWriter {
    /// Create new `Parquet` writer with background processing
    pub async fn new(config: ParquetConfig) -> Result<Self> {
        // Ensure base directory exists
        std::fs::create_dir_all(&config.base_path)
            .context("Failed to create Parquet base directory")?;

        let buffer = Arc::new(RwLock::new(Vec::with_capacity(config.batch_size * 2)));
        let (sender, receiver) = mpsc::unbounded_channel();

        let writer_handle =
            Self::spawn_writer_task(config.clone(), buffer.clone(), receiver).await?;

        Ok(Self {
            config,
            buffer,
            sender,
            _writer_handle: writer_handle,
        })
    }

    /// Record market data event (non-blocking)
    pub fn record(&self, event: MarketDataEvent) -> Result<()> {
        self.sender
            .send(event)
            .context("Failed to send market data event to writer")?;
        Ok(())
    }

    /// Get current buffer statistics
    pub async fn get_buffer_stats(&self) -> BufferStats {
        let buffer = self.buffer.read().await;
        BufferStats {
            buffered_events: buffer.len(),
            buffer_capacity: buffer.capacity(),
            utilization_percent: (buffer.len() as f64 / buffer.capacity() as f64) * 100.0,
        }
    }

    /// Spawn background writer task
    async fn spawn_writer_task(
        config: ParquetConfig,
        buffer: Arc<RwLock<Vec<MarketDataEvent>>>,
        mut receiver: mpsc::UnboundedReceiver<MarketDataEvent>,
    ) -> Result<tokio::task::JoinHandle<()>> {
        let handle = tokio::spawn(async move {
            let mut flush_interval =
                tokio::time::interval(Duration::from_millis(config.flush_interval_ms));
            let mut last_flush = Instant::now();

            loop {
                tokio::select! {
                    // Handle incoming events
                    event = receiver.recv() => {
                        match event {
                            Some(event) => {
                                let mut buffer_guard = buffer.write().await;
                                buffer_guard.push(event);

                                // Check if we should flush based on batch size
                                if buffer_guard.len() >= config.batch_size {
                                    let events = buffer_guard.drain(..).collect();
                                    drop(buffer_guard);

                                    if let Err(e) = Self::write_batch_to_parquet(&config, events).await {
                                        error!("Failed to write Parquet batch: {}", e);
                                    }
                                    last_flush = Instant::now();
                                }
                            }
                            None => {
                                info!("Parquet writer channel closed, shutting down");
                                break;
                            }
                        }
                    }

                    // Handle periodic flush
                    _ = flush_interval.tick() => {
                        if last_flush.elapsed() >= Duration::from_millis(config.flush_interval_ms) {
                            let mut buffer_guard = buffer.write().await;
                            if !buffer_guard.is_empty() {
                                let events = buffer_guard.drain(..).collect();
                                drop(buffer_guard);

                                if let Err(e) = Self::write_batch_to_parquet(&config, events).await {
                                    error!("Failed to write Parquet batch on flush: {}", e);
                                }
                                last_flush = Instant::now();
                            }
                        }
                    }
                }
            }

            // Final flush on shutdown
            let mut buffer_guard = buffer.write().await;
            if !buffer_guard.is_empty() {
                let events = buffer_guard.drain(..).collect();
                drop(buffer_guard);

                if let Err(e) = Self::write_batch_to_parquet(&config, events).await {
                    error!("Failed to write final Parquet batch: {}", e);
                }
            }
        });

        Ok(handle)
    }

    /// Write batch of events to `Parquet` file
    async fn write_batch_to_parquet(
        config: &ParquetConfig,
        events: Vec<MarketDataEvent>,
    ) -> Result<()> {
        if events.is_empty() {
            return Ok(());
        }

        let start_time = Instant::now();

        // Generate filename with timestamp
        let timestamp = events[0].timestamp_ns;
        let date = chrono::DateTime::from_timestamp_nanos(timestamp as i64).format("%Y%m%d_%H%M%S");
        let filename = format!(
            "market_data_{}_{}.parquet",
            date,
            uuid::Uuid::new_v4().simple()
        );
        let filepath = Path::new(&config.base_path).join(filename);

        // Create Arrow schema matching ParquetMarketDataEvent fields
        let schema = Arc::new(Schema::new(vec![
            Field::new(
                "timestamp_ns",
                DataType::Timestamp(TimeUnit::Nanosecond, None),
                false,
            ),
            Field::new("symbol", DataType::Utf8, false),
            Field::new("venue", DataType::Utf8, false),
            Field::new("event_type", DataType::Utf8, false),
            Field::new("price", DataType::Float64, true),
            Field::new("quantity", DataType::Float64, true),
            Field::new("sequence", DataType::UInt64, false),
            Field::new("latency_ns", DataType::UInt64, true),
            Field::new("open", DataType::Float64, true),
            Field::new("high", DataType::Float64, true),
            Field::new("low", DataType::Float64, true),
        ]));

        // Convert events to Arrow arrays
        let record_batch = Self::events_to_record_batch(&schema, events)?;

        // Write to Parquet file
        let file = File::create(&filepath)
            .with_context(|| format!("Failed to create Parquet file: {:?}", filepath))?;

        let props = WriterProperties::builder()
            .set_compression(config.compression)
            .set_dictionary_enabled(config.enable_dictionary)
            .set_statistics_enabled(config.enable_statistics)
            .build();

        let mut writer = ArrowWriter::try_new(file, schema, Some(props))
            .context("Failed to create Arrow writer")?;

        writer
            .write(&record_batch)
            .context("Failed to write record batch")?;

        writer.close().context("Failed to close Arrow writer")?;

        let duration = start_time.elapsed();
        let events_count = record_batch.num_rows();

        debug!(
            "Wrote {} events to Parquet file {:?} in {:?}",
            events_count, filepath, duration
        );

        // Update metrics
        let duration_us: u64 = duration.as_micros().try_into().unwrap_or(0);
        if duration_us > 0 {
            trading_engine::types::metrics::LATENCY_HISTOGRAMS
                .with_label_values(&["parquet_write", "data_service"])
                .observe(duration_us as f64 / 1_000_000.0);
        }

        trading_engine::types::metrics::THROUGHPUT_COUNTERS
            .with_label_values(&["parquet_events", "data_service"])
            .inc_by(events_count as u64);

        Ok(())
    }

    /// Convert events to `Arrow` RecordBatch
    fn events_to_record_batch(
        schema: &Arc<Schema>,
        events: Vec<MarketDataEvent>,
    ) -> Result<RecordBatch> {
        let len = events.len();

        // Extract data into separate vectors matching ParquetMarketDataEvent fields
        let mut timestamps = Vec::with_capacity(len);
        let mut symbols = Vec::with_capacity(len);
        let mut venues = Vec::with_capacity(len);
        let mut event_types = Vec::with_capacity(len);
        let mut prices = Vec::with_capacity(len);
        let mut quantities = Vec::with_capacity(len);
        let mut sequences = Vec::with_capacity(len);
        let mut latencies = Vec::with_capacity(len);
        let mut opens = Vec::with_capacity(len);
        let mut highs = Vec::with_capacity(len);
        let mut lows = Vec::with_capacity(len);

        for event in events {
            timestamps.push(Some(event.timestamp_ns as i64));
            symbols.push(Some(event.symbol));
            venues.push(Some(event.venue));
            // Convert MarketDataEventType enum to string
            event_types.push(Some(format!("{:?}", event.event_type)));
            prices.push(event.price);
            quantities.push(event.quantity);
            sequences.push(event.sequence);
            latencies.push(event.latency_ns);
            opens.push(event.open);
            highs.push(event.high);
            lows.push(event.low);
        }

        // Create Arrow arrays matching ParquetMarketDataEvent schema
        let timestamp_array = TimestampNanosecondArray::from(timestamps);
        let symbol_array = StringArray::from(symbols);
        let venue_array = StringArray::from(venues);
        let event_type_array = StringArray::from(event_types);
        let price_array = Float64Array::from(prices);
        let quantity_array = Float64Array::from(quantities);
        let sequence_array = UInt64Array::from(sequences);
        let latency_array = UInt64Array::from(latencies);
        let open_array = Float64Array::from(opens);
        let high_array = Float64Array::from(highs);
        let low_array = Float64Array::from(lows);

        // Create record batch
        RecordBatch::try_new(
            schema.clone(),
            vec![
                Arc::new(timestamp_array),
                Arc::new(symbol_array),
                Arc::new(venue_array),
                Arc::new(event_type_array),
                Arc::new(price_array),
                Arc::new(quantity_array),
                Arc::new(sequence_array),
                Arc::new(latency_array),
                Arc::new(open_array),
                Arc::new(high_array),
                Arc::new(low_array),
            ],
        )
        .context("Failed to create Arrow RecordBatch")
    }
}

/// Buffer statistics for monitoring
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct BufferStats {
    pub buffered_events: usize,
    pub buffer_capacity: usize,
    pub utilization_percent: f64,
}

/// `Parquet` reader for market data replay
pub struct ParquetMarketDataReader {
    base_path: String,
}

impl ParquetMarketDataReader {
    pub fn new(base_path: String) -> Self {
        Self { base_path }
    }

    /// Get the base path for this reader
    pub fn base_path(&self) -> &str {
        &self.base_path
    }

    /// Cast timestamp column to nanoseconds, supporting multiple timestamp types
    fn cast_timestamp_column(col: &Arc<dyn Array>) -> Result<Vec<i64>> {
        use arrow::array::{
            Int64Array, TimestampMicrosecondArray, TimestampMillisecondArray, TimestampSecondArray,
        };

        match col.data_type() {
            DataType::Timestamp(TimeUnit::Nanosecond, _) => {
                let array = col.as_any().downcast_ref::<TimestampNanosecondArray>()
                    .context("Failed to downcast TimestampNanosecondArray")?;
                Ok((0..array.len())
                    .map(|i| if array.is_null(i) { 0 } else { array.value(i) })
                    .collect())
            }
            DataType::Timestamp(TimeUnit::Microsecond, _) => {
                let array = col.as_any().downcast_ref::<TimestampMicrosecondArray>()
                    .context("Failed to downcast TimestampMicrosecondArray")?;
                Ok((0..array.len())
                    .map(|i| if array.is_null(i) { 0 } else { array.value(i) * 1_000 }) // μs -> ns
                    .collect())
            }
            DataType::Timestamp(TimeUnit::Millisecond, _) => {
                let array = col.as_any().downcast_ref::<TimestampMillisecondArray>()
                    .context("Failed to downcast TimestampMillisecondArray")?;
                Ok((0..array.len())
                    .map(|i| if array.is_null(i) { 0 } else { array.value(i) * 1_000_000 }) // ms -> ns
                    .collect())
            }
            DataType::Timestamp(TimeUnit::Second, _) => {
                let array = col.as_any().downcast_ref::<TimestampSecondArray>()
                    .context("Failed to downcast TimestampSecondArray")?;
                Ok((0..array.len())
                    .map(|i| if array.is_null(i) { 0 } else { array.value(i) * 1_000_000_000 }) // s -> ns
                    .collect())
            }
            DataType::UInt64 => {
                // Handle UInt64 timestamps (assume nanoseconds or convert based on magnitude)
                let array = col.as_any().downcast_ref::<UInt64Array>()
                    .context("Failed to downcast UInt64Array")?;
                Ok((0..array.len())
                    .map(|i| if array.is_null(i) { 0 } else { array.value(i) as i64 })
                    .collect())
            }
            DataType::Int64 => {
                // Handle Int64 timestamps (assume nanoseconds)
                let array = col.as_any().downcast_ref::<Int64Array>()
                    .context("Failed to downcast Int64Array")?;
                Ok((0..array.len())
                    .map(|i| if array.is_null(i) { 0 } else { array.value(i) })
                    .collect())
            }
            other => Err(anyhow::anyhow!(
                "Unsupported timestamp type: {:?}. Expected one of: Timestamp(Nanosecond|Microsecond|Millisecond|Second), UInt64, or Int64",
                other
            )),
        }
    }

    /// List available `Parquet` files for replay
    pub async fn list_available_files(&self) -> Result<Vec<String>> {
        let mut files = Vec::new();
        let entries =
            std::fs::read_dir(&self.base_path).context("Failed to read Parquet directory")?;

        for entry in entries {
            let entry = entry.context("Failed to read directory entry")?;
            let path = entry.path();

            if path.is_file() && path.extension().map_or(false, |ext| ext == "parquet") {
                if let Some(filename) = path.file_name().and_then(|f| f.to_str()) {
                    files.push(filename.to_string());
                }
            }
        }

        files.sort();
        Ok(files)
    }

    /// Read market data from `Parquet` file for replay
    pub async fn read_file(&self, filename: &str) -> Result<Vec<MarketDataEvent>> {
        let filepath = Path::new(&self.base_path).join(filename);

        debug!("Reading Parquet file: {:?}", filepath);

        // Open the Parquet file
        let file = File::open(&filepath)
            .with_context(|| format!("Failed to open Parquet file: {:?}", filepath))?;

        // Create Parquet reader
        let builder = ParquetRecordBatchReaderBuilder::try_new(file)
            .with_context(|| format!("Failed to create Parquet reader for: {:?}", filepath))?;

        let schema = builder.schema().clone();
        let reader = builder
            .build()
            .context("Failed to build Parquet record batch reader")?;

        let mut events = Vec::new();

        // Detect schema type (system format vs CSV-derived format)
        let field_names: Vec<&str> = schema.fields().iter().map(|f| f.name().as_str()).collect();
        debug!(
            "Parquet schema has {} fields: {:?}",
            field_names.len(),
            field_names
        );

        // CSV format: timestamp, open, high, low, close, volume (6 columns)
        // System format (from CSV): sequence, timestamp_ns, symbol, venue, event_type, price, quantity, latency_ns (8 columns)
        // System format (full): above + open, high, low (11 columns)
        let is_csv_derived_system_format = schema.fields().len() == 8
            && field_names.contains(&"sequence")
            && field_names.contains(&"symbol")
            && field_names.contains(&"price")
            && !field_names.contains(&"open");

        let is_pure_csv_format = schema.fields().len() == 6
            && schema.field(1).name() == "open"
            && schema.field(4).name() == "close";

        debug!(
            "Format detection: csv_derived_system={}, pure_csv={}",
            is_csv_derived_system_format, is_pure_csv_format
        );

        // Read all record batches
        for batch_result in reader {
            let batch = batch_result.context("Failed to read record batch from Parquet")?;

            if is_pure_csv_format {
                // CSV-derived format: timestamp, open, high, low, close, volume
                Self::parse_csv_format_batch(&batch, &mut events, filename)?;
            } else if is_csv_derived_system_format {
                // CSV-to-system converted format: has system fields but no OHLC columns
                // Just use the system parser (it handles missing OHLC)
                Self::parse_system_format_batch(&batch, &mut events)?;
            } else {
                // System format: full MarketDataEvent schema with OHLC
                Self::parse_system_format_batch(&batch, &mut events)?;
            }
        }

        info!(
            "Successfully read {} events from {:?}",
            events.len(),
            filepath
        );
        Ok(events)
    }

    /// Parse CSV-derived format (timestamp, open, high, low, close, volume)
    fn parse_csv_format_batch(
        batch: &RecordBatch,
        events: &mut Vec<MarketDataEvent>,
        filename: &str,
    ) -> Result<()> {
        // Extract columns
        let timestamp_col = batch.column(0);
        let timestamps = Self::cast_timestamp_column(timestamp_col).with_context(|| {
            format!(
                "Failed to cast timestamp column. Column type: {:?}",
                timestamp_col.data_type()
            )
        })?;
        let opens = batch
            .column(1)
            .as_any()
            .downcast_ref::<Float64Array>()
            .context("Failed to cast open column")?;
        let highs = batch
            .column(2)
            .as_any()
            .downcast_ref::<Float64Array>()
            .context("Failed to cast high column")?;
        let lows = batch
            .column(3)
            .as_any()
            .downcast_ref::<Float64Array>()
            .context("Failed to cast low column")?;
        let closes = batch
            .column(4)
            .as_any()
            .downcast_ref::<Float64Array>()
            .context("Failed to cast close column")?;
        let volumes = batch
            .column(5)
            .as_any()
            .downcast_ref::<Float64Array>()
            .context("Failed to cast volume column")?;

        // Extract symbol from filename (e.g., "BTC-USD_30day_2024-09.parquet" -> "BTC-USD")
        let symbol = filename.split('_').next().unwrap_or("UNKNOWN").to_string();

        // Convert rows to MarketDataEvent structs
        for i in 0..batch.num_rows() {
            let timestamp_ns = timestamps[i] as u64;
            let open = if opens.is_null(i) {
                None
            } else {
                Some(opens.value(i))
            };
            let high = if highs.is_null(i) {
                None
            } else {
                Some(highs.value(i))
            };
            let low = if lows.is_null(i) {
                None
            } else {
                Some(lows.value(i))
            };
            let price = if closes.is_null(i) {
                None
            } else {
                Some(closes.value(i))
            };
            let quantity = if volumes.is_null(i) {
                None
            } else {
                Some(volumes.value(i))
            };

            events.push(MarketDataEvent {
                timestamp_ns,
                symbol: symbol.clone(),
                venue: "exchange".to_string(), // Default venue
                event_type: trading_engine::types::metrics::MarketDataEventType::Trade,
                price,
                quantity,
                sequence: i as u64,
                latency_ns: None,
                open,
                high,
                low,
            });
        }

        Ok(())
    }

    /// Parse system format (full MarketDataEvent schema)
    fn parse_system_format_batch(
        batch: &RecordBatch,
        events: &mut Vec<MarketDataEvent>,
    ) -> Result<()> {
        use arrow::array::LargeStringArray;

        // Actual schema from files: sequence, timestamp_ns, symbol, venue, event_type, price, quantity, latency_ns
        let sequences = batch
            .column(0)
            .as_any()
            .downcast_ref::<UInt64Array>()
            .context("Failed to cast sequence column")?;
        let timestamp_col = batch.column(1);
        let timestamps = Self::cast_timestamp_column(timestamp_col).with_context(|| {
            format!(
                "Failed to cast timestamp column. Column type: {:?}",
                timestamp_col.data_type()
            )
        })?;

        // Handle both StringArray (Utf8) and LargeStringArray (LargeUtf8)
        let symbols = if let Some(arr) = batch.column(2).as_any().downcast_ref::<StringArray>() {
            arr.clone()
        } else if let Some(large_arr) = batch.column(2).as_any().downcast_ref::<LargeStringArray>()
        {
            // Convert LargeStringArray to StringArray
            let values: Vec<Option<&str>> = (0..large_arr.len())
                .map(|i| {
                    if large_arr.is_null(i) {
                        None
                    } else {
                        Some(large_arr.value(i))
                    }
                })
                .collect();
            StringArray::from(values)
        } else {
            return Err(anyhow::anyhow!(
                "Failed to cast symbol column. Column type: {:?}",
                batch.column(2).data_type()
            ));
        };

        let venues = if let Some(arr) = batch.column(3).as_any().downcast_ref::<StringArray>() {
            arr.clone()
        } else if let Some(large_arr) = batch.column(3).as_any().downcast_ref::<LargeStringArray>()
        {
            let values: Vec<Option<&str>> = (0..large_arr.len())
                .map(|i| {
                    if large_arr.is_null(i) {
                        None
                    } else {
                        Some(large_arr.value(i))
                    }
                })
                .collect();
            StringArray::from(values)
        } else {
            return Err(anyhow::anyhow!(
                "Failed to cast venue column. Column type: {:?}",
                batch.column(3).data_type()
            ));
        };

        let event_types = if let Some(arr) = batch.column(4).as_any().downcast_ref::<StringArray>()
        {
            arr.clone()
        } else if let Some(large_arr) = batch.column(4).as_any().downcast_ref::<LargeStringArray>()
        {
            let values: Vec<Option<&str>> = (0..large_arr.len())
                .map(|i| {
                    if large_arr.is_null(i) {
                        None
                    } else {
                        Some(large_arr.value(i))
                    }
                })
                .collect();
            StringArray::from(values)
        } else {
            return Err(anyhow::anyhow!(
                "Failed to cast event_type column. Column type: {:?}",
                batch.column(4).data_type()
            ));
        };
        let prices = batch
            .column(5)
            .as_any()
            .downcast_ref::<Float64Array>()
            .context("Failed to cast price column")?;
        let quantities = batch
            .column(6)
            .as_any()
            .downcast_ref::<Float64Array>()
            .context("Failed to cast quantity column")?;
        let latencies = batch
            .column(7)
            .as_any()
            .downcast_ref::<UInt64Array>()
            .context("Failed to cast latency column")?;

        // Handle optional OHLC columns (not present in all files)
        let opens = if batch.num_columns() > 8 {
            Some(
                batch
                    .column(8)
                    .as_any()
                    .downcast_ref::<Float64Array>()
                    .context("Failed to cast open column")?,
            )
        } else {
            None
        };
        let highs = if batch.num_columns() > 9 {
            Some(
                batch
                    .column(9)
                    .as_any()
                    .downcast_ref::<Float64Array>()
                    .context("Failed to cast high column")?,
            )
        } else {
            None
        };
        let lows = if batch.num_columns() > 10 {
            Some(
                batch
                    .column(10)
                    .as_any()
                    .downcast_ref::<Float64Array>()
                    .context("Failed to cast low column")?,
            )
        } else {
            None
        };

        // Convert rows to MarketDataEvent structs
        for i in 0..batch.num_rows() {
            let timestamp_ns = timestamps[i] as u64;

            let symbol = if symbols.is_null(i) {
                String::new()
            } else {
                symbols.value(i).to_string()
            };

            let venue = if venues.is_null(i) {
                String::new()
            } else {
                venues.value(i).to_string()
            };

            let event_type_str = if event_types.is_null(i) {
                "Trade"
            } else {
                event_types.value(i)
            };

            // Parse event type string back to enum
            let event_type = match event_type_str {
                "Trade" => trading_engine::types::metrics::MarketDataEventType::Trade,
                "Quote" => trading_engine::types::metrics::MarketDataEventType::Quote,
                "OrderBookUpdate" => {
                    trading_engine::types::metrics::MarketDataEventType::OrderBookUpdate
                },
                _ => trading_engine::types::metrics::MarketDataEventType::Trade,
            };

            let price = if prices.is_null(i) {
                None
            } else {
                Some(prices.value(i))
            };
            let quantity = if quantities.is_null(i) {
                None
            } else {
                Some(quantities.value(i))
            };
            let sequence = if sequences.is_null(i) {
                0
            } else {
                sequences.value(i)
            };
            let latency_ns = if latencies.is_null(i) {
                None
            } else {
                Some(latencies.value(i))
            };

            let open = opens.and_then(|arr| {
                if arr.is_null(i) {
                    None
                } else {
                    Some(arr.value(i))
                }
            });
            let high = highs.and_then(|arr| {
                if arr.is_null(i) {
                    None
                } else {
                    Some(arr.value(i))
                }
            });
            let low = lows.and_then(|arr| {
                if arr.is_null(i) {
                    None
                } else {
                    Some(arr.value(i))
                }
            });

            events.push(MarketDataEvent {
                timestamp_ns,
                symbol,
                venue,
                event_type,
                price,
                quantity,
                sequence,
                latency_ns,
                open,
                high,
                low,
            });
        }

        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use tempfile::tempdir;

    #[tokio::test]
    async fn test_parquet_writer_creation() {
        let temp_dir = tempdir().unwrap();
        let config = ParquetConfig {
            base_path: temp_dir.path().to_string_lossy().to_string(),
            ..Default::default()
        };

        let writer = ParquetMarketDataWriter::new(config).await;
        assert!(writer.is_ok());
    }

    #[tokio::test]
    async fn test_market_data_event_recording() {
        let temp_dir = tempdir().unwrap();
        let config = ParquetConfig {
            base_path: temp_dir.path().to_string_lossy().to_string(),
            batch_size: 2, // Small batch for testing
            ..Default::default()
        };

        let writer = ParquetMarketDataWriter::new(config).await.unwrap();

        let event = MarketDataEvent {
            timestamp_ns: 1234567890000000000,
            symbol: "BTCUSD".to_string(),
            venue: "binance".to_string(),
            event_type: trading_engine::types::metrics::MarketDataEventType::Trade,
            price: Some(50000.0),
            quantity: Some(0.1),
            sequence: 1,
            latency_ns: Some(1000),
            open: None,
            high: None,
            low: None,
        };

        let result = writer.record(event);
        assert!(result.is_ok());

        // Give some time for background processing
        tokio::time::sleep(Duration::from_millis(100)).await;
    }
}