Cassandra 5.0: New Features and Ecosystem Evolution

-- Enable vector search extension
-- Requires Cassandra 5.0+

-- Create table with vector column
CREATE TABLE embeddings (
    id UUID PRIMARY KEY,
    document_id UUID,
    embedding VECTOR<FLOAT, 768>,
    created_at TIMESTAMP
);

-- Create vector index (ANN - Approximate Nearest Neighbor)
CREATE CUSTOM INDEX idx_embedding 
ON embeddings USING 'StorageAttachedIndex' 
WITH OPTIONS = {
    'index_name': 'embedding_idx',
    'index_type': 'ANN',
    'similarity_function': 'cosine'
};

-- Search for similar embeddings
SELECT id, document_id, 
       similarity_cosine(embedding, ?)
FROM embeddings
ORDER BY embedding ANN OF ?
LIMIT 10;

-- Enhanced JSON functions
SELECT JSON '{"name": "John", "age": 30}';

-- Parse JSON
SELECT JSON_PARSE('{"name": "John", "age": 30}').name;

-- Convert to JSON
SELECT toJSON(name), toJSON(age) FROM users;

-- Collection functions
SELECT ARRAY_LENGTH(phone_numbers) FROM users;

-- Time functions
SELECT toDate(now());

-- Aggregate improvements
SELECT COUNT(*) FROM users;

-- Improved compaction algorithms
-- Better memory management
-- Reduced CPU overhead

-- Improved network protocol
-- Better handling of large partitions
-- Reduced memory usage

-- Better query planning
-- Improved index usage
-- Reduced read latency

-- New authentication plugins
-- Better password policies

-- Create user with password policy
CREATE ROLE appuser WITH 
    LOGIN = true 
    PASSWORD = 'secure_pass'
    AND PASSWORD EXPIRES IN 90 DAYS;

-- Transparent Data Encryption (TDE)
-- Table-level encryption

-- Enable encryption at rest
ALTER TABLE sensitive_data 
WITH encryption = {
    'key_alias': 'encryption_key'
};

-- Improved cross-DC replication
-- Reduced latency
-- Better conflict resolution

-- Better DC routing
-- Local consistency level options
-- Improved failover handling

# Connect to Astra
from cassandra.cluster import Cluster
from cassandra.auth import PlainTextAuthProvider

cloud_config = {
    'secure_connect_bundle': 'secure-connect-database.zip'
}
auth_provider = PlainTextAuthProvider('client_id', 'client_secret')
cluster = Cluster(cloud=cloud_config, auth_provider=auth_provider)

# k8ssandra.yaml
apiVersion: k8ssandra.io/v1alpha1
kind: K8ssandraCluster
metadata:
  name: my-cluster
spec:
  cassandra:
    serverVersion: "5.0"
    datacenters:
      - metadata:
          name: dc1
        size: 3

# cassandra-datacenter.yaml
apiVersion: cassandra.datastax.com/v1beta1
kind: CassandraDatacenter
metadata:
  name: dc1
spec:
  clusterName: cluster1
  serverType: cassandra
  serverVersion: "5.0"
  size: 3
  storageConfig:
    cassandraDataVolumeClaimSpec:
      storageClassName: standard
      resources:
        requests:
          storage: 10Gi

# 1. Check current version
nodetool version

# 2. Verify cluster health
nodetool status

# 3. Check for running repairs
nodetool compactionstats

# 1. Backup data
nodetool snapshot -t pre_upgrade

# 2. Update Cassandra package
apt-get update
apt-get install cassandra

# 3. Restart nodes (rolling upgrade)
# One node at a time
sudo service cassandra restart

# 4. Verify upgrade
nodetool version

-- Use appropriate partition keys
-- Avoid hot spots
-- Balance partition size

-- Good partition key example
CREATE TABLE events (
    date TEXT,
    event_id TIMEUUID,
    data TEXT,
    PRIMARY KEY ((date), event_id)
);

-- Avoid large partitions
-- Target < 100MB per partition

-- Use prepared statements
-- Batch non-logged updates
-- Monitor compaction

# RAG pipeline with Cassandra vector search
from cassandra.cluster import Cluster
import numpy as np

class CassandraVectorStore:
    """Vector store backed by Cassandra 5.0."""

    def __init__(self, session, keyspace="ai"):
        self.session = session
        self.keyspace = keyspace

    def insert_document(self, doc_id, text, embedding):
        """Insert a document with its embedding."""
        self.session.execute(
            f"INSERT INTO {self.keyspace}.embeddings "
            "(id, document_id, text, embedding) VALUES (%s, %s, %s, %s)",
            (uuid.uuid4(), doc_id, text, np.array(embedding, dtype=np.float32))
        )

    def search(self, query_embedding, limit=10):
        """Find similar documents by cosine similarity."""
        rows = self.session.execute(
            f"SELECT document_id, text, "
            f"similarity_cosine(embedding, ?) AS score "
            f"FROM {self.keyspace}.embeddings "
            f"ORDER BY embedding ANN OF ? LIMIT {limit}",
            (np.array(query_embedding, dtype=np.float32),
             np.array(query_embedding, dtype=np.float32))
        )
        return [(row.text, row.score) for row in rows]

Improvement                    Impact
---------------------------    -------------------------
Improved compaction algorithms   Faster SSTable merging
Better memory management         Lower heap pressure
Reduced CPU overhead             Higher throughput per node
Unified compaction strategy       Simplified configuration

