spiral/
table.rs

1//! Immutable table versions.
2
3use std::fmt::{self, Debug, Formatter};
4use std::sync::Arc;
5
6use uuid::Uuid;
7
8use crate::api::tasks::{
9    TableMaintenanceDefinition, TaskDefinition, TaskReadTarget, TaskReadTargetLatestOnBranch,
10};
11use crate::arrow_input::IntoArrowStream;
12use crate::catalog_types::{ProjectBranchId, ProjectCommitId, ProjectId, ResourceId};
13use crate::client::Connection;
14use crate::commit::BindingMutation;
15use crate::registration::register;
16use crate::scandal::types::{PreparedWriteReceipt, TableWriteResult};
17use crate::task_handle::submit;
18use crate::{
19    Error, Expression, Project, ResourceAddress, Run, Scan, ScanOptions, TaskHandle, WorkItem,
20};
21
22/// Stable identity of one immutable table version.
23#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
24pub struct TableIdentity {
25    /// Arbor resource identifier.
26    pub resource_id: Uuid,
27    /// Exact immutable version identifier.
28    pub version_id: Uuid,
29}
30
31/// An exact immutable table version.
32///
33/// Writing a table produces its next version without publishing it: nothing changes in the
34/// Catalog until a [`Commit`](crate::Commit) stages the new version and publishes.
35#[derive(Clone)]
36pub struct Table {
37    pub(crate) connection: Arc<Connection>,
38    identity: TableIdentity,
39    pub(crate) base: Option<CatalogBase>,
40    pub(crate) address: Option<ResourceAddress>,
41}
42
43/// The version a Catalog commit binds for a table's resource. Versions written from that
44/// table keep it, so a commit on that commit's branch can swap it for the newest one.
45#[derive(Clone)]
46pub(crate) struct CatalogBase {
47    pub(crate) project_id: ProjectId,
48    pub(crate) commit_id: ProjectCommitId,
49    /// The branch that commit was pinned from, or `None` for an exact-commit snapshot.
50    pub(crate) branch_id: Option<ProjectBranchId>,
51    pub(crate) version_id: Uuid,
52    /// The path that resolved the bound version, when it was opened by path.
53    pub(crate) address: Option<ResourceAddress>,
54}
55
56/// Rows deleted from a table, and the version without them.
57#[derive(Clone, Debug)]
58pub struct DeletedRows {
59    /// The version without the deleted rows; the original table when none matched.
60    pub table: Table,
61    /// How many visible rows were deleted.
62    pub rows: u64,
63}
64
65impl Debug for Table {
66    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
67        formatter
68            .debug_struct("Table")
69            .field("identity", &self.identity)
70            .finish()
71    }
72}
73
74impl Project {
75    /// Write the first version of a new table from `input`.
76    ///
77    /// The table has no name until [`Commit::create_table`](crate::Commit::create_table)
78    /// binds it and the commit publishes.
79    pub async fn write_table(&self, input: impl IntoArrowStream) -> Result<Table, Error> {
80        let result = self
81            .client
82            .connection
83            .tables()?
84            .create(self.id.as_str(), input.into_write_input()?)
85            .await?;
86        Ok(self
87            .client
88            .table_by_id(result.resource_id, result.version_id))
89    }
90}
91
92impl Table {
93    pub(crate) fn new(connection: Arc<Connection>, identity: TableIdentity) -> Self {
94        Self {
95            connection,
96            identity,
97            base: None,
98            address: None,
99        }
100    }
101
102    /// Exact immutable identity selected by this handle.
103    pub fn identity(&self) -> TableIdentity {
104        self.identity
105    }
106
107    /// Canonical `path@v<version>` provenance, when opened through a resource path.
108    /// Tables opened by IDs, and versions written from a table, have no resource path.
109    pub fn resolved_address(&self) -> Option<&ResourceAddress> {
110        self.address.as_ref()
111    }
112
113    /// Submit maintenance of this table's latest version on the branch it was opened from.
114    ///
115    /// Maintenance compacts and reclusters, publishing each step as a new version; its
116    /// result lists them. Returns `None` when maintenance of this table and branch is
117    /// already waiting to run.
118    pub async fn maintain(&self, run: Run) -> Result<Option<TaskHandle>, Error> {
119        let (project_id, branch_id) = self
120            .base
121            .as_ref()
122            .and_then(|base| Some((&base.project_id, base.branch_id?)))
123            .ok_or_else(|| Error::invalid("maintaining a table needs one opened from a branch"))?;
124        let definition = TaskDefinition::TableMaintenance(TableMaintenanceDefinition {
125            read_target: TaskReadTarget::LatestOnBranch(TaskReadTargetLatestOnBranch {
126                branch_id: branch_id.as_uuid(),
127            }),
128            table_resource_id: self.identity.resource_id,
129        });
130        submit(
131            &self.connection.tasks,
132            project_id.as_str().into(),
133            definition,
134            None,
135            run,
136        )
137        .await
138    }
139
140    /// Open a scan with fixed projection, predicate, schema, and execution binding.
141    ///
142    /// Metadata is resolved once. Batches are read when its streams are polled.
143    pub async fn scan(&self, options: ScanOptions) -> Result<Scan, Error> {
144        let mut scan = self
145            .connection
146            .tables()?
147            .open_scan(self.identity, options)
148            .await?;
149        scan.address = self.address.clone();
150        Ok(scan)
151    }
152
153    /// Write the next version of this table: its rows followed by `input`.
154    pub async fn append(&self, input: impl IntoArrowStream) -> Result<Table, Error> {
155        self.write(input, false).await
156    }
157
158    /// Write the next version of this table with only the rows of `input`.
159    pub async fn overwrite(&self, input: impl IntoArrowStream) -> Result<Table, Error> {
160        self.write(input, true).await
161    }
162
163    /// Write the next version of this table without the visible rows matching `filter`, or
164    /// without any rows when there is no filter.
165    ///
166    /// Retained rows keep their stable row IDs. When no row matches, nothing is written and
167    /// the result holds this table. Unsupported predicates fail before anything is written.
168    pub async fn delete_rows(&self, filter: Option<Expression>) -> Result<DeletedRows, Error> {
169        let scan = self
170            .scan(ScanOptions {
171                columns: Some(Vec::new()),
172                filter: filter.clone(),
173                ..Default::default()
174            })
175            .await?;
176        let work = scan
177            .splits()
178            .iter()
179            .enumerate()
180            .map(|(split, item)| WorkItem {
181                split,
182                rows: item.rows.clone(),
183            })
184            .collect::<Vec<_>>();
185        let rows = scan.count_rows(&work).await?.into_iter().sum();
186        if rows == 0 {
187            return Ok(DeletedRows {
188                table: self.clone(),
189                rows,
190            });
191        }
192        let result = self
193            .connection
194            .tables()?
195            .delete(self.identity, filter, rows)
196            .await?;
197        Ok(DeletedRows {
198            table: self.successor(result)?,
199            rows,
200        })
201    }
202
203    /// Register a version of this table that a write prepared but whose registration outcome
204    /// was unknown ([`Error::RegistrationUnknown`]), without uploading its rows again.
205    pub async fn recover_write(&self, receipt: PreparedWriteReceipt) -> Result<Table, Error> {
206        if receipt.resource_id != self.identity.resource_id
207            || receipt.base_version != Some(self.identity.version_id)
208        {
209            return Err(Error::invalid("receipt was not written from this table"));
210        }
211        let result = register(&self.connection.arbor, receipt).await?;
212        self.successor(result)
213    }
214
215    /// The binding that selected this table's Catalog base, to remove from the project root.
216    pub(crate) fn unbinding(&self) -> Result<BindingMutation, Error> {
217        let (parent, name) = self
218            .base
219            .as_ref()
220            .and_then(|base| base.address.as_ref())
221            .ok_or_else(|| Error::invalid("table was not opened by path"))?
222            .root_binding()?;
223        Ok(BindingMutation {
224            parent,
225            name,
226            expected: Some(ResourceId::new(self.identity.resource_id).map_err(Error::invalid)?),
227            replacement: None,
228        })
229    }
230
231    async fn write(&self, input: impl IntoArrowStream, overwrite: bool) -> Result<Self, Error> {
232        let result = self
233            .connection
234            .tables()?
235            .write(self.identity, input.into_write_input()?, overwrite)
236            .await?;
237        self.successor(result)
238    }
239
240    /// The version `result` wrote from this table, keeping this table's Catalog base.
241    pub(crate) fn successor(&self, result: TableWriteResult) -> Result<Self, Error> {
242        if result.resource_id != self.identity.resource_id {
243            return Err(Error::invalid("write changed resource identity"));
244        }
245        Ok(Self {
246            connection: Arc::clone(&self.connection),
247            identity: TableIdentity {
248                resource_id: result.resource_id,
249                version_id: result.version_id,
250            },
251            base: self.base.clone(),
252            address: None,
253        })
254    }
255}
spiral/table.rs

spiral/
table.rs
