Load a persisted MV state snapshot.
(&self, tenant_id: u64, mv_name: &str)
| 109 | |
| 110 | /// Load a persisted MV state snapshot. |
| 111 | pub fn load(&self, tenant_id: u64, mv_name: &str) -> crate::Result<Option<MvSnapshot>> { |
| 112 | let key = format!("{tenant_id}:{mv_name}"); |
| 113 | let txn = self.db.begin_read().map_err(|e| crate::Error::Storage { |
| 114 | engine: "event_plane".into(), |
| 115 | detail: format!("begin_read: {e}"), |
| 116 | })?; |
| 117 | let table = txn |
| 118 | .open_table(MV_STATE) |
| 119 | .map_err(|e| crate::Error::Storage { |
| 120 | engine: "event_plane".into(), |
| 121 | detail: format!("open_table: {e}"), |
| 122 | })?; |
| 123 | |
| 124 | match table.get(key.as_str()) { |
| 125 | Ok(Some(guard)) => { |
| 126 | let bytes: &[u8] = guard.value(); |
| 127 | let snapshot: MvSnapshot = |
| 128 | zerompk::from_msgpack(bytes).map_err(|e| crate::Error::Serialization { |
| 129 | format: "msgpack".into(), |
| 130 | detail: format!("mv_state restore: {e}"), |
| 131 | })?; |
| 132 | Ok(Some(snapshot)) |
| 133 | } |
| 134 | Ok(None) => Ok(None), |
| 135 | Err(e) => Err(crate::Error::Storage { |
| 136 | engine: "event_plane".into(), |
| 137 | detail: format!("get mv_state: {e}"), |
| 138 | }), |
| 139 | } |
| 140 | } |
| 141 | |
| 142 | /// Delete persisted state for a dropped MV. |
| 143 | pub fn delete(&self, tenant_id: u64, mv_name: &str) -> crate::Result<()> { |