Improvement                    Impact
---------------------------    -------------------------
Improved network protocol        Lower latency between nodes
Better partition handling        Fewer timeouts on large partitions
Reduced memory usage             More efficient streaming

Improvement                    Impact
---------------------------    -------------------------
Better query planning            Faster multi-partition queries
Improved index usage             More efficient range scans
Reduced read latency             Better p99 performance

-- Role with password policy
CREATE ROLE appuser WITH
    LOGIN = true
    PASSWORD = 'secure_pass'
    AND PASSWORD EXPIRES IN 90 DAYS;

-- Role with resource permissions
GRANT SELECT ON KEYSPACE app TO appuser;
GRANT MODIFY ON TABLE app.events TO writer_role;

-- Table-level encryption (TDE)
ALTER TABLE sensitive_data
WITH encryption = {
    'key_alias': 'encryption_key'
};

-- Keyspace-level encryption settings
ALTER KEYSPACE app
WITH replication = {'class': 'NetworkTopologyStrategy', 'dc1': 3}
AND durable_writes = true;

-- Read from local DC for low latency
SELECT * FROM events
USING CONSISTENCY LOCAL_QUORUM;

-- Write with LOCAL_QUORUM (fast, local only)
INSERT INTO events (id, data)
VALUES (uuid(), 'payload')
USING CONSISTENCY LOCAL_QUORUM;

-- Batch operations
BEGIN BATCH
  INSERT INTO events (id, data) VALUES (uuid(), 'a');
  INSERT INTO events (id, data) VALUES (uuid(), 'b');
APPLY BATCH;

DC1 (us-east)   DC2 (eu-west)   DC3 (ap-south)
     \            |               /
      \           |              /
       +-- Global replication --+
          LOCAL_QUORUM reads
          LOCAL_ONE writes (async)

# Check cluster health
nodetool status

# Monitor compaction
nodetool compactionstats

# Run repairs (weekly for RF=3)
nodetool repair -pr

# Flush memtables
nodetool flush

# View gossip state
nodetool gossipinfo

# cassandra.yaml tuning
num_tokens: 16
hinted_handoff_enabled: true
concurrent_reads: 32
concurrent_writes: 32
compaction_throughput_mb_per_sec: 64
read_request_timeout_in_ms: 5000
write_request_timeout_in_ms: 2000

# 1. Check current version
nodetool version

# 2. Verify cluster health
nodetool status

# 3. Check for running repairs
nodetool compactionstats

# 4. Review schema compatibility
# - Confirm no deprecated features in use
# - Check for unsupported types

# 1. Backup data
nodetool snapshot -t pre_upgrade

# 2. Update Cassandra package
apt-get update
apt-get install cassandra

# 3. Restart nodes (rolling upgrade)
# One node at a time
sudo service cassandra restart

# 4. Verify upgrade
nodetool version

# 5. Run post-upgrade repair
nodetool repair -pr

-- Time series: partition by hour/day, cluster by timestamp
CREATE TABLE sensor_readings (
    sensor_id UUID,
    day TEXT,
    reading_time TIMESTAMP,
    value DOUBLE,
    PRIMARY KEY ((sensor_id, day), reading_time)
) WITH CLUSTERING ORDER BY (reading_time DESC);

-- Event log with partition by event type
CREATE TABLE event_log (
    event_type TEXT,
    event_id TIMEUUID,
    payload TEXT,
    created_at TIMESTAMP,
    PRIMARY KEY ((event_type), event_id, created_at)
);

-- Materialized view for alternate access
CREATE TABLE users (
    user_id UUID PRIMARY KEY,
    email TEXT,
    name TEXT
);

CREATE MATERIALIZED VIEW users_by_email AS
SELECT * FROM users
WHERE email IS NOT NULL AND user_id IS NOT NULL
PRIMARY KEY (email, user_id);

Bad:  PRIMARY KEY ((date))         -- one partition per day = hot spot
Good: PRIMARY KEY ((date, user_id)) -- distributed across users
Good: PRIMARY KEY ((bucket), id)   -- add a bucketing column

# Basic write/read benchmark
from cassandra.cluster import Cluster
import uuid, time

def benchmark(host, keyspace, operations=10000):
    cluster = Cluster([host])
    session = cluster.connect(keyspace)

    # Write benchmark
    start = time.time()
    for _ in range(operations):
        session.execute(
            "INSERT INTO events (id, data) VALUES (%s, %s)",
            (uuid.uuid4(), 'payload')
        )
    write_time = time.time() - start

    # Read benchmark
    start = time.time()
    for _ in range(operations):
        session.execute("SELECT * FROM events LIMIT 1")
    read_time = time.time() - start

    print(f"Writes: {operations / write_time:.0f} ops/sec")
    print(f"Reads:  {operations / read_time:.0f} ops/sec")
    cluster.shutdown()

-- Use appropriate partition keys
-- Avoid hot spots
-- Balance partition size

-- Good partition key example
CREATE TABLE events (
    date TEXT,
    event_id TIMEUUID,
    data TEXT,
    PRIMARY KEY ((date), event_id)
);

-- Avoid large partitions
-- Target < 100MB per partition

-- Use prepared statements
-- Batch non-logged updates
-- Monitor compaction
-- Use LOCAL_QUORUM for most reads/writes
-- Right-size token count for node count
-- Run regular repairs (weekly for RF=3)
-- Monitor tombstones and set appropriate TTL