1use crate::error::FaucetError;
30use serde::{Deserialize, Serialize};
31use serde_json::Value;
32use std::collections::HashMap;
33
34pub const DEFAULT_MAX_BUILD_RECORDS: usize = 10_000_000;
36
37#[derive(
39 Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize, schemars::JsonSchema,
40)]
41#[serde(rename_all = "lowercase")]
42pub enum JoinMode {
43 #[default]
45 Inner,
46 Left,
49}
50
51impl std::fmt::Display for JoinMode {
52 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
53 f.write_str(match self {
54 JoinMode::Inner => "inner",
55 JoinMode::Left => "left",
56 })
57 }
58}
59
60#[derive(
62 Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize, schemars::JsonSchema,
63)]
64#[serde(rename_all = "lowercase")]
65pub enum OnDuplicate {
66 #[default]
68 First,
69 Cartesian,
72}
73
74#[derive(
77 Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize, schemars::JsonSchema,
78)]
79#[serde(rename_all = "lowercase")]
80pub enum OnCollision {
81 #[default]
83 Overwrite,
84 Skip,
86 Error,
88}
89
90#[derive(
92 Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize, schemars::JsonSchema,
93)]
94#[serde(rename_all = "lowercase")]
95pub enum KeyNormalize {
96 #[default]
98 Preserve,
99 Stringify,
102}
103
104#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
107pub struct Projection {
108 pub from: String,
110 #[serde(rename = "as")]
112 pub as_: String,
113}
114
115#[derive(Debug, Clone)]
117pub struct JoinConfig {
118 pub mode: JoinMode,
120 pub build_key: String,
122 pub probe_key: String,
124 pub projections: Vec<Projection>,
126 pub on_missing: Value,
128 pub on_duplicate: OnDuplicate,
130 pub on_collision: OnCollision,
132 pub key_normalize: KeyNormalize,
134 pub max_build_records: usize,
136}
137
138impl Default for JoinConfig {
139 fn default() -> Self {
140 Self {
141 mode: JoinMode::default(),
142 build_key: String::new(),
143 probe_key: String::new(),
144 projections: Vec::new(),
145 on_missing: Value::Null,
146 on_duplicate: OnDuplicate::default(),
147 on_collision: OnCollision::default(),
148 key_normalize: KeyNormalize::default(),
149 max_build_records: DEFAULT_MAX_BUILD_RECORDS,
150 }
151 }
152}
153
154#[derive(Debug, Clone, Default, PartialEq, Eq)]
156pub struct JoinStats {
157 pub build_records: u64,
159 pub build_nulls: u64,
161 pub duplicates: u64,
163 pub probe_records: u64,
165 pub matches: u64,
167 pub misses: u64,
169 pub project_misses: u64,
172}
173
174#[derive(Debug)]
176pub struct HashJoin {
177 config: JoinConfig,
178 index: HashMap<String, Vec<Value>>,
180 stats: JoinStats,
181}
182
183impl HashJoin {
184 pub fn new(config: JoinConfig) -> Self {
186 Self {
187 config,
188 index: HashMap::new(),
189 stats: JoinStats::default(),
190 }
191 }
192
193 pub fn stats(&self) -> &JoinStats {
195 &self.stats
196 }
197
198 pub fn config(&self) -> &JoinConfig {
200 &self.config
201 }
202
203 pub fn add_build_page(&mut self, records: Vec<Value>) -> Result<(), FaucetError> {
210 for rec in records {
211 self.stats.build_records += 1;
212 if self.stats.build_records as usize > self.config.max_build_records {
213 return Err(FaucetError::Transform(format!(
214 "join build side exceeded max_build_records ({}); raise the limit or partition the join",
215 self.config.max_build_records
216 )));
217 }
218 let key = match get_path(&rec, &self.config.build_key) {
219 Some(v) if !v.is_null() => v,
220 _ => {
221 self.stats.build_nulls += 1;
222 continue;
223 }
224 };
225 let ckey = match canonical_key(key, self.config.key_normalize) {
226 Some(k) => k,
227 None => {
228 self.stats.build_nulls += 1;
230 continue;
231 }
232 };
233 let bucket = self.index.entry(ckey).or_default();
234 if !bucket.is_empty() {
235 self.stats.duplicates += 1;
236 }
237 bucket.push(rec);
238 }
239 Ok(())
240 }
241
242 pub fn indexed_keys(&self) -> usize {
244 self.index.len()
245 }
246
247 pub fn probe_page(&mut self, records: Vec<Value>) -> Result<Vec<Value>, FaucetError> {
253 let mut out = Vec::with_capacity(records.len());
254 for left in records {
255 self.stats.probe_records += 1;
256 let key = get_path(&left, &self.config.probe_key).filter(|v| !v.is_null());
257 let ckey = key.and_then(|k| canonical_key(k, self.config.key_normalize));
258
259 let matches: Option<&Vec<Value>> = ckey.as_ref().and_then(|k| self.index.get(k));
260
261 match matches {
262 Some(bucket) if !bucket.is_empty() => {
263 self.stats.matches += 1;
264 let take = match self.config.on_duplicate {
265 OnDuplicate::First => &bucket[..1],
266 OnDuplicate::Cartesian => &bucket[..],
267 };
268 let rights: Vec<Value> = take.to_vec();
271 for right in &rights {
272 let mut enriched = left.clone();
273 self.apply_projection(&mut enriched, Some(right))?;
274 out.push(enriched);
275 }
276 }
277 _ => {
278 self.stats.misses += 1;
279 if self.config.mode == JoinMode::Left {
280 let mut enriched = left;
281 self.apply_projection(&mut enriched, None)?;
282 out.push(enriched);
283 }
284 }
286 }
287 }
288 Ok(out)
289 }
290
291 fn apply_projection(
294 &mut self,
295 left: &mut Value,
296 right: Option<&Value>,
297 ) -> Result<(), FaucetError> {
298 for proj in &self.config.projections {
299 let value = match right {
300 Some(r) => match get_path(r, &proj.from) {
301 Some(v) => v.clone(),
302 None => {
303 self.stats.project_misses += 1;
305 continue;
306 }
307 },
308 None => self.config.on_missing.clone(),
309 };
310 set_field(left, &proj.as_, value, self.config.on_collision)?;
311 }
312 Ok(())
313 }
314}
315
316fn get_path<'a>(value: &'a Value, path: &str) -> Option<&'a Value> {
321 if path.is_empty() {
322 return Some(value);
323 }
324 let mut cur = value;
325 for seg in path.split('.') {
326 match cur {
327 Value::Object(map) => cur = map.get(seg)?,
328 _ => return None,
329 }
330 }
331 Some(cur)
332}
333
334fn set_field(
336 left: &mut Value,
337 name: &str,
338 value: Value,
339 on_collision: OnCollision,
340) -> Result<(), FaucetError> {
341 let obj = match left {
342 Value::Object(map) => map,
343 _ => {
344 return Err(FaucetError::Transform(
345 "join can only enrich object-shaped records".into(),
346 ));
347 }
348 };
349 if obj.contains_key(name) {
350 match on_collision {
351 OnCollision::Overwrite => {
352 obj.insert(name.to_string(), value);
353 }
354 OnCollision::Skip => {}
355 OnCollision::Error => {
356 return Err(FaucetError::Transform(format!(
357 "join projection '{name}' collides with an existing field (on_collision: error)"
358 )));
359 }
360 }
361 } else {
362 obj.insert(name.to_string(), value);
363 }
364 Ok(())
365}
366
367fn canonical_key(value: &Value, mode: KeyNormalize) -> Option<String> {
373 match mode {
374 KeyNormalize::Preserve => match value {
375 Value::String(s) => Some(format!("s:{s}")),
376 Value::Number(n) => Some(format!("n:{n}")),
377 Value::Bool(b) => Some(format!("b:{b}")),
378 Value::Null => None,
379 _ => None,
380 },
381 KeyNormalize::Stringify => match value {
382 Value::String(s) => Some(s.clone()),
383 Value::Number(n) => Some(n.to_string()),
384 Value::Bool(b) => Some(b.to_string()),
385 Value::Null => None,
386 _ => None,
387 },
388 }
389}
390
391#[cfg(test)]
392mod tests {
393 use super::*;
394 use serde_json::json;
395
396 fn cfg() -> JoinConfig {
397 JoinConfig {
398 build_key: "id".into(),
399 probe_key: "customer_id".into(),
400 projections: vec![
401 Projection {
402 from: "tier".into(),
403 as_: "customer_tier".into(),
404 },
405 Projection {
406 from: "signup_date".into(),
407 as_: "customer_signup_date".into(),
408 },
409 ],
410 ..Default::default()
411 }
412 }
413
414 fn build_customers(join: &mut HashJoin) {
415 join.add_build_page(vec![
416 json!({"id": 1, "tier": "gold", "signup_date": "2020-01-01"}),
417 json!({"id": 2, "tier": "silver", "signup_date": "2021-06-15"}),
418 ])
419 .unwrap();
420 }
421
422 #[test]
425 fn get_path_top_level() {
426 let v = json!({"a": 1});
427 assert_eq!(get_path(&v, "a"), Some(&json!(1)));
428 }
429
430 #[test]
431 fn get_path_nested() {
432 let v = json!({"a": {"b": {"c": 42}}});
433 assert_eq!(get_path(&v, "a.b.c"), Some(&json!(42)));
434 }
435
436 #[test]
437 fn get_path_missing() {
438 let v = json!({"a": 1});
439 assert_eq!(get_path(&v, "b"), None);
440 assert_eq!(get_path(&v, "a.b"), None);
441 }
442
443 #[test]
444 fn get_path_empty_returns_self() {
445 let v = json!({"a": 1});
446 assert_eq!(get_path(&v, ""), Some(&v));
447 }
448
449 #[test]
452 fn inner_match_enriches() {
453 let mut j = HashJoin::new(cfg());
454 build_customers(&mut j);
455 let out = j
456 .probe_page(vec![json!({"order": "A", "customer_id": 1})])
457 .unwrap();
458 assert_eq!(out.len(), 1);
459 assert_eq!(out[0]["customer_tier"], json!("gold"));
460 assert_eq!(out[0]["customer_signup_date"], json!("2020-01-01"));
461 assert_eq!(out[0]["order"], json!("A"));
462 assert_eq!(j.stats().matches, 1);
463 assert_eq!(j.stats().misses, 0);
464 }
465
466 #[test]
467 fn inner_no_match_drops() {
468 let mut j = HashJoin::new(cfg());
469 build_customers(&mut j);
470 let out = j
471 .probe_page(vec![json!({"order": "A", "customer_id": 999})])
472 .unwrap();
473 assert!(out.is_empty());
474 assert_eq!(j.stats().matches, 0);
475 assert_eq!(j.stats().misses, 1);
476 }
477
478 #[test]
481 fn left_match_enriches() {
482 let mut c = cfg();
483 c.mode = JoinMode::Left;
484 let mut j = HashJoin::new(c);
485 build_customers(&mut j);
486 let out = j
487 .probe_page(vec![json!({"order": "A", "customer_id": 2})])
488 .unwrap();
489 assert_eq!(out.len(), 1);
490 assert_eq!(out[0]["customer_tier"], json!("silver"));
491 }
492
493 #[test]
494 fn left_no_match_passes_through_with_on_missing() {
495 let mut c = cfg();
496 c.mode = JoinMode::Left;
497 c.on_missing = json!("UNKNOWN");
498 let mut j = HashJoin::new(c);
499 build_customers(&mut j);
500 let out = j
501 .probe_page(vec![json!({"order": "A", "customer_id": 999})])
502 .unwrap();
503 assert_eq!(out.len(), 1);
504 assert_eq!(out[0]["order"], json!("A"));
505 assert_eq!(out[0]["customer_tier"], json!("UNKNOWN"));
506 assert_eq!(out[0]["customer_signup_date"], json!("UNKNOWN"));
507 assert_eq!(j.stats().misses, 1);
508 }
509
510 #[test]
511 fn left_no_match_default_on_missing_is_null() {
512 let mut c = cfg();
513 c.mode = JoinMode::Left;
514 let mut j = HashJoin::new(c);
515 build_customers(&mut j);
516 let out = j.probe_page(vec![json!({"customer_id": 999})]).unwrap();
517 assert_eq!(out[0]["customer_tier"], Value::Null);
518 }
519
520 #[test]
523 fn duplicate_first_wins() {
524 let mut c = cfg();
525 c.on_duplicate = OnDuplicate::First;
526 let mut j = HashJoin::new(c);
527 j.add_build_page(vec![
528 json!({"id": 1, "tier": "gold", "signup_date": "d1"}),
529 json!({"id": 1, "tier": "platinum", "signup_date": "d2"}),
530 ])
531 .unwrap();
532 let out = j.probe_page(vec![json!({"customer_id": 1})]).unwrap();
533 assert_eq!(out.len(), 1);
534 assert_eq!(out[0]["customer_tier"], json!("gold"));
535 assert_eq!(j.stats().duplicates, 1);
536 }
537
538 #[test]
539 fn duplicate_cartesian_emits_all() {
540 let mut c = cfg();
541 c.on_duplicate = OnDuplicate::Cartesian;
542 let mut j = HashJoin::new(c);
543 j.add_build_page(vec![
544 json!({"id": 1, "tier": "gold", "signup_date": "d1"}),
545 json!({"id": 1, "tier": "platinum", "signup_date": "d2"}),
546 ])
547 .unwrap();
548 let out = j.probe_page(vec![json!({"customer_id": 1})]).unwrap();
549 assert_eq!(out.len(), 2);
550 assert_eq!(out[0]["customer_tier"], json!("gold"));
551 assert_eq!(out[1]["customer_tier"], json!("platinum"));
552 assert_eq!(j.stats().matches, 1);
553 }
554
555 #[test]
558 fn null_build_key_is_skipped() {
559 let mut j = HashJoin::new(cfg());
560 j.add_build_page(vec![
561 json!({"id": null, "tier": "gold"}),
562 json!({"tier": "silver"}), json!({"id": 3, "tier": "bronze", "signup_date": "d"}),
564 ])
565 .unwrap();
566 assert_eq!(j.stats().build_nulls, 2);
567 assert_eq!(j.indexed_keys(), 1);
568 }
569
570 #[test]
571 fn null_probe_key_inner_drops() {
572 let mut j = HashJoin::new(cfg());
573 build_customers(&mut j);
574 let out = j.probe_page(vec![json!({"order": "A"})]).unwrap();
575 assert!(out.is_empty());
576 assert_eq!(j.stats().misses, 1);
577 }
578
579 #[test]
580 fn null_probe_key_left_emits_on_missing() {
581 let mut c = cfg();
582 c.mode = JoinMode::Left;
583 let mut j = HashJoin::new(c);
584 build_customers(&mut j);
585 let out = j.probe_page(vec![json!({"order": "A"})]).unwrap();
586 assert_eq!(out.len(), 1);
587 assert_eq!(out[0]["customer_tier"], Value::Null);
588 }
589
590 #[test]
593 fn preserve_does_not_coerce_types() {
594 let mut c = cfg();
595 c.key_normalize = KeyNormalize::Preserve;
596 let mut j = HashJoin::new(c);
597 j.add_build_page(vec![json!({"id": "42", "tier": "str", "signup_date": "d"})])
598 .unwrap();
599 let out = j.probe_page(vec![json!({"customer_id": 42})]).unwrap();
601 assert!(out.is_empty());
602 }
603
604 #[test]
605 fn stringify_coerces_types() {
606 let mut c = cfg();
607 c.key_normalize = KeyNormalize::Stringify;
608 let mut j = HashJoin::new(c);
609 j.add_build_page(vec![json!({"id": "42", "tier": "str", "signup_date": "d"})])
610 .unwrap();
611 let out = j.probe_page(vec![json!({"customer_id": 42})]).unwrap();
612 assert_eq!(out.len(), 1);
613 assert_eq!(out[0]["customer_tier"], json!("str"));
614 }
615
616 #[test]
619 fn projection_missing_source_field_is_skipped() {
620 let mut j = HashJoin::new(cfg());
621 j.add_build_page(vec![json!({"id": 1, "tier": "gold"})])
623 .unwrap();
624 let out = j.probe_page(vec![json!({"customer_id": 1})]).unwrap();
625 assert_eq!(out[0]["customer_tier"], json!("gold"));
626 assert!(out[0].get("customer_signup_date").is_none());
627 assert_eq!(j.stats().project_misses, 1);
628 }
629
630 #[test]
631 fn projection_collision_overwrite() {
632 let mut c = cfg();
633 c.on_collision = OnCollision::Overwrite;
634 let mut j = HashJoin::new(c);
635 build_customers(&mut j);
636 let out = j
637 .probe_page(vec![json!({"customer_id": 1, "customer_tier": "OLD"})])
638 .unwrap();
639 assert_eq!(out[0]["customer_tier"], json!("gold"));
640 }
641
642 #[test]
643 fn projection_collision_skip() {
644 let mut c = cfg();
645 c.on_collision = OnCollision::Skip;
646 let mut j = HashJoin::new(c);
647 build_customers(&mut j);
648 let out = j
649 .probe_page(vec![json!({"customer_id": 1, "customer_tier": "OLD"})])
650 .unwrap();
651 assert_eq!(out[0]["customer_tier"], json!("OLD"));
652 }
653
654 #[test]
655 fn projection_collision_error() {
656 let mut c = cfg();
657 c.on_collision = OnCollision::Error;
658 let mut j = HashJoin::new(c);
659 build_customers(&mut j);
660 let err = j
661 .probe_page(vec![json!({"customer_id": 1, "customer_tier": "OLD"})])
662 .unwrap_err();
663 assert!(matches!(err, FaucetError::Transform(_)));
664 assert!(err.to_string().contains("collides"));
665 }
666
667 #[test]
668 fn nested_projection_path() {
669 let mut c = cfg();
670 c.projections = vec![Projection {
671 from: "profile.tier".into(),
672 as_: "tier".into(),
673 }];
674 let mut j = HashJoin::new(c);
675 j.add_build_page(vec![json!({"id": 1, "profile": {"tier": "gold"}})])
676 .unwrap();
677 let out = j.probe_page(vec![json!({"customer_id": 1})]).unwrap();
678 assert_eq!(out[0]["tier"], json!("gold"));
679 }
680
681 #[test]
684 fn build_overflow_errors() {
685 let mut c = cfg();
686 c.max_build_records = 2;
687 let mut j = HashJoin::new(c);
688 let err = j
689 .add_build_page(vec![
690 json!({"id": 1, "tier": "a", "signup_date": "d"}),
691 json!({"id": 2, "tier": "b", "signup_date": "d"}),
692 json!({"id": 3, "tier": "c", "signup_date": "d"}),
693 ])
694 .unwrap_err();
695 assert!(matches!(err, FaucetError::Transform(_)));
696 assert!(err.to_string().contains("max_build_records"));
697 }
698
699 #[test]
700 fn build_at_exactly_limit_ok() {
701 let mut c = cfg();
702 c.max_build_records = 2;
703 let mut j = HashJoin::new(c);
704 assert!(
705 j.add_build_page(vec![
706 json!({"id": 1, "tier": "a", "signup_date": "d"}),
707 json!({"id": 2, "tier": "b", "signup_date": "d"}),
708 ])
709 .is_ok()
710 );
711 }
712
713 #[test]
716 fn build_across_multiple_pages() {
717 let mut j = HashJoin::new(cfg());
718 j.add_build_page(vec![json!({"id": 1, "tier": "gold", "signup_date": "d1"})])
719 .unwrap();
720 j.add_build_page(vec![
721 json!({"id": 2, "tier": "silver", "signup_date": "d2"}),
722 ])
723 .unwrap();
724 assert_eq!(j.indexed_keys(), 2);
725 let out = j
726 .probe_page(vec![json!({"customer_id": 1}), json!({"customer_id": 2})])
727 .unwrap();
728 assert_eq!(out.len(), 2);
729 }
730
731 #[test]
732 fn non_object_record_errors() {
733 let mut c = cfg();
737 c.mode = JoinMode::Left;
738 let mut j = HashJoin::new(c);
739 build_customers(&mut j);
740 let err = j.probe_page(vec![json!(42)]).unwrap_err();
741 assert!(matches!(err, FaucetError::Transform(_)));
742 }
743
744 #[test]
745 fn non_object_probe_inner_is_dropped_not_errored() {
746 let mut j = HashJoin::new(cfg());
747 build_customers(&mut j);
748 let out = j.probe_page(vec![json!(42)]).unwrap();
750 assert!(out.is_empty());
751 assert_eq!(j.stats().misses, 1);
752 }
753
754 #[test]
755 fn bool_key_matches() {
756 let mut c = cfg();
757 c.build_key = "active".into();
758 c.probe_key = "is_active".into();
759 c.projections = vec![Projection {
760 from: "label".into(),
761 as_: "label".into(),
762 }];
763 let mut j = HashJoin::new(c);
764 j.add_build_page(vec![json!({"active": true, "label": "yes"})])
765 .unwrap();
766 let out = j.probe_page(vec![json!({"is_active": true})]).unwrap();
767 assert_eq!(out[0]["label"], json!("yes"));
768 }
769
770 #[test]
771 fn enums_serde_roundtrip() {
772 assert_eq!(serde_json::to_value(JoinMode::Left).unwrap(), json!("left"));
773 assert_eq!(
774 serde_json::from_value::<OnDuplicate>(json!("cartesian")).unwrap(),
775 OnDuplicate::Cartesian
776 );
777 assert_eq!(
778 serde_json::from_value::<OnCollision>(json!("skip")).unwrap(),
779 OnCollision::Skip
780 );
781 assert_eq!(
782 serde_json::from_value::<KeyNormalize>(json!("stringify")).unwrap(),
783 KeyNormalize::Stringify
784 );
785 }
786
787 #[test]
788 fn join_mode_display() {
789 assert_eq!(JoinMode::Inner.to_string(), "inner");
790 assert_eq!(JoinMode::Left.to_string(), "left");
791 }
792
793 #[test]
794 fn projection_serde_uses_as_rename() {
795 let p: Projection = serde_json::from_value(json!({"from": "a", "as": "b"})).unwrap();
796 assert_eq!(p.from, "a");
797 assert_eq!(p.as_, "b");
798 }
799}