oxibase 0.2.1

// Copyright 2025 Stoolap Contributors
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

//! Parallel Query Execution
//!
//! This module provides parallel execution strategies for CPU-intensive query operations:
//!
//! - **Parallel Scan + Filter**: Process table rows in parallel chunks with WHERE evaluation
//! - **Parallel Aggregation**: Already implemented in aggregation.rs
//! - **Parallel Sort**: Parallel ORDER BY using rayon's par_sort (future)
//! - **Parallel Join**: Parallel hash join build/probe phases (future)
//!
//! # Architecture
//!
//! The parallel execution model works by:
//! 1. Collecting rows from storage (sequential - storage layer limitation)
//! 2. Splitting rows into chunks for parallel processing
//! 3. Processing each chunk independently using Rayon's work-stealing scheduler
//! 4. Merging results back together
//!
//! # Thresholds
//!
//! Parallelization has overhead, so we only use it when beneficial:
//! - Table scan + filter: 10,000+ rows
//! - Aggregation: 100,000+ rows (already in aggregation.rs)
//! - ORDER BY: 50,000+ rows
//! - Hash join: 5,000+ build rows

use rayon::prelude::*;
use rustc_hash::FxHashMap;
use std::sync::atomic::{AtomicBool, Ordering};

use crate::core::{Result, Row, Value};
use crate::functions::FunctionRegistry;
use crate::parser::ast::Expression;

use super::expression::{ExpressionEval, RowFilter};
use super::utils::{hash_composite_key, hash_row, rows_equal, verify_composite_key_equality};

// Default thresholds for parallel execution - single source of truth
// These are used by both ParallelConfig and CostEstimator
pub const DEFAULT_PARALLEL_FILTER_THRESHOLD: usize = 10_000;
pub const DEFAULT_PARALLEL_SORT_THRESHOLD: usize = 50_000;
pub const DEFAULT_PARALLEL_JOIN_THRESHOLD: usize = 5_000;
pub const DEFAULT_PARALLEL_CHUNK_SIZE: usize = 2048;

/// Configuration for parallel execution
#[derive(Clone, Debug)]
pub struct ParallelConfig {
    /// Whether parallel execution is enabled
    pub enabled: bool,
    /// Minimum rows to trigger parallel scan + filter
    pub min_rows_for_parallel_filter: usize,
    /// Minimum rows to trigger parallel sort
    pub min_rows_for_parallel_sort: usize,
    /// Minimum build rows to trigger parallel hash join
    pub min_rows_for_parallel_join: usize,
    /// Chunk size for parallel processing (rows per thread task)
    pub chunk_size: usize,
}

impl Default for ParallelConfig {
    fn default() -> Self {
        Self {
            enabled: true,
            min_rows_for_parallel_filter: DEFAULT_PARALLEL_FILTER_THRESHOLD,
            min_rows_for_parallel_sort: DEFAULT_PARALLEL_SORT_THRESHOLD,
            min_rows_for_parallel_join: DEFAULT_PARALLEL_JOIN_THRESHOLD,
            // Optimal chunk size balances:
            // - Too small: excessive task scheduling overhead
            // - Too large: poor load balancing if chunks have varying filter selectivity
            // 2048 is a good default that works well with typical L2 cache sizes
            chunk_size: DEFAULT_PARALLEL_CHUNK_SIZE,
        }
    }
}

impl ParallelConfig {
    /// Create a new parallel config with custom settings
    pub fn new(
        enabled: bool,
        min_rows_for_parallel_filter: usize,
        min_rows_for_parallel_sort: usize,
        min_rows_for_parallel_join: usize,
        chunk_size: usize,
    ) -> Self {
        Self {
            enabled,
            min_rows_for_parallel_filter,
            min_rows_for_parallel_sort,
            min_rows_for_parallel_join,
            chunk_size,
        }
    }

    /// Create a config with parallel execution disabled
    pub fn disabled() -> Self {
        Self {
            enabled: false,
            ..Default::default()
        }
    }

    /// Check if parallel filter should be used for the given row count
    #[inline]
    pub fn should_parallel_filter(&self, row_count: usize) -> bool {
        self.enabled && row_count >= self.min_rows_for_parallel_filter
    }

    /// Check if parallel sort should be used for the given row count
    #[inline]
    pub fn should_parallel_sort(&self, row_count: usize) -> bool {
        self.enabled && row_count >= self.min_rows_for_parallel_sort
    }

    /// Check if parallel join should be used for the given build side row count
    #[inline]
    pub fn should_parallel_join(&self, build_rows: usize) -> bool {
        self.enabled && build_rows >= self.min_rows_for_parallel_join
    }
}

/// Parallel filter execution for WHERE clause evaluation
///
/// This function filters rows in parallel by:
/// 1. Splitting rows into chunks
/// 2. Evaluating the WHERE predicate on each chunk in parallel
/// 3. Collecting matching rows from all chunks
///
/// # Performance
///
/// For a table with 1M rows and 50% selectivity:
/// - Sequential: ~500ms
/// - Parallel (8 cores): ~80ms (6x speedup)
///
/// The speedup depends on:
/// - Number of CPU cores
/// - Complexity of the WHERE predicate
/// - Selectivity (how many rows pass the filter)
///
/// CRITICAL: This function now returns Result to properly propagate compilation errors.
/// Previously, errors were silently swallowed which could cause data loss.
pub fn parallel_filter(
    rows: Vec<Row>,
    filter_expr: &Expression,
    columns: &[String],
    function_registry: &FunctionRegistry,
    config: &ParallelConfig,
) -> Result<Vec<Row>> {
    let row_count = rows.len();

    // Check if parallel execution is beneficial
    if !config.should_parallel_filter(row_count) {
        // Fall back to sequential filtering
        return sequential_filter(rows, filter_expr, columns, function_registry);
    }

    // Calculate optimal chunk size based on available parallelism
    // Goal: Create ~2-4 chunks per thread for good load balancing
    // Formula: max(config.chunk_size, row_count / (num_threads * 4), 512)
    // This ensures chunks aren't too small (scheduling overhead) or too large (poor balancing)
    let num_threads = rayon::current_num_threads();
    let target_chunks = num_threads * 4; // 4 chunks per thread for work-stealing
    let chunk_size = (row_count / target_chunks).max(config.chunk_size).max(512);

    // Pre-compile the filter expression once (RowFilter is Send+Sync)
    // This avoids compiling the expression in each thread
    // CRITICAL: Propagate compilation errors instead of silently falling back
    let columns_vec: Vec<String> = columns.to_vec();
    let filter = RowFilter::new(filter_expr, &columns_vec)?;

    // Process chunks in parallel
    let filtered_chunks: Vec<Vec<Row>> = rows
        .into_par_iter()
        .chunks(chunk_size)
        .map(|chunk| {
            let mut filtered = Vec::with_capacity(chunk.len() / 2); // Estimate 50% selectivity

            for row in chunk {
                // SQL semantics: NULL or error in WHERE predicate => row excluded
                // This matches standard SQL behavior where NULL is neither true nor false,
                // so rows with NULL predicate results are filtered out.
                // Errors (e.g., type mismatches) are treated the same way.
                if filter.matches(&row) {
                    filtered.push(row);
                }
            }

            filtered
        })
        .collect();

    // Merge results from all chunks
    // Pre-calculate total capacity for single allocation
    let total_size: usize = filtered_chunks.iter().map(|c| c.len()).sum();
    let mut result = Vec::with_capacity(total_size);
    for chunk in filtered_chunks {
        result.extend(chunk);
    }

    Ok(result)
}

/// Sequential filter for small datasets or when parallel is disabled
///
/// CRITICAL: This function now returns Result to properly propagate compilation errors.
/// Previously, errors were silently swallowed which could cause data loss.
fn sequential_filter(
    rows: Vec<Row>,
    filter_expr: &Expression,
    columns: &[String],
    _function_registry: &FunctionRegistry,
) -> Result<Vec<Row>> {
    let columns_vec: Vec<String> = columns.to_vec();
    // CRITICAL: Propagate compilation errors instead of silently returning empty
    let mut eval = ExpressionEval::compile(filter_expr, &columns_vec)?;

    Ok(rows.into_iter().filter(|row| eval.eval_bool(row)).collect())
}

