spiral/
local.rs

1//! Tables stored in local directories, read and written without a control plane.
2//!
3//! The client resolves no root and registers no version: the caller keeps each Table's root and
4//! exact identity, and reopens it with [`Client::local_table`]. Reads, splits, blobs and
5//! evaluations then behave as they do for managed Tables. Local Tables are for tests, examples
6//! and notebooks over local files; they have no Catalog and no commits.
7
8use std::sync::Arc;
9
10use arrow_schema::SchemaRef;
11use reqwest::Url;
12use uuid::Uuid;
13
14use crate::client::Connection;
15use crate::scandal::Scandal;
16use crate::scandal_backend::AttemptWrite;
17use crate::{Blob, BlobPack, BlobStream, Client, Error, IntoArrowStream, Table, TableIdentity};
18
19/// The API endpoint of a client without a control plane. The reserved `.invalid` name never
20/// resolves, so resource operations fail instead of reaching a server.
21const NO_CONTROL_PLANE: &str = "http://control-plane.invalid/";
22
23impl Client {
24    /// A client without a control plane, for Tables in local `file://` directories.
25    ///
26    /// `scandal` must read local files, such as a `NativeScandal` whose configuration enables
27    /// `local_files`. Resource clients and path resolution fail: no API endpoint is configured.
28    pub fn local(scandal: Arc<dyn Scandal>) -> Result<Self, Error> {
29        Self::builder()
30            .api_url(NO_CONTROL_PLANE)
31            .scandal(scandal)
32            .build()
33    }
34
35    /// The exact version `identity` of the Table stored in the local directory `root`, a
36    /// `file://` URL ending in `/`. Any client with a scandal that reads local files can open
37    /// one; no control-plane request is made.
38    pub fn local_table(&self, root: &Url, identity: TableIdentity) -> Result<Table, Error> {
39        self.connection.tables()?.add_local_root(identity, root)?;
40        Ok(Table::new(Arc::clone(&self.connection), identity))
41    }
42
43    /// Begin the first version of a new Table in the local directory `root`, which holds only
44    /// this Table, with rows of `schema`.
45    pub fn create_local_table(
46        &self,
47        root: &Url,
48        schema: SchemaRef,
49    ) -> Result<LocalTableWrite, Error> {
50        let write = self
51            .connection
52            .tables()?
53            .begin_local_create(root, &schema)?;
54        Ok(LocalTableWrite {
55            connection: Arc::clone(&self.connection),
56            root: root.clone(),
57            schema,
58            write,
59        })
60    }
61}
62
63/// The first version of a local Table: blobs uploaded first, then the rows that carry them.
64/// Nothing is readable until [`Self::finish`].
65pub struct LocalTableWrite {
66    connection: Arc<Connection>,
67    root: Url,
68    schema: SchemaRef,
69    write: AttemptWrite,
70}
71
72impl LocalTableWrite {
73    /// The new Table's resource identifier.
74    pub fn resource_id(&self) -> Uuid {
75        self.write.table_id()
76    }
77
78    /// The schema every appended batch carries.
79    pub fn schema(&self) -> SchemaRef {
80        Arc::clone(&self.schema)
81    }
82
83    /// Upload `input` as one blob of this Table in parts, however large. Rows appended afterwards
84    /// carry the returned reference, encoded with [`Blob::array`].
85    pub async fn upload_blob(&self, media_type: &str, input: BlobStream) -> Result<Blob, Error> {
86        self.write.upload_blob(media_type, input).await
87    }
88
89    /// Begin a pack of payloads to store as blobs of this Table, packed together as the inline
90    /// payloads of appended rows are.
91    pub fn blob_pack(&self) -> BlobPack<'_> {
92        BlobPack::new(&self.write)
93    }
94
95    /// Append rows of this write's schema.
96    pub async fn append(&mut self, input: impl IntoArrowStream) -> Result<(), Error> {
97        let input = input.into_write_input()?;
98        if input.schema() != self.schema {
99            return Err(Error::invalid(
100                "appended rows differ from the local Table's schema",
101            ));
102        }
103        self.write.append(input).await
104    }
105
106    /// Prepare the version from every appended row and open it.
107    pub async fn finish(self) -> Result<Table, Error> {
108        let receipt = self.write.prepare().await?;
109        let identity = TableIdentity {
110            resource_id: receipt.resource_id,
111            version_id: receipt.version_id,
112        };
113        self.connection
114            .tables()?
115            .add_local_root(identity, &self.root)?;
116        Ok(Table::new(self.connection, identity))
117    }
118}
spiral/local.rs

spiral/
local.rs
