exarrow-rs 0.20.1

//! CSV export functionality for Exasol.
//!
//! This module provides functions for exporting data from Exasol tables or query results
//! to CSV format via files, streams, in-memory lists, or custom callbacks.
//!
//! # Architecture
//!
//! The export process uses **client mode** where:
//! 1. We connect TO Exasol (outbound connection - works through firewalls)
//! 2. Perform EXA tunneling handshake to receive an internal address
//! 3. Execute an EXPORT SQL statement using the internal address
//! 4. Exasol sends data through the established connection
//! 5. We read and process the data stream
//!
//! This client mode approach works with cloud Exasol instances and through NAT/firewalls.

use std::future::Future;
use std::io;
use std::path::Path;
use std::sync::atomic::{AtomicBool, Ordering};

use bzip2::read::BzDecoder;
use flate2::read::GzDecoder;
use thiserror::Error;
use tokio::fs::File;
use tokio::io::{AsyncWrite, AsyncWriteExt, BufWriter};
use tokio::sync::mpsc;

use crate::query::export::{Compression, ExportQuery, ExportSource, RowSeparator};
use crate::transport::protocol::TransportProtocol;
use crate::transport::HttpTransportClient;

/// Default buffer size for data pipe channels (number of chunks).
const DEFAULT_PIPE_BUFFER_SIZE: usize = 16;

/// Error types for export operations.
#[derive(Error, Debug)]
pub enum ExportError {
    /// I/O error during export.
    #[error("I/O error: {0}")]
    IoError(#[from] io::Error),

    /// Transport error during export.
    #[error("Transport error: {0}")]
    TransportError(#[from] crate::error::TransportError),

    /// HTTP transport setup error.
    #[error("HTTP transport error: {message}")]
    HttpTransportError { message: String },

    /// SQL execution error.
    #[error("SQL execution error: {message}")]
    SqlExecutionError { message: String },

    /// CSV parsing error.
    #[error("CSV parsing error at row {row}: {message}")]
    CsvParseError { row: usize, message: String },

    /// Decompression error.
    #[error("Decompression error: {0}")]
    DecompressionError(String),

    /// Channel communication error.
    #[error("Channel error: {0}")]
    ChannelError(String),

    /// An explicitly configured export timeout elapsed.
    ///
    /// `transport_terminated` reports whether the driver had to terminate the
    /// transport, so a caller decides whether to reconnect without probing the
    /// connection. It is `true` when the timeout elapsed before the EXPORT
    /// response was read, because that response can no longer be matched to a
    /// request.
    #[error(
        "Export timed out after {timeout_ms}ms; {}",
        if *transport_terminated {
            "the transport was terminated, reconnect before the next operation"
        } else {
            "the transport is still usable"
        }
    )]
    Timeout {
        /// The configured timeout in milliseconds that elapsed.
        timeout_ms: u64,
        /// Whether the driver terminated the transport before returning.
        transport_terminated: bool,
    },

    /// Export was cancelled.
    #[error("Export was cancelled")]
    Cancelled,
}

/// Options for CSV export configuration.
#[derive(Debug, Clone)]
pub struct CsvExportOptions {
    /// Column separator character (default: ',').
    pub column_separator: char,

    /// Column delimiter character for quoting (default: '"').
    pub column_delimiter: char,

    /// Row separator (default: LF).
    pub row_separator: RowSeparator,

    /// Character encoding (default: "UTF-8").
    pub encoding: String,

    /// Custom NULL value representation (default: None, empty string).
    pub null_value: Option<String>,

    /// Compression type (default: None).
    pub compression: Compression,

    /// Whether to include column headers in the output (default: false).
    pub with_column_names: bool,

    /// Use TLS for the HTTP transport (default: true).
    pub use_tls: bool,

    /// Client-side timeout in milliseconds for the export operation (default: `None`).
    ///
    /// `None` arms no timer, so the export runs until the server finishes the
    /// EXPORT statement and a server-enforced query timeout is the only bound.
    /// `Some(ms)` bounds SQL execution, tunnel transfer, and the caller's
    /// callback together. When such a bound elapses before the EXPORT response
    /// has been read, the driver terminates the transport, because that
    /// response can no longer be matched to a request.
    ///
    /// Tunnel setup comes before this bound and has its own 30-second limit.
    pub timeout_ms: Option<u64>,

    /// Exasol host for HTTP transport connection.
    /// This is typically the same host as the WebSocket connection.
    pub host: String,

    /// Exasol port for HTTP transport connection.
    /// This is typically the same port as the WebSocket connection.
    pub port: u16,
}

impl Default for CsvExportOptions {
    fn default() -> Self {
        Self {
            column_separator: ',',
            column_delimiter: '"',
            row_separator: RowSeparator::LF,
            encoding: "UTF-8".to_string(),
            null_value: None,
            compression: Compression::None,
            with_column_names: false,
            use_tls: false,
            timeout_ms: None,
            host: String::new(),
            port: 0,
        }
    }
}

impl CsvExportOptions {
    /// Creates new export options with default values.
    #[must_use]
    pub fn new() -> Self {
        Self::default()
    }

    #[must_use]
    pub fn column_separator(mut self, sep: char) -> Self {
        self.column_separator = sep;
        self
    }

    #[must_use]
    pub fn column_delimiter(mut self, delim: char) -> Self {
        self.column_delimiter = delim;
        self
    }

    #[must_use]
    pub fn row_separator(mut self, sep: RowSeparator) -> Self {
        self.row_separator = sep;
        self
    }

    #[must_use]
    pub fn encoding(mut self, enc: &str) -> Self {
        self.encoding = enc.to_string();
        self
    }

    #[must_use]
    pub fn null_value(mut self, val: &str) -> Self {
        self.null_value = Some(val.to_string());
        self
    }

    #[must_use]
    pub fn compression(mut self, compression: Compression) -> Self {
        self.compression = compression;
        self
    }

    #[must_use]
    pub fn with_column_names(mut self, include: bool) -> Self {
        self.with_column_names = include;
        self
    }

    #[must_use]
    pub fn use_tls(mut self, use_tls: bool) -> Self {
        self.use_tls = use_tls;
        self
    }

    /// Bounds SQL execution, tunnel transfer, and the callback together with a
    /// client-side timer. Tunnel setup is not part of it.
    ///
    /// Unset by default. An elapsed bound can terminate the transport; see the
    /// `timeout_ms` field for when that happens.
    #[must_use]
    pub fn timeout_ms(mut self, timeout: u64) -> Self {
        self.timeout_ms = Some(timeout);
        self
    }