/// Parallel filter with ownership transfer (more efficient for large results)
///
/// This variant uses parallel iterators that take ownership of rows,
/// which is more efficient when most rows pass the filter.
pub fn parallel_filter_owned(
    rows: Vec<Row>,
    predicate: impl Fn(&Row) -> bool + Sync + Send,
    config: &ParallelConfig,
) -> Vec<Row> {
    let row_count = rows.len();

    if !config.should_parallel_filter(row_count) {
        // Sequential fallback
        return rows.into_iter().filter(|r| predicate(r)).collect();
    }

    // Use rayon's parallel filter
    rows.into_par_iter().filter(|r| predicate(r)).collect()
}

/// Parallel sort using rayon's par_sort_by
///
/// For large datasets, parallel sort can provide 2-4x speedup.
pub fn parallel_sort<F>(rows: &mut [Row], compare: F, config: &ParallelConfig)
where
    F: Fn(&Row, &Row) -> std::cmp::Ordering + Sync + Send,
{
    if config.should_parallel_sort(rows.len()) {
        rows.par_sort_by(compare);
    } else {
        rows.sort_by(compare);
    }
}

/// Parallel sort that's unstable (faster but doesn't preserve order of equal elements)
pub fn parallel_sort_unstable<F>(rows: &mut [Row], compare: F, config: &ParallelConfig)
where
    F: Fn(&Row, &Row) -> std::cmp::Ordering + Sync + Send,
{
    if config.should_parallel_sort(rows.len()) {
        rows.par_sort_unstable_by(compare);
    } else {
        rows.sort_unstable_by(compare);
    }
}

/// Parallel DISTINCT processing using hash map with proper equality checking
///
/// Removes duplicate rows in parallel by:
/// 1. Splitting rows into chunks
/// 2. Each chunk builds a local hash map (hash -> list of rows)
/// 3. Final merge pass removes cross-chunk duplicates with full equality check
///
/// This implementation correctly handles hash collisions by storing actual rows
/// and comparing them for equality, not just their hashes.
pub fn parallel_distinct(rows: Vec<Row>, config: &ParallelConfig) -> Vec<Row> {
    let row_count = rows.len();

    if !config.should_parallel_filter(row_count) {
        // Sequential distinct with proper equality checking
        return sequential_distinct(rows);
    }

    // For parallel distinct, we use a two-phase approach:
    // Phase 1: Filter locally within chunks (parallel) - uses hash map with collision lists
    // Phase 2: Final dedup across chunks (sequential) - verifies equality for hash matches

    let num_threads = rayon::current_num_threads();
    let chunk_size = config.chunk_size.max(row_count / num_threads).max(1000);

    // Phase 1: Parallel local dedup within chunks
    // Each chunk produces unique rows (within that chunk) along with their hashes
    // We store (hash, row) pairs to avoid recomputing hashes
    let deduped_chunks: Vec<Vec<(u64, Row)>> = rows
        .into_par_iter()
        .chunks(chunk_size)
        .map(|chunk| {
            // Use hash map: hash -> list of indices into unique_with_hashes
            // OPTIMIZATION: Use FxHashMap for fastest hash operations with trusted keys
            let mut hash_to_indices: FxHashMap<u64, Vec<usize>> = FxHashMap::default();
            // Store (hash, row) pairs to avoid recomputing hash later
            let mut unique_with_hashes: Vec<(u64, Row)> = Vec::with_capacity(chunk.len());

            for row in chunk {
                let hash = hash_row(&row);
                let indices = hash_to_indices.entry(hash).or_default();

                // Check if this exact row already exists (handle hash collisions)
                let is_duplicate = indices
                    .iter()
                    .any(|&idx| rows_equal(&unique_with_hashes[idx].1, &row));

                if !is_duplicate {
                    indices.push(unique_with_hashes.len());
                    unique_with_hashes.push((hash, row));
                }
            }

            unique_with_hashes
        })
        .collect();

    // Phase 2: Sequential final dedup across chunks with full equality verification
    // Use hash map: hash -> list of indices into result vec
    // OPTIMIZATION: Use FxHashMap for fastest hash operations with trusted keys
    let total_size: usize = deduped_chunks.iter().map(|chunk| chunk.len()).sum();
    // OPTIMIZATION: Estimate final size accounting for cross-chunk duplicates
    // Phase 1 removed intra-chunk dupes, but inter-chunk dupes remain
    // Empirically, 75% of phase 1 output survives phase 2 (25% are cross-chunk dupes)
    let estimated_size = (total_size * 3) / 4;
    let mut result = Vec::with_capacity(estimated_size);
    let mut hash_to_indices: FxHashMap<u64, Vec<usize>> = FxHashMap::default();

    for chunk in deduped_chunks {
        for (hash, row) in chunk {
            let indices = hash_to_indices.entry(hash).or_default();

            // Check if this exact row already exists globally (handle hash collisions)
            let is_duplicate = indices.iter().any(|&idx| rows_equal(&result[idx], &row));

            if !is_duplicate {
                indices.push(result.len());
                result.push(row); // Move, no clone
            }
        }
    }

    result
}

/// Sequential DISTINCT with proper equality checking for hash collisions
fn sequential_distinct(rows: Vec<Row>) -> Vec<Row> {
    // Use hash map: hash -> list of indices into result vec (no cloning needed)
    // OPTIMIZATION: Use FxHashMap for fastest hash operations with trusted keys
    let mut hash_to_indices: FxHashMap<u64, Vec<usize>> = FxHashMap::default();
    let mut result = Vec::with_capacity(rows.len());

    for row in rows {
        let hash = hash_row(&row);
        let indices = hash_to_indices.entry(hash).or_default();

        // Check if this exact row already exists (handle hash collisions)
        let is_duplicate = indices.iter().any(|&idx| rows_equal(&result[idx], &row));

        if !is_duplicate {
            indices.push(result.len());
            result.push(row); // Move, no clone
        }
    }

    result
}

/// Parallel projection - evaluate expressions on rows in parallel
///
/// For complex SELECT expressions (not simple column references),
/// parallel evaluation can speed up projection significantly.
pub fn parallel_project<F>(rows: Vec<Row>, project_fn: F, config: &ParallelConfig) -> Vec<Row>
where
    F: Fn(&Row) -> Row + Sync + Send,
{
    let row_count = rows.len();

    if !config.should_parallel_filter(row_count) {
        return rows.iter().map(&project_fn).collect();
    }

    rows.into_par_iter().map(|r| project_fn(&r)).collect()
}

/// Statistics about parallel execution for monitoring/debugging
#[derive(Clone, Debug, Default)]
pub struct ParallelStats {
    /// Number of rows processed
    pub rows_processed: usize,
    /// Number of rows that passed the filter
    pub rows_passed: usize,
    /// Number of chunks used
    pub chunks_used: usize,
    /// Whether parallel execution was used
    pub parallel_used: bool,
}

// ============================================================================
// Parallel Hash Join
// ============================================================================

/// Hash table storage that adapts between sequential and parallel execution
enum HashTableStorage {
    /// Sequential execution using FxHashMap (optimized for trusted keys)
    Sequential(FxHashMap<u64, Vec<usize>>),
    /// Parallel execution using DashMap (concurrent, lock-free)
    Parallel(dashmap::DashMap<u64, Vec<usize>>),
}

/// Build side match tracking using atomic operations
///
/// Uses Vec<AtomicBool> for both sequential and parallel execution to ensure
/// the type is Sync and can be safely shared across threads. The atomic overhead
/// in sequential mode is minimal (~1-2 nanoseconds per operation).
struct BuildMatchedTracker {
    matched: Vec<AtomicBool>,
}

