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
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
//! Tables stored in local directories, read and written without a control plane.
//!
//! The client resolves no root and registers no version: the caller keeps each Table's root and
//! exact identity, and reopens it with [`Client::local_table`]. Reads, splits, blobs and
//! evaluations then behave as they do for managed Tables. Local Tables are for tests, examples
//! and notebooks over local files; they have no Catalog and no commits.
use std::sync::Arc;
use arrow_schema::SchemaRef;
use reqwest::Url;
use uuid::Uuid;
use crate::client::Connection;
use crate::scandal::Scandal;
use crate::scandal_backend::AttemptWrite;
use crate::{Blob, BlobPack, BlobStream, Client, Error, IntoArrowStream, Table, TableIdentity};
/// The API endpoint of a client without a control plane. The reserved `.invalid` name never
/// resolves, so resource operations fail instead of reaching a server.
const NO_CONTROL_PLANE: &str = "http://control-plane.invalid/";
impl Client {
/// A client without a control plane, for Tables in local `file://` directories.
///
/// `scandal` must read local files, such as a `NativeScandal` whose configuration enables
/// `local_files`. Resource clients and path resolution fail: no API endpoint is configured.
pub fn local(scandal: Arc<dyn Scandal>) -> Result<Self, Error> {
Self::builder()
.api_url(NO_CONTROL_PLANE)
.scandal(scandal)
.build()
}
/// The exact version `identity` of the Table stored in the local directory `root`, a
/// `file://` URL ending in `/`. Any client with a scandal that reads local files can open
/// one; no control-plane request is made.
pub fn local_table(&self, root: &Url, identity: TableIdentity) -> Result<Table, Error> {
self.connection.tables()?.add_local_root(identity, root)?;
Ok(Table::new(Arc::clone(&self.connection), identity))
}
/// Begin the first version of a new Table in the local directory `root`, which holds only
/// this Table, with rows of `schema`.
pub fn create_local_table(
&self,
root: &Url,
schema: SchemaRef,
) -> Result<LocalTableWrite, Error> {
let write = self
.connection
.tables()?
.begin_local_create(root, &schema)?;
Ok(LocalTableWrite {
connection: Arc::clone(&self.connection),
root: root.clone(),
schema,
write,
})
}
}
/// The first version of a local Table: blobs uploaded first, then the rows that carry them.
/// Nothing is readable until [`Self::finish`].
pub struct LocalTableWrite {
connection: Arc<Connection>,
root: Url,
schema: SchemaRef,
write: AttemptWrite,
}
impl LocalTableWrite {
/// The new Table's resource identifier.
pub fn resource_id(&self) -> Uuid {
self.write.table_id()
}
/// The schema every appended batch carries.
pub fn schema(&self) -> SchemaRef {
Arc::clone(&self.schema)
}
/// Upload `input` as one blob of this Table in parts, however large. Rows appended afterwards
/// carry the returned reference, encoded with [`Blob::array`].
pub async fn upload_blob(&self, media_type: &str, input: BlobStream) -> Result<Blob, Error> {
self.write.upload_blob(media_type, input).await
}
/// Begin a pack of payloads to store as blobs of this Table, packed together as the inline
/// payloads of appended rows are.
pub fn blob_pack(&self) -> BlobPack<'_> {
BlobPack::new(&self.write)
}
/// Append rows of this write's schema.
pub async fn append(&mut self, input: impl IntoArrowStream) -> Result<(), Error> {
let input = input.into_write_input()?;
if input.schema() != self.schema {
return Err(Error::invalid(
"appended rows differ from the local Table's schema",
));
}
self.write.append(input).await
}
/// Prepare the version from every appended row and open it.
pub async fn finish(self) -> Result<Table, Error> {
let receipt = self.write.prepare().await?;
let identity = TableIdentity {
resource_id: receipt.resource_id,
version_id: receipt.version_id,
};
self.connection
.tables()?
.add_local_root(identity, &self.root)?;
Ok(Table::new(self.connection, identity))
}
}