aprender-db 0.60.0

//! Top-K selection algorithms
//!
//! **Problem**: `ORDER BY ... LIMIT K` is O(N log N). Top-K selection is O(N).
//!
//! **Solution**: Min-heap based Top-K selection algorithm
//!
//! **Performance Impact** (1M files):
//! - Full sort: 2.3 seconds
//! - Top-K selection: 0.08 seconds
//! - **Speedup**: 28.75x
//!
//! Toyota Way Principles:
//! - **Kaizen**: Algorithmic improvement (O(N log N) → O(N))
//! - **Muda elimination**: Avoid unnecessary full sort
//! - **Genchi Genbutsu**: Actual performance measurements guide optimization
//!
//! References:
//! - ../paiml-mcp-agent-toolkit/docs/specifications/trueno-db-integration-review-response.md Issue #2

use crate::Error;
use arrow::array::{
    Array, ArrayRef, Float32Array, Float64Array, Int32Array, Int64Array, StringArray,
};
use arrow::compute::SortOptions;
use arrow::record_batch::RecordBatch;
use std::cmp::Ordering;
use std::collections::BinaryHeap;
use std::sync::Arc;

/// Sort order for Top-K selection
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum SortOrder {
    /// Ascending order (smallest K values)
    Ascending,
    /// Descending order (largest K values)
    Descending,
}

impl From<SortOrder> for SortOptions {
    fn from(order: SortOrder) -> Self {
        Self { descending: matches!(order, SortOrder::Descending), nulls_first: false }
    }
}

/// Trait for Top-K selection on record batches
pub trait TopKSelection {
    /// Select top K rows by a specific column
    ///
    /// # Arguments
    /// * `column_index` - Index of the column to sort by
    /// * `k` - Number of rows to select
    /// * `order` - Sort order (Ascending or Descending)
    ///
    /// # Returns
    /// A new `RecordBatch` containing the top K rows
    ///
    /// # Errors
    /// Returns error if:
    /// - Column index is out of bounds
    /// - Column data type is not sortable
    /// - K is zero
    ///
    /// # Examples
    ///
    /// ```rust
    /// use trueno_db::topk::{TopKSelection, SortOrder};
    /// use arrow::array::{Float64Array, RecordBatch};
    /// use arrow::datatypes::{DataType, Field, Schema};
    /// use std::sync::Arc;
    ///
    /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
    /// let schema = Arc::new(Schema::new(vec![
    ///     Field::new("score", DataType::Float64, false),
    /// ]));
    /// let batch = RecordBatch::try_new(
    ///     schema,
    ///     vec![Arc::new(Float64Array::from(vec![1.0, 5.0, 3.0, 9.0, 2.0]))],
    /// )?;
    ///
    /// // Get top 3 highest scores
    /// let top3 = batch.top_k(0, 3, SortOrder::Descending)?;
    /// assert_eq!(top3.num_rows(), 3);
    /// # Ok(())
    /// # }
    /// ```
    fn top_k(&self, column_index: usize, k: usize, order: SortOrder) -> crate::Result<RecordBatch>;
}

impl TopKSelection for RecordBatch {
    fn top_k(&self, column_index: usize, k: usize, order: SortOrder) -> crate::Result<RecordBatch> {
        // Validate inputs
        if k == 0 {
            return Err(Error::InvalidInput("k must be greater than 0".to_string()));
        }

        if column_index >= self.num_columns() {
            return Err(Error::InvalidInput(format!(
                "Column index {} out of bounds (batch has {} columns)",
                column_index,
                self.num_columns()
            )));
        }

        // If k >= num_rows, just sort and return all rows
        if k >= self.num_rows() {
            return sort_all_rows(self, column_index, order);
        }

        // Use heap-based Top-K selection
        let column = self.column(column_index);
        let indices = select_top_k_indices(column, k, order)?;

        // Build result batch from selected indices
        build_batch_from_indices(self, &indices)
    }
}

