Introduction

Neumann is a unified tensor-based runtime that stores relational data, graph relationships, and vector embeddings in a single system. Instead of stitching together a SQL database, a graph store, and a vector index, Neumann gives you all three behind one query language.

Choose Your Path

What Makes Neumann Different

One system, three engines. Store a table, connect entities in a graph, and search by vector similarity without moving data between systems.

-- Relational
CREATE TABLE documents (id INT PRIMARY KEY, title TEXT, author TEXT);
INSERT INTO documents VALUES (1, 'Intro to ML', 'Alice');

-- Graph
NODE CREATE topic { name: 'machine-learning' }
ENTITY CONNECT 'doc-1' -> 'topic-ml' : covers

-- Vector
EMBED STORE 'doc-1' [0.1, 0.2, 0.3, 0.4]

-- Cross-engine: find similar documents connected to a topic
SIMILAR 'doc-1' LIMIT 5 CONNECTED TO 'topic-ml'

Encrypted vault. Store secrets with AES-256-GCM encryption and graph-based access control.

LLM cache. Cache LLM responses with exact and semantic matching to reduce API costs.

Built-in consensus. Raft-based distributed consensus with 2PC transactions for multi-node deployments.

Architecture

                    +-------------------+
                    |   neumann_shell   |    Interactive CLI
                    |   neumann_server  |    gRPC server
                    +-------------------+
                             |
                    +-------------------+
                    |   query_router    |    Unified query execution
                    +-------------------+
                             |
        +----------+---------+---------+----------+
        |          |         |         |          |
   relational   graph    vector    tensor_    tensor_
   _engine     _engine  _engine   _vault     _cache
        |          |         |         |          |
        +----------+---------+---------+----------+
                             |
                    +-------------------+
                    |   tensor_store    |    Core storage (HNSW, sharded B-trees)
                    +-------------------+

Additional subsystems: tensor_blob (S3-style blob storage), tensor_chain (blockchain with Raft), tensor_checkpoint (snapshots), tensor_compress (tensor train decomposition).

Getting Started

• Installation – Install Neumann
• Quick Start – Your first queries
• Five-Minute Tutorial – Build a mini RAG system
• Use Cases – Real-world applications
• Building from Source – Compile from source

Reference

• Query Language – Full command reference
• Data Types – Scalar, vector, and sparse types
• Functions – Aggregates, distance metrics, operators
• API Reference – Rustdoc output

Installation

Multiple installation methods are available depending on your needs.

Quick Install (Recommended)

The easiest way to install Neumann is using the install script:

curl -sSfL https://raw.githubusercontent.com/Shadylukin/Neumann/main/install.sh | bash

This script will:

• Detect your platform (Linux x86_64, macOS x86_64, macOS ARM64)
• Download a pre-built binary if available
• Fall back to building from source if needed
• Install to /usr/local/bin or ~/.local/bin
• Install shell completions and man pages

Environment Variables

Homebrew (macOS/Linux)

brew tap Shadylukin/tap
brew install neumann

Cargo (crates.io)

If you have Rust installed:

cargo install neumann_shell

To install the gRPC server:

cargo install neumann_server

Docker

Interactive CLI

docker run -it shadylukinack/neumann:latest

Server Mode

docker run -d -p 9200:9200 -v neumann-data:/var/lib/neumann shadylukinack/neumann:server

Docker Compose

# Clone the repository
git clone https://github.com/Shadylukin/Neumann.git
cd Neumann

# Start the server
docker compose up -d neumann-server

# Run the CLI
docker compose run --rm neumann-cli

From Source

Requirements

• Rust 1.75 or later
• Cargo (included with Rust)
• Git
• protobuf compiler (for gRPC)

Build Steps

# Clone the repository
git clone https://github.com/Shadylukin/Neumann.git
cd Neumann

# Build in release mode
cargo build --release --package neumann_shell

# Install locally
cargo install --path neumann_shell

Run Tests

cargo test

SDK Installation

Python

pip install neumann-db

For embedded mode (in-process database):

pip install neumann-db[native]

TypeScript / JavaScript

npm install neumann-db

Or with yarn:

yarn add neumann-db

Verify Installation

neumann --version

Platform Support

Troubleshooting

“command not found: neumann”

The binary may not be in your PATH. Try:

# Check where it was installed
which neumann || ls ~/.local/bin/neumann

# Add to PATH if needed
export PATH="$HOME/.local/bin:$PATH"

Build fails with protobuf errors

Install the protobuf compiler:

# macOS
brew install protobuf

# Ubuntu/Debian
sudo apt-get install protobuf-compiler

# Fedora
sudo dnf install protobuf-compiler

Permission denied during install

The installer tries /usr/local/bin first (requires sudo) then falls back to ~/.local/bin. You can specify a custom directory:

NEUMANN_INSTALL_DIR=~/bin \
  curl -sSfL https://raw.githubusercontent.com/Shadylukin/Neumann/main/install.sh | bash

Python SDK native module errors

If you get errors about the native module, ensure you have a Rust toolchain:

curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh
pip install neumann-db[native]

Updating

Quick Install

Re-run the install script to get the latest version:

curl -sSfL https://raw.githubusercontent.com/Shadylukin/Neumann/main/install.sh | bash

Homebrew

brew upgrade neumann

Cargo

cargo install neumann_shell --force

Python SDK

pip install --upgrade neumann-db

TypeScript SDK

npm update neumann-db

Uninstalling

Quick Install / Cargo

rm $(which neumann)

Homebrew

brew uninstall neumann

Docker

docker rmi shadylukinack/neumann:latest shadylukinack/neumann:server
docker volume rm neumann-data

Python SDK

pip uninstall neumann-db

TypeScript SDK

npm uninstall neumann-db

Next Steps

• Quick Start - Run your first queries
• Building from Source - Development setup

Quick Start

Get up and running with Neumann in under 5 minutes. This guide walks you through relational queries, graph operations, vector search, and the cross-engine “wow” moment.

Start the Shell

# In-memory (data lost on exit)
neumann

# With persistence (recommended)
neumann --wal-dir ./data

You will see:

Neumann v0.1.0
Type 'help' for available commands.
neumann>

1. Relational Queries

Create a table and insert some data:

CREATE TABLE people (id INT PRIMARY KEY, name TEXT, role TEXT, team TEXT);

INSERT INTO people VALUES (1, 'Alice', 'Staff Engineer', 'Platform');
INSERT INTO people VALUES (2, 'Bob', 'Engineering Manager', 'Platform');
INSERT INTO people VALUES (3, 'Carol', 'Senior Engineer', 'ML');
INSERT INTO people VALUES (4, 'Dave', 'Junior Engineer', 'Platform');

Query it:

SELECT * FROM people WHERE team = 'Platform';
SELECT name, role FROM people ORDER BY name;
SELECT team, COUNT(*) AS headcount FROM people GROUP BY team;

2. Graph Operations

Create nodes with labels and properties:

NODE CREATE person { name: 'Alice', role: 'Staff Engineer' }
NODE CREATE person { name: 'Bob', role: 'Engineering Manager' }
NODE CREATE person { name: 'Carol', role: 'Senior Engineer' }
NODE CREATE person { name: 'Dave', role: 'Junior Engineer' }

List the nodes to see their auto-generated IDs:

NODE LIST person

Create edges (replace the IDs with the actual values from NODE LIST):

EDGE CREATE 'alice-node-id' -> 'bob-node-id' : reports_to
EDGE CREATE 'dave-node-id' -> 'bob-node-id' : reports_to
EDGE CREATE 'alice-node-id' -> 'dave-node-id' : mentors

Traverse the graph:

NEIGHBORS 'bob-node-id' INCOMING : reports_to
PATH SHORTEST 'dave-node-id' TO 'bob-node-id'

Run graph algorithms:

PAGERANK

3. Vector Search

Store embeddings with string keys:

EMBED STORE 'alice' [0.9, 0.4, 0.1, 0.7, 0.6, 0.3]
EMBED STORE 'bob' [0.6, 0.2, 0.1, 0.5, 0.3, 0.2]
EMBED STORE 'carol' [0.3, 0.9, 0.1, 0.4, 0.8, 0.1]
EMBED STORE 'dave' [0.4, 0.1, 0.2, 0.5, 0.2, 0.1]

Find similar items by key or by vector:

SIMILAR 'alice' LIMIT 3
SIMILAR [0.8, 0.5, 0.1, 0.6, 0.5, 0.2] LIMIT 3 METRIC COSINE

Check what is stored:

SHOW EMBEDDINGS
COUNT EMBEDDINGS

4. The Cross-Engine Moment

This is where Neumann shines. Combine graph traversal with vector similarity in a single query:

SIMILAR 'alice' LIMIT 3 CONNECTED TO 'bob-node-id'

This finds embeddings similar to Alice’s that are also connected to Bob in the graph. No joins across separate databases needed.

Search across all engines with FIND:

FIND NODE person WHERE name = 'Alice'

Create unified entities that span relational, graph, and vector storage:

ENTITY CREATE 'project-x' { name: 'Project X', status: 'active' } EMBEDDING [0.5, 0.3, 0.7, 0.2, 0.4, 0.1]
ENTITY GET 'project-x'

5. Persistence

Save a checkpoint:

CHECKPOINT 'my-first-checkpoint'
CHECKPOINTS

If you started with --wal-dir, your data persists across restarts. You can also save and load binary snapshots:

SAVE 'backup.bin'
LOAD 'backup.bin'

Next Steps

• Five-Minute Tutorial – Build a mini RAG system
• Use Cases – Real-world application patterns
• Query Language Reference – Full command list
• Architecture Overview – How it works under the hood
• Python SDK – Use Neumann from Python
• TypeScript SDK – Use Neumann from TypeScript

Sample Dataset

A ready-made dataset is available in samples/knowledge-base.nql. Load it with:

neumann --wal-dir ./data

neumann> \i samples/knowledge-base.nql

Five-Minute Tutorial: Build a Mini RAG System

This tutorial builds a retrieval-augmented generation (RAG) knowledge base using all three engines. By the end, you will have documents stored relationally, linked in a graph, searchable by embeddings, and protected with vault secrets.

Prerequisites

Start the shell with persistence:

neumann --wal-dir ./rag-data

Step 1: Create the Document Store

Use a relational table for structured document metadata:

CREATE TABLE documents (
    id INT PRIMARY KEY,
    title TEXT NOT NULL,
    category TEXT,
    author TEXT,
    created TEXT
);

INSERT INTO documents VALUES (1, 'Intro to Neural Networks', 'ml', 'Alice', '2024-01-15');
INSERT INTO documents VALUES (2, 'Transformer Architecture', 'ml', 'Bob', '2024-02-20');
INSERT INTO documents VALUES (3, 'Database Indexing', 'systems', 'Carol', '2024-03-10');
INSERT INTO documents VALUES (4, 'Vector Search at Scale', 'systems', 'Alice', '2024-04-05');
INSERT INTO documents VALUES (5, 'Fine-Tuning LLMs', 'ml', 'Dave', '2024-05-12');
INSERT INTO documents VALUES (6, 'Consensus Protocols', 'distributed', 'Eve', '2024-06-01');

Verify:

SELECT * FROM documents ORDER BY id;
SELECT category, COUNT(*) FROM documents GROUP BY category;

Step 2: Add Graph Relationships

Create nodes for documents and topics, then link them:

NODE CREATE document { title: 'Intro to Neural Networks', doc_id: 1 }
NODE CREATE document { title: 'Transformer Architecture', doc_id: 2 }
NODE CREATE document { title: 'Database Indexing', doc_id: 3 }
NODE CREATE document { title: 'Vector Search at Scale', doc_id: 4 }
NODE CREATE document { title: 'Fine-Tuning LLMs', doc_id: 5 }
NODE CREATE document { title: 'Consensus Protocols', doc_id: 6 }

NODE CREATE topic { name: 'machine-learning' }
NODE CREATE topic { name: 'systems' }
NODE CREATE topic { name: 'distributed-systems' }

List nodes to get IDs:

NODE LIST document
NODE LIST topic

Create edges linking documents to topics (use actual IDs from NODE LIST):

-- Documents 1, 2, 5 cover machine-learning
-- Documents 3, 4 cover systems
-- Documents 4, 6 cover distributed-systems
-- Document 2 references document 1
-- Document 5 references documents 1 and 2

Create citation relationships between documents:

EDGE CREATE 'doc2-id' -> 'doc1-id' : cites
EDGE CREATE 'doc5-id' -> 'doc1-id' : cites
EDGE CREATE 'doc5-id' -> 'doc2-id' : cites

Check the graph:

NEIGHBORS 'doc1-id' INCOMING : cites
PATH SHORTEST 'doc5-id' TO 'doc1-id'

Step 3: Store Embeddings

Each document gets a vector representing its content. In a real system, you would generate these with an embedding model (e.g., OpenAI, Cohere, or a local model). Here we use hand-crafted 6-dimensional vectors:

-- [neural-nets, transformers, databases, vectors, llms, distributed]
EMBED STORE 'doc-1' [0.9, 0.3, 0.1, 0.2, 0.2, 0.1]
EMBED STORE 'doc-2' [0.7, 0.95, 0.1, 0.1, 0.4, 0.1]
EMBED STORE 'doc-3' [0.1, 0.1, 0.95, 0.3, 0.0, 0.2]
EMBED STORE 'doc-4' [0.2, 0.1, 0.5, 0.9, 0.1, 0.4]
EMBED STORE 'doc-5' [0.6, 0.5, 0.1, 0.1, 0.95, 0.1]
EMBED STORE 'doc-6' [0.1, 0.1, 0.3, 0.2, 0.1, 0.9]

Search by similarity:

-- Find documents similar to "Intro to Neural Networks"
SIMILAR 'doc-1' LIMIT 3

-- Search with a custom query vector (someone asking about transformers + LLMs)
SIMILAR [0.5, 0.8, 0.1, 0.1, 0.7, 0.1] LIMIT 3 METRIC COSINE

Step 4: Graph-Aware Semantic Search

Combine vector similarity with graph connectivity. Find documents similar to a query vector that are also connected to a specific topic node:

SIMILAR [0.8, 0.6, 0.1, 0.1, 0.5, 0.1] LIMIT 3 CONNECTED TO 'ml-topic-id'

This is the core RAG pattern: retrieve relevant documents using embedding similarity, then filter by graph relationships for context-aware results.

Step 5: Protect API Keys with Vault

Store the embedding API key securely:

VAULT SET 'openai_api_key' 'sk-proj-abc123...'
VAULT SET 'cohere_api_key' 'co-xyz789...'
VAULT GRANT 'alice' ON 'openai_api_key'

Retrieve when needed:

VAULT GET 'openai_api_key'
VAULT LIST

Step 6: Cache LLM Responses

Initialize the cache and store responses to avoid repeated API calls:

CACHE INIT

CACHE PUT 'what are transformers?' 'Transformers are a neural network architecture based on self-attention mechanisms...'

CACHE SEMANTIC PUT 'explain attention mechanism' 'The attention mechanism allows models to focus on relevant parts of the input...' EMBEDDING [0.6, 0.9, 0.1, 0.1, 0.3, 0.1]

On subsequent queries, check the cache first:

CACHE GET 'what are transformers?'
CACHE SEMANTIC GET 'how does attention work?' THRESHOLD 0.8

Step 7: Checkpoint

Save your work:

CHECKPOINT 'rag-setup-complete'
CHECKPOINTS

What You Built

You now have a working RAG knowledge base with:

1. Structured metadata in a relational table (searchable with SQL)
2. Semantic relationships in a graph (topics, citations, authorship)
3. Vector embeddings for similarity search
4. Graph-aware retrieval combining similarity and structure
5. Encrypted secrets for API key management
6. Response caching to reduce LLM API costs
7. Checkpoint for safe rollback

Next Steps

• Use Cases – More application patterns
• Query Language Reference – All available commands
• Python SDK – Build this in Python
• Architecture Overview – How the engines work together

Use Cases

Neumann is designed for applications that need two or more of: structured data, relationships, and semantic search. Here are concrete examples with query patterns.

RAG Application

Problem: Build a retrieval-augmented generation system that retrieves relevant context for an LLM.

Why Neumann: Traditional RAG uses a vector store for retrieval. Neumann adds graph relationships (document structure, citations, topics) and relational metadata (authors, dates, permissions) to improve retrieval quality.

Schema

CREATE TABLE documents (
    id INT PRIMARY KEY,
    title TEXT NOT NULL,
    source TEXT,
    created TEXT,
    chunk_count INT
);

NODE CREATE collection { name: 'engineering-docs' }
-- For each document:
NODE CREATE document { title: 'Design Doc: Auth System', doc_id: 1 }
-- Link to collection:
EDGE CREATE 'doc-node-id' -> 'collection-id' : belongs_to
-- Link related documents:
EDGE CREATE 'doc-a-id' -> 'doc-b-id' : references

Indexing

-- Store chunk embeddings (one per document chunk)
EMBED STORE 'doc-1-chunk-0' [0.1, 0.2, ...]
EMBED STORE 'doc-1-chunk-1' [0.3, 0.1, ...]

Retrieval

-- Basic: find similar chunks
SIMILAR [0.2, 0.3, ...] LIMIT 10 METRIC COSINE

-- Graph-aware: find chunks connected to a specific collection
SIMILAR [0.2, 0.3, ...] LIMIT 10 CONNECTED TO 'collection-id'

-- Check cache before calling LLM
CACHE SEMANTIC GET 'how does auth work?' THRESHOLD 0.85

Store LLM response

CACHE SEMANTIC PUT 'how does auth work?' 'The auth system uses JWT tokens...' EMBEDDING [0.2, 0.3, ...]

Agent Memory

Problem: Give an AI agent persistent memory that supports both exact recall and semantic search, with conversation structure.

Why Neumann: Agents need to recall specific facts (relational), navigate conversation history (graph), and find semantically related memories (vector).

Schema

CREATE TABLE memories (
    id INT PRIMARY KEY,
    content TEXT NOT NULL,
    memory_type TEXT,
    importance FLOAT,
    created TEXT
);

-- Create session nodes
NODE CREATE session { name: 'session-2024-01-15', user: 'alice' }

-- Create memory nodes
NODE CREATE memory { content: 'User prefers dark mode', type: 'preference' }

-- Link memories to sessions
EDGE CREATE 'memory-id' -> 'session-id' : observed_in

-- Link related memories
EDGE CREATE 'memory-a' -> 'memory-b' : related_to

Store a Memory

INSERT INTO memories VALUES (1, 'User prefers dark mode', 'preference', 0.8, '2024-01-15');
NODE CREATE memory { content: 'User prefers dark mode', memory_id: 1 }
EMBED STORE 'memory-1' [0.1, 0.3, ...]

Recall

-- Semantic recall: find memories similar to current context
SIMILAR [0.2, 0.3, ...] LIMIT 5

-- Structured recall: get high-importance memories
SELECT * FROM memories WHERE importance > 0.7 ORDER BY importance DESC LIMIT 10

-- Graph recall: get memories from a specific session
NEIGHBORS 'session-id' INCOMING : observed_in

Knowledge Graph

Problem: Build a knowledge graph of entities with properties, relationships, and similarity search.

Why Neumann: Knowledge graphs need rich entity properties (relational), typed relationships (graph), and entity similarity (vector) in a single queryable system.

Build the Graph

-- Create entity types
NODE CREATE company { name: 'Acme Corp', industry: 'Technology', founded: 2010 }
NODE CREATE person { name: 'Jane Smith', title: 'CEO' }
NODE CREATE product { name: 'Acme Cloud', category: 'IaaS' }

-- Create relationships
EDGE CREATE 'jane-id' -> 'acme-id' : works_at { role: 'CEO', since: '2018' }
EDGE CREATE 'acme-id' -> 'product-id' : produces

Add Embeddings

-- Embed entity descriptions for similarity search
EMBED STORE 'acme-corp' [0.8, 0.3, 0.1, ...]
EMBED STORE 'jane-smith' [0.2, 0.7, 0.5, ...]
EMBED STORE 'acme-cloud' [0.9, 0.4, 0.2, ...]

Query

-- Find entities similar to a description
SIMILAR [0.7, 0.3, 0.2, ...] LIMIT 5

-- Discover connections
NEIGHBORS 'acme-id' BOTH
PATH SHORTEST 'jane-id' TO 'product-id'

-- Graph analytics
PAGERANK
LOUVAIN
BETWEENNESS

Access-Controlled Search

Problem: Build a search system where different users see different results based on permissions.

Why Neumann: Vault stores access tokens, graph models the permission hierarchy, and vector search handles the retrieval. No external auth system needed.

Setup Permissions

-- Store API keys securely
VAULT SET 'admin_key' 'ak-admin-secret'
VAULT SET 'user_key' 'ak-user-secret'

-- Grant access based on roles
VAULT GRANT 'alice' ON 'admin_key'
VAULT GRANT 'bob' ON 'user_key'

-- Build permission graph
NODE CREATE role { name: 'admin' }
NODE CREATE role { name: 'viewer' }
NODE CREATE resource { name: 'confidential-docs' }

EDGE CREATE 'admin-role-id' -> 'resource-id' : can_access

Query with Access Check

-- Find documents similar to query
SIMILAR [0.3, 0.5, ...] LIMIT 10

-- Verify access through graph
NEIGHBORS 'admin-role-id' OUTGOING : can_access

-- Rotate keys periodically
VAULT ROTATE 'admin_key' 'ak-new-admin-secret'

Common Patterns

Checkpoint Before Migrations

CHECKPOINT 'before-schema-v2'
-- Run migration...
-- If something goes wrong:
ROLLBACK TO 'before-schema-v2'

Blob Attachments

-- Upload a file and link it to an entity
BLOB INIT
BLOB PUT 'report.pdf' FROM '/tmp/report.pdf' TAG 'quarterly' LINK 'entity-id'

-- Find all blobs for an entity
BLOBS FOR 'entity-id'

-- Find blobs by tag
BLOBS BY TAG 'quarterly'

Chain for Audit Trail

-- Start a chain transaction for auditable operations
BEGIN CHAIN TRANSACTION
INSERT INTO audit_log VALUES (1, 'user_created', 'alice', '2024-01-15');
COMMIT CHAIN

-- Verify integrity
CHAIN VERIFY
CHAIN HISTORY 'audit_log/1'

Building from Source

Development Requirements

• Rust 1.75+ (stable)
• Rust nightly (for fuzzing)
• Git

Clone and Build

git clone https://github.com/Shadylukin/Neumann.git
cd Neumann

# Debug build
cargo build

# Release build (optimized)
cargo build --release

Running Tests

# Run all tests
cargo test

# Run tests for a specific crate
cargo test -p tensor_chain

# Run with output
cargo test -- --nocapture

Quality Checks

All code must pass before commit:

# Formatting
cargo fmt --check

# Lints (warnings as errors)
cargo clippy -- -D warnings

# Documentation builds
cargo doc --no-deps

Fuzzing

Requires nightly Rust:

# Install cargo-fuzz
cargo install cargo-fuzz

# Run a fuzz target
cd fuzz
cargo +nightly fuzz run parser_parse -- -max_total_time=60

Running the Shell