impl BuildMatchedTracker {
    /// Create a new tracker
    fn new(size: usize) -> Self {
        BuildMatchedTracker {
            matched: (0..size).map(|_| AtomicBool::new(false)).collect(),
        }
    }

    /// Mark a build row as matched
    ///
    /// Uses Release ordering in parallel mode for cross-thread visibility.
    /// In sequential mode, Relaxed would suffice, but we use Release uniformly
    /// for simplicity and the overhead is negligible.
    #[inline]
    fn mark_matched(&self, idx: usize) {
        self.matched[idx].store(true, Ordering::Release);
    }

    /// Check if a build row was matched
    ///
    /// Uses Acquire ordering to synchronize with Release stores from probe phase.
    #[inline]
    fn was_matched(&self, idx: usize) -> bool {
        self.matched[idx].load(Ordering::Acquire)
    }
}

impl HashTableStorage {
    /// Get matching build row indices for a hash key
    #[inline]
    fn get(&self, key: &u64) -> Option<Vec<usize>> {
        match self {
            HashTableStorage::Sequential(map) => map.get(key).cloned(),
            HashTableStorage::Parallel(map) => map.get(key).map(|v| v.clone()),
        }
    }
}

/// Result of parallel hash table build phase
pub struct ParallelHashTable {
    /// The hash table mapping composite key hashes to row indices
    storage: HashTableStorage,
    /// Number of rows in the build side
    pub row_count: usize,
}

impl ParallelHashTable {
    /// Get matching build row indices for a hash key
    #[inline]
    pub fn get(&self, key: &u64) -> Option<Vec<usize>> {
        self.storage.get(key)
    }
}

/// Build a hash table in parallel for hash join
///
/// This function builds the hash table from the build side of a join in parallel:
/// 1. Split build rows into chunks
/// 2. Each thread computes hashes and builds a partial hash table
/// 3. Merge partial tables into a concurrent DashMap
///
/// # Performance
///
/// For 100K build rows:
/// - Sequential: ~15ms
/// - Parallel (8 cores): ~4ms (3.5x speedup)
pub fn parallel_hash_build(
    build_rows: &[Row],
    key_indices: &[usize],
    config: &ParallelConfig,
) -> ParallelHashTable {
    use dashmap::DashMap;

    let row_count = build_rows.len();

    if !config.should_parallel_join(row_count) {
        // OPTIMIZATION: Sequential build uses FxHashMap for best performance with trusted keys
        // FxHashMap is optimized for non-adversarial use cases (embedded database)
        let mut table: FxHashMap<u64, Vec<usize>> =
            FxHashMap::with_capacity_and_hasher(row_count, Default::default());
        for (idx, row) in build_rows.iter().enumerate() {
            let hash = hash_composite_key(row, key_indices);
            table.entry(hash).or_default().push(idx);
        }
        return ParallelHashTable {
            storage: HashTableStorage::Sequential(table),
            row_count,
        };
    }

    // Parallel build using DashMap's concurrent access
    let table: DashMap<u64, Vec<usize>> = DashMap::with_capacity(row_count);
    let num_threads = rayon::current_num_threads();
    let chunk_size = config.chunk_size.max(row_count / num_threads).max(1000);

    // Process chunks in parallel, each inserting directly into the concurrent hash table
    build_rows
        .par_chunks(chunk_size)
        .enumerate()
        .for_each(|(chunk_idx, chunk)| {
            let base_idx = chunk_idx * chunk_size;
            for (local_idx, row) in chunk.iter().enumerate() {
                // SAFETY: Check for index overflow (would require ~18 quintillion rows on 64-bit)
                // Use debug_assert for zero runtime cost in release builds
                debug_assert!(
                    base_idx.checked_add(local_idx).is_some(),
                    "Index overflow in parallel hash build: base_idx={} + local_idx={}",
                    base_idx,
                    local_idx
                );
                let global_idx = base_idx + local_idx;
                let hash = hash_composite_key(row, key_indices);
                table.entry(hash).or_default().push(global_idx);
            }
        });

    ParallelHashTable {
        storage: HashTableStorage::Parallel(table),
        row_count,
    }
}

/// Parallel probe phase of hash join
///
/// Probe the hash table with rows from the probe side in parallel.
/// Returns matching pairs of (probe_row_idx, build_row_idx).
///
/// # Performance
///
/// For 100K probe rows against 50K build rows:
/// - Sequential: ~25ms
/// - Parallel (8 cores): ~6ms (4x speedup)
pub fn parallel_hash_probe<F>(
    probe_rows: &[Row],
    probe_key_indices: &[usize],
    hash_table: &ParallelHashTable,
    build_rows: &[Row],
    verify_match: F,
    config: &ParallelConfig,
) -> Vec<(usize, usize)>
where
    F: Fn(&Row, &Row) -> bool + Sync + Send,
{
    let probe_count = probe_rows.len();

    if !config.should_parallel_join(probe_count) {
        // Sequential probe
        let mut matches = Vec::new();
        for (probe_idx, probe_row) in probe_rows.iter().enumerate() {
            let hash = hash_composite_key(probe_row, probe_key_indices);
            if let Some(build_indices) = hash_table.get(&hash) {
                for build_idx in build_indices {
                    if verify_match(probe_row, &build_rows[build_idx]) {
                        matches.push((probe_idx, build_idx));
                    }
                }
            }
        }
        return matches;
    }

    // Parallel probe
    let num_threads = rayon::current_num_threads();
    let chunk_size = config.chunk_size.max(probe_count / num_threads).max(1000);

    probe_rows
        .par_chunks(chunk_size)
        .enumerate()
        .flat_map(|(chunk_idx, chunk)| {
            let base_idx = chunk_idx * chunk_size;
            let mut local_matches = Vec::new();

            for (local_idx, probe_row) in chunk.iter().enumerate() {
                let probe_idx = base_idx + local_idx;
                let hash = hash_composite_key(probe_row, probe_key_indices);

                if let Some(build_indices) = hash_table.get(&hash) {
                    for build_idx in build_indices {
                        if verify_match(probe_row, &build_rows[build_idx]) {
                            local_matches.push((probe_idx, build_idx));
                        }
                    }
                }
            }

            local_matches
        })
        .collect()
}

/// Hash a row using specific key column indices.
/// Alias for hash_composite_key from utils for backward compatibility.
#[inline]
pub fn hash_row_by_keys(row: &Row, key_indices: &[usize]) -> u64 {
    hash_composite_key(row, key_indices)
}

/// Verify that two rows match on their respective key columns.
/// Alias for verify_composite_key_equality from utils for backward compatibility.
#[inline]
pub fn verify_key_match(
    probe_row: &Row,
    build_row: &Row,
    probe_key_indices: &[usize],
    build_key_indices: &[usize],
) -> bool {
    verify_composite_key_equality(probe_row, build_row, probe_key_indices, build_key_indices)
}

/// Join type for parallel hash join
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum JoinType {
    Inner,
    Left,
    Right,
    Full,
}

impl JoinType {
    /// Parse join type from string
    #[allow(clippy::should_implement_trait)]
    pub fn from_str(s: &str) -> Self {
        let upper = s.to_uppercase();
        if upper.contains("FULL") {
            JoinType::Full
        } else if upper.contains("LEFT") {
            JoinType::Left
        } else if upper.contains("RIGHT") {
            JoinType::Right
        } else {
            JoinType::Inner
        }
    }

    /// Check if this join needs unmatched probe rows (NULL-extended)
    fn needs_unmatched_probe(&self, swapped: bool) -> bool {
        match self {
            JoinType::Inner => false,
            JoinType::Left => !swapped, // LEFT JOIN: unmatched left (probe when not swapped)
            JoinType::Right => swapped, // RIGHT JOIN: unmatched right (probe when swapped)
            JoinType::Full => true,     // FULL JOIN: always needs unmatched rows
        }
    }

