mpackdb

16.4 KB

1. import { writeFile, readFile, open, rename } from 'fs/promises';
2. import { createWriteStream } from 'fs';
3. import { resolve } from 'path';
4. import { deserialize } from './mpack.js';
5. import { PrimaryKeyType, IndexType } from './MPackDB.js';
7. const BLOCK_SIZE = 4096; // 4KB
9. /**
10. * Manages indexes for the database (in-memory deltas + on-disk persistence)
11. */
12. export class IndexManager {
13. _dbDir;
14. _dbName;
15. _indexes;
16. _indexTypes;
17. _primaryKeyType;
18. _persistIntervalId = null;
20. _indexPaths = {};
21. _deltaIndexes = {};
22. _tombstones = {};
23. _persistThreshold = 1000;
24. _totalDeltaCount = 0;
26. /**
27. * Creates a new IndexManager
28. * @param {string} dbDir - Database directory path
29. * @param {string} dbName - Database name
30. * @param {string[]} indexes - Array of field names to index
31. * @param {Object} indexTypes - Map of field names to IndexType
32. * @param {number} primaryKeyType - Primary key type
33. */
34. constructor(dbDir, dbName, indexes, indexTypes, primaryKeyType) {
35. this._dbDir = dbDir;
36. this._dbName = dbName;
37. this._indexes = indexes;
38. this._indexTypes = indexTypes;
39. this._primaryKeyType = primaryKeyType;
40. }
42. /**
43. * Initializes the index manager (rebuilds indexes from data file)
44. * @param {string} dataPath - Path to the data file
45. * @returns {Promise<void>}
46. */
47. async init(dataPath) {
48. for (const field of this._indexes) {
49. this._deltaIndexes[field] = [];
50. this._tombstones[field] = new Map();
51. const indexPath = resolve(this._dbDir, `${this._dbName}.${field}.txt`);
52. this._indexPaths[field] = indexPath;
54. try {
55. const fileHandle = await open(indexPath, 'r');
56. const stats = await fileHandle.stat();
57. await fileHandle.close();
58. if (stats.size === 0) {
59. await this._rebuildIndex(dataPath, field);
60. }
61. } catch (e) {
62. if (e.code === 'ENOENT') {
63. await this._rebuildIndex(dataPath, field);
64. } else {
65. throw e;
66. }
67. }
68. }
69. }
71. /**
72. * Rebuilds an index from the data file
73. * @param {string} dataPath - Path to the data file
74. * @param {string} field - Field name to index
75. * @returns {Promise<void>}
76. * @private
77. */
78. async _rebuildIndex(dataPath, field) {
79. const tempIndexPath = `${this._indexPaths[field]}.${Date.now()}.tmp`;
80. const writeStream = createWriteStream(tempIndexPath, { flags: 'w', encoding: 'utf-8' });
82. const docIterator = this._getDocsAndLocationsFromDataFile(dataPath);
83. const entries = [];
84. for await (const { doc, loc } of docIterator) {
85. if (doc && doc[field] !== undefined) {
86. const key = doc[field];
87. entries.push({ key, loc });
88. }
89. }
91. // Sort entries before writing to the index file
92. entries.sort((a, b) => this._compareKeys(a.key, b.key));
94. for (const entry of entries) {
95. writeStream.write(`${entry.key},${entry.loc[0]},${entry.loc[1]}\n`);
96. }
98. await new Promise(resolve => writeStream.end(resolve));
99. await rename(tempIndexPath, this._indexPaths[field]);
100. }
102. /**
103. * Generator that yields documents and their locations from the data file
104. * @param {string} dataPath - Path to the data file
105. * @yields {{doc: Object, loc: Array}} Document and location tuple
106. * @private
107. */
108. async* _getDocsAndLocationsFromDataFile(dataPath) {
109. let fileHandle;
110. try {
111. fileHandle = await open(dataPath, 'r');
112. const stats = await fileHandle.stat();
113. let offset = 0;
115. while (offset < stats.size) {
116. // Read size header
117. const sizeBuffer = Buffer.alloc(4);
118. await fileHandle.read(sizeBuffer, 0, 4, offset);
119. const size = sizeBuffer.readInt32LE(0);
121. if (size <= 0 || offset + size > stats.size) {
122. break; // Corrupted or incomplete record
123. }
125. const docBuffer = Buffer.alloc(size);
126. await fileHandle.read(docBuffer, 0, size, offset);
127. const doc = deserialize(docBuffer);
129. yield { doc, loc: [offset, size] };
130. offset += size;
131. }
132. } catch (e) {
133. if (e.code !== 'ENOENT') {
134. throw e;
135. }
136. // if file does not exist, do nothing
137. } finally {
138. await fileHandle?.close();
139. }
140. }
142. /**
143. * Compares two keys for sorting
144. * @param {any} keyA - First key
145. * @param {any} keyB - Second key
146. * @returns {number} -1 if keyA < keyB, 0 if equal, 1 if keyA > keyB
147. * @private
148. */
149. _compareKeys(keyA, keyB) {
150. if (typeof keyA === 'number' && typeof keyB === 'number') {
151. return keyA - keyB;
152. }
153. return String(keyA).localeCompare(String(keyB));
154. }
156. /**
157. * Starts automatic periodic persistence of indexes
158. * @param {number} interval - Interval in milliseconds between auto-persists
159. * @param {number} threshold - Number of operations before triggering auto-persist
160. */
161. startAutoPersist(interval, threshold) {
162. this._persistThreshold = threshold;
164. // Start periodic persist
165. if (interval > 0) {
166. this._persistIntervalId = setInterval(async () => {
167. if (this._totalDeltaCount > 0) {
168. await this.persist();
169. }
170. }, interval);
171. // Don't keep process alive just for this timer
172. this._persistIntervalId.unref();
173. }
174. }
176. /**
177. * Closes the index manager (persists indexes and clears interval)
178. * @returns {Promise<void>}
179. */
180. async close() {
181. if (this._persistIntervalId) {
182. clearInterval(this._persistIntervalId);
183. this._persistIntervalId = null;
184. }
185. await this.persist();
186. }
188. /**
189. * Persists all in-memory index deltas to disk
190. * @returns {Promise<void>}
191. */
192. async persist() {
193. for (const field of this._indexes) {
194. let onDiskEntries = [];
195. const indexPath = this._indexPaths[field];
196. try {
197. const indexContent = await readFile(indexPath, 'utf-8');
198. onDiskEntries = indexContent.trim().split('\n').filter(Boolean).map(line => {
199. const [key, offset, length] = line.split(',');
200. return { key: this._parseKey(key, field), loc: [parseInt(offset), parseInt(length)] };
201. });
202. } catch (e) {
203. if (e.code !== 'ENOENT') throw e;
204. }
206. const fieldTombstones = this._tombstones[field];
207. const validDiskEntries = onDiskEntries.filter(e => !fieldTombstones.has(e.loc[0]));
209. const deltaEntries = this._deltaIndexes[field] || [];
210. const finalIndexData = [...validDiskEntries, ...deltaEntries];
212. finalIndexData.sort((a, b) => this._compareKeys(a.key, b.key));
214. if (finalIndexData.length > 0) {
215. const indexContent = finalIndexData.map(e => `${e.key},${e.loc[0]},${e.loc[1]}`).join('\n') + '\n';
216. await writeFile(indexPath, indexContent, 'utf-8');
217. } else {
218. await writeFile(indexPath, '', 'utf-8');
219. }
221. // Clear the in-memory changes now that they are persisted
222. this._deltaIndexes[field] = [];
223. this._tombstones[field].clear();
224. }
225. this._totalDeltaCount = 0;
226. }
228. /**
229. * Parses a key based on the field's index type
230. * @param {any} key - The key to parse
231. * @param {string} field - Field name
232. * @returns {any} Parsed key (number or string)
233. * @private
234. */
235. _parseKey(key, field) {
236. const indexType = this._indexTypes[field];
238. // Use index type if specified, otherwise fall back to primary key type logic
239. if (indexType === IndexType.NUMERIC) {
240. const num = parseInt(key, 10);
241. if (!isNaN(num)) return num;
242. } else if (indexType === IndexType.LEXICAL) {
243. return key;
244. }
246. // Fallback to primary key type for fields without explicit index type
247. if (this._primaryKeyType === PrimaryKeyType.NUMBER) {
248. const num = parseInt(key, 10);
249. if (!isNaN(num)) return num;
250. }
251. return key;
252. }
254. /**
255. * Performs a binary search on the blocks of an index file on disk.
256. * This is a direct port of Joshua Bloch's famously correct binary search
257. * algorithm, adapted for file blocks.
258. *
259. * @param {string} field The index field to search.
260. * @param {string|number} key The key to search for.
261. * @returns {Promise<number>} The starting offset in the file for the linear scan.
262. */
263. async _binarySearchOnDisk(field, key) {
264. try {
265. const fileHandle = await open(this._indexPaths[field], 'r');
266. const stats = await fileHandle.stat();
267. const totalBlocks = Math.ceil(stats.size / BLOCK_SIZE);
268. let low = 0;
269. let high = totalBlocks - 1;
270. const buffer = Buffer.alloc(1024);
272. while (low <= high) {
273. const mid = Math.floor((low + high) / 2);
274. const offset = mid * BLOCK_SIZE;
276. const { bytesRead } = await fileHandle.read(buffer, 0, 1024, offset);
277. if (bytesRead === 0) break;
279. const blockContent = buffer.subarray(0, bytesRead).toString('utf-8');
280. const lines = blockContent.split('\n');
282. const firstKeyInBlock = this._parseKey(lines[0].split(',')[0], field);
283. const lastKeyInBlock = this._parseKey(lines[lines.length - 2]?.split(',')[0] || firstKeyInBlock, field);
285. const cmpLast = this._compareKeys(lastKeyInBlock, key);
287. if (cmpLast < 0) {
288. // The entire block is smaller than the key
289. low = mid + 1;
290. } else {
291. // The key might be in this block or a previous one
292. high = mid - 1;
293. }
294. }
296. await fileHandle.close();
297. // low is the insertion point, start scan from the previous block to be safe.
298. const finalOffset = Math.max(0, low - 1) * BLOCK_SIZE;
299. return finalOffset;
300. } catch (e) {
301. if (e.code === 'ENOENT') {
302. return -1;
303. }
304. throw e;
305. }
306. }
308. /**
309. * Finds the last entry for a given field and key (checks both delta and disk)
310. * @param {string} field - Field name
311. * @param {any} key - Key value to search for
312. * @returns {Promise<Object|null>} Entry object with key and loc, or null if not found
313. */
314. async findLastEntry(field, key) {
315. const parsedKey = this._parseKey(key, field);
317. const deltaIndex = this._deltaIndexes[field] || [];
318. for (let i = deltaIndex.length - 1; i >= 0; i--) {
319. const entry = deltaIndex[i];
320. if (this._compareKeys(entry.key, parsedKey) === 0) {
321. return entry;
322. }
323. }
325. const onDiskEntry = await this._findLastEntryOnDisk(field, parsedKey);
326. if (!onDiskEntry) return null;
328. const isTombstoned = this._tombstones[field].has(onDiskEntry.loc[0]);
329. return isTombstoned ? null : onDiskEntry;
330. }
332. /**
333. * Finds the last entry for a key on disk (linear scan from binary search position)
334. * @param {string} field - Field name
335. * @param {any} key - Key to search for
336. * @returns {Promise<Object|null>} Entry object or null
337. * @private
338. */
339. async _findLastEntryOnDisk(field, key) {
340. const startOffset = await this._binarySearchOnDisk(field, key);
341. if (startOffset === -1) return null;
343. let lastMatch = null;
344. const fileHandle = await open(this._indexPaths[field], 'r');
345. const buffer = Buffer.alloc(BLOCK_SIZE);
346. let currentOffset = startOffset;
348. try {
349. const stats = await fileHandle.stat();
350. while (currentOffset < stats.size) {
351. const { bytesRead } = await fileHandle.read(buffer, 0, BLOCK_SIZE, currentOffset);
352. if (bytesRead === 0) break;
354. const blockContent = buffer.slice(0, bytesRead).toString('utf-8');
355. const lines = blockContent.split('\n');
356. let found = false;
357. for (const line of lines) {
358. if (!line) continue;
359. const [keyStr, offset, length] = line.split(',');
360. const lineKey = this._parseKey(keyStr, field);
361. if (this._compareKeys(lineKey, key) === 0) {
362. found = true;
363. lastMatch = { key: lineKey, loc: [parseInt(offset), parseInt(length)] };
364. }
365. }
366. if (found) break; // We can stop once we find the block with the key
367. currentOffset += BLOCK_SIZE;
368. }
369. } finally {
370. await fileHandle.close();
371. }
372. return lastMatch;
373. }
375. /**
376. * Adds a document to the indexes
377. * @param {Object} doc - The document to index
378. * @param {Array} loc - [offset, length] location in data file
379. */
380. insert(doc, loc) {
381. for (const field of this._indexes) {
382. if (doc[field] !== undefined) {
383. this._deltaIndexes[field].push({ key: doc[field], loc });
384. this._totalDeltaCount++;
385. }
386. }
388. // Auto-persist if threshold reached
389. if (this._totalDeltaCount >= this._persistThreshold) {
390. // Persist async without blocking
391. this.persist().catch(err => console.error('Index persist error:', err));
392. }
393. }
395. /**
396. * Removes a document from the indexes
397. * @param {Object} doc - The document to remove
398. * @param {string} primaryKeyField - Primary key field name
399. * @returns {Promise<void>}
400. */
401. async remove(doc, primaryKeyField) {
402. const pkValue = doc[primaryKeyField];
404. const deltaPkIndex = this._deltaIndexes[primaryKeyField] || [];
405. const indexInDelta = deltaPkIndex.findIndex(e => this._compareKeys(e.key, pkValue) === 0);
407. if (indexInDelta !== -1) {
408. const offsetToRemove = deltaPkIndex[indexInDelta].loc[0];
409. for (const field of this._indexes) {
410. this._deltaIndexes[field] = (this._deltaIndexes[field] || []).filter(e => e.loc[0] !== offsetToRemove);
411. }
412. } else {
413. const onDiskEntry = await this._findLastEntryOnDisk(primaryKeyField, pkValue);
414. if (onDiskEntry) {
415. for (const field of this._indexes) {
416. if (doc[field] !== undefined) {
417. this._tombstones[field].set(onDiskEntry.loc[0], true);
418. }
419. }
420. }
421. }
422. }
424. /**
425. * Gets all document locations from the first index (used for full scans)
426. * @returns {Promise<Array[]>} Array of all [offset, length] locations
427. */
428. async getAllLocations() {
429. const field = this._indexes[0];
430. const indexPath = this._indexPaths[field];
431. let onDiskEntries = [];
432. try {
433. const indexContent = await readFile(indexPath, 'utf-8');
434. onDiskEntries = indexContent.trim().split('\n').filter(Boolean).map(line => {
435. const [, offset, length] = line.split(',');
436. return { loc: [parseInt(offset), parseInt(length)] };
437. });
438. } catch (e) {
439. if (e.code !== 'ENOENT') throw e;
440. }
442. const fieldTombstones = this._tombstones[field] || new Map();
443. const validDiskEntries = onDiskEntries.filter(e => !fieldTombstones.has(e.loc[0]));
445. const deltaEntries = this._deltaIndexes[field] || [];
447. const finalEntries = new Map();
448. for (const entry of validDiskEntries) {
449. finalEntries.set(`${entry.loc[0]}:${entry.loc[1]}`, entry.loc);
450. }
451. for (const entry of deltaEntries) {
452. finalEntries.set(`${entry.loc[0]}:${entry.loc[1]}`, entry.loc);
453. }
454. return Array.from(finalEntries.values());
455. }
457. /**
458. * Gets document locations for a given field and key
459. * @param {string} field - Field name
460. * @param {any} key - Key value to search for
461. * @returns {Promise<Array[]>} Array of [offset, length] locations
462. */
463. async get(field, key) {
464. const parsedKey = this._parseKey(key, field);
465. const onDiskResults = await this._getOnDisk(field, parsedKey);
466. const deltaResults = (this._deltaIndexes[field] || []).filter(e => this._compareKeys(e.key, parsedKey) === 0);
467. const tombstonedOffsets = this._tombstones[field] || new Map();
469. const finalEntries = new Map();
471. for (const entry of onDiskResults) {
472. if (!tombstonedOffsets.has(entry.loc[0])) {
473. finalEntries.set(`${entry.loc[0]}:${entry.loc[1]}`, entry);
474. }
475. }
477. for (const entry of deltaResults) {
478. finalEntries.set(`${entry.loc[0]}:${entry.loc[1]}`, entry);
479. }
481. return Array.from(finalEntries.values());
482. }
484. /**
485. * Gets all entries for a key from disk
486. * @param {string} field - Field name
487. * @param {any} key - Key to search for
488. * @returns {Promise<Array>} Array of entry objects
489. * @private
490. */
491. async _getOnDisk(field, key) {
492. const startOffset = await this._binarySearchOnDisk(field, key);
493. if (startOffset === -1) return [];
495. const results = [];
496. const fileHandle = await open(this._indexPaths[field], 'r');
497. const buffer = Buffer.alloc(BLOCK_SIZE);
498. let currentOffset = startOffset;
500. try {
501. const stats = await fileHandle.stat();
502. while (currentOffset < stats.size) {
503. const { bytesRead } = await fileHandle.read(buffer, 0, BLOCK_SIZE, currentOffset);
504. if (bytesRead === 0) break;
506. const blockContent = buffer.subarray(0, bytesRead).toString('utf-8');
507. const lines = blockContent.split('\n');
508. let firstKeyInBlock;
509. let foundMatchInBlock = false;
511. for (const line of lines) {
512. if (!line) continue;
513. const [keyStr, locOffset, locLength] = line.split(',');
514. if (firstKeyInBlock === undefined) {
515. firstKeyInBlock = this._parseKey(keyStr, field);
516. }
517. const lineKey = this._parseKey(keyStr, field);
518. if (this._compareKeys(lineKey, key) === 0) {
519. foundMatchInBlock = true;
520. results.push({ key: lineKey, loc: [parseInt(locOffset), parseInt(locLength)] });
521. }
522. }
524. if (results.length > 0 && !foundMatchInBlock) {
525. break;
526. }
527. if (foundMatchInBlock === false && firstKeyInBlock && this._compareKeys(firstKeyInBlock, key) > 0) {
528. break;
529. }
530. currentOffset += BLOCK_SIZE;
531. }
532. } finally {
533. await fileHandle.close();
534. }
535. return results;
536. }
538. async* _getDocsByLocation(dataPath, locations) {
539. const fileHandle = await open(dataPath, 'r');
540. try {
541. for (const loc of locations) {
542. if (loc && loc.length === 2 && loc[1] > 0) {
543. const [offset, length] = loc;
544. const buffer = Buffer.alloc(length);
545. await fileHandle.read(buffer, 0, length, offset);
546. yield deserialize(buffer);
547. }
548. }
549. } finally {
550. await fileHandle.close();
551. }
552. }
553. }
555. export default IndexManager;