Specialized Databases - Vector, Spatial, Embedded, and Ledger

# Cosine Similarity (angle between vectors)
# Range: -1 to 1, where 1 = identical direction
cosine_sim = dot(A, B) / (norm(A) * norm(B))

# Euclidean Distance (straight-line distance)
# Range: 0 to infinity, where 0 = identical
euclidean_dist = sqrt(sum((A - B)^2))

# Dot Product (magnitude-weighted similarity)
# Range: -infinity to infinity
dot_product = sum(A * B)

# Using a vector database for semantic search
from sentence_transformers import SentenceTransformer
import numpy as np

# In practice, use actual Pinecone/Milvus client
# This demonstrates the concepts

# Load embedding model
model = SentenceTransformer('all-MiniLM-L6-v2')

# Sample documents
documents = [
    {"id": "1", "text": "How to train a machine learning model"},
    {"id": "2", "text": "Best practices for deep learning"},
    {"id": "3", "text": "Introduction to neural networks"},
    {"id": "4", "text": "Cooking delicious pasta recipes"},
    {"id": "5", "text": "Italian cuisine fundamentals"},
]

# Generate embeddings
for doc in documents:
    doc["embedding"] = model.encode(doc["text"]).tolist()

# Pinecone example (pseudo-code)
"""
import pinecone

# Initialize
pinecone.init(api_key="your-key", environment="us-west1-gcp")

# Create index
pinecone.create_index(
    name="documents",
    dimension=384,  # MiniLM dimension
    metric="cosine"
)

# Connect to index
index = pinecone.Index("documents")

# Upsert vectors
vectors = [
    (doc["id"], doc["embedding"], {"text": doc["text"]})
    for doc in documents
]
index.upsert(vectors=vectors)

# Query
query_text = "How do I build AI models?"
query_embedding = model.encode(query_text).tolist()

results = index.query(
    vector=query_embedding,
    top_k=3,
    include_metadata=True
)

for match in results["matches"]:
    print(f"Score: {match['score']:.4f} - {match['metadata']['text']}")
"""

# Milvus example
"""
from pymilvus import connections, Collection, FieldSchema, CollectionSchema, DataType

# Connect
connections.connect(host="localhost", port="19530")

# Define schema
fields = [
    FieldSchema(name="id", dtype=DataType.INT64, is_primary=True),
    FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=384),
    FieldSchema(name="text", dtype=DataType.VARCHAR, max_length=1000)
]
schema = CollectionSchema(fields)
collection = Collection(name="documents", schema=schema)

# Insert
collection.insert([
    [1, 2, 3],  # ids
    [[0.1, 0.2, ...], [0.3, 0.4, ...], [0.5, 0.6, ...]],  # embeddings
    ["text1", "text2", "text3"]  # texts
])

# Create index
collection.create_index(
    field_name="embedding",
    index_params={
        "metric_type": "COSINE",
        "index_type": "HNSW",
        "params": {"M": 16, "efConstruction": 200}
    }
)

# Search
collection.load()
results = collection.search(
    data=[query_embedding],
    anns_field="embedding",
    param={"metric_type": "COSINE", "params": {"ef": 100}},
    limit=3,
    output_fields=["text"]
)
"""

# pgvector example (PostgreSQL extension)
"""
-- Enable extension
CREATE EXTENSION vector;

-- Create table with vector column
CREATE TABLE documents (
    id SERIAL PRIMARY KEY,
    text TEXT,
    embedding vector(384)
);

-- Insert with embedding
INSERT INTO documents (text, embedding)
VALUES ('How to train a machine learning model', '[0.1, 0.2, ...]');

-- Create index for fast similarity search
CREATE INDEX ON documents USING ivfflat (embedding vector_cosine_ops)
WITH (lists = 100);

-- Or HNSW index (better accuracy)
CREATE INDEX ON documents USING hnsw (embedding vector_cosine_ops);

-- Similarity search
SELECT id, text, 1 - (embedding <=> '[0.1, 0.2, ...]') AS similarity
FROM documents
ORDER BY embedding <=> '[0.1, 0.2, ...]'
LIMIT 5;

-- Cosine distance: <=>
-- Euclidean distance: <->
-- Inner product: <#>
"""

-- Enable PostGIS extension
CREATE EXTENSION postgis;