    /// Sets the Exasol host for HTTP transport connection.
    ///
    /// This is typically the same host as the WebSocket connection.
    #[must_use]
    pub fn exasol_host(mut self, host: impl Into<String>) -> Self {
        self.host = host.into();
        self
    }

    /// Sets the Exasol port for HTTP transport connection.
    ///
    /// This is typically the same port as the WebSocket connection.
    #[must_use]
    pub fn exasol_port(mut self, port: u16) -> Self {
        self.port = port;
        self
    }
}

/// Fields the Arrow and Parquet export paths share when they build the
/// `CsvExportOptions` for their underlying CSV export. Named fields, rather
/// than five positional parameters, because two of the five are adjacent
/// `char`s and a transposed pair would compile while silently producing wrong
/// CSV.
pub(crate) struct SharedCsvExportParams<'a> {
    pub column_separator: char,
    pub column_delimiter: char,
    pub host: &'a str,
    pub port: u16,
    pub use_tls: bool,
}

/// Builds the `CsvExportOptions` shared by the Arrow and Parquet export
/// paths, which pipe their own export through a CSV export and then
/// reinterpret the rows. Column headers are always disabled, because the
/// caller reinterprets rows positionally rather than by name, and no export
/// timeout is set, because neither `ArrowExportOptions` nor
/// `ParquetExportOptions` exposes one for its caller to configure.
pub(crate) fn shared_csv_export_options(params: SharedCsvExportParams<'_>) -> CsvExportOptions {
    CsvExportOptions::default()
        .column_separator(params.column_separator)
        .column_delimiter(params.column_delimiter)
        .with_column_names(false)
        .exasol_host(params.host)
        .exasol_port(params.port)
        .use_tls(params.use_tls)
}

/// Receiver end of the data pipe for processing exported data.
pub struct DataPipeReceiver {
    rx: mpsc::Receiver<Vec<u8>>,
}

impl DataPipeReceiver {
    /// Receives the next chunk of data.
    ///
    /// Returns `None` when all data has been received.
    pub async fn recv(&mut self) -> Option<Vec<u8>> {
        self.rx.recv().await
    }
}

/// Exports data from an Exasol table or query to a file.
///
/// # Arguments
///
/// * `ws_transport` - WebSocket transport for executing SQL
/// * `source` - The data source (table or query)
/// * `file_path` - Path to the output file
/// * `options` - Export options
///
/// # Returns
///
/// The number of rows exported on success.
///
/// # Errors
///
/// Returns `ExportError` if the export fails.
///
/// # Example
///
pub async fn export_to_file<T: TransportProtocol + ?Sized>(
    ws_transport: &mut T,
    source: ExportSource,
    file_path: &Path,
    options: CsvExportOptions,
) -> Result<u64, ExportError> {
    // Create the output file
    let file = File::create(file_path).await?;
    let writer = BufWriter::new(file);

    export_to_stream(ws_transport, source, writer, options).await
}

/// Exports data from an Exasol table or query to an async writer.
///
/// # Arguments
///
/// * `ws_transport` - WebSocket transport for executing SQL
/// * `source` - The data source (table or query)
/// * `writer` - Async writer to write the CSV data to
/// * `options` - Export options
///
/// # Returns
///
/// The number of rows exported on success.
///
/// # Errors
///
/// Returns `ExportError` if the export fails.
pub async fn export_to_stream<T: TransportProtocol + ?Sized, W: AsyncWrite + Unpin>(
    ws_transport: &mut T,
    source: ExportSource,
    mut writer: W,
    options: CsvExportOptions,
) -> Result<u64, ExportError> {
    let compression = options.compression;

    // Use the callback variant to process data
    export_to_callback(ws_transport, source, options, |mut receiver| async move {
        let mut row_count = 0u64;
        let mut buffer = Vec::new();

        // Collect all data first
        while let Some(chunk) = receiver.recv().await {
            buffer.extend_from_slice(&chunk);
        }

        // Decompress if needed
        let data = decompress(buffer, compression)?;

        // Write to output and count rows
        writer.write_all(&data).await?;
        writer.flush().await?;

        // Count rows by counting newlines
        for byte in &data {
            if *byte == b'\n' {
                row_count += 1;
            }
        }

        Ok(row_count)
    })
    .await
}

/// Exports data from an Exasol table or query to an in-memory list of rows.
///
/// Each row is represented as a vector of string values.
///
/// # Arguments
///
/// * `ws_transport` - WebSocket transport for executing SQL
/// * `source` - The data source (table or query)
/// * `options` - Export options
///
/// # Returns
///
/// A vector of rows, where each row is a vector of column values.
///
/// # Errors
///
/// Returns `ExportError` if the export fails.
///
/// # Example
///
pub async fn export_to_list<T: TransportProtocol + ?Sized>(
    ws_transport: &mut T,
    source: ExportSource,
    options: CsvExportOptions,
) -> Result<Vec<Vec<String>>, ExportError> {
    let separator = options.column_separator;
    let delimiter = options.column_delimiter;
    let compression = options.compression;

    export_to_callback(ws_transport, source, options, |mut receiver| async move {
        let mut buffer = Vec::new();

        // Collect all data
        while let Some(chunk) = receiver.recv().await {
            buffer.extend_from_slice(&chunk);
        }

        // Decompress if needed
        let data = decompress(buffer, compression)?;

        // Parse CSV data
        let csv_string = String::from_utf8(data).map_err(|e| ExportError::CsvParseError {
            row: 0,
            message: format!("Invalid UTF-8: {}", e),
        })?;

        parse_csv(&csv_string, separator, delimiter)
    })
    .await
}