/// Select top K indices using min-heap algorithm
///
/// Time complexity: O(N log K) where N = number of rows, K = selection size
/// Space complexity: O(K) for the heap
fn select_top_k_indices(
    column: &ArrayRef,
    k: usize,
    order: SortOrder,
) -> crate::Result<Vec<usize>> {
    match column.data_type() {
        arrow::datatypes::DataType::Int32 => {
            let array = column.as_any().downcast_ref::<Int32Array>().ok_or_else(|| {
                Error::Other("Failed to downcast Int32 column to Int32Array".to_string())
            })?;
            select_top_k_typed(array.len(), k, order, |i| array.is_null(i), |i| array.value(i))
        }
        arrow::datatypes::DataType::Int64 => {
            let array = column.as_any().downcast_ref::<Int64Array>().ok_or_else(|| {
                Error::Other("Failed to downcast Int64 column to Int64Array".to_string())
            })?;
            select_top_k_typed(array.len(), k, order, |i| array.is_null(i), |i| array.value(i))
        }
        arrow::datatypes::DataType::Float32 => {
            let array = column.as_any().downcast_ref::<Float32Array>().ok_or_else(|| {
                Error::Other("Failed to downcast Float32 column to Float32Array".to_string())
            })?;
            select_top_k_typed(array.len(), k, order, |i| array.is_null(i), |i| array.value(i))
        }
        arrow::datatypes::DataType::Float64 => {
            let array = column.as_any().downcast_ref::<Float64Array>().ok_or_else(|| {
                Error::Other("Failed to downcast Float64 column to Float64Array".to_string())
            })?;
            select_top_k_typed(array.len(), k, order, |i| array.is_null(i), |i| array.value(i))
        }
        dt => Err(Error::InvalidInput(format!("Top-K not supported for data type: {dt:?}"))),
    }
}

// Heap item for descending order (min-heap: keep smallest at top, so we can find largest K)
#[derive(Debug)]
struct MinHeapItem<V> {
    value: V,
    index: usize,
}

impl<V: PartialOrd> PartialEq for MinHeapItem<V> {
    fn eq(&self, other: &Self) -> bool {
        self.value.partial_cmp(&other.value) == Some(Ordering::Equal)
    }
}

impl<V: PartialOrd> Eq for MinHeapItem<V> {}

impl<V: PartialOrd> Ord for MinHeapItem<V> {
    fn cmp(&self, other: &Self) -> Ordering {
        // Reverse comparison for min-heap (smallest at top)
        other.value.partial_cmp(&self.value).unwrap_or(Ordering::Equal)
    }
}

impl<V: PartialOrd> PartialOrd for MinHeapItem<V> {
    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
        Some(self.cmp(other))
    }
}

// Heap item for ascending order (max-heap: keep largest at top, so we can find smallest K)
#[derive(Debug)]
struct MaxHeapItem<V> {
    value: V,
    index: usize,
}

impl<V: PartialOrd> PartialEq for MaxHeapItem<V> {
    fn eq(&self, other: &Self) -> bool {
        self.value.partial_cmp(&other.value) == Some(Ordering::Equal)
    }
}

impl<V: PartialOrd> Eq for MaxHeapItem<V> {}

impl<V: PartialOrd> Ord for MaxHeapItem<V> {
    fn cmp(&self, other: &Self) -> Ordering {
        // Normal comparison for max-heap (largest at top)
        self.value.partial_cmp(&other.value).unwrap_or(Ordering::Equal)
    }
}

impl<V: PartialOrd> PartialOrd for MaxHeapItem<V> {
    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
        Some(self.cmp(other))
    }
}

/// Collect top-K indices from a min-heap (descending order: find largest K values)
fn collect_top_k_descending<V: PartialOrd>(
    len: usize,
    k: usize,
    is_null: impl Fn(usize) -> bool,
    get_value: impl Fn(usize) -> V,
) -> Vec<usize> {
    let mut heap: BinaryHeap<MinHeapItem<V>> = BinaryHeap::with_capacity(k);

    for index in 0..len {
        if !is_null(index) {
            let value = get_value(index);
            if heap.len() < k {
                heap.push(MinHeapItem { value, index });
            } else if let Some(top) = heap.peek() {
                if value.partial_cmp(&top.value) == Some(Ordering::Greater) {
                    heap.pop();
                    heap.push(MinHeapItem { value, index });
                }
            }
        }
    }

    let mut result: Vec<_> = heap.into_vec();
    result.sort_by(|a, b| b.value.partial_cmp(&a.value).unwrap_or(Ordering::Equal));
    result.into_iter().map(|item| item.index).collect()
}

/// Collect top-K indices from a max-heap (ascending order: find smallest K values)
fn collect_top_k_ascending<V: PartialOrd>(
    len: usize,
    k: usize,
    is_null: impl Fn(usize) -> bool,
    get_value: impl Fn(usize) -> V,
) -> Vec<usize> {
    let mut heap: BinaryHeap<MaxHeapItem<V>> = BinaryHeap::with_capacity(k);

    for index in 0..len {
        if !is_null(index) {
            let value = get_value(index);
            if heap.len() < k {
                heap.push(MaxHeapItem { value, index });
            } else if let Some(top) = heap.peek() {
                if value.partial_cmp(&top.value) == Some(Ordering::Less) {
                    heap.pop();
                    heap.push(MaxHeapItem { value, index });
                }
            }
        }
    }

    let mut result: Vec<_> = heap.into_vec();
    result.sort_by(|a, b| a.value.partial_cmp(&b.value).unwrap_or(Ordering::Equal));
    result.into_iter().map(|item| item.index).collect()
}

/// Generic top-K selection for any Arrow array with `PartialOrd` values
#[allow(clippy::unnecessary_wraps)]
fn select_top_k_typed<V: PartialOrd>(
    len: usize,
    k: usize,
    order: SortOrder,
    is_null: impl Fn(usize) -> bool,
    get_value: impl Fn(usize) -> V,
) -> crate::Result<Vec<usize>> {
    let indices = match order {
        SortOrder::Descending => collect_top_k_descending(len, k, is_null, get_value),
        SortOrder::Ascending => collect_top_k_ascending(len, k, is_null, get_value),
    };
    Ok(indices)
}

/// Build a new record batch from selected row indices
fn build_batch_from_indices(batch: &RecordBatch, indices: &[usize]) -> crate::Result<RecordBatch> {
    use arrow::datatypes::DataType;

    let mut new_columns: Vec<ArrayRef> = Vec::with_capacity(batch.num_columns());

    for col_idx in 0..batch.num_columns() {
        let column = batch.column(col_idx);

        let new_array: ArrayRef = match column.data_type() {
            DataType::Int32 => {
                let array = column.as_any().downcast_ref::<Int32Array>().ok_or_else(|| {
                    Error::Other("Failed to downcast Int32 column to Int32Array".to_string())
                })?;
                let values: Vec<i32> = indices.iter().map(|&idx| array.value(idx)).collect();
                Arc::new(Int32Array::from(values))
            }
            DataType::Int64 => {
                let array = column.as_any().downcast_ref::<Int64Array>().ok_or_else(|| {
                    Error::Other("Failed to downcast Int64 column to Int64Array".to_string())
                })?;
                let values: Vec<i64> = indices.iter().map(|&idx| array.value(idx)).collect();
                Arc::new(Int64Array::from(values))
            }
            DataType::Float32 => {
                let array = column.as_any().downcast_ref::<Float32Array>().ok_or_else(|| {
                    Error::Other("Failed to downcast Float32 column to Float32Array".to_string())
                })?;
                let values: Vec<f32> = indices.iter().map(|&idx| array.value(idx)).collect();
                Arc::new(Float32Array::from(values))
            }
            DataType::Float64 => {
                let array = column.as_any().downcast_ref::<Float64Array>().ok_or_else(|| {
                    Error::Other("Failed to downcast Float64 column to Float64Array".to_string())
                })?;
                let values: Vec<f64> = indices.iter().map(|&idx| array.value(idx)).collect();
                Arc::new(Float64Array::from(values))
            }
            DataType::Utf8 => {
                let array = column.as_any().downcast_ref::<StringArray>().ok_or_else(|| {
                    Error::Other("Failed to downcast Utf8 column to StringArray".to_string())
                })?;
                let values: Vec<&str> = indices.iter().map(|&idx| array.value(idx)).collect();
                Arc::new(StringArray::from(values))
            }
            dt => {
                return Err(Error::InvalidInput(format!(
                    "Top-K not implemented for column data type: {dt:?}"
                )));
            }
        };

        new_columns.push(new_array);
    }

    RecordBatch::try_new(batch.schema(), new_columns)
        .map_err(|e| Error::StorageError(format!("Failed to create result batch: {e}")))
}

/// Fallback: sort all rows when k >= `num_rows`
fn sort_all_rows(
    batch: &RecordBatch,
    column_index: usize,
    order: SortOrder,
) -> crate::Result<RecordBatch> {
    use arrow::compute::sort_to_indices;

    let sort_options = SortOptions::from(order);
    let indices = sort_to_indices(batch.column(column_index).as_ref(), Some(sort_options), None)
        .map_err(|e| Error::StorageError(format!("Failed to sort: {e}")))?;

    // Convert indices to usize vec
    let indices_array =
        indices.as_any().downcast_ref::<arrow::array::UInt32Array>().ok_or_else(|| {
            Error::Other(
                "Failed to downcast sort indices to UInt32Array (expected from sort_to_indices)"
                    .to_string(),
            )
        })?;
    let indices_vec: Vec<usize> =
        (0..indices_array.len()).map(|i| indices_array.value(i) as usize).collect();

    build_batch_from_indices(batch, &indices_vec)
}

#[cfg(test)]
#[allow(
    clippy::cast_possible_truncation,
    clippy::cast_possible_wrap,
    clippy::cast_precision_loss,
    clippy::float_cmp,
    clippy::redundant_closure
)]
mod tests {
    use super::*;
    use arrow::datatypes::{DataType, Field, Schema};
    use std::sync::Arc;

    fn create_test_batch(values: Vec<f64>) -> RecordBatch {
        let schema = Arc::new(Schema::new(vec![
            Field::new("id", DataType::Int32, false),
            Field::new("score", DataType::Float64, false),
        ]));

        let ids: Vec<i32> = (0..values.len() as i32).collect();

        RecordBatch::try_new(
            schema,
            vec![Arc::new(Int32Array::from(ids)), Arc::new(Float64Array::from(values))],
        )
        .unwrap()
    }

    #[test]
    fn test_top_k_descending_basic() {
        // Test: Get top 3 highest scores
        let batch = create_test_batch(vec![1.0, 5.0, 3.0, 9.0, 2.0]);
        let result = batch.top_k(1, 3, SortOrder::Descending).unwrap();

        assert_eq!(result.num_rows(), 3);

        let scores = result.column(1).as_any().downcast_ref::<Float64Array>().unwrap();
        assert_eq!(scores.value(0), 9.0);
        assert_eq!(scores.value(1), 5.0);
        assert_eq!(scores.value(2), 3.0);
    }

    #[test]
    fn test_top_k_ascending_basic() {
        // Test: Get top 3 lowest scores
        let batch = create_test_batch(vec![1.0, 5.0, 3.0, 9.0, 2.0]);
        let result = batch.top_k(1, 3, SortOrder::Ascending).unwrap();

        assert_eq!(result.num_rows(), 3);

        let scores = result.column(1).as_any().downcast_ref::<Float64Array>().unwrap();
        assert_eq!(scores.value(0), 1.0);
        assert_eq!(scores.value(1), 2.0);
        assert_eq!(scores.value(2), 3.0);
    }

    #[test]
    fn test_top_k_k_equals_length() {
        // Edge case: k equals number of rows (should return sorted batch)
        let batch = create_test_batch(vec![3.0, 1.0, 2.0]);
        let result = batch.top_k(1, 3, SortOrder::Descending).unwrap();

        assert_eq!(result.num_rows(), 3);

        let scores = result.column(1).as_any().downcast_ref::<Float64Array>().unwrap();
        assert_eq!(scores.value(0), 3.0);
        assert_eq!(scores.value(1), 2.0);
        assert_eq!(scores.value(2), 1.0);
    }

    #[test]
    fn test_top_k_k_greater_than_length() {
        // Edge case: k > number of rows (should return all rows sorted)
        let batch = create_test_batch(vec![3.0, 1.0, 2.0]);
        let result = batch.top_k(1, 10, SortOrder::Descending).unwrap();

        assert_eq!(result.num_rows(), 3);

        let scores = result.column(1).as_any().downcast_ref::<Float64Array>().unwrap();
        assert_eq!(scores.value(0), 3.0);
        assert_eq!(scores.value(1), 2.0);
        assert_eq!(scores.value(2), 1.0);
    }

    #[test]
    fn test_top_k_k_zero_fails() {
        // Error case: k = 0 should fail
        let batch = create_test_batch(vec![1.0, 2.0, 3.0]);
        let result = batch.top_k(1, 0, SortOrder::Descending);

        assert!(result.is_err());
        assert!(result.unwrap_err().to_string().contains("must be greater than 0"));
    }

