mssql-tds 0.1.0

// Copyright (c) Microsoft Corporation.
// Licensed under the MIT License.

//! Bulk copy operations for high-performance data insertion.
//!
//! This module provides the `BulkCopy` API for efficiently inserting large amounts of data
//! into SQL Server tables. It uses the TDS bulk load protocol for optimal performance.
//!
//! # Example
//!
//! ```rust,ignore
//! use mssql_tds::connection::bulk_copy::{BulkCopy, BulkCopyRow};
//! use mssql_tds::datatypes::column_values::ColumnValues;
//!
//! // Define a struct to represent your data
//! struct User {
//!     id: i32,
//!     name: String,
//!     email: String,
//! }
//!
//! // Implement BulkCopyRow for your struct
//! impl BulkCopyRow for User {
//!     fn to_column_values(&self) -> Vec<ColumnValues> {
//!         vec![
//!             ColumnValues::Int(self.id),
//!             ColumnValues::String(self.name.clone().into()),
//!             ColumnValues::String(self.email.clone().into()),
//!         ]
//!     }
//! }
//!
//! // Use bulk copy - metadata is automatically retrieved from SQL Server
//! let users = vec![
//!     User { id: 1, name: "Alice".to_string(), email: "alice@example.com".to_string() },
//!     User { id: 2, name: "Bob".to_string(), email: "bob@example.com".to_string() },
//! ];
//! let mut bulk_copy = BulkCopy::new(&mut client, "Users");
//! bulk_copy
//!     .batch_size(5000)
//!     .timeout(30);
//!
//! let result = bulk_copy.write_to_server(users.into_iter()).await?;
//! println!("Copied {} rows", result.rows_affected);
//! ```

use crate::connection::bulk_copy_state::{ATTENTION_TIMEOUT_SECONDS, BulkCopyTimeoutState};
use crate::connection::metadata_retriever::{FmtOnlyMetadataRetriever, MetadataRetriever};
use crate::connection::tds_client::TdsClient;
use crate::core::TdsResult;
use crate::datatypes::bulk_copy_metadata::{BulkCopyColumnMetadata, SqlDbType};
use crate::error::Error;
use crate::error::SqlInfoMessage;
use crate::error::bulk_copy_errors::{
    BulkCopyAttentionTimeoutError, BulkCopyError, BulkCopyTimeoutError,
};
use crate::message::transaction_management::TransactionIsolationLevel;
use async_trait::async_trait;
use std::time::{Duration, Instant};

/// Options for configuring bulk copy operations.
///
/// These options control various aspects of the bulk copy behavior,
/// such as batch size, timeout, and SQL Server-specific options.
#[derive(Debug, Clone)]
pub struct BulkCopyOptions {
    /// Number of rows in each batch. Default: 0 (all rows in one batch)
    pub batch_size: usize,

    /// Timeout for the operation in seconds. Default: 30. A value of `0`
    /// means unlimited.
    pub timeout_sec: u32,

    /// Check constraints on the destination table. Default: false
    pub check_constraints: bool,

    /// Enable triggers on the destination table. Default: false
    pub fire_triggers: bool,

    /// Preserve source identity values. Default: false
    /// If false, identity values are auto-generated by SQL Server.
    pub keep_identity: bool,

    /// Preserve NULL values regardless of defaults. Default: false
    pub keep_nulls: bool,

    /// Obtain a bulk update lock for the duration of the operation. Default: false
    pub table_lock: bool,

    /// Use an internal transaction for the bulk copy. Default: false
    ///
    /// When `true`, each batch is wrapped in its own transaction:
    /// - BEGIN TRANSACTION before each batch
    /// - COMMIT TRANSACTION on successful batch completion
    /// - ROLLBACK TRANSACTION on batch failure
    ///
    /// When `false` (the default, matching .NET SqlBulkCopy behavior):
    /// - If the connection has an active API transaction, the bulk copy
    ///   participates in that transaction
    /// - If no transaction exists, SQL Server's autocommit mode applies
    ///   (each batch is auto-committed when the DONE packet is acknowledged)
    ///
    /// **Note**: This option cannot be used when the connection already has
    /// an active transaction. Attempting to do so will result in an error.
    pub use_internal_transaction: bool,

    /// Number of rows to process before calling the progress callback.
    /// Default: 0 (no progress notifications)
    pub notification_interval: usize,

    /// Write already-encrypted (ciphertext) values into Always Encrypted
    /// columns verbatim, without the driver encrypting them. Default: false.
    ///
    /// This is the counterpart to .NET
    /// `SqlBulkCopyOptions.AllowEncryptedValueModifications`. When `true`, a
    /// value targeting an encrypted destination column is sent as-is (it must
    /// already be varbinary ciphertext) instead of being normalized and
    /// encrypted with the column's key, so the plaintext CEK is not required.
    ///
    /// # Precondition
    ///
    /// This option only takes effect when Always Encrypted is enabled on the
    /// connection (Column Encryption Setting = Enabled and the server
    /// acknowledged the feature) and the destination has encrypted columns. When
    /// column encryption is not enabled, the option is ignored and values follow
    /// the normal plaintext path.
    ///
    /// # Security
    ///
    /// The server does not validate that the supplied ciphertext matches the
    /// destination column's encryption configuration. Only enable this to move
    /// ciphertext between columns encrypted with the same key, algorithm, and
    /// encryption type; otherwise the destination will hold values that cannot
    /// be decrypted. The same caveats as the .NET option apply.
    pub allow_encrypted_value_modifications: bool,
}

impl Default for BulkCopyOptions {
    fn default() -> Self {
        Self {
            batch_size: 0,
            timeout_sec: 30,
            check_constraints: false,
            fire_triggers: false,
            keep_identity: false,
            keep_nulls: false,
            table_lock: false,
            use_internal_transaction: false, // Matches .NET SqlBulkCopy default
            notification_interval: 0,
            allow_encrypted_value_modifications: false,
        }
    }
}

impl BulkCopyOptions {
    /// Create a new `BulkCopyOptions` with default values.
    pub fn new() -> Self {
        Self::default()
    }

    /// Validate the options.
    pub fn validate(&self) -> TdsResult<()> {
        if self.batch_size > 1_000_000 {
            return Err(crate::error::Error::UsageError(
                "batch_size cannot exceed 1,000,000".to_string(),
            ));
        }
        Ok(())
    }
}

/// Specifies how to map source columns to destination columns.
#[derive(Debug, Clone)]
pub enum ColumnMappingSource {
    /// Map by column name
    Name(String),
    /// Map by column ordinal (0-based)
    Ordinal(usize),
}

/// Defines a mapping between a source column and a destination column.
#[derive(Debug, Clone)]
pub struct ColumnMapping {
    /// Source column (by name or ordinal)
    pub source: ColumnMappingSource,
    /// Destination column name
    pub destination: String,
}

impl ColumnMapping {
    /// Create a mapping from source column name to destination column name.
    pub fn by_name(source: impl Into<String>, destination: impl Into<String>) -> Self {
        Self {
            source: ColumnMappingSource::Name(source.into()),
            destination: destination.into(),
        }
    }