/// Exports data from an Exasol table or query using a custom callback.
///
/// This is the most flexible export method, allowing you to process the data
/// stream however you need.
///
/// # Timeout behavior
///
/// Tunnel setup comes first and is bounded by the 30-second setup deadline of
/// `HttpTransportClient::connect`, whatever `options.timeout_ms` says. A stalled
/// setup fails with `ExportError::HttpTransportError`.
///
/// `options.timeout_ms` bounds what follows (SQL execution, tunnel transfer,
/// and the callback) and is unset by default. The tunnel read after setup has
/// no timeout of its own, so a failing EXPORT statement aborts the pending
/// transport task instead of being awaited alongside it.
///
/// When a configured bound elapses, the transport is terminated only if the EXPORT response
/// had not been read yet, because an elapse during the callback leaves the
/// transport in sync and terminating it would break a healthy connection.
/// Tracking that with an `AtomicBool` rather than a `Cell` is deliberate:
/// `&Cell<bool>` is not `Send`, which would make this future `!Send` and stop
/// callers from spawning a long export.
///
/// # Arguments
///
/// * `ws_transport` - WebSocket transport for executing SQL
/// * `source` - The data source (table or query)
/// * `options` - Export options
/// * `callback` - A callback function that receives a `DataPipeReceiver` and processes the data
///
/// # Returns
///
/// The result of the callback function.
///
/// # Errors
///
/// Returns `ExportError` if the export fails.
pub async fn export_to_callback<T, F, Fut, R>(
    ws_transport: &mut T,
    source: ExportSource,
    options: CsvExportOptions,
    callback: F,
) -> Result<R, ExportError>
where
    T: TransportProtocol + ?Sized,
    F: FnOnce(DataPipeReceiver) -> Fut,
    Fut: Future<Output = Result<R, ExportError>>,
{
    // Connect to Exasol via HTTP transport client (performs handshake automatically)
    let mut client = HttpTransportClient::connect(&options.host, options.port, options.use_tls)
        .await
        .map_err(|e| ExportError::HttpTransportError {
            message: format!("Failed to connect to Exasol: {e}"),
        })?;

    // Get internal address and fingerprint from the handshake response
    let internal_addr = client.internal_address().to_string();
    let fingerprint = client.public_key_fingerprint().map(|s| s.to_string());

    // Build the EXPORT query
    let mut query_builder = match source {
        ExportSource::Table {
            ref schema,
            ref name,
            ref columns,
        } => {
            let mut builder = ExportQuery::from_table(name);
            if let Some(s) = schema {
                builder = builder.schema(s);
            }
            if !columns.is_empty() {
                builder = builder.columns(columns.iter().map(|s| s.as_str()).collect());
            }
            builder
        }
        ExportSource::Query { ref sql } => ExportQuery::from_query(sql),
    };

    // Apply options to the query builder, using internal address from handshake
    query_builder = query_builder
        .at_address(&internal_addr)
        .column_separator(options.column_separator)
        .column_delimiter(options.column_delimiter)
        .row_separator(options.row_separator)
        .encoding(&options.encoding)
        .with_column_names(options.with_column_names)
        .compressed(options.compression);

    if let Some(ref null_val) = options.null_value {
        query_builder = query_builder.null_value(null_val);
    }

    if let Some(ref fp) = fingerprint {
        query_builder = query_builder.with_public_key(fp);
    }

    let export_sql = query_builder.build();

    // Create channel for data transfer
    let (tx, rx) = mpsc::channel::<Vec<u8>>(DEFAULT_PIPE_BUFFER_SIZE);
    let receiver = DataPipeReceiver { rx };

    // Spawn task to handle the export request from Exasol
    // This task:
    // 1. Waits for HTTP PUT request from Exasol
    // 2. Reads CSV data from PUT request body (chunked or content-length)
    // 3. Sends HTTP 200 OK response after receiving all data
    let mut http_task =
        tokio::spawn(async move {
            // Use handle_export_request() to properly handle the EXPORT protocol
            let (_request, body) = client.handle_export_request().await.map_err(|e| {
                ExportError::HttpTransportError {
                    message: format!("Failed to handle export request: {e}"),
                }
            })?;

            // Send all data to receiver
            if !body.is_empty() && tx.send(body).await.is_err() {
                return Err(ExportError::ChannelError("Receiver dropped".to_string()));
            }

            // Shutdown connection gracefully
            let _ = client.shutdown().await;

            Ok::<(), ExportError>(())
        });

    let sql_response_read = AtomicBool::new(false);

    // Execute the EXPORT SQL in parallel
    // This triggers Exasol to send data through the established connection
    //
    // The flag is set here, inside `sql_task` itself, rather than after
    // `sql_task.await` resolves in `work` below. Setting it as part of
    // `sql_task` completing (success or error) rules out any dependence on
    // `work` reaching the next statement before a cancellation could
    // intervene, regardless of how `work`'s composition evolves later.
    let sql_task = async {
        let result = ws_transport.execute_query(&export_sql).await.map_err(|e| {
            ExportError::SqlExecutionError {
                message: e.to_string(),
            }
        });
        sql_response_read.store(true, Ordering::Relaxed);
        result
    };

    // Run callback with the receiver
    let callback_task = callback(receiver);

    let work = async {
        let sql_result = sql_task.await;

        if let Err(sql_error) = sql_result {
            http_task.abort();
            return Err(sql_error);
        }

        let (http_result, callback_result) = tokio::join!(&mut http_task, callback_task);

        http_result.map_err(|e| ExportError::HttpTransportError {
            message: format!("HTTP task panicked: {e}"),
        })??;

        callback_result
    };

    let outcome = match options.timeout_ms {
        Some(bound_ms) => tokio::time::timeout(tokio::time::Duration::from_millis(bound_ms), work)
            .await
            .map_err(|_| bound_ms),
        None => Ok(work.await),
    };

    match outcome {
        Ok(result) => result,
        Err(bound_ms) => {
            http_task.abort();
            let transport_terminated = !sql_response_read.load(Ordering::Relaxed);
            if transport_terminated {
                ws_transport.terminate();
            }
            Err(ExportError::Timeout {
                timeout_ms: bound_ms,
                transport_terminated,
            })
        }
    }
}

/// Decompresses a buffer according to the configured compression.
///
/// Returns the buffer unchanged for `Compression::None`.
fn decompress(buffer: Vec<u8>, compression: Compression) -> Result<Vec<u8>, ExportError> {
    match compression {
        Compression::Gzip => {
            let decoder = GzDecoder::new(buffer.as_slice());
            let mut decompressed = Vec::new();
            std::io::Read::read_to_end(&mut std::io::BufReader::new(decoder), &mut decompressed)
                .map_err(|e| ExportError::DecompressionError(e.to_string()))?;
            Ok(decompressed)
        }
        Compression::Bzip2 => {
            let decoder = BzDecoder::new(buffer.as_slice());
            let mut decompressed = Vec::new();
            std::io::Read::read_to_end(&mut std::io::BufReader::new(decoder), &mut decompressed)
                .map_err(|e| ExportError::DecompressionError(e.to_string()))?;
            Ok(decompressed)
        }
        Compression::None => Ok(buffer),
    }
}

