qualia_core_db/sparql_library/
range_hash_join.rs

1//! Bounded range HashJoin. The smaller TripleScan is hashed into a
2//! caller-capped table; the larger side probes it. If the build side would
3//! exceed the table, the operator fails closed so the planner can keep
4//! NestedLoopJoin. No unbounded allocation.
5
6use super::range_join_select::Q42RangeJoinSelectPlan;
7use super::range_select_apply::{apply_select_wrappers, SelectWrapperState};
8use super::sparql_ast::{BindingRow, ExpressionId, SparqlQueryContext, VariableId, MAX_BINDINGS, MAX_VARIABLES};
9use super::sparql_executor::{
10    execute_range_triple_page_into, execute_range_volume_set_triple_page_into, Q42RangeNestedLoopJoinPage,
11    Q42RangeNestedLoopJoinPlan, Q42RangeSparqlCursor, Q42RangeTriplePattern,
12    Q42RangeVolumeSetSparqlCursor,
13};
14use super::sparql_planner::{ExecutionPlan, PhysicalOperatorType};
15use crate::NQuin;
16
17#[derive(Clone, Copy, Debug, Eq, PartialEq)]
18pub struct Q42RangeHashJoinSlot {
19    pub key: u64,
20    pub row: BindingRow,
21    pub occupied: bool,
22}
23
24impl Default for Q42RangeHashJoinSlot {
25    fn default() -> Self {
26        Self {
27            key: 0,
28            row: BindingRow::default(),
29            occupied: false,
30        }
31    }
32}
33
34#[derive(Clone, Copy, Debug, Eq, PartialEq)]
35pub struct Q42RangeHashJoinPlan {
36    pub left: Q42RangeTriplePattern,
37    pub right: Q42RangeTriplePattern,
38    pub join_var: VariableId,
39    pub projection: [VariableId; MAX_VARIABLES],
40    pub projection_count: u8,
41    pub filters: [ExpressionId; 8],
42    pub filter_count: u8,
43    pub limit: u64,
44    pub offset: u64,
45}
46
47impl Q42RangeHashJoinPlan {
48    pub fn as_join_select(&self) -> Q42RangeJoinSelectPlan {
49        Q42RangeJoinSelectPlan {
50            join: Q42RangeNestedLoopJoinPlan {
51                left: self.left,
52                right: self.right,
53            },
54            projection: self.projection,
55            projection_count: self.projection_count,
56            filters: self.filters,
57            filter_count: self.filter_count,
58            limit: self.limit,
59            offset: self.offset,
60        }
61    }
62
63    pub fn from_execution_plan(plan: &ExecutionPlan) -> Result<Self, String> {
64        if plan.operator_count == 0 || plan.root_operator as usize >= plan.operator_count as usize {
65            return Err("range HashJoin requires a non-empty execution plan".into());
66        }
67        let mut operator = plan.root_operator;
68        let mut projection = [0; MAX_VARIABLES];
69        let mut projection_count = 0;
70        let mut filters = [0; 8];
71        let mut filter_count = 0usize;
72        let mut limit = u64::MAX;
73        let mut offset = 0;
74        loop {
75            match plan.operators[operator as usize].operator_type {
76                PhysicalOperatorType::Project {
77                    input,
78                    vars,
79                    var_count,
80                } => {
81                    if projection_count != 0 {
82                        return Err("range HashJoin does not support nested projections".into());
83                    }
84                    projection = vars;
85                    projection_count = var_count;
86                    operator = input;
87                }
88                PhysicalOperatorType::Limit {
89                    input,
90                    limit: configured,
91                    offset: configured_offset,
92                } => {
93                    if limit != u64::MAX || offset != 0 {
94                        return Err("range HashJoin does not support nested limits".into());
95                    }
96                    limit = configured;
97                    offset = configured_offset;
98                    operator = input;
99                }
100                PhysicalOperatorType::Filter { input, expression } => {
101                    if filter_count == filters.len() {
102                        return Err("range HashJoin supports at most eight stacked FILTER operators".into());
103                    }
104                    filters[filter_count] = expression;
105                    filter_count += 1;
106                    operator = input;
107                }
108                PhysicalOperatorType::HashJoin {
109                    left,
110                    right,
111                    join_var,
112                }
113                | PhysicalOperatorType::NestedLoopJoin {
114                    left,
115                    right,
116                    join_var,
117                } => {
118                    return Ok(Self {
119                        left: triple_scan(plan, left)?,
120                        right: triple_scan(plan, right)?,
121                        join_var,
122                        projection,
123                        projection_count,
124                        filters,
125                        filter_count: filter_count as u8,
126                        limit,
127                        offset,
128                    });
129                }
130                _ => {
131                    return Err(
132                        "range HashJoin supports Project/Filter/Limit over HashJoin or NestedLoopJoin(TripleScan, TripleScan)"
133                            .into(),
134                    );
135                }
136            }
137        }
138    }
139}
140
141#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
142pub struct Q42RangeHashJoinState {
143    pub built: bool,
144    pub build_is_right: bool,
145    pub probe_scan: Q42RangeSparqlCursor,
146    pub probe_count: usize,
147    pub probe_index: usize,
148    pub probe_exhausted: bool,
149    pub match_offset: usize,
150    pub wrappers: SelectWrapperState,
151}
152
153#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
154pub struct Q42RangeVolumeSetHashJoinState {
155    pub built: bool,
156    pub build_is_right: bool,
157    pub probe_scan: Q42RangeVolumeSetSparqlCursor,
158    pub probe_count: usize,
159    pub probe_index: usize,
160    pub probe_exhausted: bool,
161    pub match_offset: usize,
162    pub wrappers: SelectWrapperState,
163}
164
165fn triple_scan(plan: &ExecutionPlan, operator: u16) -> Result<Q42RangeTriplePattern, String> {
166    let Some(entry) = plan.operators.get(operator as usize) else {
167        return Err("range HashJoin input operator is out of bounds".into());
168    };
169    match entry.operator_type {
170        PhysicalOperatorType::TripleScan {
171            subject,
172            predicate,
173            object,
174        } => Ok(Q42RangeTriplePattern {
175            subject,
176            predicate,
177            object,
178        }),
179        _ => Err("range HashJoin currently requires TripleScan inputs".into()),
180    }
181}
182
183fn cap_error(count: usize, cap: usize) -> String {
184    format!(
185        "range HashJoin build side ({count} rows) exceeds caller cap ({cap}); planner must keep NestedLoopJoin"
186    )
187}
188
189fn term_var(term: u64, ctx: &SparqlQueryContext) -> Option<VariableId> {
190    if (term as usize) < ctx.variable_count {
191        Some(term as VariableId)
192    } else {
193        None
194    }
195}
196
197fn pattern_vars(pattern: Q42RangeTriplePattern, ctx: &SparqlQueryContext) -> ([VariableId; 3], u8) {
198    let mut vars = [0; 3];
199    let mut count = 0u8;
200    for term in [pattern.subject, pattern.predicate, pattern.object] {
201        if let Some(var) = term_var(term, ctx) {
202            if !vars[..count as usize].contains(&var) {
203                vars[count as usize] = var;
204                count += 1;
205            }
206        }
207    }
208    (vars, count)
209}
210
211fn shared_join_vars(
212    left: Q42RangeTriplePattern,
213    right: Q42RangeTriplePattern,
214    ctx: &SparqlQueryContext,
215    join_var: VariableId,
216) -> ([VariableId; 3], u8) {
217    let (left_vars, left_n) = pattern_vars(left, ctx);
218    let (right_vars, right_n) = pattern_vars(right, ctx);
219    let mut shared = [0; 3];
220    let mut count = 0u8;
221    let left_slice = &left_vars[..left_n as usize];
222    let right_slice = &right_vars[..right_n as usize];
223    if left_slice.contains(&join_var) && right_slice.contains(&join_var) {
224        shared[0] = join_var;
225        count = 1;
226    }
227    for var in left_slice {
228        if right_slice.contains(var) && !shared[..count as usize].contains(var) && count < 3 {
229            shared[count as usize] = *var;
230            count += 1;
231        }
232    }
233    (shared, count)
234}
235
236fn join_key(row: &BindingRow, vars: &[VariableId]) -> u64 {
237    for var in vars {
238        if let Some(value) = row.get(*var) {
239            return value;
240        }
241    }
242    0
243}
244
245fn rows_compatible(left: &BindingRow, right: &BindingRow) -> bool {
246    for slot in 0..MAX_BINDINGS {
247        if let (Some(a), Some(b)) = (left.slots[slot], right.slots[slot]) {
248            if a != b {
249                return false;
250            }
251        }
252    }
253    true
254}
255
256fn merge_rows(left: &BindingRow, right: &BindingRow) -> BindingRow {
257    let mut joined = BindingRow::new();
258    for slot in 0..MAX_BINDINGS {
259        joined.slots[slot] = left.slots[slot].or(right.slots[slot]);
260    }
261    joined
262}
263
264fn insert_slot(table: &mut [Q42RangeHashJoinSlot], key: u64, row: BindingRow) -> Result<(), String> {
265    if table.is_empty() {
266        return Err(cap_error(1, 0));
267    }
268    let cap = table.len();
269    let start = (key as usize) % cap;
270    for step in 0..cap {
271        let slot = &mut table[(start + step) % cap];
272        if !slot.occupied {
273            slot.occupied = true;
274            slot.key = key;
275            slot.row = row;
276            return Ok(());
277        }
278    }
279    Err(cap_error(cap + 1, cap))
280}
281
282fn count_side<C, F>(
283    mut page_fn: F,
284    pattern: Q42RangeTriplePattern,
285    scratch: &mut [BindingRow],
286) -> Result<usize, String>
287where
288    C: Copy + Default,
289    F: FnMut(Q42RangeTriplePattern, C, &mut [BindingRow]) -> Result<(usize, Option<C>), String>,
290{
291    let mut cursor = C::default();
292    let mut total = 0usize;
293    loop {
294        let (count, next) = page_fn(pattern, cursor, scratch)?;
295        total = total.saturating_add(count);
296        match next {
297            Some(next_cursor) => cursor = next_cursor,
298            None => return Ok(total),
299        }
300    }
301}
302
303fn build_side<C, F>(
304    mut page_fn: F,
305    pattern: Q42RangeTriplePattern,
306    key_vars: &[VariableId],
307    table: &mut [Q42RangeHashJoinSlot],
308    scratch: &mut [BindingRow],
309) -> Result<usize, String>
310where
311    C: Copy + Default,
312    F: FnMut(Q42RangeTriplePattern, C, &mut [BindingRow]) -> Result<(usize, Option<C>), String>,
313{
314    let mut cursor = C::default();
315    let mut inserted = 0usize;
316    loop {
317        let (count, next) = page_fn(pattern, cursor, scratch)?;
318        for row in &scratch[..count] {
319            insert_slot(table, join_key(row, key_vars), *row)?;
320            inserted += 1;
321        }
322        match next {
323            Some(next_cursor) => cursor = next_cursor,
324            None => return Ok(inserted),
325        }
326    }
327}
328
329fn emit_matches(
330    table: &[Q42RangeHashJoinSlot],
331    probe: &BindingRow,
332    key_vars: &[VariableId],
333    match_offset: &mut usize,
334    out: &mut [BindingRow],
335) -> usize {
336    if table.is_empty() {
337        *match_offset = 0;
338        return 0;
339    }
340    let key = join_key(probe, key_vars);
341    let cap = table.len();
342    let start = (key as usize) % cap;
343    let mut returned = 0usize;
344    while *match_offset < cap && returned < out.len() {
345        let slot = &table[(start + *match_offset) % cap];
346        *match_offset += 1;
347        if !slot.occupied {
348            *match_offset = cap;
349            break;
350        }
351        if slot.key != key || !rows_compatible(probe, &slot.row) {
352            continue;
353        }
354        out[returned] = merge_rows(probe, &slot.row);
355        returned += 1;
356    }
357    returned
358}
359
360pub fn execute_range_hash_join_page_into<S: crate::q42_volume::Q42RangeSource>(
361    volume: &crate::q42_volume::Q42RangeVolume<S>,
362    plan: Q42RangeHashJoinPlan,
363    ctx: &SparqlQueryContext,
364    state: &mut Q42RangeHashJoinState,
365    table: &mut [Q42RangeHashJoinSlot],
366    compressed: &mut [u8],
367    decoded: &mut [u8],
368    quin_scratch: &mut [NQuin],
369    probe_rows: &mut [BindingRow],
370    join_out: &mut [BindingRow],
371    out: &mut [BindingRow],
372) -> Result<Q42RangeNestedLoopJoinPage, String> {
373    let mut page_fn = |pattern: Q42RangeTriplePattern, cursor: Q42RangeSparqlCursor, rows: &mut [BindingRow]| {
374        execute_range_triple_page_into(
375            volume,
376            pattern.subject,
377            pattern.predicate,
378            pattern.object,
379            None,
380            ctx,
381            &BindingRow::default(),
382            cursor,
383            compressed,
384            decoded,
385            quin_scratch,
386            rows,
387        )
388        .map(|page| (page.returned, page.next_cursor))
389    };
390    if !state.built {
391        let left_n = count_side(&mut page_fn, plan.left, probe_rows)?;
392        let right_n = count_side(&mut page_fn, plan.right, probe_rows)?;
393        let build_is_right = right_n <= left_n;
394        let build_n = if build_is_right { right_n } else { left_n };
395        if build_n > table.len() {
396            return Err(cap_error(build_n, table.len()));
397        }
398        let build_pattern = if build_is_right { plan.right } else { plan.left };
399        let (key_vars, key_n) = shared_join_vars(plan.left, plan.right, ctx, plan.join_var);
400        build_side(
401            &mut page_fn,
402            build_pattern,
403            &key_vars[..key_n as usize],
404            table,
405            probe_rows,
406        )?;
407        state.built = true;
408        state.build_is_right = build_is_right;
409        state.probe_scan = Q42RangeSparqlCursor::default();
410        state.probe_exhausted = false;
411        state.probe_count = 0;
412        state.probe_index = 0;
413        state.match_offset = 0;
414    }
415    let probe_pattern = if state.build_is_right {
416        plan.left
417    } else {
418        plan.right
419    };
420    let mut returned = 0usize;
421    loop {
422        if returned == out.len() {
423            return Ok(Q42RangeNestedLoopJoinPage {
424                returned,
425                done: false,
426            });
427        }
428        if state.wrappers.emitted >= plan.limit {
429            return Ok(Q42RangeNestedLoopJoinPage {
430                returned,
431                done: true,
432            });
433        }
434        if state.probe_index >= state.probe_count {
435            if state.probe_exhausted {
436                return Ok(Q42RangeNestedLoopJoinPage {
437                    returned,
438                    done: true,
439                });
440            }
441            let (count, next) = page_fn(probe_pattern, state.probe_scan, probe_rows)?;
442            state.probe_count = count;
443            state.probe_index = 0;
444            state.match_offset = 0;
445            state.probe_scan = next.unwrap_or_default();
446            state.probe_exhausted = next.is_none();
447            if state.probe_count == 0 {
448                if state.probe_exhausted {
449                    return Ok(Q42RangeNestedLoopJoinPage {
450                        returned,
451                        done: true,
452                    });
453                }
454                continue;
455            }
456        }
457        let probe = probe_rows[state.probe_index];
458        let (key_vars, key_n) = shared_join_vars(plan.left, plan.right, ctx, plan.join_var);
459        let produced = emit_matches(
460            table,
461            &probe,
462            &key_vars[..key_n as usize],
463            &mut state.match_offset,
464            join_out,
465        );
466        let applied = apply_select_wrappers(
467            ctx,
468            &plan.filters,
469            plan.filter_count,
470            plan.projection,
471            plan.projection_count,
472            plan.limit,
473            plan.offset,
474            &mut state.wrappers,
475            &join_out[..produced],
476            &mut out[returned..],
477        )?;
478        returned += applied.returned;
479        if state.match_offset >= table.len() || table.is_empty() {
480            state.probe_index += 1;
481            state.match_offset = 0;
482        }
483        if applied.limit_reached {
484            return Ok(Q42RangeNestedLoopJoinPage {
485                returned,
486                done: true,
487            });
488        }
489        if returned == out.len() {
490            return Ok(Q42RangeNestedLoopJoinPage {
491                returned,
492                done: false,
493            });
494        }
495    }
496}
497
498pub fn execute_range_volume_set_hash_join_page_into<S: crate::q42_volume::Q42RangeSource>(
499    volumes: &crate::q42_volume::Q42RangeVolumeSet<S>,
500    plan: Q42RangeHashJoinPlan,
501    ctx: &SparqlQueryContext,
502    state: &mut Q42RangeVolumeSetHashJoinState,
503    table: &mut [Q42RangeHashJoinSlot],
504    compressed: &mut [u8],
505    decoded: &mut [u8],
506    quin_scratch: &mut [NQuin],
507    probe_rows: &mut [BindingRow],
508    join_out: &mut [BindingRow],
509    out: &mut [BindingRow],
510) -> Result<Q42RangeNestedLoopJoinPage, String> {
511    let mut page_fn =
512        |pattern: Q42RangeTriplePattern, cursor: Q42RangeVolumeSetSparqlCursor, rows: &mut [BindingRow]| {
513            execute_range_volume_set_triple_page_into(
514                volumes,
515                pattern.subject,
516                pattern.predicate,
517                pattern.object,
518                None,
519                ctx,
520                &BindingRow::default(),
521                cursor,
522                compressed,
523                decoded,
524                quin_scratch,
525                rows,
526            )
527            .map(|page| (page.returned, page.next_cursor))
528        };
529    if !state.built {
530        let left_n = count_side(&mut page_fn, plan.left, probe_rows)?;
531        let right_n = count_side(&mut page_fn, plan.right, probe_rows)?;
532        let build_is_right = right_n <= left_n;
533        let build_n = if build_is_right { right_n } else { left_n };
534        if build_n > table.len() {
535            return Err(cap_error(build_n, table.len()));
536        }
537        let build_pattern = if build_is_right { plan.right } else { plan.left };
538        let (key_vars, key_n) = shared_join_vars(plan.left, plan.right, ctx, plan.join_var);
539        build_side(
540            &mut page_fn,
541            build_pattern,
542            &key_vars[..key_n as usize],
543            table,
544            probe_rows,
545        )?;
546        state.built = true;
547        state.build_is_right = build_is_right;
548        state.probe_scan = Q42RangeVolumeSetSparqlCursor::default();
549        state.probe_exhausted = false;
550        state.probe_count = 0;
551        state.probe_index = 0;
552        state.match_offset = 0;
553    }
554    let probe_pattern = if state.build_is_right {
555        plan.left
556    } else {
557        plan.right
558    };
559    let mut returned = 0usize;
560    loop {
561        if returned == out.len() {
562            return Ok(Q42RangeNestedLoopJoinPage {
563                returned,
564                done: false,
565            });
566        }
567        if state.wrappers.emitted >= plan.limit {
568            return Ok(Q42RangeNestedLoopJoinPage {
569                returned,
570                done: true,
571            });
572        }
573        if state.probe_index >= state.probe_count {
574            if state.probe_exhausted {
575                return Ok(Q42RangeNestedLoopJoinPage {
576                    returned,
577                    done: true,
578                });
579            }
580            let (count, next) = page_fn(probe_pattern, state.probe_scan, probe_rows)?;
581            state.probe_count = count;
582            state.probe_index = 0;
583            state.match_offset = 0;
584            state.probe_scan = next.unwrap_or_default();
585            state.probe_exhausted = next.is_none();
586            if state.probe_count == 0 {
587                if state.probe_exhausted {
588                    return Ok(Q42RangeNestedLoopJoinPage {
589                        returned,
590                        done: true,
591                    });
592                }
593                continue;
594            }
595        }
596        let probe = probe_rows[state.probe_index];
597        let (key_vars, key_n) = shared_join_vars(plan.left, plan.right, ctx, plan.join_var);
598        let produced = emit_matches(
599            table,
600            &probe,
601            &key_vars[..key_n as usize],
602            &mut state.match_offset,
603            join_out,
604        );
605        let applied = apply_select_wrappers(
606            ctx,
607            &plan.filters,
608            plan.filter_count,
609            plan.projection,
610            plan.projection_count,
611            plan.limit,
612            plan.offset,
613            &mut state.wrappers,
614            &join_out[..produced],
615            &mut out[returned..],
616        )?;
617        returned += applied.returned;
618        if state.match_offset >= table.len() || table.is_empty() {
619            state.probe_index += 1;
620            state.match_offset = 0;
621        }
622        if applied.limit_reached {
623            return Ok(Q42RangeNestedLoopJoinPage {
624                returned,
625                done: true,
626            });
627        }
628        if returned == out.len() {
629            return Ok(Q42RangeNestedLoopJoinPage {
630                returned,
631                done: false,
632            });
633        }
634    }
635}
636
637#[cfg(test)]
638mod tests {
639    use super::*;
640    use crate::q42_volume::{write_unified_volume, LocalFileRangeSource, Q42RangeVolume};
641
642    fn quin(s: u64, p: u64, o: u64) -> NQuin {
643        NQuin {
644            subject: s,
645            predicate: p,
646            object: o,
647            context: 0,
648            metadata: 0,
649            parity: 0,
650        }
651    }
652
653    fn test_volume(quins: &[NQuin]) -> (tempfile::TempDir, Q42RangeVolume<LocalFileRangeSource>) {
654        let dir = tempfile::TempDir::new().unwrap();
655        let path = dir.path().join("hash-join.q42");
656        let lo = quins.iter().map(|q| q.object).min().unwrap_or(0);
657        let hi = quins.iter().map(|q| q.object).max().unwrap_or(0);
658        write_unified_volume(
659            &path,
660            &std::collections::HashMap::new(),
661            &[(lo, hi)],
662            &[quins.to_vec()],
663        )
664        .unwrap();
665        let source = LocalFileRangeSource::open(&path).unwrap();
666        (dir, Q42RangeVolume::open(source).unwrap())
667    }
668
669    fn join_plan() -> Q42RangeHashJoinPlan {
670        Q42RangeHashJoinPlan {
671            left: Q42RangeTriplePattern {
672                subject: 0,
673                predicate: 20,
674                object: 1,
675            },
676            right: Q42RangeTriplePattern {
677                subject: 1,
678                predicate: 40,
679                object: 2,
680            },
681            join_var: 1,
682            projection: [0; MAX_VARIABLES],
683            projection_count: 0,
684            filters: [0; 8],
685            filter_count: 0,
686            limit: u64::MAX,
687            offset: 0,
688        }
689    }
690
691    #[test]
692    fn accepts_nested_loop_join_tree() {
693        let mut plan = ExecutionPlan::new();
694        let left = plan
695            .add_operator(
696                PhysicalOperatorType::TripleScan {
697                    subject: 0,
698                    predicate: 20,
699                    object: 1,
700                },
701                1,
702            )
703            .unwrap();
704        let right = plan
705            .add_operator(
706                PhysicalOperatorType::TripleScan {
707                    subject: 1,
708                    predicate: 40,
709                    object: 2,
710                },
711                1,
712            )
713            .unwrap();
714        plan.root_operator = plan
715            .add_operator(
716                PhysicalOperatorType::NestedLoopJoin {
717                    left,
718                    right,
719                    join_var: 1,
720                },
721                1,
722            )
723            .unwrap();
724        let compiled = Q42RangeHashJoinPlan::from_execution_plan(&plan).unwrap();
725        assert_eq!(compiled.join_var, 1);
726        assert_eq!(compiled.right.predicate, 40);
727    }
728
729    #[test]
730    fn hashes_smaller_side_and_joins() {
731        let left = quin(10, 20, 30);
732        let right = quin(30, 40, 50);
733        let extra = quin(31, 40, 51);
734        let (_dir, volume) = test_volume(&[left, right, extra]);
735        let mut ctx = SparqlQueryContext::new();
736        ctx.variable_count = 3;
737        let mut table = [Q42RangeHashJoinSlot::default(); 8];
738        let mut compressed = [0u8; crate::q42_volume::MAX_COMPRESSED_SUPERBLOCK_SIZE];
739        let mut decoded = [0u8; crate::q42_volume::SUPERBLOCK_SIZE];
740        let mut quins = [NQuin::default(); 4];
741        let mut probe_rows = [BindingRow::default(); 4];
742        let mut join_out = [BindingRow::default(); 4];
743        let mut out = [BindingRow::default(); 4];
744        let mut state = Q42RangeHashJoinState::default();
745        let page = execute_range_hash_join_page_into(
746            &volume,
747            join_plan(),
748            &ctx,
749            &mut state,
750            &mut table,
751            &mut compressed,
752            &mut decoded,
753            &mut quins,
754            &mut probe_rows,
755            &mut join_out,
756            &mut out,
757        )
758        .unwrap();
759        assert!(page.done);
760        assert_eq!(page.returned, 1);
761        assert_eq!(out[0].get(0), Some(10));
762        assert_eq!(out[0].get(1), Some(30));
763        assert_eq!(out[0].get(2), Some(50));
764        assert!(!state.build_is_right);
765    }
766
767    #[test]
768    fn cap_exceeded_tells_planner_to_keep_nested_loop() {
769        let left_a = quin(10, 20, 30);
770        let left_b = quin(11, 20, 32);
771        let right_a = quin(30, 40, 50);
772        let right_b = quin(32, 40, 51);
773        let (_dir, volume) = test_volume(&[left_a, left_b, right_a, right_b]);
774        let mut ctx = SparqlQueryContext::new();
775        ctx.variable_count = 3;
776        let mut table = [Q42RangeHashJoinSlot::default(); 1];
777        let mut compressed = [0u8; crate::q42_volume::MAX_COMPRESSED_SUPERBLOCK_SIZE];
778        let mut decoded = [0u8; crate::q42_volume::SUPERBLOCK_SIZE];
779        let mut quins = [NQuin::default(); 4];
780        let mut probe_rows = [BindingRow::default(); 4];
781        let mut join_out = [BindingRow::default(); 4];
782        let mut out = [BindingRow::default(); 4];
783        let mut state = Q42RangeHashJoinState::default();
784        let err = execute_range_hash_join_page_into(
785            &volume,
786            join_plan(),
787            &ctx,
788            &mut state,
789            &mut table,
790            &mut compressed,
791            &mut decoded,
792            &mut quins,
793            &mut probe_rows,
794            &mut join_out,
795            &mut out,
796        )
797        .unwrap_err();
798        assert!(
799            err.contains("planner must keep NestedLoopJoin"),
800            "unexpected error: {err}"
801        );
802        assert!(!state.built);
803    }
804
805    #[test]
806    fn empty_table_is_a_cap_error() {
807        let left = quin(10, 20, 30);
808        let right = quin(30, 40, 50);
809        let (_dir, volume) = test_volume(&[left, right]);
810        let mut ctx = SparqlQueryContext::new();
811        ctx.variable_count = 3;
812        let mut table = [];
813        let mut compressed = [0u8; crate::q42_volume::MAX_COMPRESSED_SUPERBLOCK_SIZE];
814        let mut decoded = [0u8; crate::q42_volume::SUPERBLOCK_SIZE];
815        let mut quins = [NQuin::default(); 2];
816        let mut probe_rows = [BindingRow::default(); 2];
817        let mut join_out = [BindingRow::default(); 2];
818        let mut out = [BindingRow::default(); 2];
819        let mut state = Q42RangeHashJoinState::default();
820        let err = execute_range_hash_join_page_into(
821            &volume,
822            join_plan(),
823            &ctx,
824            &mut state,
825            &mut table,
826            &mut compressed,
827            &mut decoded,
828            &mut quins,
829            &mut probe_rows,
830            &mut join_out,
831            &mut out,
832        )
833        .unwrap_err();
834        assert!(err.contains("NestedLoopJoin"));
835    }
836}
qualia_core_db/sparql_library/range_hash_join.rs

qualia_core_db/sparql_library/
range_hash_join.rs