    /// Create a mapping from source column ordinal to destination column name.
    pub fn by_ordinal(source_ordinal: usize, destination: impl Into<String>) -> Self {
        Self {
            source: ColumnMappingSource::Ordinal(source_ordinal),
            destination: destination.into(),
        }
    }
}

/// A resolved mapping from source column index to destination column index.
///
/// This is the result of resolving user-provided column mappings against
/// the actual destination table metadata.
#[derive(Debug, Clone)]
pub struct ResolvedColumnMapping {
    /// Source column index (0-based)
    pub source_index: usize,
    /// Destination column index (0-based)
    pub destination_index: usize,
    /// Destination column name
    pub destination_name: String,
    /// Expected destination type
    pub destination_type: SqlDbType,
}

/// Progress information for bulk copy operations.
///
/// Passed to the progress callback to report the current state of the operation.
#[derive(Debug, Clone)]
pub struct BulkCopyProgress {
    /// Number of rows copied so far
    pub rows_copied: u64,
    /// Total number of rows (if known)
    pub total_rows: Option<u64>,
    /// Elapsed time since the operation started
    pub elapsed: Duration,
    /// Throughput in rows per second
    pub rows_per_second: f64,
}

impl BulkCopyProgress {
    /// Calculate progress percentage (0.0 to 100.0).
    pub fn percentage(&self) -> Option<f64> {
        self.total_rows
            .map(|total| (self.rows_copied as f64 / total as f64) * 100.0)
    }

    /// Estimate time remaining based on current throughput.
    pub fn estimated_time_remaining(&self) -> Option<Duration> {
        if self.rows_per_second <= 0.0 {
            return None;
        }

        self.total_rows.map(|total| {
            let remaining_rows = total.saturating_sub(self.rows_copied) as f64;
            Duration::from_secs_f64(remaining_rows / self.rows_per_second)
        })
    }
}

/// Result of a bulk copy operation.
///
/// Contains statistics about the completed operation.
#[derive(Debug, Clone)]
pub struct BulkCopyResult {
    /// Number of rows serialized to the wire by the client, matching
    /// `Microsoft.Data.SqlClient`'s `SqlBulkCopy.RowsCopied`. This is a
    /// client-side count, not a guarantee of the number of rows inserted on
    /// the server: server-side triggers or an `IGNORE_DUP_KEY` index can make
    /// the committed row count differ. It is unaffected by distributed engines
    /// that acknowledge one load with multiple DONE_COUNT tokens (issue #209).
    pub rows_affected: u64,
    /// Time taken for the operation
    pub elapsed: Duration,
    /// Throughput in rows per second
    pub rows_per_second: f64,
}

impl BulkCopyResult {
    /// Create a new result from the given statistics.
    pub fn new(rows_affected: u64, elapsed: Duration) -> Self {
        let rows_per_second = if elapsed.as_secs_f64() > 0.0 {
            rows_affected as f64 / elapsed.as_secs_f64()
        } else {
            0.0
        };

        Self {
            rows_affected,
            elapsed,
            rows_per_second,
        }
    }
}

/// High-level bulk copy API for efficiently inserting large amounts of data.
///
/// `BulkCopy` provides a convenient builder-style API for configuring and executing
/// bulk copy operations. It handles batching, progress reporting, and error handling.
///
/// # Example
///
/// ```rust,ignore
/// use mssql_tds::connection::bulk_copy::{BulkCopy, BulkCopyRow, ColumnMapping};
/// use std::time::Duration;
///
/// let mut bulk_copy = BulkCopy::new(&mut client, "Users")
///     .batch_size(5000)
///     .timeout(Duration::from_secs(60))
///     .keep_identity(true)
///     .add_column_mapping(ColumnMapping::by_name("SourceId", "Id"))
///     .add_column_mapping(ColumnMapping::by_name("SourceName", "Name"));
///
/// let rows = vec![/* your data */];
/// let result = bulk_copy.write_to_server(rows.into_iter()).await?;
/// println!("Copied {} rows in {:?}", result.rows_affected, result.elapsed);
/// ```
pub struct BulkCopy<'a> {
    /// Reference to the TDS client connection
    client: &'a mut TdsClient,

    /// Name of the destination table
    table_name: String,

    /// Bulk copy options
    options: BulkCopyOptions,

    /// Column mappings (empty means use source column order)
    column_mappings: Vec<ColumnMapping>,

    /// Progress callback
    progress_callback: Option<Box<dyn FnMut(BulkCopyProgress) + Send + 'a>>,

    /// Cached destination table metadata (retrieved from server)
    destination_metadata: Option<Vec<BulkCopyColumnMetadata>>,

    /// Metadata retriever strategy
    metadata_retriever: Box<dyn MetadataRetriever + 'a>,

    /// Timeout state for tracking operation timeout and attention handling
    timeout_state: Option<BulkCopyTimeoutState>,
}

impl<'a> BulkCopy<'a> {
    /// Create a new `BulkCopy` instance for the given table.
    ///
    /// # Arguments
    ///
    /// * `client` - A mutable reference to the TDS client connection
    /// * `table_name` - Name of the destination table (can include schema: "dbo.Users")
    ///
    /// # Example
    ///
    /// ```rust,ignore
    /// let mut bulk_copy = BulkCopy::new(&mut client, "MyTable");
    /// ```
    pub fn new(client: &'a mut TdsClient, table_name: impl Into<String>) -> Self {
        Self {
            client,
            table_name: table_name.into(),
            options: BulkCopyOptions::default(),
            column_mappings: Vec::new(),
            progress_callback: None,
            destination_metadata: None,
            metadata_retriever: Box::new(FmtOnlyMetadataRetriever::new()),
            timeout_state: None,
        }
    }

    /// Create a new `BulkCopy` instance with a custom metadata retriever.
    ///
    /// # Arguments
    ///
    /// * `client` - A mutable reference to the TDS client connection
    /// * `table_name` - Name of the destination table (can include schema: "dbo.Users")
    /// * `retriever` - Custom metadata retriever implementation
    ///
    /// # Example
    ///
    /// ```rust,ignore
    /// let custom_retriever = CachedMetadataRetriever::new();
    /// let mut bulk_copy = BulkCopy::with_retriever(
    ///     &mut client,
    ///     "MyTable",
    ///     Box::new(custom_retriever)
    /// );
    /// ```
    pub fn with_retriever(
        client: &'a mut TdsClient,
        table_name: impl Into<String>,
        retriever: Box<dyn MetadataRetriever + 'a>,
    ) -> Self {
        Self {
            client,
            table_name: table_name.into(),
            options: BulkCopyOptions::default(),
            column_mappings: Vec::new(),
            progress_callback: None,
            destination_metadata: None,
            metadata_retriever: retriever,
            timeout_state: None,
        }
    }

    /// Set the batch size (number of rows per batch).
    ///
    /// Default: 0 (all rows in one batch)
    ///
    /// # Arguments
    ///
    /// * `size` - Number of rows to include in each batch
    ///
    /// # Example
    ///
    /// ```rust,ignore
    /// bulk_copy.batch_size(5000); // Process 5000 rows at a time
    /// ```
    pub fn batch_size(mut self, size: usize) -> Self {
        self.options.batch_size = size;
        self
    }

