Core indexing logic: writes postings, doc length, and stats within a transaction. Bypasses the LSM memtable so Origin transactions can stay atomic with the document write.
(
&self,
txn: &WriteTransaction,
tid: TenantId,
collection: &str,
surrogate: Surrogate,
tokens: &[String],
)
| 61 | /// a transaction. Bypasses the LSM memtable so Origin transactions can |
| 62 | /// stay atomic with the document write. |
| 63 | fn write_index_data( |
| 64 | &self, |
| 65 | txn: &WriteTransaction, |
| 66 | tid: TenantId, |
| 67 | collection: &str, |
| 68 | surrogate: Surrogate, |
| 69 | tokens: &[String], |
| 70 | ) -> crate::Result<()> { |
| 71 | let t = tid.as_u64(); |
| 72 | |
| 73 | let mut term_postings: HashMap<&str, (u32, Vec<u32>)> = HashMap::new(); |
| 74 | for (pos, token) in tokens.iter().enumerate() { |
| 75 | let entry = term_postings |
| 76 | .entry(token.as_str()) |
| 77 | .or_insert((0, Vec::new())); |
| 78 | entry.0 += 1; |
| 79 | entry.1.push(pos as u32); |
| 80 | } |
| 81 | |
| 82 | let doc_len = tokens.len() as u32; |
| 83 | |
| 84 | let mut postings_table = txn |
| 85 | .open_table(POSTINGS) |
| 86 | .map_err(|e| inverted_err("open postings", e))?; |
| 87 | |
| 88 | for (term, (freq, positions)) in &term_postings { |
| 89 | let posting = Posting { |
| 90 | doc_id: surrogate, |
| 91 | term_freq: *freq, |
| 92 | positions: positions.clone(), |
| 93 | }; |
| 94 | |
| 95 | let mut existing: Vec<Posting> = postings_table |
| 96 | .get((t, collection, *term)) |
| 97 | .ok() |
| 98 | .flatten() |
| 99 | .and_then(|v| zerompk::from_msgpack(v.value()).ok()) |
| 100 | .unwrap_or_default(); |
| 101 | |
| 102 | existing.retain(|p| p.doc_id != surrogate); |
| 103 | existing.push(posting); |
| 104 | |
| 105 | let bytes = zerompk::to_msgpack_vec(&existing) |
| 106 | .map_err(|e| inverted_err("serialize postings", e))?; |
| 107 | postings_table |
| 108 | .insert((t, collection, *term), bytes.as_slice()) |
| 109 | .map_err(|e| inverted_err("insert posting", e))?; |
| 110 | } |
| 111 | drop(postings_table); |
| 112 | |
| 113 | let mut lengths = txn |
| 114 | .open_table(DOC_LENGTHS) |
| 115 | .map_err(|e| inverted_err("open doc_lengths", e))?; |
| 116 | let len_bytes = |
| 117 | zerompk::to_msgpack_vec(&doc_len).map_err(|e| inverted_err("serialize doc_len", e))?; |
| 118 | lengths |
| 119 | .insert((t, collection, surrogate.as_u32()), len_bytes.as_slice()) |
| 120 | .map_err(|e| inverted_err("insert doc_len", e))?; |