spiral/scandal/
mod.rs

1//! The scandal API: one set of operations that embedded and remote scandal implement alike.
2//!
3//! Every request is self-contained. Hosts resolve names, obtain storage bindings and publish
4//! prepared versions; a scandal never calls the control plane.
5
6#[path = "../api/scandal/contract.rs"]
7pub mod api;
8pub mod blob;
9pub mod emit;
10pub mod error;
11pub mod expression;
12pub mod functions;
13pub mod ordered;
14pub mod plan;
15pub mod profile;
16pub mod read_cache;
17mod safelong;
18mod storage;
19pub mod types;
20pub mod vector;
21
22use std::borrow::Cow;
23use std::ops::Range;
24use std::time::{Duration, SystemTime, UNIX_EPOCH};
25
26use arrow_array::RecordBatch;
27use arrow_schema::{Schema, SchemaRef};
28use async_trait::async_trait;
29use bytes::Bytes;
30use futures::stream::BoxStream;
31use serde::{Deserialize, Serialize};
32pub use storage::{GcsSession, S3Session, StorageConfig};
33use url::Url;
34use uuid::Uuid;
35pub use vortex_array as vortex;
36
37use self::blob::{Blob, BlobPart, BlobStat, BlobUpload};
38pub use self::error::ScandalError;
39use self::expression::NamedExpression;
40use self::functions::FunctionInfo;
41pub use self::types::{
42    FilterClause, FragmentReceipt, PlannedSplit, PreparedWriteReceipt, RowRange, Scan, ScanInfo,
43    ScanMetricsSnapshot, ScanPlan, Split, SplitTake, TableVersion,
44};
45
46pub mod wire;
47
48/// The storage transports a scandal may use. Requests never choose an endpoint.
49#[derive(Clone, Debug, Default, Deserialize, Eq, PartialEq, Serialize)]
50#[serde(rename_all = "camelCase", deny_unknown_fields)]
51pub struct ScandalConfig {
52    /// The SPFS endpoint that signs requests for `spfsm://` and `spfsu://` roots whose binding
53    /// names none.
54    #[serde(default, skip_serializing_if = "Option::is_none")]
55    pub spfs_endpoint: Option<Url>,
56    /// Read `file://` roots. Requires the runtime's `local-files` feature; hosted scandal
57    /// leaves it off.
58    #[serde(default)]
59    pub local_files: bool,
60    /// Read `s3://` and `gs://` roots with the provider session the host resolved.
61    #[serde(default)]
62    pub provider_roots: bool,
63    /// Refuse an S3 session that names its own endpoint, so that only AWS and Cloud Storage
64    /// are reached, and a binding that names an SPFS endpoint other than `spfs_endpoint`.
65    /// Hosted scandal sets it: a caller's binding must not choose where the service connects.
66    #[serde(default)]
67    pub refuse_custom_endpoints: bool,
68}
69
70pub use self::api::{
71    AlignedBase, AlignedFragment, AppendRowsRequest, ArrowSchema, AttachColumnsRequest,
72    AttachedColumns, ChangePlan, DeleteRowsRequest, DirectRead, EditBase, GenesisBase,
73    PatchRowsRequest, PlanChangesRequest, ReadLegacyTableRequest, RowIdRange, SchemaChange,
74    SearchVectorsRequest, SortColumn, StorageBinding, StorageCredential, SuccessorBase,
75    TableBinding, WriteAlignedRequest, WriteBase, WriteProperties,
76};
77
78impl StorageBinding {
79    /// A logical resource reference. A resolver-enabled host obtains its scoped storage.
80    pub fn resource(resource: Uuid) -> Self {
81        Self::new(format!("spiral://{resource}/"))
82    }
83
84    /// Storage for `root_uri`, with no alias or credential, as local files need.
85    pub fn new(root_uri: impl Into<String>) -> Self {
86        Self {
87            root_uri: root_uri.into(),
88            root_alias: None,
89            credential: None,
90            endpoint: None,
91            expires_at: None,
92            direct_read: None,
93        }
94    }
95
96    /// How long the binding stays usable from now, until its `expires_at`, or zero once that
97    /// passed; `None` when it names no deadline.
98    pub fn remaining_lifetime(&self) -> Option<Duration> {
99        let expires_at = Duration::from_secs(self.expires_at?.0);
100        let now = SystemTime::now()
101            .duration_since(UNIX_EPOCH)
102            .unwrap_or_default();
103        Some(expires_at.saturating_sub(now))
104    }
105
106    /// The operation-scoped credential: the SPFS capability of an SPFS root, the JSON
107    /// [`S3Session`] of an `s3://` root, or the JSON
108    /// [`GcsSession`] of a `gs://` root; absent for local files.
109    pub fn credential(&self) -> Option<&str> {
110        self.credential
111            .as_ref()
112            .map(|credential| credential.0.as_str())
113    }
114}
115
116impl TableBinding {
117    /// Explicit storage, or a logical resource root for the host resolver.
118    pub fn binding(&self) -> Cow<'_, StorageBinding> {
119        self.storage.as_ref().map_or_else(
120            || Cow::Owned(StorageBinding::resource(self.table_id)),
121            Cow::Borrowed,
122        )
123    }
124}
125
126/// The host's URI resolution result: an authorized source prefix and the storage that reads it.
127/// Objects below `root_uri` use the same relative path below `storage.root_uri`.
128#[derive(Clone, Debug, Serialize, Deserialize)]
129pub struct ResolvedUri {
130    /// Absolute source directory URI, ending in `/`, with no query or fragment.
131    pub root_uri: Url,
132    /// Scoped physical storage and credentials. The runtime owns its managed I/O transport.
133    pub storage: StorageBinding,
134}
135
136/// The storage authority a host must obtain for one resource operation.
137#[derive(Clone, Copy, Debug, Eq, PartialEq, Hash, Serialize, Deserialize)]
138#[serde(
139    tag = "type",
140    rename_all = "camelCase",
141    rename_all_fields = "camelCase"
142)]
143pub enum ResourceAccess {
144    /// Read objects below the resource root.
145    Read,
146    /// Read the root and create objects only within one write attempt.
147    Attempt { attempt_id: Uuid },
148    /// Read the root and create version metadata and owned blobs.
149    Publication,
150}
151
152/// Host-configured URI policy, resource lookup and credentials, outside the execution runtime.
153///
154/// A resolver belongs to one authorization context. Cloned runtimes share its resolved roots
155/// until their credentials near expiry; different callers must receive separate resolvers.
156#[async_trait]
157pub trait UriResolver: Send + Sync {
158    /// Authorize `uri` and bind the source prefix containing it. The runtime reuses this
159    /// binding for later objects under that prefix, then asks again before credentials expire.
160    async fn resolve(&self, uri: &Url) -> Result<ResolvedUri, ScandalError>;
161
162    /// Resolve the resource for a particular operation. Read-only hosts reject write access.
163    async fn resolve_for(
164        &self,
165        uri: &Url,
166        access: ResourceAccess,
167    ) -> Result<ResolvedUri, ScandalError> {
168        if access != ResourceAccess::Read {
169            return Err(ScandalError::PermissionDenied {
170                operation: "write".into(),
171                resource: uri.to_string(),
172            });
173        }
174        self.resolve(uri).await
175    }
176}
177
178impl ArrowSchema {
179    /// The contract form of an Arrow schema.
180    pub fn from_schema(schema: &Schema) -> Result<Self, ScandalError> {
181        serde_json::to_value(schema)
182            .map(Self)
183            .map_err(|error| ScandalError::Internal {
184                message: error.to_string(),
185            })
186    }
187
188    /// The Arrow schema this value describes.
189    pub fn to_schema(&self) -> Result<Schema, ScandalError> {
190        serde_json::from_value(self.0.clone()).map_err(|error| ScandalError::InvalidArgument {
191            message: format!("invalid Arrow schema: {error}"),
192        })
193    }
194}
195
196impl WriteBase {
197    /// The Table's first version, with the schema its appends used.
198    pub fn genesis(schema: &Schema) -> Result<Self, ScandalError> {
199        Ok(Self::Genesis(GenesisBase {
200            schema: ArrowSchema::from_schema(schema)?,
201        }))
202    }
203
204    /// A successor of `version_id` that keeps its rows.
205    pub fn append(version_id: Uuid) -> Self {
206        Self::Append(SuccessorBase { version_id })
207    }
208
209    /// A successor of `version_id` that replaces its rows.
210    pub fn overwrite(version_id: Uuid) -> Self {
211        Self::Overwrite(SuccessorBase { version_id })
212    }
213
214    /// The version this write builds on, absent for the Table's first version.
215    pub fn base_version(&self) -> Result<Option<Uuid>, ScandalError> {
216        match self {
217            Self::Genesis(_) => Ok(None),
218            Self::Append(base) | Self::Overwrite(base) => Ok(Some(base.version_id)),
219            Self::Delete(base) => Ok(Some(base.version_id)),
220            Self::Edit(base) => Ok(Some(base.version_id)),
221            Self::Aligned(base) => Ok(base.version_id),
222            Self::Unknown { tag, .. } => Err(ScandalError::InvalidArgument {
223                message: format!("unknown write base `{tag}`"),
224            }),
225        }
226    }
227}
228
229/// One Arrow batch of a split and the stored row of its fragment that the read reached.
230#[derive(Clone, Debug)]
231pub struct SplitBatch {
232    pub batch: RecordBatch,
233    /// The stored row after the rows this batch covered; resume the split from here. For a
234    /// take, the request's positions delivered so far; resume with the positions after them.
235    pub next_row: u64,
236    /// The counters that the split read accrued since its previous batch, when the scan
237    /// collects metrics. A last batch of no rows carries the rest. Pruning counters are the
238    /// plan's, never a split's.
239    pub metrics: Option<ScanMetricsSnapshot>,
240}
241
242pub type SplitStream = BoxStream<'static, Result<SplitBatch, ScandalError>>;
243pub type RecordBatchStream = BoxStream<'static, Result<RecordBatch, ScandalError>>;
244pub type ByteStream = BoxStream<'static, Result<Bytes, ScandalError>>;
245
246/// A split read: its output schema, then its batches in stored order.
247pub struct SplitRead {
248    pub schema: SchemaRef,
249    pub batches: SplitStream,
250}
251
252/// An evaluation's output schema, then its batches, row for row with the input.
253pub struct Evaluation {
254    pub schema: SchemaRef,
255    pub batches: RecordBatchStream,
256}
257
258/// Scans, writes, blob transfer and SPQL evaluation over exact table versions.
259///
260/// Embedded and remote scandal implement every operation with the same semantics.
261#[async_trait]
262pub trait Scandal: Send + Sync {
263    /// Whether this host resolves logical resource roots internally, including write scopes.
264    fn resolves_resources(&self) -> bool {
265        false
266    }
267
268    /// Read a bounded legacy window from host-resolved metadata and scoped storage as Arrow IPC.
269    async fn read_legacy_table(
270        &self,
271        _request: &ReadLegacyTableRequest,
272    ) -> Result<ByteStream, ScandalError> {
273        Err(ScandalError::InvalidArgument {
274            message: "Legacy reads are unavailable on this implementation.".to_owned(),
275        })
276    }
277
278    /// The SPQL functions this scandal can run.
279    async fn list_functions(&self) -> Result<Vec<FunctionInfo>, ScandalError>;
280
281    /// Search an exact index and refine candidates from its pinned source version.
282    /// Stream release stops further payload reads.
283    async fn search_vectors(
284        &self,
285        _request: &SearchVectorsRequest,
286    ) -> Result<Evaluation, ScandalError> {
287        Err(unsupported("vector search"))
288    }
289
290    /// Plan a scan: its output schema, its filter support, and one split per fragment that
291    /// survives pruning by statistics, in stored order.
292    async fn plan_scan(
293        &self,
294        scan: &Scan,
295        storage: &StorageBinding,
296    ) -> Result<ScanPlan, ScandalError>;
297
298    /// Read one split in stored order. Each batch reports the stored row it reached.
299    async fn read_scan(
300        &self,
301        scan: &Scan,
302        split: &Split,
303        storage: &StorageBinding,
304    ) -> Result<SplitRead, ScandalError>;
305
306    /// Read rows of planned splits by position, in request order: each take's rows in the order
307    /// of its positions, a repeated position's row again, the takes one after another. Positions
308    /// count the rows the version keeps in each split. A batch's
309    /// [`next_row`](SplitBatch::next_row) is how many of the request's positions, across its
310    /// takes, the batches so far delivered.
311    async fn take_rows(
312        &self,
313        scan: &Scan,
314        takes: &[SplitTake],
315        storage: &StorageBinding,
316    ) -> Result<SplitRead, ScandalError>;
317
318    /// Read a Table-owned blob, or a range of it. A whole read checks the payload's digest.
319    async fn read_blob(
320        &self,
321        blob: &Blob,
322        range: Option<Range<u64>>,
323        storage: &StorageBinding,
324    ) -> Result<ByteStream, ScandalError>;
325
326    /// What storage holds for a Table-owned blob, without reading its bytes.
327    async fn stat_blob(
328        &self,
329        blob: &Blob,
330        storage: &StorageBinding,
331    ) -> Result<BlobStat, ScandalError>;
332
333    /// Store one payload as a dedicated blob of `table`, before any row references it.
334    async fn write_blob(
335        &self,
336        table: &TableBinding,
337        media_type: &str,
338        max_bytes: u64,
339        input: ByteStream,
340    ) -> Result<Blob, ScandalError>;
341
342    /// Store the inline payloads of the `spql.Blob` columns of `input` as blobs of `table`,
343    /// packed as [`append_rows`](Self::append_rows) packs them, under `attempt`'s directory when
344    /// it names one. Returns the rows in order, each such payload replaced by its address, once
345    /// every pack is stored.
346    async fn pack_blobs(
347        &self,
348        table: &TableBinding,
349        attempt: Option<Uuid>,
350        schema: SchemaRef,
351        input: RecordBatchStream,
352    ) -> Result<Vec<RecordBatch>, ScandalError>;
353
354    /// Start a multipart upload of one blob into `attempt`'s directory of `table`, with the
355    /// binding the Tables API issued for that attempt. Parts hold `part_bytes` each, a power of
356    /// two, or [`DEFAULT_UPLOAD_PART_BYTES`](self::blob::DEFAULT_UPLOAD_PART_BYTES).
357    async fn start_blob_upload(
358        &self,
359        table: &TableBinding,
360        attempt: Uuid,
361        media_type: &str,
362        part_bytes: Option<u64>,
363    ) -> Result<BlobUpload, ScandalError>;
364
365    /// Store part `part`, counted from 1, of `upload`. Every part but the last holds exactly the
366    /// upload's part size. Sending a part again replaces it.
367    async fn upload_blob_part(
368        &self,
369        upload: &BlobUpload,
370        part: u32,
371        input: ByteStream,
372        storage: &StorageBinding,
373    ) -> Result<BlobPart, ScandalError>;
374
375    /// Join every part of `upload`, in order from part 1, into its blob, whose digest merges
376    /// the parts' hashes.
377    async fn complete_blob_upload(
378        &self,
379        upload: &BlobUpload,
380        parts: &[BlobPart],
381        storage: &StorageBinding,
382    ) -> Result<Blob, ScandalError>;
383
384    /// Discard the stored parts of `upload`. An abort removes no stored object.
385    async fn abort_blob_upload(
386        &self,
387        upload: &BlobUpload,
388        storage: &StorageBinding,
389    ) -> Result<(), ScandalError>;
390
391    /// Write rows as fragments for a successor of `base`, or for the table's first version
392    /// when there is none. Appends may run concurrently; finalize adds the kept receipts.
393    ///
394    /// With an `attempt`, the files go under the attempt's directory, which the binding the
395    /// Tables API issued for that attempt may write; without one they go under `data/`.
396    async fn append_rows(
397        &self,
398        table: &TableBinding,
399        base: Option<Uuid>,
400        attempt: Option<Uuid>,
401        schema: SchemaRef,
402        input: RecordBatchStream,
403    ) -> Result<Vec<FragmentReceipt>, ScandalError>;
404
405    /// Write one independent input stream, validating and retaining its physical properties.
406    async fn append_partition(
407        &self,
408        request: &AppendRowsRequest,
409        schema: SchemaRef,
410        input: RecordBatchStream,
411    ) -> Result<Vec<FragmentReceipt>, ScandalError> {
412        if request
413            .properties
414            .as_ref()
415            .is_some_and(|properties| !properties.ordering.is_empty())
416        {
417            return Err(ScandalError::InvalidArgument {
418                message: "this backend does not support ordered writes".into(),
419            });
420        }
421        self.append_rows(&request.table, request.base, request.attempt, schema, input)
422            .await
423    }
424
425    /// Write new values for existing rows of `request.base`, addressed by row ID in
426    /// `request.row_id_column`, as patch files. Prepare them on an `edit` base.
427    async fn patch_rows(
428        &self,
429        _request: &PatchRowsRequest,
430        _schema: SchemaRef,
431        _input: RecordBatchStream,
432    ) -> Result<Vec<FragmentReceipt>, ScandalError> {
433        Err(unsupported("patching rows"))
434    }
435
436    /// Record deletions of rows of `request.base`, addressed by row ID. Prepare them on an
437    /// `edit` base.
438    async fn delete_rows(
439        &self,
440        _request: &DeleteRowsRequest,
441        _schema: SchemaRef,
442        _input: RecordBatchStream,
443    ) -> Result<Vec<FragmentReceipt>, ScandalError> {
444        Err(unsupported("deleting rows by row ID"))
445    }
446
447    /// Attach values of the columns `request.schema_changes` adds, for every row one fragment
448    /// of `request.base` keeps, in stored order, as column files of that fragment. Prepare them
449    /// on an `edit` base.
450    async fn attach_columns(
451        &self,
452        _request: &AttachColumnsRequest,
453        _schema: SchemaRef,
454        _input: RecordBatchStream,
455    ) -> Result<AttachedColumns, ScandalError> {
456        Err(unsupported("attaching columns"))
457    }
458
459    /// Write the mirror of fragment `request.fragment_id` of `request.source` from the rows the
460    /// source keeps there, in stored order. Prepare it on an `aligned` base.
461    async fn write_aligned(
462        &self,
463        _request: &WriteAlignedRequest,
464        _schema: SchemaRef,
465        _input: RecordBatchStream,
466    ) -> Result<AlignedFragment, ScandalError> {
467        Err(unsupported("aligned writes"))
468    }
469
470    /// How one version of a Table changed since an earlier one, from metadata only.
471    async fn plan_changes(
472        &self,
473        _request: &PlanChangesRequest,
474    ) -> Result<ChangePlan, ScandalError> {
475        Err(unsupported("change plans"))
476    }
477
478    /// Prepare a version from appended fragments. The host registers and publishes it.
479    async fn finalize_write(
480        &self,
481        table: &TableBinding,
482        base: &WriteBase,
483        receipts: &[FragmentReceipt],
484    ) -> Result<PreparedWriteReceipt, ScandalError>;
485
486    /// The output schema of named expressions over an input schema, without data.
487    async fn bind_projection(
488        &self,
489        input: &Schema,
490        projection: &[NamedExpression],
491    ) -> Result<SchemaRef, ScandalError>;
492
493    /// Evaluate named expressions over the caller's batches, row for row. Blob reads use the
494    /// storage of `tables`.
495    async fn evaluate(
496        &self,
497        projection: &[NamedExpression],
498        input_schema: SchemaRef,
499        input: RecordBatchStream,
500        tables: &[TableBinding],
501    ) -> Result<Evaluation, ScandalError>;
502}
503
504/// The error of an operation that a scandal implementation does not provide.
505fn unsupported(operation: &str) -> ScandalError {
506    ScandalError::InvalidArgument {
507        message: format!("{operation} is unavailable on this implementation"),
508    }
509}
spiral/scandal/mod.rs

spiral/scandal/
mod.rs
