laminar_core/operator/

stream_join.rs

//! # Stream-Stream Join Operators
//!
//! Implementation of time-bounded joins between two event streams.
//!
//! Stream-stream joins match events from two streams based on a join key
//! and a time bound. Events are matched if they share a key and their
//! timestamps fall within the specified time window.
//!
//! ## Join Types
//!
//! - **Inner**: Only emit matched pairs
//! - **Left**: Emit all left events, with right match if exists
//! - **Right**: Emit all right events, with left match if exists
//! - **Full**: Emit all events, with matches where they exist
//! - **Left Semi**: Emit left rows that have at least one match (left columns only)
//! - **Left Anti**: Emit left rows with no match within the time bound (left columns only)
//! - **Right Semi/Anti**: Mirror of left variants
//!
//! ## Example
//!
//! ```rust,no_run
//! use laminar_core::operator::stream_join::{
//!     StreamJoinOperator, JoinType, JoinSide, StreamJoinConfig, JoinRowEncoding,
//! };
//! use std::time::Duration;
//!
//! // Basic join (backward compatible)
//! let operator = StreamJoinOperator::new(
//!     "order_id".to_string(),  // left key column
//!     "order_id".to_string(),  // right key column
//!     Duration::from_secs(3600), // 1 hour time bound
//!     JoinType::Inner,
//! );
//!
//! // Optimized join with CPU-friendly encoding
//! let config = StreamJoinConfig::builder()
//!     .left_key_column("order_id")
//!     .right_key_column("order_id")
//!     .time_bound(Duration::from_secs(3600))
//!     .join_type(JoinType::Inner)
//!     .row_encoding(JoinRowEncoding::CpuFriendly)  // 30-50% faster for memory-resident state
//!     .asymmetric_compaction(true)                  // Skip compaction on finished sides
//!     .per_key_tracking(true)                       // Aggressive cleanup for sparse keys
//!     .build()
//!     .unwrap();
//! let optimized_operator = StreamJoinOperator::from_config(config);
//! ```
//!
//! ## SQL Syntax
//!
//! ```sql
//! SELECT o.*, p.status
//! FROM orders o
//! JOIN payments p
//!     ON o.order_id = p.order_id
//!     AND p.ts BETWEEN o.ts AND o.ts + INTERVAL '1' HOUR;
//!
//! -- Session variables for optimization
//! SET streaming_join_row_encoding = 'cpu_friendly';
//! SET streaming_join_asymmetric_compaction = true;
//! ```
//!
//! ## State Management
//!
//! Events are stored in state with keys formatted as:
//! - `sjl:<key_hash>:<timestamp>:<event_id>` for left events
//! - `sjr:<key_hash>:<timestamp>:<event_id>` for right events
//!
//! State is automatically cleaned up when watermark passes
//! `event_timestamp + time_bound`.
//!
//! ## Optimizations
//!
//! - **CPU-Friendly Encoding**: Inlines primitive values for faster access (30-50% improvement)
//! - **Asymmetric Compaction**: Skips compaction on finished/idle sides
//! - **Per-Key Tracking**: Aggressive cleanup for sparse key patterns
//! - **Build-Side Pruning**: Early pruning based on probe-side watermark

use super::{
    Event, Operator, OperatorContext, OperatorError, OperatorState, Output, OutputVec, Timer,
    TimerKey,
};
use crate::state::{StateStore, StateStoreExt};
use arrow_array::{Array, ArrayRef, Float64Array, Int64Array, RecordBatch, StringArray};
use arrow_schema::{DataType, Field, Schema, SchemaRef};
use bytes::Bytes;
use rkyv::{
    rancor::Error as RkyvError, Archive, Deserialize as RkyvDeserialize, Serialize as RkyvSerialize,
};
use rustc_hash::FxHashMap;
use smallvec::SmallVec;
use std::hash::{Hash, Hasher};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;
use std::time::Duration;

/// Join type for stream-stream joins.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum JoinType {
    /// Inner join - only emit matched pairs.
    #[default]
    Inner,
    /// Left outer join - emit all left events, with right match if exists.
    Left,
    /// Right outer join - emit all right events, with left match if exists.
    Right,
    /// Full outer join - emit all events, with matches where they exist.
    Full,
    /// Left semi join - emit left rows that have at least one match (left columns only).
    LeftSemi,
    /// Left anti join - emit left rows that have no match (left columns only).
    LeftAnti,
    /// Right semi join - emit right rows that have at least one match (right columns only).
    RightSemi,
    /// Right anti join - emit right rows that have no match (right columns only).
    RightAnti,
}

impl JoinType {
    /// Returns true if unmatched left events should be emitted.
    #[must_use]
    pub fn emits_unmatched_left(&self) -> bool {
        matches!(self, JoinType::Left | JoinType::Full | JoinType::LeftAnti)
    }

    /// Returns true if unmatched right events should be emitted.
    #[must_use]
    pub fn emits_unmatched_right(&self) -> bool {
        matches!(self, JoinType::Right | JoinType::Full | JoinType::RightAnti)
    }

    /// Returns true if this is a semi join.
    #[must_use]
    pub fn is_semi(&self) -> bool {
        matches!(self, JoinType::LeftSemi | JoinType::RightSemi)
    }

    /// Returns true if this is an anti join.
    #[must_use]
    pub fn is_anti(&self) -> bool {
        matches!(self, JoinType::LeftAnti | JoinType::RightAnti)
    }

    /// For semi/anti joins, returns the side whose rows are kept in the output.
    #[must_use]
    pub fn kept_side(&self) -> Option<JoinSide> {
        match self {
            JoinType::LeftSemi | JoinType::LeftAnti => Some(JoinSide::Left),
            JoinType::RightSemi | JoinType::RightAnti => Some(JoinSide::Right),
            _ => None,
        }
    }
}

/// Identifies which side of the join an event came from.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum JoinSide {
    /// Left side of the join.
    Left,
    /// Right side of the join.
    Right,
}

// Stream Join Optimizations

/// Row encoding strategy for join state.
///
/// Controls how join rows are serialized for storage. The encoding choice
/// affects the tradeoff between memory usage and CPU access speed.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum JoinRowEncoding {
    /// Compact encoding using Arrow IPC format.
    ///
    /// - Smaller memory footprint
    /// - Higher CPU decode cost per access
    /// - Best when: state exceeds memory, disk spills frequent
    #[default]
    Compact,

    /// CPU-friendly encoding with inlined primitive values.
    ///
    /// - Larger memory footprint (~20-40% more)
    /// - Faster access (~30-50% improvement per `RisingWave` benchmarks)
    /// - Best when: state fits in memory, CPU-bound workloads
    CpuFriendly,
}

impl std::fmt::Display for JoinRowEncoding {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::Compact => write!(f, "compact"),
            Self::CpuFriendly => write!(f, "cpu_friendly"),
        }
    }
}

impl std::str::FromStr for JoinRowEncoding {
    type Err = String;

    fn from_str(s: &str) -> Result<Self, Self::Err> {
        match s.to_lowercase().as_str() {
            "compact" => Ok(Self::Compact),
            "cpu_friendly" | "cpufriendly" | "cpu-friendly" => Ok(Self::CpuFriendly),
            _ => Err(format!(
                "Unknown encoding: {s}. Expected 'compact' or 'cpu_friendly'"
            )),
        }
    }
}

/// Configuration for stream-stream joins.
///
/// Provides fine-grained control over join behavior and optimizations.
/// Use the builder pattern for convenient construction.
///
/// # Example
///
/// ```rust
/// use laminar_core::operator::stream_join::{StreamJoinConfig, JoinType, JoinRowEncoding};
/// use std::time::Duration;
///
/// let config = StreamJoinConfig::builder()
///     .left_key_column("order_id")
///     .right_key_column("order_id")
///     .time_bound(Duration::from_secs(3600))
///     .join_type(JoinType::Inner)
///     .row_encoding(JoinRowEncoding::CpuFriendly)
///     .build()
///     .unwrap();
/// ```
#[derive(Debug, Clone)]
pub struct StreamJoinConfig {
    /// Left stream key column name.
    pub left_key_column: String,
    /// Right stream key column name.
    pub right_key_column: String,
    /// Time bound for matching (milliseconds).
    pub time_bound_ms: i64,
    /// Type of join to perform.
    pub join_type: JoinType,
    /// Operator ID for checkpointing.
    pub operator_id: Option<String>,

    // Optimizations
    /// Row encoding strategy.
    pub row_encoding: JoinRowEncoding,
    /// Enable asymmetric compaction optimization.
    pub asymmetric_compaction: bool,
    /// Threshold for considering a side "finished" (ms).
    pub idle_threshold_ms: i64,
    /// Enable per-key cleanup tracking.
    pub per_key_tracking: bool,
    /// Threshold for idle key cleanup (ms).
    pub key_idle_threshold_ms: i64,
    /// Enable build-side pruning.
    pub build_side_pruning: bool,
    /// Which side to use as build side (None = auto-select based on statistics).
    pub build_side: Option<JoinSide>,
}

impl Default for StreamJoinConfig {
    fn default() -> Self {
        Self {
            left_key_column: String::new(),
            right_key_column: String::new(),
            time_bound_ms: 0,
            join_type: JoinType::Inner,
            operator_id: None,
            row_encoding: JoinRowEncoding::Compact,
            asymmetric_compaction: true,
            idle_threshold_ms: 60_000, // 1 minute
            per_key_tracking: true,
            key_idle_threshold_ms: 300_000, // 5 minutes
            build_side_pruning: true,
            build_side: None,
        }
    }
}

impl StreamJoinConfig {
    /// Creates a new configuration builder.
    #[must_use]
    pub fn builder() -> StreamJoinConfigBuilder {
        StreamJoinConfigBuilder::default()
    }
}

/// Builder for `StreamJoinConfig`.
#[derive(Debug, Default)]
pub struct StreamJoinConfigBuilder {
    config: StreamJoinConfig,
}

impl StreamJoinConfigBuilder {
    /// Sets the left stream key column name.
    #[must_use]
    pub fn left_key_column(mut self, column: impl Into<String>) -> Self {
        self.config.left_key_column = column.into();
        self
    }

    /// Sets the right stream key column name.
    #[must_use]
    pub fn right_key_column(mut self, column: impl Into<String>) -> Self {
        self.config.right_key_column = column.into();
        self
    }

    /// Sets the time bound for matching events.
    #[must_use]
    #[allow(clippy::cast_possible_truncation)] // Duration.as_millis() fits i64 for practical values
    pub fn time_bound(mut self, duration: Duration) -> Self {
        self.config.time_bound_ms = duration.as_millis() as i64;
        self
    }

    /// Sets the time bound in milliseconds.
    #[must_use]
    pub fn time_bound_ms(mut self, ms: i64) -> Self {
        self.config.time_bound_ms = ms;
        self
    }

    /// Sets the join type.
    #[must_use]
    pub fn join_type(mut self, join_type: JoinType) -> Self {
        self.config.join_type = join_type;
        self
    }

    /// Sets the operator ID for checkpointing.
    #[must_use]
    pub fn operator_id(mut self, id: impl Into<String>) -> Self {
        self.config.operator_id = Some(id.into());
        self
    }

    /// Sets the row encoding strategy.
    #[must_use]
    pub fn row_encoding(mut self, encoding: JoinRowEncoding) -> Self {
        self.config.row_encoding = encoding;
        self
    }

    /// Enables or disables asymmetric compaction.
    #[must_use]
    pub fn asymmetric_compaction(mut self, enabled: bool) -> Self {
        self.config.asymmetric_compaction = enabled;
        self
    }

    /// Sets the idle threshold for asymmetric compaction.
    #[must_use]
    #[allow(clippy::cast_possible_truncation)] // Duration.as_millis() fits i64 for practical values
    pub fn idle_threshold(mut self, duration: Duration) -> Self {
        self.config.idle_threshold_ms = duration.as_millis() as i64;
        self
    }

    /// Enables or disables per-key tracking.
    #[must_use]
    pub fn per_key_tracking(mut self, enabled: bool) -> Self {
        self.config.per_key_tracking = enabled;
        self
    }

    /// Sets the key idle threshold for cleanup.
    #[must_use]
    #[allow(clippy::cast_possible_truncation)] // Duration.as_millis() fits i64 for practical values
    pub fn key_idle_threshold(mut self, duration: Duration) -> Self {
        self.config.key_idle_threshold_ms = duration.as_millis() as i64;
        self
    }

    /// Enables or disables build-side pruning.
    #[must_use]
    pub fn build_side_pruning(mut self, enabled: bool) -> Self {
        self.config.build_side_pruning = enabled;
        self
    }

    /// Sets which side to use as the build side.
    #[must_use]
    pub fn build_side(mut self, side: JoinSide) -> Self {
        self.config.build_side = Some(side);
        self
    }

    /// Builds the configuration.
    ///
    /// # Errors
    ///
    /// Returns `OperatorError::ConfigError` if key columns are empty.
    pub fn build(self) -> std::result::Result<StreamJoinConfig, OperatorError> {
        if self.config.left_key_column.is_empty() {
            return Err(OperatorError::ConfigError(
                "left_key_column is required".into(),
            ));
        }
        if self.config.right_key_column.is_empty() {
            return Err(OperatorError::ConfigError(
                "right_key_column is required".into(),
            ));
        }
        Ok(self.config)
    }
}

/// Per-side statistics for asymmetric optimization.
#[derive(Debug, Clone, Default)]
pub struct SideStats {
    /// Total events received on this side.
    pub events_received: u64,
    /// Events in current tracking window.
    pub events_this_window: u64,
    /// Last event timestamp (processing time).
    pub last_event_time: i64,
    /// Estimated write rate (events/second).
    pub write_rate: f64,
    /// Window start time for rate calculation.
    window_start: i64,
}

impl SideStats {
    /// Creates new side statistics.
    #[must_use]
    pub fn new() -> Self {
        Self::default()
    }

    /// Records an event arrival.
    #[allow(clippy::cast_precision_loss)]
    pub fn record_event(&mut self, processing_time: i64) {
        self.events_received += 1;
        self.events_this_window += 1;
        self.last_event_time = processing_time;

        // Update write rate every 1000ms
        if self.window_start == 0 {
            self.window_start = processing_time;
        } else {
            let elapsed_ms = processing_time - self.window_start;
            if elapsed_ms >= 1000 {
                // Precision loss is acceptable for rate estimation
                self.write_rate = (self.events_this_window as f64 * 1000.0) / elapsed_ms as f64;
                self.events_this_window = 0;
                self.window_start = processing_time;
            }
        }
    }