# Debug mode
cargo run -p neumann_shell

# Release mode
cargo run --release -p neumann_shell

IDE Setup

VS Code

Install the rust-analyzer extension. Recommended settings:

{
  "rust-analyzer.checkOnSave.command": "clippy",
  "rust-analyzer.cargo.features": "all"
}

IntelliJ/CLion

Install the Rust plugin. Enable clippy in settings.

Project Structure

Neumann/
├── tensor_store/       # Core storage layer
├── relational_engine/  # SQL-like tables
├── graph_engine/       # Graph operations
├── vector_engine/      # Embeddings
├── tensor_chain/       # Distributed consensus
├── neumann_parser/     # Query parsing
├── query_router/       # Query execution
├── neumann_shell/      # CLI interface
├── tensor_compress/    # Compression
├── tensor_vault/       # Encrypted storage
├── tensor_cache/       # LLM caching
├── tensor_blob/        # Blob storage
├── tensor_checkpoint/  # Snapshots
├── tensor_unified/     # Multi-engine facade
├── fuzz/               # Fuzz targets
└── docs/               # Documentation

Architecture Overview

Neumann is a unified tensor-based runtime that stores relational data, graph relationships, and vector embeddings in a single mathematical structure.

System Architecture

flowchart TB
    subgraph Client Layer
        Shell[neumann_shell]
    end

    subgraph Query Layer
        Router[query_router]
        Parser[neumann_parser]
    end

    subgraph Engine Layer
        RE[relational_engine]
        GE[graph_engine]
        VE[vector_engine]
    end

    subgraph Storage Layer
        TS[tensor_store]
        TC[tensor_compress]
    end

    subgraph Extended Modules
        Vault[tensor_vault]
        Cache[tensor_cache]
        Blob[tensor_blob]
        Check[tensor_checkpoint]
        Unified[tensor_unified]
        Chain[tensor_chain]
    end

    Shell --> Router
    Router --> Parser
    Router --> RE
    Router --> GE
    Router --> VE
    Router --> Vault
    Router --> Cache
    Router --> Blob
    Router --> Chain

    RE --> TS
    GE --> TS
    VE --> TS
    Vault --> TS
    Cache --> TS
    Blob --> TS
    Check --> TS
    Unified --> RE
    Unified --> GE
    Unified --> VE
    Chain --> TS
    Chain --> GE
    Chain --> Check

    TS --> TC

Module Dependencies

Key Design Principles

Unified Data Model

All data is represented as tensors:

• Scalars: Single values (int, float, string, bool)
• Vectors: Dense or sparse embeddings
• Pointers: References to other entities

Thread Safety

All engines use DashMap for concurrent access:

• Sharded locks for write throughput
• No lock poisoning
• Read operations are lock-free

Composability

Engines can be composed:

• Use relational_engine alone for SQL workloads
• Combine with graph_engine for relationship queries
• Add vector_engine for similarity search

Data Flow

1. Query Parsing: neumann_parser tokenizes and parses input
2. Query Routing: query_router dispatches to appropriate engine
3. Execution: Engine performs operation using tensor_store
4. Storage: tensor_store persists data with optional compression

Distributed Architecture (tensor_chain)

For distributed deployments:

flowchart LR
    subgraph Cluster
        L[Leader]
        F1[Follower 1]
        F2[Follower 2]
    end

    C[Client] --> L
    L --> F1
    L --> F2
    F1 -.-> L
    F2 -.-> L

• Raft Consensus: Leader election and log replication
• 2PC Transactions: Cross-shard atomic operations
• SWIM Gossip: Membership and failure detection

Tensor Store Architecture

The tensor_store crate is the foundational storage layer for Neumann. It provides a unified tensor-based key-value store that holds all data - relational, graph, and vector - in a single mathematical structure. The store knows nothing about queries; it purely stores and retrieves tensors by key.

The architecture uses SlabRouter internally, which routes operations to specialized slabs based on key prefixes. This design eliminates hash table resize stalls by using BTreeMap-based storage, providing predictable O(log n) performance without the throughput cliffs caused by hash map resizing.

Core Types

TensorValue

Represents different types of values a tensor can hold.

Automatic Sparsification: Use TensorValue::from_embedding_auto(dense) to automatically choose between dense and sparse representation based on sparsity:

#![allow(unused)]
fn main() {
// Automatically uses Sparse if sparsity >= 70%
let val = TensorValue::from_embedding_auto(dense_vec);

// With custom thresholds (value_threshold, sparsity_threshold)
let val = TensorValue::from_embedding(dense_vec, 0.01, 0.8);
}

Vector Operations: TensorValue supports cross-format operations:

#![allow(unused)]
fn main() {
// Dot product works across Dense, Sparse, and mixed
let dot = tensor_a.dot(&tensor_b);

// Cosine similarity with automatic format handling
let sim = tensor_a.cosine_similarity(&tensor_b);
}

ScalarValue

TensorData

An entity that holds scalar properties, vector embeddings, and pointers to other tensors via a HashMap<String, TensorValue> internally.

Reserved Field Names

Architecture Diagram

TensorStore
  |
  +-- Arc<SlabRouter>
         |
         +-- MetadataSlab (general key-value, BTreeMap-based)
         +-- EntityIndex (sorted vocabulary + hash index)
         +-- EmbeddingSlab (dense f32 arrays)
         +-- GraphTensor (CSR format for edges)
         +-- RelationalSlab (columnar storage)
         +-- CacheRing (LRU/LFU eviction)
         +-- BlobLog (append-only blob storage)

SlabRouter Internals

SlabRouter is the core routing layer that directs operations to specialized storage backends based on key prefixes.

Key Routing Algorithm

flowchart TD
    A[put/get/delete key] --> B{Classify Key}
    B -->|emb:*| C[EmbeddingSlab + MetadataSlab]
    B -->|node:* / edge:*| D[GraphTensor via MetadataSlab]
    B -->|table:*| E[RelationalSlab via MetadataSlab]
    B -->|_cache:*| F[CacheRing]
    B -->|Everything else| G[MetadataSlab]

Key Classification

SlabRouter Operation Flow

PUT Operation:

#![allow(unused)]
fn main() {
fn put(&self, key: &str, value: TensorData) {
    match classify_key(key) {
        KeyClass::Embedding => {
            // 1. Get or create stable entity ID
            let entity_id = self.index.get_or_create(key);
            // 2. Extract and store embedding vector
            if let Some(TensorValue::Vector(vec)) = value.get("_embedding") {
                self.embeddings.set(entity_id, vec);
            }
            // 3. Store full metadata
            self.metadata.set(key, value);
        }
        KeyClass::Cache => {
            let size = estimate_size(&value);
            self.cache.put(key, value, 1.0, size);
        }
        _ => self.metadata.set(key, value),
    }
}
}

GET Operation:

#![allow(unused)]
fn main() {
fn get(&self, key: &str) -> Result<TensorData> {
    match classify_key(key) {
        KeyClass::Embedding => {
            // Try to reconstruct from embedding slab + metadata
            if let Some(entity_id) = self.index.get(key) {
                if let Some(vector) = self.embeddings.get(entity_id) {
                    let mut data = self.metadata.get(key).unwrap_or_default();
                    data.set("_embedding", TensorValue::Vector(vector));
                    return Ok(data);
                }
            }
            self.metadata.get(key)
        }
        KeyClass::Cache => self.cache.get(key),
        _ => self.metadata.get(key),
    }
}
}

Specialized Slabs

Performance Characteristics

Operation Complexity

Throughput Comparison

Optimized Scan Performance

Use scan_filter_map for selective queries to avoid cloning non-matching entries:

#![allow(unused)]
fn main() {
// Old path: 5000 clones for 5000 rows, ~2.6ms
let users = store.scan("users:");
let matches: Vec<_> = users.iter()
    .filter_map(|key| store.get(key).ok())
    .filter(|data| /* condition */)
    .collect();

// New path: 250 clones for 5% match rate, ~0.13ms (20x faster)
let matches = store.scan_filter_map("users:", |key, data| {
    if /* condition */ {
        Some(data.clone())
    } else {
        None
    }
});
}

Concurrency Model

TensorStore uses tensor-based structures instead of hash maps for predictable performance:

• No Resize Stalls: BTreeMap and sorted arrays grow incrementally
• Lock-free Reads: RwLock allows many concurrent readers
• Predictable Writes: O(log n) inserts, no amortized O(n) resizing
• Clone on Read: get() returns cloned data to avoid holding references
• Shareable Storage: TensorStore clones share the same underlying data via Arc

BloomFilter

The BloomFilter provides O(1) probabilistic rejection of non-existent keys, useful for sparse key spaces where most lookups are misses.

Mathematical Foundation

The Bloom filter uses optimal parameters calculated as:

Bit array size: m = -n * ln(p) / (ln(2)^2)

• Where n = expected items, p = false positive rate

Number of hash functions: k = (m/n) * ln(2)

• Clamped to range 1 to 16

Implementation Details

#![allow(unused)]
fn main() {
pub struct BloomFilter {
    bits: Box<[AtomicU64]>,  // Atomic u64 blocks for lock-free access
    num_bits: usize,
    num_hashes: usize,
}
}

Hash Function: Uses SipHash with different seeds for each hash function:

#![allow(unused)]
fn main() {
fn hash_index<K: Hash>(&self, key: &K, seed: usize) -> usize {
    let mut hasher = SipHasher::new_with_seed(seed as u64);
    key.hash(&mut hasher);
    (hasher.finish() as usize) % self.num_bits
}
}

Parameter Tuning Guide

Gotchas:

• Bloom filter state is not persisted in snapshots; rebuild after load
• Thread-safe via AtomicU64 with Relaxed ordering (eventual consistency)
• Cannot remove items (use counting bloom filter for that case)
• False positive rate increases if more items than expected are inserted

HNSW Index

Hierarchical Navigable Small World index for approximate nearest neighbor search with O(log n) complexity.

Algorithm Overview

flowchart TD
    subgraph "HNSW Structure"
        L3[Layer 3: Entry Point] --> L2[Layer 2: Skip connections]
        L2 --> L1[Layer 1: More connections]
        L1 --> L0[Layer 0: All nodes, dense connections]
    end

    subgraph "Search Algorithm"
        S1[Start at entry point, top layer] --> S2[Greedy descent to layer 1]
        S2 --> S3[At layer 0: ef-search candidates]
        S3 --> S4[Return top-k results]
    end

Layer Selection

New nodes are assigned layers using exponential distribution:

#![allow(unused)]
fn main() {
fn random_level(&self) -> usize {
    let f = random_float_0_1();
    let level = (-f.ln() * self.config.ml).floor() as usize;
    level.min(32)  // Cap at 32 layers
}
}

Where ml = 1 / ln(m) and m = connections per layer.

HNSWConfig Parameters

Configuration Presets

#![allow(unused)]
fn main() {
// High recall (slower, more accurate)
HNSWConfig::high_recall()  // m=32, m0=64, ef_construction=400, ef_search=200

// High speed (faster, lower recall)
HNSWConfig::high_speed()   // m=8, m0=16, ef_construction=100, ef_search=20

// Custom configuration
HNSWConfig {
    m: 24,
    m0: 48,
    ef_construction: 300,
    ef_search: 100,
    ..Default::default()
}
}

SIMD-Accelerated Distance

Dense vector operations use 8-wide SIMD (f32x8):

#![allow(unused)]
fn main() {
pub fn dot_product(a: &[f32], b: &[f32]) -> f32 {
    let chunks = a.len() / 8;
    let mut sum = f32x8::ZERO;

    for i in 0..chunks {
        let offset = i * 8;
        let va = f32x8::from(&a[offset..offset + 8]);
        let vb = f32x8::from(&b[offset..offset + 8]);
        sum += va * vb;
    }

    // Sum lanes and handle remainder
    let arr: [f32; 8] = sum.into();
    let mut result: f32 = arr.iter().sum();
    // ... scalar remainder handling
}
}

Neighbor Compression

HNSW neighbor lists use delta-varint encoding for 3-8x compression:

#![allow(unused)]
fn main() {
struct CompressedNeighbors {
    compressed: Vec<u8>,  // Delta-varint encoded neighbor IDs
}

// Decompression: O(n) where n = neighbor count
fn get(&self) -> Vec<usize> {
    decompress_ids(&self.compressed)
}

// Compression: Sort + delta encode
fn set(&mut self, ids: &[usize]) {
    let mut sorted = ids.to_vec();
    sorted.sort_unstable();
    self.compressed = compress_ids(&sorted);
}
}

Storage Types

flowchart LR
    subgraph "EmbeddingStorage"
        D[Dense: Vec f32]
        S[Sparse: SparseVector]
        DV[Delta: DeltaVector]
        TT[TensorTrain: TTVectorCached]
    end

    D --> |"sparsity > 50%"| S
    D --> |"clusters around archetype"| DV
    D --> |"high-dim 768+"| TT

Edge Cases and Gotchas

1. Delta vectors cannot be inserted directly - they require archetype registry for distance computation. Convert to Dense first.

2. TensorTrain storage - While stored in TT format, HNSW reconstructs to dense for fast distance computation during search (native TT distance is O(r^4) per comparison).

3. Capacity limits - Default max_nodes=10M prevents memory exhaustion from fuzzing/adversarial input. Use try_insert for graceful handling.

4. Empty index - Entry point is usize::MAX when empty; search returns empty results.

SparseVector

Memory-efficient storage for vectors with many zeros, based on the philosophy that “zero represents absence of information, not a stored value.”

Internal Structure

#![allow(unused)]
fn main() {
pub struct SparseVector {
    dimension: usize,      // Total dimension (shell/boundary)
    positions: Vec<u32>,   // Sorted positions of non-zero values
    values: Vec<f32>,      // Corresponding values
}
}

Operation Complexity

Sparse Arithmetic Operations

#![allow(unused)]
fn main() {
// Create delta from before/after states (only stores differences)
let delta = SparseVector::from_diff(&before, &after, threshold);

// Subtraction: self - other
let diff = a.sub(&b);

// Weighted average: (w1 * a + w2 * b) / (w1 + w2)
let merged = a.weighted_average(&b, 0.7, 0.3);

// Project out conflicting component
let orthogonal = v.project_orthogonal(&conflict_direction);
}

Distance Metrics

Security: NaN/Inf Sanitization

All similarity metrics sanitize results to prevent consensus ordering issues:

#![allow(unused)]
fn main() {
pub fn cosine_similarity(&self, other: &SparseVector) -> f32 {
    // ... computation ...

    // SECURITY: Sanitize result to valid range
    if result.is_nan() || result.is_infinite() {
        0.0
    } else {
        result.clamp(-1.0, 1.0)
    }
}
}

Memory Efficiency

#![allow(unused)]
fn main() {
let sparse = SparseVector::from_dense(&dense_vec);

// Metrics
sparse.sparsity()           // Fraction of zeros (0.0 - 1.0)
sparse.memory_bytes()       // Actual memory used
sparse.dense_memory_bytes() // Memory if stored dense
sparse.compression_ratio()  // Dense / Sparse ratio
}

For a 1000-dim vector with 90% zeros:

• Dense: 4000 bytes
• Sparse: ~800 bytes (100 positions 4 bytes + 100 values 4 bytes)
• Compression ratio: 5x

Delta Vectors and Archetype Registry

Delta encoding stores vectors as differences from reference “archetype” vectors, providing significant compression for clustered embeddings.

Concept

flowchart LR
    subgraph "Delta Encoding"
        A[Archetype Vector] --> |"+ Delta"| R[Reconstructed Vector]
        D[Delta: positions + values] --> R
    end

When many embeddings cluster around common patterns:

• Identify archetype vectors (cluster centroids via k-means)
• Store each embedding as: archetype_id + sparse_delta
• Reconstruct on demand: archetype + delta = original

DeltaVector Structure

#![allow(unused)]
fn main() {
pub struct DeltaVector {
    archetype_id: usize,       // Reference archetype
    dimension: usize,          // For reconstruction
    positions: Vec<u16>,       // Diff positions (u16 for memory)
    deltas: Vec<f32>,          // Delta values
    cached_magnitude: Option<f32>,  // For fast cosine similarity
}
}

Optimized Dot Products

#![allow(unused)]
fn main() {
// With precomputed archetype dot query
// Total: O(nnz) instead of O(dimension)
let result = delta.dot_dense_with_precomputed(query, archetype_dot_query);

// Between two deltas from SAME archetype
// dot(A, B) = dot(R, R) + dot(R, delta_b) + dot(delta_a, R) + dot(delta_a, delta_b)
let result = a.dot_same_archetype(&b, archetype, archetype_magnitude_sq);
}

ArchetypeRegistry

#![allow(unused)]
fn main() {
// Create registry with max 16 archetypes
let mut registry = ArchetypeRegistry::new(16);

// Discover archetypes via k-means clustering
let config = KMeansConfig {
    max_iterations: 100,
    convergence_threshold: 1e-4,
    seed: 42,
    init_method: KMeansInit::KMeansPlusPlus,  // Better but slower
};
registry.discover_archetypes(&embeddings, 5, config);

// Encode vectors as deltas
let delta = registry.encode(&vector, threshold)?;

// Analyze coverage
let stats = registry.analyze_coverage(&vectors, 0.01);
// stats.avg_similarity, stats.avg_compression_ratio, stats.archetype_usage
}

Persistence

#![allow(unused)]
fn main() {
// Save to TensorStore
registry.save_to_store(&store)?;

// Load from TensorStore
let registry = ArchetypeRegistry::load_from_store(&store, 16)?;
}

Tiered Storage

Two-tier storage with hot (in-memory) and cold (mmap) layers for memory-efficient storage of large datasets.

Architecture

flowchart TD
    subgraph "Hot Tier (In-Memory)"
        H[MetadataSlab]
        I[ShardAccessTracker]
    end

    subgraph "Cold Tier (Mmap)"
        C[MmapStoreMut]
        CK[cold_keys HashSet]
    end

    GET --> H
    H -->|miss| CK
    CK -->|found| C
    C -->|promote| H

    PUT --> H
    H -->|migrate_cold| C

TieredConfig

Migration Algorithm

#![allow(unused)]
fn main() {
pub fn migrate_cold(&mut self, threshold_ms: u64) -> Result<usize> {
    // 1. Find shards not accessed within threshold
    let cold_shards = self.instrumentation.cold_shards(threshold_ms);

    // 2. Collect keys belonging to cold shards
    let keys_to_migrate: Vec<String> = self.hot.scan("")
        .filter(|(key, _)| {
            let shard = shard_for_key(key);
            cold_shards.contains(&shard)
        })
        .map(|(key, _)| key)
        .collect();

    // 3. Move to cold storage
    for key in keys_to_migrate {
        cold.insert(&key, &tensor)?;
        self.cold_keys.insert(key.clone());
        self.hot.delete(&key);
    }

    cold.flush()?;
}
}

Automatic Promotion

When cold data is accessed, it’s automatically promoted back to hot:

#![allow(unused)]
fn main() {
pub fn get(&mut self, key: &str) -> Result<TensorData> {
    // Try hot first
    if let Some(data) = self.hot.get(key) {
        return Ok(data);
    }

    // Try cold
    if self.cold_keys.contains(key) {
        let tensor = self.cold.get(key)?;

        // Promote to hot
        self.hot.set(key, tensor.clone());
        self.cold_keys.remove(key);
        self.migrations_to_hot.fetch_add(1, Ordering::Relaxed);

        return Ok(tensor);
    }

    Err(TensorStoreError::NotFound(key))
}
}

Statistics

#![allow(unused)]
fn main() {
let stats = store.stats();
// stats.hot_count, stats.cold_count
// stats.hot_lookups, stats.cold_lookups, stats.cold_hits
// stats.migrations_to_cold, stats.migrations_to_hot
}

Access Instrumentation

Low-overhead tracking of shard access patterns for intelligent memory tiering.

ShardAccessTracker

#![allow(unused)]
fn main() {
pub struct ShardAccessTracker {
    shards: Box<[ShardStats]>,     // Per-shard counters
    shard_count: usize,            // Default: 16
    start_time: Instant,           // For last_access timestamps
    sample_rate: u32,              // 1 = every access, 100 = 1%
    sample_counter: AtomicU64,     // For sampling
}

// Sampling logic
fn should_sample(&self) -> bool {
    if self.sample_rate == 1 { return true; }
    self.sample_counter.fetch_add(1, Relaxed).is_multiple_of(self.sample_rate)
}
}

Hot/Cold Detection

#![allow(unused)]
fn main() {
// Get shards sorted by access count (hottest first)
let hot = tracker.hot_shards(5);  // Top 5 hottest

// Get shards not accessed within threshold
let cold = tracker.cold_shards(30_000);  // Not accessed in 30s
}

HNSW Access Stats

Specialized instrumentation for HNSW index:

#![allow(unused)]
fn main() {
pub struct HNSWAccessStats {
    entry_point_accesses: AtomicU64,
    layer0_traversals: AtomicU64,
    upper_layer_traversals: AtomicU64,
    total_searches: AtomicU64,
    distance_calculations: AtomicU64,
}

// Snapshot metrics
let stats = hnsw.access_stats()?;
stats.layer0_ratio()          // Layer 0 work fraction
stats.avg_distances_per_search  // Distance calcs per search
stats.searches_per_second()   // Throughput
}

Configuration Options

SlabRouterConfig

Usage Examples

Basic Operations

#![allow(unused)]
fn main() {
let store = TensorStore::new();

// Store a tensor
let mut user = TensorData::new();
user.set("name", TensorValue::Scalar(ScalarValue::String("Alice".into())));
user.set("age", TensorValue::Scalar(ScalarValue::Int(30)));
user.set("embedding", TensorValue::Vector(vec![0.1, 0.2, 0.3, 0.4]));
store.put("user:1", user)?;

// Retrieve
let data = store.get("user:1")?;

// Scan by prefix
let user_keys = store.scan("user:");
let count = store.scan_count("user:");
}

With Bloom Filter

#![allow(unused)]
fn main() {
// Fast rejection of non-existent keys
let store = TensorStore::with_bloom_filter(10_000, 0.01);
store.put("key:1", tensor)?;

// O(1) rejection if key definitely doesn't exist
if store.exists("key:999") { /* ... */ }
}

With Instrumentation

#![allow(unused)]
fn main() {
// Enable access tracking with 1% sampling
let store = TensorStore::with_instrumentation(100);

// After operations, check access patterns
let snapshot = store.access_snapshot()?;
println!("Hot shards: {:?}", store.hot_shards(5)?);
println!("Cold shards: {:?}", store.cold_shards(30_000)?);
}

Shared Storage Across Engines

#![allow(unused)]
fn main() {
let store = TensorStore::new();

// Clone shares the same underlying Arc<SlabRouter>
let store_clone = store.clone();

// Both see the same data
store.put("user:1", user_data)?;
assert!(store_clone.exists("user:1"));

// Use with multiple engines
let vector_engine = VectorEngine::with_store(store.clone());
let graph_engine = GraphEngine::with_store(store.clone());
}

Persistence