    #[test]
    fn test_top_k_invalid_column_index() {
        // Error case: invalid column index
        let batch = create_test_batch(vec![1.0, 2.0, 3.0]);
        let result = batch.top_k(99, 2, SortOrder::Descending);

        assert!(result.is_err());
        assert!(result.unwrap_err().to_string().contains("out of bounds"));
    }

    #[test]
    fn test_top_k_preserves_row_integrity() {
        // Test: Ensure all columns stay aligned (row integrity)
        let batch = create_test_batch(vec![1.0, 5.0, 3.0]);
        let result = batch.top_k(1, 2, SortOrder::Descending).unwrap();

        let ids = result.column(0).as_any().downcast_ref::<Int32Array>().unwrap();
        let scores = result.column(1).as_any().downcast_ref::<Float64Array>().unwrap();

        // Top 2: scores 5.0 (id=1) and 3.0 (id=2)
        assert_eq!(scores.value(0), 5.0);
        assert_eq!(ids.value(0), 1);

        assert_eq!(scores.value(1), 3.0);
        assert_eq!(ids.value(1), 2);
    }

    #[test]
    fn test_top_k_large_dataset() {
        // Performance test: 1M rows (should be O(N) vs O(N log N))
        let values: Vec<f64> = (0..1_000_000).map(|i| f64::from(i)).collect();
        let batch = create_test_batch(values);

        let result = batch.top_k(1, 10, SortOrder::Descending).unwrap();

        assert_eq!(result.num_rows(), 10);

        let scores = result.column(1).as_any().downcast_ref::<Float64Array>().unwrap();
        // Top 10 should be 999999, 999998, ..., 999990
        for i in 0..10 {
            assert_eq!(scores.value(i), 999_999.0 - i as f64);
        }

        // Performance assertion removed — flaky on shared CI runners.
        // Target for release builds: <80ms for 1M rows.
        // Verify with `cargo bench` not wall-clock assertions in tests.
        // GH-739: 633ms on loaded runner triggered false failure.
    }

    // Property-based tests
    #[cfg(test)]
    mod property_tests {
        use super::*;
        use proptest::prelude::*;

        proptest! {
            /// Property: Top-K always returns exactly K rows (or fewer if input is smaller)
            #[test]
            fn prop_top_k_returns_k_rows(
                values in prop::collection::vec(0.0f64..1000.0, 10..1000),
                k in 1usize..100
            ) {
                let batch = create_test_batch(values.clone());
                let result = batch.top_k(1, k, SortOrder::Descending).unwrap();

                let expected_rows = k.min(values.len());
                prop_assert_eq!(result.num_rows(), expected_rows);
            }

            /// Property: Top-K descending returns values in descending order
            #[test]
            fn prop_top_k_descending_is_sorted(
                values in prop::collection::vec(0.0f64..1000.0, 10..1000),
                k in 1usize..100
            ) {
                let batch = create_test_batch(values);
                let result = batch.top_k(1, k, SortOrder::Descending).unwrap();

                let scores = result.column(1).as_any().downcast_ref::<Float64Array>().unwrap();

                // Check descending order
                for i in 0..scores.len().saturating_sub(1) {
                    prop_assert!(
                        scores.value(i) >= scores.value(i + 1),
                        "Not in descending order: {} < {}",
                        scores.value(i),
                        scores.value(i + 1)
                    );
                }
            }

            /// Property: Top-K ascending returns values in ascending order
            #[test]
            fn prop_top_k_ascending_is_sorted(
                values in prop::collection::vec(0.0f64..1000.0, 10..1000),
                k in 1usize..100
            ) {
                let batch = create_test_batch(values);
                let result = batch.top_k(1, k, SortOrder::Ascending).unwrap();

                let scores = result.column(1).as_any().downcast_ref::<Float64Array>().unwrap();

                // Check ascending order
                for i in 0..scores.len().saturating_sub(1) {
                    prop_assert!(
                        scores.value(i) <= scores.value(i + 1),
                        "Not in ascending order: {} > {}",
                        scores.value(i),
                        scores.value(i + 1)
                    );
                }
            }
        }
    }

