<commit_before>var queue = require('queue-async');

module.exports = {};
module.exports.seek = seek;
module.exports.shard = shard;
module.exports.getFeature = getFeature;
module.exports.getAllFeatures = getAllFeatures;
module.exports.putFeature = putFeature;
module.exports.putFeatures = putFeatures;

// Seek to individual entry in JSON shard without doing a JSON.parse()
// against the entire shard. This is a performance optimization.
function seek(buffer, id) {
    buffer = typeof buffer === 'string' ? buffer : buffer.toString();

    var start = buffer.indexOf('"'+id+'":"{');
    if (start === -1) return false;

    start = buffer.indexOf('"{', start);
    var end = buffer.indexOf('}"', start);
    var entry = buffer.slice(start, end+2);

    return JSON.parse(JSON.parse(entry));
}

function getFeature(source, id, callback) {
    var s = shard(source._geocoder.shardlevel, id);
    source._geocoder.fcache = source._geocoder.fcache || {};
    source.getGeocoderData('feature', s, function(err, buffer) {
        if (err) return callback(err);
        var data;
        try {
            data = seek(buffer, id);
        } catch(err) {
            return callback(err);
        }
        return callback(null, data);
    });
}

function getAllFeatures(source, callback) {
    var limit = Math.pow(16, source._geocoder.shardlevel+1);
    var docs = [];
    var q = queue();
    for (var i = 0; i < limit; i++) q.defer(function(shard, callback) {
        source.getGeocoderData('feature', shard, function(err, buffer) {
            if (err) return callback(err);
            try {
                var data = buffer && buffer.length ? JSON.parse(buffer) : {};
                for (var hash in data) {
                    var chunk = JSON.parse(data[hash]);
                    for (var x in chunk) docs.push(chunk[x]);
                }
                return callback();
            } catch (err) {
                return callback(err);
            }
        });
    }, i);
    q.awaitAll(function(err) {
        if (err) return callback(err);
        callback(null, docs);
    });
}

function putFeature(source, id, data, callback) {
    var s = shard(source._geocoder.shardlevel, id);
    source.getGeocoderData('feature', s, function(err, buffer) {
        if (err) return callback(err);
        try {
            var current = buffer && buffer.length ? JSON.parse(buffer) : {};
            if (current[id]) {
                var merged = JSON.parse(current[id]);
                for (var key in data) merged[key] = data[key];
                current[id] = JSON.stringify(merged);
            } else {
                current[id] = JSON.stringify(data);
            }
            source.putGeocoderData('feature', s, JSON.stringify(current), callback);
        } catch (err) { return callback(err); }
    });
}

function putFeatures(source, docs, callback) {
    var byshard = {};
    for (var i = 0; i < docs.length; i++) {
        var doc = docs[i];
        var s = shard(source._geocoder.shardlevel, doc._hash);
        byshard[s] = byshard[s] || [];
        byshard[s].push(doc);
    }
    var q = queue(100);
    for (var s in byshard) q.defer(function(s, docs, callback) {
        source.getGeocoderData('feature', s, function(err, buffer) {
            if (err) return callback(err);
            try {
                var current = buffer && buffer.length ? JSON.parse(buffer) : {};
                for (var i = 0; i < docs.length; i++) {
                    var doc = docs[i];
                    var entry = current[doc._hash] ? JSON.parse(current[doc._hash]) : {};
                    entry[doc._id] = doc;
                    current[doc._hash] = JSON.stringify(entry, cleanDoc);
                }
                source.putGeocoderData('feature', s, JSON.stringify(current), callback);
            } catch (err) {
                return callback(err);
            }
        });
    }, s, byshard[s]);
    q.awaitAll(callback);
}

// Strip temporary indexing attributes from feature docs.
function cleanDoc(key, val) {
    if (key === '_hash') return;
    if (key === '_grid') return;
    return val;
}

// Return the shard for a given shardlevel + id.
function shard(level, id) {
    if (id === undefined) return false;
    var mod = Math.pow(16,level+1);
    return Math.floor((id%(64*mod))/64);
}<commit_msg>Mess around with putFeatures efficiency.<commit_after>var queue = require('queue-async');

module.exports = {};
module.exports.seek = seek;
module.exports.shard = shard;
module.exports.getFeature = getFeature;
module.exports.getAllFeatures = getAllFeatures;
module.exports.putFeature = putFeature;
module.exports.putFeatures = putFeatures;