#![allow(unused)]
fn main() {
// Save snapshot
store.save_snapshot("data.bin")?;

// Load snapshot
let store = TensorStore::load_snapshot("data.bin")?;

// Load with Bloom filter rebuild
let store = TensorStore::load_snapshot_with_bloom_filter(
    "data.bin",
    10_000,   // expected items
    0.01      // false positive rate
)?;

// Compressed snapshot
use tensor_compress::{CompressionConfig, QuantMode};
let config = CompressionConfig {
    vector_quantization: Some(QuantMode::Int8),  // 4x compression
    delta_encoding: true,
    rle_encoding: true,
};
store.save_snapshot_compressed("data.bin", config)?;
}

Tiered Storage

#![allow(unused)]
fn main() {
use tensor_store::{TieredStore, TieredConfig};

let config = TieredConfig {
    cold_dir: "/data/cold".into(),
    cold_capacity: 64 * 1024 * 1024,
    sample_rate: 100,
};

let mut store = TieredStore::new(config)?;
store.put("user:1", tensor);

// Migrate cold data (not accessed in 30s)
let migrated = store.migrate_cold(30_000)?;

// Check stats
let stats = store.stats();
println!("Hot: {}, Cold: {}", stats.hot_count, stats.cold_count);
}

HNSW Index

#![allow(unused)]
fn main() {
let index = HNSWIndex::with_config(HNSWConfig::default());

// Insert dense, sparse, or auto-select
index.insert(vec![0.1, 0.2, 0.3]);
index.insert_sparse(sparse_vec);
index.insert_auto(mixed_vec);  // Auto-selects dense/sparse

// With capacity checking
match index.try_insert(vec) {
    Ok(id) => println!("Inserted as node {}", id),
    Err(EmbeddingStorageError::CapacityExceeded { limit, current }) => {
        println!("Index full: {} / {}", current, limit);
    }
}

// Search with custom ef
let results = index.search_with_ef(&query, 10, 100);
for (id, similarity) in results {
    println!("Node {}: {:.4}", id, similarity);
}
}

Delta-Encoded Embeddings

#![allow(unused)]
fn main() {
let mut registry = ArchetypeRegistry::new(16);

// Discover archetypes from existing embeddings
registry.discover_archetypes(&embeddings, 5, KMeansConfig::default());

// Encode new vectors as deltas
let results = registry.encode_batch(&embeddings, 0.01);
for (delta, compression_ratio) in results {
    println!("Archetype {}, compression: {:.2}x",
             delta.archetype_id(), compression_ratio);
}
}

Error Types

TensorStoreError

SnapshotError

TieredError

EmbeddingStorageError

Related Modules

Dependencies

Relational Engine

The Relational Engine (Module 2) provides SQL-like table operations on top of the Tensor Store. It implements schema enforcement, composable condition predicates, SIMD-accelerated columnar filtering, and both hash and B-tree indexes for query acceleration.

Tables, rows, and indexes are stored as tensor data in the underlying Tensor Store, inheriting its thread safety from DashMap. The engine supports all standard CRUD operations, six SQL join types, aggregate functions, and batch operations for bulk inserts.

Architecture

flowchart TD
    subgraph RelationalEngine
        API[Public API]
        Schema[Schema Validation]
        Cond[Condition Evaluation]
        Hash[Hash Index]
        BTree[B-Tree Index]
        Columnar[Columnar SIMD]
    end

    API --> Schema
    API --> Cond
    Cond --> Hash
    Cond --> BTree
    Cond --> Columnar

    subgraph TensorStore
        Store[(DashMap Storage)]
        Meta[Table Metadata]
        Rows[Row Data]
        Idx[Index Entries]
    end

    Schema --> Meta
    API --> Rows
    Hash --> Idx
    BTree --> Idx

Query Execution Flow

flowchart TD
    Query[SELECT Query] --> ParseCond[Parse Condition]
    ParseCond --> CheckIdx{Has Index?}

    CheckIdx -->|Hash Index + Eq| HashLookup[O(1) Hash Lookup]
    CheckIdx -->|BTree + Range| BTreeRange[O(log n) Range Scan]
    CheckIdx -->|No Index| FullScan[Full Table Scan]

    HashLookup --> FilterRows[Apply Remaining Conditions]
    BTreeRange --> FilterRows
    FullScan --> SIMDFilter{Columnar Data?}

    SIMDFilter -->|Yes| VectorFilter[SIMD Vectorized Filter]
    SIMDFilter -->|No| RowFilter[Row-by-Row Filter]

    VectorFilter --> Results[Build Result Set]
    RowFilter --> Results
    FilterRows --> Results

Key Types

Column Types

Conditions

Conditions can be combined using .and() and .or() methods:

#![allow(unused)]
fn main() {
// age >= 18 AND age < 65
let condition = Condition::Ge("age".into(), Value::Int(18))
    .and(Condition::Lt("age".into(), Value::Int(65)));

// status = 'active' OR priority > 5
let condition = Condition::Eq("status".into(), Value::String("active".into()))
    .or(Condition::Gt("priority".into(), Value::Int(5)));
}

The special column _id filters by row ID and can be indexed.

Error Types

Storage Model

Tables, rows, and indexes are stored in Tensor Store with specific key patterns:

Schema metadata encodes:

• _columns: Comma-separated column names
• _col:{name}: Type and nullability for each column

Row Storage Format

Each row is stored as a TensorData object:

#![allow(unused)]
fn main() {
// Internal row structure
{
    "_id": Scalar(Int(row_id)),
    "name": Scalar(String("Alice")),
    "age": Scalar(Int(30)),
    "email": Scalar(String("alice@example.com"))
}
}

Usage Examples

Table Operations

#![allow(unused)]
fn main() {
let engine = RelationalEngine::new();

// Create table with schema
let schema = Schema::new(vec![
    Column::new("name", ColumnType::String),
    Column::new("age", ColumnType::Int),
    Column::new("email", ColumnType::String).nullable(),
]);
engine.create_table("users", schema)?;

// Check existence
engine.table_exists("users")?;  // -> bool

// List all tables
let tables = engine.list_tables();  // -> Vec<String>

// Get schema
let schema = engine.get_schema("users")?;

// Drop table (deletes all rows and indexes)
engine.drop_table("users")?;

// Row count
engine.row_count("users")?;  // -> usize
}

CRUD Operations

#![allow(unused)]
fn main() {
// INSERT
let mut values = HashMap::new();
values.insert("name".to_string(), Value::String("Alice".into()));
values.insert("age".to_string(), Value::Int(30));
let row_id = engine.insert("users", values)?;

// BATCH INSERT (59x faster for bulk inserts)
let rows: Vec<HashMap<String, Value>> = (0..1000)
    .map(|i| {
        let mut values = HashMap::new();
        values.insert("name".to_string(), Value::String(format!("User{}", i)));
        values.insert("age".to_string(), Value::Int(20 + i));
        values
    })
    .collect();
let row_ids = engine.batch_insert("users", rows)?;

// SELECT
let rows = engine.select("users", Condition::Eq("age".into(), Value::Int(30)))?;

// UPDATE
let mut updates = HashMap::new();
updates.insert("age".to_string(), Value::Int(31));
let count = engine.update(
    "users",
    Condition::Eq("name".into(), Value::String("Alice".into())),
    updates
)?;

// DELETE
let count = engine.delete_rows("users", Condition::Lt("age".into(), Value::Int(18)))?;
}

Constraints

The engine supports four constraint types for data integrity:

#![allow(unused)]
fn main() {
use relational_engine::{Constraint, ForeignKeyConstraint, ReferentialAction};

// Create table with constraints
let schema = Schema::with_constraints(
    vec![
        Column::new("id", ColumnType::Int),
        Column::new("email", ColumnType::String),
        Column::new("dept_id", ColumnType::Int).nullable(),
    ],
    vec![
        Constraint::primary_key("pk_users", vec!["id".to_string()]),
        Constraint::unique("uq_email", vec!["email".to_string()]),
    ],
);
engine.create_table("users", schema)?;

// Add constraint after table creation
engine.add_constraint("users", Constraint::not_null("nn_email", "email"))?;

// Add foreign key with referential actions
let fk = ForeignKeyConstraint::new(
    "fk_users_dept",
    vec!["dept_id".to_string()],
    "departments",
    vec!["id".to_string()],
)
.on_delete(ReferentialAction::SetNull)
.on_update(ReferentialAction::Cascade);
engine.add_constraint("users", Constraint::foreign_key(fk))?;

// Get constraints
let constraints = engine.get_constraints("users")?;

// Drop constraint
engine.drop_constraint("users", "uq_email")?;
}

Referential Actions

Foreign keys support these actions on delete/update of referenced rows:

ALTER TABLE Operations

#![allow(unused)]
fn main() {
// Add a new column (nullable or with default)
engine.add_column("users", Column::new("phone", ColumnType::String).nullable())?;

// Drop a column (fails if column has constraints)
engine.drop_column("users", "phone")?;

// Rename a column (updates constraints automatically)
engine.rename_column("users", "email", "email_address")?;
}

Joins

All six SQL join types are supported using hash join algorithm (O(n+m)):

#![allow(unused)]
fn main() {
// INNER JOIN - Only matching rows from both tables
let joined = engine.join("users", "posts", "_id", "user_id")?;
// Returns: Vec<(Row, Row)>

// LEFT JOIN - All rows from left, matching from right (or None)
let joined = engine.left_join("users", "posts", "_id", "user_id")?;
// Returns: Vec<(Row, Option<Row>)>

// RIGHT JOIN - All rows from right, matching from left (or None)
let joined = engine.right_join("users", "posts", "_id", "user_id")?;
// Returns: Vec<(Option<Row>, Row)>

// FULL JOIN - All rows from both tables
let joined = engine.full_join("users", "posts", "_id", "user_id")?;
// Returns: Vec<(Option<Row>, Option<Row>)>

// CROSS JOIN (Cartesian product)
let joined = engine.cross_join("users", "posts")?;
// Returns: Vec<(Row, Row)> with n*m rows

// NATURAL JOIN (on common column names)
let joined = engine.natural_join("users", "user_profiles")?;
// Returns: Vec<(Row, Row)> matching on all common columns
}

Aggregate Functions

#![allow(unused)]
fn main() {
// COUNT(*) - count all rows
let count = engine.count("users", Condition::True)?;

// COUNT(column) - count non-null values
let count = engine.count_column("users", "email", Condition::True)?;

// SUM - returns f64
let total = engine.sum("orders", "amount", Condition::True)?;

// AVG - returns Option<f64> (None if no matching rows)
let avg = engine.avg("orders", "amount", Condition::True)?;

// MIN/MAX - returns Option<Value>
let min = engine.min("products", "price", Condition::True)?;
let max = engine.max("products", "price", Condition::True)?;
}

Indexes

Hash Indexes

Hash indexes provide O(1) equality lookups for Condition::Eq queries:

#![allow(unused)]
fn main() {
// Create hash index
engine.create_index("users", "age")?;

// Check existence
engine.has_index("users", "age");  // -> bool

// Get indexed columns
engine.get_indexed_columns("users");  // -> Vec<String>

// Drop index
engine.drop_index("users", "age")?;
}

Hash Index Implementation Details:

graph LR
    subgraph "Hash Index Structure"
        Value[Column Value] --> Hash[hash_key()]
        Hash --> Bucket["_idx:table:col:hash"]
        Bucket --> IDs["Vec<row_id>"]
    end

The hash index uses value-specific hashing:

Hash index performance:

B-Tree Indexes

B-tree indexes accelerate range queries (Lt, Le, Gt, Ge) with O(log n + m) complexity:

#![allow(unused)]
fn main() {
// Create B-tree index
engine.create_btree_index("users", "age")?;

// Check existence
engine.has_btree_index("users", "age");  // -> bool

// Get B-tree indexed columns
engine.get_btree_indexed_columns("users");  // -> Vec<String>

// Drop index
engine.drop_btree_index("users", "age")?;

// Range queries now use the index
engine.select("users", Condition::Ge("age".into(), Value::Int(18)))?;
}

B-Tree Index Implementation Details:

The B-tree index uses a dual-storage approach:

1. In-memory BTreeMap: For O(log n) range operations
2. Persistent TensorStore: For durability and recovery

#![allow(unused)]
fn main() {
// Internal B-tree index structure
btree_indexes: RwLock<HashMap<
    (String, String),           // (table, column)
    BTreeMap<OrderedKey, Vec<u64>>  // value -> row_ids
>>
}

OrderedKey for Total Ordering:

The OrderedKey enum provides correct ordering semantics:

#![allow(unused)]
fn main() {
pub enum OrderedKey {
    Null,                    // Sorts first
    Bool(bool),              // false < true
    Int(i64),                // Standard integer ordering
    Float(OrderedFloat),     // NaN < all other values
    String(String),          // Lexicographic ordering
}
}

Sortable Key Encoding:

For persistent storage, values are encoded to maintain lexicographic ordering:

Integer encoding shifts the range from [-2^63, 2^63-1] to [0, 2^64-1] for correct lexicographic ordering of negative numbers.

B-Tree Range Operations:

#![allow(unused)]
fn main() {
// Internal range lookup
fn btree_range_lookup(&self, table: &str, column: &str,
                      value: &Value, op: RangeOp) -> Option<Vec<u64>> {
    match op {
        RangeOp::Lt => btree.range(..target),
        RangeOp::Le => btree.range(..=target),
        RangeOp::Gt => btree.range((Excluded(target), Unbounded)),
        RangeOp::Ge => btree.range(target..),
    }
}
}

Columnar Architecture

The engine uses columnar storage with SIMD-accelerated filtering:

Columnar Data Structures

graph TD
    subgraph "ColumnData"
        Name[name: String]
        RowIDs[row_ids: Vec<u64>]
        Nulls[nulls: NullBitmap]
        Values[values: ColumnValues]
    end

    subgraph "ColumnValues Variants"
        Int["Int(Vec<i64>)"]
        Float["Float(Vec<f64>)"]
        String["String { dict, indices }"]
        Bool["Bool(Vec<u64>)"]
    end

    subgraph "NullBitmap Variants"
        None["None (no nulls)"]
        Dense["Dense(Vec<u64>)"]
        Sparse["Sparse(Vec<u64>)"]
    end

    Values --> Int
    Values --> Float
    Values --> String
    Values --> Bool
    Nulls --> None
    Nulls --> Dense
    Nulls --> Sparse

Null Bitmap Selection:

• None: When column has no null values
• Sparse: When nulls are < 10% of rows (stores positions)
• Dense: When nulls are >= 10% of rows (stores bitmap)

SIMD Filtering

Column data is stored in contiguous arrays enabling 4-wide SIMD vectorized comparisons using the wide crate:

#![allow(unused)]
fn main() {
// SIMD filter implementation using wide::i64x4
pub fn filter_lt_i64(values: &[i64], threshold: i64, result: &mut [u64]) {
    let chunks = values.len() / 4;
    let threshold_vec = i64x4::splat(threshold);

    for i in 0..chunks {
        let offset = i * 4;
        let v = i64x4::new([
            values[offset],
            values[offset + 1],
            values[offset + 2],
            values[offset + 3],
        ]);
        let cmp = v.cmp_lt(threshold_vec);
        let mask_arr: [i64; 4] = cmp.into();

        for (j, &m) in mask_arr.iter().enumerate() {
            if m != 0 {
                let bit_pos = offset + j;
                result[bit_pos / 64] |= 1u64 << (bit_pos % 64);
            }
        }
    }

    // Handle remainder with scalar fallback
    let start = chunks * 4;
    for i in start..values.len() {
        if values[i] < threshold {
            result[i / 64] |= 1u64 << (i % 64);
        }
    }
}
}

Available SIMD Filter Functions:

Bitmap Operations:

#![allow(unused)]
fn main() {
// AND two selection bitmaps
pub fn bitmap_and(a: &[u64], b: &[u64], result: &mut [u64])

// OR two selection bitmaps
pub fn bitmap_or(a: &[u64], b: &[u64], result: &mut [u64])

// Count set bits
pub fn popcount(bitmap: &[u64]) -> usize

// Extract selected indices
pub fn selected_indices(bitmap: &[u64], max_count: usize) -> Vec<usize>
}

Selection Vectors

Query results use bitmap-based selection vectors to avoid copying data:

#![allow(unused)]
fn main() {
pub struct SelectionVector {
    bitmap: Vec<u64>,  // Packed bits indicating selected rows
    row_count: usize,
}

impl SelectionVector {
    // Create selection of all rows
    pub fn all(row_count: usize) -> Self;

    // Create empty selection
    pub fn none(row_count: usize) -> Self;

    // Check if row is selected
    pub fn is_selected(&self, idx: usize) -> bool;

    // Count selected rows
    pub fn count(&self) -> usize;

    // AND two selections (intersection)
    pub fn intersect(&self, other: &SelectionVector) -> SelectionVector;

    // OR two selections (union)
    pub fn union(&self, other: &SelectionVector) -> SelectionVector;
}
}

Columnar Select API

#![allow(unused)]
fn main() {
// Materialize columns for SIMD filtering
engine.materialize_columns("users", &["age", "name"])?;

// Check if columnar data exists
engine.has_columnar_data("users", "age");  // -> bool

// Select with columnar scan options
let options = ColumnarScanOptions {
    projection: Some(vec!["name".into()]),  // Only return these columns
    prefer_columnar: true,                   // Use SIMD when available
};

let rows = engine.select_columnar(
    "users",
    Condition::Gt("age".into(), Value::Int(50)),
    options
)?;

// Drop columnar data
engine.drop_columnar_data("users", "age")?;
}

Condition Evaluation

Two evaluation methods are available:

The engine automatically chooses the optimal evaluation path:

flowchart TD
    Cond[Condition] --> CheckColumnar{Columnar Data Available?}
    CheckColumnar -->|Yes| CheckType{Int Column?}
    CheckColumnar -->|No| RowEval[evaluate_tensor per row]

    CheckType -->|Yes| SIMDEval[SIMD Vectorized Filter]
    CheckType -->|No| RowEval

    SIMDEval --> Bitmap[Selection Bitmap]
    RowEval --> Filter[Filter Matching Rows]

    Bitmap --> Materialize[Materialize Results]
    Filter --> Materialize

Join Algorithm Implementations

Hash Join (INNER, LEFT, RIGHT, FULL)

All equality joins use the hash join algorithm with O(n+m) complexity:

flowchart LR
    subgraph "Build Phase"
        RightTable[Right Table] --> BuildHash[Build Hash Index]
        BuildHash --> HashIndex["HashMap<hash, Vec<idx>>"]
    end

    subgraph "Probe Phase"
        LeftTable[Left Table] --> Probe[Probe Hash Index]
        Probe --> HashIndex
        HashIndex --> Match[Find Matching Rows]
    end

    Match --> Results[Join Results]

Hash Join Implementation:

#![allow(unused)]
fn main() {
pub fn join(&self, table_a: &str, table_b: &str,
            on_a: &str, on_b: &str) -> Result<Vec<(Row, Row)>> {
    let rows_a = self.select(table_a, Condition::True)?;
    let rows_b = self.select(table_b, Condition::True)?;

    // Build phase: index the right table
    let mut index: HashMap<String, Vec<usize>> = HashMap::with_capacity(rows_b.len());
    for (i, row) in rows_b.iter().enumerate() {
        if let Some(val) = row.get_with_id(on_b) {
            let hash = val.hash_key();
            index.entry(hash).or_default().push(i);
        }
    }

    // Probe phase: scan left table and probe index
    let mut results = Vec::with_capacity(min(rows_a.len(), rows_b.len()));
    for row_a in &rows_a {
        if let Some(val) = row_a.get_with_id(on_a) {
            let hash = val.hash_key();
            if let Some(indices) = index.get(&hash) {
                for &i in indices {
                    let row_b = &rows_b[i];
                    // Verify actual equality (handles hash collisions)
                    if row_b.get_with_id(on_b).as_ref() == Some(&val) {
                        results.push((row_a.clone(), row_b.clone()));
                    }
                }
            }
        }
    }
    Ok(results)
}
}

Parallel Join Optimization:

When left table exceeds PARALLEL_THRESHOLD (1000 rows), joins use Rayon for parallel probing:

#![allow(unused)]
fn main() {
if rows_a.len() >= Self::PARALLEL_THRESHOLD {
    rows_a.par_iter()
        .flat_map(|row_a| {
            // Parallel probe of hash index
        })
        .collect()
}
}

Natural Join

Natural join finds all common column names and joins on their equality:

#![allow(unused)]
fn main() {
pub fn natural_join(&self, table_a: &str, table_b: &str) -> Result<Vec<(Row, Row)>> {
    let schema_a = self.get_schema(table_a)?;
    let schema_b = self.get_schema(table_b)?;

    // Find common columns
    let cols_a: HashSet<_> = schema_a.columns.iter().map(|c| c.name.as_str()).collect();
    let cols_b: HashSet<_> = schema_b.columns.iter().map(|c| c.name.as_str()).collect();
    let common_cols: Vec<_> = cols_a.intersection(&cols_b).copied().collect();

    // No common columns = cross join
    if common_cols.is_empty() {
        return self.cross_join(table_a, table_b);
    }

    // Build composite hash key from all common columns
    // ...
}
}

Aggregate Function Internals

Parallel Aggregation

For tables exceeding PARALLEL_THRESHOLD (1000 rows), aggregates use parallel reduction:

#![allow(unused)]
fn main() {
pub fn avg(&self, table: &str, column: &str, condition: Condition) -> Result<Option<f64>> {
    let rows = self.select(table, condition)?;

    let (total, count) = if rows.len() >= Self::PARALLEL_THRESHOLD {
        // Parallel map-reduce
        rows.par_iter()
            .map(|row| extract_numeric(row, column))
            .reduce(|| (0.0, 0u64), |(s1, c1), (s2, c2)| (s1 + s2, c1 + c2))
    } else {
        // Sequential accumulation
        let mut total = 0.0;
        let mut count = 0u64;
        for row in &rows {
            // accumulate...
        }
        (total, count)
    };

    if count == 0 { Ok(None) } else { Ok(Some(total / count as f64)) }
}
}

MIN/MAX with Parallel Reduction

#![allow(unused)]
fn main() {
pub fn min(&self, table: &str, column: &str, condition: Condition) -> Result<Option<Value>> {
    let rows = self.select(table, condition)?;

    if rows.len() >= Self::PARALLEL_THRESHOLD {
        rows.par_iter()
            .filter_map(|row| row.get(column).filter(|v| !matches!(v, Value::Null)))
            .reduce_with(|a, b| {
                if a.partial_cmp_value(&b) == Some(Ordering::Less) { a } else { b }
            })
    } else {
        // Sequential scan
    }
}
}

Performance Characteristics

Where k = number of indexes on the table, n = rows in left table, m = rows in right table.

Parallel Threshold

Operations automatically switch to parallel execution when row count exceeds PARALLEL_THRESHOLD:

#![allow(unused)]
fn main() {
impl RelationalEngine {
    const PARALLEL_THRESHOLD: usize = 1000;
}
}

Parallel Operations:

• delete_rows (parallel deletion via Rayon)
• join (parallel probe phase)
• sum, avg, min, max (parallel reduction)

Configuration

RelationalConfig

The engine can be configured with RelationalConfig:

#![allow(unused)]
fn main() {
let config = RelationalConfig {
    max_tables: Some(1000),              // Maximum tables allowed
    max_indexes_per_table: Some(10),     // Maximum indexes per table
    max_btree_entries: 10_000_000,       // Maximum B-tree index entries
    default_query_timeout_ms: Some(5000),// Default query timeout
    max_query_timeout_ms: Some(300_000), // Maximum allowed timeout (5 min)
    slow_query_threshold_ms: 100,        // Slow query warning threshold
    max_query_result_rows: Some(10_000), // Maximum rows per query
    transaction_timeout_secs: 60,        // Transaction timeout
    lock_timeout_secs: 30,               // Lock acquisition timeout
};
let engine = RelationalEngine::with_config(config);
}

Internal Constants

Observability

The observability module provides query metrics, slow query detection, and index usage tracking.

Query Metrics

#![allow(unused)]
fn main() {
use relational_engine::observability::{QueryMetrics, check_slow_query};
use std::time::Duration;

let metrics = QueryMetrics::new("users", "select")
    .with_rows_scanned(10000)
    .with_rows_returned(50)
    .with_index("idx_user_id")
    .with_duration(Duration::from_millis(25));

// Log warning if query exceeds threshold
check_slow_query(&metrics, 100); // threshold in ms
}

Index Tracking

Track index usage to identify missing indexes:

#![allow(unused)]
fn main() {
use relational_engine::observability::IndexTracker;

let tracker = IndexTracker::new();

// Record when index is used
tracker.record_hit("users", "id");

// Record when index could have been used but wasn't
tracker.record_miss("users", "email");

// Get reports of columns needing indexes
let reports = tracker.report_misses();
for report in reports {
    println!(
        "Table {}, column {}: {} misses, {} hits",
        report.table, report.column, report.miss_count, report.hit_count
    );
}

// Aggregate statistics
let total_hits = tracker.total_hits();
let total_misses = tracker.total_misses();
}

Slow Query Warnings

The check_slow_query function logs a tracing::warn! when queries exceed the threshold:

#![allow(unused)]
fn main() {
use relational_engine::observability::{check_slow_query, warn_full_table_scan};

// Warn if query took > 100ms
check_slow_query(&metrics, 100);

// Warn about full table scans on large tables (> 1000 rows)
warn_full_table_scan("users", "select", 5000);
}

Streaming Cursor API

For large result sets, use streaming cursors to avoid loading all rows into memory at once. The cursor fetches rows in configurable batches.

Basic Usage

#![allow(unused)]
fn main() {
use relational_engine::{StreamingCursor, Condition};

// Create streaming cursor with default batch size (1000)
let cursor = engine.select_streaming("users", Condition::True);

// Iterate over results
for row_result in cursor {
    let row = row_result?;
    println!("User: {:?}", row);
}
}

Custom Options

#![allow(unused)]
fn main() {
// With custom batch size
let cursor = engine.select_streaming("users", Condition::True)
    .with_batch_size(100)
    .with_max_rows(5000);

// Using the builder
let cursor = engine.select_streaming_builder("users", Condition::True)
    .batch_size(100)
    .max_rows(5000)
    .build();

// Check cursor state
let mut cursor = engine.select_streaming("users", Condition::True);
while let Some(row) = cursor.next() {
    println!("Yielded so far: {}", cursor.rows_yielded());
}
println!("Exhausted: {}", cursor.is_exhausted());
}

Cursor Methods

Edge Cases and Gotchas

NULL Handling

1. NULL in conditions: Comparisons with NULL columns return false:

   #![allow(unused)]
   fn main() {
   // If email is NULL, this returns false (not true!)
   Condition::Lt("email".into(), Value::String("z".into()))
   
   }