    /// Checks if this side is considered "finished" (no recent activity).
    #[must_use]
    pub fn is_idle(&self, current_time: i64, threshold_ms: i64) -> bool {
        if self.events_received == 0 {
            return false; // Never received events, not idle
        }
        let time_since_last = current_time - self.last_event_time;
        time_since_last > threshold_ms && self.events_this_window == 0
    }
}

/// Per-key metadata for cleanup tracking.
#[derive(Debug, Clone)]
pub struct KeyMetadata {
    /// Last event timestamp for this key (processing time).
    pub last_activity: i64,
    /// Number of events for this key.
    pub event_count: u64,
    /// Number of state entries for this key.
    pub state_entries: u64,
}

impl KeyMetadata {
    /// Creates new key metadata.
    #[must_use]
    pub fn new(processing_time: i64) -> Self {
        Self {
            last_activity: processing_time,
            event_count: 1,
            state_entries: 1,
        }
    }

    /// Records an event for this key.
    pub fn record_event(&mut self, processing_time: i64) {
        self.last_activity = processing_time;
        self.event_count += 1;
        self.state_entries += 1;
    }

    /// Decrements state entry count (called on cleanup).
    pub fn decrement_entries(&mut self) {
        self.state_entries = self.state_entries.saturating_sub(1);
    }

    /// Checks if this key is idle.
    #[must_use]
    pub fn is_idle(&self, current_time: i64, threshold_ms: i64) -> bool {
        current_time - self.last_activity > threshold_ms
    }
}

/// State key prefixes for join state.
const LEFT_STATE_PREFIX: &[u8; 4] = b"sjl:";
const RIGHT_STATE_PREFIX: &[u8; 4] = b"sjr:";

/// Timer key prefix for left-side cleanup.
const LEFT_TIMER_PREFIX: u8 = 0x10;
/// Timer key prefix for right-side cleanup.
const RIGHT_TIMER_PREFIX: u8 = 0x20;
/// Timer key prefix for unmatched event emission.
const UNMATCHED_TIMER_PREFIX: u8 = 0x30;
/// Prefix for matched-flag keys (separate from `JoinRow` to avoid serde roundtrips).
const MATCHED_FLAG_PREFIX: u8 = 0x40;

/// Build a matched-flag key from a 28-byte state key.
#[inline]
fn matched_flag_key(state_key: &[u8]) -> [u8; 29] {
    debug_assert_eq!(state_key.len(), 28, "join state keys are always 28 bytes");
    let mut k = [0u8; 29];
    k[0] = MATCHED_FLAG_PREFIX;
    k[1..].copy_from_slice(&state_key[..28]);
    k
}

/// Mark a join row as matched by writing a 1-byte flag.
#[inline]
fn mark_matched(state: &mut dyn StateStore, state_key: &[u8]) {
    let k = matched_flag_key(state_key);
    let _ = state.put(&k, Bytes::from_static(&[1]));
}

/// Check if a join row has been matched.
#[inline]
fn is_matched(state: &dyn StateStore, state_key: &[u8]) -> bool {
    let k = matched_flag_key(state_key);
    state.get_ref(&k).is_some()
}

/// Static counter for generating unique operator IDs.
static JOIN_OPERATOR_COUNTER: AtomicU64 = AtomicU64::new(0);

/// Static counter for generating unique event IDs within an operator.
static EVENT_ID_COUNTER: AtomicU64 = AtomicU64::new(0);

/// A stored join row containing serialized event data.
#[derive(Debug, Clone, Archive, RkyvSerialize, RkyvDeserialize)]
pub struct JoinRow {
    /// Event timestamp in milliseconds.
    pub timestamp: i64,
    /// Serialized key value (as bytes).
    pub key_value: Vec<u8>,
    /// Serialized record batch data.
    pub data: Vec<u8>,
    /// Whether this row has been matched (for outer joins).
    pub matched: bool,
    /// Encoding used for serialization.
    /// 0 = Compact (Arrow IPC), 1 = `CpuFriendly`
    encoding: u8,
}

/// Magic bytes for CPU-friendly encoding format.
const CPU_FRIENDLY_MAGIC: [u8; 4] = *b"CPUF";

/// Type tag for CPU-friendly encoding.
#[repr(u8)]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum CpuFriendlyType {
    Null = 0,
    Int64 = 1,
    Float64 = 2,
    Utf8 = 3,
}

impl JoinRow {
    /// Creates a new join row from an event and extracted key.
    /// Uses compact encoding by default.
    #[cfg(test)]
    fn new(timestamp: i64, key_value: Vec<u8>, batch: &RecordBatch) -> Result<Self, OperatorError> {
        Self::with_encoding(timestamp, key_value, batch, JoinRowEncoding::Compact)
    }

    /// Creates a new join row with specified encoding.
    fn with_encoding(
        timestamp: i64,
        key_value: Vec<u8>,
        batch: &RecordBatch,
        encoding: JoinRowEncoding,
    ) -> Result<Self, OperatorError> {
        let (data, encoding_byte) = match encoding {
            JoinRowEncoding::Compact => (Self::serialize_compact(batch)?, 0),
            JoinRowEncoding::CpuFriendly => (Self::serialize_cpu_friendly(batch)?, 1),
        };
        Ok(Self {
            timestamp,
            key_value,
            data,
            matched: false,
            encoding: encoding_byte,
        })
    }

    fn serialize_compact(batch: &RecordBatch) -> Result<Vec<u8>, OperatorError> {
        Ok(crate::serialization::serialize_batch_stream(batch)?)
    }

    /// Serializes using CPU-friendly format.
    ///
    /// Format:
    /// - Magic (4 bytes): "CPUF"
    /// - Num columns (2 bytes): u16
    /// - Num rows (4 bytes): u32
    /// - For each column:
    ///   - Name length (2 bytes): u16
    ///   - Name bytes (variable)
    ///   - Type tag (1 byte)
    ///   - Nullable (1 byte): 0 or 1
    ///   - Data (depends on type):
    ///     - Int64: validity bitmap + raw i64 values
    ///     - Float64: validity bitmap + raw f64 values
    ///     - Utf8: validity bitmap + offsets (u32) + data bytes
    #[allow(clippy::cast_possible_truncation)] // Wire format uses u32 for row/column counts and offsets
    fn serialize_cpu_friendly(batch: &RecordBatch) -> Result<Vec<u8>, OperatorError> {
        let schema = batch.schema();
        let num_rows = batch.num_rows();
        let num_cols = batch.num_columns();

        // Estimate capacity
        let mut buf = Vec::with_capacity(4 + 2 + 4 + num_cols * 64 + num_rows * num_cols * 8);

        // Header
        buf.extend_from_slice(&CPU_FRIENDLY_MAGIC);
        buf.extend_from_slice(&(num_cols as u16).to_le_bytes());
        buf.extend_from_slice(&(num_rows as u32).to_le_bytes());

        // Columns
        for (i, field) in schema.fields().iter().enumerate() {
            let column = batch.column(i);

            // Column name
            let name_bytes = field.name().as_bytes();
            buf.extend_from_slice(&(name_bytes.len() as u16).to_le_bytes());
            buf.extend_from_slice(name_bytes);

            // Nullable flag
            buf.push(u8::from(field.is_nullable()));

            // Type and data
            match field.data_type() {
                DataType::Int64 => {
                    buf.push(CpuFriendlyType::Int64 as u8);
                    Self::write_int64_column(&mut buf, column, num_rows)?;
                }
                DataType::Float64 => {
                    buf.push(CpuFriendlyType::Float64 as u8);
                    Self::write_float64_column(&mut buf, column, num_rows)?;
                }
                DataType::Utf8 => {
                    buf.push(CpuFriendlyType::Utf8 as u8);
                    Self::write_utf8_column(&mut buf, column, num_rows)?;
                }
                _ => {
                    // Fallback: encode as null for unsupported types
                    buf.push(CpuFriendlyType::Null as u8);
                }
            }
        }

        Ok(buf)
    }

    /// Writes an Int64 column in CPU-friendly format.
    fn write_int64_column(
        buf: &mut Vec<u8>,
        column: &ArrayRef,
        num_rows: usize,
    ) -> Result<(), OperatorError> {
        let arr = column
            .as_any()
            .downcast_ref::<Int64Array>()
            .ok_or_else(|| OperatorError::SerializationFailed("Expected Int64Array".into()))?;

        // Validity bitmap (1 bit per row, padded to bytes)
        let validity_bytes = num_rows.div_ceil(8);
        if let Some(nulls) = arr.nulls() {
            // Copy validity buffer
            let buffer = nulls.buffer();
            let slice = &buffer.as_slice()[..validity_bytes.min(buffer.len())];
            buf.extend_from_slice(slice);
            // Pad if needed
            for _ in slice.len()..validity_bytes {
                buf.push(0xFF);
            }
        } else {
            // All valid
            buf.extend(std::iter::repeat_n(0xFF, validity_bytes));
        }

        // Raw values (8 bytes each)
        // SAFETY: Converting i64 slice to bytes for zero-copy serialization.
        // The pointer cast is safe because we're reinterpreting the same memory.
        let values = arr.values();
        let value_bytes =
            unsafe { std::slice::from_raw_parts(values.as_ptr().cast::<u8>(), values.len() * 8) };
        buf.extend_from_slice(value_bytes);

        Ok(())
    }

    /// Writes a Float64 column in CPU-friendly format.
    fn write_float64_column(
        buf: &mut Vec<u8>,
        column: &ArrayRef,
        num_rows: usize,
    ) -> Result<(), OperatorError> {
        let arr = column
            .as_any()
            .downcast_ref::<Float64Array>()
            .ok_or_else(|| OperatorError::SerializationFailed("Expected Float64Array".into()))?;

        // Validity bitmap
        let validity_bytes = num_rows.div_ceil(8);
        if let Some(nulls) = arr.nulls() {
            let buffer = nulls.buffer();
            let slice = &buffer.as_slice()[..validity_bytes.min(buffer.len())];
            buf.extend_from_slice(slice);
            for _ in slice.len()..validity_bytes {
                buf.push(0xFF);
            }
        } else {
            buf.extend(std::iter::repeat_n(0xFF, validity_bytes));
        }

        // Raw values (8 bytes each)
        // SAFETY: Converting f64 slice to bytes for zero-copy serialization.
        let values = arr.values();
        let value_bytes =
            unsafe { std::slice::from_raw_parts(values.as_ptr().cast::<u8>(), values.len() * 8) };
        buf.extend_from_slice(value_bytes);

        Ok(())
    }

    /// Writes a Utf8 column in CPU-friendly format.
    #[allow(clippy::cast_sign_loss)]
    fn write_utf8_column(
        buf: &mut Vec<u8>,
        column: &ArrayRef,
        num_rows: usize,
    ) -> Result<(), OperatorError> {
        let arr = column
            .as_any()
            .downcast_ref::<StringArray>()
            .ok_or_else(|| OperatorError::SerializationFailed("Expected StringArray".into()))?;

        // Validity bitmap
        let validity_bytes = num_rows.div_ceil(8);
        if let Some(nulls) = arr.nulls() {
            let buffer = nulls.buffer();
            let slice = &buffer.as_slice()[..validity_bytes.min(buffer.len())];
            buf.extend_from_slice(slice);
            for _ in slice.len()..validity_bytes {
                buf.push(0xFF);
            }
        } else {
            buf.extend(std::iter::repeat_n(0xFF, validity_bytes));
        }

        // Offsets (u32 for each row + 1)
        // Note: Arrow string offsets are always non-negative
        let offsets = arr.offsets();
        for offset in offsets.iter() {
            buf.extend_from_slice(&(*offset as u32).to_le_bytes());
        }

        // String data
        let values = arr.values();
        buf.extend_from_slice(values.as_slice());

        Ok(())
    }

    /// Deserializes a record batch from bytes.
    fn deserialize_batch(data: &[u8], encoding: u8) -> Result<RecordBatch, OperatorError> {
        if encoding == 1 && data.starts_with(&CPU_FRIENDLY_MAGIC) {
            Self::deserialize_cpu_friendly(data)
        } else {
            Self::deserialize_compact(data)
        }
    }

    fn deserialize_compact(data: &[u8]) -> Result<RecordBatch, OperatorError> {
        Ok(crate::serialization::deserialize_batch_stream(data)?)
    }

    /// Deserializes from CPU-friendly format.
    fn deserialize_cpu_friendly(data: &[u8]) -> Result<RecordBatch, OperatorError> {
        if data.len() < 10 {
            return Err(OperatorError::SerializationFailed(
                "Buffer too short".into(),
            ));
        }

        // Parse header
        let num_cols = u16::from_le_bytes([data[4], data[5]]) as usize;
        let num_rows = u32::from_le_bytes([data[6], data[7], data[8], data[9]]) as usize;

        let mut offset = 10;
        let mut fields = Vec::with_capacity(num_cols);
        let mut columns: Vec<ArrayRef> = Vec::with_capacity(num_cols);

        for _ in 0..num_cols {
            if offset + 2 > data.len() {
                return Err(OperatorError::SerializationFailed(
                    "Truncated column header".into(),
                ));
            }

            // Read column name
            let name_len = u16::from_le_bytes([data[offset], data[offset + 1]]) as usize;
            offset += 2;

            if offset + name_len > data.len() {
                return Err(OperatorError::SerializationFailed(
                    "Truncated column name".into(),
                ));
            }
            let name = String::from_utf8_lossy(&data[offset..offset + name_len]).to_string();
            offset += name_len;

            if offset + 2 > data.len() {
                return Err(OperatorError::SerializationFailed(
                    "Truncated type info".into(),
                ));
            }

            // Read nullable and type
            let nullable = data[offset] != 0;
            offset += 1;
            let type_tag = data[offset];
            offset += 1;

            // Read column data based on type
            let validity_bytes = num_rows.div_ceil(8);

            match type_tag {
                t if t == CpuFriendlyType::Int64 as u8 => {
                    let (arr, new_offset) =
                        Self::read_int64_column(data, offset, num_rows, validity_bytes)?;
                    offset = new_offset;
                    fields.push(Field::new(&name, DataType::Int64, nullable));
                    columns.push(Arc::new(arr));
                }
                t if t == CpuFriendlyType::Float64 as u8 => {
                    let (arr, new_offset) =
                        Self::read_float64_column(data, offset, num_rows, validity_bytes)?;
                    offset = new_offset;
                    fields.push(Field::new(&name, DataType::Float64, nullable));
                    columns.push(Arc::new(arr));
                }
                t if t == CpuFriendlyType::Utf8 as u8 => {
                    let (arr, new_offset) =
                        Self::read_utf8_column(data, offset, num_rows, validity_bytes)?;
                    offset = new_offset;
                    fields.push(Field::new(&name, DataType::Utf8, nullable));
                    columns.push(Arc::new(arr));
                }
                _ => {
                    // Null/unsupported type - create null array
                    fields.push(Field::new(&name, DataType::Int64, true));
                    columns.push(Arc::new(Int64Array::from(vec![None; num_rows])));
                }
            }
        }

        let schema = Arc::new(Schema::new(fields));
        RecordBatch::try_new(schema, columns)
            .map_err(|e| OperatorError::SerializationFailed(e.to_string()))
    }

