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:
parent
634ddfb2bb
commit
46583f2d9b
16 changed files with 3846 additions and 521 deletions
|
|
@ -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
|
||||
|
|
|
|||
204
src/indexes/columnStore/ColumnManifest.ts
Normal file
204
src/indexes/columnStore/ColumnManifest.ts
Normal 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
|
||||
}
|
||||
}
|
||||
271
src/indexes/columnStore/ColumnSegmentCursor.ts
Normal file
271
src/indexes/columnStore/ColumnSegmentCursor.ts
Normal 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
|
||||
}
|
||||
}
|
||||
548
src/indexes/columnStore/ColumnSegmentFormat.ts
Normal file
548
src/indexes/columnStore/ColumnSegmentFormat.ts
Normal 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 }
|
||||
}
|
||||
695
src/indexes/columnStore/ColumnStore.ts
Normal file
695
src/indexes/columnStore/ColumnStore.ts
Normal 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
|
||||
}
|
||||
}
|
||||
}
|
||||
169
src/indexes/columnStore/ColumnTailBuffer.ts
Normal file
169
src/indexes/columnStore/ColumnTailBuffer.ts
Normal 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()
|
||||
}
|
||||
}
|
||||
38
src/indexes/columnStore/index.ts
Normal file
38
src/indexes/columnStore/index.ts
Normal 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'
|
||||
272
src/indexes/columnStore/types.ts
Normal file
272
src/indexes/columnStore/types.ts
Normal 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>
|
||||
}
|
||||
|
|
@ -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()
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue