Struct StorageManager 

pub struct StorageManager {
    pub config: UniConfig,
    pub compaction_status: Arc<Mutex<CompactionStatus>>,
    pub flush_in_progress: AtomicUsize,
    /* private fields */
}

Fields

config: UniConfig

compaction_status: Arc<Mutex<CompactionStatus>>

flush_in_progress: AtomicUsize

Counter of in-flight flush_to_l1 operations. Compaction skips delta-clear when this is non-zero to avoid wiping rows a flush is about to append. Counter (not bool) so multiple async flushes can be in flight concurrently.

Implementations

impl StorageManager

pub async fn new_with_backend( base_uri: &str, store: Arc<dyn ObjectStore>, backend: Arc<dyn StorageBackend>, schema_manager: Arc<SchemaManager>, config: UniConfig, ) -> Result<Self>

Create a new StorageManager with a pre-configured backend.

pub async fn new( base_uri: &str, schema_manager: Arc<SchemaManager>, ) -> Result<Self>

Create a new StorageManager with LanceDB integration.

pub async fn new_with_cache( base_uri: &str, schema_manager: Arc<SchemaManager>, adjacency_cache_size: usize, ) -> Result<Self>

Create a new StorageManager with custom cache size.

pub async fn new_with_config( base_uri: &str, schema_manager: Arc<SchemaManager>, config: UniConfig, ) -> Result<Self>

Create a new StorageManager with custom configuration.

pub async fn new_with_store_and_config( base_uri: &str, store: Arc<dyn ObjectStore>, schema_manager: Arc<SchemaManager>, config: UniConfig, ) -> Result<Self>

Create a new StorageManager using an already-constructed object store.

pub async fn new_with_store_and_storage_options( base_uri: &str, store: Arc<dyn ObjectStore>, schema_manager: Arc<SchemaManager>, config: UniConfig, lancedb_storage_options: Option<HashMap<String, String>>, ) -> Result<Self>

Create a new StorageManager with LanceDB storage options.

pub fn local_fs_root(&self) -> Option<PathBuf>

