Function asof_join_keyed

nodedb/src/engine/timeseries/asof_join.rs:76–129

ASOF join with multiple key columns. Groups both sides by key, then performs per-group ASOF join. `key_fn` extracts the grouping key from each row's data.

(
    left: &[TimestampedRow<L>],
    right: &[TimestampedRow<R>],
    tolerance_ms: i64,
    left_key: impl Fn(&L) -> K,
    right_key: impl Fn(&R) -> K,
)

/// Groups both sides by key, then performs per-group ASOF join.
/// `key_fn` extracts the grouping key from each row's data.
pub fn asof_join_keyed<L, R, K>(
    left: &[TimestampedRow<L>],
    right: &[TimestampedRow<R>],
    tolerance_ms: i64,
    left_key: impl Fn(&L) -> K,
    right_key: impl Fn(&R) -> K,
) -> Vec<AsofMatch<L, R>>
where
    L: Clone,
    R: Clone,
    K: Eq + std::hash::Hash + Clone,
{
    use std::collections::HashMap;

    // Group right side by key.
    let mut right_groups: HashMap<K, Vec<&TimestampedRow<R>>> = HashMap::new();
    for row in right {
        right_groups
            .entry(right_key(&row.data))
            .or_default()
            .push(row);
    }

    let mut results = Vec::with_capacity(left.len());

    for left_row in left {
        let key = left_key(&left_row.data);
        let matched = if let Some(group) = right_groups.get(&key) {
            // Binary search for largest timestamp <= left timestamp.
            let target = left_row.timestamp_ms;
            let pos = group.partition_point(|r| r.timestamp_ms <= target);
            if pos > 0 {
                let candidate = group[pos - 1];
                let gap = target - candidate.timestamp_ms;
                if tolerance_ms == i64::MAX || gap <= tolerance_ms {
                    Some(candidate.clone())
                } else {
                    None
                }
            } else {
                None
            }
        } else {
            None
        };

        results.push(AsofMatch {
            left: left_row.clone(),
            right: matched,
        });
    }

    results
}

#[cfg(test)]
mod tests {