refactor(8.0)!: remove distributed clustering subsystem — inert/orphaned, scale is single-process + native provider
The distributed-clustering subsystem never ran in production: it was inert, orphaned dead code (faked consensus, stub replication, no live wiring, and it did not interoperate with the 8.0 Db API). Brainy 8.0 is a single-process library. Scale is single-process + the optional native provider (@soulcraft/cortex, on-disk DiskANN to 10B+ vectors) + per-tenant pools + horizontal read scaling (many reader processes, one writer). Removed: - src/distributed/ entirely (coordinator, shardManager, cacheSync, readWriteSeparation, queryPlanner, healthMonitor, configManager, hashPartitioner, shardMigration, domainDetector, storageDiscovery, http/network transports). ReaderMode/HybridMode relocated to src/storage/operationalModes.ts (slimmed to the live surface). - src/types/distributedTypes.ts; config.distributed field + JSDoc; coreTypes distributedConfig; memoryStorage distributedConfig persistence. - DistributedRole enum + src/config/distributedPresets.ts and the orphaned src/config/extensibleConfig.ts (config/augmentation registry built on removed cloud adapters + distributed presets), plus their src/index.ts re-exports. - 13 BRAINY_* cluster env vars; the storage setDistributedComponents hook; enableDistributedSearch (dead config flag); the metadata partition field; the distributed_ reserved key prefix. - Orphaned src/storage/readOnlyOptimizations.ts (zero importers). - Tests targeting the subsystem: distributed-demo, distributed-cluster helper, distributed-transactions, sharding-transactions. - Docs: EXTENDING_STORAGE.md (deleted); scrubbed distributed/cluster/Raft/ shard-manager/multi-node prose from v3-features, enterprise-for-everyone, augmentations-actual, complete-feature-list, vfs/README, vfs/ROADMAP, vfs/VFS_CORE, capacity-planning, transactions, MIGRATION-V3-TO-V4, storage-architecture; reframed scale prose to the 8.0 model. Kept: src/storage/sharding.ts (local-disk 256-bucket directory sharding via getShardIdFromUuid — used live by baseStorage, unrelated to clustering); the multi-process mode: 'reader' | 'writer' roles; semantic/HNSW clustering. RELEASES.md: added a removed-surfaces row documenting the cut and the 8.0 scale model.
This commit is contained in:
parent
f8e0079d3f
commit
00d3203d68
51 changed files with 153 additions and 9889 deletions
|
|
@ -1159,13 +1159,13 @@ export abstract class BaseStorageAdapter implements StorageAdapter {
|
|||
}
|
||||
|
||||
/**
|
||||
* Increment count for entity type - O(1) operation
|
||||
* Protected by storage-specific mechanisms (mutex, distributed consensus, etc.)
|
||||
* Increment count for entity type - O(1) operation.
|
||||
* Concurrency is handled by the process-global mutex
|
||||
* ({@link incrementEntityCountSafe}); this raw form is for callers that
|
||||
* already hold it.
|
||||
* @param type The entity type
|
||||
*/
|
||||
protected incrementEntityCount(type: string): void {
|
||||
// For distributed scenarios, this is aggregated across shards
|
||||
// For single-node, this is protected by storage-specific locking
|
||||
this.entityCounts.set(type, (this.entityCounts.get(type) || 0) + 1)
|
||||
this.totalNounCount++
|
||||
// Update cache
|
||||
|
|
@ -1176,11 +1176,11 @@ export abstract class BaseStorageAdapter implements StorageAdapter {
|
|||
}
|
||||
|
||||
/**
|
||||
* Thread-safe increment for concurrent scenarios
|
||||
* Uses mutex for single-node, distributed consensus for multi-node
|
||||
* Thread-safe increment for concurrent scenarios.
|
||||
* A process-global mutex serialises the read-modify-write so concurrent
|
||||
* writers in the same process cannot lose updates.
|
||||
*/
|
||||
protected async incrementEntityCountSafe(type: string): Promise<void> {
|
||||
// Single-node mutex protection (distributed mode handled by coordinator)
|
||||
const mutex = getGlobalMutex()
|
||||
await mutex.runExclusive(`count-entity-${type}`, async () => {
|
||||
this.incrementEntityCount(type)
|
||||
|
|
@ -1225,8 +1225,10 @@ export abstract class BaseStorageAdapter implements StorageAdapter {
|
|||
}
|
||||
|
||||
/**
|
||||
* Increment verb count - O(1) operation (now synchronous)
|
||||
* Protected by storage-specific mechanisms (mutex, distributed consensus, etc.)
|
||||
* Increment verb count - O(1) operation (now synchronous).
|
||||
* Concurrency is handled by the process-global mutex
|
||||
* ({@link incrementVerbCountSafe}); this raw form is for callers that already
|
||||
* hold it.
|
||||
* @param type The verb type
|
||||
*/
|
||||
protected incrementVerbCount(type: string): void {
|
||||
|
|
@ -1240,8 +1242,9 @@ export abstract class BaseStorageAdapter implements StorageAdapter {
|
|||
}
|
||||
|
||||
/**
|
||||
* Thread-safe increment for verb counts
|
||||
* Uses mutex for single-node, distributed consensus for multi-node
|
||||
* Thread-safe increment for verb counts.
|
||||
* A process-global mutex serialises the read-modify-write so concurrent
|
||||
* writers in the same process cannot lose updates.
|
||||
* @param type The verb type
|
||||
*/
|
||||
protected async incrementVerbCountSafe(type: string): Promise<void> {
|
||||
|
|
|
|||
|
|
@ -407,13 +407,9 @@ export class MemoryStorage extends BaseStorage {
|
|||
// Include services if present
|
||||
...(statistics.services && {
|
||||
services: statistics.services.map(s => ({...s}))
|
||||
}),
|
||||
// Include distributedConfig if present
|
||||
...(statistics.distributedConfig && {
|
||||
distributedConfig: JSON.parse(JSON.stringify(statistics.distributedConfig))
|
||||
})
|
||||
}
|
||||
|
||||
|
||||
// Since this is in-memory, there's no need for time-based partitioning
|
||||
// or legacy file handling
|
||||
}
|
||||
|
|
@ -459,10 +455,6 @@ export class MemoryStorage extends BaseStorage {
|
|||
// Include services if present
|
||||
...(this.statistics.services && {
|
||||
services: this.statistics.services.map(s => ({...s}))
|
||||
}),
|
||||
// Include distributedConfig if present
|
||||
...(this.statistics.distributedConfig && {
|
||||
distributedConfig: JSON.parse(JSON.stringify(this.statistics.distributedConfig))
|
||||
})
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -144,7 +144,6 @@ const SINGLETON_SYSTEM_KEYS = new Set([
|
|||
|
||||
const SINGLETON_SYSTEM_PREFIXES = [
|
||||
'statistics_',
|
||||
'distributed_',
|
||||
]
|
||||
|
||||
function isSingletonSystemKey(key: string): boolean {
|
||||
|
|
@ -2292,11 +2291,11 @@ export abstract class BaseStorage extends BaseStorageAdapter {
|
|||
await this.ensureInitialized()
|
||||
const keyInfo = this.analyzeKey(id, 'system')
|
||||
|
||||
// Try new distributed path first
|
||||
// Try the new shard-prefixed path first
|
||||
const data = await this.readCanonicalObject(keyInfo.fullPath)
|
||||
if (data !== null) return data
|
||||
|
||||
// Backward compat: if key was distributed, fall back to legacy flat path
|
||||
// Backward compat: if the key was sharded, fall back to the legacy flat path
|
||||
if (keyInfo.shardId !== null) {
|
||||
const legacyPath = `${SYSTEM_DIR}/${id}.json`
|
||||
return this.readCanonicalObject(legacyPath)
|
||||
|
|
@ -2428,10 +2427,9 @@ export abstract class BaseStorage extends BaseStorageAdapter {
|
|||
* @returns Metadata or null if not found
|
||||
*
|
||||
* @performance
|
||||
* - O(1) direct ID lookup - always 1 read (~500ms on cloud, ~10ms local)
|
||||
* - O(1) direct ID lookup - always 1 read (~10ms on local disk)
|
||||
* - No caching complexity
|
||||
* - No type search fallbacks
|
||||
* - Works in distributed systems without sync issues
|
||||
*
|
||||
* Type-first paths (removed)
|
||||
* Promoted to fast path for brain.get() optimization
|
||||
|
|
|
|||
73
src/storage/operationalModes.ts
Normal file
73
src/storage/operationalModes.ts
Normal file
|
|
@ -0,0 +1,73 @@
|
|||
/**
|
||||
* @module storage/operationalModes
|
||||
* @description Operational modes that gate which operations a Brainy instance may
|
||||
* perform. Brainy's multi-process model runs one writer process plus any number of
|
||||
* reader processes against a single shared on-disk store; the mode object is the
|
||||
* in-process guard that enforces that contract.
|
||||
*
|
||||
* - {@link HybridMode} (the default, used by writer instances) permits reads,
|
||||
* writes, and deletes — a writer must be able to read its own data.
|
||||
* - {@link ReaderMode} (used by `mode: 'reader'` / `Brainy.openReadOnly()` and by
|
||||
* `asOf()` historical snapshots) permits reads only; every mutation throws.
|
||||
*
|
||||
* `validateOperation()` is called at the top of each mutation path, and the
|
||||
* `canWrite` flag drives the public read-only checks on the instance.
|
||||
*/
|
||||
|
||||
/**
|
||||
* @description Base class for operational modes. Concrete modes declare which of
|
||||
* read/write/delete they permit; {@link BaseOperationalMode.validateOperation}
|
||||
* turns a disallowed operation into an explicit error.
|
||||
*/
|
||||
export abstract class BaseOperationalMode {
|
||||
abstract canRead: boolean
|
||||
abstract canWrite: boolean
|
||||
abstract canDelete: boolean
|
||||
|
||||
/**
|
||||
* @description Throw if the requested operation is not allowed in this mode.
|
||||
* Called at the top of every mutation method so read-only instances fail loudly
|
||||
* rather than silently dropping writes.
|
||||
* @param operation - The operation being attempted.
|
||||
* @throws {Error} When the mode does not permit the operation.
|
||||
*/
|
||||
validateOperation(operation: 'read' | 'write' | 'delete'): void {
|
||||
switch (operation) {
|
||||
case 'read':
|
||||
if (!this.canRead) {
|
||||
throw new Error('Read operations are not allowed in write-only mode')
|
||||
}
|
||||
break
|
||||
case 'write':
|
||||
if (!this.canWrite) {
|
||||
throw new Error('Write operations are not allowed in read-only mode')
|
||||
}
|
||||
break
|
||||
case 'delete':
|
||||
if (!this.canDelete) {
|
||||
throw new Error('Delete operations are not allowed in this mode')
|
||||
}
|
||||
break
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* @description Read-only mode. Reads are permitted; writes and deletes throw.
|
||||
* Used by reader processes, `Brainy.openReadOnly()`, and historical snapshots.
|
||||
*/
|
||||
export class ReaderMode extends BaseOperationalMode {
|
||||
canRead = true
|
||||
canWrite = false
|
||||
canDelete = false
|
||||
}
|
||||
|
||||
/**
|
||||
* @description Read-write mode and the default for writer instances. Permits
|
||||
* reads, writes, and deletes — a writer needs to read its own data.
|
||||
*/
|
||||
export class HybridMode extends BaseOperationalMode {
|
||||
canRead = true
|
||||
canWrite = true
|
||||
canDelete = true
|
||||
}
|
||||
|
|
@ -1,530 +0,0 @@
|
|||
/**
|
||||
* Read-Only Storage Optimizations for Production Deployments
|
||||
* Implements compression, memory-mapping, and pre-built index segments
|
||||
*/
|
||||
|
||||
import { HNSWNoun, Vector } from '../coreTypes.js'
|
||||
|
||||
// Compression types supported
|
||||
enum CompressionType {
|
||||
NONE = 'none',
|
||||
GZIP = 'gzip',
|
||||
QUANTIZATION = 'quantization',
|
||||
HYBRID = 'hybrid'
|
||||
// BROTLI removed - was not actually implemented
|
||||
}
|
||||
|
||||
// Vector quantization methods
|
||||
enum QuantizationType {
|
||||
SCALAR = 'scalar', // 8-bit scalar quantization
|
||||
PRODUCT = 'product', // Product quantization
|
||||
BINARY = 'binary' // Binary quantization
|
||||
}
|
||||
|
||||
interface CompressionConfig {
|
||||
vectorCompression: CompressionType
|
||||
metadataCompression: CompressionType
|
||||
quantizationType?: QuantizationType
|
||||
quantizationBits?: number
|
||||
compressionLevel?: number
|
||||
}
|
||||
|
||||
interface ReadOnlyConfig {
|
||||
prebuiltIndexPath?: string
|
||||
memoryMapped?: boolean
|
||||
compression: CompressionConfig
|
||||
segmentSize?: number // For index segmentation
|
||||
prefetchSegments?: number
|
||||
cacheIndexInMemory?: boolean
|
||||
}
|
||||
|
||||
interface IndexSegment {
|
||||
id: string
|
||||
nodeCount: number
|
||||
vectorDimension: number
|
||||
compression: CompressionType
|
||||
s3Key?: string
|
||||
localPath?: string
|
||||
loadedInMemory: boolean
|
||||
lastAccessed: number
|
||||
}
|
||||
|
||||
/**
|
||||
* Read-only storage optimizations for high-performance production deployments
|
||||
*/
|
||||
export class ReadOnlyOptimizations {
|
||||
private config: Required<ReadOnlyConfig>
|
||||
private segments: Map<string, IndexSegment> = new Map()
|
||||
private compressionStats = {
|
||||
originalSize: 0,
|
||||
compressedSize: 0,
|
||||
compressionRatio: 0,
|
||||
decompressionTime: 0
|
||||
}
|
||||
|
||||
// Quantization codebooks for vector compression
|
||||
private quantizationCodebooks: Map<string, Float32Array> = new Map()
|
||||
|
||||
// Memory-mapped buffers for large datasets
|
||||
private memoryMappedBuffers: Map<string, ArrayBuffer> = new Map()
|
||||
|
||||
constructor(config: Partial<ReadOnlyConfig> = {}) {
|
||||
this.config = {
|
||||
prebuiltIndexPath: '',
|
||||
memoryMapped: true,
|
||||
compression: {
|
||||
vectorCompression: CompressionType.QUANTIZATION,
|
||||
metadataCompression: CompressionType.GZIP,
|
||||
quantizationType: QuantizationType.SCALAR,
|
||||
quantizationBits: 8,
|
||||
compressionLevel: 6
|
||||
},
|
||||
segmentSize: 10000, // 10k nodes per segment
|
||||
prefetchSegments: 3,
|
||||
cacheIndexInMemory: false,
|
||||
...config
|
||||
}
|
||||
|
||||
if (config.compression) {
|
||||
this.config.compression = { ...this.config.compression, ...config.compression }
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Compress vector data using specified compression method
|
||||
*/
|
||||
public async compressVector(vector: Vector, segmentId: string): Promise<ArrayBuffer> {
|
||||
const startTime = Date.now()
|
||||
let compressedData: ArrayBuffer
|
||||
|
||||
switch (this.config.compression.vectorCompression) {
|
||||
case CompressionType.QUANTIZATION:
|
||||
compressedData = await this.quantizeVector(vector, segmentId)
|
||||
break
|
||||
|
||||
case CompressionType.GZIP:
|
||||
const gzipBuffer = new Float32Array(vector).buffer
|
||||
compressedData = await this.gzipCompress(gzipBuffer.slice(0))
|
||||
break
|
||||
|
||||
// Brotli removed - was not implemented
|
||||
|
||||
case CompressionType.HYBRID:
|
||||
// First quantize, then compress
|
||||
const quantized = await this.quantizeVector(vector, segmentId)
|
||||
compressedData = await this.gzipCompress(quantized)
|
||||
break
|
||||
|
||||
default:
|
||||
const defaultBuffer = new Float32Array(vector).buffer
|
||||
compressedData = defaultBuffer.slice(0)
|
||||
break
|
||||
}
|
||||
|
||||
// Update compression statistics
|
||||
const originalSize = vector.length * 4 // 4 bytes per float32
|
||||
this.compressionStats.originalSize += originalSize
|
||||
this.compressionStats.compressedSize += compressedData.byteLength
|
||||
this.compressionStats.decompressionTime += Date.now() - startTime
|
||||
|
||||
this.updateCompressionRatio()
|
||||
|
||||
return compressedData
|
||||
}
|
||||
|
||||
/**
|
||||
* Decompress vector data
|
||||
*/
|
||||
public async decompressVector(
|
||||
compressedData: ArrayBuffer,
|
||||
segmentId: string,
|
||||
originalDimension: number
|
||||
): Promise<Vector> {
|
||||
switch (this.config.compression.vectorCompression) {
|
||||
case CompressionType.QUANTIZATION:
|
||||
return this.dequantizeVector(compressedData, segmentId, originalDimension)
|
||||
|
||||
case CompressionType.GZIP:
|
||||
const gzipDecompressed = await this.gzipDecompress(compressedData)
|
||||
return Array.from(new Float32Array(gzipDecompressed))
|
||||
|
||||
// Brotli removed - was not implemented
|
||||
|
||||
case CompressionType.HYBRID:
|
||||
const gzipStage = await this.gzipDecompress(compressedData)
|
||||
return this.dequantizeVector(gzipStage, segmentId, originalDimension)
|
||||
|
||||
default:
|
||||
return Array.from(new Float32Array(compressedData))
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Scalar quantization of vectors to 8-bit integers
|
||||
*/
|
||||
private async quantizeVector(vector: Vector, segmentId: string): Promise<ArrayBuffer> {
|
||||
let codebook = this.quantizationCodebooks.get(segmentId)
|
||||
|
||||
if (!codebook) {
|
||||
// Create codebook (min/max values for scaling)
|
||||
const min = Math.min(...vector)
|
||||
const max = Math.max(...vector)
|
||||
codebook = new Float32Array([min, max])
|
||||
this.quantizationCodebooks.set(segmentId, codebook)
|
||||
}
|
||||
|
||||
const [min, max] = codebook
|
||||
const scale = (max - min) / 255 // 8-bit quantization
|
||||
|
||||
const quantized = new Uint8Array(vector.length)
|
||||
for (let i = 0; i < vector.length; i++) {
|
||||
quantized[i] = Math.round((vector[i] - min) / scale)
|
||||
}
|
||||
|
||||
// Store codebook with quantized data
|
||||
const result = new ArrayBuffer(quantized.byteLength + codebook.byteLength)
|
||||
const resultView = new Uint8Array(result)
|
||||
|
||||
// First 8 bytes: codebook (min, max as float32)
|
||||
resultView.set(new Uint8Array(codebook.buffer), 0)
|
||||
// Remaining bytes: quantized vector
|
||||
resultView.set(quantized, codebook.byteLength)
|
||||
|
||||
return result
|
||||
}
|
||||
|
||||
/**
|
||||
* Dequantize 8-bit vectors back to float32
|
||||
*/
|
||||
private dequantizeVector(
|
||||
quantizedData: ArrayBuffer,
|
||||
segmentId: string,
|
||||
dimension: number
|
||||
): Vector {
|
||||
const dataView = new Uint8Array(quantizedData)
|
||||
|
||||
// Extract codebook (first 8 bytes)
|
||||
const codebookBytes = dataView.slice(0, 8)
|
||||
const codebook = new Float32Array(codebookBytes.buffer)
|
||||
const [min, max] = codebook
|
||||
|
||||
// Extract quantized vector
|
||||
const quantized = dataView.slice(8)
|
||||
const scale = (max - min) / 255
|
||||
|
||||
const result: Vector = []
|
||||
for (let i = 0; i < dimension; i++) {
|
||||
result[i] = min + quantized[i] * scale
|
||||
}
|
||||
|
||||
return result
|
||||
}
|
||||
|
||||
/**
|
||||
* GZIP compression using browser/Node.js APIs
|
||||
*/
|
||||
private async gzipCompress(data: ArrayBuffer): Promise<ArrayBuffer> {
|
||||
if (typeof CompressionStream !== 'undefined') {
|
||||
// Browser environment
|
||||
const stream = new CompressionStream('gzip')
|
||||
const writer = stream.writable.getWriter()
|
||||
const reader = stream.readable.getReader()
|
||||
|
||||
writer.write(new Uint8Array(data))
|
||||
writer.close()
|
||||
|
||||
const chunks: Uint8Array[] = []
|
||||
let result = await reader.read()
|
||||
|
||||
while (!result.done) {
|
||||
chunks.push(result.value)
|
||||
result = await reader.read()
|
||||
}
|
||||
|
||||
// Combine chunks
|
||||
const totalLength = chunks.reduce((sum, chunk) => sum + chunk.length, 0)
|
||||
const combined = new Uint8Array(totalLength)
|
||||
let offset = 0
|
||||
|
||||
for (const chunk of chunks) {
|
||||
combined.set(chunk, offset)
|
||||
offset += chunk.length
|
||||
}
|
||||
|
||||
return combined.buffer
|
||||
} else {
|
||||
// Node.js environment - would use zlib
|
||||
console.warn('GZIP compression not available, returning original data')
|
||||
return data
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* GZIP decompression
|
||||
*/
|
||||
private async gzipDecompress(compressedData: ArrayBuffer): Promise<ArrayBuffer> {
|
||||
if (typeof DecompressionStream !== 'undefined') {
|
||||
// Browser environment
|
||||
const stream = new DecompressionStream('gzip')
|
||||
const writer = stream.writable.getWriter()
|
||||
const reader = stream.readable.getReader()
|
||||
|
||||
writer.write(new Uint8Array(compressedData))
|
||||
writer.close()
|
||||
|
||||
const chunks: Uint8Array[] = []
|
||||
let result = await reader.read()
|
||||
|
||||
while (!result.done) {
|
||||
chunks.push(result.value)
|
||||
result = await reader.read()
|
||||
}
|
||||
|
||||
// Combine chunks
|
||||
const totalLength = chunks.reduce((sum, chunk) => sum + chunk.length, 0)
|
||||
const combined = new Uint8Array(totalLength)
|
||||
let offset = 0
|
||||
|
||||
for (const chunk of chunks) {
|
||||
combined.set(chunk, offset)
|
||||
offset += chunk.length
|
||||
}
|
||||
|
||||
return combined.buffer
|
||||
} else {
|
||||
console.warn('GZIP decompression not available, returning original data')
|
||||
return compressedData
|
||||
}
|
||||
}
|
||||
|
||||
// Brotli methods removed - were not implemented
|
||||
|
||||
/**
|
||||
* Create prebuilt index segments for faster loading
|
||||
*/
|
||||
public async createPrebuiltSegments(
|
||||
nodes: HNSWNoun[],
|
||||
outputPath: string
|
||||
): Promise<IndexSegment[]> {
|
||||
const segments: IndexSegment[] = []
|
||||
const segmentSize = this.config.segmentSize
|
||||
|
||||
console.log(`Creating ${Math.ceil(nodes.length / segmentSize)} prebuilt segments`)
|
||||
|
||||
for (let i = 0; i < nodes.length; i += segmentSize) {
|
||||
const segmentNodes = nodes.slice(i, i + segmentSize)
|
||||
const segmentId = `segment_${Math.floor(i / segmentSize)}`
|
||||
|
||||
const segment: IndexSegment = {
|
||||
id: segmentId,
|
||||
nodeCount: segmentNodes.length,
|
||||
vectorDimension: segmentNodes[0]?.vector.length || 0,
|
||||
compression: this.config.compression.vectorCompression,
|
||||
localPath: `${outputPath}/${segmentId}.dat`,
|
||||
loadedInMemory: false,
|
||||
lastAccessed: 0
|
||||
}
|
||||
|
||||
// Compress and serialize segment data
|
||||
const compressedData = await this.compressSegment(segmentNodes)
|
||||
|
||||
// In a real implementation, you would write this to disk/S3
|
||||
console.log(`Created segment ${segmentId} with ${compressedData.byteLength} bytes`)
|
||||
|
||||
segments.push(segment)
|
||||
this.segments.set(segmentId, segment)
|
||||
}
|
||||
|
||||
return segments
|
||||
}
|
||||
|
||||
/**
|
||||
* Compress an entire segment of nodes
|
||||
*/
|
||||
private async compressSegment(nodes: HNSWNoun[]): Promise<ArrayBuffer> {
|
||||
const serialized = JSON.stringify(nodes.map(node => ({
|
||||
id: node.id,
|
||||
vector: node.vector,
|
||||
connections: this.serializeConnections(node.connections)
|
||||
})))
|
||||
|
||||
const encoder = new TextEncoder()
|
||||
const data = encoder.encode(serialized)
|
||||
|
||||
// Apply metadata compression
|
||||
switch (this.config.compression.metadataCompression) {
|
||||
case CompressionType.GZIP:
|
||||
return this.gzipCompress(data.buffer.slice(0) as ArrayBuffer)
|
||||
// Brotli removed - was not implemented
|
||||
default:
|
||||
return data.buffer.slice(0) as ArrayBuffer
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Load a segment from storage with caching
|
||||
*/
|
||||
public async loadSegment(segmentId: string): Promise<HNSWNoun[]> {
|
||||
const segment = this.segments.get(segmentId)
|
||||
if (!segment) {
|
||||
throw new Error(`Segment ${segmentId} not found`)
|
||||
}
|
||||
|
||||
segment.lastAccessed = Date.now()
|
||||
|
||||
// Check if segment is already loaded in memory
|
||||
if (segment.loadedInMemory && this.memoryMappedBuffers.has(segmentId)) {
|
||||
return this.deserializeSegment(this.memoryMappedBuffers.get(segmentId)!)
|
||||
}
|
||||
|
||||
// Load from storage (S3, disk, etc.)
|
||||
const compressedData = await this.loadSegmentFromStorage(segment)
|
||||
|
||||
// Cache in memory if configured
|
||||
if (this.config.cacheIndexInMemory) {
|
||||
this.memoryMappedBuffers.set(segmentId, compressedData)
|
||||
segment.loadedInMemory = true
|
||||
}
|
||||
|
||||
return this.deserializeSegment(compressedData)
|
||||
}
|
||||
|
||||
/**
|
||||
* Load segment data from storage
|
||||
*/
|
||||
private async loadSegmentFromStorage(segment: IndexSegment): Promise<ArrayBuffer> {
|
||||
// Load segment from memory-mapped buffer if available
|
||||
const cached = this.memoryMappedBuffers.get(segment.id)
|
||||
if (cached) {
|
||||
return cached
|
||||
}
|
||||
|
||||
// This feature requires actual storage backend integration (S3, file system, etc)
|
||||
// Return empty buffer as this is an optional optimization feature
|
||||
console.warn(`Segment loading optimization not available for segment ${segment.id}. Using standard storage.`)
|
||||
return new ArrayBuffer(0)
|
||||
}
|
||||
|
||||
/**
|
||||
* Deserialize and decompress segment data
|
||||
*/
|
||||
private async deserializeSegment(compressedData: ArrayBuffer): Promise<HNSWNoun[]> {
|
||||
// Decompress metadata
|
||||
let decompressed: ArrayBuffer
|
||||
|
||||
switch (this.config.compression.metadataCompression) {
|
||||
case CompressionType.GZIP:
|
||||
decompressed = await this.gzipDecompress(compressedData)
|
||||
break
|
||||
// Brotli removed - was not implemented
|
||||
default:
|
||||
decompressed = compressedData
|
||||
break
|
||||
}
|
||||
|
||||
// Parse JSON
|
||||
const decoder = new TextDecoder()
|
||||
const jsonStr = decoder.decode(decompressed)
|
||||
const parsed = JSON.parse(jsonStr)
|
||||
|
||||
// Reconstruct HNSWNoun objects
|
||||
return parsed.map((item: any) => ({
|
||||
id: item.id,
|
||||
vector: item.vector,
|
||||
connections: this.deserializeConnections(item.connections)
|
||||
}))
|
||||
}
|
||||
|
||||
/**
|
||||
* Serialize connections Map for storage
|
||||
*/
|
||||
private serializeConnections(connections: Map<number, Set<string>>): Record<string, string[]> {
|
||||
const result: Record<string, string[]> = {}
|
||||
for (const [level, nodeIds] of connections.entries()) {
|
||||
result[level.toString()] = Array.from(nodeIds)
|
||||
}
|
||||
return result
|
||||
}
|
||||
|
||||
/**
|
||||
* Deserialize connections from storage format
|
||||
*/
|
||||
private deserializeConnections(serialized: Record<string, string[]>): Map<number, Set<string>> {
|
||||
const result = new Map<number, Set<string>>()
|
||||
for (const [levelStr, nodeIds] of Object.entries(serialized)) {
|
||||
result.set(parseInt(levelStr), new Set(nodeIds))
|
||||
}
|
||||
return result
|
||||
}
|
||||
|
||||
/**
|
||||
* Prefetch segments based on access patterns
|
||||
*/
|
||||
public async prefetchSegments(currentSegmentId: string): Promise<void> {
|
||||
const segment = this.segments.get(currentSegmentId)
|
||||
if (!segment) return
|
||||
|
||||
// Simple prefetching strategy - load adjacent segments
|
||||
const segmentNumber = parseInt(currentSegmentId.split('_')[1])
|
||||
const toPrefetch: string[] = []
|
||||
|
||||
for (let i = 1; i <= this.config.prefetchSegments; i++) {
|
||||
const nextId = `segment_${segmentNumber + i}`
|
||||
const prevId = `segment_${segmentNumber - i}`
|
||||
|
||||
if (this.segments.has(nextId) && !this.memoryMappedBuffers.has(nextId)) {
|
||||
toPrefetch.push(nextId)
|
||||
}
|
||||
if (this.segments.has(prevId) && !this.memoryMappedBuffers.has(prevId)) {
|
||||
toPrefetch.push(prevId)
|
||||
}
|
||||
}
|
||||
|
||||
// Prefetch in background
|
||||
for (const segmentId of toPrefetch) {
|
||||
this.loadSegment(segmentId).catch(error => {
|
||||
console.warn(`Failed to prefetch segment ${segmentId}:`, error)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Update compression statistics
|
||||
*/
|
||||
private updateCompressionRatio(): void {
|
||||
if (this.compressionStats.originalSize > 0) {
|
||||
this.compressionStats.compressionRatio =
|
||||
this.compressionStats.compressedSize / this.compressionStats.originalSize
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Get compression statistics
|
||||
*/
|
||||
public getCompressionStats(): typeof this.compressionStats & {
|
||||
segmentCount: number
|
||||
memoryUsage: number
|
||||
} {
|
||||
const memoryUsage = Array.from(this.memoryMappedBuffers.values())
|
||||
.reduce((sum, buffer) => sum + buffer.byteLength, 0)
|
||||
|
||||
return {
|
||||
...this.compressionStats,
|
||||
segmentCount: this.segments.size,
|
||||
memoryUsage
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Cleanup memory-mapped buffers
|
||||
*/
|
||||
public cleanup(): void {
|
||||
this.memoryMappedBuffers.clear()
|
||||
this.quantizationCodebooks.clear()
|
||||
|
||||
// Mark all segments as not loaded
|
||||
for (const segment of this.segments.values()) {
|
||||
segment.loadedInMemory = false
|
||||
}
|
||||
}
|
||||
}
|
||||
Loading…
Add table
Add a link
Reference in a new issue