spiral/
blob.rs

1//! Table-owned payloads uploaded independently of Arrow batch delivery.
2use std::ops::Range;
3
4use arrow_array::ArrayRef;
5use bytes::Bytes;
6use futures::stream::BoxStream;
7
8use crate::scandal_backend::AttemptWrite;
9use crate::{Error, Table};
10
11/// A bounded, backpressured byte stream. Producers should yield bounded chunks.
12pub type BlobStream = BoxStream<'static, Result<Bytes, Error>>;
13
14pub use crate::scandal::blob::{Blob, BlobPart, BlobStat, BlobUpload};
15
16/// Payloads to store as blobs of one write, packed together as the inline payloads of appended
17/// rows are. Push each payload, then [`finish`](Self::finish) to store them all.
18pub struct BlobPack<'a> {
19    write: &'a AttemptWrite,
20    payloads: Vec<(Bytes, Option<String>)>,
21}
22
23impl<'a> BlobPack<'a> {
24    pub(crate) fn new(write: &'a AttemptWrite) -> Self {
25        Self {
26            write,
27            payloads: Vec::new(),
28        }
29    }
30
31    /// Add a payload with its media type, and return the position of its blob in the column
32    /// [`finish`](Self::finish) returns.
33    pub fn push(&mut self, payload: impl Into<Bytes>, media_type: Option<&str>) -> usize {
34        self.payloads
35            .push((payload.into(), media_type.map(str::to_owned)));
36        self.payloads.len() - 1
37    }
38
39    /// Store every payload and return their blobs in push order, as a column of
40    /// [`Blob::field`]'s type that appended rows can carry. Nothing is stored before this.
41    pub async fn finish(self) -> Result<ArrayRef, Error> {
42        self.write.pack_blobs(self.payloads).await
43    }
44}
45
46impl Table {
47    /// Stream a dedicated payload into this Table's storage before publishing its row.
48    ///
49    /// Enforces `max_bytes` while consuming the stream. Failure returns no reference;
50    /// cancellation aborts an active upload when possible. A process crash may leave
51    /// an unreferenced object for retention cleanup. This does not publish a row.
52    pub async fn write_blob(
53        &self,
54        media_type: &str,
55        max_bytes: u64,
56        input: BlobStream,
57    ) -> Result<Blob, Error> {
58        self.connection
59            .tables()?
60            .write_blob(self.identity(), media_type, max_bytes, input)
61            .await
62    }
63
64    /// What storage holds for an owned payload, without reading its bytes.
65    pub async fn stat_blob(&self, blob: &Blob) -> Result<BlobStat, Error> {
66        if blob.table_id != self.identity().resource_id {
67            return Err(Error::InvalidConfiguration(
68                "blob belongs to a different Table".into(),
69            ));
70        }
71        self.connection
72            .tables()?
73            .stat_blob(self.identity(), blob)
74            .await
75    }
76
77    /// Stream an owned payload or range using this client's current read authorization.
78    /// A complete read verifies its digest; a range verifies its returned length.
79    pub async fn read_blob(
80        &self,
81        blob: &Blob,
82        range: Option<Range<u64>>,
83    ) -> Result<BlobStream, Error> {
84        if blob.table_id != self.identity().resource_id {
85            return Err(Error::InvalidConfiguration(
86                "blob belongs to a different Table".into(),
87            ));
88        }
89        if range
90            .as_ref()
91            .is_some_and(|r| r.start >= r.end || r.end > blob.size.0)
92        {
93            return Err(Error::InvalidConfiguration(
94                "blob range is outside its payload".into(),
95            ));
96        }
97        self.connection
98            .tables()?
99            .read_blob(self.identity(), blob, range)
100            .await
101    }
102}
spiral/blob.rs

spiral/
blob.rs