-- Create a table with location data
CREATE TABLE stores (
    id SERIAL PRIMARY KEY,
    name VARCHAR(100) NOT NULL,
    location GEOGRAPHY(POINT, 4326),
    service_area GEOGRAPHY(POLYGON, 4326)
);

-- Insert store locations
INSERT INTO stores (name, location) VALUES
    ('Downtown Store', ST_SetSRID(ST_MakePoint(-73.985428, 40.748817), 4326)),
    ('Midtown Store', ST_SetSRID(ST_MakePoint(-73.978873, 40.756947), 4326)),
    ('Brooklyn Store', ST_SetSRID(ST_MakePoint(-73.986229, 40.693254), 4326));

-- Create spatial index
CREATE INDEX idx_stores_location ON stores USING GIST (location);

-- Find stores within 2km of a point
SELECT 
    name,
    ST_Distance(
        location,
        ST_SetSRID(ST_MakePoint(-73.980000, 40.750000), 4326)::geography
    ) AS distance_meters
FROM stores
WHERE ST_DWithin(
    location,
    ST_SetSRID(ST_MakePoint(-73.980000, 40.750000), 4326)::geography,
    2000  -- 2km radius
)
ORDER BY distance_meters;

-- Find the nearest store
SELECT 
    name,
    ST_Distance(
        location,
        ST_SetSRID(ST_MakePoint(-73.980000, 40.750000), 4326)::geography
    ) AS distance_meters
FROM stores
ORDER BY location <-> ST_SetSRID(ST_MakePoint(-73.980000, 40.750000), 4326)::geography
LIMIT 1;

-- Create delivery zones (polygons)
CREATE TABLE delivery_zones (
    id SERIAL PRIMARY KEY,
    zone_name VARCHAR(100),
    area GEOGRAPHY(POLYGON, 4326)
);

INSERT INTO delivery_zones (zone_name, area) VALUES
    ('Manhattan', ST_SetSRID(ST_GeomFromText('POLYGON((-74.02 40.70, -74.02 40.80, -73.93 40.80, -73.93 40.70, -74.02 40.70))'), 4326));

-- Check if a point is within delivery zone
SELECT zone_name
FROM delivery_zones
WHERE ST_Contains(
    area::geometry,
    ST_SetSRID(ST_MakePoint(-73.985428, 40.748817), 4326)
);

-- Find stores within a polygon
SELECT s.name
FROM stores s
JOIN delivery_zones dz ON ST_Within(s.location::geometry, dz.area::geometry)
WHERE dz.zone_name = 'Manhattan';

-- Calculate area of delivery zone in square kilometers
SELECT 
    zone_name,
    ST_Area(area) / 1000000 AS area_sq_km
FROM delivery_zones;

-- Find routes (LineString operations)
CREATE TABLE routes (
    id SERIAL PRIMARY KEY,
    route_name VARCHAR(100),
    path GEOGRAPHY(LINESTRING, 4326)
);

INSERT INTO routes (route_name, path) VALUES
    ('Route A', ST_SetSRID(ST_MakeLine(ARRAY[
        ST_MakePoint(-73.985428, 40.748817),
        ST_MakePoint(-73.978873, 40.756947),
        ST_MakePoint(-73.986229, 40.693254)
    ]), 4326));

-- Calculate route length
SELECT 
    route_name,
    ST_Length(path) / 1000 AS length_km
FROM routes;

import sqlite3
import json

# SQLite basic operations
conn = sqlite3.connect('app_data.db')
cursor = conn.cursor()

# Create tables
cursor.execute('''
    CREATE TABLE IF NOT EXISTS users (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        email TEXT UNIQUE NOT NULL,
        name TEXT NOT NULL,
        preferences TEXT,  -- JSON stored as TEXT
        created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
    )
''')

cursor.execute('''
    CREATE TABLE IF NOT EXISTS offline_queue (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        action TEXT NOT NULL,
        payload TEXT NOT NULL,
        created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
        synced INTEGER DEFAULT 0
    )
''')

# Create indexes
cursor.execute('CREATE INDEX IF NOT EXISTS idx_users_email ON users(email)')
cursor.execute('CREATE INDEX IF NOT EXISTS idx_queue_synced ON offline_queue(synced)')

