uni_plugin/traits/
storage.rs

1//! Per-label plugin storage — `MATCH` / `CREATE` against a pluggable store.
2//!
3//! M5a (2026-05-24): the [`Storage`] trait is `#[async_trait]`. A plugin
4//! registers a [`Storage`] instance for a native label via
5//! [`crate::PluginRegistrar::label_storage`]; native-schema scans for that
6//! label are routed through the plugin's store instead of the host backend.
7
8use arrow_schema::SchemaRef;
9use async_trait::async_trait;
10use datafusion::arrow::record_batch::RecordBatch;
11use datafusion::execution::SendableRecordBatchStream;
12use datafusion::logical_expr::Expr;
13
14use crate::errors::FnError;
15
16/// Opaque write handle returned by [`Storage::write_batch`].
17#[derive(Clone, Debug)]
18pub struct WriteHandle {
19    /// Backend-specific identifier (LSN, transaction id, …).
20    pub id: u64,
21}
22
23/// Metadata returned by [`Storage::fork`] describing the newly-created branch.
24///
25/// `parent_version` is the backend version pinned as the fork-point, so
26/// callers orchestrating nested forks can chain `create_branch_from`-style
27/// calls without re-querying the backend. `branch_name` echoes the
28/// `dst_branch` argument, surfaced explicitly so backends with name
29/// canonicalization can return the resolved form.
30#[derive(Clone, Debug)]
31pub struct BranchMetadata {
32    /// Backend version pinned as the new branch's fork-point.
33    pub parent_version: u64,
34    /// Branch identifier as registered on the backend.
35    pub branch_name: String,
36}
37
38/// Per-instance storage interface.
39#[async_trait]
40pub trait Storage: Send + Sync {
41    /// Stream batches from `table` matching `predicate`.
42    ///
43    /// `predicate = None` means a full scan.
44    ///
45    /// # Errors
46    ///
47    /// Returns [`FnError`] if the read cannot start.
48    async fn read_batch(
49        &self,
50        table: &str,
51        predicate: Option<&Expr>,
52    ) -> Result<SendableRecordBatchStream, FnError>;
53
54    /// Write a single batch to `table`.
55    ///
56    /// # Errors
57    ///
58    /// Returns [`FnError`] on write failure.
59    async fn write_batch(&self, table: &str, batch: &RecordBatch) -> Result<WriteHandle, FnError>;
60
61    /// List tables known to this backend.
62    ///
63    /// # Errors
64    ///
65    /// Returns [`FnError`] if the listing cannot complete.
66    async fn list_tables(&self) -> Result<Vec<String>, FnError>;
67
68    /// Delete rows in `table` matching `predicate`. Returns the number of
69    /// rows actually deleted.
70    ///
71    /// # Errors
72    ///
73    /// Returns [`FnError`] on delete failure.
74    async fn delete(&self, table: &str, predicate: &Expr) -> Result<u64, FnError>;
75
76    /// Whether this backend supports branched / forked state.
77    fn supports_branching(&self) -> bool {
78        false
79    }
80
81    /// Fork `src_branch` of `table` into `dst_branch`. Default: unsupported.
82    ///
83    /// Granularity is per-dataset (`table`) because real branching backends
84    /// (Lance) track branches and versions independently per dataset.
85    /// Multi-dataset orchestration (atomic across all tables of a logical
86    /// fork) is the caller's responsibility.
87    ///
88    /// # Errors
89    ///
90    /// Returns [`FnError`] if branching is not supported or the fork
91    /// operation fails (missing source branch, name collision, I/O).
92    async fn fork(
93        &self,
94        _table: &str,
95        _src_branch: &str,
96        _dst_branch: &str,
97    ) -> Result<BranchMetadata, FnError> {
98        Err(FnError::new(
99            0x10,
100            "storage backend does not support branching",
101        ))
102    }
103
104    /// Backend-declared schema for `table`, if known.
105    async fn schema(&self, _table: &str) -> Option<SchemaRef> {
106        None
107    }
108}
uni_plugin/traits/storage.rs

uni_plugin/traits/
storage.rs
