sochdb 0.5.5
SochDB is an AI-native database with token-optimized output, O(|path|) lookups, built-in vector search, and durable transactions.
License: Apache-2.0
Requires Python: >=3.9
Keywords: database, llm, ai, vector-search, embedded, key-value, sochdb, context-retrieval, transactions

# SochDB Python SDK

**Dual-mode architecture: Embedded (FFI) + Server (gRPC/IPC)**  
Choose the deployment mode that fits your needs.

---

## Installation

```bash
pip install sochdb
```

Or from source:
```bash
cd sochdb-python-sdk
pip install -e .
```

> **Development builds (contributors only):** If you're modifying the Rust core and need to rebuild the native FFI libraries, run `python build_native.py --libs` before installing. This requires the Rust toolchain (`cargo`). Regular users don't need this — pre-built native libraries are bundled in the wheel.

---

## Architecture: Flexible Deployment

```
┌─────────────────────────────────────────────────────────────┐
│                    DEPLOYMENT OPTIONS                        │
├─────────────────────────────────────────────────────────────┤
│                                                               │
│  1. EMBEDDED MODE (FFI)          2. SERVER MODE (gRPC)      │
│  ┌─────────────────────┐         ┌─────────────────────┐   │
│  │   Python App        │         │   Python App        │   │
│  │   ├─ Database.open()│         │   ├─ SochDBClient() │   │
│  │   └─ Direct FFI     │         │   └─ gRPC calls     │   │
│  │         │           │         │         │           │   │
│  │         ▼           │         │         ▼           │   │
│  │   libsochdb_storage │         │   sochdb-grpc       │   │
│  │   (Rust native)     │         │   (Rust server)     │   │
│  └─────────────────────┘         └─────────────────────┘   │
│                                                               │
│  ✅ No server needed               ✅ Multi-language          │
│  ✅ Local files                    ✅ Centralized logic      │
│  ✅ Simple deployment              ✅ Production scale       │
└─────────────────────────────────────────────────────────────┘
```

### When to Use Each Mode

**Embedded Mode (FFI):**
- ✅ Local development and testing
- ✅ Jupyter notebooks and data science
- ✅ Single-process applications
- ✅ Edge deployments without network
- ✅ No server setup required

**Embedded Concurrent Mode:**
- ✅ Web applications (Flask, FastAPI, Django)
- ✅ Multi-process workers (Gunicorn, uWSGI)
- ✅ Hot reloading development servers
- ✅ Multi-reader, single-writer architecture
- ✅ Lock-free reads (~100ns latency)

**Server Mode (gRPC):**
- ✅ Production deployments
- ✅ Multi-language teams (Python, Node.js, Go)
- ✅ Distributed systems
- ✅ Centralized business logic
- ✅ Horizontal scaling

---

# SochDB Python SDK Documentation

LLM-Optimized Embedded Database with Native Vector Search

---


## Table of Contents

