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