Database Features

// Atomic operations
BEGIN TRANSACTION;
  CREATE (:Account {id: 1, balance: 1000});
  CREATE (:Account {id: 2, balance: 500});
  MATCH (a:Account {id: 1}), (b:Account {id: 2})
  SET a.balance = a.balance - 100,
      b.balance = b.balance + 100;
COMMIT;

// Rollback on error
BEGIN TRANSACTION;
  MATCH (a:Account {id: 1})
  SET a.balance = a.balance - 10000;  // Would go negative
  // Automatically rolled back if constraint fails
COMMIT;

# Python client with explicit isolation
async with client.connection() as tx:
    # Isolation is configured server-side
    await tx.begin()
    result, _ = await tx.query("MATCH (u:User {id: $id}) RETURN u", {"id": 123})
    await tx.query("CREATE (:AuditLog {user_id: $id, action: 'login'})", {"id": 123})
    await tx.commit()

BEGIN TRANSACTION;
  CREATE (:User {id: 1, name: 'Alice'});

  SAVEPOINT sp1;
  CREATE (:User {id: 2, name: 'Bob'});

  ROLLBACK TO SAVEPOINT sp1;  // Removes Bob, keeps Alice

  CREATE (:User {id: 3, name: 'Charlie'});
COMMIT;  // Commits Alice and Charlie

// B-tree index (range queries, sorting)
CREATE INDEX FOR (p:Product) ON (p.price);

// Hash index (equality lookups)
CREATE HASH INDEX FOR (u:User) ON (u.id);

// Text index (full-text search)
CREATE TEXT INDEX FOR (p:Post) ON (p.content);

// Composite index (multi-column)
CREATE INDEX FOR (u:User) ON (u.last_name, u.first_name);

// HNSW vector index (similarity search)
CREATE VECTOR INDEX FOR (d:Document) ON (d.embedding)
OPTIONS {metric: 'cosine', dimensions: 768, m: 16, ef_construction: 200};

// BM25 index (ranking search)
CREATE BM25 INDEX FOR (a:Article) ON (a.title, a.body);

// Explain query plan
EXPLAIN MATCH (u:User)-[:PURCHASED]->(p:Product)
WHERE p.price > 100
RETURN u.name, COUNT(p);

// Profile execution with metrics
PROFILE MATCH (u:User)-[:PURCHASED]->(p:Product)
WHERE p.price > 100
RETURN u.name, COUNT(p);

// Create RLS policy
CREATE POLICY tenant_isolation ON User
USING (tenant_id = current_setting('app.tenant_id'));

// Apply policy
ALTER LABEL User ENABLE ROW LEVEL SECURITY;

// Set session context
SET app.tenant_id = 'acme';

// Query only sees tenant's data
MATCH (u:User) RETURN u;  // Automatically filtered

// Transparent Data Encryption (TDE)
-- Configured at server level, encrypts all data files

// Field-Level Encryption (FLE)
CREATE (:User {
  name: 'Alice',
  ssn: ENCRYPT('123-45-6789', 'user-key'),
  email: ENCRYPT('alice@example.com', 'user-key')
});

// Decrypt on retrieval
MATCH (u:User {name: 'Alice'})
RETURN u.name, DECRYPT(u.ssn, 'user-key') AS ssn;

# Client authentication
client = Client(
    'localhost:3141',
    username='admin',
    password='secure-password',
    tls_verify=True
)

# Role-based access control
await client.execute("""
    CREATE ROLE analyst;
    GRANT SELECT ON * TO analyst;
    GRANT analyst TO user_bob;
""")

# Configure primary
geode serve --listen 0.0.0.0:3141 --replication-mode primary

# Configure replica
geode serve --listen 0.0.0.0:3142 --replication-mode replica --primary-host primary-host:3141

# Poll application ChangeLog
last_seen = "1970-01-01T00:00:00Z"
async with client.connection() as conn:
    result, _ = await conn.query(
        """
        MATCH (e:ChangeLog)
        WHERE e.emitted_at > $since AND e.label IN $labels
        RETURN e.operation AS operation,
               e.label AS label,
               e.after AS after,
               e.emitted_at AS emitted_at
        ORDER BY emitted_at
        """,
        {"since": last_seen, "labels": ["User", "Product"]},
    )
    for row in result.rows:
        print(f"Operation: {row['operation'].raw_value}")  # INSERT, UPDATE, DELETE
        print(f"Label: {row['label'].raw_value}")
        print(f"Data: {row['after'].raw_value}")
        print(f"Timestamp: {row['emitted_at'].raw_value}")