1. [Quick Start](#1-quick-start)
2. [Installation](#2-installation)
3. [Features](#3-features)
   - [Namespace API](#namespace-api---multi-tenant-isolation)
   - [Priority Queue API](#priority-queue-api---task-processing)
4. [Architecture Overview](#4-architecture-overview)
5. [Core Key-Value Operations](#5-core-key-value-operations)
6. [Transactions (ACID with SSI)](#6-transactions-acid-with-ssi)
7. [Query Builder](#7-query-builder)
8. [Prefix Scanning](#8-prefix-scanning)
9. [SQL Operations](#9-sql-operations)
10. [Table Management & Index Policies](#10-table-management--index-policies)
11. [Namespaces & Collections](#11-namespaces--collections)
12. [Priority Queues](#12-priority-queues)
13. [Vector Search](#13-vector-search)
14. [Hybrid Search (Vector + BM25)](#14-hybrid-search-vector--bm25)
15. [Graph Operations](#15-graph-operations)
16. [Temporal Graph (Time-Travel)](#16-temporal-graph-time-travel)
17. [Semantic Cache](#17-semantic-cache)
18. [Memory System](#18-memory-system)
19. [Session Management](#19-session-management)
20. [Context Query Builder (LLM Optimization)](#20-context-query-builder-llm-optimization)
21. [Atomic Multi-Index Writes](#21-atomic-multi-index-writes)
22. [Recovery & WAL Management](#22-recovery--wal-management)
23. [Checkpoints & Snapshots](#23-checkpoints--snapshots)
24. [Compression & Storage](#24-compression--storage)
25. [Statistics & Monitoring](#25-statistics--monitoring)
26. [Distributed Tracing](#26-distributed-tracing)
27. [Workflow & Run Tracking](#27-workflow--run-tracking)
28. [Server Mode (gRPC Client)](#28-server-mode-grpc-client)
29. [IPC Client (Unix Sockets)](#29-ipc-client-unix-sockets)
30. [Standalone VectorIndex](#30-standalone-vectorindex)
31. [Vector Utilities](#31-vector-utilities)
32. [Data Formats (TOON/JSON/Columnar)](#32-data-formats-toonjsoncolumnar)
33. [Policy Service](#33-policy-service)
34. [MCP (Model Context Protocol)](#34-mcp-model-context-protocol)
35. [Configuration Reference](#35-configuration-reference)
36. [Error Handling](#36-error-handling)
37. [Async Support](#37-async-support)
38. [Building & Development](#38-building--development)
39. [Complete Examples](#39-complete-examples)
40. [Migration Guide](#40-migration-guide)

---


## 1. Quick Start

### Concurrent Embedded Mode
db = Database.open_concurrent("./app_data")

# Reads are lock-free and can run in parallel (~100ns)
value = db.get(b"user:123")

# Writes are automatically coordinated (~60µs amortized)
db.put(b"user:123", b'{"name": "Alice"}')

# Check if concurrent mode is active
print(f"Concurrent mode: {db.is_concurrent}")  # True
```

### Flask Example

```python
from flask import Flask
from sochdb import Database

app = Flask(__name__)
db = Database.open_concurrent("./flask_db")

@app.route("/user/<user_id>")
def get_user(user_id):
    # Multiple concurrent requests can read simultaneously
    data = db.get(f"user:{user_id}".encode())
    return data or "Not found"

@app.route("/user/<user_id>", methods=["POST"])
def update_user(user_id):
    # Writes are serialized automatically
    db.put(f"user:{user_id}".encode(), request.data)
    return "OK"
```

### Performance

| Operation | Standard Mode | Concurrent Mode |
|-----------|---------------|-----------------|
| Read (single process) | ~100ns | ~100ns |
| Read (multi-process) | **Blocked** ❌ | ~100ns ✅ |
| Write | ~5ms (fsync) | ~60µs (amortized) |
| Max concurrent readers | 1 | 1024 |

### Gunicorn Deployment

```bash
# Install Gunicorn
pip install gunicorn

# Run with 4 worker processes (all can access same DB concurrently)
gunicorn -w 4 -b 0.0.0.0:8000 app:app

# Workers automatically share the database in concurrent mode
```

### uWSGI Deployment

```bash
# Install uWSGI
pip install uwsgi

# Run with 4 processes
uwsgi --http :8000 --wsgi-file app.py --callable app --processes 4
```

### Systemd Service Example

```ini
# /etc/systemd/system/myapp.service
[Unit]
Description=MyApp with SochDB
After=network.target

[Service]
Type=notify
User=appuser
WorkingDirectory=/opt/myapp
ExecStart=/opt/myapp/venv/bin/gunicorn -w 4 -b 0.0.0.0:8000 app:app
Restart=always

[Install]
WantedBy=multi-user.target
```

```bash
# Enable and start service
sudo systemctl enable myapp
sudo systemctl start myapp
sudo systemctl status myapp
```

### Docker Compose Example

```yaml
version: '3.8'
services:
  app:
    build: .
    environment:
      - WORKERS=4
    volumes:
      - ./data:/app/data  # Shared database volume
    ports:
      - "8000:8000"
    command: gunicorn -w 4 -b 0.0.0.0:8000 app:app
```

---

## System Requirements

### For Concurrent Mode

- **SochDB Core**: Latest version
- **Python**: 3.9+ (3.11+ recommended)
- **Native Library**: `libsochdb_storage.{dylib,so}`
- **FFI**: ctypes (built-in to Python)

**Operating Systems:**
- ✅ Linux (Ubuntu 20.04+, RHEL 8+)
- ✅ macOS (10.15+, both Intel and Apple Silicon)
- ⚠️  Windows (requires native builds)

**File Descriptors:**
- Default limit: 1024 (sufficient for most workloads)
- For high concurrency with Gunicorn: `ulimit -n 4096`

**Memory:**
- Standard mode: ~50MB base + data
- Concurrent mode: +4KB per concurrent reader slot (1024 slots = ~4MB overhead)
- Gunicorn: Each worker has independent memory

---

## Troubleshooting

### "Database is locked" Error (Standard Mode)

```
OperationalError: database is locked
```

**Solution**: Use concurrent mode for multi-process access:

```python
# ❌ Standard mode - Gunicorn workers will conflict
db = Database.open("./data.db")

# ✅ Concurrent mode - all workers can access
db = Database.open_concurrent("./data.db")
```

### Library Not Found Error

```
OSError: libsochdb_storage.dylib not found
```

**macOS**:
```bash
# Build and install library
cd /path/to/sochdb
cargo build --release
sudo cp target/release/libsochdb_storage.dylib /usr/local/lib/
```

**Linux**:
```bash
cd /path/to/sochdb
cargo build --release
sudo cp target/release/libsochdb_storage.so /usr/local/lib/
sudo ldconfig
```

**Development Mode** (no install):
```bash
export DYLD_LIBRARY_PATH=/path/to/sochdb/target/release  # macOS
export LD_LIBRARY_PATH=/path/to/sochdb/target/release    # Linux
```

### Gunicorn Worker Issues

**Symptom**: Workers crash with "database locked"

**Solution 1** - Ensure concurrent mode is used:
```python
# app.py
import os
from sochdb import Database

# Use environment variable to control mode
USE_CONCURRENT = os.getenv('USE_CONCURRENT_MODE', 'true').lower() == 'true'

if USE_CONCURRENT:
    db = Database.open_concurrent('./db')
else:
    db = Database.open('./db')

print(f"Concurrent mode: {db.is_concurrent}")  # Should be True
```

```bash
# Start with concurrent mode enabled
USE_CONCURRENT_MODE=true gunicorn -w 4 -b 0.0.0.0:8000 app:app
```

**Solution 2** - Check preload settings:
```bash
# Don't use --preload with concurrent mode
# ❌ This will cause issues:
gunicorn --preload -w 4 app:app

# ✅ Let each worker open the database:
gunicorn -w 4 app:app
```

### FastAPI with Uvicorn Workers

**Symptom**: `RuntimeError: Concurrent mode requires multi-process access`

**Solution**: Use Uvicorn workers correctly:
```bash
# ❌ Single worker (async) - doesn't need concurrent mode
uvicorn app:app --workers 1

# ✅ Multiple workers - needs concurrent mode
uvicorn app:app --workers 4
```

```python
# main.py
from fastapi import FastAPI
from sochdb import Database
import multiprocessing

app = FastAPI()

# Detect if running in multi-worker mode
workers = multiprocessing.cpu_count()
if workers > 1:
    db = Database.open_concurrent("./db")
else:
    db = Database.open("./db")
```

### Performance Issues

**Symptom**: Concurrent reads slower than expected

**Check 1** - Verify concurrent mode is active:
```python
import logging
logging.basicConfig(level=logging.INFO)

db = Database.open_concurrent("./db")
if not db.is_concurrent:
    logging.error("Database is not in concurrent mode!")
    raise RuntimeError("Expected concurrent mode")
logging.info(f"Concurrent mode active: {db.is_concurrent}")
```

**Check 2** - Monitor worker processes:
```bash
# Watch Gunicorn workers
watch -n 1 'ps aux | grep gunicorn'

# Monitor file descriptors
lsof | grep libsochdb_storage
```

**Check 3** - Batch writes:
```python
# ❌ Slow - individual writes with fsync
for item in items:
    db.put(key, value)

# ✅ Fast - batch in transaction
tx = db.begin_txn()
for item in items:
    tx.put(key, value)
tx.commit()  # Single fsync for entire batch
```

---

## API Reference

---

## 1. Quick Start

```python
from sochdb import Database

# Open (or create) a database
db = Database.open("./my_database")

# Store and retrieve data
db.put(b"hello", b"world")
value = db.get(b"hello")  # b"world"

# Use transactions for atomic operations
with db.transaction() as txn:
    txn.put(b"key1", b"value1")
    txn.put(b"key2", b"value2")
    # Auto-commits on success, auto-rollbacks on exception

# Clean up
db.delete(b"hello")
db.close()
```

**30-Second Overview:**
- **Key-Value**: Fast reads/writes with `get`/`put`/`delete`
- **Transactions**: ACID with SSI isolation
- **Vector Search**: HNSW-based semantic search
- **Hybrid Search**: Combine vectors with BM25 keyword search
- **Graph**: Build and traverse knowledge graphs
- **LLM-Optimized**: TOON format uses 40-60% fewer tokens than JSON

---

## 2. Installation

```bash
pip install sochdb
```

**Platform Support:**
| Platform | Architecture | Status |
|----------|--------------|--------|
| Linux | x86_64, aarch64 | ✅ Full support |
| macOS | x86_64, arm64 | ✅ Full support |
| Windows | x86_64 | ✅ Full support |

**Optional Dependencies:**
```bash
# For async support
pip install sochdb[async]

# For server mode
pip install sochdb[grpc]

# Everything
pip install sochdb[all]
```

---

## 3. Features

### Namespace API — Multi-Tenant Isolation

Organize data into logical namespaces with per-tenant collections, vector search, and metadata filtering.

```python
ns = db.create_namespace("tenant_123", display_name="Acme Corp", labels={"tier": "premium"})
coll = ns.create_collection("documents", dimension=384, distance_metric=DistanceMetric.COSINE)
coll.add("doc1", vector=[0.1]*384, metadata={"type": "report"})
results = coll.search(query_vector=[0.1]*384, top_k=5)
```

See [§11 Namespaces & Collections](#11-namespaces--collections) for the full API.

### Priority Queue API — Task Processing

First-class priority queue with atomic claim protocol, visibility timeouts, and at-least-once delivery.

```python
queue = PriorityQueue.from_database(db, "tasks")
task_id = queue.enqueue(priority=10, payload=b"high priority")
task = queue.dequeue(worker_id="worker-1")
queue.ack(task.task_id)   # Mark complete
```

See [§12 Priority Queues](#12-priority-queues) for the full API.

---

## 4. Architecture Overview

### Engine Internals

| Component | Status | Description |
|-----------|--------|-------------|
| **Cost-based optimizer** | ✅ Production-ready | Full cost model with cardinality estimation (HyperLogLog + histograms), join-order DP, token-budget planning, and plan caching with configurable TTL |
| **Adaptive group commit** | ✅ Implemented | Little's Law-based batch sizing with EMA arrival-rate tracking for automatic write throughput optimization |
| **WAL compaction** | ⚠️ Partially implemented | Manual `checkpoint()` + `truncate_wal()` works end-to-end; automatic background compaction planned |
| **HNSW vector index** | ✅ Production-ready | Lock-free concurrent reads, batch insert, quantization support |
| **SSI transactions** | ✅ Production-ready | Serializable Snapshot Isolation with conflict detection |

SochDB supports two deployment modes:

### Embedded Mode (Default)

Direct Rust bindings via FFI. No server required.

```python
from sochdb import Database

with Database.open("./mydb") as db:
    db.put(b"key", b"value")
    value = db.get(b"key")
```

**Best for:** Local development, notebooks, single-process applications.

### Server Mode (gRPC)

Thin client connecting to `sochdb-grpc` server.

```python
from sochdb import SochDBClient

client = SochDBClient("localhost:50051")
client.put(b"key", b"value", namespace="default")
value = client.get(b"key", namespace="default")
```

**Best for:** Production, multi-process, distributed systems.

### Feature Comparison

| Feature | Embedded | Server |
|---------|----------|--------|
| Setup | `pip install` only | Server + client |
| Performance | Fastest (in-process) | Network overhead |
| Multi-process | ❌ | ✅ |
| Horizontal scaling | ❌ | ✅ |
| Vector search | ✅ | ✅ |
| Graph operations | ✅ | ✅ |
| Semantic cache | ✅ | ✅ |
| Context service | Limited | ✅ Full |
| MCP integration | ❌ | ✅ |

```
┌─────────────────────────────────────────────────────────────┐
│                    DEPLOYMENT OPTIONS                        │
├─────────────────────────────────────────────────────────────┤
│  EMBEDDED MODE (FFI)             SERVER MODE (gRPC)         │
│  ┌─────────────────────┐         ┌─────────────────────┐   │
│  │   Python App        │         │   Python App        │   │
│  │   ├─ Database.open()│         │   ├─ SochDBClient() │   │
│  │   └─ Direct FFI     │         │   └─ gRPC calls     │   │
│  │         │           │         │         │           │   │
│  │         ▼           │         │         ▼           │   │
│  │   libsochdb_storage │         │   sochdb-grpc       │   │
│  │   (Rust native)     │         │   (Rust server)     │   │
│  └─────────────────────┘         └─────────────────────┘   │
│                                                              │
│  ✅ No server needed             ✅ Multi-language          │
│  ✅ Local files                  ✅ Centralized logic       │
│  ✅ Simple deployment            ✅ Production scale        │
└─────────────────────────────────────────────────────────────┘
```

---

## 5. Core Key-Value Operations

All keys and values are **bytes**.

### Basic Operations

```python
from sochdb import Database

db = Database.open("./my_db")

# Store data
db.put(b"user:1", b"Alice")
db.put(b"user:2", b"Bob")

# Retrieve data
user = db.get(b"user:1")  # Returns b"Alice" or None

# Check existence
exists = db.get(b"user:1") is not None  # True

# Delete data
db.delete(b"user:1")

db.close()
```

### Path-Based Keys (Hierarchical)

Organize data hierarchically with path-based access:

```python
# Store with path (strings auto-converted to bytes internally)
db.put_path("users/alice/name", b"Alice Smith")
db.put_path("users/alice/email", b"alice@example.com")
db.put_path("users/bob/name", b"Bob Jones")

# Retrieve by path
name = db.get_path("users/alice/name")  # b"Alice Smith"

# Delete by path
db.delete_path("users/alice/email")

# Scan by path prefix
results = list(db.scan_prefix(b"users/"))  # All keys under users/
```

### With TTL (Time-To-Live)

```python
# Store with expiration (seconds)
db.put(b"session:abc123", b"user_data", ttl_seconds=3600)  # Expires in 1 hour

# TTL of 0 means no expiration
db.put(b"permanent_key", b"value", ttl_seconds=0)
```

### Batch Operations

```python
# Use a transaction for efficient batch writes
with db.transaction() as txn:
    txn.put(b"key1", b"value1")
    txn.put(b"key2", b"value2")
    txn.put(b"key3", b"value3")

# Individual reads
v1 = db.get(b"key1")  # b"value1" or None
v2 = db.get(b"key2")

# Batch delete via transaction
with db.transaction() as txn:
    txn.delete(b"key1")
    txn.delete(b"key2")
    txn.delete(b"key3")
```

### Context Manager

```python
with Database.open("./my_db") as db:
    db.put(b"key", b"value")
    # Automatically closes when exiting
```

---

## 6. Transactions (ACID with SSI)

SochDB provides full ACID transactions with **Serializable Snapshot Isolation (SSI)**.

### Context Manager Pattern (Recommended)

```python
# Auto-commits on success, auto-rollbacks on exception
with db.transaction() as txn:
    txn.put(b"accounts/alice", b"1000")
    txn.put(b"accounts/bob", b"500")
    
    # Read within transaction sees your writes
    balance = txn.get(b"accounts/alice")  # b"1000"
    
    # If exception occurs, rolls back automatically
```

### Closure Pattern (Rust-Style)

```python
# Using with_transaction for automatic commit/rollback
def transfer_funds(txn):
    alice = int(txn.get(b"accounts/alice") or b"0")
    bob = int(txn.get(b"accounts/bob") or b"0")
    
    txn.put(b"accounts/alice", str(alice - 100).encode())
    txn.put(b"accounts/bob", str(bob + 100).encode())
    
    return "Transfer complete"

result = db.with_transaction(transfer_funds)
```

### Manual Transaction Control

```python
txn = db.begin_transaction()
try:
    txn.put(b"key1", b"value1")
    txn.put(b"key2", b"value2")
    
    commit_ts = txn.commit()  # Returns HLC timestamp
    print(f"Committed at: {commit_ts}")
except Exception as e:
    txn.abort()
    raise
```

### Transaction Properties

```python
txn = db.transaction()
print(f"Transaction ID: {txn.id}")      # Unique identifier
print(f"Start timestamp: {txn.start_ts}")  # HLC start time
print(f"Isolation: {txn.isolation}")    # "serializable"
```

### SSI Conflict Handling

```python
from sochdb import TransactionConflictError

MAX_RETRIES = 3

for attempt in range(MAX_RETRIES):
    try:
        with db.transaction() as txn:
            # Read and modify
            value = int(txn.get(b"counter") or b"0")
            txn.put(b"counter", str(value + 1).encode())
        break  # Success
    except TransactionConflictError:
        if attempt == MAX_RETRIES - 1:
            raise
        # Retry on conflict
        continue
```

### All Transaction Operations

```python
with db.transaction() as txn:
    # Key-value
    txn.put(key, value)
    txn.get(key)
    txn.delete(key)
    
    # Path-based
    txn.put_path(path, value)
    txn.get_path(path)
    txn.delete_path(path)
    
    # Scanning
    for k, v in txn.scan_prefix(b"prefix/"):
        print(k, v)
    
    # SQL (within transaction isolation)
    result = txn.execute("SELECT * FROM users WHERE id = 1")
```

### Isolation Levels

```python
from sochdb import IsolationLevel

# Default: Serializable (strongest)
with db.transaction(isolation=IsolationLevel.SERIALIZABLE) as txn:
    pass

# Snapshot isolation (faster, allows some anomalies)
with db.transaction(isolation=IsolationLevel.SNAPSHOT) as txn:
    pass

# Read committed (fastest, least isolation)
with db.transaction(isolation=IsolationLevel.READ_COMMITTED) as txn:
    pass
```

---

## 7. Query Builder

Fluent API for building efficient queries with predicate pushdown.

### Basic Query

```python
# Query with prefix and limit
results = db.query("users/")
    .limit(10)
    .execute()

for key, value in results:
    print(f"{key.decode()}: {value.decode()}")
```

### Filtered Query

```python
from sochdb import CompareOp

# Query with filters
results = db.query("orders/")
    .where("status", CompareOp.EQ, "pending")
    .where("amount", CompareOp.GT, 100)
    .order_by("created_at", descending=True)
    .limit(50)
    .offset(10)
    .execute()
```

### Column Selection

```python
# Select specific fields only
results = db.query("users/")
    .select(["name", "email"])  # Only fetch these columns
    .where("active", CompareOp.EQ, True)
    .execute()
```

### Aggregate Queries

```python
# Count
count = db.query("orders/")
    .where("status", CompareOp.EQ, "completed")
    .count()

# Sum (for numeric columns)
total = db.query("orders/")
    .sum("amount")

# Group by
results = db.query("orders/")
    .select(["status", "COUNT(*)", "SUM(amount)"])
    .group_by("status")
    .execute()
```

### Query in Transaction

```python
with db.transaction() as txn:
    results = txn.query("users/")
        .where("role", CompareOp.EQ, "admin")
        .execute()
```

---

## 8. Prefix Scanning

Iterate over keys with common prefixes efficiently.

### Safe Prefix Scan (Recommended)

```python
# Requires minimum 2-byte prefix (prevents accidental full scans)
for key, value in db.scan_prefix(b"users/"):
    print(f"{key.decode()}: {value.decode()}")

# Raises ValueError if prefix < 2 bytes
```

### Unchecked Prefix Scan

```python
# For internal operations needing empty/short prefixes
# WARNING: Can cause expensive full-database scans
for key, value in db.scan_prefix_unchecked(b""):
    print(f"All keys: {key}")
```

### Batched Scanning (1000x Faster)

```python
# Fetches 1000 results per FFI call instead of 1
# Performance: 10,000 results = 10 FFI calls vs 10,000 calls

for key, value in db.scan_batched(b"prefix/", batch_size=1000):
    process(key, value)
```

### Reverse Scan

```python
# Scan in reverse order (newest first)
for key, value in db.scan_prefix(b"logs/", reverse=True):
    print(key, value)
```

### Range Scan

```python
# Scan within a specific range
for key, value in db.scan_range(b"users/a", b"users/m"):
    print(key, value)  # All users from "a" to "m"
```

### Streaming Large Results

```python
# For very large result sets, use streaming to avoid memory issues
for batch in db.scan_stream(b"logs/", batch_size=10000):
    for key, value in batch:
        process(key, value)
    # Memory is freed after processing each batch
```

---

## 9. SQL Operations

Execute SQL queries for familiar relational patterns.

### Creating Tables

```python
db.execute_sql("""
    CREATE TABLE users (
        id INTEGER PRIMARY KEY,
        name TEXT NOT NULL,
        email TEXT UNIQUE,
        age INTEGER,
        created_at TEXT DEFAULT CURRENT_TIMESTAMP
    )
""")

db.execute_sql("""
    CREATE TABLE posts (
        id INTEGER PRIMARY KEY,
        user_id INTEGER REFERENCES users(id),
        title TEXT NOT NULL,
        content TEXT,
        likes INTEGER DEFAULT 0
    )
""")
```

### CRUD Operations

```python
# Insert
db.execute_sql("""
    INSERT INTO users (id, name, email, age) 
    VALUES (1, 'Alice', 'alice@example.com', 30)
""")

# Insert with parameters (prevents SQL injection)
db.execute_sql(
    "INSERT INTO users (id, name, email, age) VALUES (?, ?, ?, ?)",
    params=[2, "Bob", "bob@example.com", 25]
)

# Select
result = db.execute_sql("SELECT * FROM users WHERE age > 25")
for row in result.rows:
    print(row)  # {'id': 1, 'name': 'Alice', ...}

# Update
db.execute_sql("UPDATE users SET email = 'alice.new@example.com' WHERE id = 1")

# Delete
db.execute_sql("DELETE FROM users WHERE id = 2")
```

### Upsert (Insert or Update)

```python
# Insert or update on conflict
db.execute_sql("""
    INSERT INTO users (id, name, email) VALUES (1, 'Alice', 'alice@example.com')
    ON CONFLICT (id) DO UPDATE SET 
        name = excluded.name,
        email = excluded.email
""")
```

### Query Results

```python
from sochdb import SQLQueryResult

result = db.execute_sql("SELECT id, name FROM users")

print(f"Columns: {result.columns}")      # ['id', 'name']
print(f"Row count: {len(result.rows)}")
print(f"Execution time: {result.execution_time_ms}ms")

for row in result.rows:
    print(f"ID: {row['id']}, Name: {row['name']}")

# Convert to different formats
df = result.to_dataframe()  # pandas DataFrame
json_data = result.to_json()
```

### Index Management

```python
# Create index
db.execute_sql("CREATE INDEX idx_users_email ON users(email)")

# Create unique index
db.execute_sql("CREATE UNIQUE INDEX idx_users_email ON users(email)")

# Drop index
db.execute_sql("DROP INDEX IF EXISTS idx_users_email")

# List indexes
indexes = db.list_indexes("users")
```

### Prepared Statements

```python
# Prepare once, execute many times
stmt = db.prepare("SELECT * FROM users WHERE age > ? AND status = ?")

# Execute with different parameters
young_active = stmt.execute([25, "active"])
old_active = stmt.execute([50, "active"])

# Close when done
stmt.close()
```

### Dialect Support

SochDB auto-detects SQL dialects:

```python
# PostgreSQL style
db.execute_sql("INSERT INTO users VALUES (1, 'Alice') ON CONFLICT DO NOTHING")

# MySQL style
db.execute_sql("INSERT IGNORE INTO users VALUES (1, 'Alice')")

# SQLite style  
db.execute_sql("INSERT OR IGNORE INTO users VALUES (1, 'Alice')")
```

---

## 10. Table Management & Index Policies

### Table Information

```python
# Get table schema
schema = db.get_table_schema("users")
print(f"Columns: {schema.columns}")
print(f"Primary key: {schema.primary_key}")
print(f"Indexes: {schema.indexes}")

# List all tables
tables = db.list_tables()

# Drop table
db.execute_sql("DROP TABLE IF EXISTS old_table")
```

### Index Policies

Configure per-table indexing strategies for optimal performance:

```python
# Policy constants
Database.INDEX_WRITE_OPTIMIZED  # 0 - O(1) insert, O(N) scan
Database.INDEX_BALANCED         # 1 - O(1) amortized insert, O(log K) scan
Database.INDEX_SCAN_OPTIMIZED   # 2 - O(log N) insert, O(log N + K) scan
Database.INDEX_APPEND_ONLY      # 3 - O(1) insert, O(N) scan (time-series)

# Set by constant
db.set_table_index_policy("logs", Database.INDEX_APPEND_ONLY)

# Set by string
db.set_table_index_policy("users", "scan_optimized")

# Get current policy
policy = db.get_table_index_policy("users")
print(f"Policy: {policy}")  # "scan_optimized"
```

### Policy Selection Guide

| Policy | Insert | Scan | Best For |
|--------|--------|------|----------|
| `write_optimized` | O(1) | O(N) | High-write ingestion |
| `balanced` | O(1) amortized | O(log K) | General use (default) |
| `scan_optimized` | O(log N) | O(log N + K) | Analytics, read-heavy |
| `append_only` | O(1) | O(N) | Time-series, logs |

---

## 11. Namespaces & Collections

Organize data into logical namespaces for tenant isolation.

### Creating Namespaces

```python
from sochdb import NamespaceConfig

# Create namespace with metadata
ns = db.create_namespace(
    name="tenant_123",
    display_name="Acme Corp",
    labels={"tier": "premium", "region": "us-east"}
)

# Simple creation
ns = db.create_namespace("tenant_456")
```

### Getting Namespaces

```python
# Get existing namespace
ns = db.namespace("tenant_123")

# Get or create (idempotent)
ns = db.get_or_create_namespace("tenant_123")

# Check if exists
exists = db.namespace_exists("tenant_123")
```

### Context Manager for Scoped Operations

```python
with db.use_namespace("tenant_123") as ns:
    # All operations automatically scoped to tenant_123
    collection = ns.collection("documents")
    ns.put("config/key", b"value")
    
    # No need to specify namespace in each call
```

### Namespace Operations

```python
# List all namespaces
namespaces = db.list_namespaces()
print(namespaces)  # ['tenant_123', 'tenant_456']

# Get namespace info
info = db.namespace_info("tenant_123")
print(f"Created: {info['created_at']}")
print(f"Labels: {info['labels']}")
print(f"Size: {info['size_bytes']}")

# Update labels
db.update_namespace("tenant_123", labels={"tier": "enterprise"})

# Delete namespace (WARNING: deletes all data in namespace)
db.delete_namespace("old_tenant", force=True)
```

### Namespace-Scoped Key-Value

```python
ns = db.namespace("tenant_123")

# Operations automatically prefixed with namespace
ns.put("users/alice", b"data")      # Actually: tenant_123/users/alice
ns.get("users/alice")
ns.delete("users/alice")

# Scan within namespace
for key, value in ns.scan("users/"):
    print(key, value)  # Keys shown without namespace prefix
```

### Cross-Namespace Operations

```python
# Copy data between namespaces
db.copy_between_namespaces(
    source_ns="tenant_123",
    target_ns="tenant_456",
    prefix="shared/"
)
```

---

## 12. Priority Queues

SochDB provides a first-class priority queue implementation with atomic claim protocol for reliable distributed task processing. The queue supports both embedded (FFI) and server (gRPC) modes.

### Features

- **Priority-based ordering**: Tasks dequeued by priority, then ready time, then sequence
- **Atomic claim protocol**: Linearizable claim semantics prevent double-delivery
- **Visibility timeout**: Automatic retry for failed workers (at-least-once delivery)
- **Delayed tasks**: Schedule tasks for future execution
- **Batch operations**: Enqueue multiple tasks atomically
- **Streaming Top-K**: O(N log K) selection for efficient ranking
- **Dual-mode support**: Works with embedded Database or gRPC SochDBClient

### Quick Start

```python
from sochdb import Database, PriorityQueue, create_queue

# Create queue from database
db = Database.open("./queue_db")
queue = PriorityQueue.from_database(db, "my_queue")

# Or use convenience function (auto-detects backend)
queue = create_queue(db, "my_queue")

# Enqueue tasks with priority
task_id1 = queue.enqueue(priority=10, payload=b"high priority task")
task_id2 = queue.enqueue(priority=1, payload=b"low priority task")

# Dequeue tasks (highest priority first)
task = queue.dequeue(worker_id="worker-1")
if task:
    print(f"Processing: {task.payload}")
    # Process task...
    queue.ack(task.task_id)  # Mark as completed
```

### Enqueue Operations

```python
# Simple enqueue with priority
task_id = queue.enqueue(
    priority=10,
    payload=b"task data",
)

# Delayed task (execute after 60 seconds)
task_id = queue.enqueue(
    priority=5,
    payload=b"delayed task",
    delay_ms=60000,
)

# Batch enqueue (atomic)
task_ids = queue.enqueue_batch([
    (10, b"task 1"),
    (20, b"task 2"),
    (15, b"task 3"),
])
```

### Dequeue and Processing

```python
# Dequeue with automatic visibility timeout
task = queue.dequeue(worker_id="worker-1")

if task:
    try:
        # Process the task
        result = process_task(task.payload)
        
        # Mark as successfully completed
        queue.ack(task.task_id)
        
    except Exception as e:
        # Return to queue for retry (optionally change priority)
        queue.nack(
            task_id=task.task_id,
            new_priority=task.priority - 1  # Lower priority on retry
        )
```

### Peek and Stats

```python
# Peek at next task without claiming
task = queue.peek()
if task:
    print(f"Next task: {task.payload}, priority: {task.priority}")

# Get queue statistics
stats = queue.stats()
print(f"Pending: {stats['pending']}")
print(f"Claimed: {stats['claimed']}")
print(f"Total: {stats['total']}")

# List all tasks (for monitoring)
tasks = queue.list_tasks(limit=100)
for task in tasks:
    print(f"Task {task.task_id}: priority={task.priority}, status={task.status}")
```

### Configuration

```python
from sochdb import PriorityQueue, QueueConfig

# Custom configuration
config = QueueConfig(
    queue_id="my_queue",
    visibility_timeout_ms=30000,  # 30 seconds
    max_retries=3,
    dead_letter_queue="dlq_queue",
)

queue = PriorityQueue.from_database(db, config=config)
```

### Worker Pattern

```python
import time

def worker_loop(worker_id: str):
    """Simple worker loop."""
    while True:
        task = queue.dequeue(worker_id=worker_id)
        
        if task:
            try:
                # Process task
                result = process_task(task.payload)
                queue.ack(task.task_id)
                print(f"✓ Completed task {task.task_id}")
                
            except Exception as e:
                print(f"✗ Failed task {task.task_id}: {e}")
                queue.nack(task.task_id)
        else:
            # No tasks available, wait
            time.sleep(1)

# Start multiple workers
from concurrent.futures import ThreadPoolExecutor

with ThreadPoolExecutor(max_workers=4) as executor:
    for i in range(4):
        executor.submit(worker_loop, f"worker-{i}")
```

### Streaming Top-K Selection

The queue includes a `StreamingTopK` utility for efficient ranking with O(N log K) complexity:

```python
from sochdb.queue import StreamingTopK

# Create top-K selector (k=10, ascending order, with key function)
topk = StreamingTopK(k=10, ascending=True, key=lambda x: x[0])

# Process items one at a time
for score, item in candidates:
    topk.push((score, item))

# Get sorted top-K results
results = topk.get_sorted()

# With custom key function
topk = StreamingTopK(
    k=5, 
    ascending=False,  # Descending (highest first)
    key=lambda x: x['score']
)

for item in items:
    topk.push(item)

top_5 = topk.get_sorted()
```

### Server Mode (gRPC)

```python
from sochdb import SochDBClient, PriorityQueue

# Connect to server
client = SochDBClient("localhost:50051")

# Create queue using gRPC backend
queue = PriorityQueue.from_client(client, "distributed_queue")

# All operations work the same way
task_id = queue.enqueue(priority=10, payload=b"server task")
task = queue.dequeue(worker_id="worker-1")
if task:
    queue.ack(task.task_id)
```

### Queue Backend Architecture

```python
from sochdb.queue import (
    QueueBackend,
    FFIQueueBackend,      # For embedded Database
    GrpcQueueBackend,     # For SochDBClient
    InMemoryQueueBackend, # For testing
)

# Use specific backend
backend = FFIQueueBackend(db)
queue = PriorityQueue.from_backend(backend, "my_queue")

# Or use factory method (auto-detects)
queue = create_queue(db, "my_queue")  # Returns FFIQueueBackend
queue = create_queue(client, "my_queue")  # Returns GrpcQueueBackend
```

### Task Model

```python
# Task structure
class Task:
    task_id: str           # Unique task identifier
    priority: int          # Task priority (higher = more important)
    ready_ts: int          # When task becomes ready (epoch millis)
    sequence: int          # Sequence number for ordering
    payload: bytes         # Task data
    claim_token: Optional[ClaimToken]  # Proof of ownership
    retry_count: int       # Number of retries
    status: str           # 'pending', 'claimed', 'completed'

# Claim token (for ack/nack operations)
class ClaimToken:
    task_id: str
    owner: str
    instance: int
    created_at: int
    expires_at: int
```

### Best Practices

**1. Choose appropriate visibility timeout:**
```python
# Short tasks (< 10s)
config = QueueConfig(visibility_timeout_ms=15000)  # 15s

# Long tasks (minutes)
config = QueueConfig(visibility_timeout_ms=300000)  # 5 minutes
```

**2. Handle idempotency:**
```python
# Tasks may be redelivered, design for idempotency
def process_task(payload):
    task_id = extract_id(payload)
    
    # Check if already processed
    if is_processed(task_id):
        return  # Skip duplicate
    
    # Process and mark as done atomically
    with db.transaction() as txn:
        do_work(txn, payload)
        mark_processed(txn, task_id)
```

**3. Use dead letter queue:**
```python
config = QueueConfig(
    queue_id="main_queue",
    max_retries=3,
    dead_letter_queue="dlq_main",
)

# Monitor DLQ for failed tasks
dlq = create_queue(db, "dlq_main")
failed_tasks = dlq.list_tasks()
```

**4. Batch operations for efficiency:**
```python
# Instead of individual enqueues
for item in items:
    queue.enqueue(priority=1, payload=item)

# Use batch enqueue
tasks = [(1, item) for item in items]
queue.enqueue_batch(tasks)
```

### Performance

Based on benchmarks with `InMemoryQueueBackend`:

- **QueueKey encode/decode**: ~411K ops/s
- **Enqueue**: ~31-83K ops/s (depends on queue size)
- **Dequeue + Ack**: ~1K ops/s (includes claim protocol)
- **StreamingTopK (n=10K, k=10)**: ~212 ops/s

### Integration with Existing Features

```python
# Combine with transactions
with db.transaction() as txn:
    # Update database
    txn.put(b"status:job1", b"queued")
    
    # Enqueue task (outside transaction for reliability)
    queue.enqueue(priority=10, payload=b"job1")

# Combine with monitoring
from sochdb import TraceStore

trace = TraceStore(db)
span = trace.start_span("process_queue_task")

task = queue.dequeue("worker-1")
if task:
    try:
        process_task(task.payload)
        queue.ack(task.task_id)
        span.add_event("task_completed")
    finally:
        span.finish()
```

---

## 13. Vector Search

Collections store documents with embeddings for semantic search using HNSW.

**Strategy note:** HNSW is the default, correctness‑first navigator (training‑free, robust un