2. **NULL in joins**: NULL values never match in join conditions:

   ```rust
   // Post with user_id = NULL will not join with any user
   engine.join("users", "posts", "_id", "user_id")

1. COUNT vs COUNT(column):
   • count() counts all rows
   • count_column() counts non-null values only

Type Mismatches

Comparisons between incompatible types return false rather than error:

#![allow(unused)]
fn main() {
// Age is Int, comparing with String returns 0 matches (not error)
engine.select("users", Condition::Lt("age".into(), Value::String("30".into())));
}

Index Maintenance

Indexes are automatically maintained on INSERT, UPDATE, and DELETE:

#![allow(unused)]
fn main() {
// Creating index AFTER data exists
engine.insert("users", values)?;  // No index update
engine.create_index("users", "age")?;  // Scans all rows

// Creating index BEFORE data exists
engine.create_index("users", "age")?;  // Empty index
engine.insert("users", values)?;  // Updates index
}

Batch Insert Atomicity

batch_insert validates ALL rows upfront before inserting any:

#![allow(unused)]
fn main() {
let rows = vec![valid_row, invalid_row];
// Fails on validation - NO rows inserted (not partial insert)
engine.batch_insert("users", rows);
}

B-Tree Index Recovery

B-tree indexes maintain both in-memory and persistent state. The in-memory BTreeMap is rebuilt lazily on first access after restart.

Best Practices

Index Selection

Columnar Materialization

Materialize columns when:

• Performing many range scans on large tables
• Query selectivity is low (scanning most rows)
• Column data fits in memory

#![allow(unused)]
fn main() {
// Good: Materialize frequently-filtered columns
engine.materialize_columns("events", &["timestamp", "user_id"])?;

// Query uses SIMD acceleration
engine.select_columnar("events",
    Condition::Gt("timestamp".into(), Value::Int(cutoff)),
    ColumnarScanOptions { prefer_columnar: true, .. }
)?;
}

Batch Operations

Use batch_insert for bulk loading:

#![allow(unused)]
fn main() {
// Bad: 1000 individual inserts
for row in rows {
    engine.insert("table", row)?;  // 1000 schema lookups
}

// Good: Single batch insert
engine.batch_insert("table", rows)?;  // 1 schema lookup, 59x faster
}

SQL Features via Query Router

When using the relational engine through query_router, additional SQL features are available:

ORDER BY and OFFSET

SELECT * FROM users ORDER BY age ASC;
SELECT * FROM users ORDER BY department DESC, name ASC;
SELECT * FROM users ORDER BY email NULLS FIRST;
SELECT * FROM users ORDER BY created_at DESC LIMIT 10 OFFSET 20;

GROUP BY and HAVING

SELECT department, COUNT(*), AVG(salary) FROM employees GROUP BY department;

SELECT product, SUM(quantity) as total
FROM orders
GROUP BY product
HAVING SUM(quantity) > 100;

Related Modules

Feature Summary

Implemented

Future Considerations

Relational Engine Transactions

Local ACID transactions for single-shard relational operations. This module provides row-level locking, undo logging for rollback, and timeout-based deadlock prevention.

Transactions in the relational engine operate within a single shard and do not coordinate with other nodes. For distributed transactions across multiple shards, see Distributed Transactions.

Architecture

flowchart TD
    subgraph TransactionManager
        TxMap[Transaction Map]
        TxCounter[TX Counter]
        DefaultTimeout[Default Timeout]
    end

    subgraph RowLockManager
        LockMap["Locks: (table, row_id) -> RowLock"]
        TxLocks["TX Locks: tx_id -> Vec<(table, row_id)>"]
        LockTimeout[Lock Timeout]
    end

    subgraph Transaction
        TxId[Transaction ID]
        Phase[TxPhase State]
        UndoLog["Undo Log: Vec<UndoEntry>"]
        AffectedTables[Affected Tables]
        StartTime[Started At]
    end

    TransactionManager --> RowLockManager
    TransactionManager --> Transaction
    Transaction --> UndoLog

Component Relationships

+-------------------+
| RelationalEngine  |
+-------------------+
         |
         v
+-------------------+     +------------------+
| TransactionManager| --> | RowLockManager   |
| - begin()         |     | - try_lock()     |
| - commit()        |     | - release()      |
| - rollback()      |     | - cleanup_expired|
+-------------------+     +------------------+
         |
         v
+-------------------+
| Transaction       |
| - tx_id           |
| - phase           |
| - undo_log        |
+-------------------+

Transaction Lifecycle

Transactions follow a 5-state machine with well-defined transitions:

flowchart LR
    Active --> Committing
    Active --> Aborting
    Committing --> Committed
    Aborting --> Aborted

    style Committed fill:#9f9
    style Aborted fill:#f99

Lifecycle Flow

#![allow(unused)]
fn main() {
// Begin transaction
let tx_id = tx_manager.begin();
// Phase: Active

// Acquire row locks for modifications
lock_manager.try_lock(tx_id, &[("users", 1), ("users", 2)])?;

// Record undo entries for rollback
tx_manager.record_undo(tx_id, UndoEntry::UpdatedRow { ... });

// Option 1: Commit
tx_manager.set_phase(tx_id, TxPhase::Committing);
// Apply changes...
tx_manager.set_phase(tx_id, TxPhase::Committed);
tx_manager.release_locks(tx_id);

// Option 2: Rollback
tx_manager.set_phase(tx_id, TxPhase::Aborting);
for entry in tx_manager.get_undo_log(tx_id).iter().rev() {
    // Apply undo entry...
}
tx_manager.set_phase(tx_id, TxPhase::Aborted);
tx_manager.release_locks(tx_id);
}

Undo Log

The undo log stores entries needed to reverse each operation on rollback. Entries are applied in reverse order during rollback.

UndoEntry Variants

Undo Log Structure

Transaction Undo Log:
+---------------------------------------------------+
| Entry 0: InsertedRow { table: "users", row_id: 5 }|
+---------------------------------------------------+
| Entry 1: UpdatedRow { table: "users", row_id: 3,  |
|          old_values: [Int(25)], ... }             |
+---------------------------------------------------+
| Entry 2: DeletedRow { table: "orders", row_id: 7, |
|          old_values: [...], index_entries: [...] }|
+---------------------------------------------------+

Rollback order: Entry 2 -> Entry 1 -> Entry 0

Index Change Tracking

Updates that modify indexed columns record IndexChange entries:

#![allow(unused)]
fn main() {
pub struct IndexChange {
    pub column: String,     // Column name
    pub old_value: Value,   // Value before update
    pub new_value: Value,   // Value after update
}
}

On rollback, index entries are reverted:

1. Remove new index entry for the new value
2. Restore old index entry for the old value

Row-Level Locking

The RowLockManager provides pessimistic row-level locking with atomic multi-row acquisition.

Lock Acquisition

flowchart TD
    Request[Lock Request] --> Check{All rows available?}
    Check -->|Yes| Acquire[Acquire all locks atomically]
    Check -->|No| Conflict[Return LockConflictInfo]
    Acquire --> Success[Success]

    style Conflict fill:#f99
    style Success fill:#9f9

Locks are acquired atomically to prevent partial lock acquisition:

#![allow(unused)]
fn main() {
// Atomic multi-row locking
let rows = vec![
    ("users".to_string(), 1),
    ("users".to_string(), 2),
    ("orders".to_string(), 5),
];

match lock_manager.try_lock(tx_id, &rows) {
    Ok(()) => {
        // All locks acquired
    }
    Err(conflict) => {
        // No locks acquired, conflict info provided
        println!("Blocked by tx {}", conflict.blocking_tx);
    }
}
}

Lock Semantics

Lock Conflict Detection

When a lock conflict occurs, LockConflictInfo provides details:

#![allow(unused)]
fn main() {
pub struct LockConflictInfo {
    pub blocking_tx: u64,   // Transaction holding the lock
    pub table: String,      // Table name
    pub row_id: u64,        // Row ID
}
}

Expired Lock Handling

Locks automatically expire after the configured timeout:

#![allow(unused)]
fn main() {
// Check if lock is expired
if lock.is_expired() {
    // Lock can be acquired by another transaction
}

// Periodic cleanup of expired locks
let cleaned = lock_manager.cleanup_expired();
}

Deadline

The Deadline struct provides monotonic time-based timeout checking:

#![allow(unused)]
fn main() {
// Create deadline with timeout
let deadline = Deadline::from_timeout_ms(Some(5000));

// Check expiration
if deadline.is_expired() {
    return Err(TimeoutError);
}

// Get remaining time
if let Some(remaining) = deadline.remaining_ms() {
    println!("{} ms remaining", remaining);
}

// Never-expiring deadline
let no_deadline = Deadline::never();
}

Benefits of monotonic time (Instant):

• Immune to system clock changes
• Consistent timeout behavior
• No backwards time jumps

Configuration

Transaction Manager

Row Lock Manager

Custom Configuration

#![allow(unused)]
fn main() {
use std::time::Duration;

// Custom transaction timeout
let tx_manager = TransactionManager::with_timeout(
    Duration::from_secs(120)
);

// Custom lock timeout
let lock_manager = RowLockManager::with_default_timeout(
    Duration::from_secs(60)
);
}

Error Handling

Lock Errors

Transaction Errors

Cleanup Operations

#![allow(unused)]
fn main() {
// Clean up expired transactions (releases their locks)
let expired_count = tx_manager.cleanup_expired();

// Clean up expired locks only
let expired_locks = lock_manager.cleanup_expired();
}

Comparison with Distributed Transactions

For cross-shard transactions, use tensor_chain’s Distributed Transactions.

Usage Examples

Basic Transaction

#![allow(unused)]
fn main() {
let engine = RelationalEngine::new();
let tx_manager = engine.tx_manager();

// Begin transaction
let tx_id = tx_manager.begin();

// Perform operations (simplified)
// engine.insert_tx(tx_id, "users", values)?;
// engine.update_tx(tx_id, "users", condition, updates)?;

// Commit
tx_manager.set_phase(tx_id, TxPhase::Committing);
// Apply pending changes...
tx_manager.set_phase(tx_id, TxPhase::Committed);
tx_manager.release_locks(tx_id);
tx_manager.remove(tx_id);
}

Rollback on Error

#![allow(unused)]
fn main() {
let tx_id = tx_manager.begin();

match perform_operations(tx_id) {
    Ok(()) => {
        tx_manager.set_phase(tx_id, TxPhase::Committed);
    }
    Err(e) => {
        tx_manager.set_phase(tx_id, TxPhase::Aborting);

        // Apply undo log in reverse
        if let Some(undo_log) = tx_manager.get_undo_log(tx_id) {
            for entry in undo_log.iter().rev() {
                apply_undo(entry);
            }
        }

        tx_manager.set_phase(tx_id, TxPhase::Aborted);
    }
}

tx_manager.release_locks(tx_id);
tx_manager.remove(tx_id);
}

Checking Lock Status

#![allow(unused)]
fn main() {
let lock_manager = tx_manager.lock_manager();

// Check if row is locked
if lock_manager.is_locked("users", 42) {
    println!("Row is locked");
}

// Get lock holder
if let Some(holder_tx) = lock_manager.lock_holder("users", 42) {
    println!("Locked by transaction {}", holder_tx);
}

// Count active locks
println!("{} active locks", lock_manager.active_lock_count());
println!("{} locks held by tx {}", lock_manager.locks_held_by(tx_id), tx_id);
}

Key Types

Source References

• relational_engine/src/transaction.rs - Transaction implementation
• relational_engine/src/lib.rs - Integration with RelationalEngine

Graph Engine

The Graph Engine provides graph operations on top of the Tensor Store. It implements a labeled property graph model with support for both directed and undirected edges, BFS traversals, and shortest path finding. The engine inherits thread safety from TensorStore and supports cross-engine unified entity connections.

Design Principles

Key Types

Core Types

PropertyValue Variants

Error Types

Architecture

graph TB
    subgraph GraphEngine
        GE[GraphEngine]
        NC[Node Counter<br/>AtomicU64]
        EC[Edge Counter<br/>AtomicU64]
    end

    subgraph Storage["Storage Model"]
        NM["node:{id}"]
        NO["node:{id}:out"]
        NI["node:{id}:in"]
        EM["edge:{id}"]
    end

    subgraph Operations
        CreateNode[create_node]
        CreateEdge[create_edge]
        Neighbors[neighbors]
        Traverse[traverse]
        FindPath[find_path]
    end

    GE --> NC
    GE --> EC
    GE --> TS[TensorStore]

    CreateNode --> NM
    CreateNode --> NO
    CreateNode --> NI
    CreateEdge --> EM
    CreateEdge --> NO
    CreateEdge --> NI

    Neighbors --> NO
    Neighbors --> NI
    Traverse --> Neighbors
    FindPath --> NO

Internal Architecture

GraphEngine Struct

#![allow(unused)]
fn main() {
pub struct GraphEngine {
    store: TensorStore,           // Underlying key-value storage
    node_counter: AtomicU64,      // Atomic counter for node IDs
    edge_counter: AtomicU64,      // Atomic counter for edge IDs
}
}

The engine uses atomic counters (SeqCst ordering) to generate unique IDs:

• Node IDs start at 1 and increment monotonically
• Edge IDs are separate from node IDs
• Both counters support concurrent ID allocation

Key Generation Functions

#![allow(unused)]
fn main() {
fn node_key(id: u64) -> String { format!("node:{}", id) }
fn edge_key(id: u64) -> String { format!("edge:{}", id) }
fn outgoing_edges_key(node_id: u64) -> String { format!("node:{}:out", node_id) }
fn incoming_edges_key(node_id: u64) -> String { format!("node:{}:in", node_id) }
}

Storage Model

Nodes and edges are stored in Tensor Store using the following key patterns:

Edge List Storage Format

Edge lists are stored as TensorData with dynamically named fields:

#![allow(unused)]
fn main() {
// Each edge ID stored as: "e{edge_id}" -> edge_id
tensor.set("e1", TensorValue::Scalar(ScalarValue::Int(1)));
tensor.set("e5", TensorValue::Scalar(ScalarValue::Int(5)));
}

This format allows O(1) edge addition but O(n) edge listing. The edge retrieval scans all keys starting with ‘e’:

#![allow(unused)]
fn main() {
fn get_edge_list(&self, key: &str) -> Result<Vec<u64>> {
    let tensor = self.store.get(key)?;
    let mut edges = Vec::new();
    for k in tensor.keys() {
        if k.starts_with('e') {
            if let Some(TensorValue::Scalar(ScalarValue::Int(id))) = tensor.get(k) {
                edges.push(*id as u64);
            }
        }
    }
    Ok(edges)
}
}

API Reference

Engine Construction

#![allow(unused)]
fn main() {
// Create new engine with internal store
let engine = GraphEngine::new();

// Create engine with shared store (for cross-engine queries)
let store = TensorStore::new();
let engine = GraphEngine::with_store(store.clone());

// Access underlying store
let store = engine.store();
}

Node Operations

#![allow(unused)]
fn main() {
// Create node with properties
let mut props = HashMap::new();
props.insert("name".to_string(), PropertyValue::String("Alice".into()));
props.insert("age".to_string(), PropertyValue::Int(30));
let id = engine.create_node("Person", props)?;

// Get node by ID
let node = engine.get_node(id)?;

// Check node existence
let exists = engine.node_exists(id);

// Delete node (cascades to connected edges)
engine.delete_node(id)?;

// Count nodes in graph
let count = engine.node_count();
}

Edge Operations

#![allow(unused)]
fn main() {
// Create directed edge
let edge_id = engine.create_edge(from, to, "KNOWS", properties, true)?;

// Create undirected edge
let edge_id = engine.create_edge(from, to, "FRIENDS", properties, false)?;

// Get edge by ID
let edge = engine.get_edge(edge_id)?;
}

Undirected Edge Implementation

When an undirected edge is created, it is added to four edge lists to enable bidirectional traversal:

#![allow(unused)]
fn main() {
if !directed {
    // Add to both nodes' outgoing AND incoming lists
    self.add_edge_to_list(Self::outgoing_edges_key(to), id)?;
    self.add_edge_to_list(Self::incoming_edges_key(from), id)?;
}
}

This enables undirected edges to be traversed from either endpoint regardless of direction filter.

Traversal Operations

#![allow(unused)]
fn main() {
// Get neighbors (all edge types, both directions)
let neighbors = engine.neighbors(node_id, None, Direction::Both)?;

// Get neighbors filtered by edge type
let friends = engine.neighbors(node_id, Some("FRIENDS"), Direction::Both)?;

// BFS traversal with depth limit
let nodes = engine.traverse(start_id, Direction::Outgoing, max_depth, None)?;

// Traversal filtered by edge type
let deps = engine.traverse(start_id, Direction::Outgoing, 10, Some("DEPENDS_ON"))?;

// Find shortest path (BFS)
let path = engine.find_path(from_id, to_id)?;
}

Direction Enum

BFS Traversal Algorithm

The traverse method implements breadth-first search with depth limiting and cycle detection:

flowchart TD
    Start[Start: traverse] --> Init[Initialize visited set<br/>Initialize result vec<br/>Initialize queue with start, depth=0]
    Init --> Check{Queue empty?}
    Check -- No --> Pop[Pop current_id, depth]
    Pop --> GetNode[Get node, add to result]
    GetNode --> DepthCheck{depth >= max_depth?}
    DepthCheck -- Yes --> Check
    DepthCheck -- No --> GetNeighbors[Get neighbor IDs]
    GetNeighbors --> ForEach[For each neighbor]
    ForEach --> Visited{Already visited?}
    Visited -- Yes --> ForEach
    Visited -- No --> Add[Add to visited<br/>Push to queue with depth+1]
    Add --> ForEach
    Check -- Yes --> Return[Return result]

Implementation Details

#![allow(unused)]
fn main() {
pub fn traverse(
    &self,
    start: u64,
    direction: Direction,
    max_depth: usize,
    edge_type: Option<&str>,
) -> Result<Vec<Node>> {
    if !self.node_exists(start) {
        return Err(GraphError::NodeNotFound(start));
    }

    let mut visited = HashSet::new();
    let mut result = Vec::new();
    let mut queue = VecDeque::new();

    queue.push_back((start, 0usize));
    visited.insert(start);

    while let Some((current_id, depth)) = queue.pop_front() {
        if let Ok(node) = self.get_node(current_id) {
            result.push(node);
        }

        if depth >= max_depth {
            continue;
        }

        let neighbors = self.get_neighbor_ids(current_id, edge_type, direction)?;
        for neighbor_id in neighbors {
            if !visited.contains(&neighbor_id) {
                visited.insert(neighbor_id);
                queue.push_back((neighbor_id, depth + 1));
            }
        }
    }

    Ok(result)
}
}

Key Properties

• Cycle-Safe: The visited HashSet prevents revisiting nodes
• Depth-Limited: The max_depth parameter bounds traversal depth
• Level-Order: BFS naturally visits nodes in level order
• Start Node Included: The starting node is always in the result at depth 0

Shortest Path Algorithm

The find_path method uses BFS to find the shortest (minimum hop) path between two nodes:

flowchart TD
    Start[Start: find_path] --> Validate[Validate from and to exist]
    Validate --> SameNode{from == to?}
    SameNode -- Yes --> ReturnSingle[Return path with single node]
    SameNode -- No --> InitBFS[Initialize BFS:<br/>visited set<br/>queue with from<br/>parent map]
    InitBFS --> BFSLoop{Queue empty?}
    BFSLoop -- Yes --> NotFound[Return PathNotFound]
    BFSLoop -- No --> Dequeue[Dequeue current node]
    Dequeue --> GetEdges[Get outgoing edges]
    GetEdges --> ForEdge[For each edge]
    ForEdge --> GetNeighbor[Determine neighbor<br/>considering direction]
    GetNeighbor --> VisitedCheck{Visited?}
    VisitedCheck -- Yes --> ForEdge
    VisitedCheck -- No --> MarkVisited[Mark visited<br/>Record parent + edge]
    MarkVisited --> FoundTarget{neighbor == to?}
    FoundTarget -- Yes --> Reconstruct[Reconstruct path]
    FoundTarget -- No --> Enqueue[Enqueue neighbor]
    Enqueue --> ForEdge
    ForEdge --> BFSLoop
    Reconstruct --> Return[Return Path]

Implementation Details

#![allow(unused)]
fn main() {
pub fn find_path(&self, from: u64, to: u64) -> Result<Path> {
    // Validate endpoints exist
    if !self.node_exists(from) {
        return Err(GraphError::NodeNotFound(from));
    }
    if !self.node_exists(to) {
        return Err(GraphError::NodeNotFound(to));
    }

    // Handle trivial case
    if from == to {
        return Ok(Path {
            nodes: vec![from],
            edges: vec![],
        });
    }

    // BFS for shortest path
    let mut visited = HashSet::new();
    let mut queue = VecDeque::new();
    let mut parent: HashMap<u64, (u64, u64)> = HashMap::new(); // node -> (parent_node, edge_id)

    queue.push_back(from);
    visited.insert(from);

    while let Some(current) = queue.pop_front() {
        let out_edges = self.get_edge_list(&Self::outgoing_edges_key(current))?;

        for edge_id in out_edges {
            if let Ok(edge) = self.get_edge(edge_id) {
                let neighbor = if edge.from == current {
                    edge.to
                } else if !edge.directed && edge.to == current {
                    edge.from
                } else {
                    continue;
                };

                if !visited.contains(&neighbor) {
                    visited.insert(neighbor);
                    parent.insert(neighbor, (current, edge_id));

                    if neighbor == to {
                        return Ok(self.reconstruct_path(from, to, &parent));
                    }

                    queue.push_back(neighbor);
                }
            }
        }
    }

    Err(GraphError::PathNotFound)
}
}

Path Reconstruction

The path is reconstructed by following parent pointers backwards from the target to the source:

#![allow(unused)]
fn main() {
fn reconstruct_path(&self, from: u64, to: u64, parent: &HashMap<u64, (u64, u64)>) -> Path {
    let mut nodes = Vec::new();
    let mut edges = Vec::new();
    let mut current = to;

    // Walk backwards from target to source
    while current != from {
        nodes.push(current);
        if let Some((p, edge_id)) = parent.get(&current) {
            edges.push(*edge_id);
            current = *p;
        } else {
            break;
        }
    }
    nodes.push(from);

    // Reverse to get source-to-target order
    nodes.reverse();
    edges.reverse();

    Path { nodes, edges }
}
}

Parallel Deletion Optimization

High-degree nodes (>100 edges) use rayon’s parallel iterator for edge deletion:

#![allow(unused)]
fn main() {
const PARALLEL_THRESHOLD: usize = 100;

pub fn delete_node(&self, id: u64) -> Result<()> {
    if !self.node_exists(id) {
        return Err(GraphError::NodeNotFound(id));
    }

    // Collect all connected edges
    let out_edges = self.get_edge_list(&Self::outgoing_edges_key(id))?;
    let in_edges = self.get_edge_list(&Self::incoming_edges_key(id))?;
    let all_edges: Vec<u64> = out_edges.into_iter().chain(in_edges).collect();

    // Parallel deletion for high-degree nodes
    if all_edges.len() >= Self::PARALLEL_THRESHOLD {
        all_edges.par_iter().for_each(|edge_id| {
            let _ = self.store.delete(&Self::edge_key(*edge_id));
        });
    } else {
        for edge_id in all_edges {
            let _ = self.store.delete(&Self::edge_key(edge_id));
        }
    }

    // Delete node and edge lists
    self.store.delete(&Self::node_key(id))?;
    self.store.delete(&Self::outgoing_edges_key(id))?;
    self.store.delete(&Self::incoming_edges_key(id))?;

    Ok(())
}
}

Performance Characteristics

Unified Entity API

The Unified Entity API connects any shared entities (not just graph nodes) for cross-engine queries. Entity edges use the _out and _in reserved fields in TensorData, enabling the same entity key to have relational fields, graph connections, and a vector embedding.

graph LR
    subgraph Entity["Entity (TensorData)"]
        Fields[User Fields<br/>name, age, etc.]
        Out["_out<br/>[edge keys]"]
        In["_in<br/>[edge keys]"]
        Emb["_embedding<br/>[vector]"]
    end

    subgraph Engines
        RE[Relational Engine]
        GE[Graph Engine]
        VE[Vector Engine]
    end

    Fields --> RE
    Out --> GE
    In --> GE
    Emb --> VE

Reserved Fields

Entity Edge Key Format

Entity edges use a different key format from node-based edges:

edge:{edge_type}:{edge_id}

For example: edge:follows:42

API Reference

#![allow(unused)]
fn main() {
// Create engine with shared store
let store = TensorStore::new();
let engine = GraphEngine::with_store(store.clone());

// Add directed edge between entities
let edge_key = engine.add_entity_edge("user:1", "user:2", "follows")?;

// Add undirected edge between entities
let edge_key = engine.add_entity_edge_undirected("user:1", "user:2", "friend")?;

// Get neighbors
let neighbors = engine.get_entity_neighbors("user:1")?;
let out_neighbors = engine.get_entity_neighbors_out("user:1")?;
let in_neighbors = engine.get_entity_neighbors_in("user:1")?;

// Get edge lists
let outgoing = engine.get_entity_outgoing("user:1")?;
let incoming = engine.get_entity_incoming("user:1")?;

// Get edge details
let (from, to, edge_type, directed) = engine.get_entity_edge(&edge_key)?;

// Check if entity has edges
let has_edges = engine.entity_has_edges("user:1");

// Delete edge
engine.delete_entity_edge(&edge_key)?;

// Scan for entities with edges
let entities = engine.scan_entities_with_edges();
}

Undirected Entity Edges

For undirected entity edges, both entities receive the edge in both _out and _in:

#![allow(unused)]
fn main() {
pub fn add_entity_edge_undirected(
    &self,
    key1: &str,
    key2: &str,
    edge_type: &str,
) -> Result<String> {
    // ... create edge data ...

    // Both entities get the edge in both directions
    let mut entity1 = self.get_or_create_entity(key1);
    entity1.add_outgoing_edge(edge_key.clone());
    entity1.add_incoming_edge(edge_key.clone());

    let mut entity2 = self.get_or_create_entity(key2);
    entity2.add_outgoing_edge(edge_key.clone());
    entity2.add_incoming_edge(edge_key.clone());

    Ok(edge_key)
}
}

Cross-Engine Integration

Query Router Integration

The Query Router provides unified queries combining graph traversal with vector similarity:

#![allow(unused)]
fn main() {
// Find entities similar to query that are connected to a specific entity
let items = router.find_similar_connected("query:entity", "connected_to:entity", top_k)?;

// Find graph neighbors sorted by embedding similarity
let items = router.find_neighbors_by_similarity("entity:key", &query_vector, top_k)?;
}

Tensor Vault Integration

Tensor Vault uses GraphEngine for access control relationships:

#![allow(unused)]
fn main() {
pub struct Vault {
    store: TensorStore,
    pub graph: Arc<GraphEngine>,  // Shared graph for access edges
    // ...
}
}

Access control edges connect principals to secrets with permission metadata.

Tensor Chain Integration

Tensor Chain uses GraphEngine for block linking:

#![allow(unused)]
fn main() {
pub struct Chain {
    graph: Arc<GraphEngine>,  // Stores blocks as nodes, links as edges
    // ...
}
}

Blocks are stored with chain:block:{height} keys and linked via graph edges with type chain_next.

Performance Characteristics

Memory Characteristics

Edge Cases and Gotchas

Self-Loop Edges

Self-loops (edges from a node to itself) are valid but filtered from neighbor results:

#![allow(unused)]
fn main() {
#[test]
fn self_loop_edge() {
    let engine = GraphEngine::new();
    let n1 = engine.create_node("A", HashMap::new()).unwrap();
    engine.create_edge(n1, n1, "SELF", HashMap::new(), true).unwrap();

    // Self-loop doesn't appear in neighbors
    let neighbors = engine.neighbors(n1, None, Direction::Both).unwrap();
    assert_eq!(neighbors.len(), 0);
}
}

Same-Node Path

Finding a path from a node to itself returns a single-node path:

#![allow(unused)]
fn main() {
let path = engine.find_path(n1, n1)?;
assert_eq!(path.nodes, vec![n1]);
assert!(path.edges.is_empty());
}

Deleted Edge Orphans

When deleting a node, connected edges are deleted from storage but may remain in other nodes’ edge lists. This is a known limitation - the edge retrieval gracefully handles missing edges.

Bytes Property Conversion

ScalarValue::Bytes converts to PropertyValue::Null since PropertyValue doesn’t support binary data:

#![allow(unused)]
fn main() {
let bytes = ScalarValue::Bytes(vec![1, 2, 3]);
assert_eq!(PropertyValue::from_scalar(&bytes), PropertyValue::Null);
}

Node Count Calculation

The node_count method uses a formula based on scan counts to account for edge lists:

#![allow(unused)]
fn main() {
pub fn node_count(&self) -> usize {
    // Each node has 3 keys: node:{id}, node:{id}:out, node:{id}:in
    self.store.scan_count("node:") - self.store.scan_count("node:") / 3 * 2
}
}

Best Practices

Use Shared Store for Cross-Engine Queries

#![allow(unused)]
fn main() {
// Create shared store first
let store = TensorStore::new();

// Create engines with shared store
let graph = GraphEngine::with_store(store.clone());
let vector = VectorEngine::with_store(store.clone());

// Now entities can have both graph edges and embeddings
}

Prefer Entity API for Cross-Engine Data

Use the Unified Entity API when entities need to combine relational, graph, and vector data:

#![allow(unused)]
fn main() {
// Good: Entity API preserves all fields
engine.add_entity_edge("user:1", "user:2", "follows")?;

// Less flexible: Node API creates graph-only entities
engine.create_node("User", props)?;
}

Batch Edge Creation

When creating many edges, avoid creating them one at a time if possible. Consider the overhead of multiple store operations.

Choose Direction Wisely

• Use Direction::Outgoing for forward-only traversals (dependency graphs)
• Use Direction::Both for symmetric relationships (social graphs)
• Use Direction::Incoming for reverse lookups (finding predecessors)

Set Appropriate Traversal Depth

BFS traversal can be expensive on dense graphs. Set max_depth based on expected graph diameter:

#![allow(unused)]
fn main() {
// For typical social networks, 3-6 hops is usually sufficient
let reachable = engine.traverse(start, Direction::Both, 4, None)?;
}

Usage Examples

Social Network

#![allow(unused)]
fn main() {
let engine = GraphEngine::new();

// Create users
let alice = engine.create_node("User", user_props("Alice"))?;
let bob = engine.create_node("User", user_props("Bob"))?;
let charlie = engine.create_node("User", user_props("Charlie"))?;

// Create friendships (undirected)
engine.create_edge(alice, bob, "FRIENDS", HashMap::new(), false)?;
engine.create_edge(bob, charlie, "FRIENDS", HashMap::new(), false)?;

// Find path from Alice to Charlie
let path = engine.find_path(alice, charlie)?;
// path.nodes = [alice, bob, charlie]

// Get Alice's friends
let friends = engine.neighbors(alice, Some("FRIENDS"), Direction::Both)?;
}

Dependency Graph

#![allow(unused)]
fn main() {
let engine = GraphEngine::new();

// Create packages
let app = engine.create_node("Package", package_props("app"))?;
let lib_a = engine.create_node("Package", package_props("lib-a"))?;
let lib_b = engine.create_node("Package", package_props("lib-b"))?;

// Create dependencies (directed)
engine.create_edge(app, lib_a, "DEPENDS_ON", HashMap::new(), true)?;
engine.create_edge(app, lib_b, "DEPENDS_ON", HashMap::new(), true)?;
engine.create_edge(lib_a, lib_b, "DEPENDS_ON", HashMap::new(), true)?;

// Find all dependencies of app
let deps = engine.traverse(app, Direction::Outgoing, 10, Some("DEPENDS_ON"))?;
}

Cross-Engine Unified Entities

#![allow(unused)]
fn main() {
// Shared store for multiple engines
let store = TensorStore::new();
let graph = GraphEngine::with_store(store.clone());

// Add graph edges between entities
graph.add_entity_edge("user:1", "post:1", "created")?;
graph.add_entity_edge("user:2", "post:1", "liked")?;

// Query relationships
let creators = graph.get_entity_neighbors_in("post:1")?;
}

High-Degree Node Operations

#![allow(unused)]
fn main() {
let engine = GraphEngine::new();

// Create a hub with many connections (will use parallel deletion)
let hub = engine.create_node("Hub", HashMap::new())?;
for i in 0..150 {
    let leaf = engine.create_node("Leaf", HashMap::new())?;
    engine.create_edge(hub, leaf, "CONNECTS", HashMap::new(), true)?;
}

// Deletion will use parallel processing (150 > 100 threshold)
engine.delete_node(hub)?;
}

Configuration

The Graph Engine has minimal configuration as it inherits behavior from TensorStore:

Dependencies

Related Modules

Vector Engine

Module 4 of Neumann. Provides embeddings storage and similarity search with SIMD-accelerated distance computations.

The Vector Engine builds on tensor_store to provide k-NN search capabilities. It supports both brute-force O(n) search and HNSW O(log n) approximate search, with automatic sparse vector optimization for memory efficiency.

Design Principles

Architecture

graph TB
    subgraph VectorEngine
        VE[VectorEngine]
        SR[SearchResult]
        DM[DistanceMetric]
        VE --> |uses| SR
        VE --> |uses| DM
    end

    subgraph TensorStore
        TS[TensorStore]
        HNSW[HNSWIndex]
        SV[SparseVector]
        SIMD[SIMD Functions]
        ES[EmbeddingStorage]
    end

    VE --> |stores to| TS
    VE --> |builds| HNSW
    VE --> |uses| SV
    VE --> |uses| SIMD

    subgraph Storage
        EMB["emb:{key}"]
        ENT["entity:{key}._embedding"]
    end

    TS --> EMB
    TS --> ENT

Key Types

VectorError Variants

Configuration

VectorEngineConfig

Configuration for the Vector Engine with memory bounds and performance tuning.

Configuration Presets

Builder Methods

All builder methods are const fn for compile-time configuration:

#![allow(unused)]
fn main() {
use std::time::Duration;

let config = VectorEngineConfig::default()
    .with_default_dimension(768)
    .with_sparse_threshold(0.7)
    .with_parallel_threshold(1000)
    .with_default_metric(DistanceMetric::Cosine)
    .with_max_dimension(4096)
    .with_max_keys_per_scan(50_000)
    .with_batch_parallel_threshold(200)
    .with_search_timeout(Duration::from_secs(5));

let engine = VectorEngine::with_config(config)?;
}

Memory Bounds

For production deployments, configure memory bounds to prevent resource exhaustion:

#![allow(unused)]
fn main() {
// Reject embeddings larger than 4096 dimensions
let config = VectorEngineConfig::default()
    .with_max_dimension(4096)
    .with_max_keys_per_scan(10_000);

let engine = VectorEngine::with_config(config)?;

// This will fail with DimensionMismatch
engine.store_embedding("too_big", vec![0.0; 5000])?; // Error!
}

Search Timeout

Configure a timeout for search operations to prevent runaway queries:

#![allow(unused)]
fn main() {
use std::time::Duration;
use vector_engine::{VectorEngine, VectorEngineConfig, VectorError};

let config = VectorEngineConfig::default()
    .with_search_timeout(Duration::from_secs(5));

let engine = VectorEngine::with_config(config)?;

match engine.search_similar(&query, 10) {
    Ok(results) => { /* process results */ },
    Err(VectorError::SearchTimeout { operation, timeout_ms }) => {
        eprintln!("Search '{}' timed out after {}ms", operation, timeout_ms);
    },
    Err(e) => { /* handle other errors */ },
}
}

The timeout applies to all search methods. When a timeout occurs, no partial results are returned to prevent misleading results that may miss better matches.

Distance Metrics

All metrics return higher scores for better matches. Euclidean distance is transformed to similarity score.

Extended Distance Metrics (HNSW)

The ExtendedDistanceMetric enum provides additional metrics for HNSW-based search via search_with_hnsw_and_metric():

#![allow(unused)]
fn main() {
use vector_engine::ExtendedDistanceMetric;

let (index, keys) = engine.build_hnsw_index_default()?;

// Search with Jaccard similarity for sparse vectors
let results = engine.search_with_hnsw_and_metric(
    &index,
    &keys,
    &query,
    10,
    ExtendedDistanceMetric::Jaccard,
)?;
}

Distance Metric Implementation Details

flowchart TD
    Query[Query Vector] --> MetricCheck{Which Metric?}

    MetricCheck -->|Cosine| CosMag[Pre-compute query magnitude]
    CosMag --> CosDot[SIMD dot product]
    CosDot --> CosDiv[Divide by magnitudes]
    CosDiv --> CosScore[Score: dot / mag_a * mag_b]

    MetricCheck -->|Euclidean| EucDiff[Compute differences]
    EucDiff --> EucSum[Sum of squares]
    EucSum --> EucSqrt[Square root]
    EucSqrt --> EucScore[Score: 1 / 1 + distance]

    MetricCheck -->|DotProduct| DotSIMD[SIMD dot product]
    DotSIMD --> DotScore[Score: raw dot product]

Cosine Similarity Edge Cases

#![allow(unused)]
fn main() {
// Zero-magnitude vectors return 0.0 similarity
let zero = vec![0.0, 0.0, 0.0];
let normal = vec![1.0, 2.0, 3.0];
VectorEngine::compute_similarity(&zero, &normal)?; // Returns 0.0

// Identical vectors return 1.0
VectorEngine::compute_similarity(&normal, &normal)?; // Returns 1.0

// Opposite vectors return -1.0
let opposite = vec![-1.0, -2.0, -3.0];
VectorEngine::compute_similarity(&normal, &opposite)?; // Returns -1.0

// Orthogonal vectors return 0.0
let a = vec![1.0, 0.0];
let b = vec![0.0, 1.0];
VectorEngine::compute_similarity(&a, &b)?; // Returns 0.0
}

Euclidean Distance Transformation

The engine transforms Euclidean distance to similarity score using 1 / (1 + distance):

SIMD Implementation

The Vector Engine uses 8-wide SIMD operations via the wide crate for accelerated distance computations.

SIMD Dot Product Algorithm

#![allow(unused)]
fn main() {
// Simplified view of the SIMD implementation
pub fn dot_product(a: &[f32], b: &[f32]) -> f32 {
    let chunks = a.len() / 8;        // Process 8 floats at a time
    let remainder = a.len() % 8;

    let mut sum = f32x8::ZERO;

    // Process 8 elements at a time with SIMD
    for i in 0..chunks {
        let offset = i * 8;
        let va = f32x8::from(&a[offset..offset + 8]);
        let vb = f32x8::from(&b[offset..offset + 8]);
        sum += va * vb;  // Parallel multiply-add
    }

    // Sum SIMD lanes + handle remainder scalar
    let arr: [f32; 8] = sum.into();
    let mut result: f32 = arr.iter().sum();

    // Handle remainder with scalar operations
    let start = chunks * 8;
    for i in 0..remainder {
        result += a[start + i] * b[start + i];
    }

    result
}
}

SIMD Performance Characteristics

SIMD operations are cache-friendly due to sequential memory access patterns.

API Reference

Basic Operations

#![allow(unused)]
fn main() {
let engine = VectorEngine::new();

// Store an embedding
engine.store_embedding("doc1", vec![0.1, 0.2, 0.3])?;

// Get an embedding
let vector = engine.get_embedding("doc1")?;

// Delete an embedding
engine.delete_embedding("doc1")?;

// Check existence
engine.exists("doc1");  // -> bool

// Count embeddings
engine.count();  // -> usize

// List all keys
let keys = engine.list_keys();

// Clear all embeddings
engine.clear()?;

// Get dimension (from first embedding)
engine.dimension();  // -> Option<usize>
}

Similarity Search

#![allow(unused)]
fn main() {
// Find top-k most similar (cosine by default)
let query = vec![0.1, 0.2, 0.3];
let results = engine.search_similar(&query, 5)?;

for result in results {
    println!("Key: {}, Score: {}", result.key, result.score);
}

// Search with specific metric
let results = engine.search_similar_with_metric(
    &query,
    5,
    DistanceMetric::Euclidean
)?;

// Direct similarity computation
let similarity = VectorEngine::compute_similarity(&vec_a, &vec_b)?;
}

Filtered Search

Search with metadata filters to narrow results without post-processing:

#![allow(unused)]
fn main() {
use vector_engine::{FilterCondition, FilterValue, FilteredSearchConfig, FilterStrategy};

// Build a filter condition
let filter = FilterCondition::Eq("category".to_string(), FilterValue::String("science".to_string()))
    .and(FilterCondition::Gt("year".to_string(), FilterValue::Int(2020)));

// Search with filter (auto strategy)
let results = engine.search_similar_filtered(&query, 10, &filter, None)?;

// Search with explicit pre-filter strategy (best for selective filters)
let config = FilteredSearchConfig::pre_filter();
let results = engine.search_similar_filtered(&query, 10, &filter, Some(config))?;

// Search with post-filter and custom oversample
let config = FilteredSearchConfig::post_filter().with_oversample(5);
let results = engine.search_similar_filtered(&query, 10, &filter, Some(config))?;
}

Filter Conditions

Filter Strategies

flowchart TD
    Query[Query + Filter] --> Strategy{Which Strategy?}

    Strategy -->|Auto| Estimate[Estimate Selectivity]
    Estimate -->|< 10%| Pre[Pre-Filter]
    Estimate -->|>= 10%| Post[Post-Filter]

    Strategy -->|PreFilter| Pre
    Strategy -->|PostFilter| Post

    Pre --> Filter1[Filter all keys]
    Filter1 --> Search1[Search filtered subset]
    Search1 --> Result[Top-K Results]

    Post --> Search2[Search with oversample]
    Search2 --> Filter2[Filter candidates]
    Filter2 --> Result

Filter Helper Methods

Utilities for working with filters:

#![allow(unused)]
fn main() {
// Estimate how selective a filter is (0.0 = matches nothing, 1.0 = matches all)
let selectivity = engine.estimate_filter_selectivity(&filter);

// Count how many embeddings match a filter
let matching = engine.count_matching(&filter);

// Get keys of all matching embeddings
let keys = engine.list_keys_matching(&filter);
}

Metadata Storage

Store and retrieve metadata alongside embeddings:

#![allow(unused)]
fn main() {
use tensor_store::TensorValue;
use std::collections::HashMap;

// Store embedding with metadata
let mut metadata = HashMap::new();
metadata.insert("category".to_string(), TensorValue::from("science"));
metadata.insert("year".to_string(), TensorValue::from(2024i64));
metadata.insert("score".to_string(), TensorValue::from(0.95f64));

engine.store_embedding_with_metadata("doc1", vec![0.1, 0.2, 0.3], metadata)?;

// Get all metadata
let meta = engine.get_metadata("doc1")?;

// Get specific field
let category = engine.get_metadata_field("doc1", "category")?;

// Update metadata (merges with existing)
let mut updates = HashMap::new();
updates.insert("score".to_string(), TensorValue::from(0.98f64));
engine.update_metadata("doc1", updates)?;

// Check if metadata field exists
if engine.has_metadata_field("doc1", "category") {
    // Remove specific metadata field
    engine.remove_metadata_field("doc1", "category")?;
}
}

Batch Operations

For bulk insert and delete operations with parallel processing:

#![allow(unused)]
fn main() {
use vector_engine::EmbeddingInput;

// Batch store - validates all inputs first, then stores in parallel
let inputs = vec![
    EmbeddingInput::new("doc1", vec![0.1, 0.2, 0.3]),
    EmbeddingInput::new("doc2", vec![0.2, 0.3, 0.4]),
    EmbeddingInput::new("doc3", vec![0.3, 0.4, 0.5]),
];

let result = engine.batch_store_embeddings(inputs)?;
println!("Stored {} embeddings", result.count);  // -> 3

// Batch delete - returns count of successfully deleted
let keys = vec!["doc1".to_string(), "doc2".to_string()];
let deleted = engine.batch_delete_embeddings(keys)?;
println!("Deleted {} embeddings", deleted);  // -> 2
}

Batches larger than batch_parallel_threshold (default: 100) use parallel processing via rayon.

Pagination

For memory-efficient iteration over large datasets:

#![allow(unused)]
fn main() {
use vector_engine::Pagination;

// List keys with pagination
let page = Pagination::new(0, 100);  // skip=0, limit=100
let result = engine.list_keys_paginated(page);
println!("Items: {}, Has more: {}", result.items.len(), result.has_more);

// Get total count with pagination
let page = Pagination::new(0, 100).with_total();
let result = engine.list_keys_paginated(page);
println!("Total: {:?}", result.total_count);  // Some(total)

// Paginated similarity search
let page = Pagination::new(10, 5);  // skip first 10, return 5
let results = engine.search_similar_paginated(&query, 100, page)?;

// Paginated entity search
let results = engine.search_entities_paginated(&query, 100, page)?;
}

Use list_keys_bounded() for production to enforce max_keys_per_scan limits.

Search Flow Diagram

sequenceDiagram
    participant Client
    participant VE as VectorEngine
    participant TS as TensorStore
    participant SIMD

    Client->>VE: search_similar(query, k)
    VE->>VE: Validate query (non-empty, k > 0)
    VE->>SIMD: Pre-compute query magnitude
    VE->>TS: scan("emb:")
    TS-->>VE: List of embedding keys

    alt Dataset < 5000 vectors
        VE->>VE: Sequential search
    else Dataset >= 5000 vectors
        VE->>VE: Parallel search (rayon)
    end

    loop For each embedding
        VE->>TS: get(key)
        TS-->>VE: TensorData
        VE->>VE: Extract vector (dense or sparse)
        VE->>SIMD: cosine_similarity(query, stored)
        VE->>VE: Collect SearchResult
    end

    VE->>VE: Sort by score descending
    VE->>VE: Truncate to top k
    VE-->>Client: Vec<SearchResult>

HNSW Index

For large datasets, build an HNSW index for O(log n) search:

#![allow(unused)]
fn main() {
// Build index with default config
let (index, key_mapping) = engine.build_hnsw_index_default()?;

// Search using the index
let results = engine.search_with_hnsw(&index, &key_mapping, &query, 10)?;

// Build with custom config
let config = HNSWConfig::high_recall();
let (index, key_mapping) = engine.build_hnsw_index(config)?;

// Direct HNSW operations
let index = HNSWIndex::new();
index.insert(vec![1.0, 2.0, 3.0]);
let results = index.search(&query, 10);

// Search with custom ef (recall/speed tradeoff)
let results = index.search_with_ef(&query, 10, 200);
}

HNSW Search Flow

flowchart TD
    Query[Query Vector] --> Entry[Entry Point at Max Layer]

    Entry --> Greedy1[Greedy Search Layer L]
    Greedy1 --> |Find closest| Greedy2[Greedy Search Layer L-1]
    Greedy2 --> |...|GreedyN[Greedy Search until Layer 1]

    GreedyN --> Layer0[Full ef-Search at Layer 0]

    Layer0 --> Candidates[Candidate Pool]
    Candidates --> |BinaryHeap min-heap| Visit[Visit Neighbors]
    Visit --> Distance[Compute Distances]
    Distance --> |Update| Results[Result Pool]
    Results --> |BinaryHeap max-heap| Prune[Keep top ef]

    Prune --> |More candidates?| Visit
    Prune --> |Done| TopK[Return Top K]

Unified Entity Mode

Attach embeddings directly to entities for cross-engine queries:

#![allow(unused)]
fn main() {
let store = TensorStore::new();
let engine = VectorEngine::with_store(store.clone());

// Set embedding on an entity
engine.set_entity_embedding("user:1", vec![0.1, 0.2, 0.3])?;

// Get embedding from an entity
let embedding = engine.get_entity_embedding("user:1")?;

// Check if entity has embedding
engine.entity_has_embedding("user:1");  // -> bool

// Remove embedding (preserves other entity data)
engine.remove_entity_embedding("user:1")?;

// Search entities with embeddings
let results = engine.search_entities(&query, 5)?;

// Scan all entities with embeddings
let entity_keys = engine.scan_entities_with_embeddings();

// Count entities with embeddings
let count = engine.count_entities_with_embeddings();
}

Unified entity embeddings are stored in the _embedding field of the entity’s TensorData.

Collections

Collections provide isolated namespaces for organizing embeddings by type or purpose. Each collection can have its own dimension constraints and distance metric configuration.

Creating and Managing Collections

#![allow(unused)]
fn main() {
use vector_engine::{VectorEngine, VectorCollectionConfig, DistanceMetric};

let engine = VectorEngine::new();

// Create collection with custom config
let config = VectorCollectionConfig::default()
    .with_dimension(768)
    .with_metric(DistanceMetric::Cosine)
    .with_auto_index(5000);  // Auto-build HNSW at 5000 vectors

engine.create_collection("documents", config)?;

// List collections
let collections = engine.list_collections();

// Check if collection exists
engine.collection_exists("documents");  // -> true

// Get collection config
let config = engine.get_collection_config("documents");

// Delete collection (removes all vectors in it)
engine.delete_collection("documents")?;
}

Storing in Collections

#![allow(unused)]
fn main() {
use std::collections::HashMap;
use tensor_store::TensorValue;

// Store vector in collection
engine.store_in_collection("documents", "doc1", vec![0.1, 0.2, 0.3])?;

// Store with metadata
let mut metadata = HashMap::new();
metadata.insert("title".to_string(), TensorValue::from("Introduction to Rust"));
metadata.insert("author".to_string(), TensorValue::from("Alice"));

engine.store_in_collection_with_metadata(
    "documents",
    "doc1",
    vec![0.1, 0.2, 0.3],
    metadata
)?;
}

Searching in Collections

#![allow(unused)]
fn main() {
use vector_engine::{FilterCondition, FilterValue};

// Basic search in collection
let results = engine.search_in_collection("documents", &query, 10)?;

// Filtered search in collection
let filter = FilterCondition::Eq("author".to_string(), FilterValue::String("Alice".to_string()));
let results = engine.search_filtered_in_collection(
    "documents",
    &query,
    10,
    &filter,
    None
)?;
}

Collection Key Isolation

Collections use prefixed storage keys to ensure isolation:

VectorCollectionConfig

Index Persistence

Save and restore vector indices for fast startup:

#![allow(unused)]
fn main() {
use std::path::Path;

// Save all collections to directory (one JSON file per collection)
let saved = engine.save_all_indices(Path::new("./vector_index"))?;

// Load all indices from directory
let loaded = engine.load_all_indices(Path::new("./vector_index"))?;

// Save single collection to JSON
engine.save_index("documents", Path::new("./documents.json"))?;

// Save single collection to compact binary format
engine.save_index_binary("documents", Path::new("./documents.bin"))?;

// Load single collection from JSON (returns collection name)
let collection = engine.load_index(Path::new("./documents.json"))?;

// Load single collection from binary
let collection = engine.load_index_binary(Path::new("./documents.bin"))?;

// Get a snapshot for manual serialization
let index: PersistentVectorIndex = engine.snapshot_collection("documents");
}

PersistentVectorIndex Format

Storage Model

Automatic Sparse Storage

Vectors with >50% zeros are automatically stored as sparse vectors:

#![allow(unused)]
fn main() {
// Detection threshold: nnz * 2 <= len (i.e., sparsity >= 50%)
fn should_use_sparse(vector: &[f32]) -> bool {
    let nnz = vector.iter().filter(|&&v| v.abs() > 1e-6).count();
    nnz * 2 <= vector.len()
}

// 97% sparse vector (3 non-zeros in 100 elements)
let mut sparse = vec![0.0f32; 100];
sparse[0] = 1.0;
sparse[50] = 2.0;
sparse[99] = 3.0;

// Stored efficiently as SparseVector
engine.store_embedding("sparse_doc", sparse)?;

// Retrieved as dense for computation
let dense = engine.get_embedding("sparse_doc")?;
}

Storage Format Comparison

Example: 1000-dim vector with 100 non-zeros:

• Dense: 4000 bytes
• Sparse: 800 bytes (5x compression)

Sparse Vector Operations

Memory Layout

SparseVector {
    dimension: usize,        // Total vector dimension
    positions: Vec<u32>,     // Sorted indices of non-zeros
    values: Vec<f32>,        // Corresponding values
}

Sparse Dot Product Algorithm

#![allow(unused)]
fn main() {
// O(min(nnz_a, nnz_b)) - only overlapping positions contribute
pub fn dot(&self, other: &SparseVector) -> f32 {
    let mut result = 0.0;
    let mut i = 0;
    let mut j = 0;

    // Merge-sort style traversal
    while i < self.positions.len() && j < other.positions.len() {
        match self.positions[i].cmp(&other.positions[j]) {
            Equal => {
                result += self.values[i] * other.values[j];
                i += 1; j += 1;
            },
            Less => i += 1,
            Greater => j += 1,
        }
    }
    result
}
}

Sparse-Dense Dot Product

#![allow(unused)]
fn main() {
// O(nnz) - only iterate over sparse non-zeros
pub fn dot_dense(&self, dense: &[f32]) -> f32 {
    self.positions.iter()
        .zip(&self.values)
        .map(|(&pos, &val)| val * dense[pos as usize])
        .sum()
}
}

Sparse Distance Metrics

HNSW Configuration

Configuration Parameters

Presets

Tuning Guidelines

graph TD
    subgraph "Higher m / ef"
        A[More connections per node]
        B[Better recall]
        C[More memory]
        D[Slower insert]
    end

    subgraph "Lower m / ef"
        E[Fewer connections]
        F[Lower recall]
        G[Less memory]
        H[Faster insert]
    end

    A --> B
    A --> C
    A --> D

    E --> F
    E --> G
    E --> H

Workload-Specific Tuning

Memory vs Recall Tradeoff

Performance Characteristics

Parallel Search Threshold

Automatic parallel iteration for datasets >5000 vectors:

#![allow(unused)]
fn main() {
const PARALLEL_THRESHOLD: usize = 5000;

if keys.len() >= PARALLEL_THRESHOLD {
    // Use rayon parallel iterator
    keys.par_iter().filter_map(...)
} else {
    // Use sequential iterator
    keys.iter().filter_map(...)
}
}

Benchmark Results

Supported Embedding Dimensions

Edge Cases and Gotchas

Zero-Magnitude Vectors

Dimension Mismatch Handling

#![allow(unused)]
fn main() {
// Mismatched dimensions are silently skipped during search
engine.store_embedding("2d", vec![1.0, 0.0])?;
engine.store_embedding("3d", vec![1.0, 0.0, 0.0])?;

// Search with 2D query only matches 2D vectors
let results = engine.search_similar(&[1.0, 0.0], 10)?;
assert_eq!(results.len(), 1);  // Only "2d" matched
}

HNSW Limitations

NaN/Infinity Handling

Sparse vector operations sanitize NaN/Inf results:

#![allow(unused)]
fn main() {
// cosine_similarity returns 0.0 for NaN/Inf
if result.is_nan() || result.is_infinite() {
    0.0
} else {
    result.clamp(-1.0, 1.0)
}

// cosine_distance_dense returns 1.0 (max distance) for NaN/Inf
if similarity.is_nan() || similarity.is_infinite() {
    1.0  // Maximum distance
} else {
    1.0 - similarity.clamp(-1.0, 1.0)
}
}

Best Practices

Memory Optimization

1. Use sparse vectors for high-sparsity data: Automatic at >50% zeros
2. Batch insert for HNSW: Build index once after all data loaded
3. Choose appropriate HNSW config: Don’t over-provision m/ef
4. Monitor memory with HNSWMemoryStats: Track dense vs sparse counts

#![allow(unused)]
fn main() {
let stats = index.memory_stats();
println!("Dense: {}, Sparse: {}, Total bytes: {}",
    stats.dense_count, stats.sparse_count, stats.embedding_bytes);
}

Search Performance

1. Pre-compute query magnitude: Done automatically in search
2. Use HNSW for >10K vectors: Brute-force for smaller sets
3. Tune ef_search: Higher for recall, lower for speed
4. Parallel threshold: Automatic at 5000 vectors

Unified Entity Best Practices

1. Use for cross-engine queries: When embeddings relate to graph/relational data
2. Entity key conventions: Use prefixes like user:, doc:, item:
3. Separate embedding namespace: Use store_embedding for isolated vectors

Dependencies

Note: wide (SIMD f32x8 operations) is a transitive dependency via tensor_store.

Related Modules

• Tensor Store - Underlying storage and HNSW implementation
• Query Router - Executes SIMILAR queries using VectorEngine
• Tensor Cache - Uses vector similarity for semantic caching

Tensor Compress

Module 8 of Neumann. Provides tensor-native compression exploiting the mathematical structure of high-dimensional embeddings.

The primary compression method is Tensor Train (TT) decomposition, which decomposes vectors reshaped as tensors into a chain of smaller 3D cores using successive SVD truncations. This achieves 10-40x compression for 1024+ dimension vectors while enabling similarity computations directly in compressed space.

Design Principles

1. Tensor Mathematics: Uses Tensor Train decomposition to exploit low-rank structure
2. Higher Dimensions Are Lower: Decomposes vectors into products of smaller tensors
3. Streaming I/O: Process large snapshots without loading entire dataset
4. Incremental Updates: Delta snapshots for efficient replication
5. Pure Rust: No external LAPACK/BLAS dependencies - fully portable

Key Types

Error Types

Architecture

graph TD
    subgraph tensor_compress
        TT[tensor_train.rs<br/>TT-SVD decomposition]
        DC[decompose.rs<br/>SVD implementation]
        FMT[format.rs<br/>Snapshot format]
        STR[streaming.rs<br/>Streaming I/O]
        STT[streaming_tt.rs<br/>Streaming TT]
        INC[incremental.rs<br/>Delta snapshots]
        DLT[delta.rs<br/>Delta + varint encoding]
        RLE[rle.rs<br/>Run-length encoding]
    end

    TT --> DC
    FMT --> TT
    FMT --> DLT
    FMT --> RLE
    STR --> FMT
    STT --> TT
    INC --> FMT

Tensor Train Decomposition

Algorithm Overview

The TT-SVD algorithm (Oseledets 2011) decomposes a vector by:

1. Reshape: Convert 1D vector to multi-dimensional tensor
2. Left-to-right sweep: For each mode k from 1 to n-1:
   • Left-unfold the current tensor into a matrix
   • Compute truncated SVD: A = U S Vt
   • Store U as the k-th core
   • Multiply S * Vt to get the remainder for next iteration
3. Final core: The last remainder becomes the final core

graph LR
    subgraph "TT-SVD Algorithm"
        V[Vector 4096-dim] --> R[Reshape to 8x8x8x8]
        R --> U1[Unfold mode 1<br/>64 x 64]
        U1 --> SVD1[SVD truncate<br/>rank=8]
        SVD1 --> C1[Core 1<br/>1x8x8]
        SVD1 --> R2[Remainder<br/>8x512]
        R2 --> SVD2[SVD truncate]
        SVD2 --> C2[Core 2<br/>8x8x8]
        SVD2 --> R3[Remainder]
        R3 --> SVD3[SVD truncate]
        SVD3 --> C3[Core 3<br/>8x8x8]
        SVD3 --> C4[Core 4<br/>8x8x1]
    end

Compression Example

For a 4096-dim embedding reshaped to (8, 8, 8, 8):

Original: 4096 floats = 16 KB
TT-cores: 1x8x8 + 8x8x8 + 8x8x8 + 8x8x1 = 64 + 512 + 512 + 64 = 1152 floats
With max_rank=8: 1x8x4 + 4x8x4 + 4x8x4 + 4x8x1 = 32 + 128 + 128 + 32 = 320 floats = 1.25 KB
Compression: 12.8x

SVD Implementation Details

The module implements two SVD algorithms:

1. Power Iteration with Deflation (small matrices)

Used when matrix dimensions are <= 32 or rank is close to matrix size:

#![allow(unused)]
fn main() {
// Simplified power iteration
fn power_iteration(a: &Matrix, max_iter: usize, tol: f32) -> (sigma, u, v) {
    // Initialize v randomly (deterministic seed)
    let mut v: Vec<f32> = (0..cols).map(|i| ((i * 7 + 3) % 13) as f32 / 13.0 - 0.5).collect();
    normalize(&mut v);

    for _ in 0..max_iter {
        // u = A * v, then normalize
        u = matmul(a, v);
        new_sigma = normalize(&mut u);

        // v = A^T * u, then normalize
        v = matmul_transpose(a, u);
        normalize(&mut v);

        // Check convergence
        if (new_sigma - sigma).abs() < tol * sigma.max(1.0) {
            return (new_sigma, u, v);
        }
        sigma = new_sigma;
    }
}
}

After finding each singular triplet, the algorithm deflates: A = A - sigma u vT

2. Randomized SVD (large matrices)

Uses the Halko-Martinsson-Tropp 2011 algorithm for matrices > 32 dimensions:

graph TD
    subgraph "Randomized SVD Pipeline"
        A[Input Matrix A<br/>m x n] --> OMEGA[Generate Gaussian<br/>Omega n x k+p]
        A --> SAMPLE[Y = A * Omega<br/>m x k+p]
        SAMPLE --> QR[QR decompose Y<br/>Q = orth basis]
        QR --> PROJECT[B = Q^T * A<br/>k+p x n]
        PROJECT --> SMALL_SVD[SVD of small B<br/>power iteration]
        SMALL_SVD --> RECONSTRUCT[U = Q * U_small]
    end

Key implementation details:

• Gaussian matrix generation: Uses a Linear Congruential Generator (LCG) with Box-Muller transform for deterministic, portable random numbers
• QR orthonormalization: Modified Gram-Schmidt for numerical stability
• Oversampling: Adds 5 extra columns to improve accuracy
• Convergence: 20 iterations max (sufficient for embedding vectors)

#![allow(unused)]
fn main() {
// LCG parameters from Numerical Recipes
fn lcg_next(state: &mut u64) -> u64 {
    *state = state.wrapping_mul(6_364_136_223_846_793_005)
                  .wrapping_add(1_442_695_040_888_963_407);
    *state
}

// Box-Muller transform for Gaussian
fn box_muller(u1: f32, u2: f32) -> (f32, f32) {
    let r = (-2.0 * u1.ln()).sqrt();
    let theta = 2.0 * PI * u2;
    (r * theta.cos(), r * theta.sin())
}
}

Optimal Shape Selection

The module includes hardcoded optimal shapes for common embedding dimensions:

For non-standard dimensions, factorize_balanced finds factors close to the nth root:

#![allow(unused)]
fn main() {
fn factorize_balanced(n: usize) -> Vec<usize> {
    // Target 2-6 factors based on log2(n)
    let target_factors = (ln(n) / ln(2)).ceil().clamp(2, 6);
    let target_size = n^(1/target_factors);

    // Greedily find factors close to target_size
    // ...
}
}

TT Operations

Where: n = number of modes, d = mode size, r = TT-rank

TT Gram Matrix Computation

Computing dot products and norms in TT space uses the Gram matrix approach:

#![allow(unused)]
fn main() {
// Gram matrix propagation for dot product
fn tt_dot_product(a: &TTVector, b: &TTVector) -> f32 {
    let mut gram = vec![1.0f32];  // Start with 1x1 identity

    for (core_a, core_b) in a.cores.iter().zip(b.cores.iter()) {
        let (r1a, n, r2a) = core_a.shape;
        let (r1b, _, r2b) = core_b.shape;
        let mut new_gram = vec![0.0; r2a * r2b];

        // Contract: new_gram[a,b] = sum_{k,i,j} gram[i,j] * A[i,k,a] * B[j,k,b]
        for a_idx in 0..r2a {
            for b_idx in 0..r2b {
                for k in 0..n {
                    for ia in 0..r1a {
                        for ib in 0..r1b {
                            let g = gram[ia * r1b + ib];
                            new_gram[a_idx * r2b + b_idx] +=
                                g * core_a.get(ia, k, a_idx) * core_b.get(ib, k, b_idx);
                        }
                    }
                }
            }
        }
        gram = new_gram;
    }

    gram[0]  // Final 1x1 Gram matrix
}
}

Usage

#![allow(unused)]
fn main() {
use tensor_compress::{tt_decompose, tt_reconstruct, tt_cosine_similarity, TTConfig};

let embedding: Vec<f32> = get_embedding();  // 4096-dim
let config = TTConfig::for_dim(4096)?;

// Decompose
let tt = tt_decompose(&embedding, &config)?;
println!("Compression: {:.1}x", tt.compression_ratio());
println!("Storage: {} floats", tt.storage_size());
println!("Max rank: {}", tt.max_rank());

// Reconstruct
let restored = tt_reconstruct(&tt);

// Compute similarity without reconstruction
let tt2 = tt_decompose(&other_embedding, &config)?;
let sim = tt_cosine_similarity(&tt, &tt2)?;
}

Batch Operations

Batch operations use rayon for parallel processing when handling 4+ vectors:

#![allow(unused)]
fn main() {
use tensor_compress::{tt_decompose_batch, tt_cosine_similarity_batch, TTConfig};

let vectors: Vec<Vec<f32>> = load_embeddings();
let config = TTConfig::for_dim(4096)?;

// Batch decompose (parallel for 4+ vectors)
let refs: Vec<&[f32]> = vectors.iter().map(|v| v.as_slice()).collect();
let tts = tt_decompose_batch(&refs, &config)?;

// Batch similarity search
let query_tt = &tts[0];
let similarities = tt_cosine_similarity_batch(query_tt, &tts[1..])?;

// Find top-k
let mut indexed: Vec<_> = similarities.iter().enumerate().collect();
indexed.sort_by(|a, b| b.1.partial_cmp(a.1).unwrap());
let top_5: Vec<_> = indexed.iter().take(5).collect();
}

The parallel threshold constant is:

#![allow(unused)]
fn main() {
const PARALLEL_THRESHOLD: usize = 4;
}

Configuration

TTConfig Presets

TTConfig Validation

#![allow(unused)]
fn main() {
impl TTConfig {
    pub fn validate(&self) -> Result<(), TTError> {
        if self.shape.is_empty() {
            return Err(TTError::InvalidShape("empty shape".into()));
        }
        if self.shape.contains(&0) {
            return Err(TTError::InvalidShape("shape contains zero".into()));
        }
        if self.max_rank < 1 {
            return Err(TTError::InvalidRank);
        }
        if self.tolerance <= 0.0 || self.tolerance > 1.0 || !self.tolerance.is_finite() {
            return Err(TTError::InvalidTolerance(self.tolerance));
        }
        Ok(())
    }
}
}

CompressionConfig

#![allow(unused)]
fn main() {
pub struct CompressionConfig {
    pub tensor_mode: Option<TensorMode>,  // TT compression for vectors
    pub delta_encoding: bool,             // For sorted ID lists
    pub rle_encoding: bool,               // For repeated values
}

// Presets
CompressionConfig::high_compression()  // max_rank=4, all encodings enabled
CompressionConfig::balanced(dim)       // max_rank=8, all encodings enabled
CompressionConfig::high_accuracy(dim)  // max_rank=16, all encodings enabled
}

Dimension Presets

Streaming Operations

State Machine

stateDiagram-v2
    [*] --> Created: new()
    Created --> Writing: write_entry() / write_vector()
    Writing --> Writing: write_entry() / write_vector()
    Writing --> Finishing: finish()
    Finishing --> [*]: success

    note right of Created
        Magic bytes written
        entry_count = 0
    end note

    note right of Writing
        Length-prefixed entries
        entry_count incremented
    end note

    note right of Finishing
        Trailer written with:
        - entry_count
        - config
        - data_start offset
    end note

File Format

Uses a trailer-based header so entry count is known at the end:

+------------------------+
| Magic (NEUS/NEUT)  4B  |  Identifies streaming snapshot/TT
+------------------------+
| Entry 1 length     4B  |  Little-endian u32
+------------------------+
| Entry 1 data      var  |  Bincode-serialized entry
+------------------------+
| Entry 2 length     4B  |
+------------------------+
| Entry 2 data      var  |
+------------------------+
| ...                    |
+------------------------+
| Trailer           var  |  Bincode-serialized header
+------------------------+
| Trailer size       8B  |  Little-endian u64
+------------------------+

Security limits:

• Maximum trailer size: 1 MB (MAX_TRAILER_SIZE)
• Maximum entry size: 100 MB (MAX_ENTRY_SIZE)

Usage

#![allow(unused)]
fn main() {
use tensor_compress::streaming::{StreamingWriter, StreamingReader};

// Write entries one at a time
let mut writer = StreamingWriter::new(file, config)?;
for entry in entries {
    writer.write_entry(&entry)?;
}
writer.finish()?;

// Read entries one at a time (iterator-based)
let reader = StreamingReader::open(file)?;
println!("Entry count: {}", reader.entry_count());
for entry in reader {
    process(entry?);
}
}

Streaming TT Operations

#![allow(unused)]
fn main() {
use tensor_compress::streaming_tt::{StreamingTTWriter, StreamingTTReader,
    streaming_tt_similarity_search};

// Create streaming TT file
let config = TTConfig::for_dim(768)?;
let mut writer = StreamingTTWriter::new(file, config.clone())?;

for vector in vectors {
    writer.write_vector(&vector)?;  // Decompose on-the-fly
}
writer.finish()?;

// Similarity search without loading all into memory
let query_tt = tt_decompose(&query, &config)?;
let top_10 = streaming_tt_similarity_search(file, &query_tt, 10)?;
// Returns Vec<(index, similarity)> sorted by descending similarity
}

Merge and Convert Operations

#![allow(unused)]
fn main() {
use tensor_compress::streaming::{convert_to_streaming, read_streaming_to_snapshot,
    merge_streaming};

// Convert non-streaming snapshot to streaming format
let count = convert_to_streaming(&snapshot, output_file)?;

// Read streaming format into full snapshot (for compatibility)
let snapshot = read_streaming_to_snapshot(file)?;

// Merge multiple streaming snapshots
let count = merge_streaming(vec![file1, file2, file3], output, config)?;
}

Incremental Updates

Delta Snapshot Architecture

graph TD
    subgraph "Delta Chain"
        BASE[Base Snapshot<br/>Seq 0] --> D1[Delta 1<br/>Seq 1-10]
        D1 --> D2[Delta 2<br/>Seq 11-25]
        D2 --> D3[Delta 3<br/>Seq 26-30]
    end

    subgraph "Compaction"
        BASE2[Base] --> COMPACT[Compacted<br/>Snapshot]
        D1_2[Delta 1] --> COMPACT
        D2_2[Delta 2] --> COMPACT
        D3_2[Delta 3] --> COMPACT
    end

Delta Entry Types

#![allow(unused)]
fn main() {
pub enum ChangeType {
    Put,    // Entry was added or updated
    Delete, // Entry was deleted
}

pub struct DeltaEntry {
    pub key: String,
    pub change: ChangeType,
    pub value: Option<CompressedEntry>,  // None for Delete
    pub sequence: u64,
}
}

Usage

#![allow(unused)]
fn main() {
use tensor_compress::incremental::{DeltaBuilder, DeltaChain, apply_delta,
    merge_deltas, diff_snapshots};

// Create delta
let mut builder = DeltaBuilder::new("base_snapshot_id", sequence);
builder.put("key1", entry1);
builder.delete("key2");
let delta = builder.build();

// Apply delta
let new_snapshot = apply_delta(&base, &delta)?;

// Chain management
let mut chain = DeltaChain::new(base_snapshot);
chain.push(delta1)?;
chain.push(delta2)?;
let value = chain.get("key1");  // Checks chain then base

// Compact when chain grows long
if chain.should_compact(10) {
    let compacted = chain.compact()?;
}

// Compare two snapshots
let delta = diff_snapshots(&old_snapshot, &new_snapshot, "old_id")?;

// Merge multiple deltas into one
let merged = merge_deltas(&[delta1, delta2, delta3])?;
}

Delta Operations

Delta Format

+------------------------+
| Magic (NEUD)       4B  |
+------------------------+
| Version            2B  |
+------------------------+
| Base ID           var  |  String (length-prefixed)
+------------------------+
| Sequence Range     16B |  (start, end) u64 pair
+------------------------+
| Change Count        8B |
+------------------------+
| Created At          8B |  Unix timestamp
+------------------------+
| Entries           var  |  Bincode-serialized Vec<DeltaEntry>
+------------------------+

Lossless Compression

Delta + Varint Encoding

For sorted integer sequences (node IDs, timestamps):

graph LR
    subgraph "Delta + Varint Pipeline"
        IDS[IDs: 100, 101, 102, 105, 110] --> DELTA[Delta encode:<br/>100, 1, 1, 3, 5]
        DELTA --> VARINT[Varint encode]
        VARINT --> OUT[Bytes: ~7 bytes<br/>vs 40 bytes raw]
    end

Algorithm:

#![allow(unused)]
fn main() {
// Delta encoding: store first value, then differences
pub fn delta_encode(ids: &[u64]) -> Vec<u64> {
    let mut result = vec![ids[0]];
    for window in ids.windows(2) {
        result.push(window[1].saturating_sub(window[0]));
    }
    result
}

// Varint encoding: 7 bits per byte, high bit = continuation
pub fn varint_encode(values: &[u64]) -> Vec<u8> {
    let mut result = Vec::with_capacity(values.len() * 2);
    for &value in values {
        let mut v = value;
        loop {
            let byte = (v & 0x7f) as u8;
            v >>= 7;
            if v == 0 {
                result.push(byte);  // Final byte (no continuation)
                break;
            }
            result.push(byte | 0x80);  // Continuation bit set
        }
    }
    result
}
}

Usage:

#![allow(unused)]
fn main() {
use tensor_compress::{compress_ids, decompress_ids};

let ids: Vec<u64> = (1000..2000).collect();
let compressed = compress_ids(&ids);  // ~100 bytes vs 8000

let restored = decompress_ids(&compressed);
assert_eq!(ids, restored);
}

Varint byte sizes:

Run-Length Encoding

For repeated values:

#![allow(unused)]
fn main() {
use tensor_compress::{rle_encode, rle_decode};

let statuses = vec!["active"; 1000];
let encoded = rle_encode(&statuses);
assert_eq!(encoded.runs(), 1);  // Single run

// Storage: 1 string + 1 u32 = ~12 bytes vs 6000+ bytes
}

Internal representation:

#![allow(unused)]
fn main() {
pub struct RleEncoded<T: Eq> {
    pub values: Vec<T>,      // Unique values in order
    pub run_lengths: Vec<u32>, // Count for each value
}
}

Compression scenarios:

Sparse Vector Format

For vectors with >50% zeros:

#![allow(unused)]
fn main() {
use tensor_compress::{compress_sparse, compress_dense_as_sparse,
    should_use_sparse, should_use_sparse_threshold};

// Direct sparse compression
let positions = vec![0, 50, 99];
let values = vec![1.0, 2.0, 3.0];
let compressed = compress_sparse(100, &positions, &values);

// Auto-detect and compress
if should_use_sparse_threshold(&vector, 0.5) {
    let compressed = compress_dense_as_sparse(&vector);
}

// Check if sparse is beneficial
if should_use_sparse(dimension, non_zero_count) {
    // Use sparse format
}
}

Storage calculation:

#![allow(unused)]
fn main() {
// sparse_storage_size = 8 + 8 + nnz*2 + nnz*4 = 16 + nnz*6
// Dense storage = dimension * 4

// Sparse is better when: 16 + nnz*6 < dimension*4
// Solving: nnz < (dimension*4 - 16) / 6 = dimension*0.67 - 2.67
}

Compressed Value Types

#![allow(unused)]
fn main() {
pub enum CompressedValue {
    Scalar(CompressedScalar),           // Int, Float, String, Bool, Null
    VectorRaw(Vec<f32>),                // Uncompressed
    VectorTT { cores, original_dim, shape, ranks },  // TT-compressed
    VectorSparse { dimension, positions, values },   // Sparse
    IdList(Vec<u8>),                    // Delta + varint encoded
    RleInt(RleEncoded<i64>),            // RLE encoded integers
    Pointer(String),                    // Single pointer
    Pointers(Vec<String>),              // Multiple pointers
}
}

Automatic Format Selection

#![allow(unused)]
fn main() {
pub fn compress_vector(vector: &[f32], key: &str, field_name: &str,
    config: &CompressionConfig) -> Result<CompressedValue, FormatError> {

    // 1. Check for embedding-like keys
    let is_embedding = key.starts_with("emb:") ||
                       field_name == "_embedding" ||
                       field_name == "vector";

    if is_embedding {
        if let Some(TensorMode::TensorTrain(tt_config)) = &config.tensor_mode {
            return Ok(CompressedValue::VectorTT { ... });
        }
    }

    // 2. Check for ID list pattern
    if config.delta_encoding && looks_like_id_list(vector, field_name) {
        return Ok(CompressedValue::IdList(...));
    }

    // 3. Fall back to raw
    Ok(CompressedValue::VectorRaw(vector.to_vec()))
}
}

Performance

Benchmarks on Apple M4 (aarch64, MacBook Air 24GB), release build:

Batch operations (768-dim, 1000 vectors):

Throughput: 39,318 vectors/sec (768-dim decomposition)

Industry Comparison

Edge Cases and Gotchas

Vector Content Patterns

Numerical Edge Cases

#![allow(unused)]
fn main() {
// Very small values (denormalized floats)
let tiny: Vec<f32> = (0..64).map(|i| (i as f32) * 1e-38).collect();
// Works, but may lose precision

// Large values (1e6 range)
let large: Vec<f32> = (0..64).map(|i| (i as f32) * 1e6).collect();
// Works, no overflow

// Prime dimensions
let prime_127: Vec<f32> = (0..127).map(|i| (i as f32 * 0.1).sin()).collect();
// Works but may have poor compression
}

Streaming Gotchas

1. Incomplete files: Magic bytes are written first, but entry count is in trailer. If writer crashes before finish(), the file is corrupt.

2. Memory limits: MAX_ENTRY_SIZE = 100MB and MAX_TRAILER_SIZE = 1MB prevent allocation attacks. Exceeding these returns an error.

3. Seek requirement: StreamingReader::open requires Seek to read the trailer. For non-seekable streams, use read_streaming_to_snapshot which buffers.

Delta Chain Gotchas

1. Chain length: Default max_chain_len = 100. After this, push() returns ChainTooLong error. Call compact() periodically.

2. Sequence gaps: Deltas should have contiguous sequences. The merge_deltas function only keeps the latest state per key.

3. Base reference: Deltas store a base_id string but don’t validate it exists. Your application must track base snapshots.

Performance Tips and Best Practices

Choosing Configuration

#![allow(unused)]
fn main() {
// For search/retrieval (similarity queries)
let config = TTConfig::for_dim(dim)?;  // Balanced

// For archival/cold storage
let config = TTConfig::high_compression(dim)?;  // Smaller, slower queries

// For real-time applications
let config = TTConfig::high_accuracy(dim)?;  // Larger, faster queries
}

Batch Size Optimization

#![allow(unused)]
fn main() {
// Below parallel threshold (4), sequential is faster
// due to thread spawn overhead
let small_batch = tt_decompose_batch(&vectors[..3], &config);  // Sequential

// At threshold, parallel kicks in
let large_batch = tt_decompose_batch(&vectors, &config);  // Parallel if >= 4
}

Memory Efficiency

#![allow(unused)]
fn main() {
// Bad: Load all, then process
let all_vectors = read_streaming_tt_all(file)?;  // Loads all into memory

// Good: Stream process
for tt in StreamingTTReader::open(file)? {
    process(tt?);  // One at a time
}

// Best: Use streaming search
let results = streaming_tt_similarity_search(file, &query_tt, 10)?;
}

Delta Compaction Strategy

#![allow(unused)]
fn main() {
let mut chain = DeltaChain::new(base);

// After N deltas or M total changes
if chain.len() >= 10 || total_changes >= 10000 {
    let new_base = chain.compact()?;
    chain = DeltaChain::new(new_base);
}
}

Dependencies

• serde: Serialization traits
• bincode: Binary format
• thiserror: Error types
• rayon: Parallel batch operations

No external LAPACK/BLAS - pure Rust SVD implementation.

Related Modules

Tensor Vault

Tensor Vault provides secure secret storage with AES-256-GCM encryption and graph-based access control. Designed for multi-agent environments, it implements a zero-trust architecture where access is determined by graph topology rather than traditional ACLs.

All secrets are encrypted at rest with authenticated encryption. The vault maintains a permanent audit trail of all operations and supports features like rate limiting, TTL-based grants, and namespace isolation for multi-tenant deployments.

Design Principles

Key Types

Core Types

Cryptographic Types

Access Control Types

Audit Types

Architecture

graph TB
    subgraph "Tensor Vault"
        API[Vault API]
        AC[AccessController]
        Cipher[Cipher<br/>AES-256-GCM]
        KDF[MasterKey<br/>Argon2id + HKDF]
        Obf[Obfuscator<br/>HMAC + Padding]
        Audit[AuditLog]
        TTL[GrantTTLTracker]
        RL[RateLimiter]
    end

    subgraph "Storage"
        TS[TensorStore]
        GE[GraphEngine]
    end

    API --> AC
    API --> Cipher
    API --> Obf
    API --> Audit
    API --> TTL
    API --> RL

    AC --> GE
    Cipher --> KDF
    Obf --> KDF

    API --> TS
    Audit --> TS

Data Flow

1. Set Operation: Plaintext is padded, encrypted with random nonce, metadata obfuscated, stored via TensorStore
2. Get Operation: Rate limit check, access path verified via BFS, ciphertext decrypted, padding removed, audit logged
3. Grant Operation: Permission edge created in GraphEngine, TTL tracked if specified
4. Revoke Operation: Permission edge deleted, expired grants cleaned up

Set Operation Flow

sequenceDiagram
    participant C as Client
    participant V as Vault
    participant RL as RateLimiter
    participant AC as AccessController
    participant O as Obfuscator
    participant Ci as Cipher
    participant TS as TensorStore
    participant GE as GraphEngine
    participant A as AuditLog

    C->>V: set(requester, key, value)
    V->>RL: check_rate_limit(requester, Set)
    alt Rate Limited
        RL-->>V: RateLimited error
        V-->>C: Error
    end

    alt New Secret
        V->>V: Check requester == ROOT
        alt Not Root
            V-->>C: AccessDenied
        end
    else Update
        V->>AC: check_path_with_permission(Write)
    end

    V->>O: pad_plaintext(value)
    O-->>V: padded_value
    V->>Ci: encrypt(padded_value)
    Ci-->>V: (ciphertext, nonce)

    V->>O: generate_storage_id(key, nonce)
    O-->>V: blob_key
    V->>TS: put(blob_key, ciphertext)

    V->>O: obfuscate_key(key)
    O-->>V: obfuscated_key
    V->>O: encrypt_metadata(creator)
    V->>O: encrypt_metadata(timestamp)
    V->>TS: put(_vk:obfuscated_key, metadata)

    alt New Secret
        V->>GE: add_entity_edge(ROOT, secret_node, VAULT_ACCESS_ADMIN)
    end

    V->>A: record(requester, key, Set)
    V-->>C: Ok(())

Access Control Model

Access is determined by graph topology using BFS traversal:

node:root ──VAULT_ACCESS_ADMIN──> vault_secret:api_key
                                          ^
user:alice ──VAULT_ACCESS_READ───────────┘
                                          ^
team:devs ──VAULT_ACCESS_WRITE───────────┘
      ^
user:bob ──MEMBER────────────────────────┘

Permission Levels

Permission propagation follows graph paths. The effective permission is determined by the VAULT_ACCESS_* edge type at the end of the path.

Allowed Traversal Edges

Only these edge types can grant transitive access:

• VAULT_ACCESS - Legacy edge type (treated as Admin for backward compatibility)
• VAULT_ACCESS_READ - Read-only access
• VAULT_ACCESS_WRITE - Read + Write access
• VAULT_ACCESS_ADMIN - Full access including grant/revoke
• MEMBER - Allows group membership traversal but does NOT grant permission directly

Access Control Algorithm

The AccessController uses BFS to find the best permission level along any path:

#![allow(unused)]
fn main() {
// Simplified algorithm from access.rs
pub fn get_permission_level(graph: &GraphEngine, source: &str, target: &str) -> Option<Permission> {
    if source == target {
        return Some(Permission::Admin);  // Self-access
    }

    let mut visited = HashSet::new();
    let mut queue = VecDeque::new();
    let mut best_permission: Option<Permission> = None;

    queue.push_back(source.to_string());
    visited.insert(source.to_string());

    while let Some(current) = queue.pop_front() {
        for edge in graph.get_entity_outgoing(&current) {
            let (_, to, edge_type, _) = graph.get_entity_edge(&edge);

            // Only traverse allowed edge types
            if !is_allowed_edge_type(&edge_type) {
                continue;
            }

            // VAULT_ACCESS_* edges grant permission to target
            if edge_type.starts_with("VAULT_ACCESS") && to == target {
                if let Some(perm) = Permission::from_edge_type(&edge_type) {
                    best_permission = max(best_permission, perm);
                }
            } else if edge_type == "MEMBER" {
                // MEMBER edges allow traversal but NO permission grant
                if !visited.contains(&to) {
                    visited.insert(to.clone());
                    queue.push_back(to);
                }
            }
        }
    }

    best_permission
}
}

Security Note: MEMBER edges enable traversal through groups but do not grant permissions. Only VAULT_ACCESS_* edges grant actual permissions. This prevents privilege escalation via group membership.

Access Control Flow

flowchart TD
    Start([Check Access]) --> IsRoot{Is requester ROOT?}
    IsRoot -->|Yes| Granted([Access Granted - Admin])
    IsRoot -->|No| BFS[Start BFS from requester]

    BFS --> Queue{Queue empty?}
    Queue -->|Yes| CheckBest{Best permission found?}
    Queue -->|No| Pop[Pop next node]

    Pop --> GetEdges[Get outgoing edges]
    GetEdges --> ForEdge{For each edge}

    ForEdge --> IsAllowed{Edge type allowed?}
    IsAllowed -->|No| ForEdge
    IsAllowed -->|Yes| IsVaultAccess{VAULT_ACCESS_* ?}

    IsVaultAccess -->|Yes| IsTarget{Points to target?}
    IsTarget -->|Yes| UpdateBest[Update best permission]
    IsTarget -->|No| ForEdge
    UpdateBest --> ForEdge

    IsVaultAccess -->|No| IsMember{MEMBER edge?}
    IsMember -->|Yes| AddQueue[Add destination to queue]
    IsMember -->|No| ForEdge
    AddQueue --> ForEdge

    ForEdge -->|Done| Queue

    CheckBest -->|Yes| CheckLevel{Permission >= required?}
    CheckBest -->|No| Denied([Access Denied])
    CheckLevel -->|Yes| Granted2([Access Granted])
    CheckLevel -->|No| Insufficient([Insufficient Permission])

Storage Format

Secrets use a two-tier storage model for security:

Metadata Tensor

Storage key: _vk:{HMAC(key)} (key name obfuscated via HMAC-BLAKE2b)

Ciphertext Blob

Storage key: _vs:{HMAC(key, nonce)} (random-looking storage ID)

Storage Key Structure

_vault:salt          - Persisted 16-byte salt for key derivation
_vk:<32-hex-chars>   - Metadata tensor (HMAC of secret key)
_vs:<24-hex-chars>   - Ciphertext blob (HMAC of key + nonce)
_va:<timestamp>:<counter> - Audit log entries
_vault_ttl_grants    - Persisted TTL grants (JSON)
vault_secret:<32-hex-chars> - Secret node for graph access control

Encryption

Key Derivation

Master key derived using Argon2id with HKDF-based subkey separation:

#![allow(unused)]
fn main() {
// From key.rs - Argon2id parameters
pub const SALT_SIZE: usize = 16;  // 128-bit salt
pub const KEY_SIZE: usize = 32;   // 256-bit key (AES-256)

// Default VaultConfig values:
// argon2_memory_cost: 65536 (64 MiB)
// argon2_time_cost: 3 (iterations)
// argon2_parallelism: 4 (threads)

// Argon2id configuration
let params = Params::new(
    config.argon2_memory_cost,  // Memory in KiB
    config.argon2_time_cost,    // Iterations
    config.argon2_parallelism,  // Parallelism
    Some(KEY_SIZE),             // Output length
)?;

let argon2 = Argon2::new(Algorithm::Argon2id, Version::V0x13, params);
argon2.hash_password_into(input, salt, &mut key)?;
}

Argon2id Security Properties:

• Hybrid algorithm: Argon2i (side-channel resistant) + Argon2d (GPU resistant)
• Memory-hard: Requires 64 MiB by default, defeating GPU/ASIC attacks
• Time-hard: 3 iterations increase computation time
• Parallelism: 4 threads to utilize modern CPUs

HKDF Subkey Derivation

Each purpose gets a cryptographically independent key via HKDF-SHA256:

#![allow(unused)]
fn main() {
// From key.rs - Domain-separated subkeys
impl MasterKey {
    pub fn derive_subkey(&self, domain: &[u8]) -> [u8; KEY_SIZE] {
        let hk = Hkdf::<Sha256>::new(None, &self.bytes);
        let mut output = [0u8; KEY_SIZE];
        hk.expand(domain, &mut output).expect("HKDF expand cannot fail for 32 bytes");
        output
    }

    pub fn encryption_key(&self) -> [u8; KEY_SIZE] {
        self.derive_subkey(b"neumann_vault_encryption_v1")
    }

    pub fn obfuscation_key(&self) -> [u8; KEY_SIZE] {
        self.derive_subkey(b"neumann_vault_obfuscation_v1")
    }

    pub fn metadata_key(&self) -> [u8; KEY_SIZE] {
        self.derive_subkey(b"neumann_vault_metadata_v1")
    }
}
}

Key Hierarchy:

Master Password + Salt
        │
        ▼ Argon2id
    MasterKey (32 bytes)
        │
        ├──▶ HKDF("encryption_v1") ──▶ AES-256-GCM key
        ├──▶ HKDF("obfuscation_v1") ──▶ HMAC key for obfuscation
        └──▶ HKDF("metadata_v1") ──▶ AES-256-GCM key for metadata

Salt Persistence

The vault automatically manages salt persistence:

#![allow(unused)]
fn main() {
// From lib.rs - Salt handling on vault creation
pub fn new(master_key: &[u8], graph: Arc<GraphEngine>, store: TensorStore, config: VaultConfig) -> Result<Self> {
    let derived = if config.salt.is_some() {
        // Explicit salt provided - use it directly
        let (key, _) = MasterKey::derive(master_key, &config)?;
        key
    } else if let Some(persisted_salt) = Self::load_salt(&store) {
        // Use persisted salt for consistency across reopens
        MasterKey::derive_with_salt(master_key, &persisted_salt, &config)?
    } else {
        // Generate new random salt and persist it
        let (key, new_salt) = MasterKey::derive(master_key, &config)?;
        Self::save_salt(&store, new_salt)?;
        key
    };
    // ...
}
}

Encryption Process

1. Pad plaintext to fixed bucket size (256B, 1KB, 4KB, 16KB, 32KB, or 64KB)
2. Generate random 12-byte nonce
3. Encrypt with AES-256-GCM
4. Store ciphertext and nonce separately

#![allow(unused)]
fn main() {
// From encryption.rs
pub const NONCE_SIZE: usize = 12;  // 96-bit nonce (AES-GCM standard)

impl Cipher {
    pub fn encrypt(&self, plaintext: &[u8]) -> Result<(Vec<u8>, [u8; NONCE_SIZE])> {
        let cipher = Aes256Gcm::new_from_slice(self.key.as_bytes())?;

        // Generate random nonce - CRITICAL for security
        let mut nonce_bytes = [0u8; NONCE_SIZE];
        rand::thread_rng().fill_bytes(&mut nonce_bytes);
        let nonce = Nonce::from_slice(&nonce_bytes);

        // AES-GCM provides authenticated encryption
        // Output: ciphertext || 16-byte authentication tag
        let ciphertext = cipher.encrypt(nonce, plaintext)?;

        Ok((ciphertext, nonce_bytes))
    }

    pub fn decrypt(&self, ciphertext: &[u8], nonce_bytes: &[u8]) -> Result<Vec<u8>> {
        if nonce_bytes.len() != NONCE_SIZE {
            return Err(VaultError::CryptoError("Invalid nonce size"));
        }

        let cipher = Aes256Gcm::new_from_slice(self.key.as_bytes())?;
        let nonce = Nonce::from_slice(nonce_bytes);

        // Decryption verifies authentication tag
        // Fails if ciphertext was tampered
        cipher.decrypt(nonce, ciphertext)
    }
}
}

AES-256-GCM Security Properties:

• Authenticated encryption: Detects tampering via 128-bit authentication tag
• Nonce requirement: Each encryption MUST use a unique nonce
• Ciphertext expansion: 16 bytes larger than plaintext (auth tag)

Obfuscation Layers

Padding Bucket Sizes

#![allow(unused)]
fn main() {
// From obfuscation.rs
pub enum PaddingSize {
    Small = 256,        // API keys, tokens
    Medium = 1024,      // Certificates, small configs
    Large = 4096,       // Private keys, large configs
    ExtraLarge = 16384, // Very large secrets
    Huge = 32768,       // Oversized secrets
    Maximum = 65536,    // Maximum supported
}

// Bucket selection (includes 4-byte length prefix + 1 byte min padding)
pub fn for_length(len: usize) -> Option<Self> {
    let min_required = len + 5;  // length prefix + min padding

    if min_required <= 256 { Some(Small) }
    else if min_required <= 1024 { Some(Medium) }
    else if min_required <= 4096 { Some(Large) }
    else if min_required <= 16384 { Some(ExtraLarge) }
    else if min_required <= 32768 { Some(Huge) }
    else if min_required <= 65536 { Some(Maximum) }
    else { None }
}
}

Padding Format

+----------------+-------------------+------------------+
| Length (4B LE) | Plaintext (N B)   | Random Padding   |
+----------------+-------------------+------------------+
|<--------------- Bucket Size (256/1K/4K/...) -------->|

#![allow(unused)]
fn main() {
// From obfuscation.rs
pub fn pad_plaintext(plaintext: &[u8]) -> Result<Vec<u8>> {
    let target_size = PaddingSize::for_length(plaintext.len())? as usize;
    let padding_len = target_size - 4 - plaintext.len();  // 4 = length prefix

    let mut padded = Vec::with_capacity(target_size);

    // Store original length as u32 little-endian
    let len_bytes = (plaintext.len() as u32).to_le_bytes();
    padded.extend_from_slice(&len_bytes);

    // Original data
    padded.extend_from_slice(plaintext);

    // Random padding (not zeros - prevents padding oracle attacks)
    let mut rng_bytes = vec![0u8; padding_len];
    rand::thread_rng().fill_bytes(&mut rng_bytes);
    padded.extend_from_slice(&rng_bytes);

    Ok(padded)
}
}

HMAC-BLAKE2b Construction

#![allow(unused)]
fn main() {
// From obfuscation.rs - HMAC construction for key obfuscation
fn hmac_hash(&self, data: &[u8], domain: &[u8]) -> [u8; 32] {
    // Inner hash: H((key XOR ipad) || domain || data)
    let mut inner_key = self.obfuscation_key;
    for byte in &mut inner_key {
        *byte ^= 0x36;  // ipad
    }
    let mut inner_hasher = Blake2b::<U32>::new();
    inner_hasher.update(inner_key);
    inner_hasher.update(domain);
    inner_hasher.update(data);
    let inner_hash = inner_hasher.finalize();

    // Outer hash: H((key XOR opad) || inner_hash)
    let mut outer_key = self.obfuscation_key;
    for byte in &mut outer_key {
        *byte ^= 0x5c;  // opad
    }
    let mut outer_hasher = Blake2b::<U32>::new();
    outer_hasher.update(outer_key);
    outer_hasher.update(inner_hash);

    outer_hasher.finalize().into()
}
}

Metadata AEAD Encryption

#![allow(unused)]
fn main() {
// From obfuscation.rs - Per-record AEAD encryption
pub fn encrypt_metadata(&self, data: &[u8]) -> Result<Vec<u8>> {
    let cipher = Aes256Gcm::new_from_slice(&self.metadata_key)?;

    // Random nonce for each encryption
    let mut nonce_bytes = [0u8; 12];
    rand::thread_rng().fill_bytes(&mut nonce_bytes);
    let nonce = Nonce::from_slice(&nonce_bytes);

    let ciphertext = cipher.encrypt(nonce, data)?;

    // Format: nonce || ciphertext
    let mut result = Vec::with_capacity(12 + ciphertext.len());
    result.extend_from_slice(&nonce_bytes);
    result.extend(ciphertext);
    Ok(result)
}
}

Rate Limiting

Rate limiting uses a sliding window algorithm to prevent brute-force attacks:

#![allow(unused)]
fn main() {
// From rate_limit.rs
pub struct RateLimiter {
    // (entity, operation) -> timestamps of recent requests
    history: DashMap<(String, String), VecDeque<Instant>>,
    config: RateLimitConfig,
}

impl RateLimiter {
    pub fn check_and_record(&self, entity: &str, op: Operation) -> Result<(), String> {
        let limit = op.limit(&self.config);
        if limit == u32::MAX {
            return Ok(());  // Unlimited
        }

        let key = (entity.to_string(), op.as_str().to_string());
        let now = Instant::now();
        let window_start = now - self.config.window;

        let mut entry = self.history.entry(key).or_default();
        let timestamps = entry.value_mut();

        // Remove expired entries outside window
        while let Some(front) = timestamps.front() {
            if *front < window_start {
                timestamps.pop_front();
            } else {
                break;
            }
        }

        let count = timestamps.len() as u32;
        if count >= limit {
            Err(format!("Rate limit exceeded: {} {} calls in {:?}", count, op, self.config.window))
        } else {
            timestamps.push_back(now);  // Record this request
            Ok(())
        }
    }
}
}

Sliding Window Visualization

Window: 60 seconds
Limit: 5 requests

Timeline:
|--[req1]--[req2]---[req3]--[req4]---[req5]---|
|<------------------ Window ----------------->|
                                               ^
                                               Now (6th request blocked)

After 10 seconds:
                    |--[req2]---[req3]--[req4]---[req5]---|
   [req1] expired   |<------------------ Window --------->|
                                                           ^
                                                           Now (6th request allowed)

Rate Limit Configuration Presets

#![allow(unused)]
fn main() {
// Default configuration
impl Default for RateLimitConfig {
    fn default() -> Self {
        Self {
            max_gets: 60,    // 60 get() calls per minute
            max_lists: 10,   // 10 list() calls per minute
            max_sets: 30,    // 30 set() calls per minute
            max_grants: 20,  // 20 grant() calls per minute
            window: Duration::from_secs(60),
        }
    }
}

// Strict configuration for testing
pub fn strict() -> Self {
    Self {
        max_gets: 5,
        max_lists: 2,
        max_sets: 3,
        max_grants: 2,
        window: Duration::from_secs(60),
    }
}

// No rate limiting
pub fn unlimited() -> Self {
    Self {
        max_gets: u32::MAX,
        max_lists: u32::MAX,
        max_sets: u32::MAX,
        max_grants: u32::MAX,
        window: Duration::from_secs(60),
    }
}
}

Note: node:root is exempt from rate limiting.

TTL Grant Tracking

TTL grants use a min-heap for efficient expiration tracking:

#![allow(unused)]
fn main() {
// From ttl.rs
pub struct GrantTTLTracker {
    // Priority queue of expiration times (min-heap)
    heap: Mutex<BinaryHeap<GrantTTLEntry>>,
}

struct GrantTTLEntry {
    expires_at: Instant,
    entity: String,
    secret_key: String,
}

// Reverse ordering for min-heap (earliest expiration first)
impl Ord for GrantTTLEntry {
    fn cmp(&self, other: &Self) -> Ordering {
        other.expires_at.cmp(&self.expires_at)  // Reversed!
    }
}
}

TTL Operations

#![allow(unused)]
fn main() {
// Add a grant with TTL
pub fn add(&self, entity: &str, secret_key: &str, ttl: Duration) {
    let entry = GrantTTLEntry {
        expires_at: Instant::now() + ttl,
        entity: entity.to_string(),
        secret_key: secret_key.to_string(),
    };
    self.heap.lock().unwrap().push(entry);
}

// Efficient expiration check - O(1) to peek, O(log n) to pop
pub fn get_expired(&self) -> Vec<(String, String)> {
    let now = Instant::now();
    let mut expired = Vec::new();
    let mut heap = self.heap.lock().unwrap();

    // Pop all expired entries (they're at the top due to min-heap)
    while let Some(entry) = heap.peek() {
        if entry.expires_at <= now {
            if let Some(entry) = heap.pop() {
                expired.push((entry.entity, entry.secret_key));
            }
        } else {
            break;  // No more expired entries
        }
    }

    expired
}
}

TTL Persistence

TTL grants survive vault restarts via TensorStore persistence:

#![allow(unused)]
fn main() {
// From ttl.rs
const TTL_STORAGE_KEY: &str = "_vault_ttl_grants";

#[derive(Serialize, Deserialize)]
pub struct PersistedGrant {
    pub expires_at_ms: i64,  // Unix timestamp
    pub entity: String,
    pub secret_key: String,
}

pub fn persist(&self, store: &TensorStore) -> Result<()> {
    let grants: Vec<PersistedGrant> = self.heap.lock().unwrap()
        .iter()
        .map(|e| PersistedGrant {
            expires_at_ms: instant_to_unix_ms(e.expires_at),
            entity: e.entity.clone(),
            secret_key: e.secret_key.clone(),
        })
        .collect();

    let data = serde_json::to_vec(&grants)?;
    store.put(TTL_STORAGE_KEY, tensor_with_bytes(data))?;
    Ok(())
}

pub fn load(store: &TensorStore) -> Result<Self> {
    let tracker = Self::new();
    let grants: Vec<PersistedGrant> = load_from_store(store)?;

    for grant in grants {
        // Skip already expired grants
        if !grant.is_expired() {
            tracker.add_with_expiration(
                &grant.entity,
                &grant.secret_key,
                unix_ms_to_instant(grant.expires_at_ms),
            );
        }
    }

    Ok(tracker)
}
}

Cleanup Strategy

Expired grants are cleaned up opportunistically during get() operations:

#![allow(unused)]
fn main() {
// From lib.rs
pub fn get(&self, requester: &str, key: &str) -> Result<String> {
    // Opportunistic cleanup of expired grants
    self.cleanup_expired_grants();

    // ... rest of get operation
}

pub fn cleanup_expired_grants(&self) -> usize {
    let expired = self.ttl_tracker.get_expired();
    let mut revoked = 0;

    for (entity, key) in expired {
        let secret_node = self.secret_node_key(&key);

        // Delete the VAULT_ACCESS_* edge
        if let Ok(edges) = self.graph.get_entity_outgoing(&entity) {
            for edge_key in edges {
                if let Ok((_, to, edge_type, _)) = self.graph.get_entity_edge(&edge_key) {
                    if to == secret_node && edge_type.starts_with("VAULT_ACCESS") {
                        if self.graph.delete_entity_edge(&edge_key).is_ok() {
                            revoked += 1;
                        }
                    }
                }
            }
        }
    }

    revoked
}
}

Audit Logging

Audit Entry Storage

#![allow(unused)]
fn main() {
// From audit.rs
const AUDIT_PREFIX: &str = "_va:";
static AUDIT_COUNTER: AtomicU64 = AtomicU64::new(0);

pub fn record(&self, entity: &str, secret_key: &str, operation: &AuditOperation) {
    let timestamp = now_millis();
    let counter = AUDIT_COUNTER.fetch_add(1, Ordering::SeqCst);
    let key = format!("{AUDIT_PREFIX}{timestamp}:{counter}");

    let mut tensor = TensorData::new();
    tensor.set("_entity", entity);
    tensor.set("_secret", secret_key);  // Already obfuscated by caller
    tensor.set("_op", operation.as_str());
    tensor.set("_ts", timestamp);

    // Additional fields for grant/revoke
    match operation {
        AuditOperation::Grant { to, permission } => {
            tensor.set("_target", to);
            tensor.set("_permission", permission);
        },
        AuditOperation::Revoke { from } => {
            tensor.set("_target", from);
        },
        _ => {},
    }

    // Best effort - audit failures don't block operations
    let _ = self.store.put(&key, tensor);
}
}

Audit Query Methods

Note: Secret keys are obfuscated in audit logs to prevent leaking plaintext names.

Usage Examples

Basic Operations

#![allow(unused)]
fn main() {
use tensor_vault::{Vault, VaultConfig, Permission};
use graph_engine::GraphEngine;
use tensor_store::TensorStore;
use std::sync::Arc;

// Initialize vault
let graph = Arc::new(GraphEngine::new());
let store = TensorStore::new();
let vault = Vault::new(b"master_password", graph, store, VaultConfig::default())?;

// Store a secret (root only)
vault.set(Vault::ROOT, "api_key", "sk-secret123")?;

// Grant access with permission level
vault.grant_with_permission(Vault::ROOT, "user:alice", "api_key", Permission::Read)?;

// Retrieve secret
let value = vault.get("user:alice", "api_key")?;

// Revoke access
vault.revoke(Vault::ROOT, "user:alice", "api_key")?;
}

Permission-Based Access

#![allow(unused)]
fn main() {
// Grant different permission levels
vault.grant_with_permission(Vault::ROOT, "user:reader", "secret", Permission::Read)?;
vault.grant_with_permission(Vault::ROOT, "user:writer", "secret", Permission::Write)?;
vault.grant_with_permission(Vault::ROOT, "user:admin", "secret", Permission::Admin)?;

// Reader can only get/list
vault.get("user:reader", "secret")?;  // OK
vault.set("user:reader", "secret", "new")?;  // InsufficientPermission

// Writer can update
vault.rotate("user:writer", "secret", "new_value")?;  // OK
vault.delete("user:writer", "secret")?;  // InsufficientPermission

// Admin can do everything
vault.grant_with_permission("user:admin", "user:new", "secret", Permission::Read)?;  // OK
vault.delete("user:admin", "secret")?;  // OK
}

TTL Grants

#![allow(unused)]
fn main() {
use std::time::Duration;

// Grant temporary access (1 hour)
vault.grant_with_ttl(
    Vault::ROOT,
    "agent:temp",
    "api_key",
    Permission::Read,
    Duration::from_secs(3600),
)?;

// Access works during TTL
vault.get("agent:temp", "api_key")?;  // OK

// After 1 hour, access is automatically revoked
// (cleanup happens opportunistically on next vault operation)
}

Namespace Isolation

#![allow(unused)]
fn main() {
// Create namespaced vault for multi-tenant isolation
let backend = vault.namespace("team:backend", "user:alice");
let frontend = vault.namespace("team:frontend", "user:bob");

// Keys are automatically prefixed
backend.set("db_password", "secret1")?;   // Stored as "team:backend:db_password"
frontend.set("api_key", "secret2")?;      // Stored as "team:frontend:api_key"

// Cross-namespace access blocked
frontend.get("db_password")?;  // AccessDenied
}

Secret Versioning

#![allow(unused)]
fn main() {
// Each set/rotate creates a new version
vault.set(Vault::ROOT, "api_key", "v1")?;
vault.rotate(Vault::ROOT, "api_key", "v2")?;
vault.rotate(Vault::ROOT, "api_key", "v3")?;

// Get version info
let version = vault.current_version(Vault::ROOT, "api_key")?;  // 3
let versions = vault.list_versions(Vault::ROOT, "api_key")?;
// [VersionInfo { version: 1, created_at: ... }, ...]

// Get specific version
let old_value = vault.get_version(Vault::ROOT, "api_key", 1)?;  // "v1"

// Rollback (creates new version with old content)
vault.rollback(Vault::ROOT, "api_key", 1)?;
vault.get(Vault::ROOT, "api_key")?;  // "v1"
vault.current_version(Vault::ROOT, "api_key")?;  // 4 (rollback creates new version)
}

Audit Queries

#![allow(unused)]
fn main() {
// Query by secret
let entries = vault.audit_log("api_key");

// Query by entity
let alice_actions = vault.audit_by_entity("user:alice");

// Query by time
let recent = vault.audit_since(timestamp_millis);
let last_10 = vault.audit_recent(10);

// Audit entries include operation details
for entry in entries {
    match &entry.operation {
        AuditOperation::Grant { to, permission } => {
            println!("Granted {} to {} at {}", permission, to, entry.timestamp);
        },
        AuditOperation::Get => {
            println!("{} read secret at {}", entry.entity, entry.timestamp);
        },
        _ => {},
    }
}
}

Scoped Vault

#![allow(unused)]
fn main() {
// Create a scoped view for a specific entity
let alice = vault.scope("user:alice");

// All operations use alice as the requester
alice.get("api_key")?;  // Same as vault.get("user:alice", "api_key")
alice.list("*")?;       // Same as vault.list("user:alice", "*")
}

Configuration Options

VaultConfig

RateLimitConfig

Environment Variables

Shell Commands

VAULT INIT                              Initialize vault from NEUMANN_VAULT_KEY
VAULT IDENTITY 'node:alice'             Set current identity
VAULT NAMESPACE 'team:backend'          Set current namespace

VAULT SET 'api_key' 'sk-123'            Store encrypted secret
VAULT GET 'api_key'                     Retrieve secret
VAULT GET 'api_key' VERSION 2           Get specific version
VAULT DELETE 'api_key'                  Delete secret
VAULT LIST 'prefix:*'                   List accessible secrets
VAULT ROTATE 'api_key' 'new'            Rotate secret value
VAULT VERSIONS 'api_key'                List version history
VAULT ROLLBACK 'api_key' VERSION 2      Rollback to version

VAULT GRANT 'user:bob' ON 'api_key'              Grant admin access
VAULT GRANT 'user:bob' ON 'api_key' READ         Grant read-only access
VAULT GRANT 'user:bob' ON 'api_key' WRITE        Grant write access
VAULT GRANT 'user:bob' ON 'api_key' TTL 3600     Grant with 1-hour expiry
VAULT REVOKE 'user:bob' ON 'api_key'             Revoke access

VAULT AUDIT 'api_key'                   View audit log for secret
VAULT AUDIT BY 'user:alice'             View audit log for entity
VAULT AUDIT RECENT 10                   View last 10 operations

Security Considerations

Best Practices

1. Use strong master passwords: At least 128 bits of entropy
2. Rotate secrets regularly: Use rotate() to maintain version history
3. Grant minimal permissions: Use Read when Write/Admin not needed
4. Use TTL grants for temporary access: Prevents forgotten grants
5. Enable rate limiting in production: Prevents brute-force attacks
6. Use namespaces for multi-tenant: Enforces isolation
7. Review audit logs: Monitor for suspicious access patterns

Edge Cases and Gotchas

Threat Model

Performance

Related Modules

Dependencies

Tensor Cache Architecture

Semantic caching for LLM responses with cost tracking and background eviction. Module 10 of Neumann.

The tensor_cache module provides multi-layer caching optimized for LLM workloads. It combines O(1) exact hash lookups with O(log n) semantic similarity search via HNSW indices. All cache entries are stored as TensorData in a shared TensorStore, following the tensor-native paradigm used by tensor_vault and tensor_blob.

Design Principles

Key Types

Core Types

Configuration Types