(
&mut self,
peek_target: PeekTarget,
literal_constraints: Option<Vec<Row>>,
uuid: Uuid,
timestamp: Timestamp,
result_desc: RelationDesc,
finish
| 1734 | /// If this returns an error, then it didn't modify any `Instance` state. |
| 1735 | #[mz_ore::instrument(level = "debug")] |
| 1736 | pub fn peek( |
| 1737 | &mut self, |
| 1738 | peek_target: PeekTarget, |
| 1739 | literal_constraints: Option<Vec<Row>>, |
| 1740 | uuid: Uuid, |
| 1741 | timestamp: Timestamp, |
| 1742 | result_desc: RelationDesc, |
| 1743 | finishing: RowSetFinishing, |
| 1744 | map_filter_project: mz_expr::SafeMfpPlan, |
| 1745 | mut read_hold: ReadHold, |
| 1746 | target_replica: Option<ReplicaId>, |
| 1747 | peek_response_tx: oneshot::Sender<PeekResponse>, |
| 1748 | ) -> Result<(), PeekError> { |
| 1749 | use PeekError::*; |
| 1750 | |
| 1751 | let target_id = peek_target.id(); |
| 1752 | |
| 1753 | // Downgrade the provided read hold to the peek time. |
| 1754 | if read_hold.id() != target_id { |
| 1755 | return Err(ReadHoldIdMismatch(read_hold.id())); |
| 1756 | } |
| 1757 | read_hold |
| 1758 | .try_downgrade(Antichain::from_elem(timestamp.clone())) |
| 1759 | .map_err(|_| ReadHoldInsufficient(target_id))?; |
| 1760 | |
| 1761 | if let Some(target) = target_replica { |
| 1762 | if !self.replica_exists(target) { |
| 1763 | return Err(ReplicaMissing(target)); |
| 1764 | } |
| 1765 | } |
| 1766 | |
| 1767 | let otel_ctx = OpenTelemetryContext::obtain(); |
| 1768 | |
| 1769 | self.peeks.insert( |
| 1770 | uuid, |
| 1771 | PendingPeek { |
| 1772 | target_replica, |
| 1773 | // TODO(guswynn): can we just hold the `tracing::Span` here instead? |
| 1774 | otel_ctx: otel_ctx.clone(), |
| 1775 | requested_at: Instant::now(), |
| 1776 | read_hold, |
| 1777 | peek_response_tx, |
| 1778 | limit: finishing.limit.map(usize::cast_from), |
| 1779 | offset: finishing.offset, |
| 1780 | }, |
| 1781 | ); |
| 1782 | |
| 1783 | let peek = Peek { |
| 1784 | literal_constraints, |
| 1785 | uuid, |
| 1786 | timestamp, |
| 1787 | finishing, |
| 1788 | map_filter_project, |
| 1789 | // Obtain an `OpenTelemetryContext` from the thread-local tracing |
| 1790 | // tree to forward it on to the compute worker. |
| 1791 | otel_ctx, |
| 1792 | target: peek_target, |
| 1793 | result_desc, |
no test coverage detected