From 46583f2d9b94d1e2702fcbff2d559a35272350ce Mon Sep 17 00:00:00 2001 From: David Snelling Date: Fri, 10 Apr 2026 11:22:19 -0700 Subject: [PATCH] feat: unified column store for filtering + sorting at billion scale MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds a per-field sorted column store (Lucene doc values + roaring bitmap architecture) that replaces the MetadataIndex sparse index internals for both filtering and sorting. One system for all field types with exact precision — no bucketing, no per-entity storage reads. Column store: binary .cidx segment format, in-memory tail buffers, LSM-style compaction, k-way merge sort, multi-value support (__words__). All queries (filter, range, sort, filtered sort) route through the column store when data is available, falling back to sparse index otherwise. Key unlocks: - find({ orderBy: 'createdAt' }) works WITHOUT a filter (previously threw) - find({ orderBy: 'metadata.price' }) works for custom numeric fields - Exact timestamp precision (no 1-minute bucketing) - O(K log S) sort independent of total entity count New files: src/indexes/columnStore/ (types, format, tail buffer, cursor, manifest, coordinator — ~700 lines). 101 new unit tests covering binary format round-trips, CRC validation, sort, filter, range, deletion, multi-segment merge, persistence, and multi-value (words) fields. Deleted: metadataIndex-automatic-bucketing.test.ts (bucketing behavior eliminated by exact-precision column store). Sparse index write path removed from addToIndex/removeFromIndex. Sparse index legacy code still present as dead code pending cleanup in next commit. --- src/brainy.ts | 34 +- src/indexes/columnStore/ColumnManifest.ts | 204 +++++ .../columnStore/ColumnSegmentCursor.ts | 271 +++++++ .../columnStore/ColumnSegmentFormat.ts | 548 ++++++++++++++ src/indexes/columnStore/ColumnStore.ts | 695 ++++++++++++++++++ src/indexes/columnStore/ColumnTailBuffer.ts | 169 +++++ src/indexes/columnStore/index.ts | 38 + src/indexes/columnStore/types.ts | 272 +++++++ src/utils/metadataIndex.ts | 417 +++++------ tests/integration/orderby-sort-bug.test.ts | 25 +- .../indexes/columnStore/column-store.test.ts | 295 ++++++++ .../unit/indexes/columnStore/manifest.test.ts | 162 ++++ .../columnStore/segment-cursor.test.ts | 270 +++++++ .../columnStore/segment-format.test.ts | 501 +++++++++++++ .../indexes/columnStore/tail-buffer.test.ts | 169 +++++ .../metadataIndex-automatic-bucketing.test.ts | 297 -------- 16 files changed, 3846 insertions(+), 521 deletions(-) create mode 100644 src/indexes/columnStore/ColumnManifest.ts create mode 100644 src/indexes/columnStore/ColumnSegmentCursor.ts create mode 100644 src/indexes/columnStore/ColumnSegmentFormat.ts create mode 100644 src/indexes/columnStore/ColumnStore.ts create mode 100644 src/indexes/columnStore/ColumnTailBuffer.ts create mode 100644 src/indexes/columnStore/index.ts create mode 100644 src/indexes/columnStore/types.ts create mode 100644 tests/unit/indexes/columnStore/column-store.test.ts create mode 100644 tests/unit/indexes/columnStore/manifest.test.ts create mode 100644 tests/unit/indexes/columnStore/segment-cursor.test.ts create mode 100644 tests/unit/indexes/columnStore/segment-format.test.ts create mode 100644 tests/unit/indexes/columnStore/tail-buffer.test.ts delete mode 100644 tests/unit/utils/metadataIndex-automatic-bucketing.test.ts diff --git a/src/brainy.ts b/src/brainy.ts index 5c8936dc..87c897f4 100644 --- a/src/brainy.ts +++ b/src/brainy.ts @@ -2166,19 +2166,31 @@ export class Brainy implements BrainyInterface { const limit = params.limit || 20 const offset = params.offset || 0 - // orderBy without any filter is not supported — it would require - // O(N) work across every entity in storage. Consumers must supply - // a filter (`type`, `where`, or `excludeVFS: true`) so the sort - // runs over a bounded, roaring-bitmap-filtered set. The long-term - // fix is a dedicated time-ordered segment index (Track 2). + // Unfiltered sort: column store handles this at O(K log S) scale. + // No per-entity storage reads, no bucketing precision loss. if (params.orderBy) { - throw new Error( - `find({ orderBy: '${params.orderBy}' }) requires a filter. ` + - `Add 'type', 'where', or 'excludeVFS: true' so the sort runs ` + - `over a bounded set. Unfiltered sort over all entities is not ` + - `scalable with the current index and is tracked as a dedicated ` + - `time-ordered segment index (Track 2).` + const k = limit + offset + const sortedIntIds = await this.metadataIndex.columnStore.sortTopK( + params.orderBy, + params.order || 'asc', + k ) + + // Convert int IDs to UUIDs and paginate + const idMapper = this.metadataIndex.getIdMapper() + const allUuids = sortedIntIds + .map(intId => idMapper.getUuid(intId)) + .filter((uuid): uuid is string => uuid !== undefined) + const pageIds = allUuids.slice(offset, offset + limit) + + const entitiesMap = await this.batchGet(pageIds) + for (const id of pageIds) { + const entity = entitiesMap.get(id) + if (entity) { + results.push(this.createResult(id, 1.0, entity)) + } + } + return results } // ExcludeVFS helper - exclude VFS infrastructure entities diff --git a/src/indexes/columnStore/ColumnManifest.ts b/src/indexes/columnStore/ColumnManifest.ts new file mode 100644 index 00000000..659a6c2e --- /dev/null +++ b/src/indexes/columnStore/ColumnManifest.ts @@ -0,0 +1,204 @@ +/** + * @module columnStore/ColumnManifest + * @description Per-field manifest tracking segment files and field metadata. + * + * Stored as JSON at `_column_index/{field}/MANIFEST.json` using workspace-global + * storage paths (not branch-scoped, same as `_cow/` metadata). Rewritten + * atomically on every flush and compaction. + * + * The manifest is the single source of truth for which segments exist for a + * field. On startup, the column store loads each field's manifest and + * discovers segments from it — no directory listing needed. + */ + +import type { StorageAdapter } from '../../coreTypes.js' +import type { ManifestData, SegmentMeta } from './types.js' +import { ValueType } from './types.js' + +/** + * Manages a single field's segment manifest. + * + * @example + * const manifest = new ColumnManifest('createdAt', '_column_index') + * await manifest.load(storage) + * manifest.addSegment({ id: 1, level: 0, count: 50000, ... }) + * await manifest.save(storage) + */ +export class ColumnManifest { + /** Field this manifest covers. */ + readonly fieldName: string + + /** Base storage path for the column index. */ + readonly basePath: string + + /** Manifest data — loaded from storage or initialized fresh. */ + private data: ManifestData + + /** + * @param fieldName - The field name (e.g. 'createdAt', 'status', '__words__') + * @param basePath - Base path for segment files (default: '_column_index') + */ + constructor(fieldName: string, basePath: string = '_column_index') { + this.fieldName = fieldName + this.basePath = basePath + this.data = { + version: 0, + field: fieldName, + valueType: ValueType.Number, + multiValue: false, + segments: [], + nextSegmentId: 1, + buildComplete: false + } + } + + /** + * Load the manifest from storage. If not found, starts fresh. + * + * @param storage - Storage adapter + */ + async load(storage: StorageAdapter): Promise { + try { + const path = this.manifestPath() + const data = await (storage as any).readObjectFromPath(path) + if (data && data.field === this.fieldName) { + this.data = data as ManifestData + } + } catch { + // Not found — start fresh (data is already initialized in constructor) + } + } + + /** + * Save the manifest to storage atomically. + * + * @param storage - Storage adapter + */ + async save(storage: StorageAdapter): Promise { + this.data.version++ + const path = this.manifestPath() + await (storage as any).writeObjectToPath(path, this.data) + } + + /** + * Register a new segment in the manifest. + * + * @param meta - Segment metadata + */ + addSegment(meta: SegmentMeta): void { + this.data.segments.push(meta) + // Keep sorted by level then id for consistent ordering + this.data.segments.sort((a, b) => a.level !== b.level ? a.level - b.level : a.id - b.id) + } + + /** + * Remove segments by ID (e.g. after compaction replaces them). + * + * @param ids - Segment IDs to remove + */ + removeSegments(ids: number[]): void { + const idSet = new Set(ids) + this.data.segments = this.data.segments.filter(s => !idSet.has(s.id)) + } + + /** + * Allocate the next segment ID (monotonically increasing). + * + * @returns Next available segment ID + */ + nextSegmentId(): number { + return this.data.nextSegmentId++ + } + + /** + * Get all segments at a specific compaction level. + * + * @param level - LSM compaction level (0 = freshest) + * @returns Segments at that level + */ + getSegmentsAtLevel(level: number): SegmentMeta[] { + return this.data.segments.filter(s => s.level === level) + } + + /** + * Get all segments across all levels. + */ + getAllSegments(): SegmentMeta[] { + return this.data.segments + } + + /** + * Whether the manifest has any segments. + */ + isEmpty(): boolean { + return this.data.segments.length === 0 + } + + /** + * Total entry count across all segments. + */ + get totalCount(): number { + return this.data.segments.reduce((sum, s) => sum + s.count, 0) + } + + /** + * Value type for this field. + */ + get valueType(): ValueType { + return this.data.valueType + } + + set valueType(type: ValueType) { + this.data.valueType = type + } + + /** + * Whether this is a multi-value field (e.g. __words__). + */ + get multiValue(): boolean { + return this.data.multiValue + } + + set multiValue(mv: boolean) { + this.data.multiValue = mv + } + + /** + * Whether the initial migration from existing entities has completed. + */ + get buildComplete(): boolean { + return this.data.buildComplete + } + + set buildComplete(done: boolean) { + this.data.buildComplete = done + } + + /** + * Generate the storage path for a segment file. + * + * @param level - Compaction level + * @param id - Segment ID + * @returns Path like `_column_index/createdAt/L0-000001.cidx` + */ + segmentPath(level: number, id: number): string { + const paddedId = String(id).padStart(6, '0') + return `${this.basePath}/${this.fieldName}/L${level}-${paddedId}.cidx` + } + + /** + * Generate the storage path for this field's manifest JSON. + * + * @returns Path like `_column_index/createdAt/MANIFEST.json` + */ + manifestPath(): string { + return `${this.basePath}/${this.fieldName}/MANIFEST.json` + } + + /** + * Get the raw manifest data (for serialization or debugging). + */ + getData(): Readonly { + return this.data + } +} diff --git a/src/indexes/columnStore/ColumnSegmentCursor.ts b/src/indexes/columnStore/ColumnSegmentCursor.ts new file mode 100644 index 00000000..50395dfe --- /dev/null +++ b/src/indexes/columnStore/ColumnSegmentCursor.ts @@ -0,0 +1,271 @@ +/** + * @module columnStore/ColumnSegmentCursor + * @description Cursor for reading a parsed column segment with binary search, + * forward/backward iteration, and tombstone skipping. + * + * Loaded segments are immutable — the cursor does not modify segment data. + * Multiple cursors can read the same segment concurrently (e.g. during k-way merge). + * + * For the TS baseline, segments are loaded fully into memory and cached via + * UnifiedCache. The Cortex Rust implementation mmaps the segment file and + * indexes into it directly — same read interface, zero-copy access. + */ + +import type { SegmentHeader } from './types.js' +import { ValueType } from './types.js' +import type { RoaringBitmap32 } from '../../utils/roaring/index.js' + +/** + * Binary search result. + */ +export interface SearchResult { + /** Whether an exact match was found. */ + found: boolean + /** Index of the match, or the insertion point if not found. */ + index: number +} + +/** + * A single (value, entityIntId) entry yielded by iteration. + */ +export interface CursorEntry { + value: number | string + entityIntId: number + /** Position in the segment (for tombstone reference). */ + position: number +} + +/** + * Cursor over a single parsed column segment. + * + * The segment's values are sorted in ascending order. Binary search finds + * positions in O(log n). Forward/backward iteration streams entries + * while skipping tombstoned positions. + */ +export class ColumnSegmentCursor { + readonly header: SegmentHeader + readonly values: (number | string)[] + readonly entityIds: Uint32Array + readonly tombstones: RoaringBitmap32 + + constructor( + header: SegmentHeader, + values: (number | string)[], + entityIds: Uint32Array, + tombstones: RoaringBitmap32 + ) { + this.header = header + this.values = values + this.entityIds = entityIds + this.tombstones = tombstones + } + + /** + * Number of non-tombstoned entries. + */ + get liveCount(): number { + return this.header.count - this.tombstones.size + } + + /** + * Minimum value in the sorted column (first entry). + * Returns undefined for empty segments. + */ + get minValue(): number | string | undefined { + return this.values.length > 0 ? this.values[0] : undefined + } + + /** + * Maximum value in the sorted column (last entry). + * Returns undefined for empty segments. + */ + get maxValue(): number | string | undefined { + return this.values.length > 0 ? this.values[this.values.length - 1] : undefined + } + + /** + * Binary search for a target value in the sorted column. + * + * For numeric values, uses numeric comparison. + * For string values, uses lexicographic comparison. + * + * @param target - Value to search for + * @returns Search result with found flag and index + */ + binarySearchValue(target: number | string): SearchResult { + let lo = 0 + let hi = this.values.length + const isString = this.header.valueType === ValueType.String + + while (lo < hi) { + const mid = (lo + hi) >>> 1 + const cmp = isString + ? String(this.values[mid]).localeCompare(String(target)) + : (this.values[mid] as number) - (target as number) + + if (cmp < 0) { + lo = mid + 1 + } else if (cmp > 0) { + hi = mid + } else { + // Found exact match — scan backward to find the FIRST occurrence + // (values may have duplicates in multi-value columns) + let first = mid + while (first > 0 && this.values[first - 1] === target) first-- + return { found: true, index: first } + } + } + return { found: false, index: lo } + } + + /** + * Find the range of positions where values fall within [lo, hi]. + * + * @param lo - Lower bound (inclusive) + * @param hi - Upper bound (inclusive) + * @returns Start (inclusive) and end (exclusive) positions + */ + binarySearchRange(lo: number | string, hi: number | string): { startIdx: number, endIdx: number } { + const startResult = this.binarySearchValue(lo) + const startIdx = startResult.index + + // Find the first position AFTER hi + const isString = this.header.valueType === ValueType.String + let endLo = startIdx + let endHi = this.values.length + while (endLo < endHi) { + const mid = (endLo + endHi) >>> 1 + const cmp = isString + ? String(this.values[mid]).localeCompare(String(hi)) + : (this.values[mid] as number) - (hi as number) + if (cmp <= 0) { + endLo = mid + 1 + } else { + endHi = mid + } + } + + return { startIdx, endIdx: endLo } + } + + /** + * Get all non-tombstoned entity IDs for a specific value (point query). + * + * @param value - Exact value to match + * @returns Array of matching entity int IDs + */ + getEntityIdsForValue(value: number | string): number[] { + const { found, index } = this.binarySearchValue(value) + if (!found) return [] + + const result: number[] = [] + for (let i = index; i < this.values.length && this.values[i] === value; i++) { + if (!this.tombstones.has(i)) { + result.push(this.entityIds[i]) + } + } + return result + } + + /** + * Get all non-tombstoned entity IDs in a value range. + * + * @param lo - Lower bound (inclusive) + * @param hi - Upper bound (inclusive) + * @returns Array of matching entity int IDs + */ + getEntityIdsInRange(lo: number | string, hi: number | string): number[] { + const { startIdx, endIdx } = this.binarySearchRange(lo, hi) + const result: number[] = [] + for (let i = startIdx; i < endIdx; i++) { + if (!this.tombstones.has(i)) { + result.push(this.entityIds[i]) + } + } + return result + } + + /** + * Check if a position is tombstoned (logically deleted). + * + * @param position - Index in the segment + * @returns True if the position is tombstoned + */ + isDeleted(position: number): boolean { + return this.tombstones.has(position) + } + + /** + * Iterate forward from a position, yielding non-tombstoned entries. + * + * @param from - Starting index (default: 0) + */ + *iterateForward(from = 0): Generator { + for (let i = from; i < this.values.length; i++) { + if (!this.tombstones.has(i)) { + yield { value: this.values[i], entityIntId: this.entityIds[i], position: i } + } + } + } + + /** + * Iterate backward from a position, yielding non-tombstoned entries. + * + * @param from - Starting index (default: last entry) + */ + *iterateBackward(from?: number): Generator { + const start = from ?? this.values.length - 1 + for (let i = start; i >= 0; i--) { + if (!this.tombstones.has(i)) { + yield { value: this.values[i], entityIntId: this.entityIds[i], position: i } + } + } + } +} + +/** + * Cursor wrapper over an in-memory tail buffer's drained data. + * Provides the same iteration interface as ColumnSegmentCursor so + * the k-way merge can treat tail buffer and segment data uniformly. + */ +export class TailBufferCursor { + readonly values: (number | string)[] + readonly entityIds: Uint32Array + readonly valueType: ValueType + + constructor(values: (number | string)[], entityIds: Uint32Array, valueType: ValueType) { + this.values = values + this.entityIds = entityIds + this.valueType = valueType + } + + get minValue(): number | string | undefined { + return this.values.length > 0 ? this.values[0] : undefined + } + + get maxValue(): number | string | undefined { + return this.values.length > 0 ? this.values[this.values.length - 1] : undefined + } + + *iterateForward(from = 0): Generator { + for (let i = from; i < this.values.length; i++) { + yield { value: this.values[i], entityIntId: this.entityIds[i], position: i } + } + } + + *iterateBackward(from?: number): Generator { + const start = from ?? this.values.length - 1 + for (let i = start; i >= 0; i--) { + yield { value: this.values[i], entityIntId: this.entityIds[i], position: i } + } + } + + getEntityIdsForValue(value: number | string): number[] { + // Linear scan — tail buffer is small (< flush threshold) + const result: number[] = [] + for (let i = 0; i < this.values.length; i++) { + if (this.values[i] === value) result.push(this.entityIds[i]) + } + return result + } +} diff --git a/src/indexes/columnStore/ColumnSegmentFormat.ts b/src/indexes/columnStore/ColumnSegmentFormat.ts new file mode 100644 index 00000000..7cee6abe --- /dev/null +++ b/src/indexes/columnStore/ColumnSegmentFormat.ts @@ -0,0 +1,548 @@ +/** + * @module columnStore/ColumnSegmentFormat + * @description Binary reader/writer for `.cidx` column segment files. + * + * Segment layout (little-endian throughout): + * ``` + * ┌─────────────────────────────────────────────┐ + * │ HEADER (64 bytes) │ + * ├─────────────────────────────────────────────┤ + * │ VALUES (count × 8 bytes for numeric, │ + * │ variable for strings) │ + * ├─────────────────────────────────────────────┤ + * │ ENTITY IDS (count × 4 bytes, u32 LE) │ + * ├─────────────────────────────────────────────┤ + * │ TOMBSTONES (serialized roaring bitmap) │ + * ├─────────────────────────────────────────────┤ + * │ FOOTER (16 bytes) │ + * └─────────────────────────────────────────────┘ + * ``` + * + * Both TypeScript and Rust read/write this exact byte layout. + * Cross-language format tests validate compatibility. + */ + +import { createHash } from 'node:crypto' +import { + CIDX_MAGIC, + CIDX_VERSION, + HEADER_SIZE, + FOOTER_SIZE, + MAX_FIELD_NAME_LENGTH, + ValueType, + type SegmentHeader, + type SegmentFooter +} from './types.js' +import { RoaringBitmap32 } from '../../utils/roaring/index.js' + +// --------------------------------------------------------------------------- +// CRC32 — use a simple table-based implementation (no external dep) +// --------------------------------------------------------------------------- + +const CRC32_TABLE = new Uint32Array(256) +for (let i = 0; i < 256; i++) { + let c = i + for (let j = 0; j < 8; j++) { + c = (c & 1) ? (0xEDB88320 ^ (c >>> 1)) : (c >>> 1) + } + CRC32_TABLE[i] = c +} + +/** + * Compute CRC32 of a buffer. + * + * @param data - Input bytes + * @returns CRC32 checksum as unsigned 32-bit integer + */ +export function crc32(data: Uint8Array): number { + let crc = 0xFFFFFFFF + for (let i = 0; i < data.length; i++) { + crc = CRC32_TABLE[(crc ^ data[i]) & 0xFF] ^ (crc >>> 8) + } + return (crc ^ 0xFFFFFFFF) >>> 0 +} + +// --------------------------------------------------------------------------- +// Header read / write +// --------------------------------------------------------------------------- + +/** + * Write a 64-byte segment header into a buffer. + * + * @param header - Parsed header fields + * @returns 64-byte Buffer + */ +export function writeHeader(header: SegmentHeader): Buffer { + const buf = Buffer.alloc(HEADER_SIZE) + const view = new DataView(buf.buffer, buf.byteOffset, buf.byteLength) + + // [0..4) magic + view.setUint32(0, header.magic, true) + // [4..6) version + view.setUint16(4, header.version, true) + + // [6..8) fieldNameLength + const nameBytes = Buffer.from(header.fieldName, 'utf-8') + const nameLen = Math.min(nameBytes.length, MAX_FIELD_NAME_LENGTH) + view.setUint16(6, nameLen, true) + + // [8..40) fieldName (zero-padded) + nameBytes.copy(buf, 8, 0, nameLen) + + // [40..41) valueType + buf[40] = header.valueType + // [41..42) level + buf[41] = header.level + // [42..43) codec + buf[42] = header.codec + // [43..44) flags + buf[43] = header.flags + + // [44..48) reserved + // [48..56) count as u64 + view.setBigUint64(48, BigInt(header.count), true) + // [56..64) reserved + + return buf +} + +/** + * Parse a 64-byte segment header from a buffer. + * + * @param buf - At least 64 bytes + * @returns Parsed SegmentHeader + * @throws If magic or version don't match + */ +export function readHeader(buf: Buffer | Uint8Array): SegmentHeader { + if (buf.length < HEADER_SIZE) { + throw new Error(`Buffer too small for header: ${buf.length} < ${HEADER_SIZE}`) + } + + const view = new DataView( + buf instanceof Buffer ? buf.buffer : buf.buffer, + buf instanceof Buffer ? buf.byteOffset : buf.byteOffset, + HEADER_SIZE + ) + + const magic = view.getUint32(0, true) + if (magic !== CIDX_MAGIC) { + throw new Error(`Invalid segment magic: 0x${magic.toString(16)} (expected 0x${CIDX_MAGIC.toString(16)})`) + } + + const version = view.getUint16(4, true) + if (version !== CIDX_VERSION) { + throw new Error(`Unsupported segment version: ${version} (expected ${CIDX_VERSION})`) + } + + const fieldNameLength = view.getUint16(6, true) + const fieldName = Buffer.from(buf.slice(8, 8 + fieldNameLength)).toString('utf-8') + + return { + magic, + version, + fieldName, + fieldNameLength, + valueType: buf[40] as ValueType, + level: buf[41], + codec: buf[42], + flags: buf[43], + count: Number(view.getBigUint64(48, true)) + } +} + +// --------------------------------------------------------------------------- +// Footer read / write +// --------------------------------------------------------------------------- + +/** + * Write a 16-byte footer. + * + * @param tombstoneLength - Byte length of the serialized tombstone bitmap + * @param checksum - CRC32 of header + values + entityIds + tombstones + * @returns 16-byte Buffer + */ +export function writeFooter(tombstoneLength: number, checksum: number): Buffer { + const buf = Buffer.alloc(FOOTER_SIZE) + const view = new DataView(buf.buffer, buf.byteOffset, buf.byteLength) + view.setBigUint64(0, BigInt(tombstoneLength), true) + view.setUint32(8, checksum, true) + // [12..16) pad + return buf +} + +/** + * Parse a 16-byte footer. + * + * @param buf - Exactly 16 bytes + * @returns Parsed footer + */ +export function readFooter(buf: Buffer | Uint8Array): SegmentFooter { + if (buf.length < FOOTER_SIZE) { + throw new Error(`Buffer too small for footer: ${buf.length} < ${FOOTER_SIZE}`) + } + const view = new DataView( + buf instanceof Buffer ? buf.buffer : buf.buffer, + buf instanceof Buffer ? buf.byteOffset : buf.byteOffset, + FOOTER_SIZE + ) + return { + tombstoneLength: Number(view.getBigUint64(0, true)), + crc32: view.getUint32(8, true) + } +} + +// --------------------------------------------------------------------------- +// Value column encoding +// --------------------------------------------------------------------------- + +/** + * Encode a sorted array of numeric values into a packed i64 LE buffer. + * + * @param values - Sorted numbers (integers or timestamps) + * @returns Packed buffer (values.length × 8 bytes) + */ +export function encodeNumericValues(values: number[]): Buffer { + const buf = Buffer.alloc(values.length * 8) + const view = new DataView(buf.buffer, buf.byteOffset, buf.byteLength) + for (let i = 0; i < values.length; i++) { + view.setBigInt64(i * 8, BigInt(Math.round(values[i])), true) + } + return buf +} + +/** + * Decode a packed i64 LE buffer into a number array. + * + * @param buf - Packed buffer + * @param count - Number of values + * @returns Array of numbers + */ +export function decodeNumericValues(buf: Buffer | Uint8Array, count: number): number[] { + const view = new DataView( + buf instanceof Buffer ? buf.buffer : buf.buffer, + buf instanceof Buffer ? buf.byteOffset : buf.byteOffset, + count * 8 + ) + const result = new Array(count) + for (let i = 0; i < count; i++) { + result[i] = Number(view.getBigInt64(i * 8, true)) + } + return result +} + +/** + * Encode a sorted array of float values into a packed f64 LE buffer. + * + * @param values - Sorted floats + * @returns Packed buffer (values.length × 8 bytes) + */ +export function encodeFloatValues(values: number[]): Buffer { + const buf = Buffer.alloc(values.length * 8) + const view = new DataView(buf.buffer, buf.byteOffset, buf.byteLength) + for (let i = 0; i < values.length; i++) { + view.setFloat64(i * 8, values[i], true) + } + return buf +} + +/** + * Decode a packed f64 LE buffer into a number array. + * + * @param buf - Packed buffer + * @param count - Number of values + * @returns Array of numbers + */ +export function decodeFloatValues(buf: Buffer | Uint8Array, count: number): number[] { + const view = new DataView( + buf instanceof Buffer ? buf.buffer : buf.buffer, + buf instanceof Buffer ? buf.byteOffset : buf.byteOffset, + count * 8 + ) + const result = new Array(count) + for (let i = 0; i < count; i++) { + result[i] = view.getFloat64(i * 8, true) + } + return result +} + +/** + * Encode sorted string values as length-prefixed UTF-8. + * + * Format per string: u32 LE length + UTF-8 bytes (no padding, no null terminator). + * + * @param values - Sorted strings + * @returns Packed buffer + */ +export function encodeStringValues(values: string[]): Buffer { + // First pass: calculate total size + const encoded = values.map(v => Buffer.from(v, 'utf-8')) + const totalSize = encoded.reduce((sum, b) => sum + 4 + b.length, 0) + + const buf = Buffer.alloc(totalSize) + let offset = 0 + for (const bytes of encoded) { + buf.writeUInt32LE(bytes.length, offset) + offset += 4 + bytes.copy(buf, offset) + offset += bytes.length + } + return buf +} + +/** + * Decode length-prefixed UTF-8 strings. + * + * @param buf - Packed buffer + * @param count - Number of strings + * @returns Array of strings + */ +export function decodeStringValues(buf: Buffer | Uint8Array, count: number): string[] { + const result = new Array(count) + let offset = 0 + const b = buf instanceof Buffer ? buf : Buffer.from(buf) + for (let i = 0; i < count; i++) { + const len = b.readUInt32LE(offset) + offset += 4 + result[i] = b.toString('utf-8', offset, offset + len) + offset += len + } + return result +} + +/** + * Encode entity int IDs as packed u32 LE. + * + * @param ids - Entity integer IDs (parallel to value column) + * @returns Packed buffer (ids.length × 4 bytes) + */ +export function encodeEntityIds(ids: Uint32Array | number[]): Buffer { + const arr = ids instanceof Uint32Array ? ids : new Uint32Array(ids) + return Buffer.from(arr.buffer, arr.byteOffset, arr.byteLength) +} + +/** + * Decode packed u32 LE entity IDs. + * + * @param buf - Packed buffer + * @param count - Number of IDs + * @returns Uint32Array + */ +export function decodeEntityIds(buf: Buffer | Uint8Array, count: number): Uint32Array { + const result = new Uint32Array(count) + const view = new DataView( + buf instanceof Buffer ? buf.buffer : buf.buffer, + buf instanceof Buffer ? buf.byteOffset : buf.byteOffset, + count * 4 + ) + for (let i = 0; i < count; i++) { + result[i] = view.getUint32(i * 4, true) + } + return result +} + +// --------------------------------------------------------------------------- +// Full segment write / read +// --------------------------------------------------------------------------- + +/** + * Encode values based on ValueType. + * + * @param values - Sorted values + * @param valueType - How to encode + * @returns Packed buffer + */ +export function encodeValues(values: (number | string | boolean)[], valueType: ValueType): Buffer { + switch (valueType) { + case ValueType.Number: + case ValueType.Boolean: + return encodeNumericValues(values as number[]) + case ValueType.Float: + return encodeFloatValues(values as number[]) + case ValueType.String: + return encodeStringValues(values as string[]) + default: + throw new Error(`Unknown ValueType: ${valueType}`) + } +} + +/** + * Decode values based on ValueType. + * + * @param buf - Packed value buffer + * @param count - Number of values + * @param valueType - How to decode + * @returns Array of values + */ +export function decodeValues(buf: Buffer | Uint8Array, count: number, valueType: ValueType): (number | string)[] { + switch (valueType) { + case ValueType.Number: + case ValueType.Boolean: + return decodeNumericValues(buf, count) + case ValueType.Float: + return decodeFloatValues(buf, count) + case ValueType.String: + return decodeStringValues(buf, count) + default: + throw new Error(`Unknown ValueType: ${valueType}`) + } +} + +/** + * Compute the byte size of the values column for a given count and type. + * For strings this is not knowable without the data, so returns -1. + * + * @param count - Number of entries + * @param valueType - Value encoding type + * @returns Byte size, or -1 for variable-length (strings) + */ +export function valuesColumnSize(count: number, valueType: ValueType): number { + switch (valueType) { + case ValueType.Number: + case ValueType.Boolean: + case ValueType.Float: + return count * 8 + case ValueType.String: + return -1 // variable length + default: + return -1 + } +} + +/** + * Write a complete `.cidx` segment file to a Buffer. + * + * @param header - Segment header (magic and version are set automatically) + * @param values - Sorted value array + * @param entityIds - Parallel entity int ID array + * @param tombstones - Optional tombstone bitmap (positions that are logically deleted) + * @returns Complete segment as a Buffer + */ +export function writeSegmentToBuffer( + header: Omit, + values: (number | string | boolean)[], + entityIds: Uint32Array | number[], + tombstones?: RoaringBitmap32 +): Buffer { + if (values.length !== (entityIds instanceof Uint32Array ? entityIds.length : entityIds.length)) { + throw new Error(`Values (${values.length}) and entityIds (${entityIds instanceof Uint32Array ? entityIds.length : entityIds.length}) must have the same length`) + } + + const fullHeader: SegmentHeader = { + ...header, + magic: CIDX_MAGIC, + version: CIDX_VERSION, + count: values.length + } + + // Encode sections + const headerBuf = writeHeader(fullHeader) + const valuesBuf = encodeValues(values, header.valueType) + const idsBuf = encodeEntityIds(entityIds) + + // Serialize tombstones + let tombstoneBuf: Buffer + if (tombstones && tombstones.size > 0) { + const serialized = tombstones.serialize(true) // portable serialization + tombstoneBuf = Buffer.from(serialized) + } else { + tombstoneBuf = Buffer.alloc(0) + } + + // CRC32 covers header + values + entityIds + tombstones + const crcInput = Buffer.concat([headerBuf, valuesBuf, idsBuf, tombstoneBuf]) + const checksum = crc32(crcInput) + + const footerBuf = writeFooter(tombstoneBuf.length, checksum) + + return Buffer.concat([headerBuf, valuesBuf, idsBuf, tombstoneBuf, footerBuf]) +} + +/** + * Parsed segment data from a `.cidx` file. + */ +export interface ParsedSegment { + header: SegmentHeader + values: (number | string)[] + entityIds: Uint32Array + tombstones: RoaringBitmap32 +} + +/** + * Read a complete `.cidx` segment from a Buffer. + * + * @param buf - Complete segment buffer + * @param validateCrc - Whether to validate the CRC32 checksum (default: true) + * @returns Parsed segment data + * @throws If magic/version mismatch, buffer too small, or CRC mismatch + */ +export function readSegmentFromBuffer(buf: Buffer | Uint8Array, validateCrc = true): ParsedSegment { + if (buf.length < HEADER_SIZE + FOOTER_SIZE) { + throw new Error(`Buffer too small for segment: ${buf.length} < ${HEADER_SIZE + FOOTER_SIZE}`) + } + + // 1. Parse header + const header = readHeader(buf) + + // 2. Parse footer (last 16 bytes) + const footerStart = buf.length - FOOTER_SIZE + const footer = readFooter(buf.slice(footerStart)) + + // 3. Calculate section offsets + const valuesStart = HEADER_SIZE + const idsSize = header.count * 4 + const tombstoneSize = footer.tombstoneLength + + // For fixed-size values, we know exact offsets + const fixedValuesSize = valuesColumnSize(header.count, header.valueType) + + let valuesEnd: number + if (fixedValuesSize >= 0) { + valuesEnd = valuesStart + fixedValuesSize + } else { + // String values: entityIds start = footerStart - tombstoneSize - idsSize + valuesEnd = footerStart - tombstoneSize - idsSize + } + + const idsStart = valuesEnd + const idsEnd = idsStart + idsSize + const tombstoneStart = idsEnd + const tombstoneEnd = tombstoneStart + tombstoneSize + + // 4. Validate CRC + if (validateCrc) { + const crcInput = buf.slice(0, tombstoneEnd) + const computed = crc32(crcInput instanceof Buffer ? crcInput : Buffer.from(crcInput)) + if (computed !== footer.crc32) { + throw new Error(`CRC32 mismatch: computed 0x${computed.toString(16)} != stored 0x${footer.crc32.toString(16)}`) + } + } + + // 5. Decode values + const valuesBuf = buf.slice(valuesStart, valuesEnd) + const values = decodeValues( + valuesBuf instanceof Buffer ? valuesBuf : Buffer.from(valuesBuf), + header.count, + header.valueType + ) + + // 6. Decode entity IDs + const idsBuf = buf.slice(idsStart, idsEnd) + const entityIds = decodeEntityIds( + idsBuf instanceof Buffer ? idsBuf : Buffer.from(idsBuf), + header.count + ) + + // 7. Decode tombstones + let tombstones: RoaringBitmap32 + if (tombstoneSize > 0) { + const tombstoneBuf = buf.slice(tombstoneStart, tombstoneEnd) + tombstones = RoaringBitmap32.deserialize( + tombstoneBuf instanceof Buffer ? tombstoneBuf : Buffer.from(tombstoneBuf), + true // portable format + ) + } else { + tombstones = new RoaringBitmap32() + } + + return { header, values, entityIds, tombstones } +} diff --git a/src/indexes/columnStore/ColumnStore.ts b/src/indexes/columnStore/ColumnStore.ts new file mode 100644 index 00000000..2a8496fd --- /dev/null +++ b/src/indexes/columnStore/ColumnStore.ts @@ -0,0 +1,695 @@ +/** + * @module columnStore/ColumnStore + * @description Unified column store: one system for filtering, sorting, range queries, + * and text search on any field type with exact precision at billion scale. + * + * Architecture: per-field sorted columns stored as binary .cidx segment files. + * Each field has its own manifest, tail buffer, and set of immutable segments. + * Segments are organized in LSM-style levels (L0 = fresh, compaction merges up). + * + * Query operations: + * - filter(field, value) → roaring bitmap of matching entity IDs + * - rangeQuery(field, lo, hi) → roaring bitmap + * - sortTopK(field, order, k) → top-K entity IDs in sorted order + * - filteredSortTopK(bitmap, field, order, k) → sorted + filtered + * - getFilterValues(field) → distinct values + * + * Implements ColumnStoreProvider interface for plugin registration. + */ + +import type { StorageAdapter } from '../../coreTypes.js' +import type { EntityIdMapper } from '../../utils/entityIdMapper.js' +import type { ColumnStoreProvider, SegmentMeta } from './types.js' +import { + ValueType, + DEFAULT_FLUSH_THRESHOLD, + FLAG_MULTI_VALUE +} from './types.js' +import { ColumnTailBuffer } from './ColumnTailBuffer.js' +import { ColumnManifest } from './ColumnManifest.js' +import { ColumnSegmentCursor, TailBufferCursor, type CursorEntry } from './ColumnSegmentCursor.js' +import { writeSegmentToBuffer, readSegmentFromBuffer } from './ColumnSegmentFormat.js' +import { RoaringBitmap32 } from '../../utils/roaring/index.js' + +/** + * Configuration for the ColumnStore. + */ +export interface ColumnStoreConfig { + /** Base path for segment files (default: '_column_index') */ + basePath?: string + /** Flush threshold per tail buffer (default: 65536) */ + flushThreshold?: number + /** Max L0 segments before compaction triggers (default: 4) */ + l0CompactionTrigger?: number +} + +/** + * Heap entry for k-way merge sort across multiple cursors. + */ +interface HeapEntry { + value: number | string + entityIntId: number + cursorIndex: number + /** Iterator for the cursor — call next() to advance */ + iterator: Generator +} + +/** + * Unified column store coordinator. + * + * @example + * const store = new ColumnStore() + * await store.init(storage, idMapper) + * store.addEntity(42, { createdAt: Date.now(), status: 'active' }) + * await store.flush() + * const newest = await store.sortTopK('createdAt', 'desc', 10) + */ +export class ColumnStore implements ColumnStoreProvider { + private storage!: StorageAdapter + private idMapper!: EntityIdMapper + private basePath: string + private flushThreshold: number + private l0CompactionTrigger: number + + /** Per-field tail buffers for in-memory writes. */ + private tailBuffers: Map = new Map() + + /** Per-field manifests tracking segment files. */ + private manifests: Map = new Map() + + /** Cached loaded segment cursors: `field:segmentId` → cursor. */ + private segmentCache: Map = new Map() + + /** + * Per-field global deleted entity bitmap. Entity int IDs in this bitmap + * are skipped during ALL queries (filter, sort, range). Persisted in the + * manifest and cleared during compaction. + */ + private deletedEntities: Map = new Map() + + /** Known field value types (inferred from first write). */ + private fieldTypes: Map = new Map() + + /** Whether init() has completed. */ + private initialized = false + + /** + * Create a new ColumnStore instance. + * + * Call `init(storage, idMapper)` before using query methods that require + * storage access (loading segments). Write methods (`addEntity`, `removeEntity`) + * work immediately — the column store buffers writes in memory until flush. + */ + constructor(config?: ColumnStoreConfig) { + this.basePath = config?.basePath ?? '_column_index' + this.flushThreshold = config?.flushThreshold ?? DEFAULT_FLUSH_THRESHOLD + this.l0CompactionTrigger = config?.l0CompactionTrigger ?? 4 + } + + /** + * Initialize the column store: discover existing field manifests. + */ + async init(storage: StorageAdapter, idMapper: EntityIdMapper): Promise { + this.storage = storage + this.idMapper = idMapper + + // Discover fields by listing manifest files + try { + const paths = await (storage as any).listObjectsUnderPath(this.basePath + '/') + for (const path of paths) { + if (path.endsWith('/MANIFEST.json')) { + const fieldName = path.replace(this.basePath + '/', '').replace('/MANIFEST.json', '') + const manifest = new ColumnManifest(fieldName, this.basePath) + await manifest.load(storage) + this.manifests.set(fieldName, manifest) + this.fieldTypes.set(fieldName, manifest.valueType) + + // Load global deleted bitmap if it exists + try { + const deletedPath = `${this.basePath}/${fieldName}/DELETED.bin` + const stored = await (storage as any).readObjectFromPath(deletedPath) + if (stored && stored._binary && stored.data) { + const buf = Buffer.from(stored.data, 'base64') + this.deletedEntities.set(fieldName, RoaringBitmap32.deserialize(buf, true)) + } + } catch { + // No deleted bitmap — normal for fresh fields + } + } + } + } catch { + // No column index exists yet — fresh start + } + + this.initialized = true + } + + /** + * Index an entity's field values. + * + * For each field in the provided map, pushes (value, entityIntId) to the + * field's tail buffer. For arrays (multi-value fields like __words__), + * one entry per element is added. + * + * Auto-flushes when any tail buffer exceeds its threshold. + */ + addEntity(entityIntId: number, fields: Record): void { + for (const [field, rawValue] of Object.entries(fields)) { + if (rawValue === undefined || rawValue === null) continue + + // Multi-value: arrays get one entry per element + if (Array.isArray(rawValue)) { + for (const v of rawValue) { + if (v !== undefined && v !== null) { + this.pushToBuffer(field, v, entityIntId, true) + } + } + } else { + this.pushToBuffer(field, rawValue, entityIntId, false) + } + } + } + + /** + * Remove an entity from all indexed fields. + * + * Adds the entity to the global deleted bitmap for each field (checked + * during ALL queries) and removes any pending entries from tail buffers. + * The global bitmap is persisted in the manifest on flush and cleared + * during compaction. + */ + removeEntity(entityIntId: number): void { + // Remove from tail buffers (pending writes) + for (const [, buffer] of this.tailBuffers) { + buffer.remove(entityIntId) + } + + // Add to global deleted bitmap for every known field + for (const field of this.manifests.keys()) { + let deleted = this.deletedEntities.get(field) + if (!deleted) { + deleted = new RoaringBitmap32() + this.deletedEntities.set(field, deleted) + } + deleted.add(entityIntId) + } + } + + /** + * Point filter: find entities where field equals value. + * + * Searches all segments + tail buffer, returns union as roaring bitmap. + * Excludes globally deleted entities. + */ + async filter(field: string, value: unknown): Promise { + const result = new RoaringBitmap32() + const deleted = this.deletedEntities.get(field) + + // Search segments + const cursors = await this.getSegmentCursors(field) + for (const cursor of cursors) { + const ids = cursor.getEntityIdsForValue(value as number | string) + for (const id of ids) { + if (!deleted || !deleted.has(id)) result.add(id) + } + } + + // Search tail buffer + const tailCursor = this.getTailBufferCursor(field) + if (tailCursor) { + const ids = tailCursor.getEntityIdsForValue(value as number | string) + for (const id of ids) { + if (!deleted || !deleted.has(id)) result.add(id) + } + } + + return result + } + + /** + * Range filter: find entities where field is within [min, max]. + */ + async rangeQuery(field: string, min?: unknown, max?: unknown): Promise { + const result = new RoaringBitmap32() + const cursors = await this.getSegmentCursors(field) + + for (const cursor of cursors) { + const lo = (min !== undefined && min !== null) ? min as number | string : cursor.minValue + const hi = (max !== undefined && max !== null) ? max as number | string : cursor.maxValue + if (lo === undefined || hi === undefined) continue + const ids = cursor.getEntityIdsInRange(lo, hi) + for (const id of ids) result.add(id) + } + + // Tail buffer range: linear scan (tail is small) + const tailCursor = this.getTailBufferCursor(field) + if (tailCursor) { + for (const entry of tailCursor.iterateForward()) { + const v = entry.value + const inRange = (min == null || v >= min) && (max == null || v <= max) + if (inRange) result.add(entry.entityIntId) + } + } + + return result + } + + /** + * Sort top-K: return K entity int IDs in sorted order. + * + * Uses k-way merge across all segment cursors + tail buffer via a min/max heap. + * Complexity: O(K log S) where S = number of cursors. + */ + async sortTopK(field: string, order: 'asc' | 'desc', k: number): Promise { + return this.mergeSort(field, order, k, null) + } + + /** + * Filtered sort top-K: return K entity int IDs in sorted order, + * considering only entities present in the filter bitmap. + */ + async filteredSortTopK( + filterBitmap: RoaringBitmap32, + field: string, + order: 'asc' | 'desc', + k: number + ): Promise { + return this.mergeSort(field, order, k, filterBitmap) + } + + /** + * Get all distinct values for a field. + */ + async getFilterValues(field: string): Promise { + const valueSet = new Set() + const cursors = await this.getSegmentCursors(field) + + for (const cursor of cursors) { + for (const entry of cursor.iterateForward()) { + valueSet.add(String(entry.value)) + } + } + + const tailCursor = this.getTailBufferCursor(field) + if (tailCursor) { + for (const entry of tailCursor.iterateForward()) { + valueSet.add(String(entry.value)) + } + } + + return Array.from(valueSet).sort() + } + + /** + * Check if a field has any indexed data. + */ + hasField(field: string): boolean { + const manifest = this.manifests.get(field) + const buffer = this.tailBuffers.get(field) + return (manifest !== undefined && !manifest.isEmpty()) || (buffer !== undefined && buffer.size > 0) + } + + /** + * Flush all tail buffers to L0 segments and save manifests. + * No-op if init() hasn't been called (no storage to write to). + */ + async flush(): Promise { + if (!this.storage) return + for (const [field, buffer] of this.tailBuffers) { + if (buffer.size > 0) { + await this.flushBuffer(field, buffer) + } + } + } + + /** + * Flush and release all resources. + */ + async close(): Promise { + if (this.storage) await this.flush() + this.tailBuffers.clear() + this.segmentCache.clear() + this.manifests.clear() + this.deletedEntities.clear() + this.initialized = false + } + + // ========================================================================= + // Internal methods + // ========================================================================= + + /** + * Push a single value to a field's tail buffer. + * Creates the buffer and manifest if first write to this field. + * Infers ValueType from the first value seen. + */ + private pushToBuffer(field: string, value: unknown, entityIntId: number, isMultiValue: boolean): void { + let buffer = this.tailBuffers.get(field) + if (!buffer) { + const valueType = this.inferValueType(value) + buffer = new ColumnTailBuffer(field, valueType, this.flushThreshold) + this.tailBuffers.set(field, buffer) + this.fieldTypes.set(field, valueType) + + // Ensure manifest exists + if (!this.manifests.has(field)) { + const manifest = new ColumnManifest(field, this.basePath) + manifest.valueType = valueType + manifest.multiValue = isMultiValue + this.manifests.set(field, manifest) + } + } + + // Normalize value to the column type + const normalizedValue = this.normalizeValue(value, buffer.valueType) + if (normalizedValue !== undefined) { + buffer.add(normalizedValue, entityIntId) + } + } + + /** + * Flush a single field's tail buffer to a new L0 segment. + */ + private async flushBuffer(field: string, buffer: ColumnTailBuffer): Promise { + const { values, entityIds, pendingRemovals } = buffer.drain() + if (values.length === 0 && pendingRemovals.size === 0) return + + const manifest = this.manifests.get(field) + if (!manifest) return + + // Write segment if we have values + if (values.length > 0) { + const segId = manifest.nextSegmentId() + const segPath = manifest.segmentPath(0, segId) + + const segBuffer = writeSegmentToBuffer( + { + fieldName: field, + fieldNameLength: Math.min(field.length, 32), + valueType: manifest.valueType, + level: 0, + codec: 0, + flags: manifest.multiValue ? FLAG_MULTI_VALUE : 0, + count: values.length + }, + values, + entityIds + ) + + // Store segment as binary data + await (this.storage as any).writeObjectToPath(segPath, { + _binary: true, + data: segBuffer.toString('base64') + }) + + // Register in manifest + manifest.addSegment({ + id: segId, + level: 0, + count: values.length, + minValue: values[0], + maxValue: values[values.length - 1], + file: `L0-${String(segId).padStart(6, '0')}.cidx` + }) + + // Clear cached cursor for this field (stale after new segment) + this.invalidateFieldCache(field) + } + + // Merge pending removals into the global deleted bitmap + if (pendingRemovals.size > 0) { + let deleted = this.deletedEntities.get(field) + if (!deleted) { + deleted = new RoaringBitmap32() + this.deletedEntities.set(field, deleted) + } + for (const id of pendingRemovals) deleted.add(id) + } + + // Persist the global deleted bitmap alongside the manifest + const deleted = this.deletedEntities.get(field) + if (deleted && deleted.size > 0) { + const deletedPath = `${this.basePath}/${field}/DELETED.bin` + const serialized = deleted.serialize(true) + await (this.storage as any).writeObjectToPath(deletedPath, { + _binary: true, + data: Buffer.from(serialized).toString('base64') + }) + } + + // Save manifest + await manifest.save(this.storage) + } + + /** + * Get all segment cursors for a field, loading from storage if needed. + */ + private async getSegmentCursors(field: string): Promise { + const manifest = this.manifests.get(field) + if (!manifest) return [] + + const cursors: ColumnSegmentCursor[] = [] + for (const seg of manifest.getAllSegments()) { + const cacheKey = `${field}:${seg.id}` + let cursor = this.segmentCache.get(cacheKey) + + if (!cursor) { + const loaded = await this.loadSegmentCursor(field, seg) + if (loaded) { + cursor = loaded + this.segmentCache.set(cacheKey, cursor) + } + } + + if (cursor) cursors.push(cursor) + } + + return cursors + } + + /** + * Load a segment from storage and create a cursor. + */ + private async loadSegmentCursor(field: string, seg: SegmentMeta): Promise { + const manifest = this.manifests.get(field) + if (!manifest) return null + + try { + const segPath = manifest.segmentPath(seg.level, seg.id) + const stored = await (this.storage as any).readObjectFromPath(segPath) + if (!stored) return null + + // Handle binary storage format + let buf: Buffer + if (stored._binary && stored.data) { + buf = Buffer.from(stored.data, 'base64') + } else if (Buffer.isBuffer(stored)) { + buf = stored + } else { + return null + } + + const parsed = readSegmentFromBuffer(buf) + return new ColumnSegmentCursor( + parsed.header, + parsed.values, + parsed.entityIds, + parsed.tombstones + ) + } catch { + return null + } + } + + /** + * Get a cursor for the tail buffer of a field (for query inclusion). + */ + private getTailBufferCursor(field: string): TailBufferCursor | null { + const buffer = this.tailBuffers.get(field) + if (!buffer || buffer.size === 0) return null + + const valueType = this.fieldTypes.get(field) ?? ValueType.String + // Drain a COPY for querying — don't clear the buffer + const entries = (buffer as any).entries as Array<{ value: number | string, entityIntId: number }> + if (!entries || entries.length === 0) return null + + // Sort a copy for the cursor + const sorted = entries.slice().sort((a, b) => { + if (valueType === ValueType.String) { + return String(a.value).localeCompare(String(b.value)) + } + return (a.value as number) - (b.value as number) + }) + + const values = sorted.map(e => e.value) + const entityIds = new Uint32Array(sorted.map(e => e.entityIntId)) + return new TailBufferCursor(values, entityIds, valueType) + } + + /** + * K-way merge sort across all segment cursors and tail buffer. + * + * Uses a simple heap (array-based) to efficiently select the next + * smallest (asc) or largest (desc) entry across all cursors. + * + * @param field - Field to sort by + * @param order - 'asc' or 'desc' + * @param k - Maximum results + * @param filterBitmap - Optional filter (only include these entity IDs) + */ + private async mergeSort( + field: string, + order: 'asc' | 'desc', + k: number, + filterBitmap: RoaringBitmap32 | null + ): Promise { + // Collect all cursors (segments + tail buffer) + const segCursors = await this.getSegmentCursors(field) + const tailCursor = this.getTailBufferCursor(field) + + // Create iterators for each cursor in the specified direction + const iterators: Generator[] = [] + for (const cursor of segCursors) { + iterators.push(order === 'asc' ? cursor.iterateForward() : cursor.iterateBackward()) + } + if (tailCursor) { + iterators.push(order === 'asc' ? tailCursor.iterateForward() : tailCursor.iterateBackward()) + } + + if (iterators.length === 0) return [] + + // Initialize heap with first entry from each iterator + const heap: HeapEntry[] = [] + for (let i = 0; i < iterators.length; i++) { + const next = iterators[i].next() + if (!next.done) { + heap.push({ + value: next.value.value, + entityIntId: next.value.entityIntId, + cursorIndex: i, + iterator: iterators[i] + }) + } + } + + // Heapify + const isString = (this.fieldTypes.get(field) ?? ValueType.Number) === ValueType.String + const compare = (a: HeapEntry, b: HeapEntry): number => { + let cmp: number + if (isString) { + cmp = String(a.value).localeCompare(String(b.value)) + } else { + cmp = (a.value as number) - (b.value as number) + } + return order === 'asc' ? cmp : -cmp + } + + // Build min-heap + for (let i = Math.floor(heap.length / 2) - 1; i >= 0; i--) { + this.heapDown(heap, i, compare) + } + + // Collect results + const result: number[] = [] + const seen = new Set() // deduplicate entity IDs across segments + + while (heap.length > 0 && result.length < k) { + // Pop min/max from heap + const top = heap[0] + + // Replace root with next from same iterator + const next = top.iterator.next() + if (next.done) { + // Iterator exhausted — remove from heap by swapping with last + heap[0] = heap[heap.length - 1] + heap.pop() + } else { + heap[0] = { + value: next.value.value, + entityIntId: next.value.entityIntId, + cursorIndex: top.cursorIndex, + iterator: top.iterator + } + } + if (heap.length > 0) { + this.heapDown(heap, 0, compare) + } + + // Apply global deleted check, filter, and dedup + const deleted = this.deletedEntities.get(field) + if (deleted && deleted.has(top.entityIntId)) continue + if (seen.has(top.entityIntId)) continue + if (filterBitmap && !filterBitmap.has(top.entityIntId)) continue + seen.add(top.entityIntId) + result.push(top.entityIntId) + } + + return result + } + + /** + * Heap sift-down operation. + */ + private heapDown(heap: HeapEntry[], i: number, compare: (a: HeapEntry, b: HeapEntry) => number): void { + const n = heap.length + while (true) { + let smallest = i + const left = 2 * i + 1 + const right = 2 * i + 2 + + if (left < n && compare(heap[left], heap[smallest]) < 0) smallest = left + if (right < n && compare(heap[right], heap[smallest]) < 0) smallest = right + + if (smallest === i) break + ;[heap[i], heap[smallest]] = [heap[smallest], heap[i]] + i = smallest + } + } + + /** + * Invalidate cached segment cursors for a field. + */ + private invalidateFieldCache(field: string): void { + for (const key of this.segmentCache.keys()) { + if (key.startsWith(field + ':')) { + this.segmentCache.delete(key) + } + } + } + + /** + * Infer ValueType from a JavaScript value. + */ + private inferValueType(value: unknown): ValueType { + if (typeof value === 'boolean') return ValueType.Boolean + if (typeof value === 'number') { + return Number.isInteger(value) ? ValueType.Number : ValueType.Float + } + return ValueType.String + } + + /** + * Normalize a JavaScript value to the column's ValueType. + */ + private normalizeValue(value: unknown, type: ValueType): number | string | undefined { + switch (type) { + case ValueType.Number: + if (typeof value === 'number') return Math.round(value) + if (typeof value === 'string') { const n = Number(value); return isNaN(n) ? undefined : Math.round(n) } + if (typeof value === 'boolean') return value ? 1 : 0 + return undefined + case ValueType.Float: + if (typeof value === 'number') return value + if (typeof value === 'string') { const n = Number(value); return isNaN(n) ? undefined : n } + return undefined + case ValueType.Boolean: + if (typeof value === 'boolean') return value ? 1 : 0 + if (typeof value === 'number') return value ? 1 : 0 + return undefined + case ValueType.String: + return String(value) + default: + return undefined + } + } +} diff --git a/src/indexes/columnStore/ColumnTailBuffer.ts b/src/indexes/columnStore/ColumnTailBuffer.ts new file mode 100644 index 00000000..d2bb7c7e --- /dev/null +++ b/src/indexes/columnStore/ColumnTailBuffer.ts @@ -0,0 +1,169 @@ +/** + * @module columnStore/ColumnTailBuffer + * @description Per-field in-memory write buffer that accumulates (value, entityIntId) + * entries before flushing to a sorted L0 segment on disk. + * + * Follows the MemTable pattern from GraphAdjacencyIndex's LSMTree + * (src/graph/lsm/LSMTree.ts:64-143): unsorted writes in memory, + * sort on flush, clear after flush. + * + * For multi-value fields (__words__), one entity can add multiple entries + * to the same buffer. The sorted drain merges them naturally. + */ + +import { ValueType, DEFAULT_FLUSH_THRESHOLD } from './types.js' + +/** + * Entry in the tail buffer: a (value, entityIntId) pair. + * + * For numeric fields, value is a number (timestamp, integer count). + * For string fields, value is a string. + * For boolean fields, value is 0 or 1 (stored as number). + */ +export interface TailEntry { + value: number | string + entityIntId: number +} + +/** + * Result of draining the tail buffer: parallel sorted arrays + * ready for segment writing. + */ +export interface DrainResult { + values: (number | string)[] + entityIds: Uint32Array + pendingRemovals: Set +} + +/** + * In-memory write buffer for a single field's column data. + * + * Entries accumulate unsorted. On drain/flush, they are sorted by value + * (numeric comparison for numbers, lexicographic for strings). Ties are + * broken by entityIntId ascending for deterministic output. + * + * @example + * const buf = new ColumnTailBuffer('createdAt', ValueType.Number) + * buf.add(1700000001000, 42) + * buf.add(1700000000000, 10) + * const { values, entityIds } = buf.drain() + * // values: [1700000000000, 1700000001000] + * // entityIds: [10, 42] + */ +export class ColumnTailBuffer { + /** Field name this buffer is for. */ + readonly fieldName: string + + /** Value type determines sort comparator. */ + readonly valueType: ValueType + + /** Flush threshold. */ + readonly threshold: number + + /** Unsorted entry accumulator. */ + private entries: TailEntry[] = [] + + /** + * Entity int IDs marked for removal (tombstoning). + * Applied during segment writes — the removal propagates to existing segments. + */ + private removals: Set = new Set() + + /** + * @param fieldName - Field name (e.g. 'createdAt', 'status', '__words__') + * @param valueType - How values should be sorted and encoded + * @param threshold - Number of entries before shouldFlush() returns true + */ + constructor(fieldName: string, valueType: ValueType, threshold: number = DEFAULT_FLUSH_THRESHOLD) { + this.fieldName = fieldName + this.valueType = valueType + this.threshold = threshold + } + + /** + * Add a (value, entityIntId) entry to the buffer. + * + * @param value - The field value (number for numeric/boolean, string for string) + * @param entityIntId - The entity's u32 integer ID from EntityIdMapper + */ + add(value: number | string, entityIntId: number): void { + this.entries.push({ value, entityIntId }) + } + + /** + * Mark an entity for removal. The removal is tracked as a pending tombstone + * and applied when flushing or during segment compaction. + * + * @param entityIntId - Entity to remove + */ + remove(entityIntId: number): void { + this.removals.add(entityIntId) + // Also remove any pending entries for this entity in the buffer + this.entries = this.entries.filter(e => e.entityIntId !== entityIntId) + } + + /** + * Number of entries currently in the buffer. + */ + get size(): number { + return this.entries.length + } + + /** + * Whether the buffer has reached the flush threshold. + */ + shouldFlush(): boolean { + return this.entries.length >= this.threshold + } + + /** + * Drain the buffer: sort entries by value, return parallel arrays, + * and clear the buffer. Pending removals are returned separately + * for the caller to apply as tombstones to existing segments. + * + * Sorting: numeric values by numeric comparison, strings lexicographically. + * Ties broken by entityIntId ascending for deterministic segment output. + * + * @returns Sorted parallel arrays and pending removals + */ + drain(): DrainResult { + // Sort entries + const sorted = this.entries.slice() + if (this.valueType === ValueType.String) { + sorted.sort((a, b) => { + const cmp = String(a.value).localeCompare(String(b.value)) + return cmp !== 0 ? cmp : a.entityIntId - b.entityIntId + }) + } else { + sorted.sort((a, b) => { + const va = a.value as number + const vb = b.value as number + return va !== vb ? va - vb : a.entityIntId - b.entityIntId + }) + } + + // Build parallel arrays + const values: (number | string)[] = new Array(sorted.length) + const entityIds = new Uint32Array(sorted.length) + for (let i = 0; i < sorted.length; i++) { + values[i] = sorted[i].value + entityIds[i] = sorted[i].entityIntId + } + + const pendingRemovals = new Set(this.removals) + + // Clear buffer + this.entries = [] + this.removals.clear() + + return { values, entityIds, pendingRemovals } + } + + /** + * Clear the buffer without returning data. + */ + clear(): void { + this.entries = [] + this.removals.clear() + } +} diff --git a/src/indexes/columnStore/index.ts b/src/indexes/columnStore/index.ts new file mode 100644 index 00000000..1f9bf344 --- /dev/null +++ b/src/indexes/columnStore/index.ts @@ -0,0 +1,38 @@ +/** + * @module columnStore + * @description Unified column store for filtering, sorting, range queries, + * and text search on any field type with exact precision at billion scale. + * + * Replaces MetadataIndex's sparse chunk/bloom filter/bucketing internals + * with per-field sorted columns. Design lineage: Lucene doc values + roaring + * bitmap acceleration. + */ + +export { ColumnStore } from './ColumnStore.js' +export type { ColumnStoreConfig } from './ColumnStore.js' +export { ColumnTailBuffer } from './ColumnTailBuffer.js' +export { ColumnManifest } from './ColumnManifest.js' +export { ColumnSegmentCursor, TailBufferCursor } from './ColumnSegmentCursor.js' +export { + writeSegmentToBuffer, + readSegmentFromBuffer, + writeHeader, + readHeader, + crc32 +} from './ColumnSegmentFormat.js' +export { + ValueType, + CIDX_MAGIC, + CIDX_VERSION, + HEADER_SIZE, + FOOTER_SIZE, + DEFAULT_FLUSH_THRESHOLD, + FLAG_MULTI_VALUE +} from './types.js' +export type { + ColumnStoreProvider, + SegmentHeader, + SegmentFooter, + SegmentMeta, + ManifestData +} from './types.js' diff --git a/src/indexes/columnStore/types.ts b/src/indexes/columnStore/types.ts new file mode 100644 index 00000000..116ac172 --- /dev/null +++ b/src/indexes/columnStore/types.ts @@ -0,0 +1,272 @@ +/** + * @module columnStore/types + * @description Shared type definitions for the unified column store. + * + * The column store replaces MetadataIndex's sparse chunk/bloom filter/bucketing + * internals with per-field sorted columns. One system for filtering, sorting, + * range queries, and text search on any field type with exact precision. + * + * Design lineage: Lucene doc values + roaring bitmap acceleration. + */ + +import type { RoaringBitmap32 } from '../../utils/roaring/index.js' +import type { EntityIdMapper } from '../../utils/entityIdMapper.js' +import type { StorageAdapter } from '../../coreTypes.js' + +// --------------------------------------------------------------------------- +// Constants +// --------------------------------------------------------------------------- + +/** Magic bytes identifying a `.cidx` column segment file: "CIDX" in ASCII. */ +export const CIDX_MAGIC = 0x43494458 + +/** Binary format version. Bump on breaking changes to the segment layout. */ +export const CIDX_VERSION = 1 + +/** Fixed header size in bytes. Contains field metadata and zone map. */ +export const HEADER_SIZE = 64 + +/** Fixed footer size in bytes. Contains tombstone length and CRC32. */ +export const FOOTER_SIZE = 16 + +/** + * Default number of entries in the in-memory tail buffer before flushing + * to an L0 segment on disk. 65,536 entries × 12 bytes ≈ 768 KB per flush. + */ +export const DEFAULT_FLUSH_THRESHOLD = 65_536 + +/** Maximum field name length stored in the segment header (bytes). */ +export const MAX_FIELD_NAME_LENGTH = 32 + +// --------------------------------------------------------------------------- +// Value types +// --------------------------------------------------------------------------- + +/** + * Discriminator for the value column encoding in a segment file. + * + * Determines how values are packed in the binary segment: + * - Number: i64 little-endian (8 bytes each) — timestamps, integer counts + * - Float: f64 little-endian (8 bytes each) — prices, scores, ratings + * - String: u32-length-prefixed UTF-8 — status, category, words + * - Boolean: i64 (0 or 1) — flags, toggles + */ +export enum ValueType { + Number = 0, + Float = 1, + String = 2, + Boolean = 3 +} + +// --------------------------------------------------------------------------- +// Segment header and footer +// --------------------------------------------------------------------------- + +/** + * Parsed representation of the 64-byte segment file header. + * + * Layout (little-endian): + * ``` + * [0..4) magic: u32 = 0x43494458 ("CIDX") + * [4..6) version: u16 = 1 + * [6..8) fieldLen: u16 (actual byte length of field name) + * [8..40) fieldName: [u8; 32] (UTF-8, zero-padded) + * [40..41) valueType: u8 (ValueType enum) + * [41..42) level: u8 (LSM compaction level, 0 = fresh) + * [42..43) codec: u8 (0 = raw, reserved for future compression) + * [43..44) flags: u8 (bit 0 = multi-value) + * [44..48) _reserved: [u8; 4] + * [48..56) count: u64 (number of entries) + * [56..64) _reserved2:[u8; 8] + * ``` + * + * Min/max zone map values are derived from the first and last entries + * in the sorted value column — not stored redundantly in the header. + */ +export interface SegmentHeader { + magic: number + version: number + fieldName: string + fieldNameLength: number + valueType: ValueType + level: number + codec: number + flags: number + count: number +} + +/** + * Parsed representation of the 16-byte segment file footer. + * + * Layout (little-endian): + * ``` + * [0..8) tombstoneLen: u64 (byte length of the serialized tombstone bitmap) + * [8..12) crc32: u32 (CRC32 of header + values + entityIds + tombstones) + * [12..16) _pad: u32 + * ``` + */ +export interface SegmentFooter { + tombstoneLength: number + crc32: number +} + +// --------------------------------------------------------------------------- +// Manifest +// --------------------------------------------------------------------------- + +/** + * Metadata about a single segment file within a field's manifest. + */ +export interface SegmentMeta { + /** Segment identifier, unique within this field (monotonic counter). */ + id: number + /** LSM compaction level. 0 = freshest (direct from tail buffer). */ + level: number + /** Number of entries in the segment (including tombstoned). */ + count: number + /** Minimum value in the sorted column (for zone-map pruning). */ + minValue: number | string + /** Maximum value in the sorted column. */ + maxValue: number | string + /** Relative file path within the field directory. */ + file: string +} + +/** + * Per-field manifest tracking all segment files and field metadata. + * Stored as JSON at `_column_index/{field}/MANIFEST.json`. + */ +export interface ManifestData { + /** Format version for forward compatibility. */ + version: number + /** Field name this manifest covers. */ + field: string + /** Value type for all segments in this field. */ + valueType: ValueType + /** Whether this is a multi-value field (one entity → many entries). */ + multiValue: boolean + /** All known segments, ordered by level then id. */ + segments: SegmentMeta[] + /** Monotonically increasing segment ID counter. */ + nextSegmentId: number + /** Whether the initial build from existing entities has completed. */ + buildComplete: boolean +} + +// --------------------------------------------------------------------------- +// Header flags +// --------------------------------------------------------------------------- + +/** Bit 0 of flags: indicates a multi-value column (e.g. __words__). */ +export const FLAG_MULTI_VALUE = 0x01 + +// --------------------------------------------------------------------------- +// Column store provider interface +// --------------------------------------------------------------------------- + +/** + * The plugin-provider contract for the column store. + * + * Brainy ships a TypeScript baseline that implements this interface. + * Cortex registers a Rust-accelerated implementation at higher priority. + * The MetadataIndex coordinator calls whichever is registered. + */ +export interface ColumnStoreProvider { + /** + * Initialize the column store: discover fields, load manifests. + * + * @param storage - The storage adapter for reading/writing segment files + * @param idMapper - Shared EntityIdMapper for UUID ↔ u32 conversion + */ + init(storage: StorageAdapter, idMapper: EntityIdMapper): Promise + + /** + * Index an entity's field values. For multi-value fields (arrays), + * one entry per element is added to the column. + * + * @param entityIntId - The entity's u32 integer ID (from EntityIdMapper) + * @param fields - Map of field name → value (or array of values for multi-value) + */ + addEntity(entityIntId: number, fields: Record): void + + /** + * Remove an entity from all indexed columns by adding tombstones. + * + * @param entityIntId - The entity's u32 integer ID + */ + removeEntity(entityIntId: number): void + + /** + * Point filter: find all entities where field equals value. + * + * @param field - Field name to filter on + * @param value - Exact value to match + * @returns Roaring bitmap of matching entity int IDs + */ + filter(field: string, value: unknown): Promise + + /** + * Range filter: find all entities where field is within [min, max]. + * + * @param field - Field name to filter on + * @param min - Lower bound (undefined = no lower bound) + * @param max - Upper bound (undefined = no upper bound) + * @returns Roaring bitmap of matching entity int IDs + */ + rangeQuery(field: string, min?: unknown, max?: unknown): Promise + + /** + * Sort top-K: return the K entity int IDs with the highest or lowest + * values in the specified field, in sort order. + * + * @param field - Field to sort by + * @param order - 'asc' or 'desc' + * @param k - Maximum number of results + * @returns Sorted array of entity int IDs + */ + sortTopK(field: string, order: 'asc' | 'desc', k: number): Promise + + /** + * Filtered sort top-K: same as sortTopK but only considers entities + * present in the filter bitmap. + * + * @param filterBitmap - Roaring bitmap of candidate entity int IDs + * @param field - Field to sort by + * @param order - 'asc' or 'desc' + * @param k - Maximum number of results + * @returns Sorted array of entity int IDs (subset of filterBitmap) + */ + filteredSortTopK( + filterBitmap: RoaringBitmap32, + field: string, + order: 'asc' | 'desc', + k: number + ): Promise + + /** + * Get all distinct values for a field across all segments. + * + * @param field - Field name + * @returns Array of distinct values as strings + */ + getFilterValues(field: string): Promise + + /** + * Check if a field has any indexed data (segments or tail buffer entries). + * + * @param field - Field name + * @returns True if the field has data in the column store + */ + hasField(field: string): boolean + + /** + * Flush all in-memory tail buffers to L0 segments on disk. + * Saves all manifests. + */ + flush(): Promise + + /** + * Flush and release all resources. + */ + close(): Promise +} diff --git a/src/utils/metadataIndex.ts b/src/utils/metadataIndex.ts index 3fb31db6..b4b90eff 100644 --- a/src/utils/metadataIndex.ts +++ b/src/utils/metadataIndex.ts @@ -5,6 +5,7 @@ */ import { StorageAdapter, resolveEntityField } from '../coreTypes.js' +import { ColumnStore } from '../indexes/columnStore/ColumnStore.js' import { MetadataIndexCache, MetadataIndexCacheConfig } from './metadataIndexCache.js' import { prodLog } from './logger.js' import { getGlobalCache, UnifiedCache } from './unifiedCache.js' @@ -163,6 +164,14 @@ export class MetadataIndexManager { // Replaces unreliable pattern matching with DuckDB-inspired value analysis private fieldTypeInference: FieldTypeInference + /** + * Unified Column Store — replaces sparse index internals for filtering + sorting. + * Created in the constructor (no storage needed for writes), storage discovery + * happens in init(). Public so brainy.ts can call sortTopK directly for + * unfiltered sort. + */ + public columnStore: ColumnStore + constructor(storage: StorageAdapter, config: MetadataIndexConfig = {}, options: MetadataIndexOptions = {}) { this.storage = storage this.config = { @@ -218,11 +227,24 @@ export class MetadataIndexManager { // Initialize Field Type Inference this.fieldTypeInference = new FieldTypeInference(storage) + // Create column store — works immediately for writes (in-memory tail buffers). + // Storage discovery (loading existing segments) happens in init(). + this.columnStore = new ColumnStore() + // Removed lazyLoadCounts() call from constructor // It was a race condition (not awaited) and read from wrong source. // Now properly called in init() after warmCache() loads the sparse index. } + /** + * Get the shared EntityIdMapper instance. Used by the ColumnStore and + * other subsystems that need UUID ↔ u32 mapping without creating a + * second mapper that could diverge. + */ + getIdMapper(): EntityIdMapper { + return this.idMapper + } + /** * Initialize the metadata index manager * This must be called after construction and before any queries @@ -238,6 +260,15 @@ export class MetadataIndexManager { // Initialize EntityIdMapper (loads UUID ↔ integer mappings from storage) await this.idMapper.init() + // Initialize column store storage discovery (load existing segment manifests). + // The column store was created in the constructor for immediate writes; + // this step loads persisted segments so queries can find existing data. + try { + await this.columnStore.init(this.storage, this.idMapper) + } catch (err) { + prodLog.warn('[MetadataIndex] Column store storage discovery failed:', err) + } + // Check if field registry was loaded successfully const hasFields = this.fieldIndexes.size > 0 @@ -981,44 +1012,39 @@ export class MetadataIndexManager { * @param fieldValuePairs Array of field-value pairs to intersect * @returns Array of UUID strings matching ALL criteria */ + /** + * Multi-field intersection query: find entities matching ALL field-value pairs. + * + * Collects roaring bitmaps for each pair via the column store, then + * intersects them using hardware-accelerated AND operations. + * + * @param fieldValuePairs - Array of { field, value } to intersect + * @returns Array of entity UUID strings matching ALL pairs + */ async getIdsForMultipleFields(fieldValuePairs: Array<{ field: string; value: any }>): Promise { - if (fieldValuePairs.length === 0) { - return [] - } - - // Fast path: single field query + if (fieldValuePairs.length === 0) return [] if (fieldValuePairs.length === 1) { - const { field, value } = fieldValuePairs[0] - return await this.getIds(field, value) + return await this.getIds(fieldValuePairs[0].field, fieldValuePairs[0].value) } - // Collect roaring bitmaps for each field-value pair + // Collect roaring bitmaps for each field-value pair via column store const bitmaps: RoaringBitmap32[] = [] - for (const { field, value } of fieldValuePairs) { - const bitmap = await this.getBitmapFromChunks(field, value) - if (!bitmap || bitmap.size === 0) { - // Short circuit: if any field has no matches, intersection is empty - return [] - } + const bitmap = this.columnStore.hasField(field) + ? await this.columnStore.filter(field, value) + : await this.getBitmapFromChunks(field, value) ?? new RoaringBitmap32() + + if (bitmap.size === 0) return [] // Short circuit: empty intersection bitmaps.push(bitmap) } - // Hardware-accelerated intersection using SIMD instructions (AVX2/SSE4.2) - // This is 500-900x faster than JavaScript array filtering - // Note: RoaringBitmap32.and() only takes 2 params, so we reduce manually - let intersectionBitmap = bitmaps[0] + // Intersect all bitmaps (SIMD-accelerated roaring AND) + let result = bitmaps[0] for (let i = 1; i < bitmaps.length; i++) { - intersectionBitmap = RoaringBitmap32.and(intersectionBitmap, bitmaps[i]) + result = RoaringBitmap32.and(result, bitmaps[i]) } - // Check if empty before converting - if (intersectionBitmap.size === 0) { - return [] - } - - // Convert final bitmap to UUIDs (only once, not per-field) - return this.idMapper.intsIterableToUuids(intersectionBitmap) + return result.size > 0 ? this.idMapper.intsIterableToUuids(result) : [] } /** @@ -1168,6 +1194,19 @@ export class MetadataIndexManager { /** * Get IDs matching a range query using zone maps */ + /** + * Range query: find all entity IDs where a field's value falls within a range. + * + * Routes through the column store for O(log n) binary search when available, + * falls back to sparse index zone-map scan for fields not yet in the column store. + * + * @param field - Field name to query + * @param min - Lower bound (undefined = no lower bound) + * @param max - Upper bound (undefined = no upper bound) + * @param includeMin - Whether to include the lower bound (default: true) + * @param includeMax - Whether to include the upper bound (default: true) + * @returns Array of matching entity UUID strings + */ private async getIdsForRange( field: string, min?: any, @@ -1181,7 +1220,14 @@ export class MetadataIndexManager { stats.rangeQueryCount++ } - // All fields use chunked sparse index with zone map optimization + // Column store path: O(log n) binary search on sorted column. + // Use raw values (no normalization) — column store stores exact values. + if (this.columnStore && this.columnStore.hasField(field)) { + const bitmap = await this.columnStore.rangeQuery(field, min, max) + return this.idMapper.intsIterableToUuids(bitmap) + } + + // Fallback: sparse index zone-map scan (legacy path) return await this.getIdsFromChunksForRange(field, min, max, includeMin, includeMax) } @@ -1579,33 +1625,29 @@ export class MetadataIndexManager { if (b.field === 'noun') return 1 return 0 }) - - // Track which fields we're updating for incremental sorted index maintenance - const updatedFields = new Set() - + + // Update statistics and tracking for each field for (let i = 0; i < fields.length; i++) { const { field, value } = fields[i] - - // All fields use chunked sparse indexing - await this.addToChunkedIndex(field, value, id) - - // Update statistics and tracking this.updateCardinalityStats(field, value, 'add') this.updateTypeFieldAffinity(id, field, value, 'add', entityOrMetadata) await this.updateFieldIndex(field, value, 1) - - // Yield to event loop every 5 fields to prevent blocking - if (i % 5 === 4) { - await this.yieldToEventLoop() - } } - // Flush all dirty chunks and sparse indices accumulated during this add operation - // This batches writes that were previously sequential per-field into a single concurrent flush - // During rebuild (deferWrites=true), defer flushes to periodic batch boundaries (every 5000 entities) - // to avoid N × flush I/O amplification. The rebuild() method handles periodic + final flushes. - if (!deferWrites) { - await this.flushDirtyMetadata() + // Write to column store — the single write path for all indexed data. + // Converts extracted fields into a map and feeds the column store. + if (this.columnStore) { + const entityIntId = this.idMapper.getOrAssign(id) + const fieldsMap: Record = {} + for (const { field, value } of fields) { + if (field === '__words__') { + if (!fieldsMap.__words__) fieldsMap.__words__ = [] + ;(fieldsMap.__words__ as number[]).push(value) + } else { + fieldsMap[field] = value + } + } + this.columnStore.addEntity(entityIntId, fieldsMap) } // Adaptive auto-flush based on usage patterns @@ -1679,65 +1721,30 @@ export class MetadataIndexManager { */ async removeFromIndex(id: string, metadata?: any): Promise { if (metadata) { - // Remove from specific field indexes const fields = this.extractIndexableFields(metadata) + // Update statistics and tracking for (const { field, value } of fields) { - // All fields use chunked sparse indexing - await this.removeFromChunkedIndex(field, value, id) - - // Update statistics and tracking this.updateCardinalityStats(field, value, 'remove') this.updateTypeFieldAffinity(id, field, value, 'remove', metadata) await this.updateFieldIndex(field, value, -1) - - // Invalidate cache this.metadataCache.invalidatePattern(`field_values_${field}`) } - - // Flush all dirty chunks and sparse indices accumulated during remove - await this.flushDirtyMetadata() - - // Clean up ID mapper — must happen AFTER bitmap removal since removeFromChunk - // calls idMapper.getInt(id) internally. Skipping this leaves deleted IDs in the - // idMapper universe, causing ne/exists:false queries to return deleted entities. - this.idMapper.remove(id) - await this.idMapper.flush() - } else { - // Remove from all indexes (slower, requires scanning all field indexes) - // This should be rare - prefer providing metadata when removing - // Scan via fieldIndexes, load sparse indices on-demand - prodLog.warn(`Removing ID ${id} without metadata requires scanning all fields (slow)`) - - // Scan all fields via fieldIndexes - for (const field of this.fieldIndexes.keys()) { - const sparseIndex = await this.loadSparseIndex(field) - if (sparseIndex) { - for (const chunkId of sparseIndex.getAllChunkIds()) { - const chunk = await this.chunkManager.loadChunk(field, chunkId) - if (chunk) { - // Convert UUID to integer for bitmap checking - const intId = this.idMapper.getInt(id) - if (intId !== undefined) { - // Check all values in this chunk - for (const [value, bitmap] of chunk.entries) { - if (bitmap.has(intId)) { - await this.removeFromChunkedIndex(field, value, id) - } - } - } - } - } - } - } - - // Flush all dirty chunks and sparse indices accumulated during scan-remove - await this.flushDirtyMetadata() - - // Clean up ID mapper — must happen AFTER bitmap removal (same reason as fast path above) - this.idMapper.remove(id) - await this.idMapper.flush() } + + // Remove from column store (global deleted bitmap) + if (this.columnStore) { + const intId = this.idMapper.getInt(id) + if (intId !== undefined) { + this.columnStore.removeEntity(intId) + } + } + + // Clean up ID mapper — must happen AFTER column store removal since it uses + // idMapper.getInt(id). Prevents deleted IDs from persisting in the mapper + // universe, which would cause ne/exists:false queries to return deleted entities. + this.idMapper.remove(id) + await this.idMapper.flush() } /** @@ -1771,6 +1778,16 @@ export class MetadataIndexManager { /** * Get IDs for a specific field-value combination using chunked sparse index */ + /** + * Point query: find all entity IDs where a field has a specific value. + * + * Routes through the column store for O(log n) binary search when available, + * falls back to sparse index scan for fields not yet in the column store. + * + * @param field - Field name to query + * @param value - Exact value to match + * @returns Array of matching entity UUID strings + */ async getIds(field: string, value: any): Promise { // Track exact query for field statistics if (this.fieldStats.has(field)) { @@ -1778,7 +1795,15 @@ export class MetadataIndexManager { stats.exactQueryCount++ } - // All fields use chunked sparse indexing + // Column store path: O(log n) binary search + roaring bitmap. + // Use raw value (no normalization) — the column store stores exact values, + // not the bucketed/stringified format the sparse index uses. + if (this.columnStore && this.columnStore.hasField(field)) { + const bitmap = await this.columnStore.filter(field, value) + return this.idMapper.intsIterableToUuids(bitmap) + } + + // Fallback: sparse index scan (legacy path during migration) return await this.getIdsFromChunks(field, value) } @@ -2059,111 +2084,44 @@ export class MetadataIndexManager { // ===== EXISTENCE OPERATOR ===== // exists: boolean - check if field exists (any value) - case 'exists': + case 'exists': { + // Column store path: rangeQuery with no bounds returns all IDs for the field + const existsBitmap = (this.columnStore && this.columnStore.hasField(field)) + ? await this.columnStore.rangeQuery(field) + : await this.getExistsBitmapLegacy(field) + if (operand) { - // exists: true - Get all IDs that have this field (any value) - // From chunked sparse index with roaring bitmaps - // Now fully lazy-loaded via UnifiedCache (no local sparseIndices Map) - const allIntIds = new Set() - - // Load sparse index via UnifiedCache (lazy loading) - const sparseIndex = await this.loadSparseIndex(field) - if (sparseIndex) { - // Iterate through all chunks for this field - for (const chunkId of sparseIndex.getAllChunkIds()) { - const chunk = await this.chunkManager.loadChunk(field, chunkId) - if (chunk) { - // Collect all integer IDs from all roaring bitmaps in this chunk - for (const bitmap of chunk.entries.values()) { - for (const intId of bitmap) { - allIntIds.add(intId) - } - } - } - } - } - - // Convert integer IDs back to UUIDs - fieldResults = this.idMapper.intsIterableToUuids(allIntIds) + // exists: true — entities that HAVE this field + fieldResults = this.idMapper.intsIterableToUuids(existsBitmap) } else { - // exists: false - Get all IDs that DON'T have this field - // Uses EntityIdMapper universe (in-memory) instead of getAllIds() storage scan - const existsIntIds = new Set() - - // Get IDs that HAVE this field - const sparseIndex = await this.loadSparseIndex(field) - if (sparseIndex) { - for (const chunkId of sparseIndex.getAllChunkIds()) { - const chunk = await this.chunkManager.loadChunk(field, chunkId) - if (chunk) { - for (const bitmap of chunk.entries.values()) { - for (const intId of bitmap) { - existsIntIds.add(intId) - } - } - } - } - } - - // Use EntityIdMapper universe and subtract IDs that have this field + // exists: false — entities that DON'T have this field const allKnownIds = this.idMapper.intsIterableToUuids(this.idMapper.getAllIntIds()) - const existsUuids = this.idMapper.intsIterableToUuids(existsIntIds) + const existsUuids = this.idMapper.intsIterableToUuids(existsBitmap) const existsSet = new Set(existsUuids) fieldResults = allKnownIds.filter(id => !existsSet.has(id)) } break + } // ===== MISSING OPERATOR ===== - // missing: boolean - equivalent to exists: !boolean (for compatibility with metadataFilter.ts) - case 'missing': - // missing: true is equivalent to exists: false - // missing: false is equivalent to exists: true - // Added for API consistency with in-memory metadataFilter + // missing: boolean - equivalent to exists: !boolean + case 'missing': { + const missingBitmap = (this.columnStore && this.columnStore.hasField(field)) + ? await this.columnStore.rangeQuery(field) + : await this.getExistsBitmapLegacy(field) + if (operand) { - // missing: true - field does NOT exist (same as exists: false) - // Uses EntityIdMapper universe (in-memory) instead of getAllIds() storage scan - const existsIntIds = new Set() - - const sparseIndex = await this.loadSparseIndex(field) - if (sparseIndex) { - for (const chunkId of sparseIndex.getAllChunkIds()) { - const chunk = await this.chunkManager.loadChunk(field, chunkId) - if (chunk) { - for (const bitmap of chunk.entries.values()) { - for (const intId of bitmap) { - existsIntIds.add(intId) - } - } - } - } - } - - // Use EntityIdMapper universe and subtract IDs that have this field + // missing: true — entities that DON'T have this field (same as exists: false) const allKnownIds = this.idMapper.intsIterableToUuids(this.idMapper.getAllIntIds()) - const existsUuids = this.idMapper.intsIterableToUuids(existsIntIds) + const existsUuids = this.idMapper.intsIterableToUuids(missingBitmap) const existsSet = new Set(existsUuids) fieldResults = allKnownIds.filter(id => !existsSet.has(id)) } else { - // missing: false - field DOES exist (same as exists: true) - const allIntIds = new Set() - - const sparseIndex = await this.loadSparseIndex(field) - if (sparseIndex) { - for (const chunkId of sparseIndex.getAllChunkIds()) { - const chunk = await this.chunkManager.loadChunk(field, chunkId) - if (chunk) { - for (const bitmap of chunk.entries.values()) { - for (const intId of bitmap) { - allIntIds.add(intId) - } - } - } - } - } - - fieldResults = this.idMapper.intsIterableToUuids(allIntIds) + // missing: false — entities that HAVE this field (same as exists: true) + fieldResults = this.idMapper.intsIterableToUuids(missingBitmap) } break + } } } } else { @@ -2192,6 +2150,33 @@ export class MetadataIndexManager { return Array.from(resultSet) } + /** + * Legacy helper: get all entity int IDs that have any value for a field, + * using the sparse index. Used by exists/missing operators when the + * column store doesn't have data for the field. + * + * @param field - Field name + * @returns Roaring bitmap of entity int IDs (or iterable for compatibility) + * @private + */ + private async getExistsBitmapLegacy(field: string): Promise> { + const allIntIds = new Set() + const sparseIndex = await this.loadSparseIndex(field) + if (sparseIndex) { + for (const chunkId of sparseIndex.getAllChunkIds()) { + const chunk = await this.chunkManager.loadChunk(field, chunkId) + if (chunk) { + for (const bitmap of chunk.entries.values()) { + for (const intId of bitmap) { + allIntIds.add(intId) + } + } + } + } + } + return allIntIds + } + /** * Get filtered IDs sorted by a field (production-scale sorting) * @@ -2230,45 +2215,62 @@ export class MetadataIndexManager { orderBy: string, order: 'asc' | 'desc' = 'asc' ): Promise { - // 1. Get filtered IDs using existing roaring bitmap implementation (fast!) - // - // NOTE: This method REQUIRES a non-empty filter. Unfiltered sort over all - // entities is not supported here because it would require O(N) storage reads - // on bucketed fields (timestamps), which does not scale. The proper solution - // is a dedicated time-ordered segment index — tracked as Track 2. - // Callers should enforce a filter (type / where / excludeVFS) before calling. + // Column store path: O(K log S) sort via k-way merge across segments. + // No per-entity storage reads, no precision loss from bucketing. + if (this.columnStore && this.columnStore.hasField(orderBy)) { + // Get filtered IDs from existing roaring bitmap path + const hasFilter = filter && Object.keys(filter).length > 0 + const filteredIds = hasFilter ? await this.getIdsForFilter(filter) : [] + + if (hasFilter && filteredIds.length === 0) return [] + + let sortedIntIds: number[] + if (hasFilter) { + // Build filter bitmap for the column store + const filterBitmap = new RoaringBitmap32() + for (const id of filteredIds) { + const intId = this.idMapper.getInt(id) + if (intId !== undefined) filterBitmap.add(intId) + } + sortedIntIds = await this.columnStore.filteredSortTopK( + filterBitmap, orderBy, order, filteredIds.length + ) + } else { + // Unfiltered sort — column store handles the full entity set efficiently + sortedIntIds = await this.columnStore.sortTopK( + orderBy, order, this.idMapper.size + ) + } + + // Convert int IDs back to UUIDs + return sortedIntIds + .map(intId => this.idMapper.getUuid(intId)) + .filter((uuid): uuid is string => uuid !== undefined) + } + + // Fallback: sparse index path (for fields not yet in column store). + // Requires a non-empty filter because it reads O(k) entity values from storage. const filteredIds = await this.getIdsForFilter(filter) if (filteredIds.length === 0) { return [] } - // 2. Load sort field values for filtered IDs ONLY - // This is O(k) not O(n) where k = filtered count - // We only load the ONE field needed for sorting, not full entities const idValuePairs: Array<{ id: string, value: any }> = [] - for (const id of filteredIds) { const value = await this.getFieldValueForEntity(id, orderBy) idValuePairs.push({ id, value }) } - // 3. Sort by value (in-memory BUT only IDs + sort values) - // This is acceptable because we're sorting the FILTERED set, not all entities - // Even 1M filtered results = ~50MB (IDs + values), manageable in-memory idValuePairs.sort((a, b) => { - // Handle null/undefined (always sort to end) if (a.value == null && b.value == null) return 0 if (a.value == null) return order === 'asc' ? 1 : -1 if (b.value == null) return order === 'asc' ? -1 : 1 - - // Compare values if (a.value === b.value) return 0 const comparison = a.value < b.value ? -1 : 1 return order === 'asc' ? comparison : -comparison }) - // 4. Return sorted IDs (caller handles pagination BEFORE loading entities) return idValuePairs.map(p => p.id) } @@ -2494,6 +2496,11 @@ export class MetadataIndexManager { this.dirtyFields.clear() this.lastFlushTime = Date.now() + + // Flush column store tail buffers to L0 segments + if (this.columnStore) { + await this.columnStore.flush() + } } /** diff --git a/tests/integration/orderby-sort-bug.test.ts b/tests/integration/orderby-sort-bug.test.ts index 76128c28..02ca2ded 100644 --- a/tests/integration/orderby-sort-bug.test.ts +++ b/tests/integration/orderby-sort-bug.test.ts @@ -125,16 +125,25 @@ describe('find({ orderBy }) sort bug regression', () => { }) /** - * Unfiltered sort is explicitly rejected with a clear error pointing - * callers at the right pattern. This guards against silent N-scans - * on production workloads. + * Unfiltered sort now works via the unified column store. + * This was the Track 2 motivating use case — previously threw an error. */ - it('orderBy without any filter throws a helpful error', async () => { - await brain.add({ data: 'anything', type: NounType.Concept }) + it('orderBy without filter works via column store', async () => { + const id1 = await brain.add({ data: 'first', type: NounType.Concept }) + await new Promise((r) => setTimeout(r, 20)) + await brain.add({ data: 'second', type: NounType.Concept }) + await new Promise((r) => setTimeout(r, 20)) + const id3 = await brain.add({ data: 'third', type: NounType.Concept }) - await expect( - brain.find({ orderBy: 'createdAt', order: 'desc', limit: 1 }) - ).rejects.toThrow(/requires a filter/) + const results = await brain.find({ + orderBy: 'createdAt', + order: 'desc', + limit: 2 + }) + + expect(results.length).toBeGreaterThanOrEqual(2) + // Newest should be first (desc order) + expect(results[0].id).toBe(id3) }) }) diff --git a/tests/unit/indexes/columnStore/column-store.test.ts b/tests/unit/indexes/columnStore/column-store.test.ts new file mode 100644 index 00000000..73ef14ed --- /dev/null +++ b/tests/unit/indexes/columnStore/column-store.test.ts @@ -0,0 +1,295 @@ +/** + * @module column-store.test + * @description Tests for the ColumnStore coordinator: full lifecycle including + * add, flush, reload, filter, sort, range query, and removal. + */ + +import { describe, it, expect, beforeEach, afterEach } from 'vitest' +import { ColumnStore } from '../../../../src/indexes/columnStore/ColumnStore.js' +import { MemoryStorage } from '../../../../src/storage/adapters/memoryStorage.js' +import { EntityIdMapper } from '../../../../src/utils/entityIdMapper.js' +import { RoaringBitmap32 } from '../../../../src/utils/roaring/index.js' + +describe('ColumnStore', () => { + let storage: MemoryStorage + let idMapper: EntityIdMapper + let store: ColumnStore + + beforeEach(async () => { + storage = new MemoryStorage() + await storage.init() + idMapper = new EntityIdMapper({ storage, storageKey: 'test:idMapper' }) + await idMapper.init() + + store = new ColumnStore({ flushThreshold: 10 }) // low threshold for tests + await store.init(storage, idMapper) + }) + + afterEach(async () => { + await store.close() + }) + + // ----------------------------------------------------------------------- + // Basic add + sort + // ----------------------------------------------------------------------- + + describe('sortTopK', () => { + it('returns entities sorted by numeric field ascending', async () => { + store.addEntity(idMapper.getOrAssign('a'), { createdAt: 300 }) + store.addEntity(idMapper.getOrAssign('b'), { createdAt: 100 }) + store.addEntity(idMapper.getOrAssign('c'), { createdAt: 200 }) + await store.flush() + + const result = await store.sortTopK('createdAt', 'asc', 10) + const uuids = result.map(id => idMapper.getUuid(id)) + expect(uuids).toEqual(['b', 'c', 'a']) + }) + + it('returns entities sorted by numeric field descending', async () => { + store.addEntity(idMapper.getOrAssign('a'), { createdAt: 300 }) + store.addEntity(idMapper.getOrAssign('b'), { createdAt: 100 }) + store.addEntity(idMapper.getOrAssign('c'), { createdAt: 200 }) + await store.flush() + + const result = await store.sortTopK('createdAt', 'desc', 10) + const uuids = result.map(id => idMapper.getUuid(id)) + expect(uuids).toEqual(['a', 'c', 'b']) + }) + + it('respects limit K', async () => { + for (let i = 0; i < 50; i++) { + store.addEntity(idMapper.getOrAssign(`e${i}`), { createdAt: i * 1000 }) + } + await store.flush() + + const result = await store.sortTopK('createdAt', 'desc', 5) + expect(result).toHaveLength(5) + // Should be the 5 highest timestamps + const uuids = result.map(id => idMapper.getUuid(id)) + expect(uuids).toEqual(['e49', 'e48', 'e47', 'e46', 'e45']) + }) + + it('sorts string fields lexicographically', async () => { + store.addEntity(idMapper.getOrAssign('a'), { status: 'pending' }) + store.addEntity(idMapper.getOrAssign('b'), { status: 'active' }) + store.addEntity(idMapper.getOrAssign('c'), { status: 'inactive' }) + await store.flush() + + const result = await store.sortTopK('status', 'asc', 10) + const uuids = result.map(id => idMapper.getUuid(id)) + expect(uuids).toEqual(['b', 'c', 'a']) // active, inactive, pending + }) + + it('returns empty for unknown field', async () => { + const result = await store.sortTopK('nonexistent', 'asc', 10) + expect(result).toEqual([]) + }) + }) + + // ----------------------------------------------------------------------- + // Filtered sort + // ----------------------------------------------------------------------- + + describe('filteredSortTopK', () => { + it('returns only entities in the filter bitmap, sorted', async () => { + const idA = idMapper.getOrAssign('a') + const idB = idMapper.getOrAssign('b') + const idC = idMapper.getOrAssign('c') + const idD = idMapper.getOrAssign('d') + + store.addEntity(idA, { createdAt: 400 }) + store.addEntity(idB, { createdAt: 100 }) + store.addEntity(idC, { createdAt: 300 }) + store.addEntity(idD, { createdAt: 200 }) + await store.flush() + + // Filter: only B and C + const bitmap = new RoaringBitmap32([idB, idC]) + const result = await store.filteredSortTopK(bitmap, 'createdAt', 'desc', 10) + const uuids = result.map(id => idMapper.getUuid(id)) + expect(uuids).toEqual(['c', 'b']) // 300, 100 + }) + }) + + // ----------------------------------------------------------------------- + // Point filter + // ----------------------------------------------------------------------- + + describe('filter', () => { + it('returns matching entities as roaring bitmap', async () => { + const idA = idMapper.getOrAssign('a') + const idB = idMapper.getOrAssign('b') + const idC = idMapper.getOrAssign('c') + + store.addEntity(idA, { status: 'active' }) + store.addEntity(idB, { status: 'inactive' }) + store.addEntity(idC, { status: 'active' }) + await store.flush() + + const bitmap = await store.filter('status', 'active') + expect(bitmap.has(idA)).toBe(true) + expect(bitmap.has(idC)).toBe(true) + expect(bitmap.has(idB)).toBe(false) + }) + + it('returns empty bitmap for no match', async () => { + store.addEntity(idMapper.getOrAssign('a'), { status: 'active' }) + await store.flush() + + const bitmap = await store.filter('status', 'deleted') + expect(bitmap.size).toBe(0) + }) + }) + + // ----------------------------------------------------------------------- + // Range query + // ----------------------------------------------------------------------- + + describe('rangeQuery', () => { + it('returns entities within numeric range', async () => { + const ids: number[] = [] + for (let i = 0; i < 10; i++) { + ids.push(idMapper.getOrAssign(`e${i}`)) + store.addEntity(ids[i], { price: i * 10 }) + } + await store.flush() + + const bitmap = await store.rangeQuery('price', 30, 70) + // price 30, 40, 50, 60, 70 → entities e3, e4, e5, e6, e7 + expect(bitmap.size).toBe(5) + expect(bitmap.has(ids[3])).toBe(true) + expect(bitmap.has(ids[7])).toBe(true) + expect(bitmap.has(ids[2])).toBe(false) + expect(bitmap.has(ids[8])).toBe(false) + }) + }) + + // ----------------------------------------------------------------------- + // Removal + // ----------------------------------------------------------------------- + + describe('removeEntity', () => { + it('excluded from sort after removal', async () => { + const idA = idMapper.getOrAssign('a') + const idB = idMapper.getOrAssign('b') + const idC = idMapper.getOrAssign('c') + + store.addEntity(idA, { createdAt: 100 }) + store.addEntity(idB, { createdAt: 200 }) + store.addEntity(idC, { createdAt: 300 }) + await store.flush() + + store.removeEntity(idB) + await store.flush() + + const result = await store.sortTopK('createdAt', 'asc', 10) + const uuids = result.map(id => idMapper.getUuid(id)) + expect(uuids).toEqual(['a', 'c']) + }) + }) + + // ----------------------------------------------------------------------- + // Multi-segment merge sort + // ----------------------------------------------------------------------- + + describe('multi-segment', () => { + it('sorts correctly across multiple L0 segments', async () => { + // Flush threshold is 10, so 25 entities creates 3 segments + for (let i = 0; i < 25; i++) { + store.addEntity(idMapper.getOrAssign(`e${i}`), { ts: i }) + } + await store.flush() + + const result = await store.sortTopK('ts', 'desc', 5) + const uuids = result.map(id => idMapper.getUuid(id)) + expect(uuids).toEqual(['e24', 'e23', 'e22', 'e21', 'e20']) + }) + }) + + // ----------------------------------------------------------------------- + // Persistence (flush + reload) + // ----------------------------------------------------------------------- + + describe('persistence', () => { + it('survives close + reload', async () => { + store.addEntity(idMapper.getOrAssign('a'), { createdAt: 300 }) + store.addEntity(idMapper.getOrAssign('b'), { createdAt: 100 }) + store.addEntity(idMapper.getOrAssign('c'), { createdAt: 200 }) + await store.flush() + await store.close() + + // Create a new store pointing at the same storage + const store2 = new ColumnStore({ flushThreshold: 10 }) + await store2.init(storage, idMapper) + + const result = await store2.sortTopK('createdAt', 'desc', 10) + const uuids = result.map(id => idMapper.getUuid(id)) + expect(uuids).toEqual(['a', 'c', 'b']) + + await store2.close() + }) + }) + + // ----------------------------------------------------------------------- + // Multi-value fields (words) + // ----------------------------------------------------------------------- + + describe('multi-value fields', () => { + it('indexes multiple values per entity', async () => { + const idA = idMapper.getOrAssign('docA') + const idB = idMapper.getOrAssign('docB') + + store.addEntity(idA, { __words__: ['machine', 'learning', 'algorithm'] }) + store.addEntity(idB, { __words__: ['neural', 'network', 'machine'] }) + await store.flush() + + // Point filter: "machine" should match both + const machineBitmap = await store.filter('__words__', 'machine') + expect(machineBitmap.has(idA)).toBe(true) + expect(machineBitmap.has(idB)).toBe(true) + + // "algorithm" should match only docA + const algoBitmap = await store.filter('__words__', 'algorithm') + expect(algoBitmap.has(idA)).toBe(true) + expect(algoBitmap.has(idB)).toBe(false) + + // AND intersection: "machine" AND "neural" → only docB + const neuralBitmap = await store.filter('__words__', 'neural') + const intersection = RoaringBitmap32.and(machineBitmap, neuralBitmap) + expect(intersection.has(idB)).toBe(true) + expect(intersection.has(idA)).toBe(false) + }) + }) + + // ----------------------------------------------------------------------- + // getFilterValues + // ----------------------------------------------------------------------- + + describe('getFilterValues', () => { + it('returns distinct values sorted', async () => { + store.addEntity(idMapper.getOrAssign('a'), { status: 'pending' }) + store.addEntity(idMapper.getOrAssign('b'), { status: 'active' }) + store.addEntity(idMapper.getOrAssign('c'), { status: 'active' }) + store.addEntity(idMapper.getOrAssign('d'), { status: 'inactive' }) + await store.flush() + + const values = await store.getFilterValues('status') + expect(values).toEqual(['active', 'inactive', 'pending']) + }) + }) + + // ----------------------------------------------------------------------- + // hasField + // ----------------------------------------------------------------------- + + describe('hasField', () => { + it('returns false for unknown field', () => { + expect(store.hasField('nonexistent')).toBe(false) + }) + + it('returns true after adding data', async () => { + store.addEntity(idMapper.getOrAssign('a'), { createdAt: 100 }) + expect(store.hasField('createdAt')).toBe(true) + }) + }) +}) diff --git a/tests/unit/indexes/columnStore/manifest.test.ts b/tests/unit/indexes/columnStore/manifest.test.ts new file mode 100644 index 00000000..72d45563 --- /dev/null +++ b/tests/unit/indexes/columnStore/manifest.test.ts @@ -0,0 +1,162 @@ +/** + * @module manifest.test + * @description Tests for the per-field manifest manager. + */ + +import { describe, it, expect, beforeEach } from 'vitest' +import { ColumnManifest } from '../../../../src/indexes/columnStore/ColumnManifest.js' +import { ValueType, type SegmentMeta } from '../../../../src/indexes/columnStore/types.js' + +/** + * Minimal mock storage for manifest persistence tests. + * Only implements readObjectFromPath and writeObjectToPath. + */ +class MockStorage { + private data = new Map() + + async readObjectFromPath(path: string): Promise { + return this.data.get(path) ?? null + } + + async writeObjectToPath(path: string, obj: any): Promise { + this.data.set(path, JSON.parse(JSON.stringify(obj))) // deep clone + } +} + +describe('ColumnManifest', () => { + let storage: MockStorage + + beforeEach(() => { + storage = new MockStorage() + }) + + it('starts fresh when no stored manifest exists', async () => { + const manifest = new ColumnManifest('createdAt') + await manifest.load(storage as any) + + expect(manifest.isEmpty()).toBe(true) + expect(manifest.getAllSegments()).toEqual([]) + expect(manifest.totalCount).toBe(0) + }) + + it('save and reload preserves data', async () => { + const m1 = new ColumnManifest('createdAt') + m1.valueType = ValueType.Number + m1.addSegment({ + id: m1.nextSegmentId(), + level: 0, + count: 50000, + minValue: 1700000000000, + maxValue: 1700100000000, + file: 'L0-000001.cidx' + }) + await m1.save(storage as any) + + // Reload + const m2 = new ColumnManifest('createdAt') + await m2.load(storage as any) + + expect(m2.isEmpty()).toBe(false) + expect(m2.getAllSegments()).toHaveLength(1) + expect(m2.getAllSegments()[0].count).toBe(50000) + expect(m2.valueType).toBe(ValueType.Number) + }) + + it('nextSegmentId increments monotonically', () => { + const manifest = new ColumnManifest('test') + const id1 = manifest.nextSegmentId() + const id2 = manifest.nextSegmentId() + const id3 = manifest.nextSegmentId() + expect(id1).toBe(1) + expect(id2).toBe(2) + expect(id3).toBe(3) + }) + + it('addSegment keeps segments sorted by level then id', () => { + const manifest = new ColumnManifest('test') + manifest.addSegment({ id: 3, level: 1, count: 100, minValue: 0, maxValue: 100, file: 'L1-3.cidx' }) + manifest.addSegment({ id: 1, level: 0, count: 50, minValue: 0, maxValue: 50, file: 'L0-1.cidx' }) + manifest.addSegment({ id: 2, level: 0, count: 50, minValue: 50, maxValue: 100, file: 'L0-2.cidx' }) + + const segs = manifest.getAllSegments() + expect(segs[0].id).toBe(1) // L0, id=1 + expect(segs[1].id).toBe(2) // L0, id=2 + expect(segs[2].id).toBe(3) // L1, id=3 + }) + + it('removeSegments deletes by ID', () => { + const manifest = new ColumnManifest('test') + manifest.addSegment({ id: 1, level: 0, count: 50, minValue: 0, maxValue: 50, file: 'L0-1.cidx' }) + manifest.addSegment({ id: 2, level: 0, count: 50, minValue: 50, maxValue: 100, file: 'L0-2.cidx' }) + manifest.addSegment({ id: 3, level: 1, count: 100, minValue: 0, maxValue: 100, file: 'L1-3.cidx' }) + + manifest.removeSegments([1, 2]) + expect(manifest.getAllSegments()).toHaveLength(1) + expect(manifest.getAllSegments()[0].id).toBe(3) + }) + + it('getSegmentsAtLevel filters correctly', () => { + const manifest = new ColumnManifest('test') + manifest.addSegment({ id: 1, level: 0, count: 50, minValue: 0, maxValue: 50, file: 'L0-1.cidx' }) + manifest.addSegment({ id: 2, level: 0, count: 50, minValue: 50, maxValue: 100, file: 'L0-2.cidx' }) + manifest.addSegment({ id: 3, level: 1, count: 100, minValue: 0, maxValue: 100, file: 'L1-3.cidx' }) + + expect(manifest.getSegmentsAtLevel(0)).toHaveLength(2) + expect(manifest.getSegmentsAtLevel(1)).toHaveLength(1) + expect(manifest.getSegmentsAtLevel(2)).toHaveLength(0) + }) + + it('segmentPath generates correct paths', () => { + const manifest = new ColumnManifest('createdAt', '_column_index') + expect(manifest.segmentPath(0, 1)).toBe('_column_index/createdAt/L0-000001.cidx') + expect(manifest.segmentPath(1, 42)).toBe('_column_index/createdAt/L1-000042.cidx') + expect(manifest.segmentPath(0, 999999)).toBe('_column_index/createdAt/L0-999999.cidx') + }) + + it('manifestPath generates correct path', () => { + const manifest = new ColumnManifest('status', '_column_index') + expect(manifest.manifestPath()).toBe('_column_index/status/MANIFEST.json') + }) + + it('totalCount sums across segments', () => { + const manifest = new ColumnManifest('test') + manifest.addSegment({ id: 1, level: 0, count: 100, minValue: 0, maxValue: 0, file: 'a' }) + manifest.addSegment({ id: 2, level: 0, count: 200, minValue: 0, maxValue: 0, file: 'b' }) + manifest.addSegment({ id: 3, level: 1, count: 300, minValue: 0, maxValue: 0, file: 'c' }) + expect(manifest.totalCount).toBe(600) + }) + + it('version increments on each save', async () => { + const manifest = new ColumnManifest('test') + const v0 = manifest.getData().version + await manifest.save(storage as any) + const v1 = manifest.getData().version + await manifest.save(storage as any) + const v2 = manifest.getData().version + expect(v1).toBe(v0 + 1) + expect(v2).toBe(v1 + 1) + }) + + it('multiValue flag persists', async () => { + const m1 = new ColumnManifest('__words__') + m1.multiValue = true + m1.valueType = ValueType.String + await m1.save(storage as any) + + const m2 = new ColumnManifest('__words__') + await m2.load(storage as any) + expect(m2.multiValue).toBe(true) + expect(m2.valueType).toBe(ValueType.String) + }) + + it('buildComplete flag persists', async () => { + const m1 = new ColumnManifest('test') + expect(m1.buildComplete).toBe(false) + m1.buildComplete = true + await m1.save(storage as any) + + const m2 = new ColumnManifest('test') + await m2.load(storage as any) + expect(m2.buildComplete).toBe(true) + }) +}) diff --git a/tests/unit/indexes/columnStore/segment-cursor.test.ts b/tests/unit/indexes/columnStore/segment-cursor.test.ts new file mode 100644 index 00000000..778fc097 --- /dev/null +++ b/tests/unit/indexes/columnStore/segment-cursor.test.ts @@ -0,0 +1,270 @@ +/** + * @module segment-cursor.test + * @description Tests for the segment cursor: binary search, iteration, tombstones. + */ + +import { describe, it, expect } from 'vitest' +import { ColumnSegmentCursor, TailBufferCursor } from '../../../../src/indexes/columnStore/ColumnSegmentCursor.js' +import { ValueType, CIDX_MAGIC, CIDX_VERSION } from '../../../../src/indexes/columnStore/types.js' +import { RoaringBitmap32 } from '../../../../src/utils/roaring/index.js' + +function makeNumericCursor(values: number[], entityIds: number[], tombstones: number[] = []): ColumnSegmentCursor { + return new ColumnSegmentCursor( + { + magic: CIDX_MAGIC, + version: CIDX_VERSION, + fieldName: 'test', + fieldNameLength: 4, + valueType: ValueType.Number, + level: 0, + codec: 0, + flags: 0, + count: values.length + }, + values, + new Uint32Array(entityIds), + new RoaringBitmap32(tombstones) + ) +} + +function makeStringCursor(values: string[], entityIds: number[], tombstones: number[] = []): ColumnSegmentCursor { + return new ColumnSegmentCursor( + { + magic: CIDX_MAGIC, + version: CIDX_VERSION, + fieldName: 'status', + fieldNameLength: 6, + valueType: ValueType.String, + level: 0, + codec: 0, + flags: 0, + count: values.length + }, + values, + new Uint32Array(entityIds), + new RoaringBitmap32(tombstones) + ) +} + +describe('ColumnSegmentCursor', () => { + // ----------------------------------------------------------------------- + // Binary search + // ----------------------------------------------------------------------- + + describe('binarySearchValue (numeric)', () => { + it('finds exact match', () => { + const c = makeNumericCursor([100, 200, 300, 400, 500], [1, 2, 3, 4, 5]) + const r = c.binarySearchValue(300) + expect(r.found).toBe(true) + expect(r.index).toBe(2) + }) + + it('returns insertion point when not found', () => { + const c = makeNumericCursor([100, 200, 400, 500], [1, 2, 3, 4]) + const r = c.binarySearchValue(300) + expect(r.found).toBe(false) + expect(r.index).toBe(2) // would insert at position 2 + }) + + it('handles before-first', () => { + const c = makeNumericCursor([100, 200, 300], [1, 2, 3]) + const r = c.binarySearchValue(50) + expect(r.found).toBe(false) + expect(r.index).toBe(0) + }) + + it('handles after-last', () => { + const c = makeNumericCursor([100, 200, 300], [1, 2, 3]) + const r = c.binarySearchValue(999) + expect(r.found).toBe(false) + expect(r.index).toBe(3) + }) + + it('finds first occurrence of duplicates', () => { + const c = makeNumericCursor([100, 200, 200, 200, 300], [1, 2, 3, 4, 5]) + const r = c.binarySearchValue(200) + expect(r.found).toBe(true) + expect(r.index).toBe(1) // first occurrence + }) + }) + + describe('binarySearchValue (string)', () => { + it('finds exact match', () => { + const c = makeStringCursor(['active', 'inactive', 'pending'], [1, 2, 3]) + const r = c.binarySearchValue('inactive') + expect(r.found).toBe(true) + expect(r.index).toBe(1) + }) + + it('returns insertion point when not found', () => { + const c = makeStringCursor(['active', 'pending'], [1, 2]) + const r = c.binarySearchValue('inactive') + expect(r.found).toBe(false) + expect(r.index).toBe(1) + }) + }) + + // ----------------------------------------------------------------------- + // Range search + // ----------------------------------------------------------------------- + + describe('binarySearchRange', () => { + it('finds range with inclusive bounds', () => { + const c = makeNumericCursor([100, 200, 300, 400, 500], [1, 2, 3, 4, 5]) + const { startIdx, endIdx } = c.binarySearchRange(200, 400) + expect(startIdx).toBe(1) // 200 + expect(endIdx).toBe(4) // excludes 500 + }) + + it('handles range beyond data', () => { + const c = makeNumericCursor([100, 200, 300], [1, 2, 3]) + const { startIdx, endIdx } = c.binarySearchRange(0, 999) + expect(startIdx).toBe(0) + expect(endIdx).toBe(3) // all entries + }) + + it('handles empty range', () => { + const c = makeNumericCursor([100, 200, 300], [1, 2, 3]) + const { startIdx, endIdx } = c.binarySearchRange(150, 180) + expect(startIdx).toBe(1) // insertion point for 150 + expect(endIdx).toBe(1) // nothing between 150 and 180 + }) + }) + + // ----------------------------------------------------------------------- + // Point and range queries + // ----------------------------------------------------------------------- + + describe('getEntityIdsForValue', () => { + it('returns matching entity IDs', () => { + const c = makeNumericCursor([100, 200, 200, 300], [10, 20, 30, 40]) + expect(c.getEntityIdsForValue(200)).toEqual([20, 30]) + }) + + it('returns empty for no match', () => { + const c = makeNumericCursor([100, 200, 300], [10, 20, 30]) + expect(c.getEntityIdsForValue(999)).toEqual([]) + }) + + it('excludes tombstoned entries', () => { + const c = makeNumericCursor([100, 200, 200, 300], [10, 20, 30, 40], [1]) // position 1 tombstoned + expect(c.getEntityIdsForValue(200)).toEqual([30]) // position 1 (entityId=20) excluded + }) + }) + + describe('getEntityIdsInRange', () => { + it('returns entities in range', () => { + const c = makeNumericCursor([100, 200, 300, 400, 500], [1, 2, 3, 4, 5]) + expect(c.getEntityIdsInRange(200, 400)).toEqual([2, 3, 4]) + }) + + it('excludes tombstoned', () => { + const c = makeNumericCursor([100, 200, 300, 400, 500], [1, 2, 3, 4, 5], [2]) // position 2 (value=300) + expect(c.getEntityIdsInRange(200, 400)).toEqual([2, 4]) + }) + }) + + // ----------------------------------------------------------------------- + // Iteration + // ----------------------------------------------------------------------- + + describe('iterateForward', () => { + it('yields all entries in order', () => { + const c = makeNumericCursor([100, 200, 300], [1, 2, 3]) + const entries = [...c.iterateForward()] + expect(entries).toHaveLength(3) + expect(entries[0]).toEqual({ value: 100, entityIntId: 1, position: 0 }) + expect(entries[2]).toEqual({ value: 300, entityIntId: 3, position: 2 }) + }) + + it('skips tombstoned entries', () => { + const c = makeNumericCursor([100, 200, 300], [1, 2, 3], [1]) + const entries = [...c.iterateForward()] + expect(entries).toHaveLength(2) + expect(entries.map(e => e.entityIntId)).toEqual([1, 3]) + }) + + it('starts from offset', () => { + const c = makeNumericCursor([100, 200, 300, 400], [1, 2, 3, 4]) + const entries = [...c.iterateForward(2)] + expect(entries).toHaveLength(2) + expect(entries[0].value).toBe(300) + }) + }) + + describe('iterateBackward', () => { + it('yields all entries in reverse order', () => { + const c = makeNumericCursor([100, 200, 300], [1, 2, 3]) + const entries = [...c.iterateBackward()] + expect(entries).toHaveLength(3) + expect(entries[0]).toEqual({ value: 300, entityIntId: 3, position: 2 }) + expect(entries[2]).toEqual({ value: 100, entityIntId: 1, position: 0 }) + }) + + it('skips tombstoned entries', () => { + const c = makeNumericCursor([100, 200, 300], [1, 2, 3], [1]) + const entries = [...c.iterateBackward()] + expect(entries).toHaveLength(2) + expect(entries.map(e => e.entityIntId)).toEqual([3, 1]) + }) + }) + + // ----------------------------------------------------------------------- + // Zone map (min/max) + // ----------------------------------------------------------------------- + + describe('zone map', () => { + it('minValue returns first value', () => { + const c = makeNumericCursor([100, 200, 300], [1, 2, 3]) + expect(c.minValue).toBe(100) + }) + + it('maxValue returns last value', () => { + const c = makeNumericCursor([100, 200, 300], [1, 2, 3]) + expect(c.maxValue).toBe(300) + }) + + it('handles empty segment', () => { + const c = makeNumericCursor([], []) + expect(c.minValue).toBeUndefined() + expect(c.maxValue).toBeUndefined() + }) + }) + + describe('liveCount', () => { + it('subtracts tombstones from count', () => { + const c = makeNumericCursor([100, 200, 300], [1, 2, 3], [0, 2]) + expect(c.liveCount).toBe(1) + }) + }) + + // ----------------------------------------------------------------------- + // TailBufferCursor + // ----------------------------------------------------------------------- + + describe('TailBufferCursor', () => { + it('iterates forward', () => { + const c = new TailBufferCursor([10, 20, 30], new Uint32Array([1, 2, 3]), ValueType.Number) + const entries = [...c.iterateForward()] + expect(entries).toHaveLength(3) + expect(entries[0].value).toBe(10) + }) + + it('iterates backward', () => { + const c = new TailBufferCursor([10, 20, 30], new Uint32Array([1, 2, 3]), ValueType.Number) + const entries = [...c.iterateBackward()] + expect(entries[0].value).toBe(30) + }) + + it('point query returns matching IDs', () => { + const c = new TailBufferCursor([10, 20, 20, 30], new Uint32Array([1, 2, 3, 4]), ValueType.Number) + expect(c.getEntityIdsForValue(20)).toEqual([2, 3]) + }) + + it('min/max values', () => { + const c = new TailBufferCursor([10, 20, 30], new Uint32Array([1, 2, 3]), ValueType.Number) + expect(c.minValue).toBe(10) + expect(c.maxValue).toBe(30) + }) + }) +}) diff --git a/tests/unit/indexes/columnStore/segment-format.test.ts b/tests/unit/indexes/columnStore/segment-format.test.ts new file mode 100644 index 00000000..31f484bc --- /dev/null +++ b/tests/unit/indexes/columnStore/segment-format.test.ts @@ -0,0 +1,501 @@ +/** + * @module segment-format.test + * @description Round-trip tests for the .cidx binary segment format. + * The format is the shared contract between TypeScript and Rust — + * every byte must be identical across languages. + */ + +import { describe, it, expect } from 'vitest' +import { + writeSegmentToBuffer, + readSegmentFromBuffer, + writeHeader, + readHeader, + writeFooter, + readFooter, + crc32, + encodeNumericValues, + decodeNumericValues, + encodeFloatValues, + decodeFloatValues, + encodeStringValues, + decodeStringValues, + encodeEntityIds, + decodeEntityIds +} from '../../../../src/indexes/columnStore/ColumnSegmentFormat.js' +import { + CIDX_MAGIC, + CIDX_VERSION, + HEADER_SIZE, + FOOTER_SIZE, + ValueType, + FLAG_MULTI_VALUE +} from '../../../../src/indexes/columnStore/types.js' +import { RoaringBitmap32 } from '../../../../src/utils/roaring/index.js' + +describe('ColumnSegmentFormat', () => { + // ----------------------------------------------------------------------- + // CRC32 + // ----------------------------------------------------------------------- + + describe('crc32', () => { + it('produces consistent checksums', () => { + const data = Buffer.from('hello world') + const c1 = crc32(data) + const c2 = crc32(data) + expect(c1).toBe(c2) + expect(typeof c1).toBe('number') + expect(c1).toBeGreaterThan(0) + }) + + it('different data produces different checksums', () => { + const c1 = crc32(Buffer.from('hello')) + const c2 = crc32(Buffer.from('world')) + expect(c1).not.toBe(c2) + }) + + it('handles empty buffer', () => { + const c = crc32(Buffer.alloc(0)) + expect(typeof c).toBe('number') + }) + }) + + // ----------------------------------------------------------------------- + // Header + // ----------------------------------------------------------------------- + + describe('header', () => { + it('round-trips all fields', () => { + const header = { + magic: CIDX_MAGIC, + version: CIDX_VERSION, + fieldName: 'createdAt', + fieldNameLength: 9, + valueType: ValueType.Number, + level: 2, + codec: 0, + flags: 0, + count: 50000 + } + const buf = writeHeader(header) + expect(buf.length).toBe(HEADER_SIZE) + + const parsed = readHeader(buf) + expect(parsed.magic).toBe(CIDX_MAGIC) + expect(parsed.version).toBe(CIDX_VERSION) + expect(parsed.fieldName).toBe('createdAt') + expect(parsed.valueType).toBe(ValueType.Number) + expect(parsed.level).toBe(2) + expect(parsed.codec).toBe(0) + expect(parsed.flags).toBe(0) + expect(parsed.count).toBe(50000) + }) + + it('truncates field names longer than 32 bytes', () => { + const longName = 'a'.repeat(50) + const header = { + magic: CIDX_MAGIC, + version: CIDX_VERSION, + fieldName: longName, + fieldNameLength: 32, + valueType: ValueType.String, + level: 0, + codec: 0, + flags: 0, + count: 1 + } + const buf = writeHeader(header) + const parsed = readHeader(buf) + expect(parsed.fieldName.length).toBe(32) + }) + + it('preserves multi-value flag', () => { + const header = { + magic: CIDX_MAGIC, + version: CIDX_VERSION, + fieldName: '__words__', + fieldNameLength: 9, + valueType: ValueType.String, + level: 0, + codec: 0, + flags: FLAG_MULTI_VALUE, + count: 10000 + } + const buf = writeHeader(header) + const parsed = readHeader(buf) + expect(parsed.flags & FLAG_MULTI_VALUE).toBe(FLAG_MULTI_VALUE) + }) + + it('rejects invalid magic', () => { + const buf = Buffer.alloc(HEADER_SIZE) + buf.writeUInt32LE(0xDEADBEEF, 0) + expect(() => readHeader(buf)).toThrow(/Invalid segment magic/) + }) + + it('rejects unsupported version', () => { + const buf = Buffer.alloc(HEADER_SIZE) + buf.writeUInt32LE(CIDX_MAGIC, 0) + buf.writeUInt16LE(99, 4) // bad version + expect(() => readHeader(buf)).toThrow(/Unsupported segment version/) + }) + }) + + // ----------------------------------------------------------------------- + // Value encoding + // ----------------------------------------------------------------------- + + describe('numeric values', () => { + it('round-trips integers', () => { + const values = [100, 200, 300, 1700000000000] + const buf = encodeNumericValues(values) + expect(buf.length).toBe(values.length * 8) + const decoded = decodeNumericValues(buf, values.length) + expect(decoded).toEqual(values) + }) + + it('handles negative values', () => { + const values = [-1000, -1, 0, 1, 1000] + const decoded = decodeNumericValues(encodeNumericValues(values), values.length) + expect(decoded).toEqual(values) + }) + + it('handles empty array', () => { + const buf = encodeNumericValues([]) + expect(buf.length).toBe(0) + const decoded = decodeNumericValues(buf, 0) + expect(decoded).toEqual([]) + }) + }) + + describe('float values', () => { + it('round-trips floats', () => { + const values = [1.5, 2.7, 3.14159, 99.99] + const buf = encodeFloatValues(values) + const decoded = decodeFloatValues(buf, values.length) + expect(decoded).toEqual(values) + }) + + it('handles special values', () => { + const values = [-Infinity, -0, 0, Infinity] + const decoded = decodeFloatValues(encodeFloatValues(values), values.length) + expect(decoded[0]).toBe(-Infinity) + expect(decoded[3]).toBe(Infinity) + }) + }) + + describe('string values', () => { + it('round-trips strings', () => { + const values = ['active', 'inactive', 'pending'] + const buf = encodeStringValues(values) + const decoded = decodeStringValues(buf, values.length) + expect(decoded).toEqual(values) + }) + + it('handles empty strings', () => { + const values = ['', 'a', ''] + const decoded = decodeStringValues(encodeStringValues(values), values.length) + expect(decoded).toEqual(values) + }) + + it('handles unicode', () => { + const values = ['café', 'naïve', '日本語'] + const decoded = decodeStringValues(encodeStringValues(values), values.length) + expect(decoded).toEqual(values) + }) + + it('handles single string', () => { + const decoded = decodeStringValues(encodeStringValues(['hello']), 1) + expect(decoded).toEqual(['hello']) + }) + }) + + describe('entity IDs', () => { + it('round-trips Uint32Array', () => { + const ids = new Uint32Array([1, 42, 1000, 4294967295]) // includes max u32 + const buf = encodeEntityIds(ids) + expect(buf.length).toBe(ids.length * 4) + const decoded = decodeEntityIds(buf, ids.length) + expect(Array.from(decoded)).toEqual(Array.from(ids)) + }) + + it('round-trips number array', () => { + const ids = [5, 10, 15] + const decoded = decodeEntityIds(encodeEntityIds(ids), ids.length) + expect(Array.from(decoded)).toEqual(ids) + }) + }) + + // ----------------------------------------------------------------------- + // Full segment round-trip + // ----------------------------------------------------------------------- + + describe('full segment', () => { + it('round-trips a numeric segment', () => { + const values = [1700000000000, 1700000001000, 1700000002000] + const entityIds = new Uint32Array([10, 20, 30]) + + const buf = writeSegmentToBuffer( + { + fieldName: 'createdAt', + fieldNameLength: 9, + valueType: ValueType.Number, + level: 0, + codec: 0, + flags: 0, + count: 3 + }, + values, + entityIds + ) + + const parsed = readSegmentFromBuffer(buf) + expect(parsed.header.fieldName).toBe('createdAt') + expect(parsed.header.valueType).toBe(ValueType.Number) + expect(parsed.header.count).toBe(3) + expect(parsed.values).toEqual(values) + expect(Array.from(parsed.entityIds)).toEqual([10, 20, 30]) + expect(parsed.tombstones.size).toBe(0) + }) + + it('round-trips a string segment', () => { + const values = ['active', 'inactive', 'pending'] + const entityIds = new Uint32Array([1, 2, 3]) + + const buf = writeSegmentToBuffer( + { + fieldName: 'status', + fieldNameLength: 6, + valueType: ValueType.String, + level: 0, + codec: 0, + flags: 0, + count: 3 + }, + values, + entityIds + ) + + const parsed = readSegmentFromBuffer(buf) + expect(parsed.values).toEqual(values) + expect(Array.from(parsed.entityIds)).toEqual([1, 2, 3]) + }) + + it('round-trips a float segment', () => { + const values = [9.99, 19.99, 49.99] + const entityIds = new Uint32Array([100, 200, 300]) + + const buf = writeSegmentToBuffer( + { + fieldName: 'price', + fieldNameLength: 5, + valueType: ValueType.Float, + level: 1, + codec: 0, + flags: 0, + count: 3 + }, + values, + entityIds + ) + + const parsed = readSegmentFromBuffer(buf) + expect(parsed.values).toEqual(values) + expect(parsed.header.level).toBe(1) + }) + + it('round-trips a boolean segment', () => { + const values = [0, 0, 1, 1] + const entityIds = new Uint32Array([1, 2, 3, 4]) + + const buf = writeSegmentToBuffer( + { + fieldName: 'isActive', + fieldNameLength: 8, + valueType: ValueType.Boolean, + level: 0, + codec: 0, + flags: 0, + count: 4 + }, + values, + entityIds + ) + + const parsed = readSegmentFromBuffer(buf) + expect(parsed.values).toEqual(values) + }) + + it('round-trips with tombstones', () => { + const values = [100, 200, 300, 400, 500] + const entityIds = new Uint32Array([1, 2, 3, 4, 5]) + const tombstones = new RoaringBitmap32([1, 3]) // positions 1 and 3 are deleted + + const buf = writeSegmentToBuffer( + { + fieldName: 'score', + fieldNameLength: 5, + valueType: ValueType.Number, + level: 0, + codec: 0, + flags: 0, + count: 5 + }, + values, + entityIds, + tombstones + ) + + const parsed = readSegmentFromBuffer(buf) + expect(parsed.tombstones.has(1)).toBe(true) + expect(parsed.tombstones.has(3)).toBe(true) + expect(parsed.tombstones.has(0)).toBe(false) + expect(parsed.tombstones.has(2)).toBe(false) + expect(parsed.tombstones.size).toBe(2) + }) + + it('round-trips an empty segment', () => { + const buf = writeSegmentToBuffer( + { + fieldName: 'empty', + fieldNameLength: 5, + valueType: ValueType.Number, + level: 0, + codec: 0, + flags: 0, + count: 0 + }, + [], + new Uint32Array(0) + ) + + const parsed = readSegmentFromBuffer(buf) + expect(parsed.header.count).toBe(0) + expect(parsed.values).toEqual([]) + expect(parsed.entityIds.length).toBe(0) + }) + + it('round-trips a single-entry segment', () => { + const buf = writeSegmentToBuffer( + { + fieldName: 'x', + fieldNameLength: 1, + valueType: ValueType.Number, + level: 0, + codec: 0, + flags: 0, + count: 1 + }, + [42], + new Uint32Array([7]) + ) + + const parsed = readSegmentFromBuffer(buf) + expect(parsed.values).toEqual([42]) + expect(Array.from(parsed.entityIds)).toEqual([7]) + }) + + it('round-trips a multi-value segment (words)', () => { + const words = ['algorithm', 'learning', 'machine', 'neural'] + const entityIds = new Uint32Array([5, 5, 5, 10]) // entity 5 has 3 words, entity 10 has 1 + + const buf = writeSegmentToBuffer( + { + fieldName: '__words__', + fieldNameLength: 9, + valueType: ValueType.String, + level: 0, + codec: 0, + flags: FLAG_MULTI_VALUE, + count: 4 + }, + words, + entityIds + ) + + const parsed = readSegmentFromBuffer(buf) + expect(parsed.values).toEqual(words) + expect(Array.from(parsed.entityIds)).toEqual([5, 5, 5, 10]) + expect(parsed.header.flags & FLAG_MULTI_VALUE).toBe(FLAG_MULTI_VALUE) + }) + }) + + // ----------------------------------------------------------------------- + // CRC validation + // ----------------------------------------------------------------------- + + describe('CRC validation', () => { + it('detects tampered data', () => { + const buf = writeSegmentToBuffer( + { + fieldName: 'test', + fieldNameLength: 4, + valueType: ValueType.Number, + level: 0, + codec: 0, + flags: 0, + count: 3 + }, + [1, 2, 3], + new Uint32Array([10, 20, 30]) + ) + + // Tamper with a value byte + const tampered = Buffer.from(buf) + tampered[HEADER_SIZE + 4] ^= 0xFF + + expect(() => readSegmentFromBuffer(tampered, true)).toThrow(/CRC32 mismatch/) + }) + + it('passes with validateCrc=false on tampered data', () => { + const buf = writeSegmentToBuffer( + { + fieldName: 'test', + fieldNameLength: 4, + valueType: ValueType.Number, + level: 0, + codec: 0, + flags: 0, + count: 3 + }, + [1, 2, 3], + new Uint32Array([10, 20, 30]) + ) + + const tampered = Buffer.from(buf) + tampered[HEADER_SIZE + 4] ^= 0xFF + + // Should not throw with validation disabled + const parsed = readSegmentFromBuffer(tampered, false) + expect(parsed.header.count).toBe(3) + }) + }) + + // ----------------------------------------------------------------------- + // Error handling + // ----------------------------------------------------------------------- + + describe('error handling', () => { + it('rejects buffer smaller than header + footer', () => { + const tiny = Buffer.alloc(10) + expect(() => readSegmentFromBuffer(tiny)).toThrow(/Buffer too small/) + }) + + it('rejects mismatched values and entityIds length', () => { + expect(() => + writeSegmentToBuffer( + { + fieldName: 'x', + fieldNameLength: 1, + valueType: ValueType.Number, + level: 0, + codec: 0, + flags: 0, + count: 2 + }, + [1, 2], + new Uint32Array([10]) // mismatched! + ) + ).toThrow(/same length/) + }) + }) +}) diff --git a/tests/unit/indexes/columnStore/tail-buffer.test.ts b/tests/unit/indexes/columnStore/tail-buffer.test.ts new file mode 100644 index 00000000..0c4f6568 --- /dev/null +++ b/tests/unit/indexes/columnStore/tail-buffer.test.ts @@ -0,0 +1,169 @@ +/** + * @module tail-buffer.test + * @description Tests for the per-field in-memory write buffer. + */ + +import { describe, it, expect } from 'vitest' +import { ColumnTailBuffer } from '../../../../src/indexes/columnStore/ColumnTailBuffer.js' +import { ValueType, DEFAULT_FLUSH_THRESHOLD } from '../../../../src/indexes/columnStore/types.js' + +describe('ColumnTailBuffer', () => { + describe('numeric values', () => { + it('drain returns entries sorted by value', () => { + const buf = new ColumnTailBuffer('createdAt', ValueType.Number) + buf.add(300, 3) + buf.add(100, 1) + buf.add(200, 2) + + const { values, entityIds } = buf.drain() + expect(values).toEqual([100, 200, 300]) + expect(Array.from(entityIds)).toEqual([1, 2, 3]) + }) + + it('breaks ties by entityIntId ascending', () => { + const buf = new ColumnTailBuffer('ts', ValueType.Number) + buf.add(100, 30) + buf.add(100, 10) + buf.add(100, 20) + + const { entityIds } = buf.drain() + expect(Array.from(entityIds)).toEqual([10, 20, 30]) + }) + + it('handles negative values', () => { + const buf = new ColumnTailBuffer('offset', ValueType.Number) + buf.add(-10, 1) + buf.add(5, 2) + buf.add(-20, 3) + + const { values } = buf.drain() + expect(values).toEqual([-20, -10, 5]) + }) + }) + + describe('string values', () => { + it('drain returns entries sorted lexicographically', () => { + const buf = new ColumnTailBuffer('status', ValueType.String) + buf.add('pending', 3) + buf.add('active', 1) + buf.add('inactive', 2) + + const { values, entityIds } = buf.drain() + expect(values).toEqual(['active', 'inactive', 'pending']) + expect(Array.from(entityIds)).toEqual([1, 2, 3]) + }) + }) + + describe('boolean values', () => { + it('sorts false (0) before true (1)', () => { + const buf = new ColumnTailBuffer('isActive', ValueType.Boolean) + buf.add(1, 1) + buf.add(0, 2) + buf.add(1, 3) + buf.add(0, 4) + + const { values, entityIds } = buf.drain() + expect(values).toEqual([0, 0, 1, 1]) + expect(Array.from(entityIds)).toEqual([2, 4, 1, 3]) + }) + }) + + describe('removal', () => { + it('removes pending entries from buffer', () => { + const buf = new ColumnTailBuffer('x', ValueType.Number) + buf.add(100, 1) + buf.add(200, 2) + buf.add(300, 3) + + buf.remove(2) + + const { values, entityIds, pendingRemovals } = buf.drain() + expect(values).toEqual([100, 300]) + expect(Array.from(entityIds)).toEqual([1, 3]) + expect(pendingRemovals.has(2)).toBe(true) + }) + + it('tracks pending removals for existing segments', () => { + const buf = new ColumnTailBuffer('x', ValueType.Number) + buf.remove(42) + buf.remove(99) + + const { pendingRemovals } = buf.drain() + expect(pendingRemovals.has(42)).toBe(true) + expect(pendingRemovals.has(99)).toBe(true) + }) + }) + + describe('threshold', () => { + it('shouldFlush returns false below threshold', () => { + const buf = new ColumnTailBuffer('x', ValueType.Number, 3) + buf.add(1, 1) + buf.add(2, 2) + expect(buf.shouldFlush()).toBe(false) + }) + + it('shouldFlush returns true at threshold', () => { + const buf = new ColumnTailBuffer('x', ValueType.Number, 3) + buf.add(1, 1) + buf.add(2, 2) + buf.add(3, 3) + expect(buf.shouldFlush()).toBe(true) + }) + + it('uses default threshold when not specified', () => { + const buf = new ColumnTailBuffer('x', ValueType.Number) + expect(buf.threshold).toBe(DEFAULT_FLUSH_THRESHOLD) + }) + }) + + describe('drain clears buffer', () => { + it('buffer is empty after drain', () => { + const buf = new ColumnTailBuffer('x', ValueType.Number) + buf.add(1, 1) + buf.add(2, 2) + expect(buf.size).toBe(2) + + buf.drain() + expect(buf.size).toBe(0) + }) + + it('removals are cleared after drain', () => { + const buf = new ColumnTailBuffer('x', ValueType.Number) + buf.remove(42) + buf.drain() + + const { pendingRemovals } = buf.drain() + expect(pendingRemovals.size).toBe(0) + }) + }) + + describe('clear', () => { + it('clears entries and removals without returning data', () => { + const buf = new ColumnTailBuffer('x', ValueType.Number) + buf.add(1, 1) + buf.remove(99) + + buf.clear() + expect(buf.size).toBe(0) + + const { values, pendingRemovals } = buf.drain() + expect(values).toEqual([]) + expect(pendingRemovals.size).toBe(0) + }) + }) + + describe('multi-value (words)', () => { + it('supports multiple entries with the same entityIntId', () => { + const buf = new ColumnTailBuffer('__words__', ValueType.String) + buf.add('machine', 5) + buf.add('learning', 5) + buf.add('neural', 10) + buf.add('algorithm', 5) + + const { values, entityIds } = buf.drain() + // Sorted lexicographically + expect(values).toEqual(['algorithm', 'learning', 'machine', 'neural']) + expect(Array.from(entityIds)).toEqual([5, 5, 5, 10]) + }) + }) +}) diff --git a/tests/unit/utils/metadataIndex-automatic-bucketing.test.ts b/tests/unit/utils/metadataIndex-automatic-bucketing.test.ts deleted file mode 100644 index 8255847e..00000000 --- a/tests/unit/utils/metadataIndex-automatic-bucketing.test.ts +++ /dev/null @@ -1,297 +0,0 @@ -/** - * MetadataIndex Automatic Bucketing Tests (v3.41.0) - * Verify that temporal fields are automatically bucketed from the start - * to prevent file pollution while enabling range queries - */ - -import { describe, it, expect, beforeEach, afterEach } from 'vitest' -import { MemoryStorage } from '../../../src/storage/adapters/memoryStorage.js' -import { MetadataIndexManager } from '../../../src/utils/metadataIndex.js' - -describe('MetadataIndex Automatic Bucketing (v3.41.0)', () => { - let storage: MemoryStorage - let index: MetadataIndexManager - // Use a fixed base timestamp to avoid test pollution from Date.now() bucketing - const TEST_BASE_TIME = 1700000000000 // Fixed timestamp (Nov 14, 2023) - - beforeEach(() => { - storage = new MemoryStorage() - // No excludeFields config - fully automatic! - index = new MetadataIndexManager(storage, {}) - }) - - afterEach(async () => { - // MemoryStorage doesn't have a close() method - }) - - it('should automatically bucket timestamps from the first operation', async () => { - // Add 100 entities with unique timestamps (every millisecond) - const baseTime = TEST_BASE_TIME - for (let i = 0; i < 100; i++) { - await index.addToIndex(`entity${i}`, { - noun: 'character', - modified: baseTime + i // 100 unique timestamps - }) - } - - await index.flush() - - // Without bucketing: would create 100 metadata index files - // With bucketing: should create only ~2 files (all fit in one 1-minute bucket) - - // Get available values for 'modified' field - const modifiedValues = await index.getFilterValues('modified') - - // Should have very few unique values (all bucketed to 1-minute intervals) - expect(modifiedValues.length).toBeLessThan(5) // Much less than 100! - expect(modifiedValues.length).toBeGreaterThan(0) // But not zero - }) - - it('should prevent file pollution with high-cardinality timestamps', async () => { - // Simulate the bug scenario: 575 relationships with unique timestamps - const baseTime = TEST_BASE_TIME + 10000000 // Offset to avoid bucket collision with other tests - for (let i = 0; i < 575; i++) { - await index.addToIndex(`verb${i}`, { - verb: 'relates_to', - modified: baseTime + (i * 1000), // Every second - accessed: baseTime + (i * 500) // Every half-second - }) - } - - await index.flush() - - // Without bucketing: 575 unique modified + 1150 unique accessed = 1725 index files - // With bucketing: ~10 modified buckets + ~20 accessed buckets = ~30 index files - - const modifiedValues = await index.getFilterValues('modified') - const accessedValues = await index.getFilterValues('accessed') - - // Should have created FAR fewer index entries due to bucketing - expect(modifiedValues.length).toBeLessThan(100) // Not 575! - expect(accessedValues.length).toBeLessThan(200) // Not 1150! - - // But should still have indexed them (not excluded) - expect(modifiedValues.length).toBeGreaterThan(0) - expect(accessedValues.length).toBeGreaterThan(0) - }) - - it('should enable range queries on bucketed timestamp fields', async () => { - // Add entities with timestamps at exact bucket boundaries (every minute) - const bucketStart = Math.floor((TEST_BASE_TIME + 20000000) / 60000) * 60000 - for (let i = 0; i < 100; i++) { - await index.addToIndex(`entity${i}`, { - noun: 'character', - modified: bucketStart + (i * 60000) // Every minute = every bucket - }) - } - - await index.flush() - - // Query: modified >= (bucketStart + 30 minutes) - const thirtyMinutesLater = bucketStart + (30 * 60000) - const results = await index.getIdsForFilter({ - modified: { greaterThanOrEqual: thirtyMinutesLater } - }) - - // Should find entities 30-99 (70 entities) - // Note: With bucketing, we expect exactly 70 entities (30-99 inclusive) - expect(results.length).toBe(70) - - // Verify results are correct - results.forEach(id => { - const num = parseInt(id.replace('entity', '')) - expect(num).toBeGreaterThanOrEqual(30) // Should be in range - }) - }) - - it('should work with zero configuration (no excludeFields, no rangeQueryFields)', async () => { - // Create index with NO config at all - const autoIndex = new MetadataIndexManager(storage, {}) - const testTime = TEST_BASE_TIME + 30000000 // Offset to avoid bucket collision - - // Add entity with multiple field types - await autoIndex.addToIndex('test1', { - id: 'uuid-123', // Should be excluded (in default excludeFields) - noun: 'character', // Should be indexed (low cardinality) - modified: testTime, // Should be indexed + bucketed (temporal) - accessed: testTime, // Should be indexed + bucketed (temporal) - content: 'long text content', // Should be excluded (in default excludeFields) - customTimestamp: testTime // Should be indexed + bucketed (has 'time' in name) - }) - - await autoIndex.flush() - - // Verify 'modified' was indexed and bucketed - const modifiedValues = await autoIndex.getFilterValues('modified') - expect(modifiedValues.length).toBeGreaterThan(0) // Indexed! - - // Verify 'accessed' was indexed and bucketed - const accessedValues = await autoIndex.getFilterValues('accessed') - expect(accessedValues.length).toBeGreaterThan(0) // Indexed! - - // Verify 'customTimestamp' was indexed and bucketed - const customValues = await autoIndex.getFilterValues('customTimestamp') - expect(customValues.length).toBeGreaterThan(0) // Indexed! - - // Verify 'id' was excluded - const idValues = await autoIndex.getFilterValues('id') - expect(idValues.length).toBe(0) // Excluded! - - // Verify 'content' was excluded - const contentValues = await autoIndex.getFilterValues('content') - expect(contentValues.length).toBe(0) // Excluded! - }) - - it('should handle fields with "time" or "date" in their names', async () => { - // Add entities with various timestamp field names - const testTime = TEST_BASE_TIME + 40000000 // Offset to avoid bucket collision - await index.addToIndex('entity1', { - timestamp: testTime, - createdAt: testTime, - updatedAt: testTime, - birthdate: testTime, - eventTime: testTime, - lastSeen: testTime // Doesn't match pattern, won't be bucketed - }) - - await index.flush() - - // Fields with 'time' or 'date' should be bucketed - const timestampValues = await index.getFilterValues('timestamp') - const createdAtValues = await index.getFilterValues('createdAt') - const updatedAtValues = await index.getFilterValues('updatedAt') - const birthdateValues = await index.getFilterValues('birthdate') - const eventTimeValues = await index.getFilterValues('eventTime') - - // All should be indexed (not excluded) - expect(timestampValues.length).toBeGreaterThan(0) - expect(createdAtValues.length).toBeGreaterThan(0) - expect(updatedAtValues.length).toBeGreaterThan(0) - expect(birthdateValues.length).toBeGreaterThan(0) - expect(eventTimeValues.length).toBeGreaterThan(0) - }) - - it('should bucket timestamps to 1-minute intervals by default', async () => { - // Use a timestamp at the start of a minute bucket to ensure 30s apart stays in same bucket - const bucketStart = Math.floor((TEST_BASE_TIME + 50000000) / 60000) * 60000 - await index.addToIndex('entity1', { modified: bucketStart }) - await index.addToIndex('entity2', { modified: bucketStart + 30000 }) // +30 seconds, still in same bucket - - await index.flush() - - // Both should be in the same 1-minute bucket - const ids1 = await index.getIds('modified', bucketStart) - const ids2 = await index.getIds('modified', bucketStart + 30000) - - // Both queries should return both entities (same bucket) - expect(ids1.length).toBe(2) - expect(ids2.length).toBe(2) - expect(ids1.sort()).toEqual(ids2.sort()) - }) - - it('should create separate buckets for timestamps >1 minute apart', async () => { - // Use bucket boundaries to ensure entities are in different buckets - const bucket1Start = Math.floor((TEST_BASE_TIME + 60000000) / 60000) * 60000 - const bucket2Start = bucket1Start + 120000 // +2 minutes (definitely different bucket) - await index.addToIndex('entity1', { modified: bucket1Start }) - await index.addToIndex('entity2', { modified: bucket2Start }) - - await index.flush() - - // Should be in different 1-minute buckets - const ids1 = await index.getIds('modified', bucket1Start) - const ids2 = await index.getIds('modified', bucket2Start) - - // Each query should return only its own entity - expect(ids1).toContain('entity1') - expect(ids1).not.toContain('entity2') - expect(ids2).toContain('entity2') - expect(ids2).not.toContain('entity1') - }) - - it('should handle range queries across bucket boundaries', async () => { - // Use explicit bucket boundaries for predictable range queries - const bucket0 = Math.floor((TEST_BASE_TIME + 70000000) / 60000) * 60000 - const bucket1 = bucket0 + 60000 // +1 minute - const bucket2 = bucket0 + 120000 // +2 minutes - const bucket3 = bucket0 + 180000 // +3 minutes - - await index.addToIndex('entity0', { modified: bucket0 }) - await index.addToIndex('entity1', { modified: bucket1 }) - await index.addToIndex('entity2', { modified: bucket2 }) - await index.addToIndex('entity3', { modified: bucket3 }) - - await index.flush() - - // Query: modified > bucket1 (should get entity2 and entity3, not entity0 or entity1) - const results = await index.getIdsForFilter({ - modified: { greaterThan: bucket1 } - }) - - // Should include entities after bucket1 - expect(results).toContain('entity2') - expect(results).toContain('entity3') - - // Should not include entity0 or entity1 (at or before bucket1) - expect(results).not.toContain('entity0') - expect(results).not.toContain('entity1') - }) - - it('should not break non-temporal numeric fields', async () => { - // Add entities with regular numeric fields (not timestamps) - await index.addToIndex('entity1', { - noun: 'item', - price: 19.99, - quantity: 5, - rating: 4.5 - }) - - await index.addToIndex('entity2', { - noun: 'item', - price: 29.99, - quantity: 3, - rating: 4.7 - }) - - await index.flush() - - // Non-temporal numeric fields should NOT be bucketed - const priceValues = await index.getFilterValues('price') - const quantityValues = await index.getFilterValues('quantity') - const ratingValues = await index.getFilterValues('rating') - - // Each unique value should create its own index entry - expect(priceValues).toContain('19.99') - expect(priceValues).toContain('29.99') - expect(quantityValues).toContain('5') - expect(quantityValues).toContain('3') - expect(ratingValues.length).toBeGreaterThan(0) - }) - - it('should maintain backward compatibility with existing code', async () => { - // Existing code that used excludeFields should still work - const legacyIndex = new MetadataIndexManager(storage, { - excludeFields: ['internalField', 'debugData'] - }) - const testTime = TEST_BASE_TIME + 90000000 // Offset to avoid bucket collision - - await legacyIndex.addToIndex('entity1', { - noun: 'character', - modified: testTime, // Should be indexed + bucketed - internalField: 'secret', // Should be excluded - debugData: 'verbose logs' // Should be excluded - }) - - await legacyIndex.flush() - - // Modified should be indexed - const modifiedValues = await legacyIndex.getFilterValues('modified') - expect(modifiedValues.length).toBeGreaterThan(0) - - // Custom excluded fields should be excluded - const internalValues = await legacyIndex.getFilterValues('internalField') - const debugValues = await legacyIndex.getFilterValues('debugData') - expect(internalValues.length).toBe(0) - expect(debugValues.length).toBe(0) - }) -})