uqa-execution 0.4.9

Volcano physical operators with row-batch pipelines
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90

//
// Unified Query Algebra
//
// Copyright (c) 2023-2026 Cognica, Inc.
//

use std::cell::RefCell;

use super::*;
use uqa_storage::document_store::Document;

#[derive(Default)]
struct TextRecorder {
    batches: RefCell<Vec<(String, PreparedFtsDocuments)>>,
}

impl MutationTextIndex for TextRecorder {
    fn text_fields(
        &self,
        _table: &str,
        _document: &Document,
    ) -> Result<BTreeMap<String, String>, SQLError> {
        unreachable!("the test supplies already evaluated fields")
    }

    fn add_documents(&self, table: &str, documents: PreparedFtsDocuments) -> Result<(), SQLError> {
        self.batches.borrow_mut().push((table.into(), documents));
        Ok(())
    }
}

#[test]
fn repeated_identity_publishes_intermediate_text_before_replacement() {
    let text = TextRecorder::default();
    let mut batch = MutationPublicationBatch::default();
    let mut add = |table: &str, doc_id, value: &str| {
        batch.before_document(&text, table, doc_id).unwrap();
        batch.push_fts(
            table.into(),
            doc_id,
            BTreeMap::from([("body".into(), value.into())]),
        );
    };
    add("a", 9, "first");
    add("a", 2, "second");
    add("b", 9, "separate relation");
    assert!(text.batches.borrow().is_empty());
    add("a", 9, "replacement");
    assert_eq!(text.batches.borrow().len(), 2);
    batch.flush_fts(&text).unwrap();
    let published = text.batches.borrow();
    assert_eq!(published[0].0, "a");
    assert_eq!(published[0].1[0].1["body"], "first");
    assert_eq!(published[0].1[1].0, 2);
    assert_eq!(published[1].0, "b");
    assert_eq!(published[2].1[0].1["body"], "replacement");
}

#[test]
fn bounded_batches_retain_empty_replacements_and_reset_identity_tracking() {
    let text = TextRecorder::default();
    let mut batch = MutationPublicationBatch::default();
    for id in (0..PREPARED_FTS_BATCH_DOCUMENTS as u64).rev() {
        batch.before_document(&text, "docs", id).unwrap();
        batch.push_fts("docs".into(), id, BTreeMap::new());
    }
    assert!(batch.fts_is_full());
    assert!(text.batches.borrow().is_empty());
    batch.flush_fts(&text).unwrap();
    assert!(!batch.fts_is_full());
    batch.before_document(&text, "docs", 0).unwrap();
    batch.push_fts("docs".into(), 0, BTreeMap::new());
    batch.flush_fts(&text).unwrap();
    batch.flush_fts(&text).unwrap();
    let published = text.batches.borrow();
    assert_eq!(published.len(), 2);
    assert_eq!(published[0].1.len(), PREPARED_FTS_BATCH_DOCUMENTS);
    assert_eq!(published[1].1.len(), 1);
    assert!(published[0].1.iter().all(|(_, fields)| fields.is_empty()));
}

#[test]
fn a_new_document_without_indexed_text_leaves_the_text_index_alone() {
    let text = BTreeMap::from([("body".to_string(), "text".to_string())]);
    assert!(!document_changes_text_index(true, &BTreeMap::new()));
    assert!(document_changes_text_index(true, &text));
    // A document that may replace an earlier version removes that version's postings even when it has no text itself.
    assert!(document_changes_text_index(false, &BTreeMap::new()));
    assert!(document_changes_text_index(false, &text));
}