/// Parses CSV data into a vector of rows.
fn parse_csv(
    data: &str,
    separator: char,
    delimiter: char,
) -> Result<Vec<Vec<String>>, ExportError> {
    csv_rows(data, separator, delimiter).collect()
}

/// Reads CSV text one parsed row at a time.
///
/// A row ends at a line feed, a carriage return and line feed pair, or a lone carriage return
/// outside quotes. Line breaks inside quotes stay in the field. A doubled delimiter inside
/// quotes is one literal delimiter. The iterator yields an error when the text ends inside a
/// quote, naming the 0-based index of the row that was open, and ends after that error.
pub(crate) fn csv_rows(data: &str, separator: char, delimiter: char) -> CsvRows<'_> {
    CsvRows {
        chars: data.chars().peekable(),
        separator,
        delimiter,
        rows_yielded: 0,
    }
}

/// Row iterator returned by [`csv_rows`].
pub(crate) struct CsvRows<'a> {
    chars: std::iter::Peekable<std::str::Chars<'a>>,
    separator: char,
    delimiter: char,
    rows_yielded: usize,
}

impl CsvRows<'_> {
    fn complete_row(&mut self, mut row: Vec<String>, last_field: String) -> Vec<String> {
        row.push(last_field);
        self.rows_yielded += 1;
        row
    }
}

impl Iterator for CsvRows<'_> {
    type Item = Result<Vec<String>, ExportError>;

    fn next(&mut self) -> Option<Self::Item> {
        let mut row = Vec::new();
        let mut field = String::new();
        let mut in_quotes = false;
        let mut consumed_any = false;

        while let Some(c) = self.chars.next() {
            consumed_any = true;
            if in_quotes {
                if c != self.delimiter {
                    field.push(c);
                } else if self.chars.peek() == Some(&self.delimiter) {
                    field.push(self.delimiter);
                    self.chars.next();
                } else {
                    in_quotes = false;
                }
            } else if c == self.delimiter {
                in_quotes = true;
            } else if c == self.separator {
                row.push(std::mem::take(&mut field));
            } else if c == '\n' {
                return Some(Ok(self.complete_row(row, field)));
            } else if c == '\r' {
                // A CR before LF belongs to the CRLF pair, and the LF ends the row.
                if self.chars.peek() != Some(&'\n') {
                    return Some(Ok(self.complete_row(row, field)));
                }
            } else {
                field.push(c);
            }
        }

        if in_quotes {
            return Some(Err(ExportError::CsvParseError {
                row: self.rows_yielded,
                message: "Unclosed quote at end of data".to_string(),
            }));
        }
        if !consumed_any {
            return None;
        }
        Some(Ok(self.complete_row(row, field)))
    }
}