Filesystem root backing this manager’s object store, when the store is a local filesystem (the non-:// branch of build_store_from_uri). Used to fsync WAL segments after PUT — object_store::LocalFileSystem does not fsync on its own. None for remote/URL-based stores.

pub fn pinned(&self, snapshot: SnapshotManifest) -> Self

pub fn pinned_at_version(&self, hwm: u64) -> Self

Construct a clone of this StorageManager pinned to a row-version high-water mark (C2: transaction-level L1 pinning).

Unlike Self::pinned, this needs no SnapshotManifest: scans filter to _version <= hwm via Self::version_high_water_mark. A read-write transaction builds one of these at begin with SnapshotView.started_at_version, so an L0→L1 flush completing mid-transaction cannot leak post-snapshot rows into its L1 scans (the L0 tier is pinned separately by the SnapshotView).

Unlike Self::pinned, the live AdjacencyManager is SHARED, not fresh: commits replay their edges into the live manager’s overlay, which is the traversal path’s only source for L0-resident edges — a fresh manager would make every unflushed edge invisible to the transaction. The cost is that the edge tier is not version-pinned (post-snapshot edges remain visible to traversals, exactly as before C2); edge reads are recorded in the OCC read-set, so a conflicting read-modify-write still aborts at commit.

pub fn at_fork(&self, scope: Arc<ForkScope>) -> Self

Construct a fork-scoped clone of this StorageManager.

All reads through dataset factories and through backend() on the returned manager route through the fork’s Lance branches via base_paths. The AdjacencyManager is fresh (per Issue #73 reasoning — same as pinned) to prevent primary’s CSR from leaking into the fork. fork_scope and pinned_snapshot are mutually exclusive.

The backend is wrapped in crate::backend::branched::BranchedBackend so that every ScanRequest constructed anywhere (PropertyManager, MainVertexDataset static methods, etc.) automatically picks up the fork’s branch for tables the fork has branched. Untracked tables fall back to primary, matching Phase 1 read semantics.

pub fn at_fork_with_schema( &self, scope: Arc<ForkScope>, merged_schema: Arc<SchemaManager>, ) -> Self

Variant of Self::at_fork that uses an explicit merged_schema for the fork’s storage rather than primary’s schema_manager. Used by UniInner::at_fork so that the fork-side strict-schema checks (in uni-query / uni-store’s writer) see fork-local labels and edge types added through Session::fork_schema(). Without this, those checks would route through primary’s schema and reject fork-local labels.

pub fn fork_scope(&self) -> Option<&Arc<ForkScope>>

Borrow the active fork scope, if any.

pub fn fork_index_exists( &self, label: &str, column: &str, ) -> Option<ForkLocalIndexKind>

Phase 5a: query whether a fork-local index exists for the (label, column) pair on the active fork scope. Returns None outside a fork or when no fork-local build has completed for that pair.

The planner consults this to decide whether to emit FusedIndexScan (returns Some) or fall back to the inherited primary index via base_paths (returns None). The lookup is a DashMap::get on ForkScope — O(1) and safe to call per query without caching above this layer.

pub fn base_uri(&self) -> &str

Base URI for this storage manager (the directory or remote prefix under which dataset directories live).

pub fn get_edge_version_by_id(&self, edge_type_id: u32) -> Option<u64>

pub fn version_high_water_mark(&self) -> Option<u64>

Returns the version high-water mark from the pinned snapshot or the transaction-level version pin, if present.

Used by the SCAN tier (vertex tables, property reads) to filter data by version: when set, only rows with _version <= version_high_water_mark are visible. The edge/adjacency path must use Self::snapshot_version_hwm instead.

pub fn snapshot_version_hwm(&self) -> Option<u64>

Version high-water mark from a manifest-pinned (time-travel) snapshot ONLY — never from a transaction-level version pin.

The edge/adjacency read path switches to version-filtered CSR reads and skips the L0 overlays when a hwm is present. That is correct for time-travel (a snapshot is flushed state, with its own fresh AdjacencyManager), but a transaction pin shares the LIVE adjacency manager and needs live CSR + L0 overlays + its tx-L0 — filtering there would hide unflushed edges and poison the shared warm cache. The edge tier is deliberately not version-pinned for transactions (see Self::pinned_at_version).

pub fn apply_version_filter(&self, base_filter: String) -> String

Apply version filtering to a base filter expression.

If a snapshot is pinned, wraps base_filter with an additional _version <= hwm clause. Otherwise returns base_filter unchanged.

pub fn store(&self) -> Arc<dyn ObjectStore>

pub fn compaction_status(&self) -> Result<CompactionStatus, LockPoisonedError>

Get current compaction status.

Errors

Returns error if the compaction status lock is poisoned (see issue #18/#150).

pub async fn compact(&self) -> Result<CompactionStats>

pub async fn compact_label(&self, label: &str) -> Result<CompactionStats>

pub async fn compact_edge_type( &self, edge_type: &str, ) -> Result<CompactionStats>

pub async fn wait_for_compaction(&self) -> Result<()>

pub fn start_background_compaction( self: Arc<Self>, shutdown_rx: Receiver<()>, ) -> JoinHandle<()>

pub fn trigger_async_compaction(self: &Arc<Self>)

Trigger L1 compaction asynchronously without blocking the caller. Safe to call frequently — CompactionGuard prevents concurrent runs.

pub fn invalidate_table_cache(&self, label: &str)

Open a LanceDB table for a label.

Invalidate cached table state (call after writes).

pub fn base_path(&self) -> &str

pub fn schema_manager(&self) -> &SchemaManager

pub fn schema_manager_arc(&self) -> Arc<SchemaManager>

pub fn schema_manager_arc_ref(&self) -> &Arc<SchemaManager>

Returns the backing Arc<SchemaManager> by reference.

Unlike Self::schema_manager (which derefs to &SchemaManager), this preserves the Arc’s pointer identity. A pinned transaction and the live session clone the same schema_manager Arc, while forks hold a distinct one — so this is the correct registry key for the projection store (see uni-query’s projection_store::for_storage).

pub fn adjacency_manager(&self) -> Arc<AdjacencyManager>

Get the adjacency manager for the dual-CSR architecture.

pub async fn warm_adjacency( &self, edge_type_id: u32, direction: Direction, version: Option<u64>, ) -> Result<()>

Warm the adjacency manager for a specific edge type and direction.

Builds the Main CSR from L2 adjacency + L1 delta data in storage. Called lazily on first access per edge type or at startup.

pub async fn warm_adjacency_coalesced( &self, edge_type_id: u32, direction: Direction, version: Option<u64>, ) -> Result<()>

Coalesced warm_adjacency() to prevent cache stampede (Issue #13).

Uses double-checked locking to ensure only one concurrent warm() per (edge_type, direction) key. Subsequent callers wait for the first to complete.

pub fn has_adjacency_csr(&self, edge_type_id: u32, direction: Direction) -> bool

Check whether the adjacency manager has a CSR for the given edge type and direction.

pub fn get_neighbors_at_version( &self, vid: Vid, edge_type: u32, direction: Direction, version: u64, ) -> Vec<(Vid, Eid)>

Get neighbors at a specific version for snapshot queries.

pub fn backend(&self) -> &dyn StorageBackend

Get the storage backend.

pub fn backend_arc(&self) -> Arc<dyn StorageBackend>

Get the storage backend as an Arc.

pub fn get_labels_from_index(&self, vid: Vid) -> Option<Vec<String>>

Get labels for a VID from the in-memory index. Returns None if the index is disabled or the VID is not found.

pub fn update_vid_labels_index(&self, vid: Vid, labels: Vec<String>)

Update the VID-to-labels mapping in the index. No-op if the index is disabled.

pub fn remove_from_vid_labels_index(&self, vid: Vid)

Remove a VID from the labels index. No-op if the index is disabled.

pub async fn load_subgraph_cached( &self, start_vids: &[Vid], edge_types: &[u32], max_hops: usize, direction: GraphDirection, _l0: Option<Arc<RwLock<L0Buffer>>>, ) -> Result<WorkingGraph>

pub fn snapshot_manager(&self) -> &SnapshotManager

pub fn index_manager(&self) -> IndexManager

pub async fn scan_vertex_table( &self, label: &str, columns: &[&str], additional_filter: Option<&str>, ) -> Result<Option<RecordBatch>>

Scan a per-label vertex table. Returns None if the table doesn’t exist.

Internally opens the table, filters requested columns to those that physically exist, and applies the version HWM filter for snapshot isolation.

pub async fn scan_delta_table( &self, edge_type: &str, direction: &str, columns: &[&str], additional_filter: Option<&str>, ) -> Result<Option<RecordBatch>>

Scan a delta table for an edge type + direction. Returns None if the table doesn’t exist.

pub async fn scan_main_vertex_table( &self, columns: &[&str], filter: Option<&str>, ) -> Result<Option<RecordBatch>>

Scan the unified main vertex table. Returns None if table doesn’t exist.

Applies version HWM filter internally for snapshot isolation, combined with any caller-provided filter (label conditions, etc.).

pub async fn scan_main_edge_table_stream( &self, filter: Option<&str>, ) -> Result<Option<Pin<Box<dyn Stream<Item = Result<RecordBatch>> + Send>>>>

Scan the main edge table as a stream. Returns None if table doesn’t exist.

pub async fn scan_vertex_table_stream( &self, label: &str, ) -> Result<Option<Pin<Box<dyn Stream<Item = Result<RecordBatch>> + Send>>>>

Scan a per-label vertex table as a stream. Returns None if table doesn’t exist.

pub async fn find_vertex_by_ext_id(&self, ext_id: &str) -> Result<Option<Vid>>

Find a vertex VID by external ID. Uses pinned snapshot HWM if present.

pub async fn find_vertex_labels_by_vid( &self, vid: Vid, ) -> Result<Option<Vec<String>>>

Find labels for a vertex by VID. Uses pinned snapshot HWM if present.

pub async fn find_edges_by_type_names( &self, type_names: &[&str], endpoint_filter: Option<(EndpointSide, &[Vid])>, ) -> Result<Vec<(Eid, Vid, Vid, String, Properties)>>

Find edges from the main edge table by type names, optionally pushing a bounded endpoint vid set into the scan (review perf #5).

pub async fn scan_vertex_candidates( &self, label: &str, filter: Option<&str>, ) -> Result<Vec<Vid>>

Scan vertex candidates matching a filter. Returns VIDs where _deleted = false.

pub fn vertex_dataset(&self, label: &str) -> Result<VertexDataset>

pub fn edge_dataset( &self, edge_type: &str, src_label: &str, dst_label: &str, ) -> Result<EdgeDataset>

pub fn delta_dataset( &self, edge_type: &str, direction: &str, ) -> Result<DeltaDataset>

pub fn adjacency_dataset( &self, edge_type: &str, label: &str, direction: &str, ) -> Result<AdjacencyDataset>

pub fn main_vertex_dataset(&self) -> MainVertexDataset

Get the main vertex dataset for unified vertex storage.

The main vertex dataset contains all vertices regardless of label, enabling fast ID-based lookups without knowing the label.

pub fn main_edge_dataset(&self) -> MainEdgeDataset

Get the main edge dataset for unified edge storage.

The main edge dataset contains all edges regardless of type, enabling fast ID-based lookups without knowing the edge type.

pub fn uid_index(&self, label: &str) -> Result<UidIndex>

pub async fn inverted_index( &self, label: &str, property: &str, ) -> Result<InvertedIndex>

pub async fn vector_search( &self, label: &str, property: &str, query: &[f32], k: usize, filter: Option<&str>, ctx: Option<&QueryContext>, ) -> Result<Vec<(Vid, f32)>>

pub async fn fts_search( &self, label: &str, property: &str, query: &str, k: usize, filter: Option<&str>, ctx: Option<&QueryContext>, ) -> Result<Vec<(Vid, f32)>>

Perform a full-text search with BM25 scoring.

Returns vertices matching the search query along with their BM25 scores. Results are sorted by score descending (most relevant first).

Arguments

• label - The label to search within
• property - The property column to search (must have FTS index)
• query - The search query text
• k - Maximum number of results to return
• filter - Optional Lance filter expression
• ctx - Optional query context for visibility checks

Returns

Vector of (Vid, score) tuples, where score is the BM25 relevance score.

pub async fn get_vertex_by_uid( &self, uid: &UniId, label: &str, ) -> Result<Option<Vid>>

pub async fn insert_vertex_with_uid( &self, label: &str, vid: Vid, uid: UniId, ) -> Result<()>

pub async fn load_subgraph( &self, start_vids: &[Vid], edge_types: &[u32], max_hops: usize, direction: GraphDirection, l0: Option<&L0Buffer>, ) -> Result<WorkingGraph>