spiral-sdk 0.1.9

//! The scandal API: one set of operations that embedded and remote scandal implement alike.
//!
//! Every request is self-contained. Hosts resolve names, obtain storage bindings and publish
//! prepared versions; a scandal never calls the control plane.

#[path = "../api/scandal/contract.rs"]
pub mod api;
pub mod blob;
pub mod emit;
pub mod error;
pub mod expression;
pub mod functions;
pub mod ordered;
pub mod read_cache;
mod safelong;
mod storage;
pub mod types;

use std::borrow::Cow;
use std::ops::Range;

use arrow_array::RecordBatch;
use arrow_schema::{Schema, SchemaRef};
use async_trait::async_trait;
use bytes::Bytes;
use futures::stream::BoxStream;
use serde::{Deserialize, Serialize};
pub use storage::{GcsSession, S3Session, StorageConfig};
use url::Url;
use uuid::Uuid;
pub use vortex_array as vortex;

use self::blob::{Blob, BlobPart, BlobStat, BlobUpload};
pub use self::error::ScandalError;
use self::expression::NamedExpression;
use self::functions::FunctionInfo;
pub use self::types::{
    FilterClause, FragmentReceipt, PlannedSplit, PreparedWriteReceipt, RowRange, Scan, ScanInfo,
    ScanMetricsSnapshot, ScanPlan, Split, SplitTake, TableVersion,
};

pub mod wire;

/// The storage transports a scandal may use. Requests never choose an endpoint.
#[derive(Clone, Debug, Default, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
pub struct ScandalConfig {
    /// The SPFS endpoint that signs requests for `spfsm://` and `spfsu://` roots.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub spfs_endpoint: Option<Url>,
    /// Read `file://` roots. Requires the runtime's `local-files` feature; hosted scandal
    /// leaves it off.
    #[serde(default)]
    pub local_files: bool,
    /// Read `s3://` and `gs://` roots with the provider session the host resolved.
    #[serde(default)]
    pub provider_roots: bool,
    /// Refuse an S3 session that names its own endpoint, so that only AWS and Cloud Storage
    /// are reached. Hosted scandal sets it: a caller's session must not choose where the
    /// service connects.
    #[serde(default)]
    pub refuse_custom_endpoints: bool,
}

pub use self::api::{
    AppendRowsRequest, ArrowSchema, AttachColumnsRequest, AttachedColumns, ChangePlan,
    DeleteRowsRequest, EditBase, GenesisBase, PatchRowsRequest, PlanChangesRequest,
    ReadLegacyTableRequest, RowIdRange, SchemaChange, SortColumn, StorageBinding,
    StorageCredential, SuccessorBase, TableBinding, WriteBase, WriteProperties,
};

impl StorageBinding {
    /// A logical resource reference. A resolver-enabled host obtains its scoped storage.
    pub fn resource(resource: Uuid) -> Self {
        Self::new(format!("spiral://{resource}/"))
    }

    /// Storage for `root_uri`, with no alias or credential, as local files need.
    pub fn new(root_uri: impl Into<String>) -> Self {
        Self {
            root_uri: root_uri.into(),
            root_alias: None,
            credential: None,
        }
    }

    /// The operation-scoped credential: the SPFS capability of an SPFS root, the JSON
    /// [`S3Session`] of an `s3://` root, or the JSON
    /// [`GcsSession`] of a `gs://` root; absent for local files.
    pub fn credential(&self) -> Option<&str> {
        self.credential
            .as_ref()
            .map(|credential| credential.0.as_str())
    }
}

impl TableBinding {
    /// Explicit storage, or a logical resource root for the host resolver.
    pub fn binding(&self) -> Cow<'_, StorageBinding> {
        self.storage.as_ref().map_or_else(
            || Cow::Owned(StorageBinding::resource(self.table_id)),
            Cow::Borrowed,
        )
    }
}

/// The host's URI resolution result: an authorized source prefix and the storage that reads it.
/// Objects below `root_uri` use the same relative path below `storage.root_uri`.
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct ResolvedUri {
    /// Absolute source directory URI, ending in `/`, with no query or fragment.
    pub root_uri: Url,
    /// Scoped physical storage and credentials. The runtime owns its managed I/O transport.
    pub storage: StorageBinding,
}