# Create backup
geode backup create --destination /backups/geode-20240615

# Restore from backup
geode backup restore --source /backups/geode-20240615 --target /data/geode-restored

# Point-in-time restore
geode backup restore --source /backups/geode-20240615 --timestamp '2024-06-15T14:30:00Z'

# Configure continuous backup
geode serve --backup-enabled --backup-interval 3600 --backup-destination /backups

# Prepare statement once
stmt = await client.prepare("""
    MATCH (u:User {id: $id})
    RETURN u.name, u.email
""")

# Execute multiple times with different parameters
result1, _ = await stmt.execute({"id": 1})
result2, _ = await stmt.execute({"id": 2})
result3, _ = await stmt.execute({"id": 3})

// Create procedure
CREATE PROCEDURE transfer_funds(from_id INT, to_id INT, amount FLOAT)
LANGUAGE GQL
AS $$
BEGIN TRANSACTION;
  MATCH (from:Account {id: $from_id}), (to:Account {id: $to_id})
  WHERE from.balance >= $amount
  SET from.balance = from.balance - $amount,
      to.balance = to.balance + $amount;
  RETURN from, to;
COMMIT;
$$;

// Call procedure
CALL transfer_funds(1, 2, 100.0);

# Enable query cache
client = Client('localhost:3141', enable_cache=True, cache_ttl=300)

# First execution hits database
result1 = await client.query("MATCH (u:User) WHERE u.active = true RETURN COUNT(u)")

# Second execution uses cache (if within TTL)
result2 = await client.query("MATCH (u:User) WHERE u.active = true RETURN COUNT(u)")

// Query current metrics
SHOW METRICS;

// Results include:
// - queries_per_second
// - average_query_latency_ms
// - active_connections
// - cache_hit_rate
// - index_usage_stats
// - transaction_throughput

# Enable slow query log
geode serve --slow-query-log --slow-query-threshold 1000  # Log queries > 1s

# Health check endpoint
health = await client.health_check()
print(f"Status: {health['status']}")  # 'healthy', 'degraded', 'unhealthy'
print(f"Uptime: {health['uptime_seconds']}")
print(f"Version: {health['version']}")

// Unique constraints
CREATE CONSTRAINT ON (u:User) ASSERT u.email IS UNIQUE;

// Existence constraints
CREATE CONSTRAINT ON (u:User) ASSERT EXISTS(u.email);

// Property type constraints
CREATE CONSTRAINT ON (p:Product) ASSERT p.price IS :: FLOAT;

// Check constraints
CREATE CONSTRAINT ON (u:User) ASSERT u.age >= 0 AND u.age <= 150;

// Create versioned schema change
BEGIN SCHEMA MIGRATION 'v2.0.0';
  CREATE INDEX FOR (p:Product) ON (p.category);
  ALTER LABEL User ADD PROPERTY created_at DATETIME DEFAULT NOW();
COMMIT SCHEMA MIGRATION;

// View migration history
SHOW SCHEMA MIGRATIONS;

# Import from CSV
geode import --format csv --file users.csv --label User --mapping id:id,name:name,email:email

# Import from JSON
geode import --format json --file data.json

# Import from Cypher/GQL dump
geode import --format gql --file database.gql

# Export to CSV
geode export --format csv --query "MATCH (u:User) RETURN u.id, u.name, u.email" --output users.csv

# Export entire database
geode export --format gql --output database-backup.gql

# Export to JSON
geode export --format json --query "MATCH (u:User) RETURN u" --output users.json

// Schema-based multi-tenancy
CREATE GRAPH tenant_acme;
CREATE GRAPH tenant_globex;

// Switch context
USE GRAPH tenant_acme;
MATCH (u:User) RETURN u;  // Only sees Acme data

// RLS-based multi-tenancy
CREATE POLICY tenant_filter ON *
USING (tenant_id = current_setting('app.tenant_id'));

# Python (async)
from geode_client import Client
client = Client(host="localhost", port=3141)
async with client.connection() as conn:
    result, _ = await conn.query("MATCH (u:User) RETURN u")

// Go (database/sql)
import "database/sql"
import _ "geodedb.com/geode"

db, _ := sql.Open("geode", "quic://localhost:3141")
rows, _ := db.Query("MATCH (u:User) RETURN u.name")

// Rust (async)
use geode_client::Client;

let client = Client::from_dsn("localhost:3141")?;
let mut conn = client.connect().await?;
let (page, _) = conn.query("MATCH (u:User) RETURN u").await?;