    /// Reads an Int64 column from CPU-friendly format.
    fn read_int64_column(
        data: &[u8],
        offset: usize,
        num_rows: usize,
        validity_bytes: usize,
    ) -> Result<(Int64Array, usize), OperatorError> {
        let mut pos = offset;

        // Skip validity bitmap (we don't reconstruct it for simplicity)
        if pos + validity_bytes > data.len() {
            return Err(OperatorError::SerializationFailed(
                "Truncated validity".into(),
            ));
        }
        pos += validity_bytes;

        // Read values
        let values_bytes = num_rows * 8;
        if pos + values_bytes > data.len() {
            return Err(OperatorError::SerializationFailed(
                "Truncated int64 values".into(),
            ));
        }

        let mut values = Vec::with_capacity(num_rows);
        for i in 0..num_rows {
            let start = pos + i * 8;
            let bytes = [
                data[start],
                data[start + 1],
                data[start + 2],
                data[start + 3],
                data[start + 4],
                data[start + 5],
                data[start + 6],
                data[start + 7],
            ];
            values.push(i64::from_le_bytes(bytes));
        }
        pos += values_bytes;

        Ok((Int64Array::from(values), pos))
    }

    /// Reads a Float64 column from CPU-friendly format.
    fn read_float64_column(
        data: &[u8],
        offset: usize,
        num_rows: usize,
        validity_bytes: usize,
    ) -> Result<(Float64Array, usize), OperatorError> {
        let mut pos = offset;

        // Skip validity bitmap
        if pos + validity_bytes > data.len() {
            return Err(OperatorError::SerializationFailed(
                "Truncated validity".into(),
            ));
        }
        pos += validity_bytes;

        // Read values
        let values_bytes = num_rows * 8;
        if pos + values_bytes > data.len() {
            return Err(OperatorError::SerializationFailed(
                "Truncated float64 values".into(),
            ));
        }

        let mut values = Vec::with_capacity(num_rows);
        for i in 0..num_rows {
            let start = pos + i * 8;
            let bytes = [
                data[start],
                data[start + 1],
                data[start + 2],
                data[start + 3],
                data[start + 4],
                data[start + 5],
                data[start + 6],
                data[start + 7],
            ];
            values.push(f64::from_le_bytes(bytes));
        }
        pos += values_bytes;

        Ok((Float64Array::from(values), pos))
    }

    /// Reads a Utf8 column from CPU-friendly format.
    fn read_utf8_column(
        data: &[u8],
        offset: usize,
        num_rows: usize,
        validity_bytes: usize,
    ) -> Result<(StringArray, usize), OperatorError> {
        let mut pos = offset;

        // Skip validity bitmap
        if pos + validity_bytes > data.len() {
            return Err(OperatorError::SerializationFailed(
                "Truncated validity".into(),
            ));
        }
        pos += validity_bytes;

        // Read offsets
        let offsets_bytes = (num_rows + 1) * 4;
        if pos + offsets_bytes > data.len() {
            return Err(OperatorError::SerializationFailed(
                "Truncated offsets".into(),
            ));
        }

        let mut offsets = Vec::with_capacity(num_rows + 1);
        for i in 0..=num_rows {
            let start = pos + i * 4;
            let bytes = [
                data[start],
                data[start + 1],
                data[start + 2],
                data[start + 3],
            ];
            offsets.push(u32::from_le_bytes(bytes) as usize);
        }
        pos += offsets_bytes;

        // Calculate data length and read string data
        let data_len = offsets.last().copied().unwrap_or(0);
        if pos + data_len > data.len() {
            return Err(OperatorError::SerializationFailed(
                "Truncated string data".into(),
            ));
        }

        let string_data = &data[pos..pos + data_len];
        pos += data_len;

        // Build strings
        let mut strings = Vec::with_capacity(num_rows);
        for i in 0..num_rows {
            let start = offsets[i];
            let end = offsets[i + 1];
            let s = String::from_utf8_lossy(&string_data[start..end]).to_string();
            strings.push(s);
        }

        Ok((StringArray::from(strings), pos))
    }

    /// Converts this join row back to a record batch.
    ///
    /// # Errors
    ///
    /// Returns `OperatorError::SerializationFailed` if the batch data is invalid.
    pub fn to_batch(&self) -> Result<RecordBatch, OperatorError> {
        Self::deserialize_batch(&self.data, self.encoding)
    }

    /// Returns the encoding used for this row.
    #[must_use]
    pub fn encoding(&self) -> JoinRowEncoding {
        if self.encoding == 1 {
            JoinRowEncoding::CpuFriendly
        } else {
            JoinRowEncoding::Compact
        }
    }
}

/// Metrics for tracking join operations.
#[derive(Debug, Clone, Default)]
pub struct JoinMetrics {
    /// Number of left events processed.
    pub left_events: u64,
    /// Number of right events processed.
    pub right_events: u64,
    /// Number of join matches produced.
    pub matches: u64,
    /// Number of unmatched left events emitted (left/full joins).
    pub unmatched_left: u64,
    /// Number of unmatched right events emitted (right/full joins).
    pub unmatched_right: u64,
    /// Number of late events dropped.
    pub late_events: u64,
    /// Number of state entries cleaned up.
    pub state_cleanups: u64,

    // Optimization Metrics
    /// Rows encoded with CPU-friendly format.
    pub cpu_friendly_encodes: u64,
    /// Rows encoded with compact format.
    pub compact_encodes: u64,
    /// Compactions skipped due to asymmetric optimization.
    pub asymmetric_skips: u64,
    /// Idle keys cleaned up.
    pub idle_key_cleanups: u64,
    /// Build-side entries pruned early.
    pub build_side_prunes: u64,
    /// Current number of tracked keys (for per-key tracking).
    pub tracked_keys: u64,
}

impl JoinMetrics {
    /// Creates new metrics.
    #[must_use]
    pub fn new() -> Self {
        Self::default()
    }

    /// Resets all counters.
    pub fn reset(&mut self) {
        *self = Self::default();
    }
}

/// Stream-stream join operator.
///
/// Joins events from two streams based on a key column and time bound.
/// Events are matched if they share a key value and their timestamps
/// are within the specified time window.
///
/// # State Management
///
/// Events from both sides are stored in state until they can no longer
/// produce matches (watermark passes `timestamp + time_bound`). State
/// is automatically cleaned up via timers.
///
/// # Performance Considerations
///
/// - State grows linearly with the number of events within the time window
/// - For high-cardinality joins, consider using shorter time bounds
/// - Inner joins use less state than outer joins (no unmatched tracking)
///
/// # Optimizations
///
/// - **CPU-Friendly Encoding**: Use `JoinRowEncoding::CpuFriendly` for 30-50% faster access
/// - **Asymmetric Compaction**: Automatically skips compaction on finished/idle sides
/// - **Per-Key Tracking**: Aggressive cleanup for sparse key patterns
/// - **Build-Side Pruning**: Early pruning based on probe-side watermark progress
pub struct StreamJoinOperator {
    /// Left stream key column name.
    left_key_column: String,
    /// Right stream key column name.
    right_key_column: String,
    /// Time bound for matching (events match if within this duration).
    time_bound_ms: i64,
    /// Type of join to perform.
    join_type: JoinType,
    /// Operator ID for checkpointing.
    operator_id: String,
    /// Metrics for monitoring.
    metrics: JoinMetrics,
    /// Output schema (lazily initialized).
    output_schema: Option<SchemaRef>,
    /// Left schema (captured from first left event).
    left_schema: Option<SchemaRef>,
    /// Right schema (captured from first right event).
    right_schema: Option<SchemaRef>,

    // Optimization Fields
    /// Row encoding strategy.
    row_encoding: JoinRowEncoding,
    /// Enable asymmetric compaction.
    asymmetric_compaction: bool,
    /// Idle threshold for asymmetric compaction (ms).
    idle_threshold_ms: i64,
    /// Enable per-key tracking.
    per_key_tracking: bool,
    /// Key idle threshold (ms).
    key_idle_threshold_ms: i64,
    /// Enable build-side pruning.
    build_side_pruning: bool,
    /// Configured build side.
    build_side: Option<JoinSide>,
    /// Left-side statistics.
    left_stats: SideStats,
    /// Right-side statistics.
    right_stats: SideStats,
    /// Per-key metadata (`key_hash` -> metadata).
    key_metadata: FxHashMap<u64, KeyMetadata>,
    /// Left-side watermark.
    left_watermark: i64,
    /// Right-side watermark.
    right_watermark: i64,
    /// Reusable buffer for `prune_build_side` to avoid per-call allocation.
    prune_buffer: Vec<Bytes>,
    /// Cached column index for left key — resolved on first left event.
    left_key_index: Option<usize>,
    /// Cached column index for right key — resolved on first right event.
    right_key_index: Option<usize>,
}

impl StreamJoinOperator {
    /// Creates a new stream join operator.
    ///
    /// # Arguments
    ///
    /// * `left_key_column` - Name of the key column in left stream events
    /// * `right_key_column` - Name of the key column in right stream events
    /// * `time_bound` - Maximum time difference for matching events
    /// * `join_type` - Type of join to perform
    #[must_use]
    #[allow(clippy::cast_possible_truncation)] // Duration.as_millis() fits i64 for practical values
    pub fn new(
        left_key_column: String,
        right_key_column: String,
        time_bound: Duration,
        join_type: JoinType,
    ) -> Self {
        let operator_num = JOIN_OPERATOR_COUNTER.fetch_add(1, Ordering::Relaxed);
        Self {
            left_key_column,
            right_key_column,
            time_bound_ms: time_bound.as_millis() as i64,
            join_type,
            operator_id: format!("stream_join_{operator_num}"),
            metrics: JoinMetrics::new(),
            output_schema: None,
            left_schema: None,
            right_schema: None,
            // Default optimizations
            row_encoding: JoinRowEncoding::Compact,
            asymmetric_compaction: true,
            idle_threshold_ms: 60_000,
            per_key_tracking: true,
            key_idle_threshold_ms: 300_000,
            build_side_pruning: true,
            build_side: None,
            left_stats: SideStats::new(),
            right_stats: SideStats::new(),
            key_metadata: FxHashMap::default(),
            left_watermark: i64::MIN,
            right_watermark: i64::MIN,
            prune_buffer: Vec::with_capacity(100),
            left_key_index: None,
            right_key_index: None,
        }
    }

    /// Creates a new stream join operator with a custom operator ID.
    #[must_use]
    #[allow(clippy::cast_possible_truncation)] // Duration.as_millis() fits i64 for practical values
    pub fn with_id(
        left_key_column: String,
        right_key_column: String,
        time_bound: Duration,
        join_type: JoinType,
        operator_id: String,
    ) -> Self {
        Self {
            left_key_column,
            right_key_column,
            time_bound_ms: time_bound.as_millis() as i64,
            join_type,
            operator_id,
            metrics: JoinMetrics::new(),
            output_schema: None,
            left_schema: None,
            right_schema: None,
            // Default optimizations
            row_encoding: JoinRowEncoding::Compact,
            asymmetric_compaction: true,
            idle_threshold_ms: 60_000,
            per_key_tracking: true,
            key_idle_threshold_ms: 300_000,
            build_side_pruning: true,
            build_side: None,
            left_stats: SideStats::new(),
            right_stats: SideStats::new(),
            key_metadata: FxHashMap::default(),
            left_watermark: i64::MIN,
            right_watermark: i64::MIN,
            prune_buffer: Vec::with_capacity(100),
            left_key_index: None,
            right_key_index: None,
        }
    }

    /// Creates a new stream join operator from configuration.
    ///
    /// This is the recommended constructor for production use, allowing
    /// fine-grained control over optimization settings.
    ///
    /// # Example
    ///
    /// ```rust
    /// use laminar_core::operator::stream_join::{
    ///     StreamJoinOperator, StreamJoinConfig, JoinType, JoinRowEncoding,
    /// };
    /// use std::time::Duration;
    ///
    /// let config = StreamJoinConfig::builder()
    ///     .left_key_column("order_id")
    ///     .right_key_column("order_id")
    ///     .time_bound(Duration::from_secs(3600))
    ///     .join_type(JoinType::Inner)
    ///     .row_encoding(JoinRowEncoding::CpuFriendly)
    ///     .build()
    ///     .unwrap();
    ///
    /// let operator = StreamJoinOperator::from_config(config);
    /// ```
    #[must_use]
    pub fn from_config(config: StreamJoinConfig) -> Self {
        let operator_id = config.operator_id.unwrap_or_else(|| {
            let num = JOIN_OPERATOR_COUNTER.fetch_add(1, Ordering::Relaxed);
            format!("stream_join_{num}")
        });

        Self {
            left_key_column: config.left_key_column,
            right_key_column: config.right_key_column,
            time_bound_ms: config.time_bound_ms,
            join_type: config.join_type,
            operator_id,
            metrics: JoinMetrics::new(),
            output_schema: None,
            left_schema: None,
            right_schema: None,
            row_encoding: config.row_encoding,
            asymmetric_compaction: config.asymmetric_compaction,
            idle_threshold_ms: config.idle_threshold_ms,
            per_key_tracking: config.per_key_tracking,
            key_idle_threshold_ms: config.key_idle_threshold_ms,
            build_side_pruning: config.build_side_pruning,
            build_side: config.build_side,
            left_stats: SideStats::new(),
            right_stats: SideStats::new(),
            key_metadata: FxHashMap::default(),
            left_watermark: i64::MIN,
            right_watermark: i64::MIN,
            prune_buffer: Vec::with_capacity(100),
            left_key_index: None,
            right_key_index: None,
        }
    }