    /// Convert a timeout duration to the whole-second representation used internally.
    ///
    /// A zero duration represents an unlimited timeout. Any positive duration less
    /// than one second is rounded up to one second so it cannot be interpreted as
    /// the unlimited-timeout sentinel. Durations exceeding the maximum `u32`
    /// timeout are saturated at `u32::MAX`.
    fn duration_to_timeout_seconds(timeout: Duration) -> u32 {
        if timeout > Duration::ZERO {
            u32::try_from(timeout.as_secs().max(1)).unwrap_or(u32::MAX)
        } else {
            0
        }
    }

    /// Set the operation timeout.
    ///
    /// Default: 30 seconds
    ///
    /// A timeout of `Duration::ZERO` disables the timeout. Positive durations
    /// shorter than one second are treated as a one-second timeout. Durations
    /// exceeding the maximum supported timeout are clamped to that maximum.
    ///
    /// # Arguments
    ///
    /// * `timeout` - Maximum duration for the operation
    ///
    /// # Example
    ///
    /// ```rust,ignore
    /// use std::time::Duration;
    /// bulk_copy.timeout(Duration::from_secs(120)); // 2 minute timeout
    /// ```
    pub fn timeout(mut self, timeout: Duration) -> Self {
        self.options.timeout_sec = Self::duration_to_timeout_seconds(timeout);
        self
    }

    /// Set whether to check constraints on the destination table.
    ///
    /// Default: false
    ///
    /// # Arguments
    ///
    /// * `enabled` - If true, check constraints will be enforced
    ///
    /// # Example
    ///
    /// ```rust,ignore
    /// bulk_copy.check_constraints(true);
    /// ```
    pub fn check_constraints(mut self, enabled: bool) -> Self {
        self.options.check_constraints = enabled;
        self
    }

    /// Set whether to fire triggers on the destination table.
    ///
    /// Default: false
    ///
    /// # Arguments
    ///
    /// * `enabled` - If true, triggers will be fired for inserted rows
    ///
    /// # Example
    ///
    /// ```rust,ignore
    /// bulk_copy.fire_triggers(true);
    /// ```
    pub fn fire_triggers(mut self, enabled: bool) -> Self {
        self.options.fire_triggers = enabled;
        self
    }

    /// Set whether to preserve source identity values.
    ///
    /// Default: false (identity values are auto-generated by SQL Server)
    ///
    /// # Arguments
    ///
    /// * `enabled` - If true, source identity values will be preserved
    ///
    /// # Example
    ///
    /// ```rust,ignore
    /// bulk_copy.keep_identity(true); // Preserve source identity values
    /// ```
    pub fn keep_identity(mut self, enabled: bool) -> Self {
        self.options.keep_identity = enabled;
        self
    }

    /// Set whether to preserve NULL values regardless of column defaults.
    ///
    /// Default: false
    ///
    /// # Arguments
    ///
    /// * `enabled` - If true, NULLs will be preserved even if columns have defaults
    ///
    /// # Example
    ///
    /// ```rust,ignore
    /// bulk_copy.keep_nulls(true);
    /// ```
    pub fn keep_nulls(mut self, enabled: bool) -> Self {
        self.options.keep_nulls = enabled;
        self
    }

    /// Write already-encrypted (ciphertext) values into Always Encrypted
    /// columns without the driver re-encrypting them.
    ///
    /// Default: false. This option only takes effect when Always Encrypted is
    /// enabled on the connection and the destination has encrypted columns;
    /// otherwise it is ignored. See
    /// [`BulkCopyOptions::allow_encrypted_value_modifications`] for the security
    /// caveats — only use this to move ciphertext between columns that share the
    /// same column encryption key, algorithm, and encryption type.
    ///
    /// # Example
    ///
    /// ```rust,ignore
    /// bulk_copy.allow_encrypted_value_modifications(true);
    /// ```
    pub fn allow_encrypted_value_modifications(mut self, enabled: bool) -> Self {
        self.options.allow_encrypted_value_modifications = enabled;
        self
    }

    /// Set whether to obtain a bulk update lock for the operation.
    ///
    /// Default: false
    ///
    /// # Arguments
    ///
    /// * `enabled` - If true, a table lock will be obtained
    ///
    /// # Example
    ///
    /// ```rust,ignore
    /// bulk_copy.table_lock(true); // Lock the table during bulk copy
    /// ```
    pub fn table_lock(mut self, enabled: bool) -> Self {
        self.options.table_lock = enabled;
        self
    }

    /// Set whether to use an internal transaction.
    ///
    /// Default: false (matches .NET SqlBulkCopy behavior)
    ///
    /// When `true`, each batch is wrapped in its own transaction:
    /// - BEGIN TRANSACTION before each batch
    /// - COMMIT TRANSACTION on successful batch completion
    /// - ROLLBACK TRANSACTION on batch failure
    ///
    /// When `false` (the default):
    /// - If the connection has an active API transaction, bulk copy participates in it
    /// - If no transaction exists, SQL Server's autocommit mode applies
    ///
    /// **Note**: Cannot be used when the connection already has an active transaction.
    ///
    /// # Arguments
    ///
    /// * `enabled` - If true, use internal per-batch transactions
    ///
    /// # Example
    ///
    /// ```rust,ignore
    /// bulk_copy.use_internal_transaction(true); // Use internal transaction per batch
    /// ```
    pub fn use_internal_transaction(mut self, enabled: bool) -> Self {
        self.options.use_internal_transaction = enabled;
        self
    }

    /// Set the notification interval for progress callbacks.
    ///
    /// Default: 0 (no progress notifications)
    ///
    /// # Arguments
    ///
    /// * `interval` - Number of rows to process before calling the progress callback
    ///
    /// # Example
    ///
    /// ```rust,ignore
    /// bulk_copy.notification_interval(1000); // Report progress every 1000 rows
    /// ```
    pub fn notification_interval(mut self, interval: usize) -> Self {
        self.options.notification_interval = interval;
        self
    }

    /// Add a column mapping from source to destination.
    ///
    /// If no mappings are specified, columns are mapped by ordinal position.
    ///
    /// # Arguments
    ///
    /// * `mapping` - Column mapping specification
    ///
    /// # Example
    ///
    /// ```rust,ignore
    /// use mssql_tds::connection::bulk_copy::ColumnMapping;
    ///
    /// bulk_copy
    ///     .add_column_mapping(ColumnMapping::by_name("SourceId", "Id"))
    ///     .add_column_mapping(ColumnMapping::by_ordinal(1, "Name"));
    /// ```
    pub fn add_column_mapping(mut self, mapping: ColumnMapping) -> Self {
        self.column_mappings.push(mapping);
        self
    }