/// Parses a single CSV line into its fields, handling RFC-style quoting.
///
/// A field is opened/closed by `delimiter`; a doubled delimiter inside a quoted
/// field is an escaped delimiter. `separator` splits fields when not inside
/// quotes. This operates on a single already-split line and therefore does not
/// handle newlines embedded in quoted fields (callers split on newlines first).
///
/// Returns the parsed fields together with a flag indicating whether the line
/// ended while still inside a quoted field (unterminated quote). Callers decide
/// whether an unterminated quote is an error.
pub(crate) fn parse_csv_row(line: &str, separator: char, delimiter: char) -> (Vec<String>, bool) {
    let mut fields = Vec::new();
    let mut current_field = String::new();
    let mut in_quotes = false;
    let mut chars = line.chars().peekable();

    while let Some(c) = chars.next() {
        if in_quotes {
            if c == delimiter {
                // Check for escaped quote (doubled delimiter)
                if chars.peek() == Some(&delimiter) {
                    current_field.push(delimiter);
                    chars.next();
                } else {
                    in_quotes = false;
                }
            } else {
                current_field.push(c);
            }
        } else if c == delimiter {
            in_quotes = true;
        } else if c == separator {
            fields.push(current_field);
            current_field = String::new();
        } else {
            current_field.push(c);
        }
    }

    // Don't forget the last field
    fields.push(current_field);

    (fields, in_quotes)
}

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

    #[test]
    fn test_csv_rows_yields_rows_one_at_a_time_keeping_quoted_line_breaks() {
        let data = "1,\"a\nb\r c\"\n2,plain\r\n3,last";
        let mut rows = csv_rows(data, ',', '"');

        assert_eq!(rows.next().unwrap().unwrap(), vec!["1", "a\nb\r c"]);
        assert_eq!(rows.next().unwrap().unwrap(), vec!["2", "plain"]);
        assert_eq!(rows.next().unwrap().unwrap(), vec!["3", "last"]);
        assert!(rows.next().is_none());
    }

    #[test]
    fn test_csv_rows_keeps_a_final_quoted_empty_field() {
        let rows: Vec<_> = csv_rows("\"\"", ',', '"')
            .collect::<Result<_, _>>()
            .unwrap();
        assert_eq!(rows, vec![vec![String::new()]]);
    }

    #[test]
    fn test_csv_rows_ends_after_an_unclosed_quote_error() {
        let mut rows = csv_rows("1,x\n2,\"open", ',', '"');

        assert_eq!(rows.next().unwrap().unwrap(), vec!["1", "x"]);
        match rows.next().unwrap() {
            Err(ExportError::CsvParseError { row, .. }) => assert_eq!(row, 1),
            other => panic!("expected an unclosed quote error, got {other:?}"),
        }
        assert!(rows.next().is_none());
    }

    #[test]
    fn test_csv_rows_ends_a_row_at_a_lone_carriage_return_outside_quotes() {
        let mut rows = csv_rows("1,a\r2,b\r", ',', '"');

        assert_eq!(rows.next().unwrap().unwrap(), vec!["1", "a"]);
        assert_eq!(rows.next().unwrap().unwrap(), vec!["2", "b"]);
        assert!(rows.next().is_none());
    }
    use crate::error::TransportError;
    use crate::transport::protocol::QueryResult;
    use crate::transport::test_support::{FakeExasolServer, MockTransport, StalledQueryTransport};
    use std::time::Duration;

    // Tests for CsvExportOptions

    #[test]
    fn test_csv_export_options_default() {
        let options = CsvExportOptions::default();

        assert_eq!(options.column_separator, ',');
        assert_eq!(options.column_delimiter, '"');
        assert_eq!(options.row_separator, RowSeparator::LF);
        assert_eq!(options.encoding, "UTF-8");
        assert!(options.null_value.is_none());
        assert_eq!(options.compression, Compression::None);
        assert!(!options.with_column_names);
        assert!(!options.use_tls);
        assert_eq!(options.timeout_ms, None);
        assert_eq!(options.host, "");
        assert_eq!(options.port, 0);
    }

    #[test]
    fn test_csv_export_options_builder() {
        let options = CsvExportOptions::new()
            .column_separator(';')
            .column_delimiter('\'')
            .row_separator(RowSeparator::CRLF)
            .encoding("ISO-8859-1")
            .null_value("NULL")
            .compression(Compression::Gzip)
            .with_column_names(true)
            .use_tls(false)
            .timeout_ms(60_000)
            .exasol_host("exasol.example.com")
            .exasol_port(8563);

        assert_eq!(options.column_separator, ';');
        assert_eq!(options.column_delimiter, '\'');
        assert_eq!(options.row_separator, RowSeparator::CRLF);
        assert_eq!(options.encoding, "ISO-8859-1");
        assert_eq!(options.null_value, Some("NULL".to_string()));
        assert_eq!(options.compression, Compression::Gzip);
        assert!(options.with_column_names);
        assert!(!options.use_tls);
        assert_eq!(options.timeout_ms, Some(60_000));
        assert_eq!(options.host, "exasol.example.com");
        assert_eq!(options.port, 8563);
    }

    #[test]
    fn test_shared_csv_options_configure_no_export_timeout() {
        let options = shared_csv_export_options(SharedCsvExportParams {
            column_separator: ';',
            column_delimiter: '\'',
            host: "exasol.example.com",
            port: 8563,
            use_tls: true,
        });

        assert_eq!(options.timeout_ms, None);
    }

    // Tests for export_to_callback

    /// Pins the `Send` guarantee of the export future, which every public
    /// `Connection::export_*` method awaits and which a caller may hand to
    /// `tokio::spawn`. The type check is the whole assertion: the future is
    /// never polled, so no transport call happens.
    #[test]
    fn test_export_to_callback_future_is_send() {
        fn assert_send<T: Send>(_: T) {}

        let mut transport = MockTransport::new();
        let export = export_to_callback(
            &mut transport,
            ExportSource::Query {
                sql: "SELECT 1".to_string(),
            },
            CsvExportOptions::default(),
            |_receiver: DataPipeReceiver| async { Ok::<(), ExportError>(()) },
        );

        assert_send(export);
    }

    /// Short enough to keep the deadline tests quick, yet orders of magnitude
    /// above the loopback round trip they race against, so which side wins is
    /// decided by the scenario and not by scheduling noise.
    const SHORT_DEADLINE_MS: u64 = 50;

    /// Longer than any loopback exchange these tests perform, so an export
    /// that completes normally can never trip it.
    const UNREACHABLE_DEADLINE_MS: u64 = 30_000;

    /// Far beyond the test's own lifetime: whatever awaits it never finishes.
    const NEVER: Duration = Duration::from_secs(3600);

    /// The fixed five-minute default bound this release removed (#52).
    const REMOVED_DEFAULT_BOUND_MS: u64 = 300_000;

    fn tunnel_options(server: &FakeExasolServer) -> CsvExportOptions {
        CsvExportOptions::default()
            .exasol_host(&server.host)
            .exasol_port(server.port)
            .use_tls(false)
    }

    async fn collect_body(mut receiver: DataPipeReceiver) -> Result<String, ExportError> {
        let mut collected = Vec::new();
        while let Some(chunk) = receiver.recv().await {
            collected.extend_from_slice(&chunk);
        }
        String::from_utf8(collected).map_err(|e| ExportError::CsvParseError {
            row: 0,
            message: e.to_string(),
        })
    }

    fn succeeding_transport() -> MockTransport {
        let mut transport = MockTransport::new();
        transport
            .expect_execute_query()
            .returning(|_| Ok(QueryResult::row_count(1)));
        transport
    }

    fn unwrap_timeout(error: ExportError) -> (u64, bool) {
        let ExportError::Timeout {
            timeout_ms,
            transport_terminated,
        } = error
        else {
            panic!("expected a Timeout, got {error:?}");
        };
        (timeout_ms, transport_terminated)
    }

    #[tokio::test]
    async fn test_export_to_callback_without_a_deadline_delivers_the_tunnel_body() {
        let server = FakeExasolServer::serving_csv("1,alice\n2,bob\n").await;
        let mut transport = succeeding_transport();

        let delivered = export_to_callback(
            &mut transport,
            ExportSource::Query {
                sql: "SELECT * FROM users".to_string(),
            },
            tunnel_options(&server),
            collect_body,
        )
        .await
        .expect("an export with no deadline must run to completion");

        assert_eq!(delivered, "1,alice\n2,bob\n");
    }

    /// The unset default arms no timer at all, not merely a longer one — the
    /// executable guard on `No client-side export timeout by default`
    /// (`specs/import-export/csv-export-timeout`), which asserting the field
    /// alone would not provide.
    ///
    /// The clock is paused only after the export has had real time to open its
    /// tunnel and enter the SQL leg: auto-advance would otherwise jump past a
    /// timer that was not armed yet. Starting late cannot turn this red, since
    /// `StalledQueryTransport::execute_query` never resolves.
    ///
    /// Scenario: No client-side export timeout by default
    #[tokio::test]
    async fn test_export_to_callback_arms_no_timer_when_the_deadline_is_left_unset() {
        let server = FakeExasolServer::silent_after_handshake().await;
        let mut mock = MockTransport::new();
        // Only reached if the guard regresses; allowed so the panic below
        // reports that rather than an unexpected mock call.
        mock.expect_terminate().return_const(());
        let mut transport = StalledQueryTransport::new(mock);

        let options = tunnel_options(&server);
        assert_eq!(
            options.timeout_ms, None,
            "the export below only tests the unset default if the options carry it"
        );

        let export = export_to_callback(
            &mut transport,
            ExportSource::Query {
                sql: "SELECT * FROM users".to_string(),
            },
            options,
            collect_body,
        );
        tokio::pin!(export);

        let started = tokio::time::timeout(Duration::from_millis(500), &mut export).await;
        assert!(
            started.is_err(),
            "an EXPORT statement that never responds cannot let the export finish"
        );

        tokio::time::pause();
        tokio::time::advance(Duration::from_millis(REMOVED_DEFAULT_BOUND_MS * 2)).await;

        if let Ok(outcome) = tokio::time::timeout(Duration::from_millis(50), &mut export).await {
            panic!(
                "the export stopped waiting after {}ms, so an unset timeout_ms still armed a \
                 client-side timer: {outcome:?}",
                REMOVED_DEFAULT_BOUND_MS * 2
            );
        }
    }

    /// Starts an export against a tunnel peer that never answers the magic
    /// packet and returns its error once the paused clock reaches the setup
    /// deadline. The clock pauses only after the peer holds the magic packet,
    /// because auto-advance would otherwise fire the deadline during the connect.
    async fn export_against_a_stalled_tunnel_setup(timeout_ms: Option<u64>) -> ExportError {
        let mut server = FakeExasolServer::silent_before_handshake().await;
        // No expectations: the EXPORT statement must not be sent.
        let mut transport = MockTransport::new();
        let mut options = tunnel_options(&server);
        if let Some(timeout_ms) = timeout_ms {
            options = options.timeout_ms(timeout_ms);
        }

        let export = export_to_callback(
            &mut transport,
            ExportSource::Query {
                sql: "SELECT * FROM users".to_string(),
            },
            options,
            collect_body,
        );
        tokio::pin!(export);

        tokio::select! {
            outcome = &mut export => panic!("the export ended before the peer saw its handshake: {outcome:?}"),
            () = server.wait_for_magic_packet() => {}
        }
        tokio::time::pause();
        export
            .await
            .expect_err("a stalled tunnel setup must fail the export")
    }

    fn assert_tunnel_setup_timed_out_at_the_exa_handshake(error: ExportError) {
        let ExportError::HttpTransportError { message } = error else {
            panic!("expected an HttpTransportError, got {error:?}");
        };
        assert!(
            message.contains("HTTP tunnel setup timeout after 30000ms (EXA handshake)"),
            "{message}"
        );
    }

    /// Scenario: Tunnel setup is bounded by 30 seconds by default
    /// Scenario: No client-side export timeout by default
    #[tokio::test]
    async fn test_export_to_callback_bounds_a_stalled_tunnel_setup_by_30_seconds() {
        let error = export_against_a_stalled_tunnel_setup(None).await;

        assert_tunnel_setup_timed_out_at_the_exa_handshake(error);
    }

    /// Scenario: Tunnel setup is bounded by 30 seconds by default
    #[tokio::test]
    async fn test_export_to_callback_bounds_a_stalled_tunnel_setup_even_with_an_export_timeout() {
        let error = export_against_a_stalled_tunnel_setup(Some(1_000)).await;

        assert_tunnel_setup_timed_out_at_the_exa_handshake(error);
    }

    #[tokio::test]
    async fn test_export_to_callback_within_its_deadline_delivers_the_tunnel_body() {
        let server = FakeExasolServer::serving_csv("1,alice\n").await;
        let mut transport = succeeding_transport();

        let delivered = export_to_callback(
            &mut transport,
            ExportSource::Query {
                sql: "SELECT * FROM users".to_string(),
            },
            tunnel_options(&server).timeout_ms(UNREACHABLE_DEADLINE_MS),
            collect_body,
        )
        .await
        .expect("an export that beats its deadline must not report a timeout");

        assert_eq!(delivered, "1,alice\n");
    }

    /// The tunnel read has no deadline of its own, so a failing EXPORT
    /// statement has to abort the tunnel task instead of being awaited
    /// alongside it. The peer here never sends the `PUT`, so anything less
    /// than a hard abort hangs.
    #[tokio::test]
    async fn test_export_to_callback_abandons_the_tunnel_when_the_export_statement_fails() {
        let server = FakeExasolServer::silent_after_handshake().await;
        let mut transport = MockTransport::new();
        transport.expect_execute_query().returning(|_| {
            Err(TransportError::ProtocolError(
                "object USERS not found".to_string(),
            ))
        });

        let error = tokio::time::timeout(
            Duration::from_secs(10),
            export_to_callback(
                &mut transport,
                ExportSource::Query {
                    sql: "SELECT * FROM users".to_string(),
                },
                tunnel_options(&server),
                collect_body,
            ),
        )
        .await
        .expect("a failed EXPORT statement must not leave the export waiting on the tunnel")
        .expect_err("a failed EXPORT statement must fail the export");

        let ExportError::SqlExecutionError { message } = error else {
            panic!("expected a SqlExecutionError, got {error:?}");
        };
        assert!(
            message.contains("object USERS not found"),
            "the statement failure must reach the caller intact: {message}"
        );
    }

    #[tokio::test]
    async fn test_export_to_callback_terminates_the_transport_when_the_deadline_precedes_the_sql_response(
    ) {
        let server = FakeExasolServer::silent_after_handshake().await;
        let mut mock = MockTransport::new();
        mock.expect_terminate().times(1).return_const(());
        let mut transport = StalledQueryTransport::new(mock);

        let error = export_to_callback(
            &mut transport,
            ExportSource::Query {
                sql: "SELECT * FROM users".to_string(),
            },
            tunnel_options(&server).timeout_ms(SHORT_DEADLINE_MS),
            collect_body,
        )
        .await
        .expect_err("an outstanding EXPORT statement must fail once the deadline elapses");

        let (timeout_ms, transport_terminated) = unwrap_timeout(error);
        assert_eq!(timeout_ms, SHORT_DEADLINE_MS);
        assert!(
            transport_terminated,
            "an unread EXPORT response can no longer be matched to a request"
        );
    }

    #[tokio::test]
    async fn test_export_to_callback_keeps_the_transport_when_the_deadline_falls_in_the_callback() {
        let server = FakeExasolServer::serving_csv("1,alice\n").await;
        let mut transport = succeeding_transport();
        transport.expect_terminate().never();

        let error = export_to_callback(
            &mut transport,
            ExportSource::Query {
                sql: "SELECT * FROM users".to_string(),
            },
            tunnel_options(&server).timeout_ms(SHORT_DEADLINE_MS),
            |_receiver: DataPipeReceiver| async {
                tokio::time::sleep(NEVER).await;
                Ok::<String, ExportError>(String::new())
            },
        )
        .await
        .expect_err("a callback that outlives the deadline must fail the export");

        let (timeout_ms, transport_terminated) = unwrap_timeout(error);
        assert_eq!(timeout_ms, SHORT_DEADLINE_MS);
        assert!(
            !transport_terminated,
            "the EXPORT response was already read, so the transport is still in sync"
        );
    }

    // Tests for CSV parsing

    #[test]
    fn test_parse_csv_simple() {
        let data = "a,b,c\n1,2,3\n";
        let result = parse_csv(data, ',', '"').unwrap();

        assert_eq!(result.len(), 2);
        assert_eq!(result[0], vec!["a", "b", "c"]);
        assert_eq!(result[1], vec!["1", "2", "3"]);
    }

    #[test]
    fn test_parse_csv_with_quotes() {
        let data = "\"hello\",\"world\"\n\"foo\",\"bar\"\n";
        let result = parse_csv(data, ',', '"').unwrap();

        assert_eq!(result.len(), 2);
        assert_eq!(result[0], vec!["hello", "world"]);
        assert_eq!(result[1], vec!["foo", "bar"]);
    }

    #[test]
    fn test_parse_csv_with_escaped_quotes() {
        let data = "\"hello \"\"world\"\"\",normal\n";
        let result = parse_csv(data, ',', '"').unwrap();

        assert_eq!(result.len(), 1);
        assert_eq!(result[0], vec!["hello \"world\"", "normal"]);
    }

    #[test]
    fn test_parse_csv_with_separator_in_quotes() {
        let data = "\"a,b,c\",\"d,e\"\n";
        let result = parse_csv(data, ',', '"').unwrap();

        assert_eq!(result.len(), 1);
        assert_eq!(result[0], vec!["a,b,c", "d,e"]);
    }

    #[test]
    fn test_parse_csv_with_newline_in_quotes() {
        let data = "\"line1\nline2\",normal\n";
        let result = parse_csv(data, ',', '"').unwrap();

        assert_eq!(result.len(), 1);
        assert_eq!(result[0], vec!["line1\nline2", "normal"]);
    }

    #[test]
    fn test_parse_csv_with_crlf() {
        let data = "a,b\r\nc,d\r\n";
        let result = parse_csv(data, ',', '"').unwrap();

        assert_eq!(result.len(), 2);
        assert_eq!(result[0], vec!["a", "b"]);
        assert_eq!(result[1], vec!["c", "d"]);
    }

    #[test]
    fn test_parse_csv_with_semicolon_separator() {
        let data = "a;b;c\n1;2;3\n";
        let result = parse_csv(data, ';', '"').unwrap();

        assert_eq!(result.len(), 2);
        assert_eq!(result[0], vec!["a", "b", "c"]);
        assert_eq!(result[1], vec!["1", "2", "3"]);
    }

    #[test]
    fn test_parse_csv_empty_fields() {
        let data = "a,,c\n,b,\n";
        let result = parse_csv(data, ',', '"').unwrap();

        assert_eq!(result.len(), 2);
        assert_eq!(result[0], vec!["a", "", "c"]);
        assert_eq!(result[1], vec!["", "b", ""]);
    }

    #[test]
    fn test_parse_csv_unclosed_quote_error() {
        let data = "\"unclosed,quote\n";
        let result = parse_csv(data, ',', '"');

        assert!(result.is_err());
        if let Err(ExportError::CsvParseError { row, message }) = result {
            assert_eq!(row, 0);
            assert!(message.contains("Unclosed quote"));
        } else {
            panic!("Expected CsvParseError");
        }
    }

    #[test]
    fn test_parse_csv_empty_data() {
        let data = "";
        let result = parse_csv(data, ',', '"').unwrap();

        assert!(result.is_empty());
    }

    #[test]
    fn test_parse_csv_single_field() {
        let data = "single\n";
        let result = parse_csv(data, ',', '"').unwrap();

        assert_eq!(result.len(), 1);
        assert_eq!(result[0], vec!["single"]);
    }

    #[test]
    fn test_parse_csv_no_trailing_newline() {
        let data = "a,b,c";
        let result = parse_csv(data, ',', '"').unwrap();

        assert_eq!(result.len(), 1);
        assert_eq!(result[0], vec!["a", "b", "c"]);
    }

    // Tests for ExportError

    #[test]
    fn test_export_error_display() {
        let err = ExportError::IoError(io::Error::new(io::ErrorKind::NotFound, "file not found"));
        assert!(err.to_string().contains("I/O error"));

        let err = ExportError::HttpTransportError {
            message: "connection refused".to_string(),
        };
        assert!(err.to_string().contains("HTTP transport error"));
        assert!(err.to_string().contains("connection refused"));

        let err = ExportError::CsvParseError {
            row: 5,
            message: "invalid data".to_string(),
        };
        assert!(err.to_string().contains("row 5"));
        assert!(err.to_string().contains("invalid data"));

        let err = ExportError::Timeout {
            timeout_ms: 5000,
            transport_terminated: true,
        };
        assert_eq!(
            err.to_string(),
            "Export timed out after 5000ms; the transport was terminated, \
             reconnect before the next operation"
        );

        let err = ExportError::Timeout {
            timeout_ms: 5000,
            transport_terminated: false,
        };
        assert_eq!(
            err.to_string(),
            "Export timed out after 5000ms; the transport is still usable"
        );
    }

    #[test]
    fn test_export_error_from_io_error() {
        let io_err = io::Error::new(io::ErrorKind::PermissionDenied, "access denied");
        let export_err: ExportError = io_err.into();

        assert!(matches!(export_err, ExportError::IoError(_)));
    }

    // Tests for DataPipeReceiver

    #[tokio::test]
    async fn test_data_pipe_receiver_recv() {
        let (tx, rx) = mpsc::channel::<Vec<u8>>(16);
        let mut receiver = DataPipeReceiver { rx };

        tx.send(vec![1, 2, 3]).await.unwrap();
        tx.send(vec![4, 5, 6]).await.unwrap();
        drop(tx);

        let chunk1 = receiver.recv().await;
        assert_eq!(chunk1, Some(vec![1, 2, 3]));

        let chunk2 = receiver.recv().await;
        assert_eq!(chunk2, Some(vec![4, 5, 6]));

        let chunk3 = receiver.recv().await;
        assert!(chunk3.is_none());
    }

    #[tokio::test]
    async fn test_data_pipe_receiver_empty() {
        let (tx, rx) = mpsc::channel::<Vec<u8>>(16);
        let mut receiver = DataPipeReceiver { rx };

        drop(tx);

        let chunk = receiver.recv().await;
        assert!(chunk.is_none());
    }

    // Tests for CSV parsing with column headers

    #[test]
    fn test_parse_csv_with_column_headers() {
        let data = "id,name,email\n1,Alice,alice@example.com\n2,Bob,bob@example.com\n";
        let result = parse_csv(data, ',', '"').unwrap();

        assert_eq!(result.len(), 3);
        // First row is the header
        assert_eq!(result[0], vec!["id", "name", "email"]);
        // Data rows
        assert_eq!(result[1], vec!["1", "Alice", "alice@example.com"]);
        assert_eq!(result[2], vec!["2", "Bob", "bob@example.com"]);
    }

    #[test]
    fn test_parse_csv_with_quoted_headers() {
        let data = "\"Column A\",\"Column B\",\"Column C\"\n1,2,3\n";
        let result = parse_csv(data, ',', '"').unwrap();

        assert_eq!(result.len(), 2);
        assert_eq!(result[0], vec!["Column A", "Column B", "Column C"]);
        assert_eq!(result[1], vec!["1", "2", "3"]);
    }

    // Tests for decompression

    #[test]
    fn test_gzip_decompression() {
        use flate2::write::GzEncoder;
        use std::io::Write;

        // Create gzip compressed data
        let original = b"hello,world\n1,2\n";
        let mut encoder = GzEncoder::new(Vec::new(), flate2::Compression::default());
        encoder.write_all(original).unwrap();
        let compressed = encoder.finish().unwrap();

        // Decompress using our logic
        let decoder = GzDecoder::new(compressed.as_slice());
        let mut decompressed = Vec::new();
        std::io::Read::read_to_end(&mut std::io::BufReader::new(decoder), &mut decompressed)
            .unwrap();

        assert_eq!(decompressed, original);

        // Parse the decompressed CSV
        let csv_string = String::from_utf8(decompressed).unwrap();
        let rows = parse_csv(&csv_string, ',', '"').unwrap();
        assert_eq!(rows.len(), 2);
        assert_eq!(rows[0], vec!["hello", "world"]);
        assert_eq!(rows[1], vec!["1", "2"]);
    }

    #[test]
    fn test_bzip2_decompression() {
        use bzip2::write::BzEncoder;
        use std::io::Write;

        // Create bzip2 compressed data
        let original = b"foo,bar\na,b\n";
        let mut encoder = BzEncoder::new(Vec::new(), bzip2::Compression::default());
        encoder.write_all(original).unwrap();
        let compressed = encoder.finish().unwrap();

        // Decompress using our logic
        let decoder = BzDecoder::new(compressed.as_slice());
        let mut decompressed = Vec::new();
        std::io::Read::read_to_end(&mut std::io::BufReader::new(decoder), &mut decompressed)
            .unwrap();

        assert_eq!(decompressed, original);

        // Parse the decompressed CSV
        let csv_string = String::from_utf8(decompressed).unwrap();
        let rows = parse_csv(&csv_string, ',', '"').unwrap();
        assert_eq!(rows.len(), 2);
        assert_eq!(rows[0], vec!["foo", "bar"]);
        assert_eq!(rows[1], vec!["a", "b"]);
    }

    // Tests for export options with column names

    #[test]
    fn test_csv_export_options_with_column_names() {
        let options = CsvExportOptions::new().with_column_names(true);
        assert!(options.with_column_names);

        let options = CsvExportOptions::new().with_column_names(false);
        assert!(!options.with_column_names);
    }

    // Tests for export options with compression

    #[test]
    fn test_csv_export_options_compression_gzip() {
        let options = CsvExportOptions::new().compression(Compression::Gzip);
        assert_eq!(options.compression, Compression::Gzip);
    }

    #[test]
    fn test_csv_export_options_compression_bzip2() {
        let options = CsvExportOptions::new().compression(Compression::Bzip2);
        assert_eq!(options.compression, Compression::Bzip2);
    }

    #[test]
    fn test_csv_export_options_compression_none() {
        let options = CsvExportOptions::new().compression(Compression::None);
        assert_eq!(options.compression, Compression::None);
    }

    // Tests for error types

    #[test]
    fn test_export_error_decompression() {
        let err = ExportError::DecompressionError("invalid gzip header".to_string());
        assert!(err.to_string().contains("Decompression error"));
        assert!(err.to_string().contains("invalid gzip header"));
    }

    #[test]
    fn test_export_error_channel() {
        let err = ExportError::ChannelError("receiver dropped".to_string());
        assert!(err.to_string().contains("Channel error"));
        assert!(err.to_string().contains("receiver dropped"));
    }

    #[test]
    fn test_export_error_cancelled() {
        let err = ExportError::Cancelled;
        assert!(err.to_string().contains("cancelled"));
    }

    // Test DataPipeReceiver with multiple chunks

    #[tokio::test]
    async fn test_data_pipe_receiver_multiple_chunks() {
        let (tx, rx) = mpsc::channel::<Vec<u8>>(DEFAULT_PIPE_BUFFER_SIZE);
        let mut receiver = DataPipeReceiver { rx };

        // Send CSV data in chunks
        let chunk1 = b"id,name\n".to_vec();
        let chunk2 = b"1,Alice\n".to_vec();
        let chunk3 = b"2,Bob\n".to_vec();

        tx.send(chunk1.clone()).await.unwrap();
        tx.send(chunk2.clone()).await.unwrap();
        tx.send(chunk3.clone()).await.unwrap();
        drop(tx);

        // Receive and concatenate
        let mut buffer = Vec::new();
        while let Some(chunk) = receiver.recv().await {
            buffer.extend_from_slice(&chunk);
        }

        let csv_string = String::from_utf8(buffer).unwrap();
        let rows = parse_csv(&csv_string, ',', '"').unwrap();

        assert_eq!(rows.len(), 3);
        assert_eq!(rows[0], vec!["id", "name"]);
        assert_eq!(rows[1], vec!["1", "Alice"]);
        assert_eq!(rows[2], vec!["2", "Bob"]);
    }
}