getCandidates is now async

pull/2/head
Louis Chatriot 9 years ago
parent 65024439c6
commit 645d87504d
  1. 111
      lib/cursor.js
  2. 188
      lib/datastore.js
  3. 2
      lib/indexes.js
  4. 85
      test/db.test.js

@ -103,76 +103,83 @@ Cursor.prototype.project = function (candidates) {
* *
* @param {Function} callback - Signature: err, results * @param {Function} callback - Signature: err, results
*/ */
Cursor.prototype._exec = function(callback) { Cursor.prototype._exec = function(_callback) {
var candidates = this.db.getCandidates(this.query) var res = [], added = 0, skipped = 0, self = this
, res = [], added = 0, skipped = 0, self = this
, error = null , error = null
, i, keys, key , i, keys, key
; ;
try { function callback (error, res) {
for (i = 0; i < candidates.length; i += 1) { if (self.execFn) {
if (model.match(candidates[i], this.query)) { return self.execFn(error, res, _callback);
// If a sort is defined, wait for the results to be sorted before applying limit and skip } else {
if (!this._sort) { return _callback(error, res);
if (this._skip && this._skip > skipped) { }
skipped += 1; }
this.db.getCandidates(this.query, function (err, candidates) {
if (err) { return callback(err); }
try {
for (i = 0; i < candidates.length; i += 1) {
if (model.match(candidates[i], self.query)) {
// If a sort is defined, wait for the results to be sorted before applying limit and skip
if (!self._sort) {
if (self._skip && self._skip > skipped) {
skipped += 1;
} else {
res.push(candidates[i]);
added += 1;
if (self._limit && self._limit <= added) { break; }
}
} else { } else {
res.push(candidates[i]); res.push(candidates[i]);
added += 1;
if (this._limit && this._limit <= added) { break; }
} }
} else {
res.push(candidates[i]);
} }
} }
} catch (err) {
return callback(err);
} }
} catch (err) {
return callback(err);
}
// Apply all sorts // Apply all sorts
if (this._sort) { if (self._sort) {
keys = Object.keys(this._sort); keys = Object.keys(self._sort);
// Sorting // Sorting
var criteria = []; var criteria = [];
for (i = 0; i < keys.length; i++) { for (i = 0; i < keys.length; i++) {
key = keys[i]; key = keys[i];
criteria.push({ key: key, direction: self._sort[key] }); criteria.push({ key: key, direction: self._sort[key] });
}
res.sort(function(a, b) {
var criterion, compare, i;
for (i = 0; i < criteria.length; i++) {
criterion = criteria[i];
compare = criterion.direction * model.compareThings(model.getDotValue(a, criterion.key), model.getDotValue(b, criterion.key), self.db.compareStrings);
if (compare !== 0) {
return compare;
}
} }
return 0; res.sort(function(a, b) {
}); var criterion, compare, i;
for (i = 0; i < criteria.length; i++) {
criterion = criteria[i];
compare = criterion.direction * model.compareThings(model.getDotValue(a, criterion.key), model.getDotValue(b, criterion.key), self.db.compareStrings);
if (compare !== 0) {
return compare;
}
}
return 0;
});
// Applying limit and skip // Applying limit and skip
var limit = this._limit || res.length var limit = self._limit || res.length
, skip = this._skip || 0; , skip = self._skip || 0;
res = res.slice(skip, skip + limit); res = res.slice(skip, skip + limit);
} }
// Apply projection // Apply projection
try { try {
res = this.project(res); res = self.project(res);
} catch (e) { } catch (e) {
error = e; error = e;
res = undefined; res = undefined;
} }
if (this.execFn) {
return this.execFn(error, res, callback);
} else {
return callback(error, res); return callback(error, res);
} });
}; };
Cursor.prototype.exec = function () { Cursor.prototype.exec = function () {

@ -115,6 +115,7 @@ Datastore.prototype.resetIndexes = function (newData) {
* @param {String} options.fieldName * @param {String} options.fieldName
* @param {Boolean} options.unique * @param {Boolean} options.unique
* @param {Boolean} options.sparse * @param {Boolean} options.sparse
* @param {Number} options.expireAfterSeconds - Optional, if set this index becomes a TTL index
* @param {Function} cb Optional callback, signature: err * @param {Function} cb Optional callback, signature: err
*/ */
Datastore.prototype.ensureIndex = function (options, cb) { Datastore.prototype.ensureIndex = function (options, cb) {
@ -243,50 +244,63 @@ Datastore.prototype.updateIndexes = function (oldDoc, newDoc) {
* One way to make it better would be to enable the use of multiple indexes if the first usable index * One way to make it better would be to enable the use of multiple indexes if the first usable index
* returns too much data. I may do it in the future. * returns too much data. I may do it in the future.
* *
* TODO: needs to be moved to the Cursor module * Returned candidates will be scanned to find and remove all expired documents
*
* @param {Query} query
* @param {Function} callback Signature err, docs
*/ */
Datastore.prototype.getCandidates = function (query) { Datastore.prototype.getCandidates = function (query, callback) {
var indexNames = Object.keys(this.indexes) var indexNames = Object.keys(this.indexes)
, self = this
, usableQueryKeys; , usableQueryKeys;
// For a basic match
usableQueryKeys = []; async.waterfall([
Object.keys(query).forEach(function (k) { // STEP 1: get candidates list by checking indexes from most to least frequent usecase
if (typeof query[k] === 'string' || typeof query[k] === 'number' || typeof query[k] === 'boolean' || util.isDate(query[k]) || query[k] === null) { function (cb) {
usableQueryKeys.push(k); // For a basic match
usableQueryKeys = [];
Object.keys(query).forEach(function (k) {
if (typeof query[k] === 'string' || typeof query[k] === 'number' || typeof query[k] === 'boolean' || util.isDate(query[k]) || query[k] === null) {
usableQueryKeys.push(k);
}
});
usableQueryKeys = _.intersection(usableQueryKeys, indexNames);
if (usableQueryKeys.length > 0) {
return cb(null, self.indexes[usableQueryKeys[0]].getMatching(query[usableQueryKeys[0]]));
} }
});
usableQueryKeys = _.intersection(usableQueryKeys, indexNames);
if (usableQueryKeys.length > 0) {
return this.indexes[usableQueryKeys[0]].getMatching(query[usableQueryKeys[0]]);
}
// For a $in match // For a $in match
usableQueryKeys = []; usableQueryKeys = [];
Object.keys(query).forEach(function (k) { Object.keys(query).forEach(function (k) {
if (query[k] && query[k].hasOwnProperty('$in')) { if (query[k] && query[k].hasOwnProperty('$in')) {
usableQueryKeys.push(k); usableQueryKeys.push(k);
}
});
usableQueryKeys = _.intersection(usableQueryKeys, indexNames);
if (usableQueryKeys.length > 0) {
return cb(null, self.indexes[usableQueryKeys[0]].getMatching(query[usableQueryKeys[0]].$in));
} }
});
usableQueryKeys = _.intersection(usableQueryKeys, indexNames);
if (usableQueryKeys.length > 0) {
return this.indexes[usableQueryKeys[0]].getMatching(query[usableQueryKeys[0]].$in);
}
// For a comparison match // For a comparison match
usableQueryKeys = []; usableQueryKeys = [];
Object.keys(query).forEach(function (k) { Object.keys(query).forEach(function (k) {
if (query[k] && (query[k].hasOwnProperty('$lt') || query[k].hasOwnProperty('$lte') || query[k].hasOwnProperty('$gt') || query[k].hasOwnProperty('$gte'))) { if (query[k] && (query[k].hasOwnProperty('$lt') || query[k].hasOwnProperty('$lte') || query[k].hasOwnProperty('$gt') || query[k].hasOwnProperty('$gte'))) {
usableQueryKeys.push(k); usableQueryKeys.push(k);
}
});
usableQueryKeys = _.intersection(usableQueryKeys, indexNames);
if (usableQueryKeys.length > 0) {
return cb(null, self.indexes[usableQueryKeys[0]].getBetweenBounds(query[usableQueryKeys[0]]));
} }
});
usableQueryKeys = _.intersection(usableQueryKeys, indexNames);
if (usableQueryKeys.length > 0) {
return this.indexes[usableQueryKeys[0]].getBetweenBounds(query[usableQueryKeys[0]]);
}
// By default, return all the DB data // By default, return all the DB data
return this.getAllData(); return cb(null, self.getAllData());
}
// STEP 2: remove all expired documents
, function (docs) {
return callback(null, docs);
}]);
}; };
@ -548,48 +562,49 @@ Datastore.prototype._update = function (query, updateQuery, options, cb) {
}); });
} }
, function () { // Perform the update , function () { // Perform the update
var modifiedDoc var modifiedDoc , modifications = [];
, candidates = self.getCandidates(query)
, modifications = []
;
// Preparing update (if an error is thrown here neither the datafile nor self.getCandidates(query, function (err, candidates) {
// the in-memory indexes are affected) if (err) { return callback(err); }
try {
for (i = 0; i < candidates.length; i += 1) { // Preparing update (if an error is thrown here neither the datafile nor
if (model.match(candidates[i], query) && (multi || numReplaced === 0)) { // the in-memory indexes are affected)
numReplaced += 1; try {
modifiedDoc = model.modify(candidates[i], updateQuery); for (i = 0; i < candidates.length; i += 1) {
if (self.timestampData) { modifiedDoc.updatedAt = new Date(); } if (model.match(candidates[i], query) && (multi || numReplaced === 0)) {
modifications.push({ oldDoc: candidates[i], newDoc: modifiedDoc }); numReplaced += 1;
modifiedDoc = model.modify(candidates[i], updateQuery);
if (self.timestampData) { modifiedDoc.updatedAt = new Date(); }
modifications.push({ oldDoc: candidates[i], newDoc: modifiedDoc });
}
} }
} catch (err) {
return callback(err);
} }
} catch (err) {
return callback(err);
}
// Change the docs in memory // Change the docs in memory
try { try {
self.updateIndexes(modifications); self.updateIndexes(modifications);
} catch (err) { } catch (err) {
return callback(err); return callback(err);
}
// Update the datafile
var updatedDocs = _.pluck(modifications, 'newDoc');
self.persistence.persistNewState(updatedDocs, function (err) {
if (err) { return callback(err); }
if (!options.returnUpdatedDocs) {
return callback(null, numReplaced);
} else {
var updatedDocsDC = [];
updatedDocs.forEach(function (doc) { updatedDocsDC.push(model.deepCopy(doc)); });
return callback(null, numReplaced, updatedDocsDC);
} }
// Update the datafile
var updatedDocs = _.pluck(modifications, 'newDoc');
self.persistence.persistNewState(updatedDocs, function (err) {
if (err) { return callback(err); }
if (!options.returnUpdatedDocs) {
return callback(null, numReplaced);
} else {
var updatedDocsDC = [];
updatedDocs.forEach(function (doc) { updatedDocsDC.push(model.deepCopy(doc)); });
return callback(null, numReplaced, updatedDocsDC);
}
});
}); });
} }]);
]);
}; };
Datastore.prototype.update = function () { Datastore.prototype.update = function () {
this.executor.push({ this: this, fn: this._update, arguments: arguments }); this.executor.push({ this: this, fn: this._update, arguments: arguments });
}; };
@ -607,32 +622,33 @@ Datastore.prototype.update = function () {
*/ */
Datastore.prototype._remove = function (query, options, cb) { Datastore.prototype._remove = function (query, options, cb) {
var callback var callback
, self = this , self = this, numRemoved = 0, removedDocs = [], multi
, numRemoved = 0
, multi
, removedDocs = []
, candidates = this.getCandidates(query)
; ;
if (typeof options === 'function') { cb = options; options = {}; } if (typeof options === 'function') { cb = options; options = {}; }
callback = cb || function () {}; callback = cb || function () {};
multi = options.multi !== undefined ? options.multi : false; multi = options.multi !== undefined ? options.multi : false;
try { this.getCandidates(query, function (err, candidates) {
candidates.forEach(function (d) {
if (model.match(d, query) && (multi || numRemoved === 0)) {
numRemoved += 1;
removedDocs.push({ $$deleted: true, _id: d._id });
self.removeFromIndexes(d);
}
});
} catch (err) { return callback(err); }
self.persistence.persistNewState(removedDocs, function (err) {
if (err) { return callback(err); } if (err) { return callback(err); }
return callback(null, numRemoved);
try {
candidates.forEach(function (d) {
if (model.match(d, query) && (multi || numRemoved === 0)) {
numRemoved += 1;
removedDocs.push({ $$deleted: true, _id: d._id });
self.removeFromIndexes(d);
}
});
} catch (err) { return callback(err); }
self.persistence.persistNewState(removedDocs, function (err) {
if (err) { return callback(err); }
return callback(null, numRemoved);
});
}); });
}; };
Datastore.prototype.remove = function () { Datastore.prototype.remove = function () {
this.executor.push({ this: this, fn: this._remove, arguments: arguments }); this.executor.push({ this: this, fn: this._remove, arguments: arguments });
}; };

@ -32,11 +32,13 @@ function projectForUnique (elt) {
* @param {String} options.fieldName On which field should the index apply (can use dot notation to index on sub fields) * @param {String} options.fieldName On which field should the index apply (can use dot notation to index on sub fields)
* @param {Boolean} options.unique Optional, enforce a unique constraint (default: false) * @param {Boolean} options.unique Optional, enforce a unique constraint (default: false)
* @param {Boolean} options.sparse Optional, allow a sparse index (we can have documents for which fieldName is undefined) (default: false) * @param {Boolean} options.sparse Optional, allow a sparse index (we can have documents for which fieldName is undefined) (default: false)
* @param {Number} options.expireAfterSeconds - Optional, if set this index becomes a TTL index
*/ */
function Index (options) { function Index (options) {
this.fieldName = options.fieldName; this.fieldName = options.fieldName;
this.unique = options.unique || false; this.unique = options.unique || false;
this.sparse = options.sparse || false; this.sparse = options.sparse || false;
if (options.expireAfterSeconds !== undefined) { this.expireAfterSeconds = options.expireAfterSeconds; }
this.treeOptions = { unique: this.unique, compareKeys: model.compareThings, checkValueEquality: checkValueEquality }; this.treeOptions = { unique: this.unique, compareKeys: model.compareThings, checkValueEquality: checkValueEquality };

@ -432,16 +432,17 @@ describe('Database', function () {
d.insert({ tf: 6 }, function () { d.insert({ tf: 6 }, function () {
d.insert({ tf: 4, an: 'other' }, function (err, _doc2) { d.insert({ tf: 4, an: 'other' }, function (err, _doc2) {
d.insert({ tf: 9 }, function () { d.insert({ tf: 9 }, function () {
var data = d.getCandidates({ r: 6, tf: 4 }) d.getCandidates({ r: 6, tf: 4 }, function (err, data) {
, doc1 = _.find(data, function (d) { return d._id === _doc1._id; }) var doc1 = _.find(data, function (d) { return d._id === _doc1._id; })
, doc2 = _.find(data, function (d) { return d._id === _doc2._id; }) , doc2 = _.find(data, function (d) { return d._id === _doc2._id; })
; ;
data.length.should.equal(2); data.length.should.equal(2);
assert.deepEqual(doc1, { _id: doc1._id, tf: 4 }); assert.deepEqual(doc1, { _id: doc1._id, tf: 4 });
assert.deepEqual(doc2, { _id: doc2._id, tf: 4, an: 'other' }); assert.deepEqual(doc2, { _id: doc2._id, tf: 4, an: 'other' });
done(); done();
});
}); });
}); });
}); });
@ -455,16 +456,17 @@ describe('Database', function () {
d.insert({ tf: 6 }, function (err, _doc1) { d.insert({ tf: 6 }, function (err, _doc1) {
d.insert({ tf: 4, an: 'other' }, function (err) { d.insert({ tf: 4, an: 'other' }, function (err) {
d.insert({ tf: 9 }, function (err, _doc2) { d.insert({ tf: 9 }, function (err, _doc2) {
var data = d.getCandidates({ r: 6, tf: { $in: [6, 9, 5] } }) d.getCandidates({ r: 6, tf: { $in: [6, 9, 5] } }, function (err, data) {
, doc1 = _.find(data, function (d) { return d._id === _doc1._id; }) var doc1 = _.find(data, function (d) { return d._id === _doc1._id; })
, doc2 = _.find(data, function (d) { return d._id === _doc2._id; }) , doc2 = _.find(data, function (d) { return d._id === _doc2._id; })
; ;
data.length.should.equal(2); data.length.should.equal(2);
assert.deepEqual(doc1, { _id: doc1._id, tf: 6 }); assert.deepEqual(doc1, { _id: doc1._id, tf: 6 });
assert.deepEqual(doc2, { _id: doc2._id, tf: 9 }); assert.deepEqual(doc2, { _id: doc2._id, tf: 9 });
done(); done();
});
}); });
}); });
}); });
@ -478,20 +480,21 @@ describe('Database', function () {
d.insert({ tf: 6 }, function (err, _doc2) { d.insert({ tf: 6 }, function (err, _doc2) {
d.insert({ tf: 4, an: 'other' }, function (err, _doc3) { d.insert({ tf: 4, an: 'other' }, function (err, _doc3) {
d.insert({ tf: 9 }, function (err, _doc4) { d.insert({ tf: 9 }, function (err, _doc4) {
var data = d.getCandidates({ r: 6, notf: { $in: [6, 9, 5] } }) d.getCandidates({ r: 6, notf: { $in: [6, 9, 5] } }, function (err, data) {
, doc1 = _.find(data, function (d) { return d._id === _doc1._id; }) var doc1 = _.find(data, function (d) { return d._id === _doc1._id; })
, doc2 = _.find(data, function (d) { return d._id === _doc2._id; }) , doc2 = _.find(data, function (d) { return d._id === _doc2._id; })
, doc3 = _.find(data, function (d) { return d._id === _doc3._id; }) , doc3 = _.find(data, function (d) { return d._id === _doc3._id; })
, doc4 = _.find(data, function (d) { return d._id === _doc4._id; }) , doc4 = _.find(data, function (d) { return d._id === _doc4._id; })
; ;
data.length.should.equal(4); data.length.should.equal(4);
assert.deepEqual(doc1, { _id: doc1._id, tf: 4 }); assert.deepEqual(doc1, { _id: doc1._id, tf: 4 });
assert.deepEqual(doc2, { _id: doc2._id, tf: 6 }); assert.deepEqual(doc2, { _id: doc2._id, tf: 6 });
assert.deepEqual(doc3, { _id: doc3._id, tf: 4, an: 'other' }); assert.deepEqual(doc3, { _id: doc3._id, tf: 4, an: 'other' });
assert.deepEqual(doc4, { _id: doc4._id, tf: 9 }); assert.deepEqual(doc4, { _id: doc4._id, tf: 9 });
done(); done();
});
}); });
}); });
}); });
@ -505,16 +508,17 @@ describe('Database', function () {
d.insert({ tf: 6 }, function (err, _doc2) { d.insert({ tf: 6 }, function (err, _doc2) {
d.insert({ tf: 4, an: 'other' }, function (err, _doc3) { d.insert({ tf: 4, an: 'other' }, function (err, _doc3) {
d.insert({ tf: 9 }, function (err, _doc4) { d.insert({ tf: 9 }, function (err, _doc4) {
var data = d.getCandidates({ r: 6, tf: { $lte: 9, $gte: 6 } }) d.getCandidates({ r: 6, tf: { $lte: 9, $gte: 6 } }, function (err, data) {
, doc2 = _.find(data, function (d) { return d._id === _doc2._id; }) var doc2 = _.find(data, function (d) { return d._id === _doc2._id; })
, doc4 = _.find(data, function (d) { return d._id === _doc4._id; }) , doc4 = _.find(data, function (d) { return d._id === _doc4._id; })
; ;
data.length.should.equal(2); data.length.should.equal(2);
assert.deepEqual(doc2, { _id: doc2._id, tf: 6 }); assert.deepEqual(doc2, { _id: doc2._id, tf: 6 });
assert.deepEqual(doc4, { _id: doc4._id, tf: 9 }); assert.deepEqual(doc4, { _id: doc4._id, tf: 9 });
done(); done();
});
}); });
}); });
}); });
@ -2649,9 +2653,10 @@ describe('Database', function () {
it('Results of getMatching should never contain duplicates', function (done) { it('Results of getMatching should never contain duplicates', function (done) {
d.ensureIndex({ fieldName: 'bad' }); d.ensureIndex({ fieldName: 'bad' });
d.insert({ bad: ['a', 'b'] }, function () { d.insert({ bad: ['a', 'b'] }, function () {
var res = d.getCandidates({ bad: { $in: ['a', 'b'] } }); d.getCandidates({ bad: { $in: ['a', 'b'] } }, function (err, res) {
res.length.should.equal(1); res.length.should.equal(1);
done(); done();
});
}); });
}); });

Loading…
Cancel
Save