feat: unified column store for filtering + sorting at billion scale

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.
This commit is contained in:
David Snelling 2026-04-10 11:22:19 -07:00
parent 634ddfb2bb
commit 46583f2d9b
16 changed files with 3846 additions and 521 deletions

View file

@ -2166,19 +2166,31 @@ export class Brainy<T = any> implements BrainyInterface<T> {
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

View file

@ -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<void> {
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<void> {
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<ManifestData> {
return this.data
}
}

View file

@ -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<CursorEntry> {
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<CursorEntry> {
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<CursorEntry> {
for (let i = from; i < this.values.length; i++) {
yield { value: this.values[i], entityIntId: this.entityIds[i], position: i }
}
}
*iterateBackward(from?: number): Generator<CursorEntry> {
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
}
}

View file

@ -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<number>(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<number>(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<string>(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<SegmentHeader, 'magic' | 'version'>,
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 }
}

View file

@ -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<CursorEntry>
}
/**
* 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<string, ColumnTailBuffer> = new Map()
/** Per-field manifests tracking segment files. */
private manifests: Map<string, ColumnManifest> = new Map()
/** Cached loaded segment cursors: `field:segmentId` → cursor. */
private segmentCache: Map<string, ColumnSegmentCursor> = 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<string, RoaringBitmap32> = new Map()
/** Known field value types (inferred from first write). */
private fieldTypes: Map<string, ValueType> = 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<void> {
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<string, unknown>): 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<RoaringBitmap32> {
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<RoaringBitmap32> {
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<number[]> {
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<number[]> {
return this.mergeSort(field, order, k, filterBitmap)
}
/**
* Get all distinct values for a field.
*/
async getFilterValues(field: string): Promise<string[]> {
const valueSet = new Set<string>()
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<void> {
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<void> {
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<void> {
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<ColumnSegmentCursor[]> {
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<ColumnSegmentCursor | null> {
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<number[]> {
// 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<CursorEntry>[] = []
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<number>() // 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
}
}
}

View file

@ -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<number>
}
/**
* 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<number> = 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()
}
}

View file

@ -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'

View file

@ -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<void>
/**
* 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<string, unknown>): 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<RoaringBitmap32>
/**
* 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<RoaringBitmap32>
/**
* 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<number[]>
/**
* 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<number[]>
/**
* 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<string[]>
/**
* 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<void>
/**
* Flush and release all resources.
*/
close(): Promise<void>
}

View file

@ -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<string[]> {
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<string>()
// 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<string, unknown> = {}
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<void> {
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<string[]> {
// 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<number>()
// 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<number>()
// 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<number>()
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<number>()
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<Iterable<number>> {
const allIntIds = new Set<number>()
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<string[]> {
// 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()
}
}
/**

View file

@ -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)
})
})

View file

@ -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)
})
})
})

View file

@ -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<string, any>()
async readObjectFromPath(path: string): Promise<any> {
return this.data.get(path) ?? null
}
async writeObjectToPath(path: string, obj: any): Promise<void> {
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)
})
})

View file

@ -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)
})
})
})

View file

@ -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/)
})
})
})

View file

@ -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])
})
})
})

View file

@ -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)
})
})