path: root/jstests/sharding/zbigMapReduce.js

/**
 * This test is labeled resource intensive because its total io_write is 625MB compared to a median
 * of 5MB across all sharding tests in wiredTiger.
 * @tags: [resource_intensive]
 */
(function() {
'use strict';

let s = new ShardingTest({
    shards: 2,
    mongos: 1,
    other: {
        rs: true,
        numReplicas: 2,
        chunkSize: 1,
        rsOptions: {oplogSize: 50},
        enableAutoSplit: true,
    }
});

assert.commandWorked(s.s0.adminCommand({enablesharding: "test"}));
s.ensurePrimaryShard('test', s.shard0.shardName);
assert.commandWorked(s.s0.adminCommand({shardcollection: "test.foo", key: {"_id": 1}}));

let testDb = s.getDB("test");

jsTest.log("Inserting a lot of documents into test.foo");

// Make each document data to be 5K so that the total size is ~250MB
const str = "#".repeat(5 * 1024);

var idInc = 0;
var valInc = 0;

var bulk = testDb.foo.initializeUnorderedBulkOp();
for (var j = 0; j < 100; j++) {
    for (var i = 0; i < 512; i++) {
        bulk.insert({i: idInc++, val: valInc++, y: str});
    }
}
assert.writeOK(bulk.execute({w: 2, wtimeout: 10 * 60 * 1000}));

jsTest.log("Documents inserted, doing double-checks of insert...");

// Collect some useful stats to figure out what happened
if (testDb.foo.find().itcount() != 51200) {
    s.printShardingStatus(true);

    print("Shard 0: " + s.shard0.getCollection(testDb.foo + "").find().itcount());
    print("Shard 1: " + s.shard1.getCollection(testDb.foo + "").find().itcount());

    for (var i = 0; i < 51200; i++) {
        if (!testDb.foo.findOne({i: i}, {i: 1})) {
            print("Could not find: " + i);
        }

        if (i % 100 == 0)
            print("Checked " + i);
    }

    assert(false, 'Incorect number of chunks found!');
}

s.printChunks(testDb.foo.getFullName());
s.printChangeLog();

function map() {
    emit('count', 1);
}
function reduce(key, values) {
    return Array.sum(values);
}

// Let chunks move around while map reduce is running
s.startBalancer();

jsTest.log("Test basic mapreduce...");

// Test basic mapReduce
for (var iter = 0; iter < 5; iter++) {
    print("Test #" + iter);
    testDb.foo.mapReduce(map, reduce, "big_out");
}

print("Testing output to different db...");

// Test output to a different DB - do it multiple times so that the merging shard changes
for (var iter = 0; iter < 5; iter++) {
    print("Test #" + iter);

    assert.eq(51200, testDb.foo.find().itcount(), "Not all data was found!");

    let outCollStr = "mr_replace_col_" + iter;
    let outDbStr = "mr_db_" + iter;

    print("Testing mr replace into DB " + iter);

    var res = testDb.foo.mapReduce(map, reduce, {out: {replace: outCollStr, db: outDbStr}});
    printjson(res);

    var outDb = s.getDB(outDbStr);
    var outColl = outDb[outCollStr];

    var obj = outColl.convertToSingleObject("value");
    assert.eq(51200, obj.count, "Received wrong result " + obj.count);

    print("Checking result field");
    assert.eq(res.result.collection, outCollStr, "Wrong collection " + res.result.collection);
    assert.eq(res.result.db, outDbStr, "Wrong db " + res.result.db);
}

jsTest.log("Verifying nonatomic M/R throws...");

// Check nonAtomic output
assert.throws(function() {
    testDb.foo.mapReduce(map, reduce, {out: {replace: "big_out", nonAtomic: true}});
});

jsTest.log("Adding documents");

// Add docs with dup "i"
valInc = 0;
for (var j = 0; j < 100; j++) {
    print("Inserted document: " + (j * 100));
    var bulk = testDb.foo.initializeUnorderedBulkOp();
    for (i = 0; i < 512; i++) {
        bulk.insert({i: idInc++, val: valInc++, y: str});
    }
    assert.writeOK(bulk.execute({w: 2, wtimeout: 10 * 60 * 1000}));
}

jsTest.log("No errors...");

function map2() {
    emit(this.val, 1);
}
function reduce2(key, values) {
    return Array.sum(values);
}

// Test merge
let outColMerge = 'big_out_merge';

// M/R quarter of the docs
{
    jsTestLog("Test A");
    var out =
        testDb.foo.mapReduce(map2, reduce2, {query: {i: {$lt: 25600}}, out: {merge: outColMerge}});
    printjson(out);
    assert.eq(25600, out.counts.emit, "Received wrong result");
    assert.eq(25600, out.counts.output, "Received wrong result");
}

// M/R further docs
{
    jsTestLog("Test B");
    var out = testDb.foo.mapReduce(
        map2, reduce2, {query: {i: {$gte: 25600, $lt: 51200}}, out: {merge: outColMerge}});
    printjson(out);
    assert.eq(25600, out.counts.emit, "Received wrong result");
    assert.eq(51200, out.counts.output, "Received wrong result");
}

// M/R do 2nd half of docs
{
    jsTestLog("Test C");
    var out = testDb.foo.mapReduce(
        map2, reduce2, {query: {i: {$gte: 51200}}, out: {merge: outColMerge, nonAtomic: true}});
    printjson(out);
    assert.eq(51200, out.counts.emit, "Received wrong result");
    assert.eq(51200, out.counts.output, "Received wrong result");
    assert.eq(1, testDb[outColMerge].findOne().value, "Received wrong result");
}

// Test reduce
let outColReduce = "big_out_reduce";

// M/R quarter of the docs
{
    jsTestLog("Test D");
    var out = testDb.foo.mapReduce(
        map2, reduce2, {query: {i: {$lt: 25600}}, out: {reduce: outColReduce}});
    printjson(out);
    assert.eq(25600, out.counts.emit, "Received wrong result");
    assert.eq(25600, out.counts.output, "Received wrong result");
}

// M/R further docs
{
    jsTestLog("Test E");
    var out = testDb.foo.mapReduce(
        map2, reduce2, {query: {i: {$gte: 25600, $lt: 51200}}, out: {reduce: outColReduce}});
    printjson(out);
    assert.eq(25600, out.counts.emit, "Received wrong result");
    assert.eq(51200, out.counts.output, "Received wrong result");
}

// M/R do 2nd half of docs
{
    jsTestLog("Test F");
    var out = testDb.foo.mapReduce(
        map2, reduce2, {query: {i: {$gte: 51200}}, out: {reduce: outColReduce, nonAtomic: true}});
    printjson(out);
    assert.eq(51200, out.counts.emit, "Received wrong result");
    assert.eq(51200, out.counts.output, "Received wrong result");
    assert.eq(2, testDb[outColReduce].findOne().value, "Received wrong result");
}

// Verify that data is also on secondary
{
    jsTestLog("Test G");
    var primary = s.rs0._master;
    var secondaries = s.rs0._slaves;

    // Stop the balancer to prevent new writes from happening and make sure that replication can
    // keep up even on slow machines
    s.stopBalancer();
    s.rs0.awaitReplication();
    assert.eq(51200, primary.getDB("test")[outColReduce].find().itcount(), "Wrong count");

    for (var i = 0; i < secondaries.length; ++i) {
        assert.eq(
            51200, secondaries[i].getDB("test")[outColReduce].find().itcount(), "Wrong count");
    }
}

s.stop();
})();