// Seek to individual entry in JSON shard without doing a JSON.parse()
// against the entire shard. This is a performance optimization.
function seek(buffer, id) {
    buffer = typeof buffer === 'string' ? buffer : buffer.toString();

    var start = buffer.indexOf('"'+id+'":"{');
    if (start === -1) return false;

    start = buffer.indexOf('"{', start);
    var end = buffer.indexOf('}"', start);
    var entry = buffer.slice(start, end+2);

    return JSON.parse(JSON.parse(entry));
}

function getFeature(source, id, callback) {
    var s = shard(source._geocoder.shardlevel, id);
    source._geocoder.fcache = source._geocoder.fcache || {};
    source.getGeocoderData('feature', s, function(err, buffer) {
        if (err) return callback(err);
        var data;
        try {
            data = seek(buffer, id);
        } catch(err) {
            return callback(err);
        }
        return callback(null, data);
    });
}

function getAllFeatures(source, callback) {
    var limit = Math.pow(16, source._geocoder.shardlevel+1);
    var docs = [];
    var q = queue();
    for (var i = 0; i < limit; i++) q.defer(function(shard, callback) {
        source.getGeocoderData('feature', shard, function(err, buffer) {
            if (err) return callback(err);
            try {
                var data = buffer && buffer.length ? JSON.parse(buffer) : {};
                for (var hash in data) {
                    var chunk = JSON.parse(data[hash]);
                    for (var x in chunk) docs.push(chunk[x]);
                }
                return callback();
            } catch (err) {
                return callback(err);
            }
        });
    }, i);
    q.awaitAll(function(err) {
        if (err) return callback(err);
        callback(null, docs);
    });
}

function putFeature(source, id, data, callback) {
    var s = shard(source._geocoder.shardlevel, id);
    source.getGeocoderData('feature', s, function(err, buffer) {
        if (err) return callback(err);
        try {
            var current = buffer && buffer.length ? JSON.parse(buffer) : {};
            if (current[id]) {
                var merged = JSON.parse(current[id]);
                for (var key in data) merged[key] = data[key];
                current[id] = JSON.stringify(merged);
            } else {
                current[id] = JSON.stringify(data);
            }
            source.putGeocoderData('feature', s, JSON.stringify(current), callback);
        } catch (err) { return callback(err); }
    });
}

function putFeatures(source, docs, callback) {
    var byshard = {};
    for (var i = 0; i < docs.length; i++) {
        var doc = docs[i];
        var s = shard(source._geocoder.shardlevel, doc._hash);
        byshard[s] = byshard[s] || [];
        byshard[s].push(doc);
    }
    var q = queue(100);
    for (var s in byshard) q.defer(function(s, docs, callback) {
        source.getGeocoderData('feature', s, function(err, buffer) {
            if (err) return callback(err);
            try {
                var current = buffer && buffer.length ? JSON.parse(buffer) : {};
                var hashes = [];
                var hashesUniq = {};
                for (var i = 0; i < docs.length; i++) {
                    var doc = docs[i];
                    if (typeof current[doc._hash] === 'string') {
                        current[doc._hash] = JSON.parse(current[doc._hash]);
                    } else if (typeof current[doc._hash] === 'undefined') {
                        current[doc._hash] = {};
                    }
                    if (!hashesUniq[doc._hash]) {
                        hashes.push(doc._hash);
                        hashesUniq[doc._hash] = true;
                    }
                    current[doc._hash][doc._id] = doc;
                }
                for (var i = 0; i < hashes.length; i++) {
                    current[hashes[i]] = JSON.stringify(current[hashes[i]], cleanDoc);
                }
                source.putGeocoderData('feature', s, JSON.stringify(current), callback);
            } catch (err) {
                return callback(err);
            }
        });
    }, s, byshard[s]);
    q.awaitAll(callback);
}

// Strip temporary indexing attributes from feature docs.
function cleanDoc(key, val) {
    if (key === '_hash') return;
    if (key === '_grid') return;
    return val;
}

// Return the shard for a given shardlevel + id.
function shard(level, id) {
    if (id === undefined) return false;
    var mod = Math.pow(16,level+1);
    return Math.floor((id%(64*mod))/64);
}
{'fraction_non_alphanumeric': 0.14378448555663745, 'fraction_numerical': 0.004003029319485016, 'mean_word_length': 2.058901389808074, 'pattern_counts': {'":': 2, '<': 10, '<?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': '2314754', 'n_tokens_mistral': 2792, 'n_tokens_neox': 2684, 'n_words': 680}