Vector Storage Patterns
Efficient patterns and architectures for storing and managing vector data at scale
🗄️ Vector Storage Fundamentals
Definition: Strategies and patterns for efficiently storing, organizing, and accessing high-dimensional vector data
Simple Analogy: Like designing a well-organized warehouse where similar items are stored together, with efficient systems for finding and retrieving what you need quickly.
Core Storage Challenges
High Dimensionality
- Challenge: Vectors often have 256-4096 dimensions
- Impact: Traditional indexing strategies become inefficient
- Solution: Specialized indexing and compression techniques
Volume Scale
- Challenge: Millions to billions of vectors
- Impact: Memory and storage requirements grow rapidly
- Solution: Hierarchical storage and partitioning strategies
Access Patterns
- Challenge: Need for both batch processing and real-time queries
- Impact: Different storage optimizations needed
- Solution: Hybrid storage architectures
Storage Architecture Patterns
1. Flat Storage Pattern
Simple approach where all vectors are stored in a single, flat structure.
python
import numpy as np
import pickle
import h5py
class FlatVectorStorage:
def __init__(self, dimension):
self.dimension = dimension
self.vectors = []
self.metadata = []
def add_vector(self, vector, metadata=None):
"""Add a single vector with optional metadata"""
if len(vector) != self.dimension:
raise ValueError(f"Vector dimension {len(vector)} doesn't match expected {self.dimension}")
self.vectors.append(vector)
self.metadata.append(metadata or {})
return len(self.vectors) - 1 # Return index
def add_batch(self, vectors, metadata_list=None):
"""Add multiple vectors at once"""
if metadata_list is None:
metadata_list = [{}] * len(vectors)
for vector, metadata in zip(vectors, metadata_list):
self.add_vector(vector, metadata)
def get_vector(self, index):
"""Retrieve vector by index"""
if 0 <= index < len(self.vectors):
return self.vectors[index], self.metadata[index]
return None, None
def save_to_disk(self, filepath):
"""Save vectors to disk using HDF5"""
with h5py.File(filepath, 'w') as f:
# Store vectors as dataset
vectors_array = np.array(self.vectors)
f.create_dataset('vectors', data=vectors_array)
f.create_dataset('dimension', data=self.dimension)
# Store metadata separately (pickle for complex objects)
metadata_bytes = pickle.dumps(self.metadata)
f.create_dataset('metadata', data=np.void(metadata_bytes))
def load_from_disk(self, filepath):
"""Load vectors from disk"""
with h5py.File(filepath, 'r') as f:
self.vectors = f['vectors'][:]
self.dimension = f['dimension'][()]
metadata_bytes = f['metadata'][()].tobytes()
self.metadata = pickle.loads(metadata_bytes)
# Example usage
storage = FlatVectorStorage(dimension=128)
# Add vectors
vectors = np.random.random((1000, 128))
metadata = [{'id': i, 'category': f'cat_{i%5}'} for i in range(1000)]
storage.add_batch(vectors, metadata)
print(f"Stored {len(storage.vectors)} vectors")
# Save and load
storage.save_to_disk('vectors.h5')
print("Vectors saved to disk")Pros:
- Simple to implement and understand
- Good for small to medium datasets
- Fast sequential access
Cons:
- Poor performance for similarity search
- No indexing for fast retrieval
- Memory limitations for large datasets
2. Partitioned Storage Pattern
Divides vectors into partitions based on various criteria.
python
import hashlib
from collections import defaultdict
class PartitionedVectorStorage:
def __init__(self, dimension, num_partitions=16):
self.dimension = dimension
self.num_partitions = num_partitions
self.partitions = defaultdict(list)
self.metadata_partitions = defaultdict(list)
self.partition_stats = defaultdict(int)
def get_partition_key(self, vector, method='hash'):
"""Determine which partition a vector belongs to"""
if method == 'hash':
# Hash-based partitioning
vector_bytes = vector.tobytes()
hash_value = int(hashlib.md5(vector_bytes).hexdigest(), 16)
return hash_value % self.num_partitions
elif method == 'random':
# Random partitioning
return np.random.randint(0, self.num_partitions)
elif method == 'clustering':
# Cluster-based partitioning (simplified)
# In practice, you'd use K-means or similar
first_dim_value = vector[0]
return int((first_dim_value + 1) * self.num_partitions / 2) % self.num_partitions
def add_vector(self, vector, metadata=None, partition_method='hash'):
"""Add vector to appropriate partition"""
partition_key = self.get_partition_key(vector, partition_method)
self.partitions[partition_key].append(vector)
self.metadata_partitions[partition_key].append(metadata or {})
self.partition_stats[partition_key] += 1
return partition_key, len(self.partitions[partition_key]) - 1
def get_partition_info(self):
"""Get statistics about partitions"""
stats = {}
for partition_id in range(self.num_partitions):
count = self.partition_stats[partition_id]
stats[partition_id] = {
'count': count,
'memory_mb': count * self.dimension * 4 / (1024 * 1024) # Assuming float32
}
return stats
def search_partition(self, partition_id, query_vector, k=5):
"""Search within a specific partition"""
if partition_id not in self.partitions:
return [], []
partition_vectors = np.array(self.partitions[partition_id])
if len(partition_vectors) == 0:
return [], []
# Calculate cosine similarities
from sklearn.metrics.pairwise import cosine_similarity
similarities = cosine_similarity([query_vector], partition_vectors)[0]
# Get top-k indices
top_indices = similarities.argsort()[-k:][::-1]
# Return results with metadata
results = []
scores = []
for idx in top_indices:
if idx < len(self.metadata_partitions[partition_id]):
results.append(self.metadata_partitions[partition_id][idx])
scores.append(similarities[idx])
return results, scores
# Example usage
partitioned_storage = PartitionedVectorStorage(dimension=128, num_partitions=8)
# Add vectors to partitions
vectors = np.random.random((1000, 128))
for i, vector in enumerate(vectors):
metadata = {'id': i, 'value': f'item_{i}'}
partition_id, local_idx = partitioned_storage.add_vector(vector, metadata)
# Check partition distribution
stats = partitioned_storage.get_partition_info()
print("Partition Statistics:")
for pid, stat in stats.items():
print(f" Partition {pid}: {stat['count']} vectors, {stat['memory_mb']:.2f} MB")
# Search within partitions
query = np.random.random(128)
for partition_id in range(3): # Search first 3 partitions
results, scores = partitioned_storage.search_partition(partition_id, query, k=3)
print(f"Partition {partition_id} top results: {len(results)} items")3. Hierarchical Storage Pattern
Multi-level storage with different performance characteristics.
python
import time
from abc import ABC, abstractmethod
class StorageTier(ABC):
@abstractmethod
def store(self, key, data):
pass
@abstractmethod
def retrieve(self, key):
pass
@abstractmethod
def exists(self, key):
pass
class MemoryTier(StorageTier):
"""Fast in-memory storage tier"""
def __init__(self, max_size=1000):
self.data = {}
self.access_times = {}
self.max_size = max_size
def store(self, key, data):
if len(self.data) >= self.max_size:
self._evict_lru()
self.data[key] = data
self.access_times[key] = time.time()
def retrieve(self, key):
if key in self.data:
self.access_times[key] = time.time()
return self.data[key]
return None
def exists(self, key):
return key in self.data
def _evict_lru(self):
"""Evict least recently used item"""
if not self.data:
return
lru_key = min(self.access_times.keys(), key=lambda k: self.access_times[k])
del self.data[lru_key]
del self.access_times[lru_key]
class DiskTier(StorageTier):
"""Slower disk-based storage tier"""
def __init__(self, base_path='./disk_storage'):
import os
self.base_path = base_path
os.makedirs(base_path, exist_ok=True)
def store(self, key, data):
filepath = f"{self.base_path}/{key}.npy"
np.save(filepath, data)
def retrieve(self, key):
import os
filepath = f"{self.base_path}/{key}.npy"
if os.path.exists(filepath):
return np.load(filepath)
return None
def exists(self, key):
import os
filepath = f"{self.base_path}/{key}.npy"
return os.path.exists(filepath)
class HierarchicalVectorStorage:
def __init__(self, memory_size=100):
self.memory_tier = MemoryTier(max_size=memory_size)
self.disk_tier = DiskTier()
self.stats = {
'memory_hits': 0,
'disk_hits': 0,
'misses': 0
}
def store_vector(self, vector_id, vector, metadata=None):
"""Store vector with automatic tiering"""
data = {
'vector': vector,
'metadata': metadata or {},
'timestamp': time.time()
}
# Always store in memory first
self.memory_tier.store(vector_id, data)
# Also store on disk for persistence
self.disk_tier.store(vector_id, data)
def retrieve_vector(self, vector_id):
"""Retrieve vector with automatic tier promotion"""
# Try memory first
data = self.memory_tier.retrieve(vector_id)
if data is not None:
self.stats['memory_hits'] += 1
return data
# Try disk second
data = self.disk_tier.retrieve(vector_id)
if data is not None:
self.stats['disk_hits'] += 1
# Promote to memory
self.memory_tier.store(vector_id, data)
return data
# Not found
self.stats['misses'] += 1
return None
def get_stats(self):
"""Get performance statistics"""
total_requests = sum(self.stats.values())
if total_requests == 0:
return self.stats
return {
**self.stats,
'memory_hit_rate': self.stats['memory_hits'] / total_requests,
'total_hit_rate': (self.stats['memory_hits'] + self.stats['disk_hits']) / total_requests
}
# Example usage
hierarchical_storage = HierarchicalVectorStorage(memory_size=50)
# Store vectors
for i in range(100):
vector = np.random.random(128)
metadata = {'id': i, 'type': 'test'}
hierarchical_storage.store_vector(f'vec_{i}', vector, metadata)
# Access patterns - some frequently, some rarely
frequently_accessed = ['vec_0', 'vec_1', 'vec_2']
rarely_accessed = ['vec_90', 'vec_91', 'vec_92']
# Simulate access patterns
for _ in range(10):
for vec_id in frequently_accessed:
data = hierarchical_storage.retrieve_vector(vec_id)
for vec_id in rarely_accessed:
data = hierarchical_storage.retrieve_vector(vec_id)
stats = hierarchical_storage.get_stats()
print("Storage Performance Stats:")
for key, value in stats.items():
if 'rate' in key:
print(f" {key}: {value:.2%}")
else:
print(f" {key}: {value}")Advanced Storage Patterns
4. Compressed Storage Pattern
Reduces storage requirements through various compression techniques.
python
class CompressedVectorStorage:
def __init__(self, dimension, compression_method='pca'):
self.dimension = dimension
self.compression_method = compression_method
self.compressed_vectors = []
self.metadata = []
self.compression_model = None
def train_compression(self, training_vectors, target_dimension=64):
"""Train compression model on sample data"""
if self.compression_method == 'pca':
from sklearn.decomposition import PCA
self.compression_model = PCA(n_components=target_dimension)
self.compression_model.fit(training_vectors)
self.compressed_dimension = target_dimension
elif self.compression_method == 'quantization':
# Simple scalar quantization
self.compression_model = {
'min_values': np.min(training_vectors, axis=0),
'max_values': np.max(training_vectors, axis=0)
}
self.compressed_dimension = self.dimension
print(f"Compression trained: {self.dimension} -> {self.compressed_dimension} dimensions")
if hasattr(self.compression_model, 'explained_variance_ratio_'):
total_variance = sum(self.compression_model.explained_variance_ratio_)
print(f"Explained variance: {total_variance:.2%}")
def compress_vector(self, vector):
"""Compress a single vector"""
if self.compression_method == 'pca':
return self.compression_model.transform([vector])[0]
elif self.compression_method == 'quantization':
# Quantize to 8-bit integers
min_vals = self.compression_model['min_values']
max_vals = self.compression_model['max_values']
# Normalize to 0-255 range
normalized = (vector - min_vals) / (max_vals - min_vals + 1e-8)
quantized = (normalized * 255).astype(np.uint8)
return quantized
def decompress_vector(self, compressed_vector):
"""Decompress a vector (approximate reconstruction)"""
if self.compression_method == 'pca':
return self.compression_model.inverse_transform([compressed_vector])[0]
elif self.compression_method == 'quantization':
# Dequantize from 8-bit integers
min_vals = self.compression_model['min_values']
max_vals = self.compression_model['max_values']
# Convert back to float range
normalized = compressed_vector.astype(np.float32) / 255.0
reconstructed = normalized * (max_vals - min_vals) + min_vals
return reconstructed
def add_vector(self, vector, metadata=None):
"""Add compressed vector"""
compressed = self.compress_vector(vector)
self.compressed_vectors.append(compressed)
self.metadata.append(metadata or {})
return len(self.compressed_vectors) - 1
def get_vector(self, index, decompress=True):
"""Retrieve vector, optionally decompressed"""
if 0 <= index < len(self.compressed_vectors):
compressed = self.compressed_vectors[index]
if decompress:
return self.decompress_vector(compressed), self.metadata[index]
else:
return compressed, self.metadata[index]
return None, None
def calculate_compression_ratio(self):
"""Calculate storage savings"""
original_size = self.dimension * 4 # float32
if self.compression_method == 'pca':
compressed_size = self.compressed_dimension * 4 # float32
elif self.compression_method == 'quantization':
compressed_size = self.dimension * 1 # uint8
ratio = original_size / compressed_size
savings = (1 - compressed_size / original_size) * 100
return {
'compression_ratio': ratio,
'space_savings_percent': savings,
'original_bytes': original_size,
'compressed_bytes': compressed_size
}
# Example usage
# Generate training data
training_data = np.random.random((1000, 128))
# PCA compression
pca_storage = CompressedVectorStorage(dimension=128, compression_method='pca')
pca_storage.train_compression(training_data, target_dimension=32)
# Add vectors
test_vectors = np.random.random((100, 128))
for i, vector in enumerate(test_vectors):
pca_storage.add_vector(vector, {'id': i})
# Check compression effectiveness
pca_stats = pca_storage.calculate_compression_ratio()
print("PCA Compression Stats:")
for key, value in pca_stats.items():
if 'percent' in key:
print(f" {key}: {value:.1f}%")
elif 'ratio' in key:
print(f" {key}: {value:.2f}x")
else:
print(f" {key}: {value}")
# Test reconstruction quality
original = test_vectors[0]
reconstructed, _ = pca_storage.get_vector(0, decompress=True)
reconstruction_error = np.mean((original - reconstructed) ** 2)
print(f"Reconstruction MSE: {reconstruction_error:.6f}")5. Distributed Storage Pattern
Spreads vectors across multiple nodes for scalability.
python
class DistributedVectorStorage:
def __init__(self, nodes):
"""
Initialize distributed storage
nodes: list of node identifiers/addresses
"""
self.nodes = nodes
self.node_storage = {node: {} for node in nodes}
self.hash_ring = self._create_hash_ring()
self.replication_factor = min(3, len(nodes)) # Replicate to up to 3 nodes
def _create_hash_ring(self):
"""Create consistent hash ring for node selection"""
import hashlib
ring = {}
# Add multiple virtual nodes per physical node for better distribution
virtual_nodes_per_node = 100
for node in self.nodes:
for i in range(virtual_nodes_per_node):
virtual_node = f"{node}:{i}"
hash_value = int(hashlib.md5(virtual_node.encode()).hexdigest(), 16)
ring[hash_value] = node
# Sort by hash value
self.sorted_hashes = sorted(ring.keys())
return ring
def _get_nodes_for_key(self, key):
"""Get nodes responsible for storing a key"""
import hashlib
key_hash = int(hashlib.md5(str(key).encode()).hexdigest(), 16)
# Find position in hash ring
idx = 0
for i, hash_val in enumerate(self.sorted_hashes):
if key_hash <= hash_val:
idx = i
break
# Get nodes for replication
selected_nodes = []
unique_nodes = set()
for i in range(len(self.sorted_hashes)):
ring_idx = (idx + i) % len(self.sorted_hashes)
hash_val = self.sorted_hashes[ring_idx]
node = self.hash_ring[hash_val]
if node not in unique_nodes:
selected_nodes.append(node)
unique_nodes.add(node)
if len(selected_nodes) >= self.replication_factor:
break
return selected_nodes
def store_vector(self, vector_id, vector, metadata=None):
"""Store vector across multiple nodes"""
nodes = self._get_nodes_for_key(vector_id)
data = {
'vector': vector,
'metadata': metadata or {},
'timestamp': time.time()
}
stored_nodes = []
for node in nodes:
try:
# In a real implementation, this would be a network call
self.node_storage[node][vector_id] = data
stored_nodes.append(node)
except Exception as e:
print(f"Failed to store on node {node}: {e}")
return {
'vector_id': vector_id,
'stored_on': stored_nodes,
'replication_level': len(stored_nodes)
}
def retrieve_vector(self, vector_id):
"""Retrieve vector from any available replica"""
nodes = self._get_nodes_for_key(vector_id)
for node in nodes:
try:
if vector_id in self.node_storage[node]:
return self.node_storage[node][vector_id]
except Exception as e:
print(f"Failed to retrieve from node {node}: {e}")
continue
return None
def get_cluster_stats(self):
"""Get statistics about the distributed storage"""
stats = {
'nodes': len(self.nodes),
'total_vectors': 0,
'node_distribution': {},
'replication_factor': self.replication_factor
}
for node in self.nodes:
count = len(self.node_storage[node])
stats['node_distribution'][node] = count
stats['total_vectors'] += count
# Calculate distribution balance
if self.nodes:
avg_per_node = stats['total_vectors'] / len(self.nodes)
max_deviation = max(abs(count - avg_per_node)
for count in stats['node_distribution'].values())
stats['load_balance_deviation'] = max_deviation / avg_per_node if avg_per_node > 0 else 0
return stats
# Example usage
# Simulate a 4-node cluster
nodes = ['node-1', 'node-2', 'node-3', 'node-4']
distributed_storage = DistributedVectorStorage(nodes)
# Store vectors
for i in range(100):
vector = np.random.random(128)
metadata = {'id': i, 'category': f'type_{i%5}'}
result = distributed_storage.store_vector(f'vec_{i}', vector, metadata)
# Check distribution
stats = distributed_storage.get_cluster_stats()
print("Distributed Storage Stats:")
print(f" Total vectors: {stats['total_vectors']}")
print(f" Replication factor: {stats['replication_factor']}")
print(f" Load balance deviation: {stats['load_balance_deviation']:.2%}")
print(" Node distribution:")
for node, count in stats['node_distribution'].items():
print(f" {node}: {count} vectors")
# Test retrieval
retrieved = distributed_storage.retrieve_vector('vec_0')
if retrieved:
print(f"Successfully retrieved vector with metadata: {retrieved['metadata']}")Storage Performance Optimization
Memory Management
python
class MemoryEfficientVectorStorage:
def __init__(self, dimension, batch_size=1000):
self.dimension = dimension
self.batch_size = batch_size
self.batches = []
self.current_batch = []
self.current_metadata = []
self.total_vectors = 0
def add_vector(self, vector, metadata=None):
"""Add vector with automatic batching"""
self.current_batch.append(vector)
self.current_metadata.append(metadata or {})
if len(self.current_batch) >= self.batch_size:
self._flush_batch()
self.total_vectors += 1
return self.total_vectors - 1
def _flush_batch(self):
"""Convert current batch to numpy array and store"""
if self.current_batch:
batch_array = np.array(self.current_batch, dtype=np.float32)
self.batches.append({
'vectors': batch_array,
'metadata': self.current_metadata.copy()
})
# Clear current batch
self.current_batch = []
self.current_metadata = []
def get_vector(self, index):
"""Retrieve vector by global index"""
if index >= self.total_vectors:
return None, None
# Find which batch contains this index
batch_idx = index // self.batch_size
local_idx = index % self.batch_size
# Handle current unflushed batch
if batch_idx >= len(self.batches):
if local_idx < len(self.current_batch):
return self.current_batch[local_idx], self.current_metadata[local_idx]
return None, None
# Get from flushed batch
batch = self.batches[batch_idx]
if local_idx < len(batch['vectors']):
return batch['vectors'][local_idx], batch['metadata'][local_idx]
return None, None
def finalize(self):
"""Flush any remaining vectors and optimize storage"""
self._flush_batch()
# Optional: consolidate small batches
self._consolidate_small_batches()
def _consolidate_small_batches(self, min_size=None):
"""Merge small batches to improve memory efficiency"""
if min_size is None:
min_size = self.batch_size // 2
new_batches = []
accumulated_vectors = []
accumulated_metadata = []
for batch in self.batches:
vectors = batch['vectors']
metadata = batch['metadata']
if len(vectors) < min_size and accumulated_vectors:
# Add to accumulator
accumulated_vectors.extend(vectors)
accumulated_metadata.extend(metadata)
# Check if accumulator is big enough
if len(accumulated_vectors) >= self.batch_size:
# Flush accumulator
new_batches.append({
'vectors': np.array(accumulated_vectors, dtype=np.float32),
'metadata': accumulated_metadata
})
accumulated_vectors = []
accumulated_metadata = []
else:
# Flush accumulator if exists
if accumulated_vectors:
new_batches.append({
'vectors': np.array(accumulated_vectors, dtype=np.float32),
'metadata': accumulated_metadata
})
accumulated_vectors = []
accumulated_metadata = []
# Add current batch
new_batches.append(batch)
# Handle any remaining accumulated items
if accumulated_vectors:
new_batches.append({
'vectors': np.array(accumulated_vectors, dtype=np.float32),
'metadata': accumulated_metadata
})
self.batches = new_batches
def get_memory_usage(self):
"""Calculate memory usage statistics"""
total_vectors = sum(len(batch['vectors']) for batch in self.batches)
total_vectors += len(self.current_batch)
vector_memory = total_vectors * self.dimension * 4 # float32
metadata_memory = len(str(self.current_metadata)) * len(self.batches) # Rough estimate
return {
'total_vectors': total_vectors,
'vector_memory_mb': vector_memory / (1024 * 1024),
'metadata_memory_mb': metadata_memory / (1024 * 1024),
'num_batches': len(self.batches),
'avg_batch_size': total_vectors / len(self.batches) if self.batches else 0
}
# Example usage
efficient_storage = MemoryEfficientVectorStorage(dimension=128, batch_size=500)
# Add many vectors
for i in range(2500):
vector = np.random.random(128)
metadata = {'id': i, 'batch': i // 500}
efficient_storage.add_vector(vector, metadata)
# Finalize storage
efficient_storage.finalize()
# Check memory usage
memory_stats = efficient_storage.get_memory_usage()
print("Memory-Efficient Storage Stats:")
for key, value in memory_stats.items():
if 'mb' in key:
print(f" {key}: {value:.2f} MB")
else:
print(f" {key}: {value}")🎯 Key Takeaways
Storage Pattern Selection
- Flat Storage: Best for small datasets and simple use cases
- Partitioned Storage: Good for medium-scale applications with known access patterns
- Hierarchical Storage: Ideal for applications with mixed hot/cold data access
- Compressed Storage: Essential for large-scale deployments with storage constraints
- Distributed Storage: Required for massive scale and high availability
Performance Considerations
- Memory vs Disk: Balance between access speed and storage cost
- Compression Trade-offs: Storage savings vs computational overhead
- Replication: Availability vs storage cost
- Indexing: Query speed vs storage overhead
Scalability Strategies
- Horizontal Partitioning: Split data across multiple storage units
- Vertical Partitioning: Separate frequently and rarely accessed data
- Caching Layers: Use memory tiers for frequently accessed vectors
- Batch Processing: Optimize for bulk operations when possible
Next Steps:
- Similarity Search & Indexing: Learn efficient search algorithms for your stored vectors
- Advanced RAG Patterns: See how storage patterns enable advanced retrieval systems
- Production Deployment: Deploy vector storage systems in production