/// The storage authority a host must obtain for one resource operation.
#[derive(Clone, Copy, Debug, Eq, PartialEq, Hash, Serialize, Deserialize)]
#[serde(
    tag = "type",
    rename_all = "camelCase",
    rename_all_fields = "camelCase"
)]
pub enum ResourceAccess {
    /// Read objects below the resource root.
    Read,
    /// Read the root and create objects only within one write attempt.
    Attempt { attempt_id: Uuid },
    /// Read the root and create version metadata and owned blobs.
    Publication,
}

/// Host-configured URI policy, resource lookup and credentials, outside the execution runtime.
///
/// A resolver belongs to one authorization context. Cloned runtimes share its resolved roots
/// until their credentials near expiry; different callers must receive separate resolvers.
#[async_trait]
pub trait UriResolver: Send + Sync {
    /// Authorize `uri` and bind the source prefix containing it. The runtime reuses this
    /// binding for later objects under that prefix, then asks again before credentials expire.
    async fn resolve(&self, uri: &Url) -> Result<ResolvedUri, ScandalError>;

    /// Resolve the resource for a particular operation. Read-only hosts reject write access.
    async fn resolve_for(
        &self,
        uri: &Url,
        access: ResourceAccess,
    ) -> Result<ResolvedUri, ScandalError> {
        if access != ResourceAccess::Read {
            return Err(ScandalError::PermissionDenied {
                operation: "write".into(),
                resource: uri.to_string(),
            });
        }
        self.resolve(uri).await
    }
}

impl ArrowSchema {
    /// The contract form of an Arrow schema.
    pub fn from_schema(schema: &Schema) -> Result<Self, ScandalError> {
        serde_json::to_value(schema)
            .map(Self)
            .map_err(|error| ScandalError::Internal {
                message: error.to_string(),
            })
    }

    /// The Arrow schema this value describes.
    pub fn to_schema(&self) -> Result<Schema, ScandalError> {
        serde_json::from_value(self.0.clone()).map_err(|error| ScandalError::InvalidArgument {
            message: format!("invalid Arrow schema: {error}"),
        })
    }
}

impl WriteBase {
    /// The Table's first version, with the schema its appends used.
    pub fn genesis(schema: &Schema) -> Result<Self, ScandalError> {
        Ok(Self::Genesis(GenesisBase {
            schema: ArrowSchema::from_schema(schema)?,
        }))
    }

    /// A successor of `version_id` that keeps its rows.
    pub fn append(version_id: Uuid) -> Self {
        Self::Append(SuccessorBase { version_id })
    }

    /// A successor of `version_id` that replaces its rows.
    pub fn overwrite(version_id: Uuid) -> Self {
        Self::Overwrite(SuccessorBase { version_id })
    }

    /// The version this write builds on, absent for the Table's first version.
    pub fn base_version(&self) -> Result<Option<Uuid>, ScandalError> {
        match self {
            Self::Genesis(_) => Ok(None),
            Self::Append(base) | Self::Overwrite(base) => Ok(Some(base.version_id)),
            Self::Delete(base) => Ok(Some(base.version_id)),
            Self::Edit(base) => Ok(Some(base.version_id)),
            Self::Unknown { tag, .. } => Err(ScandalError::InvalidArgument {
                message: format!("unknown write base `{tag}`"),
            }),
        }
    }
}

/// One Arrow batch of a split and the stored row of its fragment that the read reached.
#[derive(Clone, Debug)]
pub struct SplitBatch {
    pub batch: RecordBatch,
    /// The stored row after the rows this batch covered; resume the split from here. For a
    /// take, the request's positions delivered so far; resume with the positions after them.
    pub next_row: u64,
    /// The split read's counters, on its last batch when the scan collects metrics. That
    /// batch may hold no rows. Pruning counters are the plan's, never a split's.
    pub metrics: Option<ScanMetricsSnapshot>,
}

pub type SplitStream = BoxStream<'static, Result<SplitBatch, ScandalError>>;
pub type RecordBatchStream = BoxStream<'static, Result<RecordBatch, ScandalError>>;
pub type ByteStream = BoxStream<'static, Result<Bytes, ScandalError>>;

/// A split read: its output schema, then its batches in stored order.
pub struct SplitRead {
    pub schema: SchemaRef,
    pub batches: SplitStream,
}