    /// Resolve user-provided column mappings against destination metadata.
    ///
    /// This method converts ColumnMapping (which can specify source by name or ordinal)
    /// into ResolvedColumnMapping (which uses ordinals for both source and destination).
    ///
    /// # Arguments
    ///
    /// * `destination_metadata` - Metadata for the destination table columns
    ///
    /// # Returns
    ///
    /// A vector of resolved mappings, sorted by destination column index.
    ///
    /// # Errors
    ///
    /// Returns an error if a destination column name is not found in the metadata.
    fn resolve_column_mappings(
        &self,
        destination_metadata: &[BulkCopyColumnMetadata],
    ) -> TdsResult<Vec<ResolvedColumnMapping>> {
        let mut resolved = Vec::with_capacity(self.column_mappings.len());

        for mapping in &self.column_mappings {
            // Get source index (for ordinal-based mappings, use the ordinal directly)
            let source_index = match &mapping.source {
                ColumnMappingSource::Ordinal(idx) => *idx,
                ColumnMappingSource::Name(_) => {
                    // For name-based source mappings, we can't resolve without source metadata
                    // This is typically used with DataReader sources, not iterator-based sources
                    return Err(Error::UsageError(
                        "Source column name mappings are not supported with iterator-based bulk copy".to_string()
                    ));
                }
            };

            // Find destination column index by name
            let dest_column = destination_metadata
                .iter()
                .enumerate()
                .find(|(_, col)| col.column_name == mapping.destination)
                .ok_or_else(|| {
                    Error::UsageError(format!(
                        "Destination column '{}' not found in table",
                        mapping.destination
                    ))
                })?;

            resolved.push(ResolvedColumnMapping {
                source_index,
                destination_index: dest_column.0,
                destination_name: dest_column.1.column_name.clone(),
                destination_type: dest_column.1.sql_type,
            });
        }

        // Sort by destination index to ensure we write columns in the correct order
        resolved.sort_by_key(|m| m.destination_index);

        Ok(resolved)
    }

    /// Validate transaction state before starting bulk copy.
    ///
    /// This method checks for conflicting transaction configurations and ensures
    /// the connection is in a valid state for bulk copy operations.
    ///
    /// # Validation Rules
    ///
    /// 1. If `use_internal_transaction` is true and the connection already has
    ///    an active transaction, this is an error (mirrors .NET Transaction4.cs test:
    ///    "BulkLoadConflictingTransactionOption")
    ///
    /// 2. If the connection has an active transaction but `use_internal_transaction`
    ///    is false, the bulk copy will participate in that external transaction
    ///    (this is valid - mirrors .NET Transaction2.cs behavior)
    ///
    /// # Errors
    ///
    /// Returns `Error::UsageError` if the transaction state is invalid.
    fn validate_transaction_state(&self) -> TdsResult<()> {
        let has_active_transaction = self.client.has_active_transaction();
        let use_internal_transaction = self.options.use_internal_transaction;

        // Check for conflicting configuration:
        // Cannot use internal transaction when connection already has a transaction
        // This mirrors .NET SqlBulkCopy constructor validation (line 264-269):
        // "Must not specify SqlBulkCopyOption.UseInternalTransaction and pass an
        //  external Transaction at the same time."
        if use_internal_transaction && has_active_transaction {
            return Err(Error::UsageError(
                "Cannot use UseInternalTransaction when the connection already has an active transaction. \
                 Either commit/rollback the existing transaction, or set use_internal_transaction to false \
                 to participate in the existing transaction.".to_string()
            ));
        }

        Ok(())
    }

    /// Get resolved column mappings for the current configuration.
    ///
    /// This method retrieves destination metadata (if not already cached) and
    /// resolves the column mappings, returning them for use by the row adapter.
    ///
    /// # Returns
    ///
    /// A vector of resolved column mappings.
    ///
    /// # Errors
    ///
    /// Returns an error if metadata retrieval or mapping resolution fails.
    pub async fn get_resolved_mappings(&mut self) -> TdsResult<Vec<ResolvedColumnMapping>> {
        // Ensure we have destination metadata
        let destination_metadata = self.retrieve_destination_metadata().await?;

        // Resolve the mappings
        self.resolve_column_mappings(&destination_metadata)
    }

    /// Set a progress callback to receive notifications during the operation.
    ///
    /// The callback is invoked every `notification_interval` rows (if set).
    ///
    /// # Arguments
    ///
    /// * `callback` - Function to call with progress information
    ///
    /// # Example
    ///
    /// ```rust,ignore
    /// bulk_copy
    ///     .notification_interval(1000)
    ///     .on_progress(|progress| {
    ///         println!("Copied {} rows ({:.1}%)",
    ///             progress.rows_copied,
    ///             progress.percentage().unwrap_or(0.0));
    ///     });
    /// ```
    pub fn on_progress<F>(mut self, callback: F) -> Self
    where
        F: FnMut(BulkCopyProgress) + Send + 'a,
    {
        self.progress_callback = Some(Box::new(callback));
        self
    }

    /// Retrieve destination table metadata from SQL Server.
    ///
    /// This uses the configured metadata retriever strategy to get column information
    /// for the destination table. The metadata is cached for subsequent operations.
    ///
    /// # Returns
    ///
    /// A vector of `BulkCopyColumnMetadata` containing column information
    ///
    /// # Errors
    ///
    /// Returns an error if:
    /// - Table does not exist
    /// - No permission to access metadata
    /// - Network errors during query execution
    ///
    /// # Example
    ///
    /// ```rust,ignore
    /// let metadata = bulk_copy.retrieve_destination_metadata().await?;
    /// for col in &metadata {
    ///     println!("Column: {}, Type: {:?}", col.column_name, col.sql_type);
    /// }
    /// ```
    pub async fn retrieve_destination_metadata(
        &mut self,
    ) -> TdsResult<Vec<BulkCopyColumnMetadata>> {
        // Check if we already have cached metadata
        if let Some(ref metadata) = self.destination_metadata {
            return Ok(metadata.clone());
        }

        // Use the metadata retriever strategy
        let metadata = self
            .metadata_retriever
            .retrieve_metadata(self.client, &self.table_name, self.options.timeout_sec)
            .await?;

        // Cache the metadata
        self.destination_metadata = Some(metadata.clone());

        Ok(metadata)
    }