    /// Check if this join needs unmatched build rows (NULL-extended)
    fn needs_unmatched_build(&self, swapped: bool) -> bool {
        match self {
            JoinType::Inner => false,
            JoinType::Left => swapped, // LEFT JOIN: unmatched left (build when swapped)
            JoinType::Right => !swapped, // RIGHT JOIN: unmatched right (build when not swapped)
            JoinType::Full => true,    // FULL JOIN: always needs unmatched rows
        }
    }
}

/// Parallel hash join result
pub struct ParallelJoinResult {
    /// The joined rows
    pub rows: Vec<Row>,
    /// Whether parallel execution was used
    pub parallel_used: bool,
    /// Number of probe rows processed
    pub probe_rows_processed: usize,
    /// Number of build rows
    pub build_rows_count: usize,
    /// Number of matches found
    pub matches_found: usize,
}

/// Execute a complete parallel hash join
///
/// This is the main entry point for parallel hash join execution. It handles:
/// - Parallel hash table build
/// - Parallel probe phase (for INNER joins)
/// - Sequential probe with atomic tracking (for OUTER joins)
/// - Proper NULL handling for unmatched rows
///
/// # Arguments
/// * `probe_rows` - Rows from the probe side (typically larger)
/// * `build_rows` - Rows from the build side (typically smaller)
/// * `probe_key_indices` - Column indices for join keys in probe rows
/// * `build_key_indices` - Column indices for join keys in build rows
/// * `join_type` - Type of join (INNER, LEFT, RIGHT, FULL)
/// * `probe_col_count` - Number of columns in probe rows
/// * `build_col_count` - Number of columns in build rows
/// * `swapped` - Whether left/right were swapped for build side optimization
/// * `config` - Parallel execution configuration
#[allow(clippy::too_many_arguments)]
pub fn parallel_hash_join(
    probe_rows: &[Row],
    build_rows: &[Row],
    probe_key_indices: &[usize],
    build_key_indices: &[usize],
    join_type: JoinType,
    probe_col_count: usize,
    build_col_count: usize,
    swapped: bool,
    config: &ParallelConfig,
) -> ParallelJoinResult {
    let probe_count = probe_rows.len();
    let build_count = build_rows.len();

    // Determine if we should use parallel execution
    let use_parallel =
        config.should_parallel_join(build_count) || config.should_parallel_join(probe_count);

    // Build phase: Create hash table (parallel if large enough)
    let hash_table = parallel_hash_build(build_rows, build_key_indices, config);

    // For OUTER joins, we need to track which build rows were matched
    // Uses Vec<AtomicBool> for both sequential and parallel execution (minimal overhead)
    let build_matched: Option<BuildMatchedTracker> = if join_type.needs_unmatched_build(swapped) {
        Some(BuildMatchedTracker::new(build_count))
    } else {
        None
    };

    // Probe phase
    let (matched_rows, unmatched_probe_rows) = if use_parallel && join_type == JoinType::Inner {
        // For INNER joins, we can fully parallelize the probe phase
        let matches: Vec<Row> = probe_rows
            .par_chunks(config.chunk_size.max(1000))
            .flat_map(|chunk| {
                let mut local_results = Vec::new();
                for probe_row in chunk {
                    let hash = hash_row_by_keys(probe_row, probe_key_indices);
                    if let Some(build_indices) = hash_table.get(&hash) {
                        for build_idx in build_indices {
                            let build_row = &build_rows[build_idx];
                            if verify_key_match(
                                probe_row,
                                build_row,
                                probe_key_indices,
                                build_key_indices,
                            ) {
                                let combined = combine_join_rows(
                                    probe_row,
                                    build_row,
                                    probe_col_count,
                                    build_col_count,
                                    swapped,
                                );
                                local_results.push(Row::from_values(combined));
                            }
                        }
                    }
                }
                local_results
            })
            .collect();
        (matches, Vec::new())
    } else if use_parallel {
        // For OUTER joins with parallel execution, use atomic tracking for build side
        // and collect unmatched probe rows directly in parallel
        let needs_unmatched_probe = join_type.needs_unmatched_probe(swapped);

        // Each chunk returns: (matched_rows, unmatched_probe_rows)
        let chunk_results: Vec<(Vec<Row>, Vec<Row>)> = probe_rows
            .par_chunks(config.chunk_size.max(1000))
            .map(|chunk| {
                let mut matched_results = Vec::new();
                let mut unmatched_results = Vec::new();

                for probe_row in chunk.iter() {
                    let mut matched = false;
                    let hash = hash_row_by_keys(probe_row, probe_key_indices);

                    if let Some(build_indices) = hash_table.get(&hash) {
                        for build_idx in build_indices {
                            let build_row = &build_rows[build_idx];
                            if verify_key_match(
                                probe_row,
                                build_row,
                                probe_key_indices,
                                build_key_indices,
                            ) {
                                matched = true;
                                // Mark build row as matched (uses atomic Release in parallel mode)
                                if let Some(ref tracker) = build_matched {
                                    tracker.mark_matched(build_idx);
                                }
                                let combined = combine_join_rows(
                                    probe_row,
                                    build_row,
                                    probe_col_count,
                                    build_col_count,
                                    swapped,
                                );
                                matched_results.push(Row::from_values(combined));
                            }
                        }
                    }

                    // Add unmatched probe row directly (no need for second pass)
                    if !matched && needs_unmatched_probe {
                        let values = combine_with_nulls(
                            probe_row,
                            probe_col_count,
                            build_col_count,
                            swapped,
                        );
                        unmatched_results.push(Row::from_values(values));
                    }
                }

                (matched_results, unmatched_results)
            })
            .collect();

        // Merge results from all chunks
        let total_matched: usize = chunk_results.iter().map(|(m, _)| m.len()).sum();
        let total_unmatched: usize = chunk_results.iter().map(|(_, u)| u.len()).sum();

        let mut matched_rows = Vec::with_capacity(total_matched);
        let mut unmatched_rows = Vec::with_capacity(total_unmatched);

        for (matched, unmatched) in chunk_results {
            matched_rows.extend(matched);
            unmatched_rows.extend(unmatched);
        }

        // SYNCHRONIZATION NOTE:
        // CRITICAL: This Acquire fence is REQUIRED for correctness, not optional.
        //
        // Memory Ordering Justification:
        // 1. Parallel probe writes to build_matched[] use Release ordering (line 737)
        // 2. This Acquire fence establishes a happens-before relationship
        // 3. All Release stores in parallel threads are visible after this fence
        //
        // Why we can't rely solely on Rayon's barrier:
        // - While Rayon's collect() does join all threads, this is an implementation detail
        // - Rayon's API does not formally guarantee memory ordering semantics
        // - The fence makes the synchronization contract explicit and compiler-verifiable
        //
        // Without this fence: The sequential scan below might read stale values from
        // build_matched[], causing incorrect LEFT/FULL JOIN results (missing rows).
        std::sync::atomic::fence(Ordering::Acquire);

        (matched_rows, unmatched_rows)
    } else {
        // Sequential execution for small datasets
        let mut matched_rows = Vec::new();
        let needs_unmatched_probe = join_type.needs_unmatched_probe(swapped);

        for probe_row in probe_rows.iter() {
            let hash = hash_row_by_keys(probe_row, probe_key_indices);
            let mut matched = false;

            if let Some(build_indices) = hash_table.get(&hash) {
                for build_idx in build_indices {
                    let build_row = &build_rows[build_idx];
                    if verify_key_match(probe_row, build_row, probe_key_indices, build_key_indices)
                    {
                        matched = true;
                        // Mark build row as matched (uses plain bool in sequential mode)
                        if let Some(ref tracker) = build_matched {
                            tracker.mark_matched(build_idx);
                        }
                        let combined = combine_join_rows(
                            probe_row,
                            build_row,
                            probe_col_count,
                            build_col_count,
                            swapped,
                        );
                        matched_rows.push(Row::from_values(combined));
                    }
                }
            }

            // Handle unmatched probe row for OUTER joins
            if !matched && needs_unmatched_probe {
                let values =
                    combine_with_nulls(probe_row, probe_col_count, build_col_count, swapped);
                matched_rows.push(Row::from_values(values));
            }
        }

        (matched_rows, Vec::new())
    };

    let mut result_rows = matched_rows;
    result_rows.extend(unmatched_probe_rows);

    // Handle unmatched build rows for OUTER joins
    // The Acquire fence at line 794 ensures all parallel stores are visible
    if let Some(ref tracker) = build_matched {
        for (build_idx, build_row) in build_rows.iter().enumerate() {
            if !tracker.was_matched(build_idx) {
                let values =
                    combine_build_with_nulls(build_row, build_col_count, probe_col_count, swapped);
                result_rows.push(Row::from_values(values));
            }
        }
    }

    let matches_found = result_rows.len();

    ParallelJoinResult {
        rows: result_rows,
        parallel_used: use_parallel,
        probe_rows_processed: probe_count,
        build_rows_count: build_count,
        matches_found,
    }
}