    // Additional tests for all data types
    #[test]
    fn test_top_k_int32() {
        use arrow::array::Int32Array;
        use arrow::datatypes::{DataType, Field, Schema};
        use std::sync::Arc;

        let schema = Schema::new(vec![Field::new("value", DataType::Int32, false)]);
        let values = Int32Array::from(vec![5, 2, 8, 1, 9, 3]);
        let batch = RecordBatch::try_new(Arc::new(schema), vec![Arc::new(values)]).unwrap();

        let result = batch.top_k(0, 3, SortOrder::Descending).unwrap();
        assert_eq!(result.num_rows(), 3);

        let col = result.column(0).as_any().downcast_ref::<Int32Array>().unwrap();
        assert_eq!(col.value(0), 9);
        assert_eq!(col.value(1), 8);
        assert_eq!(col.value(2), 5);
    }

    #[test]
    fn test_top_k_int32_ascending() {
        use arrow::array::Int32Array;
        use arrow::datatypes::{DataType, Field, Schema};
        use std::sync::Arc;

        let schema = Schema::new(vec![Field::new("value", DataType::Int32, false)]);
        let values = Int32Array::from(vec![5, 2, 8, 1, 9, 3]);
        let batch = RecordBatch::try_new(Arc::new(schema), vec![Arc::new(values)]).unwrap();

        let result = batch.top_k(0, 3, SortOrder::Ascending).unwrap();
        assert_eq!(result.num_rows(), 3);

        let col = result.column(0).as_any().downcast_ref::<Int32Array>().unwrap();
        assert_eq!(col.value(0), 1);
        assert_eq!(col.value(1), 2);
        assert_eq!(col.value(2), 3);
    }

    #[test]
    fn test_top_k_int64() {
        use arrow::array::Int64Array;
        use arrow::datatypes::{DataType, Field, Schema};
        use std::sync::Arc;

        let schema = Schema::new(vec![Field::new("value", DataType::Int64, false)]);
        let values = Int64Array::from(vec![100i64, 200, 50, 300, 150]);
        let batch = RecordBatch::try_new(Arc::new(schema), vec![Arc::new(values)]).unwrap();

        let result = batch.top_k(0, 2, SortOrder::Ascending).unwrap();
        assert_eq!(result.num_rows(), 2);

        let col = result.column(0).as_any().downcast_ref::<Int64Array>().unwrap();
        assert_eq!(col.value(0), 50);
        assert_eq!(col.value(1), 100);
    }

    #[test]
    fn test_top_k_int64_descending() {
        use arrow::array::Int64Array;
        use arrow::datatypes::{DataType, Field, Schema};
        use std::sync::Arc;

        let schema = Schema::new(vec![Field::new("value", DataType::Int64, false)]);
        let values = Int64Array::from(vec![100i64, 200, 50, 300, 150]);
        let batch = RecordBatch::try_new(Arc::new(schema), vec![Arc::new(values)]).unwrap();

        let result = batch.top_k(0, 2, SortOrder::Descending).unwrap();
        assert_eq!(result.num_rows(), 2);

        let col = result.column(0).as_any().downcast_ref::<Int64Array>().unwrap();
        assert_eq!(col.value(0), 300);
        assert_eq!(col.value(1), 200);
    }

    #[test]
    fn test_top_k_float32() {
        use arrow::array::Float32Array;
        use arrow::datatypes::{DataType, Field, Schema};
        use std::sync::Arc;

        let schema = Schema::new(vec![Field::new("value", DataType::Float32, false)]);
        let values = Float32Array::from(vec![1.5f32, 2.7, 0.3, 4.2, 3.1]);
        let batch = RecordBatch::try_new(Arc::new(schema), vec![Arc::new(values)]).unwrap();

        let result = batch.top_k(0, 3, SortOrder::Descending).unwrap();
        assert_eq!(result.num_rows(), 3);

        let col = result.column(0).as_any().downcast_ref::<Float32Array>().unwrap();
        assert!((col.value(0) - 4.2).abs() < 0.001);
        assert!((col.value(1) - 3.1).abs() < 0.001);
        assert!((col.value(2) - 2.7).abs() < 0.001);
    }