    /// Writes rows to the server using zero-copy bulk load operations.
    ///
    /// This method implements the zero-copy optimization path that eliminates per-row
    /// Vec allocations by serializing rows directly to the packet writer.
    ///
    /// # Performance
    ///
    /// This method provides superior performance compared to `write_to_server` by:
    /// - Eliminating the `dest_buffer.clone()` allocation per row (0 allocations vs 1)
    /// - Writing columns directly to the TDS packet without intermediate buffering
    /// - Reusing column contexts across all rows (created once during metadata phase)
    ///
    /// # Type Parameters
    ///
    /// * `I` - Iterator over rows
    /// * `R` - Row type implementing `BulkLoadRow` trait
    ///
    /// # Examples
    ///
    /// ```rust,ignore
    /// use mssql_tds::connection::bulk_copy::{BulkLoadRow, BulkRowWriter};
    ///
    /// struct User {
    ///     id: i32,
    ///     name: String,
    ///     active: bool,
    /// }
    ///
    /// #[async_trait]
    /// impl BulkLoadRow for User {
    ///     async fn write_to_packet(&self, writer: &mut BulkRowWriter<'_>) -> TdsResult<()> {
    ///         writer.write_int(0, self.id).await?;
    ///         writer.write_string(1, &self.name).await?;
    ///         writer.write_bit(2, self.active).await?;
    ///         Ok(())
    ///     }
    /// }
    ///
    /// let users = vec![
    ///     User { id: 1, name: "Alice".to_string(), active: true },
    ///     User { id: 2, name: "Bob".to_string(), active: false },
    /// ];
    ///
    /// bulk_copy.write_to_server_zerocopy(users.into_iter()).await?;
    /// ```
    ///
    /// # Errors
    ///
    /// Returns an error if:
    /// - Column mapping fails
    /// - Type conversion fails
    /// - SQL Server returns an error (constraints, type mismatches, etc.)
    /// - Network error occurs
    /// - Operation is cancelled or times out
    pub async fn write_to_server_zerocopy<I, R>(&mut self, rows: I) -> TdsResult<BulkCopyResult>
    where
        I: IntoIterator<Item = R>,
        R: BulkLoadRow,
    {
        let start_time = Instant::now();
        let mut total_rows = 0u64;

        // Initialize timeout state for this operation
        // A timeout of 0 means infinite (no timeout)
        self.timeout_state = Some(BulkCopyTimeoutState::from_seconds(self.options.timeout_sec));

        // ═══════════════════════════════════════════════════════════════════════
        // VALIDATION: Check for conflicting transaction states
        // This mirrors .NET SqlBulkCopy behavior from AnalyzeTargetAndCreateUpdateBulkCommand
        // ═══════════════════════════════════════════════════════════════════════
        self.validate_transaction_state()?;

        // Peek-able iterator for batch boundary detection
        let mut rows = rows.into_iter().peekable();

        // Ensure destination table exists and retrieve metadata
        if self.destination_metadata.is_none() {
            self.retrieve_destination_metadata().await?;
        }

        let _server_metadata = self
            .destination_metadata
            .as_ref()
            .ok_or_else(|| Error::UsageError("Destination metadata not available".to_string()))?;

        // If no column mappings are configured, peek at first row to determine source column count
        // and create ordinal mappings for min(source_columns, destination_columns)
        if self.column_mappings.is_empty() {
            let destination_metadata = self.retrieve_destination_metadata().await?;

            // Filter metadata based on keep_identity option for auto-mapping
            // When keep_identity is false, skip identity columns (they will be auto-generated)
            let filtered_metadata: Vec<_> = if self.options.keep_identity {
                destination_metadata.clone()
            } else {
                destination_metadata
                    .iter()
                    .filter(|col| !col.is_identity)
                    .cloned()
                    .collect()
            };

            // Peek at first row to determine source column count
            let source_column_count = if let Some(_first_row) = rows.peek() {
                // For BulkLoadRow trait, we can't easily determine column count without consuming
                // So we'll map all destination columns and let the row writer handle it
                // This is a limitation of the current design - the Python layer handles this better
                filtered_metadata.len()
            } else {
                // No rows, doesn't matter
                0
            };

            let mapping_count = std::cmp::min(source_column_count, filtered_metadata.len());
            self.column_mappings.reserve(mapping_count);
            for (i, col) in filtered_metadata.iter().enumerate().take(mapping_count) {
                self.column_mappings
                    .push(ColumnMapping::by_ordinal(i, col.column_name.clone()));
            }
        }

        // Prepare metadata only if we have rows to process
        if rows.peek().is_some() {
            // Retrieve destination metadata
            let destination_metadata = self.retrieve_destination_metadata().await?;

            // Filter metadata based on keep_identity option
            // When keep_identity is false, skip identity columns (they will be auto-generated)
            let filtered_metadata: Vec<_> = if self.options.keep_identity {
                destination_metadata.clone()
            } else {
                destination_metadata
                    .iter()
                    .filter(|col| !col.is_identity)
                    .cloned()
                    .collect()
            };

            // Resolve column mappings using user-provided mappings (using filtered metadata)
            let resolved_mappings = self.resolve_column_mappings(&filtered_metadata)?;

            // Use filtered metadata for the bulk copy operation
            let dest_column_metadata = filtered_metadata;

            // Process rows with zero-copy path
            self.write_rows_to_server_zerocopy(
                rows,
                &resolved_mappings,
                dest_column_metadata,
                &mut total_rows,
                start_time,
            )
            .await?;
        }

        let elapsed = start_time.elapsed();
        Ok(BulkCopyResult::new(total_rows, elapsed))
    }