/// Combine probe and build rows into a single row, respecting swap order
#[inline]
fn combine_join_rows(
    probe_row: &Row,
    build_row: &Row,
    probe_col_count: usize,
    build_col_count: usize,
    swapped: bool,
) -> Vec<Value> {
    let mut combined = Vec::with_capacity(probe_col_count + build_col_count);
    if swapped {
        // Build was originally left, probe was originally right
        for i in 0..build_col_count {
            combined.push(
                build_row
                    .get(i)
                    .cloned()
                    .unwrap_or_else(Value::null_unknown),
            );
        }
        for i in 0..probe_col_count {
            combined.push(
                probe_row
                    .get(i)
                    .cloned()
                    .unwrap_or_else(Value::null_unknown),
            );
        }
    } else {
        // Probe is left, build is right
        for i in 0..probe_col_count {
            combined.push(
                probe_row
                    .get(i)
                    .cloned()
                    .unwrap_or_else(Value::null_unknown),
            );
        }
        for i in 0..build_col_count {
            combined.push(
                build_row
                    .get(i)
                    .cloned()
                    .unwrap_or_else(Value::null_unknown),
            );
        }
    }
    combined
}

/// Combine probe row with NULLs for unmatched probe side in OUTER joins
#[inline]
fn combine_with_nulls(
    probe_row: &Row,
    probe_col_count: usize,
    build_col_count: usize,
    swapped: bool,
) -> Vec<Value> {
    let mut combined = Vec::with_capacity(probe_col_count + build_col_count);
    if swapped {
        // Build (left) is NULL, probe (right) has values
        for _ in 0..build_col_count {
            combined.push(Value::null_unknown());
        }
        for i in 0..probe_col_count {
            combined.push(
                probe_row
                    .get(i)
                    .cloned()
                    .unwrap_or_else(Value::null_unknown),
            );
        }
    } else {
        // Probe (left) has values, build (right) is NULL
        for i in 0..probe_col_count {
            combined.push(
                probe_row
                    .get(i)
                    .cloned()
                    .unwrap_or_else(Value::null_unknown),
            );
        }
        for _ in 0..build_col_count {
            combined.push(Value::null_unknown());
        }
    }
    combined
}

/// Combine build row with NULLs for unmatched build side in OUTER joins
#[inline]
fn combine_build_with_nulls(
    build_row: &Row,
    build_col_count: usize,
    probe_col_count: usize,
    swapped: bool,
) -> Vec<Value> {
    let mut combined = Vec::with_capacity(probe_col_count + build_col_count);
    if swapped {
        // Build (left) has values, probe (right) is NULL
        for i in 0..build_col_count {
            combined.push(
                build_row
                    .get(i)
                    .cloned()
                    .unwrap_or_else(Value::null_unknown),
            );
        }
        for _ in 0..probe_col_count {
            combined.push(Value::null_unknown());
        }
    } else {
        // Probe (left) is NULL, build (right) has values
        for _ in 0..probe_col_count {
            combined.push(Value::null_unknown());
        }
        for i in 0..build_col_count {
            combined.push(
                build_row
                    .get(i)
                    .cloned()
                    .unwrap_or_else(Value::null_unknown),
            );
        }
    }
    combined
}

// ============================================================================
// Parallel ORDER BY
// ============================================================================

/// Sort direction for ORDER BY
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum SortDirection {
    Ascending,
    Descending,
}

/// Sort specification for a single column
#[derive(Clone, Debug)]
pub struct SortSpec {
    /// Column index to sort by
    pub column_index: usize,
    /// Sort direction (ASC or DESC)
    pub direction: SortDirection,
    /// NULLS FIRST or NULLS LAST
    pub nulls_first: bool,
}

/// Parallel ORDER BY execution
///
/// Sorts rows in parallel using rayon's par_sort when the dataset is large enough.
/// Supports multi-column sorting with mixed ASC/DESC directions and NULL handling.
///
/// # Performance
///
/// For 100K rows:
/// - Sequential: ~50ms
/// - Parallel (8 cores): ~15ms (3x speedup)
pub fn parallel_order_by(rows: &mut [Row], sort_specs: &[SortSpec], config: &ParallelConfig) {
    let compare = |a: &Row, b: &Row| -> std::cmp::Ordering {
        for spec in sort_specs {
            let a_val = a.get(spec.column_index);
            let b_val = b.get(spec.column_index);

            // Check for NULL values (either missing column or Value::Null)
            let a_is_null = a_val.map(|v| v.is_null()).unwrap_or(true);
            let b_is_null = b_val.map(|v| v.is_null()).unwrap_or(true);

            let ordering = match (a_is_null, b_is_null) {
                (true, true) => std::cmp::Ordering::Equal,
                (true, false) => {
                    if spec.nulls_first {
                        std::cmp::Ordering::Less
                    } else {
                        std::cmp::Ordering::Greater
                    }
                }
                (false, true) => {
                    if spec.nulls_first {
                        std::cmp::Ordering::Greater
                    } else {
                        std::cmp::Ordering::Less
                    }
                }
                (false, false) => {
                    // Both non-null, compare values
                    let a_v = a_val.unwrap();
                    let b_v = b_val.unwrap();
                    a_v.partial_cmp(b_v).unwrap_or(std::cmp::Ordering::Equal)
                }
            };

            let ordering = if spec.direction == SortDirection::Descending {
                ordering.reverse()
            } else {
                ordering
            };

            if ordering != std::cmp::Ordering::Equal {
                return ordering;
            }
        }
        std::cmp::Ordering::Equal
    };

    if config.should_parallel_sort(rows.len()) {
        rows.par_sort_by(compare);
    } else {
        rows.sort_by(compare);
    }
}

/// Parallel ORDER BY with a custom comparator function
///
/// More flexible version that accepts any comparison function.
pub fn parallel_order_by_fn<F>(rows: &mut [Row], compare: F, config: &ParallelConfig)
where
    F: Fn(&Row, &Row) -> std::cmp::Ordering + Sync + Send,
{
    if config.should_parallel_sort(rows.len()) {
        rows.par_sort_by(compare);
    } else {
        rows.sort_by(compare);
    }
}

/// Parallel ORDER BY with unstable sort (faster, but doesn't preserve order of equal elements)
pub fn parallel_order_by_unstable<F>(rows: &mut [Row], compare: F, config: &ParallelConfig)
where
    F: Fn(&Row, &Row) -> std::cmp::Ordering + Sync + Send,
{
    if config.should_parallel_sort(rows.len()) {
        rows.par_sort_unstable_by(compare);
    } else {
        rows.sort_unstable_by(compare);
    }
}

