use nodedb_types::result::QueryResult;
use crate::error::LiteError;
use crate::query::engine::LiteQueryEngine;
use crate::storage::engine::StorageEngine;
use super::common::{hash_join, maps_to_result, scan_collection};
pub async fn execute_shuffle_join<S: StorageEngine>(
engine: &LiteQueryEngine<S>,
left_collection: &str,
right_collection: &str,
on: &[(String, String)],
join_type: &str,
limit: usize,
_target_core: usize,
) -> Result<QueryResult, LiteError> {
let left_rows = scan_collection(engine, left_collection).await?;
let right_rows = scan_collection(engine, right_collection).await?;
let left_keys: Vec<String> = on.iter().map(|(l, _)| l.clone()).collect();
let right_keys: Vec<String> = on.iter().map(|(_, r)| r.clone()).collect();
let effective_limit = if limit == 0 { usize::MAX } else { limit };
let joined = hash_join(
right_rows,
left_rows,
&right_keys,
&left_keys,
join_type,
None,
effective_limit,
);
Ok(maps_to_result(joined))
}
#[cfg(test)]
mod tests {
use super::super::common::hash_join;
use nodedb_types::value::Value;
use std::collections::HashMap;
fn row(fields: &[(&str, Value)]) -> HashMap<String, Value> {
fields
.iter()
.map(|(k, v)| (k.to_string(), v.clone()))
.collect()
}
#[test]
fn shuffle_reduces_to_hash() {
let left = vec![row(&[
("id", Value::Integer(1)),
("val", Value::Integer(10)),
])];
let right = vec![row(&[
("lid", Value::Integer(1)),
("extra", Value::Integer(99)),
])];
let result = hash_join(
right,
left,
&["lid".into()],
&["id".into()],
"inner",
None,
usize::MAX,
);
assert_eq!(result.len(), 1);
assert_eq!(result[0]["val"], Value::Integer(10));
assert_eq!(result[0]["extra"], Value::Integer(99));
}
}