    /// Internal method to write rows to the server using zero-copy path.
    ///
    /// This method implements the zero-copy optimization:
    /// - Rows are serialized directly to packet writer via BulkLoadRow trait
    /// - No intermediate Vec allocations (0 allocations per row)
    /// - Column contexts are created once and reused across all rows
    async fn write_rows_to_server_zerocopy<I, R>(
        &mut self,
        mut rows: std::iter::Peekable<I>,
        resolved_mappings: &[ResolvedColumnMapping],
        dest_column_metadata: Vec<BulkCopyColumnMetadata>,
        total_rows: &mut u64,
        start_time: Instant,
    ) -> TdsResult<()>
    where
        I: Iterator<Item = R>,
        R: BulkLoadRow,
    {
        // Determine batch size (0 means all rows in one batch)
        let batch_size = if self.options.batch_size == 0 {
            usize::MAX
        } else {
            self.options.batch_size
        };

        let timeout_sec = if self.options.timeout_sec > 0 {
            Some(self.options.timeout_sec)
        } else {
            None
        };

        // Capture use_internal_transaction flag at start (matches .NET _savedBatchSize pattern)
        let use_internal_transaction = self.options.use_internal_transaction;

        // Accumulate INFO messages across all batches so the whole bulk copy operation's
        // diagnostics are retrievable via `client.info_messages()` after it returns.
        // Each `execute_bulk_load_streaming_zerocopy` call resets the client's buffer via
        // `begin_command`, and the internal `commit_transaction` (when enabled) would clear
        // it as well, so we drain each batch's INFO before commit and restore the full set
        // once the loop completes.
        let mut accumulated_info: Vec<SqlInfoMessage> = Vec::new();

        loop {
            if rows.peek().is_none() {
                break;
            }

            // Check timeout before starting next batch
            if let Some(ref mut state) = self.timeout_state
                && state.is_expired()
                && !state.is_attention_sent()
            {
                // Mark that we're sending attention due to timeout
                state.mark_bulk_copy_write_timeout();
                state.begin_sending_attention();

                // Send attention and wait for ACK with ATTENTION_TIMEOUT_SECONDS timeout
                // per SqlClient behavior
                let attention_timeout = Duration::from_secs(ATTENTION_TIMEOUT_SECONDS);
                let ack_received = self
                    .client
                    .send_attention_with_timeout(attention_timeout)
                    .await
                    .unwrap_or(false);

                state.mark_attention_sent();

                if ack_received {
                    state.mark_attention_received();
                    // Attention acknowledged - return timeout error
                    let err = BulkCopyTimeoutError::new(
                        *total_rows,
                        self.options.timeout_sec,
                        Some("Bulk copy operation timed out".to_string()),
                    );
                    return Err(Error::BulkCopyError(BulkCopyError::Timeout(err)));
                } else {
                    // No attention ACK within 5 seconds - connection is broken
                    // Close the connection to ensure it can't be reused
                    let _ = self.client.close_connection().await;

                    let err = BulkCopyAttentionTimeoutError::new(true);
                    return Err(Error::BulkCopyError(BulkCopyError::AttentionTimeout(err)));
                }
            }

            // Create a batch iterator that yields up to batch_size rows
            // The take() adaptor just sets an upper bound, but the actual iteration ends when the input is finished.
            let batch_iter = (&mut rows).take(batch_size);

            // ═══════════════════════════════════════════════════════════════════
            // BEGIN TRANSACTION: Start internal transaction before each batch
            // (if UseInternalTransaction is enabled)
            // This mirrors .NET SqlBulkCopy.CopyBatchesAsync behavior
            // ═══════════════════════════════════════════════════════════════════
            if use_internal_transaction {
                self.client
                    .begin_transaction(TransactionIsolationLevel::ReadCommitted, None)
                    .await?;
            }

            // Execute streaming bulk load with zero-copy path
            let batch_result = self
                .client
                .execute_bulk_load_streaming_zerocopy(
                    self.table_name.clone(),
                    dest_column_metadata.clone(),
                    self.options.clone(),
                    timeout_sec,
                    None,
                    batch_iter,
                    resolved_mappings,
                )
                .await;

            // Handle batch result with transaction management
            match batch_result {
                Ok(batch_count) => {
                    // Drain this batch's INFO messages before the internal commit (which
                    // resets the client's buffer) can clear them.
                    accumulated_info.extend(self.client.take_info_messages());

                    // ═══════════════════════════════════════════════════════════
                    // COMMIT TRANSACTION: Commit on successful batch completion
                    // This mirrors .NET SqlBulkCopy.CommitTransaction() behavior
                    // ═══════════════════════════════════════════════════════════
                    if use_internal_transaction
                        && let Err(e) = self.client.commit_transaction(None, None).await
                    {
                        // The commit itself failed. Restore the INFO accumulated up to
                        // and including this batch before propagating the error, so it
                        // stays retrievable via `client.info_messages()` — consistent
                        // with the Err(e) arm below. Drop any residual INFO left by the
                        // failed commit first.
                        let _ = self.client.take_info_messages();
                        self.client
                            .extend_info_messages(std::mem::take(&mut accumulated_info));
                        return Err(e);
                    }

                    *total_rows += batch_count;
                }
                Err(e) => {
                    // Capture this failing batch's INFO before rollback (which resets
                    // the client's buffer via begin_command) can clear it, so it is
                    // retained alongside the already-accumulated prior batches' INFO.
                    accumulated_info.extend(self.client.take_info_messages());

                    // ═══════════════════════════════════════════════════════════
                    // ROLLBACK TRANSACTION: Rollback on batch failure
                    // This mirrors .NET SqlBulkCopy.AbortTransaction() behavior
                    // Note: Only rollback if we have an active internal transaction
                    // ═══════════════════════════════════════════════════════════
                    if use_internal_transaction && self.client.has_active_transaction() {
                        // Attempt rollback, but don't mask the original error
                        let _ = self.client.rollback_transaction(None, None).await;
                    }

                    // Restore the INFO accumulated up to the failure so a mid-stream
                    // failure still leaves the completed batches' diagnostics (plus this
                    // batch's INFO emitted before the error) retrievable via
                    // `client.info_messages()`. The returned error carries the failure
                    // itself; INFO is a separate, complementary channel (INFO tokens vs
                    // ERROR tokens) and this mirrors .NET's InfoMessage events having
                    // already fired for the batches that completed. Drop any residual
                    // INFO left by the rollback first.
                    let _ = self.client.take_info_messages();
                    self.client
                        .extend_info_messages(std::mem::take(&mut accumulated_info));

                    return Err(e);
                }
            }

            // Report progress if callback is configured
            if let Some(ref mut callback) = self.progress_callback
                && self.options.notification_interval > 0
                && (*total_rows).is_multiple_of(self.options.notification_interval as u64)
            {
                let elapsed = start_time.elapsed();
                let rows_per_second = if elapsed.as_secs_f64() > 0.0 {
                    *total_rows as f64 / elapsed.as_secs_f64()
                } else {
                    0.0
                };

                callback(BulkCopyProgress {
                    rows_copied: *total_rows,
                    total_rows: None,
                    elapsed,
                    rows_per_second,
                });
            }
        }

        // Restore the INFO messages accumulated across all batches so the caller can read
        // the full bulk copy diagnostics via `client.info_messages()`. Drop any residual
        // buffer left by the final internal commit first, so only bulk-load INFO remains.
        let _ = self.client.take_info_messages();
        self.client.extend_info_messages(accumulated_info);

        Ok(())
    }

    /// Check if the bulk copy operation has timed out.
    ///
    /// This can be called during long-running operations to check if the
    /// configured timeout has expired.
    ///
    /// # Returns
    ///
    /// * `true` - if timeout has expired
    /// * `false` - if timeout has not expired or no timeout was configured
    pub fn is_timed_out(&self) -> bool {
        self.timeout_state
            .as_ref()
            .is_some_and(|state| state.is_expired())
    }

    /// Get the remaining timeout in milliseconds.
    ///
    /// # Returns
    ///
    /// * `Some(ms)` - remaining milliseconds until timeout
    /// * `None` - if no timeout was configured (infinite timeout)
    pub fn remaining_timeout_ms(&self) -> Option<u64> {
        self.timeout_state
            .as_ref()
            .and_then(|state| state.remaining_ms())
    }
}

/// Zero-copy bulk copy trait for direct serialization to TDS packets.
///
/// This trait enables true zero-copy bulk insert by allowing rows to write
/// directly to the packet writer without allocating intermediate Vec<ColumnValues>.
/// This eliminates the per-row allocation overhead present in the `BulkCopyRow` trait.
///
/// # Performance Benefits
///
/// - **Zero allocations per row** - no Vec allocation needed
/// - **Direct serialization** - writes straight to TDS packet buffer
/// - **Lower memory pressure** - no intermediate storage of column values
///
/// # Example
///
/// ```rust,ignore
/// use mssql_tds::connection::bulk_copy::{BulkLoadRow, BulkRowWriter};
/// use mssql_tds::core::TdsResult;
///
/// struct Product {
///     id: i32,
///     name: String,
///     price: f64,
/// }
///
/// impl BulkLoadRow for Product {
///     async fn write_to_packet(&self, writer: &mut BulkRowWriter<'_>) -> TdsResult<()> {
///         writer.write_int(self.id).await?;
///         writer.write_string(&self.name).await?;
///         writer.write_float(self.price).await?;
///         Ok(())
///     }
/// }
///
/// // Use it:
/// let products = vec![Product { id: 1, name: "Widget".to_string(), price: 9.99 }];
/// let bulk_copy = BulkCopy::new(&mut client, "Products");
/// bulk_copy.write_to_server_zerocopy(products.into_iter()).await?;
/// ```
#[async_trait]
pub trait BulkLoadRow {
    /// Write this row's column values directly to the packet writer.
    ///
    /// The order of writes must match the order of columns in the destination table.
    /// Each column value is serialized directly to the TDS packet without intermediate
    /// allocations.
    ///
    /// # Arguments
    ///
    /// * `writer` - Streaming bulk load writer
    /// * `column_index` - Mutable reference to track current column index
    ///
    /// # Errors
    ///
    /// Returns an error if:
    /// - Network errors occur during transmission
    /// - Type conversion errors occur
    /// - Column count doesn't match metadata
    async fn write_to_packet(
        &self,
        writer: &mut crate::message::bulk_load::StreamingBulkLoadWriter<'_>,
        column_index: &mut usize,
    ) -> TdsResult<()>;
}