/// An evaluation's output schema, then its batches, row for row with the input.
pub struct Evaluation {
    pub schema: SchemaRef,
    pub batches: RecordBatchStream,
}

/// Scans, writes, blob transfer and SPQL evaluation over exact table versions.
///
/// Embedded and remote scandal implement every operation with the same semantics.
#[async_trait]
pub trait Scandal: Send + Sync {
    /// Whether this host resolves logical resource roots internally, including write scopes.
    fn resolves_resources(&self) -> bool {
        false
    }

    /// Read a bounded legacy window from host-resolved metadata and scoped storage as Arrow IPC.
    async fn read_legacy_table(
        &self,
        _request: &ReadLegacyTableRequest,
    ) -> Result<ByteStream, ScandalError> {
        Err(ScandalError::InvalidArgument {
            message: "Legacy reads are unavailable on this implementation.".to_owned(),
        })
    }

    /// The SPQL functions this scandal can run.
    async fn list_functions(&self) -> Result<Vec<FunctionInfo>, ScandalError>;

    /// Plan a scan: its output schema, its filter support, and one split per fragment that
    /// survives pruning by statistics, in stored order.
    async fn plan_scan(
        &self,
        scan: &Scan,
        storage: &StorageBinding,
    ) -> Result<ScanPlan, ScandalError>;

    /// Read one split in stored order. Each batch reports the stored row it reached.
    async fn read_scan(
        &self,
        scan: &Scan,
        split: &Split,
        storage: &StorageBinding,
    ) -> Result<SplitRead, ScandalError>;

    /// Read rows of planned splits by position, in request order: each take's rows in the order
    /// of its positions, a repeated position's row again, the takes one after another. Positions
    /// count the rows the version keeps in each split. A batch's
    /// [`next_row`](SplitBatch::next_row) is how many of the request's positions, across its
    /// takes, the batches so far delivered.
    async fn take_rows(
        &self,
        scan: &Scan,
        takes: &[SplitTake],
        storage: &StorageBinding,
    ) -> Result<SplitRead, ScandalError>;

    /// Read a Table-owned blob, or a range of it. A whole read checks the payload's digest.
    async fn read_blob(
        &self,
        blob: &Blob,
        range: Option<Range<u64>>,
        storage: &StorageBinding,
    ) -> Result<ByteStream, ScandalError>;

    /// What storage holds for a Table-owned blob, without reading its bytes.
    async fn stat_blob(
        &self,
        blob: &Blob,
        storage: &StorageBinding,
    ) -> Result<BlobStat, ScandalError>;

    /// Store one payload as a dedicated blob of `table`, before any row references it.
    async fn write_blob(
        &self,
        table: &TableBinding,
        media_type: &str,
        max_bytes: u64,
        input: ByteStream,
    ) -> Result<Blob, ScandalError>;

    /// Store the inline payloads of the `spql.Blob` columns of `input` as blobs of `table`,
    /// packed as [`append_rows`](Self::append_rows) packs them, under `attempt`'s directory when
    /// it names one. Returns the rows in order, each such payload replaced by its address, once
    /// every pack is stored.
    async fn pack_blobs(
        &self,
        table: &TableBinding,
        attempt: Option<Uuid>,
        schema: SchemaRef,
        input: RecordBatchStream,
    ) -> Result<Vec<RecordBatch>, ScandalError>;

    /// Start a multipart upload of one blob into `attempt`'s directory of `table`, with the
    /// binding the Tables API issued for that attempt. Parts hold `part_bytes` each, a power of
    /// two, or [`DEFAULT_UPLOAD_PART_BYTES`](self::blob::DEFAULT_UPLOAD_PART_BYTES).
    async fn start_blob_upload(
        &self,
        table: &TableBinding,
        attempt: Uuid,
        media_type: &str,
        part_bytes: Option<u64>,
    ) -> Result<BlobUpload, ScandalError>;

    /// Store part `part`, counted from 1, of `upload`. Every part but the last holds exactly the
    /// upload's part size. Sending a part again replaces it.
    async fn upload_blob_part(
        &self,
        upload: &BlobUpload,
        part: u32,
        input: ByteStream,
        storage: &StorageBinding,
    ) -> Result<BlobPart, ScandalError>;