    /// Returns the join type.
    #[must_use]
    pub fn join_type(&self) -> JoinType {
        self.join_type
    }

    /// Returns the time bound in milliseconds.
    #[must_use]
    pub fn time_bound_ms(&self) -> i64 {
        self.time_bound_ms
    }

    /// Returns the metrics.
    #[must_use]
    pub fn metrics(&self) -> &JoinMetrics {
        &self.metrics
    }

    /// Resets the metrics.
    pub fn reset_metrics(&mut self) {
        self.metrics.reset();
    }

    /// Returns the row encoding strategy.
    #[must_use]
    pub fn row_encoding(&self) -> JoinRowEncoding {
        self.row_encoding
    }

    /// Returns whether asymmetric compaction is enabled.
    #[must_use]
    pub fn asymmetric_compaction_enabled(&self) -> bool {
        self.asymmetric_compaction
    }

    /// Returns whether per-key tracking is enabled.
    #[must_use]
    pub fn per_key_tracking_enabled(&self) -> bool {
        self.per_key_tracking
    }

    /// Returns the left-side statistics.
    #[must_use]
    pub fn left_stats(&self) -> &SideStats {
        &self.left_stats
    }

    /// Returns the right-side statistics.
    #[must_use]
    pub fn right_stats(&self) -> &SideStats {
        &self.right_stats
    }

    /// Returns the number of tracked keys.
    #[must_use]
    pub fn tracked_key_count(&self) -> usize {
        self.key_metadata.len()
    }

    /// Checks if a side is considered "finished" (idle).
    #[must_use]
    pub fn is_side_idle(&self, side: JoinSide, current_time: i64) -> bool {
        match side {
            JoinSide::Left => self
                .left_stats
                .is_idle(current_time, self.idle_threshold_ms),
            JoinSide::Right => self
                .right_stats
                .is_idle(current_time, self.idle_threshold_ms),
        }
    }

    /// Determines the effective build side based on configuration or heuristics.
    #[must_use]
    pub fn effective_build_side(&self) -> JoinSide {
        // Use configured build side if set
        if let Some(side) = self.build_side {
            return side;
        }

        // Auto-select based on statistics: smaller side is typically better as build
        if self.left_stats.events_received < self.right_stats.events_received {
            JoinSide::Left
        } else {
            JoinSide::Right
        }
    }

    /// Processes an event from either the left or right side.
    ///
    /// This is the main entry point for the join operator. Call this with
    /// the appropriate `JoinSide` to indicate which stream the event came from.
    pub fn process_side(
        &mut self,
        event: &Event,
        side: JoinSide,
        ctx: &mut OperatorContext,
    ) -> OutputVec {
        match side {
            JoinSide::Left => self.process_left(event, ctx),
            JoinSide::Right => self.process_right(event, ctx),
        }
    }

    /// Processes a left-side event.
    fn process_left(&mut self, event: &Event, ctx: &mut OperatorContext) -> OutputVec {
        self.metrics.left_events += 1;

        // Track side statistics for asymmetric compaction
        self.left_stats.record_event(ctx.processing_time);

        // Capture left schema on first event
        if self.left_schema.is_none() {
            self.left_schema = Some(event.data.schema());
            self.update_output_schema();
        }

        self.process_event(event, JoinSide::Left, ctx)
    }

    /// Processes a right-side event.
    fn process_right(&mut self, event: &Event, ctx: &mut OperatorContext) -> OutputVec {
        self.metrics.right_events += 1;

        // Track side statistics for asymmetric compaction
        self.right_stats.record_event(ctx.processing_time);

        // Capture right schema on first event
        if self.right_schema.is_none() {
            self.right_schema = Some(event.data.schema());
            self.update_output_schema();
        }

        self.process_event(event, JoinSide::Right, ctx)
    }

    /// Updates the output schema when both input schemas are known.
    fn update_output_schema(&mut self) {
        if let (Some(left), Some(right)) = (&self.left_schema, &self.right_schema) {
            // Semi/anti joins output only the kept side's columns
            match self.join_type.kept_side() {
                Some(JoinSide::Left) => {
                    self.output_schema = Some(Arc::clone(left));
                    return;
                }
                Some(JoinSide::Right) => {
                    self.output_schema = Some(Arc::clone(right));
                    return;
                }
                None => {}
            }

            let mut fields: Vec<Field> =
                Vec::with_capacity(left.fields().len() + right.fields().len());
            fields.extend(left.fields().iter().map(|f| f.as_ref().clone()));

            // Add right fields, prefixing duplicates
            for field in right.fields() {
                let name = if left.field_with_name(field.name()).is_ok() {
                    format!("right_{}", field.name())
                } else {
                    field.name().clone()
                };
                fields.push(Field::new(
                    name,
                    field.data_type().clone(),
                    true, // Nullable for outer joins
                ));
            }

            self.output_schema = Some(Arc::new(Schema::new(fields)));
        }
    }

    /// Core event processing logic.
    fn process_event(
        &mut self,
        event: &Event,
        side: JoinSide,
        ctx: &mut OperatorContext,
    ) -> OutputVec {
        let mut output = OutputVec::new();
        let event_time = event.timestamp;

        // Update watermark
        let emitted_watermark = ctx.watermark_generator.on_event(event_time);

        // Track per-side watermarks for build-side pruning using actual watermark
        let current_wm = ctx.watermark_generator.current_watermark();
        match side {
            JoinSide::Left => self.left_watermark = self.left_watermark.max(current_wm),
            JoinSide::Right => self.right_watermark = self.right_watermark.max(current_wm),
        }

        // Check if event is too late
        if current_wm > i64::MIN && event_time + self.time_bound_ms < current_wm {
            self.metrics.late_events += 1;
            output.push(Output::LateEvent(event.clone()));
            return output;
        }

        // Extract join key
        let (key_column, cached_index) = match side {
            JoinSide::Left => (&self.left_key_column, &mut self.left_key_index),
            JoinSide::Right => (&self.right_key_column, &mut self.right_key_index),
        };
        let Some(key_value) = Self::extract_key(&event.data, key_column, cached_index) else {
            // Can't extract key, skip this event
            return output;
        };

        // Compute key hash for per-key tracking
        let key_hash = {
            let mut hasher = rustc_hash::FxHasher::default();
            key_value.hash(&mut hasher);
            hasher.finish()
        };

        // Track per-key metadata
        if self.per_key_tracking {
            self.key_metadata
                .entry(key_hash)
                .and_modify(|meta| meta.record_event(ctx.processing_time))
                .or_insert_with(|| KeyMetadata::new(ctx.processing_time));
            self.metrics.tracked_keys = self.key_metadata.len() as u64;
        }

        // Create join row with configured encoding
        let join_row = match JoinRow::with_encoding(
            event_time,
            key_value.to_vec(),
            &event.data,
            self.row_encoding,
        ) {
            Ok(row) => {
                // Track encoding metrics
                match self.row_encoding {
                    JoinRowEncoding::Compact => self.metrics.compact_encodes += 1,
                    JoinRowEncoding::CpuFriendly => self.metrics.cpu_friendly_encodes += 1,
                }
                row
            }
            Err(_) => return output,
        };

        // Store the event in state
        let state_key = Self::make_state_key(side, &key_value, event_time);
        if ctx.state.put_typed(&state_key, &join_row).is_err() {
            return output;
        }

        // Register cleanup timer
        let cleanup_time = event_time + self.time_bound_ms;
        let timer_key = Self::make_timer_key(side, &state_key);
        ctx.timers
            .register_timer(cleanup_time, Some(timer_key), Some(ctx.operator_index));

        // For outer joins, register unmatched emission timer
        if (side == JoinSide::Left && self.join_type.emits_unmatched_left())
            || (side == JoinSide::Right && self.join_type.emits_unmatched_right())
        {
            let unmatched_timer_key = Self::make_unmatched_timer_key(side, &state_key);
            ctx.timers.register_timer(
                cleanup_time,
                Some(unmatched_timer_key),
                Some(ctx.operator_index),
            );
        }

        // Build-side pruning - prune entries that can no longer produce matches
        if self.build_side_pruning {
            self.prune_build_side(side, ctx);
        }

        // Probe the opposite side for matches
        let matches = self.probe_opposite_side(side, &key_value, event_time, ctx.state);

        self.process_matches(
            matches,
            side,
            event,
            event_time,
            &join_row,
            &state_key,
            ctx,
            &mut output,
        );

        // Emit watermark if generated
        if let Some(wm) = emitted_watermark {
            output.push(Output::Watermark(wm.timestamp()));
        }

        output
    }

    /// Processes matched rows according to the join type (semi, anti, or standard).
    #[allow(clippy::too_many_arguments)]
    fn process_matches(
        &mut self,
        matches: SmallVec<[([u8; 28], JoinRow); 4]>,
        side: JoinSide,
        event: &Event,
        event_time: i64,
        join_row: &JoinRow,
        state_key: &[u8],
        ctx: &mut OperatorContext,
        output: &mut OutputVec,
    ) {
        if self.join_type.is_semi() {
            // Semi join: emit the kept side's row on first match only
            let kept_side = self.join_type.kept_side().unwrap();
            for (other_row_key, other_row) in matches {
                self.metrics.matches += 1;
                let was_other_matched = is_matched(&*ctx.state, &other_row_key);
                mark_matched(&mut *ctx.state, &other_row_key);

                let was_our_matched = is_matched(&*ctx.state, state_key);
                mark_matched(&mut *ctx.state, state_key);

                if side == kept_side && !was_our_matched {
                    output.push(Output::Event(Event::new(
                        event_time,
                        RecordBatch::clone(&event.data),
                    )));
                    break; // One match is enough for semi
                } else if side != kept_side && !was_other_matched {
                    if let Ok(batch) = other_row.to_batch() {
                        output.push(Output::Event(Event::new(other_row.timestamp, batch)));
                    }
                }
            }
        } else if self.join_type.is_anti() {
            // Anti join: mark rows as matched (suppresses unmatched timer emission)
            for (other_row_key, _other_row) in matches {
                self.metrics.matches += 1;
                mark_matched(&mut *ctx.state, &other_row_key);
                mark_matched(&mut *ctx.state, state_key);
            }
        } else {
            // Standard join: emit concatenated left+right rows
            let needs_matched_flag =
                self.join_type.emits_unmatched_left() || self.join_type.emits_unmatched_right();
            for (other_row_key, other_row) in matches {
                self.metrics.matches += 1;

                // Mark both sides as matched (only needed for outer joins)
                if needs_matched_flag {
                    mark_matched(&mut *ctx.state, &other_row_key);
                    mark_matched(&mut *ctx.state, state_key);
                }

                // Create joined output
                if let Some(joined_event) = self.create_joined_event(
                    side,
                    join_row,
                    &other_row,
                    std::cmp::max(event_time, other_row.timestamp),
                ) {
                    output.push(Output::Event(joined_event));
                }
            }
        }
    }

    /// Prunes build-side entries that cannot produce future matches.
    ///
    /// An entry can be pruned if its `timestamp + time_bound < probe_side_watermark`,
    /// meaning the probe side has advanced beyond any possible match window.
    fn prune_build_side(&mut self, current_side: JoinSide, ctx: &mut OperatorContext) {
        let build_side = self.effective_build_side();

        // Only prune when processing from probe side
        if current_side == build_side {
            return;
        }

        // Get probe side watermark
        let probe_watermark = match build_side {
            JoinSide::Left => self.right_watermark,
            JoinSide::Right => self.left_watermark,
        };

        if probe_watermark == i64::MIN {
            return;
        }

        // Calculate prune threshold
        let prune_threshold = probe_watermark - self.time_bound_ms;
        if prune_threshold == i64::MIN {
            return;
        }

        // For inner joins, we can prune more aggressively
        if self.join_type == JoinType::Inner {
            let prefix = match build_side {
                JoinSide::Left => LEFT_STATE_PREFIX,
                JoinSide::Right => RIGHT_STATE_PREFIX,
            };

            // Reuse prune_buffer to avoid per-call allocation
            self.prune_buffer.clear();
            let time_bound = self.time_bound_ms;
            for (key, value) in ctx.state.prefix_scan(prefix) {
                if self.prune_buffer.len() >= 100 {
                    break; // Limit per-event pruning to avoid latency spikes
                }
                // Try to get timestamp from the key (bytes 12-20)
                if key.len() >= 20 {
                    if let Ok(ts_bytes) = <[u8; 8]>::try_from(&key[12..20]) {
                        let timestamp = i64::from_be_bytes(ts_bytes);
                        if timestamp + time_bound < prune_threshold {
                            // Also verify via deserialization
                            if let Ok(row) =
                                rkyv::access::<rkyv::Archived<JoinRow>, RkyvError>(&value)
                                    .and_then(rkyv::deserialize::<JoinRow, RkyvError>)
                            {
                                if row.timestamp + time_bound < prune_threshold {
                                    self.prune_buffer.push(key);
                                }
                            }
                        }
                    }
                }
            }

            for key in &self.prune_buffer {
                if ctx.state.delete(key).is_ok() {
                    let _ = ctx.state.delete(&matched_flag_key(key));
                    self.metrics.build_side_prunes += 1;
                }
            }
        }
    }

    /// Scans for idle keys and cleans them up aggressively.
    ///
    /// Called periodically (e.g., on timer) to identify keys with no recent
    /// activity and remove their state entries.
    pub fn scan_idle_keys(&mut self, ctx: &mut OperatorContext) {
        if !self.per_key_tracking {
            return;
        }

        let threshold = ctx.processing_time - self.key_idle_threshold_ms;

        // Find idle keys
        let idle_keys: Vec<u64> = self
            .key_metadata
            .iter()
            .filter(|(_, meta)| meta.last_activity < threshold && meta.state_entries == 0)
            .map(|(k, _)| *k)
            .collect();

        // Remove idle key metadata
        for key_hash in idle_keys {
            self.key_metadata.remove(&key_hash);
            self.metrics.idle_key_cleanups += 1;
        }

        self.metrics.tracked_keys = self.key_metadata.len() as u64;
    }

