Index a document's field value in all applicable GSIs. Called on every PointPut. For each GSI declared on this collection, extracts the indexed field value and inserts/updates the GSI entry.
(
&self,
tenant_id: u64,
collection: &str,
document_id: &str,
shard_id: u16,
doc: &serde_json::Value,
indexes: &[GsiMeta],
)
| 159 | /// Called on every PointPut. For each GSI declared on this collection, |
| 160 | /// extracts the indexed field value and inserts/updates the GSI entry. |
| 161 | pub fn index_document( |
| 162 | &self, |
| 163 | tenant_id: u64, |
| 164 | collection: &str, |
| 165 | document_id: &str, |
| 166 | shard_id: u16, |
| 167 | doc: &serde_json::Value, |
| 168 | indexes: &[GsiMeta], |
| 169 | ) -> crate::Result<()> { |
| 170 | if indexes.is_empty() { |
| 171 | return Ok(()); |
| 172 | } |
| 173 | |
| 174 | let write_txn = self.db.begin_write().map_err(|e| crate::Error::Storage { |
| 175 | engine: "gsi".into(), |
| 176 | detail: format!("write: {e}"), |
| 177 | })?; |
| 178 | { |
| 179 | let mut table = write_txn |
| 180 | .open_table(GSI_TABLE) |
| 181 | .map_err(|e| crate::Error::Storage { |
| 182 | engine: "gsi".into(), |
| 183 | detail: format!("open entries: {e}"), |
| 184 | })?; |
| 185 | |
| 186 | for idx_meta in indexes { |
| 187 | let value = doc.get(&idx_meta.field).and_then(|v| match v { |
| 188 | serde_json::Value::String(s) => Some(s.clone()), |
| 189 | serde_json::Value::Number(n) => Some(n.to_string()), |
| 190 | serde_json::Value::Bool(b) => Some(b.to_string()), |
| 191 | _ => None, |
| 192 | }); |
| 193 | let Some(value_str) = value else { continue }; |
| 194 | |
| 195 | let key = format!("{}:{}", idx_meta.index_name, value_str); |
| 196 | let mut entries: Vec<GsiEntry> = match table.get(key.as_str()) { |
| 197 | Ok(Some(guard)) => match zerompk::from_msgpack(guard.value()) { |
| 198 | Ok(v) => v, |
| 199 | Err(e) => { |
| 200 | tracing::warn!(index = %key, error = %e, "GSI entry deserialization failed, starting fresh"); |
| 201 | Vec::new() |
| 202 | } |
| 203 | }, |
| 204 | _ => Vec::new(), |
| 205 | }; |
| 206 | |
| 207 | // Remove existing entry for this doc (update case). |
| 208 | entries.retain(|e| e.document_id != document_id); |
| 209 | entries.push(GsiEntry { |
| 210 | tenant_id, |
| 211 | collection: collection.to_string(), |
| 212 | document_id: document_id.to_string(), |
| 213 | shard_id, |
| 214 | }); |
| 215 | |
| 216 | let bytes = |
| 217 | zerompk::to_msgpack_vec(&entries).map_err(|e| crate::Error::Serialization { |
| 218 | format: "msgpack".into(), |