#[cfg(test)]
mod tests {
    use crate::datatypes::bulk_copy_metadata::SystemTypeId;

    use super::*;

    #[test]
    fn test_bulk_copy_options_default() {
        let opts = BulkCopyOptions::default();
        assert_eq!(opts.batch_size, 0);
        assert_eq!(opts.timeout_sec, 30);
        assert!(!opts.check_constraints);
        assert!(!opts.fire_triggers);
        assert!(!opts.keep_identity);
        assert!(!opts.keep_nulls);
        assert!(!opts.table_lock);
        assert!(!opts.use_internal_transaction); // Matches .NET default
        assert_eq!(opts.notification_interval, 0);
    }

    #[test]
    fn test_bulk_copy_options_validate() {
        let opts = BulkCopyOptions {
            batch_size: 5000,
            ..Default::default()
        };
        assert!(opts.validate().is_ok());

        let invalid_opts = BulkCopyOptions {
            batch_size: 2_000_000,
            ..Default::default()
        };
        assert!(invalid_opts.validate().is_err());
    }

    #[test]
    fn test_column_mapping_by_name() {
        let mapping = ColumnMapping::by_name("source_col", "dest_col");
        assert!(matches!(
            mapping.source,
            ColumnMappingSource::Name(ref name) if name == "source_col"
        ));
        assert_eq!(mapping.destination, "dest_col");
    }

    #[test]
    fn test_column_mapping_by_ordinal() {
        let mapping = ColumnMapping::by_ordinal(2, "dest_col");
        assert!(matches!(mapping.source, ColumnMappingSource::Ordinal(2)));
        assert_eq!(mapping.destination, "dest_col");
    }

    #[test]
    fn test_bulk_copy_progress_percentage() {
        let progress = BulkCopyProgress {
            rows_copied: 500,
            total_rows: Some(1000),
            elapsed: Duration::from_secs(10),
            rows_per_second: 50.0,
        };

        assert_eq!(progress.percentage(), Some(50.0));

        let progress_no_total = BulkCopyProgress {
            rows_copied: 500,
            total_rows: None,
            elapsed: Duration::from_secs(10),
            rows_per_second: 50.0,
        };

        assert_eq!(progress_no_total.percentage(), None);
    }

    #[test]
    fn test_bulk_copy_progress_estimated_time() {
        let progress = BulkCopyProgress {
            rows_copied: 1000,
            total_rows: Some(2000),
            elapsed: Duration::from_secs(10),
            rows_per_second: 100.0,
        };

        let estimated = progress.estimated_time_remaining().unwrap();
        assert_eq!(estimated.as_secs(), 10); // 1000 remaining rows / 100 rows/sec = 10 sec
    }

    #[test]
    fn test_bulk_copy_result() {
        let result = BulkCopyResult::new(10000, Duration::from_secs(10));
        assert_eq!(result.rows_affected, 10000);
        assert_eq!(result.rows_per_second, 1000.0);
    }

    #[test]
    fn test_bulk_copy_metadata_creation() {
        use crate::datatypes::bulk_copy_metadata::TypeLength;

        let bulk_meta = BulkCopyColumnMetadata::new("TestColumn", SqlDbType::Int, 0x26)
            .with_length(4, TypeLength::Fixed(4))
            .with_nullable(true);

        assert_eq!(bulk_meta.column_name, "TestColumn");
        assert_eq!(bulk_meta.sql_type, SqlDbType::Int);
        assert_eq!(bulk_meta.tds_type, 0x26); // TDS type for IntN (nullable int)
        assert!(bulk_meta.is_nullable);
    }

    #[test]
    fn test_system_type_id_to_sql_db_type_conversion() {
        // Test common types using TryFrom trait with SystemTypeId wrapper
        assert_eq!(
            SqlDbType::try_from(SystemTypeId(48)).unwrap(),
            SqlDbType::TinyInt
        );
        assert_eq!(
            SqlDbType::try_from(SystemTypeId(56)).unwrap(),
            SqlDbType::Int
        );
        assert_eq!(
            SqlDbType::try_from(SystemTypeId(127)).unwrap(),
            SqlDbType::BigInt
        );
        assert_eq!(
            SqlDbType::try_from(SystemTypeId(231)).unwrap(),
            SqlDbType::NVarChar
        );
        assert_eq!(
            SqlDbType::try_from(SystemTypeId(167)).unwrap(),
            SqlDbType::VarChar
        );
        assert_eq!(
            SqlDbType::try_from(SystemTypeId(36)).unwrap(),
            SqlDbType::UniqueIdentifier
        );

        // Test unsupported type
        assert!(SqlDbType::try_from(SystemTypeId(255)).is_err());
    }

    // Tests for type compatibility checking
    // Note: These test the internal logic without needing a full BulkCopy instance

    fn check_compat(source: SqlDbType, dest: SqlDbType) -> bool {
        // Exact match
        if source == dest {
            return true;
        }

        // Test the compatibility rules
        matches!(
            (source, dest),
            // Numeric promotions
            (SqlDbType::TinyInt, SqlDbType::SmallInt | SqlDbType::Int | SqlDbType::BigInt) |
            (SqlDbType::SmallInt, SqlDbType::Int | SqlDbType::BigInt) |
            (SqlDbType::Int, SqlDbType::BigInt) |
            // Numeric to float
            (SqlDbType::TinyInt | SqlDbType::SmallInt | SqlDbType::Int | SqlDbType::BigInt,
             SqlDbType::Real | SqlDbType::Float) |
            (SqlDbType::Real, SqlDbType::Float) |
            // Decimal/Numeric
            (SqlDbType::Decimal, SqlDbType::Numeric) | (SqlDbType::Numeric, SqlDbType::Decimal) |
            // String conversions
            (SqlDbType::Char, SqlDbType::VarChar) |
            (SqlDbType::NChar, SqlDbType::NVarChar) |
            (SqlDbType::VarChar, SqlDbType::NVarChar) |
            (SqlDbType::Text, SqlDbType::VarChar | SqlDbType::NVarChar) |
            (SqlDbType::NText, SqlDbType::NVarChar) |
            (SqlDbType::VarChar, SqlDbType::Text) |
            (SqlDbType::NVarChar, SqlDbType::NText) |
            // Binary conversions
            (SqlDbType::Binary, SqlDbType::VarBinary) |
            (SqlDbType::VarBinary, SqlDbType::Image) |
            (SqlDbType::Image, SqlDbType::VarBinary) |
            // DateTime conversions
            (SqlDbType::SmallDateTime, SqlDbType::DateTime | SqlDbType::DateTime2) |
            (SqlDbType::DateTime, SqlDbType::DateTime2) |
            (SqlDbType::Date, SqlDbType::DateTime | SqlDbType::DateTime2) |
            // Money conversions
            (SqlDbType::SmallMoney, SqlDbType::Money)
        )
    }

    #[test]
    fn test_type_compatibility_exact_match() {
        // Exact type matches should always be compatible
        assert!(check_compat(SqlDbType::Int, SqlDbType::Int));
        assert!(check_compat(SqlDbType::NVarChar, SqlDbType::NVarChar));
        assert!(check_compat(SqlDbType::DateTime2, SqlDbType::DateTime2));
    }

