(
&mut self,
id: GlobalId,
new_description: ExportDescription,
)
| 1541 | } |
| 1542 | |
| 1543 | async fn alter_export( |
| 1544 | &mut self, |
| 1545 | id: GlobalId, |
| 1546 | new_description: ExportDescription, |
| 1547 | ) -> Result<(), StorageError> { |
| 1548 | let from_id = new_description.sink.from; |
| 1549 | |
| 1550 | // Acquire read holds at StorageCollections to ensure that the |
| 1551 | // sinked collection is not dropped while we're sinking it. |
| 1552 | let desired_read_holds = vec![from_id.clone(), id.clone()]; |
| 1553 | let [input_hold, self_hold] = self |
| 1554 | .storage_collections |
| 1555 | .acquire_read_holds(desired_read_holds) |
| 1556 | .expect("missing dependency") |
| 1557 | .try_into() |
| 1558 | .expect("expected number of holds"); |
| 1559 | let from_storage_metadata = self.storage_collections.collection_metadata(from_id)?; |
| 1560 | let to_storage_metadata = self.storage_collections.collection_metadata(id)?; |
| 1561 | |
| 1562 | // Check whether the sink's write frontier is beyond the read hold we got |
| 1563 | let cur_export = self.export_mut(id)?; |
| 1564 | let input_readable = cur_export |
| 1565 | .write_frontier |
| 1566 | .iter() |
| 1567 | .all(|t| input_hold.since().less_than(t)); |
| 1568 | if !input_readable { |
| 1569 | return Err(StorageError::ReadBeforeSince(from_id)); |
| 1570 | } |
| 1571 | |
| 1572 | let new_export = ExportState { |
| 1573 | read_capabilities: cur_export.read_capabilities.clone(), |
| 1574 | cluster_id: new_description.instance_id, |
| 1575 | derived_since: cur_export.derived_since.clone(), |
| 1576 | read_holds: [input_hold, self_hold], |
| 1577 | read_policy: cur_export.read_policy.clone(), |
| 1578 | write_frontier: cur_export.write_frontier.clone(), |
| 1579 | }; |
| 1580 | *cur_export = new_export; |
| 1581 | |
| 1582 | // For `ALTER SINK`, the snapshot should only occur if the sink has not made any progress. |
| 1583 | // This prevents unnecessary decoding in the sink. |
| 1584 | // If the write frontier of the sink is strictly larger than its read hold, it must have at |
| 1585 | // least written out its snapshot, and we can skip reading it; otherwise assume we may have |
| 1586 | // to replay from the beginning. |
| 1587 | // TODO(database-issues#10002): unify this with run_export, if possible |
| 1588 | let with_snapshot = new_description.sink.with_snapshot |
| 1589 | && !PartialOrder::less_than(&new_description.sink.as_of, &cur_export.write_frontier); |
| 1590 | |
| 1591 | let cmd = RunSinkCommand { |
| 1592 | id, |
| 1593 | description: StorageSinkDesc { |
| 1594 | from: from_id, |
| 1595 | from_desc: new_description.sink.from_desc, |
| 1596 | connection: new_description.sink.connection, |
| 1597 | envelope: new_description.sink.envelope, |
| 1598 | as_of: new_description.sink.as_of, |
| 1599 | version: new_description.sink.version, |
| 1600 | from_storage_metadata, |
no test coverage detected