    /// Join every part of `upload`, in order from part 1, into its blob, whose digest merges
    /// the parts' hashes.
    async fn complete_blob_upload(
        &self,
        upload: &BlobUpload,
        parts: &[BlobPart],
        storage: &StorageBinding,
    ) -> Result<Blob, ScandalError>;

    /// Discard the stored parts of `upload`. An abort removes no stored object.
    async fn abort_blob_upload(
        &self,
        upload: &BlobUpload,
        storage: &StorageBinding,
    ) -> Result<(), ScandalError>;

    /// Write rows as fragments for a successor of `base`, or for the table's first version
    /// when there is none. Appends may run concurrently; finalize adds the kept receipts.
    ///
    /// With an `attempt`, the files go under the attempt's directory, which the binding the
    /// Tables API issued for that attempt may write; without one they go under `data/`.
    async fn append_rows(
        &self,
        table: &TableBinding,
        base: Option<Uuid>,
        attempt: Option<Uuid>,
        schema: SchemaRef,
        input: RecordBatchStream,
    ) -> Result<Vec<FragmentReceipt>, ScandalError>;

    /// Write one independent input stream, validating and retaining its physical properties.
    async fn append_partition(
        &self,
        request: &AppendRowsRequest,
        schema: SchemaRef,
        input: RecordBatchStream,
    ) -> Result<Vec<FragmentReceipt>, ScandalError> {
        if request
            .properties
            .as_ref()
            .is_some_and(|properties| !properties.ordering.is_empty())
        {
            return Err(ScandalError::InvalidArgument {
                message: "this backend does not support ordered writes".into(),
            });
        }
        self.append_rows(&request.table, request.base, request.attempt, schema, input)
            .await
    }

    /// Write new values for existing rows of `request.base`, addressed by row ID in
    /// `request.row_id_column`, as patch files. Prepare them on an `edit` base.
    async fn patch_rows(
        &self,
        _request: &PatchRowsRequest,
        _schema: SchemaRef,
        _input: RecordBatchStream,
    ) -> Result<Vec<FragmentReceipt>, ScandalError> {
        Err(unsupported("patching rows"))
    }

    /// Record deletions of rows of `request.base`, addressed by row ID. Prepare them on an
    /// `edit` base.
    async fn delete_rows(
        &self,
        _request: &DeleteRowsRequest,
        _schema: SchemaRef,
        _input: RecordBatchStream,
    ) -> Result<Vec<FragmentReceipt>, ScandalError> {
        Err(unsupported("deleting rows by row ID"))
    }

    /// Attach values of the columns `request.schema_changes` adds, for every row one fragment
    /// of `request.base` keeps, in stored order, as column files of that fragment. Prepare them
    /// on an `edit` base.
    async fn attach_columns(
        &self,
        _request: &AttachColumnsRequest,
        _schema: SchemaRef,
        _input: RecordBatchStream,
    ) -> Result<AttachedColumns, ScandalError> {
        Err(unsupported("attaching columns"))
    }

    /// How one version of a Table changed since an earlier one, from metadata only.
    async fn plan_changes(
        &self,
        _request: &PlanChangesRequest,
    ) -> Result<ChangePlan, ScandalError> {
        Err(unsupported("change plans"))
    }

    /// Prepare a version from appended fragments. The host registers and publishes it.
    async fn finalize_write(
        &self,
        table: &TableBinding,
        base: &WriteBase,
        receipts: &[FragmentReceipt],
    ) -> Result<PreparedWriteReceipt, ScandalError>;

    /// The output schema of named expressions over an input schema, without data.
    async fn bind_projection(
        &self,
        input: &Schema,
        projection: &[NamedExpression],
    ) -> Result<SchemaRef, ScandalError>;

    /// Evaluate named expressions over the caller's batches, row for row. Blob reads use the
    /// storage of `tables`.
    async fn evaluate(
        &self,
        projection: &[NamedExpression],
        input_schema: SchemaRef,
        input: RecordBatchStream,
        tables: &[TableBinding],
    ) -> Result<Evaluation, ScandalError>;
}

/// The error of an operation that a scandal implementation does not provide.
fn unsupported(operation: &str) -> ScandalError {
    ScandalError::InvalidArgument {
        message: format!("{operation} is unavailable on this implementation"),
    }
}