    #[test]
    fn test_type_compatibility_numeric_promotions() {
        // TinyInt can promote to larger integer types
        assert!(check_compat(SqlDbType::TinyInt, SqlDbType::SmallInt));
        assert!(check_compat(SqlDbType::TinyInt, SqlDbType::Int));
        assert!(check_compat(SqlDbType::TinyInt, SqlDbType::BigInt));

        // SmallInt can promote to larger integer types
        assert!(check_compat(SqlDbType::SmallInt, SqlDbType::Int));
        assert!(check_compat(SqlDbType::SmallInt, SqlDbType::BigInt));

        // Int can promote to BigInt
        assert!(check_compat(SqlDbType::Int, SqlDbType::BigInt));

        // But not the reverse
        assert!(!check_compat(SqlDbType::BigInt, SqlDbType::Int));
        assert!(!check_compat(SqlDbType::Int, SqlDbType::SmallInt));
    }

    #[test]
    fn test_type_compatibility_numeric_to_float() {
        // Integer types can convert to float types
        assert!(check_compat(SqlDbType::TinyInt, SqlDbType::Real));
        assert!(check_compat(SqlDbType::SmallInt, SqlDbType::Float));
        assert!(check_compat(SqlDbType::Int, SqlDbType::Float));
        assert!(check_compat(SqlDbType::BigInt, SqlDbType::Real));

        // Real can promote to Float
        assert!(check_compat(SqlDbType::Real, SqlDbType::Float));

        // But not the reverse
        assert!(!check_compat(SqlDbType::Float, SqlDbType::Real));
    }

    #[test]
    fn test_type_compatibility_string_types() {
        // Char types can convert to Varchar
        assert!(check_compat(SqlDbType::Char, SqlDbType::VarChar));
        assert!(check_compat(SqlDbType::NChar, SqlDbType::NVarChar));

        // ASCII to Unicode conversion
        assert!(check_compat(SqlDbType::VarChar, SqlDbType::NVarChar));

        // Text type conversions
        assert!(check_compat(SqlDbType::Text, SqlDbType::VarChar));
        assert!(check_compat(SqlDbType::Text, SqlDbType::NVarChar));
        assert!(check_compat(SqlDbType::NText, SqlDbType::NVarChar));
        assert!(check_compat(SqlDbType::VarChar, SqlDbType::Text));
        assert!(check_compat(SqlDbType::NVarChar, SqlDbType::NText));
    }

    #[test]
    fn test_type_compatibility_binary_types() {
        assert!(check_compat(SqlDbType::Binary, SqlDbType::VarBinary));
        assert!(check_compat(SqlDbType::VarBinary, SqlDbType::Image));
        assert!(check_compat(SqlDbType::Image, SqlDbType::VarBinary));
    }

    #[test]
    fn test_type_compatibility_datetime_types() {
        // SmallDateTime can convert to DateTime and DateTime2
        assert!(check_compat(SqlDbType::SmallDateTime, SqlDbType::DateTime));
        assert!(check_compat(SqlDbType::SmallDateTime, SqlDbType::DateTime2));

        // DateTime can convert to DateTime2
        assert!(check_compat(SqlDbType::DateTime, SqlDbType::DateTime2));

        // Date can convert to DateTime types
        assert!(check_compat(SqlDbType::Date, SqlDbType::DateTime));
        assert!(check_compat(SqlDbType::Date, SqlDbType::DateTime2));
    }

    #[test]
    fn test_type_compatibility_incompatible_types() {
        // String to numeric should not be compatible
        assert!(!check_compat(SqlDbType::VarChar, SqlDbType::Int));
        assert!(!check_compat(SqlDbType::NVarChar, SqlDbType::BigInt));

        // Numeric to string should not be compatible
        assert!(!check_compat(SqlDbType::Int, SqlDbType::VarChar));

        // DateTime to numeric should not be compatible
        assert!(!check_compat(SqlDbType::DateTime, SqlDbType::Int));

        // Binary to string should not be compatible
        assert!(!check_compat(SqlDbType::VarBinary, SqlDbType::VarChar));
    }

    // =========================================================================
    // Transaction-related unit tests
    // =========================================================================

    #[test]
    fn test_use_internal_transaction_default_is_false() {
        // Verify the default matches .NET SqlBulkCopy behavior:
        // SqlBulkCopyOptions.Default = 0 means no flags are set,
        // so UseInternalTransaction is off by default.
        let opts = BulkCopyOptions::default();
        assert!(
            !opts.use_internal_transaction,
            "use_internal_transaction should default to false (matching .NET)"
        );
    }

    #[test]
    fn test_batch_size_zero_means_single_batch() {
        let opts = BulkCopyOptions::default();
        assert_eq!(opts.batch_size, 0, "batch_size should default to 0");

        // When batch_size is 0, effective batch size is usize::MAX (single batch)
        let effective_batch_size = if opts.batch_size == 0 {
            usize::MAX
        } else {
            opts.batch_size
        };
        assert_eq!(
            effective_batch_size,
            usize::MAX,
            "batch_size=0 should mean single batch (usize::MAX)"
        );
    }

    #[test]
    fn test_batch_size_positive_value() {
        let opts = BulkCopyOptions {
            batch_size: 1000,
            ..Default::default()
        };
        assert_eq!(opts.batch_size, 1000);

        // Effective batch size should be the configured value
        let effective_batch_size = if opts.batch_size == 0 {
            usize::MAX
        } else {
            opts.batch_size
        };
        assert_eq!(effective_batch_size, 1000);
    }

    #[test]
    fn test_use_internal_transaction_can_be_enabled() {
        // UseInternalTransaction can be explicitly enabled
        let opts = BulkCopyOptions {
            use_internal_transaction: true,
            ..Default::default()
        };
        assert!(
            opts.use_internal_transaction,
            "use_internal_transaction should be configurable to true"
        );
    }

    #[test]
    fn test_duration_to_timeout_seconds() {
        assert_eq!(BulkCopy::duration_to_timeout_seconds(Duration::ZERO), 0);
        assert_eq!(
            BulkCopy::duration_to_timeout_seconds(Duration::from_millis(1)),
            1
        );
        assert_eq!(
            BulkCopy::duration_to_timeout_seconds(Duration::from_millis(500)),
            1
        );
        assert_eq!(
            BulkCopy::duration_to_timeout_seconds(Duration::from_millis(999)),
            1
        );
        assert_eq!(
            BulkCopy::duration_to_timeout_seconds(Duration::from_secs(1)),
            1
        );
        assert_eq!(
            BulkCopy::duration_to_timeout_seconds(Duration::from_secs(2)),
            2
        );
        assert_eq!(
            BulkCopy::duration_to_timeout_seconds(Duration::from_millis(2500)),
            2
        );
        assert_eq!(
            BulkCopy::duration_to_timeout_seconds(Duration::from_secs(u32::MAX as u64)),
            u32::MAX
        );
        assert_eq!(
            BulkCopy::duration_to_timeout_seconds(Duration::from_secs(1_u64 << 32)),
            u32::MAX
        );
    }

    // Note: Full integration tests for BulkCopy transaction behavior require
    // a TdsClient connection and will be added in the integration test phase.
    // See the Implementation Guide for comprehensive integration test specifications.
}