# Insert with JSON
user_prefs = json.dumps({"theme": "dark", "notifications": True})
cursor.execute(
    'INSERT INTO users (email, name, preferences) VALUES (?, ?, ?)',
    ('alice@example.com', 'Alice Smith', user_prefs)
)

# Offline-first pattern: queue operations for sync
def queue_action(action, payload):
    cursor.execute(
        'INSERT INTO offline_queue (action, payload) VALUES (?, ?)',
        (action, json.dumps(payload))
    )
    conn.commit()

queue_action('create_order', {'product_id': 123, 'quantity': 2})

# Sync pending operations
def get_pending_sync():
    cursor.execute(
        'SELECT id, action, payload FROM offline_queue WHERE synced = 0 ORDER BY created_at'
    )
    return cursor.fetchall()

def mark_synced(ids):
    cursor.execute(
        f'UPDATE offline_queue SET synced = 1 WHERE id IN ({",".join("?" * len(ids))})',
        ids
    )
    conn.commit()

# JSON operations (SQLite 3.38+)
cursor.execute('''
    SELECT 
        name,
        json_extract(preferences, '$.theme') as theme
    FROM users
    WHERE json_extract(preferences, '$.notifications') = 1
''')

# Full-text search with FTS5
cursor.execute('''
    CREATE VIRTUAL TABLE IF NOT EXISTS documents_fts USING fts5(
        title,
        content,
        content='documents',
        content_rowid='id'
    )
''')

# Transaction management
try:
    cursor.execute('BEGIN TRANSACTION')
    cursor.execute('UPDATE users SET name = ? WHERE id = ?', ('Alice Johnson', 1))
    cursor.execute('INSERT INTO offline_queue (action, payload) VALUES (?, ?)', 
                   ('update_user', '{"id": 1, "name": "Alice Johnson"}'))
    conn.commit()
except Exception as e:
    conn.rollback()
    raise

# Performance settings for SQLite
cursor.execute('PRAGMA journal_mode = WAL')  # Write-ahead logging
cursor.execute('PRAGMA synchronous = NORMAL')  # Balance safety/speed
cursor.execute('PRAGMA cache_size = -64000')  # 64MB cache
cursor.execute('PRAGMA temp_store = MEMORY')  # Temp tables in memory

conn.close()

// DuckDB for embedded analytics (JavaScript example)
import * as duckdb from 'duckdb';

const db = new duckdb.Database(':memory:');
const conn = db.connect();

// Create and load data
conn.run(`
    CREATE TABLE sales (
        date DATE,
        product VARCHAR,
        region VARCHAR,
        amount DECIMAL(10,2)
    );
    
    INSERT INTO sales VALUES
        ('2024-01-01', 'Widget', 'US', 100.00),
        ('2024-01-01', 'Gadget', 'US', 150.00),
        ('2024-01-02', 'Widget', 'EU', 120.00),
        ('2024-01-02', 'Gadget', 'EU', 180.00);
`);

// Analytical queries on local data
const result = conn.all(`
    SELECT 
        region,
        SUM(amount) as total_sales,
        COUNT(*) as transaction_count
    FROM sales
    GROUP BY region
    ORDER BY total_sales DESC
`);

// Import from Parquet (local file analytics)
conn.run(`
    CREATE TABLE logs AS 
    SELECT * FROM read_parquet('logs/*.parquet');
`);

// Export results
conn.run(`
    COPY (SELECT * FROM sales) TO 'sales_export.csv' (HEADER, DELIMITER ',');
`);

-- Amazon QLDB-style operations (conceptual SQL)

-- QLDB uses PartiQL (SQL-compatible)
-- Create a ledger table
CREATE TABLE accounts;
CREATE TABLE transactions;

-- Insert initial account
INSERT INTO accounts VALUE {
    'AccountId': 'ACC001',
    'Owner': 'Alice Smith',
    'Balance': 1000.00,
    'Status': 'Active'
};

-- Transfer funds (creates immutable history)
-- Step 1: Debit source account
UPDATE accounts SET Balance = Balance - 100.00
WHERE AccountId = 'ACC001';

-- Step 2: Credit destination account
UPDATE accounts SET Balance = Balance + 100.00
WHERE AccountId = 'ACC002';

-- Step 3: Record transaction
INSERT INTO transactions VALUE {
    'TransactionId': 'TXN001',
    'FromAccount': 'ACC001',
    'ToAccount': 'ACC002',
    'Amount': 100.00,
    'Timestamp': `2024-01-15T10:30:00Z`
};

-- Query current state
SELECT * FROM accounts WHERE AccountId = 'ACC001';

-- Query history (QLDB-specific)
SELECT * FROM history(accounts) AS h
WHERE h.data.AccountId = 'ACC001';

-- Verify document integrity (get document hash)
SELECT h.metadata.id, h.hash, h.data
FROM history(accounts) AS h
WHERE h.data.AccountId = 'ACC001';

-- ImmuDB example (immutable key-value + SQL)
/*
# Using immudb Python client
from immudb import ImmudbClient

client = ImmudbClient()
client.login("immudb", "immudb")
client.useDatabase("defaultdb")

# Set with verified proof
client.verifiedSet(b"account:ACC001", b"1000.00")

# Get with proof verification
entry = client.verifiedGet(b"account:ACC001")
print(f"Value: {entry.value}, Verified: {entry.verified}")

# SQL interface
client.sqlExec("""
    CREATE TABLE IF NOT EXISTS audit_log (
        id INTEGER AUTO_INCREMENT,
        action VARCHAR[50],
        user_id VARCHAR[50],
        details VARCHAR[500],
        timestamp TIMESTAMP,
        PRIMARY KEY (id)
    )
""")

# Insert audit entry (immutable)
client.sqlExec("""
    INSERT INTO audit_log (action, user_id, details, timestamp)
    VALUES ('LOGIN', 'user123', 'Successful login from IP 192.168.1.1', NOW())
""")

# Query with time travel
result = client.sqlQuery("""
    SELECT * FROM audit_log 
    BEFORE TX 1000
    WHERE user_id = 'user123'
""")

# History of a specific key
history = client.history(b"account:ACC001")
for entry in history:
    print(f"TX: {entry.tx}, Value: {entry.value}")
*/

# Using pgvector with PostgreSQL
import psycopg2
from sentence_transformers import SentenceTransformer

# Setup
model = SentenceTransformer('all-MiniLM-L6-v2')

conn = psycopg2.connect("postgresql://user:pass@localhost/vectordb")
cur = conn.cursor()

# Create table
cur.execute("""
    CREATE EXTENSION IF NOT EXISTS vector;
    
    CREATE TABLE IF NOT EXISTS documents (
        id SERIAL PRIMARY KEY,
        title TEXT NOT NULL,
        content TEXT NOT NULL,
        category VARCHAR(50),
        embedding vector(384),
        created_at TIMESTAMP DEFAULT NOW()
    );
    
    CREATE INDEX IF NOT EXISTS idx_docs_embedding 
    ON documents USING hnsw (embedding vector_cosine_ops);
""")

# Insert documents
documents = [
    ("Machine Learning Basics", "Introduction to supervised and unsupervised learning algorithms", "tech"),
    ("Deep Learning with Python", "Building neural networks using TensorFlow and PyTorch", "tech"),
    ("Healthy Cooking Tips", "How to prepare nutritious meals at home", "lifestyle"),
    ("Database Optimization", "Techniques for improving SQL query performance", "tech"),
]

for title, content, category in documents:
    embedding = model.encode(f"{title} {content}").tolist()
    cur.execute(
        "INSERT INTO documents (title, content, category, embedding) VALUES (%s, %s, %s, %s)",
        (title, content, category, embedding)
    )

conn.commit()

# Search function
def semantic_search(query, category=None, limit=5):
    query_embedding = model.encode(query).tolist()
    
    if category:
        cur.execute("""
            SELECT title, content, 1 - (embedding <=> %s::vector) AS similarity
            FROM documents
            WHERE category = %s
            ORDER BY embedding <=> %s::vector
            LIMIT %s
        """, (query_embedding, category, query_embedding, limit))
    else:
        cur.execute("""
            SELECT title, content, 1 - (embedding <=> %s::vector) AS similarity
            FROM documents
            ORDER BY embedding <=> %s::vector
            LIMIT %s
        """, (query_embedding, query_embedding, limit))
    
    return cur.fetchall()