/// Parallel filter with statistics collection
///
/// CRITICAL: This function now returns Result to properly propagate compilation errors.
pub fn parallel_filter_with_stats(
    rows: Vec<Row>,
    filter_expr: &Expression,
    columns: &[String],
    function_registry: &FunctionRegistry,
    config: &ParallelConfig,
) -> Result<(Vec<Row>, ParallelStats)> {
    let row_count = rows.len();
    let parallel_used = config.should_parallel_filter(row_count);

    let num_threads = rayon::current_num_threads();
    let chunk_size = if parallel_used {
        config
            .chunk_size
            .max(row_count / (num_threads * 4))
            .max(512)
    } else {
        row_count // Single chunk for sequential
    };

    let chunks_used = row_count.div_ceil(chunk_size);

    // CRITICAL: Propagate errors with ? instead of silently swallowing them
    let result = if parallel_used {
        parallel_filter(rows, filter_expr, columns, function_registry, config)?
    } else {
        sequential_filter(rows, filter_expr, columns, function_registry)?
    };

    let stats = ParallelStats {
        rows_processed: row_count,
        rows_passed: result.len(),
        chunks_used,
        parallel_used,
    };

    Ok((result, stats))
}

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

    fn make_test_rows(count: usize) -> Vec<Row> {
        (0..count)
            .map(|i| {
                Row::from_values(vec![
                    Value::Integer(i as i64),
                    Value::Integer(i as i64 % 10),
                ])
            })
            .collect()
    }

    #[test]
    fn test_parallel_distinct() {
        // Create rows with duplicates
        let mut rows = Vec::new();
        for i in 0..1000 {
            // Each value appears 10 times
            rows.push(Row::from_values(vec![Value::Integer(i % 100)]));
        }

        let config = ParallelConfig {
            min_rows_for_parallel_filter: 100, // Lower threshold for test
            ..Default::default()
        };

        let result = parallel_distinct(rows, &config);

        // Should have 100 unique values
        assert_eq!(result.len(), 100);
    }

    #[test]
    fn test_parallel_sort() {
        let mut rows: Vec<Row> = (0..1000)
            .rev()
            .map(|i| Row::from_values(vec![Value::Integer(i)]))
            .collect();

        let config = ParallelConfig {
            min_rows_for_parallel_sort: 100,
            ..Default::default()
        };

        parallel_sort(
            &mut rows,
            |a, b| {
                let a_val = a.get(0).and_then(|v| v.as_int64()).unwrap_or(0);
                let b_val = b.get(0).and_then(|v| v.as_int64()).unwrap_or(0);
                a_val.cmp(&b_val)
            },
            &config,
        );

        // Verify sorted
        for (i, row) in rows.iter().enumerate() {
            assert_eq!(row.get(0), Some(&Value::Integer(i as i64)));
        }
    }

    #[test]
    fn test_parallel_config_thresholds() {
        let config = ParallelConfig::default();

        assert!(!config.should_parallel_filter(1000)); // Below threshold
        assert!(config.should_parallel_filter(20_000)); // Above threshold

        assert!(!config.should_parallel_sort(10_000)); // Below threshold
        assert!(config.should_parallel_sort(100_000)); // Above threshold

        let disabled = ParallelConfig::disabled();
        assert!(!disabled.should_parallel_filter(1_000_000)); // Always false when disabled
    }

    #[test]
    fn test_parallel_filter_owned() {
        let rows = make_test_rows(50_000);
        let config = ParallelConfig::default();

        // Filter rows where second column (value) < 5
        let result = parallel_filter_owned(
            rows,
            |row| {
                if let Some(Value::Integer(v)) = row.get(1) {
                    *v < 5
                } else {
                    false
                }
            },
            &config,
        );

        // 50% selectivity: values 0-4 pass out of 0-9
        assert_eq!(result.len(), 25_000);
    }

    #[test]
    fn test_sequential_fallback_small_dataset() {
        let rows = make_test_rows(100); // Below threshold
        let config = ParallelConfig::default();

        // Should use sequential path
        let result = parallel_filter_owned(
            rows,
            |row| {
                if let Some(Value::Integer(v)) = row.get(1) {
                    *v < 5
                } else {
                    false
                }
            },
            &config,
        );

        // 50% selectivity
        assert_eq!(result.len(), 50);
    }

    #[test]
    fn test_parallel_hash_build() {
        // Build side: 10K rows with id as key
        let build_rows: Vec<Row> = (0..10_000)
            .map(|i| {
                Row::from_values(vec![
                    Value::Integer(i),
                    Value::Text(format!("build_{}", i).into()),
                ])
            })
            .collect();

        let config = ParallelConfig {
            min_rows_for_parallel_join: 1000,
            ..Default::default()
        };

        let hash_table = parallel_hash_build(&build_rows, &[0], &config);

        assert_eq!(hash_table.row_count, 10_000);
        // Verify some lookups work
        let test_hash = hash_row_by_keys(&build_rows[500], &[0]);
        assert!(hash_table.get(&test_hash).is_some());
    }

    #[test]
    fn test_parallel_hash_probe() {
        // Build side: 5K rows
        let build_rows: Vec<Row> = (0..5_000)
            .map(|i| {
                Row::from_values(vec![
                    Value::Integer(i),
                    Value::Text(format!("build_{}", i).into()),
                ])
            })
            .collect();

        // Probe side: 10K rows, half will match (even numbers only)
        let probe_rows: Vec<Row> = (0..10_000)
            .map(|i| {
                Row::from_values(vec![
                    Value::Integer(i * 2 % 5_000), // Maps to 0-4999 with duplicates
                    Value::Text(format!("probe_{}", i).into()),
                ])
            })
            .collect();

        let config = ParallelConfig {
            min_rows_for_parallel_join: 1000,
            ..Default::default()
        };

        // Build hash table
        let hash_table = parallel_hash_build(&build_rows, &[0], &config);

        // Probe
        let matches = parallel_hash_probe(
            &probe_rows,
            &[0],
            &hash_table,
            &build_rows,
            |probe, build| {
                // Verify actual equality
                probe.get(0) == build.get(0)
            },
            &config,
        );

        // Each probe row should match exactly one build row
        // 10K probes, each maps to one of 5K build rows
        assert_eq!(matches.len(), 10_000);
    }

    #[test]
    fn test_verify_key_match() {
        let row1 = Row::from_values(vec![Value::Integer(1), Value::Text("a".to_string().into())]);
        let row2 = Row::from_values(vec![Value::Integer(1), Value::Text("b".to_string().into())]);
        let row3 = Row::from_values(vec![Value::Integer(2), Value::Text("a".to_string().into())]);

        // Same key column 0
        assert!(verify_key_match(&row1, &row2, &[0], &[0]));

        // Different key column 0
        assert!(!verify_key_match(&row1, &row3, &[0], &[0]));

        // Same value in column 1
        assert!(verify_key_match(&row1, &row3, &[1], &[1]));
    }

    #[test]
    fn test_parallel_order_by() {
        // Create rows with random-ish order
        let mut rows: Vec<Row> = (0..1000)
            .map(|i| {
                Row::from_values(vec![
                    Value::Integer((i * 7 + 13) % 1000), // Scrambled order
                    Value::Text(format!("row_{}", i).into()),
                ])
            })
            .collect();

        let config = ParallelConfig {
            min_rows_for_parallel_sort: 100,
            ..Default::default()
        };

        let sort_specs = vec![SortSpec {
            column_index: 0,
            direction: SortDirection::Ascending,
            nulls_first: false,
        }];

        parallel_order_by(&mut rows, &sort_specs, &config);

        // Verify sorted ascending
        for i in 1..rows.len() {
            let prev = rows[i - 1].get(0).and_then(|v| v.as_int64()).unwrap();
            let curr = rows[i].get(0).and_then(|v| v.as_int64()).unwrap();
            assert!(prev <= curr, "Row {} should be <= row {}", i - 1, i);
        }
    }

    #[test]
    fn test_parallel_order_by_descending() {
        let mut rows: Vec<Row> = (0..500)
            .map(|i| Row::from_values(vec![Value::Integer(i)]))
            .collect();

        let config = ParallelConfig {
            min_rows_for_parallel_sort: 100,
            ..Default::default()
        };

        let sort_specs = vec![SortSpec {
            column_index: 0,
            direction: SortDirection::Descending,
            nulls_first: false,
        }];

        parallel_order_by(&mut rows, &sort_specs, &config);

        // Verify sorted descending
        for i in 1..rows.len() {
            let prev = rows[i - 1].get(0).and_then(|v| v.as_int64()).unwrap();
            let curr = rows[i].get(0).and_then(|v| v.as_int64()).unwrap();
            assert!(prev >= curr, "Row {} should be >= row {}", i - 1, i);
        }
    }

    #[test]
    fn test_parallel_order_by_with_nulls() {
        let mut rows: Vec<Row> = vec![
            Row::from_values(vec![Value::Integer(3)]),
            Row::from_values(vec![Value::null_unknown()]),
            Row::from_values(vec![Value::Integer(1)]),
            Row::from_values(vec![Value::null_unknown()]),
            Row::from_values(vec![Value::Integer(2)]),
        ];

        let config = ParallelConfig {
            min_rows_for_parallel_sort: 1, // Force parallel for test
            ..Default::default()
        };

        // NULLS FIRST
        let sort_specs = vec![SortSpec {
            column_index: 0,
            direction: SortDirection::Ascending,
            nulls_first: true,
        }];

        parallel_order_by(&mut rows, &sort_specs, &config);

        // First two should be NULL
        assert!(rows[0].get(0).map(|v| v.is_null()).unwrap_or(false));
        assert!(rows[1].get(0).map(|v| v.is_null()).unwrap_or(false));
        // Then 1, 2, 3
        assert_eq!(rows[2].get(0), Some(&Value::Integer(1)));
        assert_eq!(rows[3].get(0), Some(&Value::Integer(2)));
        assert_eq!(rows[4].get(0), Some(&Value::Integer(3)));
    }

    // ========================================================================
    // Edge Case Tests for Hash Collisions and OUTER Joins
    // ========================================================================

    /// Test that DISTINCT handles hash collisions correctly
    /// Two different rows with the same hash should both be preserved
    #[test]
    fn test_distinct_hash_collision_handling() {
        // Create rows that might have hash collisions by having different values
        // that could theoretically hash to the same value
        let rows = vec![
            Row::from_values(vec![Value::Integer(1), Value::Text("a".into())]),
            Row::from_values(vec![Value::Integer(1), Value::Text("b".into())]), // Different from above
            Row::from_values(vec![Value::Integer(1), Value::Text("a".into())]), // Duplicate of first
            Row::from_values(vec![Value::Integer(2), Value::Text("a".into())]), // Different from all
            Row::from_values(vec![Value::Integer(2), Value::Text("a".into())]), // Duplicate
        ];

        let config = ParallelConfig {
            min_rows_for_parallel_filter: 1, // Force parallel path
            ..Default::default()
        };

        let result = parallel_distinct(rows, &config);

        // Should have 3 unique rows:
        // (1, "a"), (1, "b"), (2, "a")
        assert_eq!(result.len(), 3, "Should have 3 unique rows");

        // Verify all three distinct combinations are present
        let has_1_a = result.iter().any(|r| {
            r.get(0) == Some(&Value::Integer(1)) && r.get(1) == Some(&Value::Text("a".into()))
        });
        let has_1_b = result.iter().any(|r| {
            r.get(0) == Some(&Value::Integer(1)) && r.get(1) == Some(&Value::Text("b".into()))
        });
        let has_2_a = result.iter().any(|r| {
            r.get(0) == Some(&Value::Integer(2)) && r.get(1) == Some(&Value::Text("a".into()))
        });

        assert!(has_1_a, "Should contain (1, 'a')");
        assert!(has_1_b, "Should contain (1, 'b')");
        assert!(has_2_a, "Should contain (2, 'a')");
    }

    /// Test sequential distinct also handles hash collisions
    #[test]
    fn test_sequential_distinct_hash_collision() {
        let rows = vec![
            Row::from_values(vec![Value::Integer(100)]),
            Row::from_values(vec![Value::Integer(200)]),
            Row::from_values(vec![Value::Integer(100)]), // Duplicate
            Row::from_values(vec![Value::Integer(300)]),
        ];

        // Use high threshold to force sequential path
        let config = ParallelConfig {
            min_rows_for_parallel_filter: 10000,
            ..Default::default()
        };

        let result = parallel_distinct(rows, &config);
        assert_eq!(
            result.len(),
            3,
            "Should have 3 unique values: 100, 200, 300"
        );
    }

    /// Test parallel hash join with hash collisions on join keys
    #[test]
    fn test_parallel_hash_join_collision_handling() {
        // Build side: rows with varying second columns but same join key
        let build_rows: Vec<Row> = vec![
            Row::from_values(vec![Value::Integer(1), Value::Text("build_a".into())]),
            Row::from_values(vec![Value::Integer(2), Value::Text("build_b".into())]),
            Row::from_values(vec![Value::Integer(3), Value::Text("build_c".into())]),
        ];

        // Probe side: rows that should match build side
        let probe_rows: Vec<Row> = vec![
            Row::from_values(vec![Value::Integer(1), Value::Text("probe_x".into())]),
            Row::from_values(vec![Value::Integer(2), Value::Text("probe_y".into())]),
            Row::from_values(vec![Value::Integer(4), Value::Text("probe_z".into())]), // No match
        ];

        let config = ParallelConfig {
            min_rows_for_parallel_join: 1, // Force parallel path
            ..Default::default()
        };

        // INNER JOIN on first column
        let result = parallel_hash_join(
            &probe_rows,
            &build_rows,
            &[0], // probe key
            &[0], // build key
            JoinType::Inner,
            2, // probe col count
            2, // build col count
            false,
            &config,
        );

        // Should have 2 matches (id=1 and id=2)
        assert_eq!(result.rows.len(), 2, "INNER JOIN should have 2 matches");

        // Verify the joined rows have correct values
        for row in &result.rows {
            // Combined row should have 4 columns (2 from probe + 2 from build)
            assert_eq!(row.len(), 4);
        }
    }

    /// Test LEFT OUTER join with unmatched probe rows
    #[test]
    fn test_parallel_left_join_unmatched() {
        let build_rows: Vec<Row> = vec![Row::from_values(vec![
            Value::Integer(1),
            Value::Text("match".into()),
        ])];

        let probe_rows: Vec<Row> = vec![
            Row::from_values(vec![Value::Integer(1), Value::Text("p1".into())]), // Matches
            Row::from_values(vec![Value::Integer(2), Value::Text("p2".into())]), // No match
            Row::from_values(vec![Value::Integer(3), Value::Text("p3".into())]), // No match
        ];

        let config = ParallelConfig {
            min_rows_for_parallel_join: 1,
            ..Default::default()
        };

        let result = parallel_hash_join(
            &probe_rows,
            &build_rows,
            &[0],
            &[0],
            JoinType::Left,
            2,
            2,
            false,
            &config,
        );

        // Should have 3 rows: 1 matched + 2 unmatched with NULL build columns
        assert_eq!(result.rows.len(), 3, "LEFT JOIN should have 3 rows");

        // Count rows with NULL in build columns (last 2 columns)
        let null_count = result
            .rows
            .iter()
            .filter(|r| {
                r.get(2).map(|v| v.is_null()).unwrap_or(false)
                    && r.get(3).map(|v| v.is_null()).unwrap_or(false)
            })
            .count();
        assert_eq!(
            null_count, 2,
            "Should have 2 unmatched rows with NULL build columns"
        );
    }

    /// Test RIGHT OUTER join with unmatched build rows
    #[test]
    fn test_parallel_right_join_unmatched() {
        let build_rows: Vec<Row> = vec![
            Row::from_values(vec![Value::Integer(1), Value::Text("b1".into())]), // Matches
            Row::from_values(vec![Value::Integer(2), Value::Text("b2".into())]), // No match
            Row::from_values(vec![Value::Integer(3), Value::Text("b3".into())]), // No match
        ];

        let probe_rows: Vec<Row> = vec![Row::from_values(vec![
            Value::Integer(1),
            Value::Text("p1".into()),
        ])];

        let config = ParallelConfig {
            min_rows_for_parallel_join: 1,
            ..Default::default()
        };

        let result = parallel_hash_join(
            &probe_rows,
            &build_rows,
            &[0],
            &[0],
            JoinType::Right,
            2,
            2,
            false,
            &config,
        );

        // Should have 3 rows: 1 matched + 2 unmatched with NULL probe columns
        assert_eq!(result.rows.len(), 3, "RIGHT JOIN should have 3 rows");

        // Count rows with NULL in probe columns (first 2 columns)
        let null_count = result
            .rows
            .iter()
            .filter(|r| {
                r.get(0).map(|v| v.is_null()).unwrap_or(false)
                    && r.get(1).map(|v| v.is_null()).unwrap_or(false)
            })
            .count();
        assert_eq!(
            null_count, 2,
            "Should have 2 unmatched rows with NULL probe columns"
        );
    }

    /// Test FULL OUTER join with unmatched rows on both sides
    #[test]
    fn test_parallel_full_outer_join() {
        let build_rows: Vec<Row> = vec![
            Row::from_values(vec![Value::Integer(1), Value::Text("b1".into())]), // Matches
            Row::from_values(vec![Value::Integer(3), Value::Text("b3".into())]), // No match
        ];

        let probe_rows: Vec<Row> = vec![
            Row::from_values(vec![Value::Integer(1), Value::Text("p1".into())]), // Matches
            Row::from_values(vec![Value::Integer(2), Value::Text("p2".into())]), // No match
        ];

        let config = ParallelConfig {
            min_rows_for_parallel_join: 1,
            ..Default::default()
        };

        let result = parallel_hash_join(
            &probe_rows,
            &build_rows,
            &[0],
            &[0],
            JoinType::Full,
            2,
            2,
            false,
            &config,
        );

        // Should have 3 rows:
        // 1 matched (id=1)
        // 1 unmatched probe (id=2, build NULL)
        // 1 unmatched build (id=3, probe NULL)
        assert_eq!(result.rows.len(), 3, "FULL OUTER JOIN should have 3 rows");
    }

    /// Test join with empty tables
    #[test]
    fn test_parallel_join_empty_tables() {
        let config = ParallelConfig::default();

        // Empty probe
        let result = parallel_hash_join(
            &[],
            &[Row::from_values(vec![Value::Integer(1)])],
            &[0],
            &[0],
            JoinType::Inner,
            1,
            1,
            false,
            &config,
        );
        assert_eq!(
            result.rows.len(),
            0,
            "Empty probe should give empty result for INNER"
        );

        // Empty build
        let result = parallel_hash_join(
            &[Row::from_values(vec![Value::Integer(1)])],
            &[],
            &[0],
            &[0],
            JoinType::Inner,
            1,
            1,
            false,
            &config,
        );
        assert_eq!(
            result.rows.len(),
            0,
            "Empty build should give empty result for INNER"
        );

        // LEFT JOIN with empty build should preserve all probe rows
        let result = parallel_hash_join(
            &[
                Row::from_values(vec![Value::Integer(1)]),
                Row::from_values(vec![Value::Integer(2)]),
            ],
            &[],
            &[0],
            &[0],
            JoinType::Left,
            1,
            1,
            false,
            &config,
        );
        assert_eq!(
            result.rows.len(),
            2,
            "LEFT JOIN with empty build should have all probe rows"
        );
    }

    /// Test join with swapped build/probe sides
    /// When swapped=true, the roles of probe and build are swapped, but the join type
    /// semantics stay the same. LEFT JOIN with swapped=true means:
    /// - Build side is actually "left" in the original query
    /// - Probe side is "right"
    /// - LEFT JOIN needs unmatched LEFT (build) rows, not probe rows
    #[test]
    fn test_parallel_join_swapped() {
        // When swapped=true:
        // - build_rows (1 row) = original left side
        // - probe_rows (2 rows) = original right side
        let build_rows: Vec<Row> = vec![
            Row::from_values(vec![Value::Integer(1), Value::Text("b1".into())]),
            Row::from_values(vec![Value::Integer(3), Value::Text("b3".into())]), // Unmatched left
        ];

        let probe_rows: Vec<Row> = vec![
            Row::from_values(vec![Value::Integer(1), Value::Text("p1".into())]),
            Row::from_values(vec![Value::Integer(2), Value::Text("p2".into())]), // Unmatched right
        ];

        let config = ParallelConfig {
            min_rows_for_parallel_join: 1,
            ..Default::default()
        };

        // LEFT JOIN with swapped=true:
        // - Build is "left", so unmatched build rows need NULL right columns
        // - Probe is "right", so unmatched probe rows are NOT included (LEFT JOIN only keeps left)
        let result = parallel_hash_join(
            &probe_rows,
            &build_rows,
            &[0],
            &[0],
            JoinType::Left,
            2,
            2,
            true, // swapped: build=left, probe=right
            &config,
        );

        // Should have 2 rows:
        // 1 matched row (id=1)
        // 1 unmatched build row (id=3) with NULL probe columns
        assert_eq!(result.rows.len(), 2, "LEFT JOIN swapped should have 2 rows");

        // Verify column order is correct (build columns first when swapped)
        // Row structure: [build_col0, build_col1, probe_col0, probe_col1]
        let matched_row = result
            .rows
            .iter()
            .find(|r| r.get(0) == Some(&Value::Integer(1)) && r.get(2) == Some(&Value::Integer(1)));
        assert!(matched_row.is_some(), "Should have a matched row with id=1");

        // Verify unmatched build row has NULL in probe columns
        let unmatched_row = result
            .rows
            .iter()
            .find(|r| r.get(0) == Some(&Value::Integer(3)));
        assert!(
            unmatched_row.is_some(),
            "Should have unmatched build row with id=3"
        );
        let unmatched = unmatched_row.unwrap();
        assert!(
            unmatched.get(2).map(|v| v.is_null()).unwrap_or(false),
            "Probe col should be NULL"
        );
        assert!(
            unmatched.get(3).map(|v| v.is_null()).unwrap_or(false),
            "Probe col should be NULL"
        );
    }

    /// Test rows_equal function directly
    #[test]
    fn test_rows_equal() {
        let row1 = Row::from_values(vec![Value::Integer(1), Value::Text("a".into())]);
        let row2 = Row::from_values(vec![Value::Integer(1), Value::Text("a".into())]);
        let row3 = Row::from_values(vec![Value::Integer(1), Value::Text("b".into())]);
        let row4 = Row::from_values(vec![Value::Integer(1)]);

        assert!(rows_equal(&row1, &row2), "Identical rows should be equal");
        assert!(
            !rows_equal(&row1, &row3),
            "Different values should not be equal"
        );
        assert!(
            !rows_equal(&row1, &row4),
            "Different lengths should not be equal"
        );
    }

    /// Test rows_equal with NULL values
    #[test]
    fn test_rows_equal_with_nulls() {
        let row_with_null1 = Row::from_values(vec![Value::Integer(1), Value::null_unknown()]);
        let row_with_null2 = Row::from_values(vec![Value::Integer(1), Value::null_unknown()]);
        let row_no_null = Row::from_values(vec![Value::Integer(1), Value::Integer(2)]);

        // NULL == NULL in this context (for deduplication purposes)
        assert!(
            rows_equal(&row_with_null1, &row_with_null2),
            "Rows with same NULL positions should be equal"
        );
        assert!(
            !rows_equal(&row_with_null1, &row_no_null),
            "NULL should not equal non-NULL"
        );
    }
}