    /// Checks if compaction should be skipped for a side due to asymmetric optimization.
    #[must_use]
    pub fn should_skip_compaction(&self, side: JoinSide, current_time: i64) -> bool {
        if !self.asymmetric_compaction {
            return false;
        }

        // Skip compaction if side is idle
        let is_idle = match side {
            JoinSide::Left => self
                .left_stats
                .is_idle(current_time, self.idle_threshold_ms),
            JoinSide::Right => self
                .right_stats
                .is_idle(current_time, self.idle_threshold_ms),
        };

        if is_idle {
            // Note: metrics update happens in the caller
            true
        } else {
            false
        }
    }

    /// Extracts the join key value from a record batch.
    ///
    /// Caches the column index on first call to avoid per-event schema lookups.
    fn extract_key(
        batch: &RecordBatch,
        column_name: &str,
        cached_index: &mut Option<usize>,
    ) -> Option<SmallVec<[u8; 16]>> {
        let column_index = if let Some(idx) = *cached_index {
            idx
        } else {
            let idx = batch.schema().index_of(column_name).ok()?;
            *cached_index = Some(idx);
            idx
        };
        let column = batch.column(column_index);

        // Handle different column types
        if let Some(string_array) = column.as_any().downcast_ref::<StringArray>() {
            if string_array.is_empty() || string_array.is_null(0) {
                return None;
            }
            return Some(SmallVec::from_slice(string_array.value(0).as_bytes()));
        }

        if let Some(int_array) = column.as_any().downcast_ref::<Int64Array>() {
            if int_array.is_empty() || int_array.is_null(0) {
                return None;
            }
            return Some(SmallVec::from_slice(&int_array.value(0).to_le_bytes()));
        }

        // For other types, use the raw bytes if available
        // This is a fallback - in practice, keys should be string or integer
        None
    }

    /// Creates a state key for storing a join row.
    #[allow(clippy::cast_sign_loss)]
    fn make_state_key(side: JoinSide, key_value: &[u8], timestamp: i64) -> [u8; 28] {
        let prefix = match side {
            JoinSide::Left => LEFT_STATE_PREFIX,
            JoinSide::Right => RIGHT_STATE_PREFIX,
        };

        let event_id = EVENT_ID_COUNTER.fetch_add(1, Ordering::Relaxed);

        // Key format: prefix (4) + key_hash (8) + timestamp (8) + event_id (8) = 28 bytes
        let mut key = [0u8; 28];
        key[..4].copy_from_slice(prefix);

        // Use FxHash for the key value
        let key_hash = {
            let mut hasher = rustc_hash::FxHasher::default();
            key_value.hash(&mut hasher);
            hasher.finish()
        };
        key[4..12].copy_from_slice(&key_hash.to_be_bytes());
        key[12..20].copy_from_slice(&timestamp.to_be_bytes());
        key[20..28].copy_from_slice(&event_id.to_be_bytes());

        key
    }

    /// Creates a timer key for cleanup.
    fn make_timer_key(side: JoinSide, state_key: &[u8]) -> TimerKey {
        let prefix = match side {
            JoinSide::Left => LEFT_TIMER_PREFIX,
            JoinSide::Right => RIGHT_TIMER_PREFIX,
        };

        let mut key = TimerKey::new();
        key.push(prefix);
        key.extend_from_slice(state_key);
        key
    }

    /// Creates a timer key for unmatched event emission.
    fn make_unmatched_timer_key(side: JoinSide, state_key: &[u8]) -> TimerKey {
        let side_byte = match side {
            JoinSide::Left => 0x01,
            JoinSide::Right => 0x02,
        };

        let mut key = TimerKey::new();
        key.push(UNMATCHED_TIMER_PREFIX);
        key.push(side_byte);
        key.extend_from_slice(state_key);
        key
    }

    /// Probes the opposite side for matching events.
    fn probe_opposite_side(
        &self,
        current_side: JoinSide,
        key_value: &[u8],
        timestamp: i64,
        state: &dyn StateStore,
    ) -> SmallVec<[([u8; 28], JoinRow); 4]> {
        let mut matches = SmallVec::new();

        let prefix = match current_side {
            JoinSide::Left => RIGHT_STATE_PREFIX,
            JoinSide::Right => LEFT_STATE_PREFIX,
        };

        // Build prefix for scanning: prefix + key_hash
        let key_hash = {
            let mut hasher = rustc_hash::FxHasher::default();
            key_value.hash(&mut hasher);
            hasher.finish()
        };
        let mut scan_prefix = [0u8; 12];
        scan_prefix[..4].copy_from_slice(prefix);
        scan_prefix[4..12].copy_from_slice(&key_hash.to_be_bytes());

        // Scan for matching keys
        for (state_key, value) in state.prefix_scan(&scan_prefix) {
            // Deserialize the join row
            let Ok(row) = rkyv::access::<rkyv::Archived<JoinRow>, RkyvError>(&value)
                .and_then(rkyv::deserialize::<JoinRow, RkyvError>)
            else {
                continue;
            };

            // Check if timestamps are within time bound
            let time_diff = (timestamp - row.timestamp).abs();
            if time_diff <= self.time_bound_ms {
                // Verify key matches (in case of hash collision)
                if row.key_value == key_value {
                    let mut key_buf = [0u8; 28];
                    if state_key.len() == 28 {
                        key_buf.copy_from_slice(&state_key);
                    }
                    matches.push((key_buf, row));
                }
            }
        }

        matches
    }

    /// Creates a joined event from two matching rows.
    fn create_joined_event(
        &self,
        current_side: JoinSide,
        current_row: &JoinRow,
        other_row: &JoinRow,
        output_timestamp: i64,
    ) -> Option<Event> {
        let (left_row, right_row) = match current_side {
            JoinSide::Left => (current_row, other_row),
            JoinSide::Right => (other_row, current_row),
        };

        let left_batch = left_row.to_batch().ok()?;
        let right_batch = right_row.to_batch().ok()?;

        let joined_batch = self.concat_batches(&left_batch, &right_batch)?;

        Some(Event::new(output_timestamp, joined_batch))
    }

    /// Concatenates two batches horizontally.
    fn concat_batches(&self, left: &RecordBatch, right: &RecordBatch) -> Option<RecordBatch> {
        let schema = self.output_schema.as_ref()?;

        let mut columns: Vec<ArrayRef> =
            Vec::with_capacity(left.num_columns() + right.num_columns());
        columns.extend(left.columns().iter().cloned());
        columns.extend(right.columns().iter().cloned());

        RecordBatch::try_new(Arc::clone(schema), columns).ok()
    }

    /// Creates an unmatched event for outer joins.
    fn create_unmatched_event(&self, side: JoinSide, row: &JoinRow) -> Option<Event> {
        let batch = row.to_batch().ok()?;
        let schema = self.output_schema.as_ref()?;

        let num_rows = batch.num_rows();
        let mut columns: Vec<ArrayRef> = Vec::with_capacity(schema.fields().len());

        match side {
            JoinSide::Left => {
                // Left columns are populated, right columns are null
                columns.extend(batch.columns().iter().cloned());

                // Add null columns for right side
                if let Some(right_schema) = &self.right_schema {
                    for field in right_schema.fields() {
                        columns.push(Self::create_null_array(field.data_type(), num_rows));
                    }
                }
            }
            JoinSide::Right => {
                // Left columns are null, right columns are populated
                if let Some(left_schema) = &self.left_schema {
                    for field in left_schema.fields() {
                        columns.push(Self::create_null_array(field.data_type(), num_rows));
                    }
                }

                columns.extend(batch.columns().iter().cloned());
            }
        }

        let joined_batch = RecordBatch::try_new(Arc::clone(schema), columns).ok()?;

        Some(Event::new(row.timestamp, joined_batch))
    }

    /// Creates a null array of the given type and length.
    fn create_null_array(data_type: &DataType, num_rows: usize) -> ArrayRef {
        match data_type {
            DataType::Utf8 => Arc::new(StringArray::from(vec![None::<&str>; num_rows])) as ArrayRef,
            // Default to Int64 for all other types (add more specific cases as needed)
            _ => Arc::new(Int64Array::from(vec![None; num_rows])) as ArrayRef,
        }
    }

    /// Handles cleanup timer expiration.
    fn handle_cleanup_timer(
        &mut self,
        side: JoinSide,
        state_key: &[u8],
        ctx: &mut OperatorContext,
    ) -> OutputVec {
        let output = OutputVec::new();

        // Check asymmetric compaction - skip if side is idle
        if self.should_skip_compaction(side, ctx.processing_time) {
            self.metrics.asymmetric_skips += 1;
            // Don't skip the actual cleanup, but could be used for compaction
        }

        // Update per-key metadata before deleting
        if self.per_key_tracking && state_key.len() >= 12 {
            // Extract key hash from state key (bytes 4-12)
            if let Ok(hash_bytes) = state_key[4..12].try_into() {
                let key_hash = u64::from_be_bytes(hash_bytes);
                if let Some(meta) = self.key_metadata.get_mut(&key_hash) {
                    meta.decrement_entries();
                }
            }
        }

        // For outer/anti joins, defer state deletion to the unmatched timer handler
        // so the row is still readable when the unmatched timer fires.
        let defer_cleanup = match side {
            JoinSide::Left => self.join_type.emits_unmatched_left(),
            JoinSide::Right => self.join_type.emits_unmatched_right(),
        };
        if !defer_cleanup && ctx.state.delete(state_key).is_ok() {
            let _ = ctx.state.delete(&matched_flag_key(state_key));
            self.metrics.state_cleanups += 1;
        }

        output
    }

    /// Handles unmatched timer expiration for outer/anti joins.
    fn handle_unmatched_timer(
        &mut self,
        side: JoinSide,
        state_key: &[u8],
        ctx: &mut OperatorContext,
    ) -> OutputVec {
        let mut output = OutputVec::new();

        // Get the join row
        let Ok(Some(row)) = ctx.state.get_typed::<JoinRow>(state_key) else {
            return output;
        };

        // Only emit if not matched (check separate flag key, not JoinRow field)
        if !is_matched(&*ctx.state, state_key) {
            match side {
                JoinSide::Left if self.join_type.emits_unmatched_left() => {
                    self.metrics.unmatched_left += 1;
                    if self.join_type.is_anti() {
                        // Anti join: emit just the kept side's data
                        if let Ok(batch) = row.to_batch() {
                            output.push(Output::Event(Event::new(row.timestamp, batch)));
                        }
                    } else if let Some(event) = self.create_unmatched_event(side, &row) {
                        output.push(Output::Event(event));
                    }
                }
                JoinSide::Right if self.join_type.emits_unmatched_right() => {
                    self.metrics.unmatched_right += 1;
                    if self.join_type.is_anti() {
                        // Anti join: emit just the kept side's data
                        if let Ok(batch) = row.to_batch() {
                            output.push(Output::Event(Event::new(row.timestamp, batch)));
                        }
                    } else if let Some(event) = self.create_unmatched_event(side, &row) {
                        output.push(Output::Event(event));
                    }
                }
                _ => {}
            }
        }

        // Clean up state (deferred from cleanup timer for outer/anti joins)
        if ctx.state.delete(state_key).is_ok() {
            let _ = ctx.state.delete(&matched_flag_key(state_key));
            self.metrics.state_cleanups += 1;
        }

        output
    }

    /// Parses a timer key to determine its type and extract the state key.
    fn parse_timer_key(key: &[u8]) -> Option<(TimerKeyType, JoinSide, Vec<u8>)> {
        if key.is_empty() {
            return None;
        }

        match key[0] {
            LEFT_TIMER_PREFIX => {
                let state_key = key[1..].to_vec();
                Some((TimerKeyType::Cleanup, JoinSide::Left, state_key))
            }
            RIGHT_TIMER_PREFIX => {
                let state_key = key[1..].to_vec();
                Some((TimerKeyType::Cleanup, JoinSide::Right, state_key))
            }
            UNMATCHED_TIMER_PREFIX => {
                if key.len() < 2 {
                    return None;
                }
                let side = match key[1] {
                    0x01 => JoinSide::Left,
                    0x02 => JoinSide::Right,
                    _ => return None,
                };
                let state_key = key[2..].to_vec();
                Some((TimerKeyType::Unmatched, side, state_key))
            }
            _ => None,
        }
    }
}

/// Type of timer key.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum TimerKeyType {
    /// Cleanup timer - delete state entry.
    Cleanup,
    /// Unmatched timer - emit unmatched event for outer joins.
    Unmatched,
}

impl Operator for StreamJoinOperator {
    fn process(&mut self, event: &Event, ctx: &mut OperatorContext) -> OutputVec {
        // Default to left side - in practice, users should call process_side directly
        self.process_left(event, ctx)
    }

    fn on_timer(&mut self, timer: Timer, ctx: &mut OperatorContext) -> OutputVec {
        let Some((timer_type, side, state_key)) = Self::parse_timer_key(&timer.key) else {
            return OutputVec::new();
        };

        match timer_type {
            TimerKeyType::Cleanup => self.handle_cleanup_timer(side, &state_key, ctx),
            TimerKeyType::Unmatched => self.handle_unmatched_timer(side, &state_key, ctx),
        }
    }

    fn checkpoint(&self) -> OperatorState {
        // Checkpoint the metrics, configuration, and optimization state
        // Use nested tuples to stay within rkyv's tuple size limit (max 12)
        // Format: ((config), (core_metrics), (f057_metrics, side_stats))
        let checkpoint_data = (
            // Part 1: Configuration (3 elements)
            (
                self.left_key_column.clone(),
                self.right_key_column.clone(),
                self.time_bound_ms,
            ),
            // Part 2: Core metrics (3 elements)
            (
                self.metrics.left_events,
                self.metrics.right_events,
                self.metrics.matches,
            ),
            // Part 3: optimization metrics (5 elements)
            (
                self.metrics.cpu_friendly_encodes,
                self.metrics.compact_encodes,
                self.metrics.asymmetric_skips,
                self.metrics.idle_key_cleanups,
                self.metrics.build_side_prunes,
            ),
            // Part 4: side stats (4 elements)
            (
                self.left_stats.events_received,
                self.right_stats.events_received,
                self.left_watermark,
                self.right_watermark,
            ),
        );

        let data = rkyv::to_bytes::<RkyvError>(&checkpoint_data)
            .map(|v| v.to_vec())
            .unwrap_or_default();

        OperatorState {
            operator_id: self.operator_id.clone(),
            data,
        }
    }

