Skip to main content

faucet_core/
join.rs

1//! Pure hash-join logic for the topology `join` node (issue #72).
2//!
3//! A [`HashJoin`] buffers one upstream (the **build** side) into an in-memory
4//! index keyed by a configurable dotted path, then enriches records streamed
5//! from the other upstream (the **probe** side) with projected fields looked
6//! up from the matching build record. This is the classic *hash-join* shape:
7//! the build side is fully materialized before the probe side starts emitting.
8//!
9//! This module is **pure data logic** — no I/O, no channels, no `async`. The
10//! topology executor ([`crate::topology`]) owns the streaming/channel plumbing
11//! and calls [`HashJoin::add_build_page`] / [`HashJoin::probe_page`] as pages
12//! arrive. Keeping it pure makes every join semantic unit-testable in
13//! isolation (see the extensive tests at the bottom of this file).
14//!
15//! ## Semantics
16//!
17//! - **`inner`** drops probe records with no build-side match.
18//! - **`left`** passes probe records through for non-matches, filling the
19//!   projected fields with [`JoinConfig::on_missing`].
20//! - **`on_duplicate`** decides what happens when one probe key matches more
21//!   than one build record: [`OnDuplicate::First`] keeps the first, and
22//!   [`OnDuplicate::Cartesian`] emits one enriched record per match.
23//! - **`on_collision`** decides what happens when a projected `as` name
24//!   already exists on the probe record.
25//! - **`key_normalize`** controls whether `"42"` (string) and `42` (number)
26//!   are treated as the same key ([`KeyNormalize::Stringify`]) or as distinct
27//!   keys ([`KeyNormalize::Preserve`], the default — no coercion).
28
29use crate::error::FaucetError;
30use serde::{Deserialize, Serialize};
31use serde_json::Value;
32use std::collections::HashMap;
33
34/// Default safety cap on build-side records (10M).
35pub const DEFAULT_MAX_BUILD_RECORDS: usize = 10_000_000;
36
37/// Join mode — how probe records with no build match are handled.
38#[derive(
39    Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize, schemars::JsonSchema,
40)]
41#[serde(rename_all = "lowercase")]
42pub enum JoinMode {
43    /// Drop probe records that have no matching build record.
44    #[default]
45    Inner,
46    /// Pass probe records through even without a match, filling projected
47    /// fields from [`JoinConfig::on_missing`].
48    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/// What to do when one probe key matches more than one build record.
61#[derive(
62    Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize, schemars::JsonSchema,
63)]
64#[serde(rename_all = "lowercase")]
65pub enum OnDuplicate {
66    /// Keep only the first build-side match (deterministic given build order).
67    #[default]
68    First,
69    /// Emit one enriched record per build-side match (may duplicate the probe
70    /// record).
71    Cartesian,
72}
73
74/// What to do when a projected `as` name collides with an existing field on
75/// the probe record.
76#[derive(
77    Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize, schemars::JsonSchema,
78)]
79#[serde(rename_all = "lowercase")]
80pub enum OnCollision {
81    /// Overwrite the probe record's field with the projected value.
82    #[default]
83    Overwrite,
84    /// Leave the probe record's field untouched; skip the projection.
85    Skip,
86    /// Fail the record with a typed error.
87    Error,
88}
89
90/// How to normalize keys before comparison.
91#[derive(
92    Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize, schemars::JsonSchema,
93)]
94#[serde(rename_all = "lowercase")]
95pub enum KeyNormalize {
96    /// Compare keys as their JSON value (no coercion): `"42"` != `42`.
97    #[default]
98    Preserve,
99    /// Coerce scalar keys to their string form before comparison so `"42"`
100    /// and `42` match.
101    Stringify,
102}
103
104/// A single field projection: copy `from` (a dotted path into the build
105/// record) onto the probe record under the name `as_`.
106#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
107pub struct Projection {
108    /// Dotted path inside each build (right) record.
109    pub from: String,
110    /// Output field name written onto the probe (left) record.
111    #[serde(rename = "as")]
112    pub as_: String,
113}
114
115/// Compiled join configuration. Dotted key paths are pre-split at construction.
116#[derive(Debug, Clone)]
117pub struct JoinConfig {
118    /// Inner vs left.
119    pub mode: JoinMode,
120    /// Dotted path to the build-side (right) key.
121    pub build_key: String,
122    /// Dotted path to the probe-side (left) key.
123    pub probe_key: String,
124    /// Fields to copy from the build record onto the probe record.
125    pub projections: Vec<Projection>,
126    /// Value used to fill projected fields on a `left`-mode non-match.
127    pub on_missing: Value,
128    /// Multi-match policy.
129    pub on_duplicate: OnDuplicate,
130    /// Projection-collision policy.
131    pub on_collision: OnCollision,
132    /// Key-normalization policy.
133    pub key_normalize: KeyNormalize,
134    /// Safety cap on build-side records.
135    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/// Running counters for a join, mirroring the `faucet_join_*` metrics.
155#[derive(Debug, Clone, Default, PartialEq, Eq)]
156pub struct JoinStats {
157    /// Records ingested by the build side (before null-key skipping).
158    pub build_records: u64,
159    /// Build records skipped because their key resolved to null/absent.
160    pub build_nulls: u64,
161    /// Distinct build keys that appeared more than once (duplicate keys).
162    pub duplicates: u64,
163    /// Records ingested by the probe side.
164    pub probe_records: u64,
165    /// Probe records that matched at least one build record.
166    pub matches: u64,
167    /// Probe records with no build-side match.
168    pub misses: u64,
169    /// Projections skipped because the `from` path was absent on the build
170    /// record.
171    pub project_misses: u64,
172}
173
174/// A hash join: build the index, then probe it.
175#[derive(Debug)]
176pub struct HashJoin {
177    config: JoinConfig,
178    /// Build index: canonical key → build records (in build order).
179    index: HashMap<String, Vec<Value>>,
180    stats: JoinStats,
181}
182
183impl HashJoin {
184    /// Create an empty join ready to receive build-side pages.
185    pub fn new(config: JoinConfig) -> Self {
186        Self {
187            config,
188            index: HashMap::new(),
189            stats: JoinStats::default(),
190        }
191    }
192
193    /// Access the running counters.
194    pub fn stats(&self) -> &JoinStats {
195        &self.stats
196    }
197
198    /// The compiled config.
199    pub fn config(&self) -> &JoinConfig {
200        &self.config
201    }
202
203    /// Ingest a page of build-side records into the index.
204    ///
205    /// Records whose build key resolves to null/absent are skipped (counted in
206    /// [`JoinStats::build_nulls`]). Returns [`FaucetError::Transform`] with a
207    /// `JoinBuildOverflow`-style message once the cumulative build count
208    /// exceeds [`JoinConfig::max_build_records`].
209    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                    // Non-scalar key under stringify, or otherwise unrepresentable.
229                    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    /// Number of distinct keys indexed so far.
243    pub fn indexed_keys(&self) -> usize {
244        self.index.len()
245    }
246
247    /// Enrich a page of probe-side records against the built index.
248    ///
249    /// Must be called only after every build page has been ingested. Returns
250    /// the enriched output records (0..N per input record depending on mode
251    /// and duplicate policy).
252    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                    // Clone the build records we need up front so we no longer
269                    // borrow `self.index` while mutating `self.stats`.
270                    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                    // inner mode: drop.
285                }
286            }
287        }
288        Ok(out)
289    }
290
291    /// Apply the configured projections onto `left`. `right = None` fills every
292    /// projected field with `on_missing` (the `left`-mode non-match path).
293    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                        // Field absent on the matched build record — skip it.
304                        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
316/// Resolve a dotted path against a JSON value, traversing objects only.
317///
318/// `"a.b.c"` descends three object levels. A missing key or a non-object
319/// intermediate yields `None`. An empty path yields the value itself.
320fn 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
334/// Set `left[name] = value` at the top level, applying the collision policy.
335fn 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
367/// Turn a JSON scalar into a canonical hash-map key string.
368///
369/// `Preserve` prefixes a type tag so `"42"` and `42` never collide. `Stringify`
370/// coerces scalars to their plain string form so they do. Non-scalar keys
371/// (objects/arrays) return `None` — they are not valid join keys.
372fn 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    // ── get_path ─────────────────────────────────────────────────────────────
423
424    #[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    // ── inner join ─────────────────────────────────────────────────────────────
450
451    #[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    // ── left join ─────────────────────────────────────────────────────────────
479
480    #[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    // ── duplicate build keys ────────────────────────────────────────────────────
521
522    #[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    // ── null keys ───────────────────────────────────────────────────────────────
556
557    #[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"}), // missing id
563            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    // ── key normalization ────────────────────────────────────────────────────────
591
592    #[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        // Probe with numeric 42 must NOT match the string "42".
600        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    // ── projection edge cases ─────────────────────────────────────────────────────
617
618    #[test]
619    fn projection_missing_source_field_is_skipped() {
620        let mut j = HashJoin::new(cfg());
621        // Build record lacks `signup_date`.
622        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    // ── overflow ─────────────────────────────────────────────────────────────────
682
683    #[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    // ── multi-page build ───────────────────────────────────────────────────────────
714
715    #[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        // A non-object probe record resolves to a null key (miss). Under inner
734        // mode it is simply dropped; under left mode we attempt to project onto
735        // it, which is where the "object-shaped records only" guard fires.
736        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        // inner mode: non-object → null key → miss → dropped, no error.
749        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}