| 2080 | |
| 2081 | #[instrument(level = "debug")] |
| 2082 | fn append_table( |
| 2083 | &mut self, |
| 2084 | write_ts: Timestamp, |
| 2085 | advance_to: Timestamp, |
| 2086 | commands: Vec<(GlobalId, Vec<TableData>)>, |
| 2087 | ) -> Result<tokio::sync::oneshot::Receiver<Result<(), StorageError>>, StorageError> { |
| 2088 | if self.read_only { |
| 2089 | // While in read only mode, ONLY collections that have been migrated |
| 2090 | // and need to be re-hydrated in read only mode can be written to. |
| 2091 | if !commands |
| 2092 | .iter() |
| 2093 | .all(|(id, _)| id.is_system() && self.migrated_storage_collections.contains(id)) |
| 2094 | { |
| 2095 | return Err(StorageError::ReadOnly); |
| 2096 | } |
| 2097 | } |
| 2098 | |
| 2099 | // TODO(petrosagg): validate appends against the expected RelationDesc of the collection |
| 2100 | for (id, updates) in commands.iter() { |
| 2101 | if !updates.is_empty() { |
| 2102 | if !write_ts.less_than(&advance_to) { |
| 2103 | return Err(StorageError::UpdateBeyondUpper(*id)); |
| 2104 | } |
| 2105 | } |
| 2106 | } |
| 2107 | |
| 2108 | Ok(self |
| 2109 | .persist_table_worker |
| 2110 | .append(write_ts, advance_to, commands)) |
| 2111 | } |
| 2112 | |
| 2113 | fn monotonic_appender(&self, id: GlobalId) -> Result<MonotonicAppender, StorageError> { |
| 2114 | self.collection_manager.monotonic_appender(id) |