    #[test]
    fn test_top_k_float32_ascending() {
        use arrow::array::Float32Array;
        use arrow::datatypes::{DataType, Field, Schema};
        use std::sync::Arc;

        let schema = Schema::new(vec![Field::new("value", DataType::Float32, false)]);
        let values = Float32Array::from(vec![1.5f32, 2.7, 0.3, 4.2, 3.1]);
        let batch = RecordBatch::try_new(Arc::new(schema), vec![Arc::new(values)]).unwrap();

        let result = batch.top_k(0, 3, SortOrder::Ascending).unwrap();
        assert_eq!(result.num_rows(), 3);

        let col = result.column(0).as_any().downcast_ref::<Float32Array>().unwrap();
        assert!((col.value(0) - 0.3).abs() < 0.001);
        assert!((col.value(1) - 1.5).abs() < 0.001);
        assert!((col.value(2) - 2.7).abs() < 0.001);
    }

    #[test]
    fn test_top_k_unsupported_type() {
        use arrow::array::StringArray;
        use arrow::datatypes::{DataType, Field, Schema};
        use std::sync::Arc;

        let schema = Schema::new(vec![Field::new("value", DataType::Utf8, false)]);
        let values = StringArray::from(vec!["a", "b", "c"]);
        let batch = RecordBatch::try_new(Arc::new(schema), vec![Arc::new(values)]).unwrap();

        let result = batch.top_k(0, 2, SortOrder::Descending);
        assert!(result.is_err());
        assert!(result.unwrap_err().to_string().contains("Top-K not supported for data type"));
    }

    // ========================================================================
    // Heap Item Trait Tests (for coverage of MinHeapItem/MaxHeapItem)
    // ========================================================================

    #[test]
    fn test_min_heap_item_eq() {
        let item1 = MinHeapItem { value: 42i32, index: 0 };
        let item2 = MinHeapItem { value: 42i32, index: 1 };
        let item3 = MinHeapItem { value: 43i32, index: 2 };

        assert_eq!(item1, item2);
        assert_ne!(item1, item3);
    }

    #[test]
    fn test_min_heap_item_ord() {
        let item1 = MinHeapItem { value: 10i32, index: 0 };
        let item2 = MinHeapItem { value: 20i32, index: 1 };
        let item3 = MinHeapItem { value: 30i32, index: 2 };

        // Min-heap: reverse ordering (smaller values at top)
        assert!(item3 < item2); // 30 < 20 in min-heap ordering
        assert!(item2 < item1); // 20 < 10 in min-heap ordering
    }

    #[test]
    fn test_min_heap_item_partial_ord() {
        let item1 = MinHeapItem { value: 5i32, index: 0 };
        let item2 = MinHeapItem { value: 10i32, index: 1 };

        assert!(item1.partial_cmp(&item2) == Some(Ordering::Greater));
    }

    #[test]
    fn test_max_heap_item_eq() {
        let item1 = MaxHeapItem { value: 42i32, index: 0 };
        let item2 = MaxHeapItem { value: 42i32, index: 1 };
        let item3 = MaxHeapItem { value: 43i32, index: 2 };

        assert_eq!(item1, item2);
        assert_ne!(item1, item3);
    }

    #[test]
    fn test_max_heap_item_ord() {
        let item1 = MaxHeapItem { value: 10i32, index: 0 };
        let item2 = MaxHeapItem { value: 20i32, index: 1 };
        let item3 = MaxHeapItem { value: 30i32, index: 2 };

        // Max-heap: normal ordering (larger values at top)
        assert!(item3 > item2);
        assert!(item2 > item1);
    }

    #[test]
    fn test_max_heap_item_partial_ord() {
        let item1 = MaxHeapItem { value: 5i32, index: 0 };
        let item2 = MaxHeapItem { value: 10i32, index: 1 };

        assert!(item1.partial_cmp(&item2) == Some(Ordering::Less));
    }

    #[test]
    fn test_heap_item_with_floats() {
        let item1 = MinHeapItem { value: 1.5f64, index: 0 };
        let item2 = MinHeapItem { value: 2.5f64, index: 1 };

        assert_ne!(item1, item2);
        assert!(item2 < item1); // Min-heap: reverse ordering
    }

    #[test]
    fn test_heap_item_eq_method_with_floats() {
        let item1 = MaxHeapItem { value: 3.25f64, index: 0 };
        let item2 = MaxHeapItem { value: 3.25f64, index: 1 };
        let item3 = MaxHeapItem { value: 2.75f64, index: 2 };

        assert!(item1.eq(&item2));
        assert!(!item1.eq(&item3));
    }
}