(
datadriven: &MachineState,
args: DirectiveArgs<'_>,
)
| 1708 | } |
| 1709 | |
| 1710 | pub async fn fetch_batch( |
| 1711 | datadriven: &MachineState, |
| 1712 | args: DirectiveArgs<'_>, |
| 1713 | ) -> Result<String, anyhow::Error> { |
| 1714 | let input = args.expect_str("input"); |
| 1715 | let stats = args.optional_str("stats"); |
| 1716 | let batch = datadriven.batches.get(input).expect("unknown batch"); |
| 1717 | |
| 1718 | let mut s = String::new(); |
| 1719 | let mut stream = pin!( |
| 1720 | batch |
| 1721 | .batch |
| 1722 | .part_stream( |
| 1723 | datadriven.shard_id, |
| 1724 | &*datadriven.state_versions.blob, |
| 1725 | &*datadriven.state_versions.metrics |
| 1726 | ) |
| 1727 | .enumerate() |
| 1728 | ); |
| 1729 | while let Some((idx, part)) = stream.next().await { |
| 1730 | let part = &*part?; |
| 1731 | write!(s, "<part {idx}>\n"); |
| 1732 | |
| 1733 | let lower = match part { |
| 1734 | BatchPart::Inline { updates, .. } => { |
| 1735 | let updates: BlobTraceBatchPart<u64> = |
| 1736 | updates.decode(&datadriven.client.metrics.columnar)?; |
| 1737 | updates.structured_key_lower() |
| 1738 | } |
| 1739 | other @ BatchPart::Hollow(_) => other.structured_key_lower(), |
| 1740 | }; |
| 1741 | |
| 1742 | if let Some(lower) = lower { |
| 1743 | if stats == Some("lower") { |
| 1744 | writeln!(s, "<key lower={}>", lower.get()) |
| 1745 | } |
| 1746 | } |
| 1747 | |
| 1748 | match part { |
| 1749 | BatchPart::Hollow(part) => { |
| 1750 | let blob_batch = datadriven |
| 1751 | .client |
| 1752 | .blob |
| 1753 | .get(&part.key.complete(&datadriven.shard_id)) |
| 1754 | .await; |
| 1755 | match blob_batch { |
| 1756 | Ok(Some(_)) | Err(_) => {} |
| 1757 | // don't try to fetch/print the keys of the batch part |
| 1758 | // if the blob store no longer has it |
| 1759 | Ok(None) => { |
| 1760 | s.push_str("<empty>\n"); |
| 1761 | continue; |
| 1762 | } |
| 1763 | }; |
| 1764 | } |
| 1765 | BatchPart::Inline { .. } => {} |
| 1766 | }; |
| 1767 | let part = EncodedPart::fetch( |
no test coverage detected