    fn restore(&mut self, state: OperatorState) -> Result<(), OperatorError> {
        // Extended checkpoint data type with optimization fields using nested tuples
        type CheckpointData = (
            (String, String, i64),     // config
            (u64, u64, u64),           // core metrics
            (u64, u64, u64, u64, u64), // optimization metrics
            (u64, u64, i64, i64),      // side stats
        );
        // Legacy checkpoint type for backward compatibility
        type LegacyCheckpointData = (String, String, i64, u64, u64, u64);

        if state.operator_id != self.operator_id {
            return Err(OperatorError::StateAccessFailed(format!(
                "Operator ID mismatch: expected {}, got {}",
                self.operator_id, state.operator_id
            )));
        }

        // Try to restore full optimization checkpoint first
        if let Ok(archived) = rkyv::access::<rkyv::Archived<CheckpointData>, RkyvError>(&state.data)
        {
            if let Ok(data) = rkyv::deserialize::<CheckpointData, RkyvError>(archived) {
                let (
                    _config,
                    (left_events, right_events, matches),
                    (
                        cpu_friendly_encodes,
                        compact_encodes,
                        asymmetric_skips,
                        idle_key_cleanups,
                        build_side_prunes,
                    ),
                    (left_received, right_received, left_wm, right_wm),
                ) = data;

                self.metrics.left_events = left_events;
                self.metrics.right_events = right_events;
                self.metrics.matches = matches;
                self.metrics.cpu_friendly_encodes = cpu_friendly_encodes;
                self.metrics.compact_encodes = compact_encodes;
                self.metrics.asymmetric_skips = asymmetric_skips;
                self.metrics.idle_key_cleanups = idle_key_cleanups;
                self.metrics.build_side_prunes = build_side_prunes;
                self.left_stats.events_received = left_received;
                self.right_stats.events_received = right_received;
                self.left_watermark = left_wm;
                self.right_watermark = right_wm;

                return Ok(());
            }
        }

        // Fall back to legacy checkpoint format
        let archived = rkyv::access::<rkyv::Archived<LegacyCheckpointData>, RkyvError>(&state.data)
            .map_err(|e| OperatorError::SerializationFailed(e.to_string()))?;
        let (_, _, _, left_events, right_events, matches) =
            rkyv::deserialize::<LegacyCheckpointData, RkyvError>(archived)
                .map_err(|e| OperatorError::SerializationFailed(e.to_string()))?;

        self.metrics.left_events = left_events;
        self.metrics.right_events = right_events;
        self.metrics.matches = matches;

        Ok(())
    }
}

#[cfg(test)]
#[allow(clippy::cast_possible_wrap)]
mod tests {
    use super::*;
    use crate::state::InMemoryStore;
    use crate::time::{BoundedOutOfOrdernessGenerator, TimerService, WatermarkGenerator};
    use arrow_array::{Int64Array, StringArray};
    use arrow_schema::{DataType, Field, Schema};

    fn create_order_event(timestamp: i64, order_id: &str, amount: i64) -> Event {
        let schema = Arc::new(Schema::new(vec![
            Field::new("order_id", DataType::Utf8, false),
            Field::new("amount", DataType::Int64, false),
        ]));
        let batch = RecordBatch::try_new(
            schema,
            vec![
                Arc::new(StringArray::from(vec![order_id])),
                Arc::new(Int64Array::from(vec![amount])),
            ],
        )
        .unwrap();
        Event::new(timestamp, batch)
    }

    fn create_payment_event(timestamp: i64, order_id: &str, status: &str) -> Event {
        let schema = Arc::new(Schema::new(vec![
            Field::new("order_id", DataType::Utf8, false),
            Field::new("status", DataType::Utf8, false),
        ]));
        let batch = RecordBatch::try_new(
            schema,
            vec![
                Arc::new(StringArray::from(vec![order_id])),
                Arc::new(StringArray::from(vec![status])),
            ],
        )
        .unwrap();
        Event::new(timestamp, batch)
    }