class MultiTenantClient:
    def __init__(self, base_client):
        self.client = base_client
    
    async def setup_tenant_isolation(self):
        """Configure RLS policies for multi-tenancy"""
        await self.client.execute("""
            -- Create RLS policy for users
            CREATE POLICY tenant_users ON User
            USING (tenant_id = current_setting('app.tenant_id'));
            
            -- Create RLS policy for orders
            CREATE POLICY tenant_orders ON Order
            USING (tenant_id = current_setting('app.tenant_id'));
            
            -- Enable RLS
            ALTER LABEL User ENABLE ROW LEVEL SECURITY;
            ALTER LABEL Order ENABLE ROW LEVEL SECURITY;
        """)
    
    async def execute_for_tenant(self, tenant_id, query, params=None):
        """Execute query in tenant context"""
        async with self.client.connection() as conn:
            await conn.begin()
            try:
                # Set tenant context
                await conn.execute("""
                    SET app.tenant_id = $tenant_id
                """, {"tenant_id": tenant_id})

                # Execute query (RLS auto-applied)
                result, _ = await conn.query(query, params)
                await conn.commit()
                return result
            except Exception:
                await conn.rollback()
                raise

# Usage
client = MultiTenantClient(base_client)
await client.setup_tenant_isolation()

# Queries automatically filtered by tenant
result = await client.execute_for_tenant("tenant_123", """
    MATCH (u:User) RETURN u.name
""")  # Only sees tenant_123 users

-- Point-in-time query
MATCH (u:User)
WHERE u.valid_from <= datetime('2024-01-15')
  AND (u.valid_to IS NULL OR u.valid_to > datetime('2024-01-15'))
RETURN u.name, u.email

-- Temporal join
MATCH (e:Employee)-[r:WORKS_FOR]->(c:Company)
WHERE r.start_date <= datetime('2023-06-01')
  AND (r.end_date IS NULL OR r.end_date >= datetime('2023-06-01'))
RETURN e.name, c.name

-- Audit trail
MATCH (a:Account {id: $account_id})-[:HAS_HISTORY]->(h:AccountHistory)
WHERE h.timestamp BETWEEN datetime('2024-01-01') AND datetime('2024-12-31')
RETURN h.balance, h.timestamp
ORDER BY h.timestamp

-- Create full-text index
CREATE TEXT INDEX FOR (p:Post) ON (p.title, p.content);
CREATE TEXT INDEX FOR (a:Article) ON (a.body)
OPTIONS {language: 'english', stemming: true};

-- Basic text search
MATCH (p:Post)
WHERE p.content CONTAINS 'graph database'
RETURN p.title, p.content

-- Ranked search with BM25
CREATE BM25 INDEX FOR (d:Document) ON (d.title, d.body);

MATCH (d:Document)
WHERE d MATCH 'graph AND (database OR query)'
RETURN d.title, score(d) as relevance
ORDER BY relevance DESC
LIMIT 10

-- Phrase search
MATCH (a:Article)
WHERE a.body CONTAINS PHRASE 'machine learning'
RETURN a.title

-- Fuzzy search
MATCH (u:User)
WHERE u.name =~ fuzzy('Alise', max_edit_distance: 2)
RETURN u.name

async def semantic_search(client, query_text, limit=10):
    """Perform semantic search using vector embeddings"""
    # Generate query embedding
    query_embedding = await generate_embedding(query_text)
    
    # Vector similarity search
    results, _ = await client.query("""
        MATCH (d:Document)
        WITH d, vector_similarity(d.embedding, $query_embedding, 'cosine') as similarity
        WHERE similarity > 0.7
        RETURN d.id, d.title, d.content, similarity
        ORDER BY similarity DESC
        LIMIT $limit
    """, {
        "query_embedding": query_embedding,
        "limit": limit
    })
    
    return results

# HNSW index for fast approximate search
await client.execute("""
    CREATE VECTOR INDEX document_embeddings
    FOR (d:Document) ON (d.embedding)
    OPTIONS {
        metric: 'cosine',
        dimensions: 768,
        m: 16,
        ef_construction: 200,
        ef_search: 100
    }
""")

-- PageRank
CALL algo.pageRank('User', 'FOLLOWS', {
    iterations: 20,
    dampingFactor: 0.85
}) YIELD node, score
RETURN node.name, score
ORDER BY score DESC
LIMIT 10

-- Community Detection (Louvain)
CALL algo.louvain('User', 'FRIENDS_WITH', {
    resolution: 1.0
}) YIELD node, community
RETURN community, COUNT(node) as size
ORDER BY size DESC

