<commit_before>/**
 * Manage access to data, be it to find, update or remove it
 */
var document = require('./document')
  , events = require('events')
  , util = require('util')
  , _ = require('lodash')
  ;


var INDEX_BUILDING_DEBOUNCE = 10;
var LIVE_QUERY_DEBOUNCE = 30;
var AUTO_INDEXING = true;


/**
 * Create a new cursor for this collection
 * @param {Datastore} db - The datastore this cursor is bound to
 * @param {Query} query - The query this cursor will operate on
 * @param {Function} execDn - Handler to be executed after cursor has found the results and before the callback passed to find/findOne/update/remove
 */
function Cursor (db, query, execFn) {
  this.db = db;
  this.query = query || {};
  if (execFn) { this.execFn = execFn; }
}


/**
 * Set a limit to the number of results
 */
Cursor.prototype.limit = function(limit) {
  this._limit = limit;
  return this;
};


/**
 * Skip a the number of results
 */
Cursor.prototype.skip = function(skip) {
  this._skip = skip;
  return this;
};


/**
 * Filter the results of the query - must be a function
 */
Cursor.prototype.filter = function(filter) {
  if (typeof(filter) != "function") return this;
  this._filter = filter;
  return this;
};


/**
 * Re-define the query itself; used for live queries
 */
Cursor.prototype.find = function(query) {
  this.query = query || this.query;
  return this;
};

/**
 * Sort results of the query
 * @param {SortQuery} sortQuery - SortQuery is { field: order }, field can use the dot-notation, order is 1 for ascending and -1 for descending
 */
Cursor.prototype.sort = function(sortQuery) {
  this._sort = sortQuery;
  return this;
};


/**
 * Add the use of a mapping function
 * @param {Function} map - map function, takes the object as the first argument
 */
Cursor.prototype.map = function(map) {
  if (typeof(map)=="function") this._map = map;
  return this;
};

/**
 * Add the use of a reducing function
 * @param {Function} reduce - reduce function, takes the two objects
 */
Cursor.prototype.reduce = function(reduce, initial) {
  if (typeof(reduce)=="function") {
    this._reduce = initial===undefined ? [reduce] : [reduce,initial];
  }
  return this;
};


/**
 * Aggregating function
 */
Cursor.prototype.aggregate = function(aggr) {
  this._aggregate = aggr;
  return this;
};

/**
 * Get all matching elements
 * Will return pointers to matched elements (shallow copies), returning full copies is the role of find or findOne
 *
 * @param {Function} callback - Signature: err, results
 */