    fn create_test_context<'a>(
        timers: &'a mut TimerService,
        state: &'a mut dyn StateStore,
        watermark_gen: &'a mut dyn WatermarkGenerator,
    ) -> OperatorContext<'a> {
        OperatorContext {
            event_time: 0,
            processing_time: 0,
            timers,
            state,
            watermark_generator: watermark_gen,
            operator_index: 0,
        }
    }

    #[test]
    fn test_join_type_properties() {
        assert!(!JoinType::Inner.emits_unmatched_left());
        assert!(!JoinType::Inner.emits_unmatched_right());

        assert!(JoinType::Left.emits_unmatched_left());
        assert!(!JoinType::Left.emits_unmatched_right());

        assert!(!JoinType::Right.emits_unmatched_left());
        assert!(JoinType::Right.emits_unmatched_right());

        assert!(JoinType::Full.emits_unmatched_left());
        assert!(JoinType::Full.emits_unmatched_right());

        // Semi joins: never emit unmatched
        assert!(!JoinType::LeftSemi.emits_unmatched_left());
        assert!(!JoinType::LeftSemi.emits_unmatched_right());
        assert!(!JoinType::RightSemi.emits_unmatched_left());
        assert!(!JoinType::RightSemi.emits_unmatched_right());

        // Anti joins: emit unmatched for the kept side only
        assert!(JoinType::LeftAnti.emits_unmatched_left());
        assert!(!JoinType::LeftAnti.emits_unmatched_right());
        assert!(!JoinType::RightAnti.emits_unmatched_left());
        assert!(JoinType::RightAnti.emits_unmatched_right());

        // Semi/Anti helpers
        assert!(JoinType::LeftSemi.is_semi());
        assert!(JoinType::RightSemi.is_semi());
        assert!(!JoinType::LeftAnti.is_semi());
        assert!(JoinType::LeftAnti.is_anti());
        assert!(JoinType::RightAnti.is_anti());
        assert!(!JoinType::Inner.is_semi());
        assert!(!JoinType::Inner.is_anti());

        // Kept side
        assert_eq!(JoinType::LeftSemi.kept_side(), Some(JoinSide::Left));
        assert_eq!(JoinType::LeftAnti.kept_side(), Some(JoinSide::Left));
        assert_eq!(JoinType::RightSemi.kept_side(), Some(JoinSide::Right));
        assert_eq!(JoinType::RightAnti.kept_side(), Some(JoinSide::Right));
        assert_eq!(JoinType::Inner.kept_side(), None);
    }

    #[test]
    fn test_left_semi_join_only_emits_matched() {
        let mut operator = StreamJoinOperator::with_id(
            "order_id".to_string(),
            "order_id".to_string(),
            Duration::from_secs(3600),
            JoinType::LeftSemi,
            "test_semi".to_string(),
        );

        let mut timers = TimerService::new();
        let mut state = InMemoryStore::new();
        let mut watermark_gen = BoundedOutOfOrdernessGenerator::new(100);

        // Left event (order) — no match yet
        let order = create_order_event(1000, "order_1", 100);
        {
            let mut ctx = create_test_context(&mut timers, &mut state, &mut watermark_gen);
            let outputs = operator.process_side(&order, JoinSide::Left, &mut ctx);
            let events: Vec<_> = outputs
                .iter()
                .filter(|o| matches!(o, Output::Event(_)))
                .collect();
            assert_eq!(events.len(), 0, "semi join should not emit without match");
        }

        // Right event (payment) — match found, should emit left row
        let payment = create_payment_event(2000, "order_1", "paid");
        let outputs = {
            let mut ctx = create_test_context(&mut timers, &mut state, &mut watermark_gen);
            operator.process_side(&payment, JoinSide::Right, &mut ctx)
        };
        let events: Vec<_> = outputs
            .iter()
            .filter_map(|o| match o {
                Output::Event(e) => Some(e),
                _ => None,
            })
            .collect();
        assert_eq!(events.len(), 1, "semi join should emit left row on match");
        // Output should only have left columns (order_id, amount) — NOT concatenated
        assert_eq!(events[0].data.num_columns(), 2);
        assert_eq!(events[0].data.schema().field(0).name(), "order_id");
        assert_eq!(events[0].data.schema().field(1).name(), "amount");

        // Second right match for same left — should NOT emit again
        let payment2 = create_payment_event(2500, "order_1", "refunded");
        {
            let mut ctx = create_test_context(&mut timers, &mut state, &mut watermark_gen);
            let outputs = operator.process_side(&payment2, JoinSide::Right, &mut ctx);
            let events: Vec<_> = outputs
                .iter()
                .filter(|o| matches!(o, Output::Event(_)))
                .collect();
            assert_eq!(events.len(), 0, "semi join should not emit duplicate");
        }
    }

    #[test]
    fn test_left_anti_join_only_emits_unmatched() {
        let mut operator = StreamJoinOperator::with_id(
            "order_id".to_string(),
            "order_id".to_string(),
            Duration::from_secs(3600),
            JoinType::LeftAnti,
            "test_anti".to_string(),
        );

        let mut timers = TimerService::new();
        let mut state = InMemoryStore::new();
        let mut watermark_gen = BoundedOutOfOrdernessGenerator::new(100);

        // Left event for a key that WILL have a match (should NOT be emitted)
        let matched_order = create_order_event(1000, "order_1", 100);
        {
            let mut ctx = create_test_context(&mut timers, &mut state, &mut watermark_gen);
            let outputs = operator.process_side(&matched_order, JoinSide::Left, &mut ctx);
            let events: Vec<_> = outputs
                .iter()
                .filter(|o| matches!(o, Output::Event(_)))
                .collect();
            assert_eq!(events.len(), 0, "anti join should not emit on insert");
        }

        // Left event for a key that will NOT have a match (should be emitted at timer)
        let unmatched_order = create_order_event(1100, "order_2", 200);
        {
            let mut ctx = create_test_context(&mut timers, &mut state, &mut watermark_gen);
            let outputs = operator.process_side(&unmatched_order, JoinSide::Left, &mut ctx);
            let events: Vec<_> = outputs
                .iter()
                .filter(|o| matches!(o, Output::Event(_)))
                .collect();
            assert_eq!(events.len(), 0, "anti join should not emit on insert");
        }

        // Right event matches order_1 — suppresses anti emission
        let payment = create_payment_event(2000, "order_1", "paid");
        {
            let mut ctx = create_test_context(&mut timers, &mut state, &mut watermark_gen);
            let outputs = operator.process_side(&payment, JoinSide::Right, &mut ctx);
            let events: Vec<_> = outputs
                .iter()
                .filter(|o| matches!(o, Output::Event(_)))
                .collect();
            assert_eq!(events.len(), 0, "anti join should never emit on match");
        }

        // Fire timers — order_2 should be emitted (unmatched), order_1 should NOT
        let pending_timers = timers.poll_timers(i64::MAX);
        let mut anti_outputs = 0;
        for timer_reg in pending_timers {
            let timer = Timer {
                key: timer_reg.key.unwrap_or_default(),
                timestamp: timer_reg.timestamp,
            };
            let mut ctx = create_test_context(&mut timers, &mut state, &mut watermark_gen);
            let outputs = operator.on_timer(timer, &mut ctx);
            for out in &outputs {
                if let Output::Event(e) = out {
                    anti_outputs += 1;
                    // Should only have left columns
                    assert_eq!(e.data.num_columns(), 2);
                    assert_eq!(e.data.schema().field(0).name(), "order_id");
                }
            }
        }
        assert!(
            anti_outputs >= 1,
            "anti join should emit unmatched left rows on timer"
        );
    }

    #[test]
    fn test_join_operator_creation() {
        let operator = StreamJoinOperator::new(
            "order_id".to_string(),
            "order_id".to_string(),
            Duration::from_secs(3600),
            JoinType::Inner,
        );

        assert_eq!(operator.join_type(), JoinType::Inner);
        assert_eq!(operator.time_bound_ms(), 3_600_000);
    }

    #[test]
    fn test_join_operator_with_id() {
        let operator = StreamJoinOperator::with_id(
            "order_id".to_string(),
            "order_id".to_string(),
            Duration::from_secs(3600),
            JoinType::Left,
            "test_join".to_string(),
        );

        assert_eq!(operator.operator_id, "test_join");
        assert_eq!(operator.join_type(), JoinType::Left);
    }

    #[test]
    fn test_inner_join_basic() {
        let mut operator = StreamJoinOperator::with_id(
            "order_id".to_string(),
            "order_id".to_string(),
            Duration::from_secs(3600),
            JoinType::Inner,
            "test_join".to_string(),
        );

        let mut timers = TimerService::new();
        let mut state = InMemoryStore::new();
        let mut watermark_gen = BoundedOutOfOrdernessGenerator::new(100);

        // Process left event (order)
        let order = create_order_event(1000, "order_1", 100);
        {
            let mut ctx = create_test_context(&mut timers, &mut state, &mut watermark_gen);
            let outputs = operator.process_side(&order, JoinSide::Left, &mut ctx);
            // No match yet, should produce no output
            assert!(
                outputs
                    .iter()
                    .filter(|o| matches!(o, Output::Event(_)))
                    .count()
                    == 0
            );
        }

        // Process right event (payment) - should produce a match
        let payment = create_payment_event(2000, "order_1", "paid");
        {
            let mut ctx = create_test_context(&mut timers, &mut state, &mut watermark_gen);
            let outputs = operator.process_side(&payment, JoinSide::Right, &mut ctx);
            // Should have one match
            assert_eq!(
                outputs
                    .iter()
                    .filter(|o| matches!(o, Output::Event(_)))
                    .count(),
                1
            );
        }

        assert_eq!(operator.metrics().matches, 1);
        assert_eq!(operator.metrics().left_events, 1);
        assert_eq!(operator.metrics().right_events, 1);
    }

    #[test]
    fn test_inner_join_no_match_different_key() {
        let mut operator = StreamJoinOperator::with_id(
            "order_id".to_string(),
            "order_id".to_string(),
            Duration::from_secs(3600),
            JoinType::Inner,
            "test_join".to_string(),
        );

        let mut timers = TimerService::new();
        let mut state = InMemoryStore::new();
        let mut watermark_gen = BoundedOutOfOrdernessGenerator::new(100);

        // Process left event
        let order = create_order_event(1000, "order_1", 100);
        {
            let mut ctx = create_test_context(&mut timers, &mut state, &mut watermark_gen);
            operator.process_side(&order, JoinSide::Left, &mut ctx);
        }

        // Process right event with different key
        let payment = create_payment_event(2000, "order_2", "paid");
        {
            let mut ctx = create_test_context(&mut timers, &mut state, &mut watermark_gen);
            let outputs = operator.process_side(&payment, JoinSide::Right, &mut ctx);
            // No match due to different key
            assert_eq!(
                outputs
                    .iter()
                    .filter(|o| matches!(o, Output::Event(_)))
                    .count(),
                0
            );
        }

        assert_eq!(operator.metrics().matches, 0);
    }

    #[test]
    fn test_inner_join_no_match_outside_time_bound() {
        let mut operator = StreamJoinOperator::with_id(
            "order_id".to_string(),
            "order_id".to_string(),
            Duration::from_secs(1), // 1 second time bound
            JoinType::Inner,
            "test_join".to_string(),
        );

        let mut timers = TimerService::new();
        let mut state = InMemoryStore::new();
        let mut watermark_gen = BoundedOutOfOrdernessGenerator::new(100);

        // Process left event at t=1000
        let order = create_order_event(1000, "order_1", 100);
        {
            let mut ctx = create_test_context(&mut timers, &mut state, &mut watermark_gen);
            operator.process_side(&order, JoinSide::Left, &mut ctx);
        }

        // Process right event at t=5000 (4 seconds later, outside 1s bound)
        let payment = create_payment_event(5000, "order_1", "paid");
        {
            let mut ctx = create_test_context(&mut timers, &mut state, &mut watermark_gen);
            let outputs = operator.process_side(&payment, JoinSide::Right, &mut ctx);
            // No match due to time bound
            assert_eq!(
                outputs
                    .iter()
                    .filter(|o| matches!(o, Output::Event(_)))
                    .count(),
                0
            );
        }

        assert_eq!(operator.metrics().matches, 0);
    }

    #[test]
    fn test_join_multiple_matches() {
        let mut operator = StreamJoinOperator::with_id(
            "order_id".to_string(),
            "order_id".to_string(),
            Duration::from_secs(3600),
            JoinType::Inner,
            "test_join".to_string(),
        );

        let mut timers = TimerService::new();
        let mut state = InMemoryStore::new();
        let mut watermark_gen = BoundedOutOfOrdernessGenerator::new(100);

        // Process two left events with same key
        for ts in [1000, 2000] {
            let order = create_order_event(ts, "order_1", 100);
            let mut ctx = create_test_context(&mut timers, &mut state, &mut watermark_gen);
            operator.process_side(&order, JoinSide::Left, &mut ctx);
        }

        // Process right event - should match both
        let payment = create_payment_event(1500, "order_1", "paid");
        {
            let mut ctx = create_test_context(&mut timers, &mut state, &mut watermark_gen);
            let outputs = operator.process_side(&payment, JoinSide::Right, &mut ctx);
            // Should have two matches
            assert_eq!(
                outputs
                    .iter()
                    .filter(|o| matches!(o, Output::Event(_)))
                    .count(),
                2
            );
        }

        assert_eq!(operator.metrics().matches, 2);
    }

    #[test]
    fn test_join_late_event() {
        let mut operator = StreamJoinOperator::with_id(
            "order_id".to_string(),
            "order_id".to_string(),
            Duration::from_secs(1),
            JoinType::Inner,
            "test_join".to_string(),
        );

        let mut timers = TimerService::new();
        let mut state = InMemoryStore::new();
        let mut watermark_gen = BoundedOutOfOrdernessGenerator::new(0);

        // Advance watermark significantly
        let future_order = create_order_event(10000, "order_2", 200);
        {
            let mut ctx = create_test_context(&mut timers, &mut state, &mut watermark_gen);
            operator.process_side(&future_order, JoinSide::Left, &mut ctx);
        }

        // Process very late event
        let late_payment = create_payment_event(100, "order_1", "paid");
        {
            let mut ctx = create_test_context(&mut timers, &mut state, &mut watermark_gen);
            let outputs = operator.process_side(&late_payment, JoinSide::Right, &mut ctx);
            // Should be marked as late
            assert!(outputs.iter().any(|o| matches!(o, Output::LateEvent(_))));
        }

        assert_eq!(operator.metrics().late_events, 1);
    }

    #[test]
    fn test_join_row_serialization() {
        let schema = Arc::new(Schema::new(vec![
            Field::new("id", DataType::Utf8, false),
            Field::new("value", DataType::Int64, false),
        ]));
        let batch = RecordBatch::try_new(
            schema,
            vec![
                Arc::new(StringArray::from(vec!["test"])),
                Arc::new(Int64Array::from(vec![42])),
            ],
        )
        .unwrap();

        let row = JoinRow::new(1000, b"key".to_vec(), &batch).unwrap();

        // Verify we can deserialize back
        let restored_batch = row.to_batch().unwrap();
        assert_eq!(restored_batch.num_rows(), 1);
        assert_eq!(restored_batch.num_columns(), 2);
    }

    #[test]
    fn test_cleanup_timer() {
        let mut operator = StreamJoinOperator::with_id(
            "order_id".to_string(),
            "order_id".to_string(),
            Duration::from_secs(1),
            JoinType::Inner,
            "test_join".to_string(),
        );

        let mut timers = TimerService::new();
        let mut state = InMemoryStore::new();
        let mut watermark_gen = BoundedOutOfOrdernessGenerator::new(100);

        // Process an event
        let order = create_order_event(1000, "order_1", 100);
        {
            let mut ctx = create_test_context(&mut timers, &mut state, &mut watermark_gen);
            operator.process_side(&order, JoinSide::Left, &mut ctx);
        }

        // State should have one entry
        assert!(state.len() > 0);
        let initial_state_len = state.len();

        // Get the registered timers and fire them
        let registered_timers = timers.poll_timers(2001); // After cleanup time
        assert!(!registered_timers.is_empty());

        // Fire the cleanup timer - convert TimerRegistration to Timer
        for timer_reg in registered_timers {
            let timer = Timer {
                key: timer_reg.key.unwrap_or_default(),
                timestamp: timer_reg.timestamp,
            };
            let mut ctx = create_test_context(&mut timers, &mut state, &mut watermark_gen);
            operator.on_timer(timer, &mut ctx);
        }

        // Verify cleanup happened (state decreased)
        assert!(state.len() < initial_state_len || operator.metrics().state_cleanups > 0);
    }

    #[test]
    fn test_checkpoint_restore() {
        let mut operator = StreamJoinOperator::with_id(
            "order_id".to_string(),
            "order_id".to_string(),
            Duration::from_secs(3600),
            JoinType::Inner,
            "test_join".to_string(),
        );

        // Simulate some activity
        operator.metrics.left_events = 10;
        operator.metrics.right_events = 5;
        operator.metrics.matches = 3;

        // Checkpoint
        let checkpoint = operator.checkpoint();

        // Create new operator and restore
        let mut restored = StreamJoinOperator::with_id(
            "order_id".to_string(),
            "order_id".to_string(),
            Duration::from_secs(3600),
            JoinType::Inner,
            "test_join".to_string(),
        );

        restored.restore(checkpoint).unwrap();

        assert_eq!(restored.metrics().left_events, 10);
        assert_eq!(restored.metrics().right_events, 5);
        assert_eq!(restored.metrics().matches, 3);
    }

    #[test]
    fn test_metrics_reset() {
        let mut operator = StreamJoinOperator::new(
            "order_id".to_string(),
            "order_id".to_string(),
            Duration::from_secs(3600),
            JoinType::Inner,
        );

        operator.metrics.left_events = 10;
        operator.metrics.matches = 5;

        operator.reset_metrics();

        assert_eq!(operator.metrics().left_events, 0);
        assert_eq!(operator.metrics().matches, 0);
    }

    #[test]
    fn test_bidirectional_join() {
        let mut operator = StreamJoinOperator::with_id(
            "order_id".to_string(),
            "order_id".to_string(),
            Duration::from_secs(3600),
            JoinType::Inner,
            "test_join".to_string(),
        );

        let mut timers = TimerService::new();
        let mut state = InMemoryStore::new();
        let mut watermark_gen = BoundedOutOfOrdernessGenerator::new(100);

        // Right event arrives first
        let payment = create_payment_event(1000, "order_1", "paid");
        {
            let mut ctx = create_test_context(&mut timers, &mut state, &mut watermark_gen);
            let outputs = operator.process_side(&payment, JoinSide::Right, &mut ctx);
            assert_eq!(
                outputs
                    .iter()
                    .filter(|o| matches!(o, Output::Event(_)))
                    .count(),
                0
            );
        }

        // Left event arrives and matches
        let order = create_order_event(1500, "order_1", 100);
        {
            let mut ctx = create_test_context(&mut timers, &mut state, &mut watermark_gen);
            let outputs = operator.process_side(&order, JoinSide::Left, &mut ctx);
            assert_eq!(
                outputs
                    .iter()
                    .filter(|o| matches!(o, Output::Event(_)))
                    .count(),
                1
            );
        }

        assert_eq!(operator.metrics().matches, 1);
    }

    #[test]
    fn test_integer_key_join() {
        fn create_int_key_event(timestamp: i64, key: i64, value: i64) -> Event {
            let schema = Arc::new(Schema::new(vec![
                Field::new("key", DataType::Int64, false),
                Field::new("value", DataType::Int64, false),
            ]));
            let batch = RecordBatch::try_new(
                schema,
                vec![
                    Arc::new(Int64Array::from(vec![key])),
                    Arc::new(Int64Array::from(vec![value])),
                ],
            )
            .unwrap();
            Event::new(timestamp, batch)
        }

        let mut operator = StreamJoinOperator::with_id(
            "key".to_string(),
            "key".to_string(),
            Duration::from_secs(3600),
            JoinType::Inner,
            "test_join".to_string(),
        );

        let mut timers = TimerService::new();
        let mut state = InMemoryStore::new();
        let mut watermark_gen = BoundedOutOfOrdernessGenerator::new(100);

        // Process left with integer key
        let left = create_int_key_event(1000, 42, 100);
        {
            let mut ctx = create_test_context(&mut timers, &mut state, &mut watermark_gen);
            operator.process_side(&left, JoinSide::Left, &mut ctx);
        }

        // Process right with same integer key
        let right = create_int_key_event(1500, 42, 200);
        {
            let mut ctx = create_test_context(&mut timers, &mut state, &mut watermark_gen);
            let outputs = operator.process_side(&right, JoinSide::Right, &mut ctx);
            assert_eq!(
                outputs
                    .iter()
                    .filter(|o| matches!(o, Output::Event(_)))
                    .count(),
                1
            );
        }

        assert_eq!(operator.metrics().matches, 1);
    }

    // Stream Join Optimization Tests

    #[test]
    fn test_f057_join_row_encoding_enum() {
        assert_eq!(JoinRowEncoding::default(), JoinRowEncoding::Compact);
        assert_eq!(format!("{}", JoinRowEncoding::Compact), "compact");
        assert_eq!(format!("{}", JoinRowEncoding::CpuFriendly), "cpu_friendly");

        assert_eq!(
            "compact".parse::<JoinRowEncoding>().unwrap(),
            JoinRowEncoding::Compact
        );
        assert_eq!(
            "cpu_friendly".parse::<JoinRowEncoding>().unwrap(),
            JoinRowEncoding::CpuFriendly
        );
        assert_eq!(
            "cpu-friendly".parse::<JoinRowEncoding>().unwrap(),
            JoinRowEncoding::CpuFriendly
        );
        assert!("invalid".parse::<JoinRowEncoding>().is_err());
    }

    #[test]
    fn test_f057_config_builder() {
        let config = StreamJoinConfig::builder()
            .left_key_column("order_id")
            .right_key_column("payment_id")
            .time_bound(Duration::from_secs(3600))
            .join_type(JoinType::Left)
            .operator_id("test_join")
            .row_encoding(JoinRowEncoding::CpuFriendly)
            .asymmetric_compaction(true)
            .idle_threshold(Duration::from_secs(120))
            .per_key_tracking(true)
            .key_idle_threshold(Duration::from_secs(600))
            .build_side_pruning(true)
            .build_side(JoinSide::Left)
            .build()
            .unwrap();

        assert_eq!(config.left_key_column, "order_id");
        assert_eq!(config.right_key_column, "payment_id");
        assert_eq!(config.time_bound_ms, 3_600_000);
        assert_eq!(config.join_type, JoinType::Left);
        assert_eq!(config.operator_id, Some("test_join".to_string()));
        assert_eq!(config.row_encoding, JoinRowEncoding::CpuFriendly);
        assert!(config.asymmetric_compaction);
        assert_eq!(config.idle_threshold_ms, 120_000);
        assert!(config.per_key_tracking);
        assert_eq!(config.key_idle_threshold_ms, 600_000);
        assert!(config.build_side_pruning);
        assert_eq!(config.build_side, Some(JoinSide::Left));
    }

    #[test]
    fn test_f057_from_config() {
        let config = StreamJoinConfig::builder()
            .left_key_column("key")
            .right_key_column("key")
            .time_bound(Duration::from_secs(60))
            .join_type(JoinType::Inner)
            .row_encoding(JoinRowEncoding::CpuFriendly)
            .build()
            .unwrap();

        let operator = StreamJoinOperator::from_config(config);

        assert_eq!(operator.row_encoding(), JoinRowEncoding::CpuFriendly);
        assert!(operator.asymmetric_compaction_enabled());
        assert!(operator.per_key_tracking_enabled());
    }

    #[test]
    fn test_f057_cpu_friendly_encoding_roundtrip() {
        let schema = Arc::new(Schema::new(vec![
            Field::new("id", DataType::Utf8, false),
            Field::new("value", DataType::Int64, false),
            Field::new("price", DataType::Float64, false),
        ]));
        let batch = RecordBatch::try_new(
            schema,
            vec![
                Arc::new(StringArray::from(vec!["test_key"])),
                Arc::new(Int64Array::from(vec![42])),
                Arc::new(Float64Array::from(vec![99.99])),
            ],
        )
        .unwrap();

        // Test CPU-friendly encoding
        let row =
            JoinRow::with_encoding(1000, b"key".to_vec(), &batch, JoinRowEncoding::CpuFriendly)
                .unwrap();
        assert_eq!(row.encoding(), JoinRowEncoding::CpuFriendly);

        // Verify roundtrip
        let restored = row.to_batch().unwrap();
        assert_eq!(restored.num_rows(), 1);
        assert_eq!(restored.num_columns(), 3);

        // Verify values
        let id_col = restored
            .column(0)
            .as_any()
            .downcast_ref::<StringArray>()
            .unwrap();
        assert_eq!(id_col.value(0), "test_key");

        let value_col = restored
            .column(1)
            .as_any()
            .downcast_ref::<Int64Array>()
            .unwrap();
        assert_eq!(value_col.value(0), 42);

        let price_col = restored
            .column(2)
            .as_any()
            .downcast_ref::<Float64Array>()
            .unwrap();
        assert!((price_col.value(0) - 99.99).abs() < 0.001);
    }

    #[test]
    fn test_f057_compact_encoding_still_works() {
        let schema = Arc::new(Schema::new(vec![
            Field::new("id", DataType::Utf8, false),
            Field::new("value", DataType::Int64, false),
        ]));
        let batch = RecordBatch::try_new(
            schema,
            vec![
                Arc::new(StringArray::from(vec!["test"])),
                Arc::new(Int64Array::from(vec![100])),
            ],
        )
        .unwrap();

        let row = JoinRow::with_encoding(1000, b"key".to_vec(), &batch, JoinRowEncoding::Compact)
            .unwrap();
        assert_eq!(row.encoding(), JoinRowEncoding::Compact);

        let restored = row.to_batch().unwrap();
        assert_eq!(restored.num_rows(), 1);
    }

    #[test]
    fn test_f057_side_stats_tracking() {
        let mut stats = SideStats::new();
        assert_eq!(stats.events_received, 0);
        assert!(!stats.is_idle(1000, 60_000)); // No events yet, not idle

        // Record events
        stats.record_event(1000);
        assert_eq!(stats.events_received, 1);
        assert_eq!(stats.last_event_time, 1000);

        stats.record_event(2000);
        assert_eq!(stats.events_received, 2);
        assert_eq!(stats.last_event_time, 2000);

        // Check idle detection
        assert!(!stats.is_idle(2000, 60_000)); // Just received event
        assert!(!stats.is_idle(50_000, 60_000)); // Within threshold

        // After threshold with no new events in window
        stats.events_this_window = 0;
        assert!(stats.is_idle(100_000, 60_000)); // Past threshold
    }

    #[test]
    fn test_f057_key_metadata_tracking() {
        let mut meta = KeyMetadata::new(1000);
        assert_eq!(meta.last_activity, 1000);
        assert_eq!(meta.event_count, 1);
        assert_eq!(meta.state_entries, 1);

        meta.record_event(2000);
        assert_eq!(meta.last_activity, 2000);
        assert_eq!(meta.event_count, 2);
        assert_eq!(meta.state_entries, 2);

        meta.decrement_entries();
        assert_eq!(meta.state_entries, 1);

        assert!(!meta.is_idle(2000, 60_000));
        assert!(meta.is_idle(100_000, 60_000));
    }

    #[test]
    fn test_f057_per_key_tracking_in_operator() {
        let config = StreamJoinConfig::builder()
            .left_key_column("order_id")
            .right_key_column("order_id")
            .time_bound(Duration::from_secs(3600))
            .join_type(JoinType::Inner)
            .per_key_tracking(true)
            .build()
            .unwrap();

        let mut operator = StreamJoinOperator::from_config(config);
        let mut timers = TimerService::new();
        let mut state = InMemoryStore::new();
        let mut watermark_gen = BoundedOutOfOrdernessGenerator::new(100);

        // Process events with different keys
        for (i, key) in ["order_1", "order_2", "order_3"].iter().enumerate() {
            let event = create_order_event(1000 + i as i64 * 100, key, 100);
            let mut ctx = create_test_context(&mut timers, &mut state, &mut watermark_gen);
            operator.process_side(&event, JoinSide::Left, &mut ctx);
        }

        // Verify key tracking
        assert_eq!(operator.tracked_key_count(), 3);
        assert_eq!(operator.metrics().tracked_keys, 3);
    }

    #[test]
    fn test_f057_encoding_metrics() {
        // Test compact encoding
        let mut compact_op = StreamJoinOperator::from_config(
            StreamJoinConfig::builder()
                .left_key_column("order_id")
                .right_key_column("order_id")
                .time_bound(Duration::from_secs(3600))
                .join_type(JoinType::Inner)
                .row_encoding(JoinRowEncoding::Compact)
                .build()
                .unwrap(),
        );

        let mut timers = TimerService::new();
        let mut state = InMemoryStore::new();
        let mut watermark_gen = BoundedOutOfOrdernessGenerator::new(100);

        let event = create_order_event(1000, "order_1", 100);
        {
            let mut ctx = create_test_context(&mut timers, &mut state, &mut watermark_gen);
            compact_op.process_side(&event, JoinSide::Left, &mut ctx);
        }
        assert_eq!(compact_op.metrics().compact_encodes, 1);
        assert_eq!(compact_op.metrics().cpu_friendly_encodes, 0);

        // Test CPU-friendly encoding
        let mut cpu_op = StreamJoinOperator::from_config(
            StreamJoinConfig::builder()
                .left_key_column("order_id")
                .right_key_column("order_id")
                .time_bound(Duration::from_secs(3600))
                .join_type(JoinType::Inner)
                .row_encoding(JoinRowEncoding::CpuFriendly)
                .build()
                .unwrap(),
        );

        let mut state2 = InMemoryStore::new();
        {
            let mut ctx = create_test_context(&mut timers, &mut state2, &mut watermark_gen);
            cpu_op.process_side(&event, JoinSide::Left, &mut ctx);
        }
        assert_eq!(cpu_op.metrics().cpu_friendly_encodes, 1);
        assert_eq!(cpu_op.metrics().compact_encodes, 0);
    }

    #[test]
    fn test_f057_asymmetric_compaction_detection() {
        let config = StreamJoinConfig::builder()
            .left_key_column("order_id")
            .right_key_column("order_id")
            .time_bound(Duration::from_secs(60))
            .join_type(JoinType::Inner)
            .asymmetric_compaction(true)
            .idle_threshold(Duration::from_secs(10)) // 10 seconds
            .build()
            .unwrap();

        let mut operator = StreamJoinOperator::from_config(config);
        let mut timers = TimerService::new();
        let mut state = InMemoryStore::new();
        let mut watermark_gen = BoundedOutOfOrdernessGenerator::new(100);

        // Process left events
        for i in 0..5 {
            let event = create_order_event(1000 + i * 100, "order_1", 100);
            let mut ctx = create_test_context(&mut timers, &mut state, &mut watermark_gen);
            ctx.processing_time = 1000 + i * 100;
            operator.process_side(&event, JoinSide::Left, &mut ctx);
        }

        // Left side is not idle (just processed events)
        assert!(!operator.is_side_idle(JoinSide::Left, 1500));

        // Right side has no events - but is_idle returns false when no events received
        assert!(!operator.is_side_idle(JoinSide::Right, 1500));

        // Simulate time passing with no left events
        operator.left_stats.events_this_window = 0;
        assert!(operator.is_side_idle(JoinSide::Left, 100_000));
    }

    #[test]
    fn test_f057_effective_build_side_selection() {
        // Test configured build side
        let config = StreamJoinConfig::builder()
            .left_key_column("key")
            .right_key_column("key")
            .time_bound(Duration::from_secs(60))
            .join_type(JoinType::Inner)
            .build_side(JoinSide::Right)
            .build()
            .unwrap();

        let operator = StreamJoinOperator::from_config(config);
        assert_eq!(operator.effective_build_side(), JoinSide::Right);

        // Test auto-selection (smaller side)
        let config2 = StreamJoinConfig::builder()
            .left_key_column("key")
            .right_key_column("key")
            .time_bound(Duration::from_secs(60))
            .join_type(JoinType::Inner)
            .build()
            .unwrap();

        let mut operator2 = StreamJoinOperator::from_config(config2);
        operator2.left_stats.events_received = 100;
        operator2.right_stats.events_received = 1000;

        // Left is smaller, so should be build side
        assert_eq!(operator2.effective_build_side(), JoinSide::Left);
    }

    #[test]
    fn test_f057_join_with_cpu_friendly_encoding() {
        let config = StreamJoinConfig::builder()
            .left_key_column("order_id")
            .right_key_column("order_id")
            .time_bound(Duration::from_secs(3600))
            .join_type(JoinType::Inner)
            .row_encoding(JoinRowEncoding::CpuFriendly)
            .build()
            .unwrap();

        let mut operator = StreamJoinOperator::from_config(config);
        let mut timers = TimerService::new();
        let mut state = InMemoryStore::new();
        let mut watermark_gen = BoundedOutOfOrdernessGenerator::new(100);

        // Process left event
        let order = create_order_event(1000, "order_1", 100);
        {
            let mut ctx = create_test_context(&mut timers, &mut state, &mut watermark_gen);
            operator.process_side(&order, JoinSide::Left, &mut ctx);
        }

        // Process right event - should produce a match
        let payment = create_payment_event(2000, "order_1", "paid");
        {
            let mut ctx = create_test_context(&mut timers, &mut state, &mut watermark_gen);
            let outputs = operator.process_side(&payment, JoinSide::Right, &mut ctx);
            assert_eq!(
                outputs
                    .iter()
                    .filter(|o| matches!(o, Output::Event(_)))
                    .count(),
                1
            );
        }

        assert_eq!(operator.metrics().matches, 1);
        assert_eq!(operator.metrics().cpu_friendly_encodes, 2); // Both events encoded
    }

    #[test]
    fn test_f057_checkpoint_restore_with_optimization_state() {
        let config = StreamJoinConfig::builder()
            .left_key_column("key")
            .right_key_column("key")
            .time_bound(Duration::from_secs(60))
            .join_type(JoinType::Inner)
            .operator_id("test_join")
            .build()
            .unwrap();

        let mut operator = StreamJoinOperator::from_config(config);

        // Simulate activity
        operator.metrics.left_events = 100;
        operator.metrics.right_events = 50;
        operator.metrics.matches = 25;
        operator.metrics.cpu_friendly_encodes = 10;
        operator.metrics.compact_encodes = 140;
        operator.metrics.asymmetric_skips = 5;
        operator.metrics.idle_key_cleanups = 3;
        operator.metrics.build_side_prunes = 2;
        operator.left_stats.events_received = 100;
        operator.right_stats.events_received = 50;
        operator.left_watermark = 5000;
        operator.right_watermark = 4000;

        // Checkpoint
        let checkpoint = operator.checkpoint();

        // Restore
        let config2 = StreamJoinConfig::builder()
            .left_key_column("key")
            .right_key_column("key")
            .time_bound(Duration::from_secs(60))
            .join_type(JoinType::Inner)
            .operator_id("test_join")
            .build()
            .unwrap();

        let mut restored = StreamJoinOperator::from_config(config2);
        restored.restore(checkpoint).unwrap();

        // Verify optimization state was restored
        assert_eq!(restored.metrics().left_events, 100);
        assert_eq!(restored.metrics().right_events, 50);
        assert_eq!(restored.metrics().matches, 25);
        assert_eq!(restored.metrics().cpu_friendly_encodes, 10);
        assert_eq!(restored.metrics().compact_encodes, 140);
        assert_eq!(restored.metrics().asymmetric_skips, 5);
        assert_eq!(restored.metrics().idle_key_cleanups, 3);
        assert_eq!(restored.metrics().build_side_prunes, 2);
        assert_eq!(restored.left_stats.events_received, 100);
        assert_eq!(restored.right_stats.events_received, 50);
        assert_eq!(restored.left_watermark, 5000);
        assert_eq!(restored.right_watermark, 4000);
    }

    #[test]
    fn test_f057_should_skip_compaction() {
        let config = StreamJoinConfig::builder()
            .left_key_column("key")
            .right_key_column("key")
            .time_bound(Duration::from_secs(60))
            .join_type(JoinType::Inner)
            .asymmetric_compaction(true)
            .idle_threshold(Duration::from_secs(10))
            .build()
            .unwrap();

        let mut operator = StreamJoinOperator::from_config(config);

        // Record some left events
        operator.left_stats.record_event(1000);
        operator.left_stats.events_this_window = 0; // Simulate window rollover

        // Should skip compaction for idle left side
        assert!(operator.should_skip_compaction(JoinSide::Left, 100_000));

        // Should not skip when asymmetric compaction is disabled
        operator.asymmetric_compaction = false;
        assert!(!operator.should_skip_compaction(JoinSide::Left, 100_000));
    }

    #[test]
    fn test_f057_multiple_rows_cpu_friendly() {
        // Test with multiple values in arrays
        let schema = Arc::new(Schema::new(vec![
            Field::new("id", DataType::Int64, false),
            Field::new("name", DataType::Utf8, false),
        ]));

        // Single row (typical for streaming)
        let batch = RecordBatch::try_new(
            schema.clone(),
            vec![
                Arc::new(Int64Array::from(vec![1])),
                Arc::new(StringArray::from(vec!["Alice"])),
            ],
        )
        .unwrap();

        let row =
            JoinRow::with_encoding(1000, b"key".to_vec(), &batch, JoinRowEncoding::CpuFriendly)
                .unwrap();
        let restored = row.to_batch().unwrap();

        let id_col = restored
            .column(0)
            .as_any()
            .downcast_ref::<Int64Array>()
            .unwrap();
        let name_col = restored
            .column(1)
            .as_any()
            .downcast_ref::<StringArray>()
            .unwrap();

        assert_eq!(id_col.value(0), 1);
        assert_eq!(name_col.value(0), "Alice");
    }
}