-- Shortest Path
MATCH path = shortestPath((start:User {id: $id1})-[:KNOWS*]-(end:User {id: $id2}))
RETURN [node IN nodes(path) | node.name] as path, length(path) as distance

-- Betweenness Centrality
CALL algo.betweenness('User', 'KNOWS') YIELD node, score
RETURN node.name, score
ORDER BY score DESC
LIMIT 20

-- Uniqueness constraints
CREATE CONSTRAINT ON (u:User) ASSERT u.email IS UNIQUE;
CREATE CONSTRAINT ON (p:Product) ASSERT (p.sku, p.variant) IS UNIQUE;

-- Existence constraints
CREATE CONSTRAINT ON (u:User) ASSERT EXISTS(u.email);
CREATE CONSTRAINT ON (o:Order) ASSERT EXISTS(o.customer_id);

-- Type constraints
CREATE CONSTRAINT ON (p:Product) ASSERT p.price IS :: FLOAT;
CREATE CONSTRAINT ON (u:User) ASSERT p.created_at IS :: DATETIME;

-- Check constraints
CREATE CONSTRAINT ON (u:User) ASSERT u.age BETWEEN 0 AND 150;
CREATE CONSTRAINT ON (p:Product) ASSERT p.price > 0;
CREATE CONSTRAINT ON (r:Rating) ASSERT r.score IN [1,2,3,4,5];

-- Pattern constraints
CREATE CONSTRAINT ON (u:User) 
ASSERT u.email =~ '^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\\.[a-zA-Z]{2,}$';

-- Custom validation functions
CREATE FUNCTION validate_credit_card(card_number STRING) RETURNS BOOLEAN
AS $$
    -- Luhn algorithm implementation
    RETURN luhn_check(card_number);
$$;

CREATE CONSTRAINT ON (p:Payment)
ASSERT validate_credit_card(p.card_number);

-- Create spatial index
CREATE SPATIAL INDEX FOR (l:Location) ON (l.coordinates);

-- Point creation
CREATE (store:Store {
    name: 'Main Street Store',
    coordinates: point({latitude: 37.7749, longitude: -122.4194})
});

-- Distance queries
MATCH (u:User {id: $user_id}), (store:Store)
WITH u, store, distance(u.location, store.location) as dist_meters
WHERE dist_meters < 5000  // Within 5km
RETURN store.name, dist_meters
ORDER BY dist_meters

-- Polygon containment
MATCH (p:Place)
WHERE contains(
    polygon([
        point({latitude: 37.8, longitude: -122.5}),
        point({latitude: 37.8, longitude: -122.3}),
        point({latitude: 37.7, longitude: -122.3}),
        point({latitude: 37.7, longitude: -122.5})
    ]),
    p.location
)
RETURN p.name

from functools import lru_cache
import hashlib

class CachedClient:
    def __init__(self, client, cache_backend):
        self.client = client
        self.cache = cache_backend
    
    def cache_key(self, query, params):
        """Generate cache key from query and params"""
        content = f"{query}:{json.dumps(params, sort_keys=True)}"
        return hashlib.sha256(content.encode()).hexdigest()
    
    async def execute_cached(self, query, params=None, ttl=300):
        """Execute query with result caching"""
        key = self.cache_key(query, params)
        
        # Check cache
        cached = await self.cache.get(key)
        if cached:
            return json.loads(cached)
        
        # Execute query
        result, _ = await self.client.query(query, params)
        
        # Store in cache
        await self.cache.set(key, json.dumps(result), ex=ttl)
        
        return result
    
    async def invalidate_pattern(self, pattern):
        """Invalidate cached queries matching pattern"""
        keys = await self.cache.keys(f"{pattern}*")
        if keys:
            await self.cache.delete(*keys)

async def batch_insert_users(client, users, batch_size=1000):
    """Insert users in batches"""
    for i in range(0, len(users), batch_size):
        batch = users[i:i + batch_size]
        
        await client.execute("""
            UNWIND $users AS user_data
            CREATE (u:User {
                id: user_data.id,
                name: user_data.name,
                email: user_data.email,
                created_at: datetime()
            })
        """, {"users": batch})

async def batch_update_prices(client, price_updates):
    """Update prices in batch"""
    await client.execute("""
        UNWIND $updates AS update
        MATCH (p:Product {id: update.product_id})
        SET p.price = update.new_price,
            p.updated_at = datetime()
    """, {"updates": price_updates})