# Test search
results = semantic_search("How do I build AI models?", category="tech")
for title, content, similarity in results:
    print(f"{similarity:.4f}: {title}")

-- Create schema
CREATE EXTENSION IF NOT EXISTS postgis;

CREATE TABLE stores (
    id SERIAL PRIMARY KEY,
    name VARCHAR(100) NOT NULL,
    address TEXT,
    location GEOGRAPHY(POINT, 4326),
    delivery_radius_km FLOAT DEFAULT 5.0,
    is_open BOOLEAN DEFAULT TRUE
);

CREATE TABLE delivery_zones (
    id SERIAL PRIMARY KEY,
    store_id INTEGER REFERENCES stores(id),
    zone_type VARCHAR(20), -- 'express', 'standard', 'extended'
    area GEOGRAPHY(POLYGON, 4326),
    delivery_fee DECIMAL(10,2)
);

-- Create indexes
CREATE INDEX idx_stores_location ON stores USING GIST (location);
CREATE INDEX idx_zones_area ON delivery_zones USING GIST (area);

-- Insert sample stores
INSERT INTO stores (name, address, location, delivery_radius_km) VALUES
    ('Central Kitchen', '123 Main St', ST_SetSRID(ST_MakePoint(-73.985, 40.748), 4326), 5.0),
    ('East Side Kitchen', '456 East Ave', ST_SetSRID(ST_MakePoint(-73.960, 40.755), 4326), 4.0),
    ('Brooklyn Hub', '789 Brooklyn Blvd', ST_SetSRID(ST_MakePoint(-73.986, 40.693), 4326), 6.0);

-- Create delivery zones from radius
INSERT INTO delivery_zones (store_id, zone_type, area, delivery_fee)
SELECT 
    id,
    'standard',
    ST_Buffer(location::geometry, delivery_radius_km * 1000)::geography,
    2.99
FROM stores;

-- Function: Find nearest open stores
CREATE OR REPLACE FUNCTION find_nearest_stores(
    user_lat FLOAT,
    user_lng FLOAT,
    max_distance_km FLOAT DEFAULT 10.0,
    limit_count INT DEFAULT 5
)
RETURNS TABLE (
    store_id INT,
    store_name VARCHAR,
    distance_km FLOAT,
    can_deliver BOOLEAN
) AS $$
BEGIN
    RETURN QUERY
    SELECT 
        s.id,
        s.name,
        ST_Distance(
            s.location,
            ST_SetSRID(ST_MakePoint(user_lng, user_lat), 4326)::geography
        ) / 1000 AS dist_km,
        ST_DWithin(
            s.location,
            ST_SetSRID(ST_MakePoint(user_lng, user_lat), 4326)::geography,
            s.delivery_radius_km * 1000
        ) AS can_deliver
    FROM stores s
    WHERE s.is_open = TRUE
      AND ST_DWithin(
          s.location,
          ST_SetSRID(ST_MakePoint(user_lng, user_lat), 4326)::geography,
          max_distance_km * 1000
      )
    ORDER BY dist_km
    LIMIT limit_count;
END;
$$ LANGUAGE plpgsql;

-- Use the function
SELECT * FROM find_nearest_stores(40.750, -73.980);

-- Check delivery zone and fee
CREATE OR REPLACE FUNCTION get_delivery_options(
    user_lat FLOAT,
    user_lng FLOAT
)
RETURNS TABLE (
    store_id INT,
    store_name VARCHAR,
    zone_type VARCHAR,
    delivery_fee DECIMAL,
    estimated_time_min INT
) AS $$
BEGIN
    RETURN QUERY
    SELECT 
        s.id,
        s.name,
        dz.zone_type,
        dz.delivery_fee,
        CASE dz.zone_type
            WHEN 'express' THEN 15
            WHEN 'standard' THEN 30
            WHEN 'extended' THEN 45
        END AS est_time
    FROM stores s
    JOIN delivery_zones dz ON s.id = dz.store_id
    WHERE s.is_open = TRUE
      AND ST_Contains(
          dz.area::geometry,
          ST_SetSRID(ST_MakePoint(user_lng, user_lat), 4326)
      )
    ORDER BY dz.delivery_fee;
END;
$$ LANGUAGE plpgsql;

SELECT * FROM get_delivery_options(40.750, -73.980);