(
&mut self,
task: &ExecutionTask,
tid: u64,
collection: &str,
document_id: &str,
surrogate: Surrogate,
value: &[u8],
on_conflict_update
| 25 | /// tuples for existing docs, merge, and re-encode via `apply_point_put`. |
| 26 | #[allow(clippy::too_many_arguments)] |
| 27 | pub(in crate::data::executor) fn execute_upsert( |
| 28 | &mut self, |
| 29 | task: &ExecutionTask, |
| 30 | tid: u64, |
| 31 | collection: &str, |
| 32 | document_id: &str, |
| 33 | surrogate: Surrogate, |
| 34 | value: &[u8], |
| 35 | on_conflict_updates: &[(String, nodedb_physical::physical_plan::UpdateValue)], |
| 36 | ) -> Response { |
| 37 | let row_key = surrogate_to_doc_id(surrogate); |
| 38 | let row_key = row_key.as_str(); |
| 39 | debug!( |
| 40 | core = self.core_id, |
| 41 | %collection, |
| 42 | %document_id, |
| 43 | has_on_conflict = !on_conflict_updates.is_empty(), |
| 44 | "upsert" |
| 45 | ); |
| 46 | |
| 47 | // Detect strict storage mode for this collection. |
| 48 | let config_key = (crate::types::TenantId::new(tid), collection.to_string()); |
| 49 | let strict_schema = self.doc_configs.get(&config_key).and_then(|config| { |
| 50 | if let nodedb_physical::physical_plan::StorageMode::Strict { ref schema } = |
| 51 | config.storage_mode |
| 52 | { |
| 53 | Some(schema.clone()) |
| 54 | } else { |
| 55 | None |
| 56 | } |
| 57 | }); |
| 58 | |
| 59 | // Check if document already exists. Bitemporal collections consult |
| 60 | // the versioned table's current-state view (reverse-scan to newest |
| 61 | // non-tombstone); non-bitemporal collections use the legacy point |
| 62 | // lookup. |
| 63 | let bitemporal = self.is_bitemporal(tid, collection); |
| 64 | let existing = if bitemporal { |
| 65 | self.sparse.versioned_get_current(tid, collection, row_key) |
| 66 | } else { |
| 67 | self.sparse.get(tid, collection, row_key) |
| 68 | }; |
| 69 | |
| 70 | match existing { |
| 71 | Ok(Some(current_bytes)) => { |
| 72 | // Decode existing document to nodedb_types::Value. |
| 73 | let existing_val = if let Some(ref schema) = strict_schema { |
| 74 | // Strict: binary tuple → Value via schema. |
| 75 | match super::super::strict_format::binary_tuple_to_value(¤t_bytes, schema) |
| 76 | { |
| 77 | Some(v) => v, |
| 78 | None => { |
| 79 | // Fallback: try msgpack (migration case). |
| 80 | match nodedb_types::value_from_msgpack(¤t_bytes) { |
| 81 | Ok(v) => v, |
| 82 | Err(_) => { |
| 83 | return self.response_error( |
| 84 | task, |
no test coverage detected