Cursor.prototype.exec = function(callback) {
  var res = [], resIds = {}, resById = {}, self = this
    , err = null;

  var sort, sorter;
  if (typeof(this._sort) == "function") sorter = this._sort;
  else sort = this._sort;

  var reducer;
  if (self._reduce) reducer = _.cloneDeep(self._reduce);

  if (! this._quiet) this.db.emit("query", this.query);

  var stream = Cursor.getMatchesStream(this.db, this.query, sort, this._prefetched);
  stream.on("error", function(e) { callback(e) });
  stream.removeListener("ids", stream.trigger);
  stream.on("ids", function(ids) {
    var indexed = ids._indexed, sorted = ids._sorted; // for some reason we cannot access those in on 'ready' - maybe properties get erased from arrays?
    var earlyLimit = indexed && sorted && !self._filter && !sorter; // query is indexed, sorter and we don't have special filter/sort funcs; TODO: we should also set this to true if there's no limit/skip
    var earlySort = (sorted || !sort) && !sorter; // If ids are already sorted
    var earlyMapReduce = earlyLimit && earlySort && self._map && reducer; // special mode - run the map/reduce directly on data event
    // TODO: earlyMapReduce even without the map

    if (earlyLimit) ids = limit(ids, self._limit || ids.length, self._skip || 0);

    // No need to go further, sort and limits are applied, we only need count
    if (earlyLimit && self._count) { res = ids; return ready(); }

    // Start retrieving the objects for those IDs; res is an array that will hold all results
    stream.trigger(ids);
    res = new Array(ids.length);

    // Special case: res becomes the cursor of reduce if we're running a map/reduce
    if (earlyMapReduce) res = reducer[1] || undefined;

    // Error catcher
    var catcher = function(e) { stream.close(); err = e; res = undefined };

    stream.on("data", function(d) {
      var v = d.val();
<commit_msg>deepCopy the reduce start object<commit_after>/**
 * Manage access to data, be it to find, update or remove it
 */
var document = require('./document')
  , events = require('events')
  , util = require('util')
  , _ = require('lodash')
  ;


var INDEX_BUILDING_DEBOUNCE = 10;
var LIVE_QUERY_DEBOUNCE = 30;
var AUTO_INDEXING = true;


/**
 * Create a new cursor for this collection
 * @param {Datastore} db - The datastore this cursor is bound to
 * @param {Query} query - The query this cursor will operate on
 * @param {Function} execDn - Handler to be executed after cursor has found the results and before the callback passed to find/findOne/update/remove
 */
function Cursor (db, query, execFn) {
  this.db = db;
  this.query = query || {};
  if (execFn) { this.execFn = execFn; }
}


/**
 * Set a limit to the number of results
 */
Cursor.prototype.limit = function(limit) {
  this._limit = limit;
  return this;
};


/**
 * Skip a the number of results
 */
Cursor.prototype.skip = function(skip) {
  this._skip = skip;
  return this;
};


/**
 * Filter the results of the query - must be a function
 */
Cursor.prototype.filter = function(filter) {
  if (typeof(filter) != "function") return this;
  this._filter = filter;
  return this;
};


/**
 * Re-define the query itself; used for live queries
 */
Cursor.prototype.find = function(query) {
  this.query = query || this.query;
  return this;
};

/**
 * Sort results of the query
 * @param {SortQuery} sortQuery - SortQuery is { field: order }, field can use the dot-notation, order is 1 for ascending and -1 for descending
 */
Cursor.prototype.sort = function(sortQuery) {
  this._sort = sortQuery;
  return this;
};


/**
 * Add the use of a mapping function
 * @param {Function} map - map function, takes the object as the first argument
 */
Cursor.prototype.map = function(map) {
  if (typeof(map)=="function") this._map = map;
  return this;
};

/**
 * Add the use of a reducing function
 * @param {Function} reduce - reduce function, takes the two objects
 */
Cursor.prototype.reduce = function(reduce, initial) {
  if (typeof(reduce)=="function") {
    this._reduce = initial===undefined ? [reduce] : [reduce,initial];
  }
  return this;
};


/**
 * Aggregating function
 */
Cursor.prototype.aggregate = function(aggr) {
  this._aggregate = aggr;
  return this;
};

/**
 * Get all matching elements
 * Will return pointers to matched elements (shallow copies), returning full copies is the role of find or findOne
 *
 * @param {Function} callback - Signature: err, results
 */
Cursor.prototype.exec = function(callback) {
  var res = [], resIds = {}, resById = {}, self = this
    , err = null;

  var sort, sorter;
  if (typeof(this._sort) == "function") sorter = this._sort;
  else sort = this._sort;

  var reducer;
  if (self._reduce) reducer = _.cloneDeep(self._reduce);

  if (! this._quiet) this.db.emit("query", this.query);

  var stream = Cursor.getMatchesStream(this.db, this.query, sort, this._prefetched);
  stream.on("error", function(e) { callback(e) });
  stream.removeListener("ids", stream.trigger);
  stream.on("ids", function(ids) {
    var indexed = ids._indexed, sorted = ids._sorted; // for some reason we cannot access those in on 'ready' - maybe properties get erased from arrays?
    var earlyLimit = indexed && sorted && !self._filter && !sorter; // query is indexed, sorter and we don't have special filter/sort funcs; TODO: we should also set this to true if there's no limit/skip
    var earlySort = (sorted || !sort) && !sorter; // If ids are already sorted
    var earlyMapReduce = earlyLimit && earlySort && self._map && reducer; // special mode - run the map/reduce directly on data event
    // TODO: earlyMapReduce even without the map

    if (earlyLimit) ids = limit(ids, self._limit || ids.length, self._skip || 0);

    // No need to go further, sort and limits are applied, we only need count
    if (earlyLimit && self._count) { res = ids; return ready(); }

    // Start retrieving the objects for those IDs; res is an array that will hold all results
    stream.trigger(ids);
    res = new Array(ids.length);

    // Special case: res becomes the cursor of reduce if we're running a map/reduce
    if (earlyMapReduce) res = reducer[1] ? document.deepCopy(reducer[1]) : undefined;

    // Error catcher
    var catcher = function(e) { stream.close(); err = e; res = undefined };

    stream.on("data", function(d) {
      var v = d.val();

{'fraction_non_alphanumeric': 0.12853267013339362, 'fraction_numerical': 0.0019217725525661315, 'mean_word_length': 3.4726996966632964, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4065719', 'n_tokens_mistral': 2729, 'n_tokens_neox': 2638, 'n_words': 969}