path: root/jstests/sharding/zbigMapReduce.js

// This test is skipped on 32-bit platforms

function setupTest() {
    var s = new ShardingTest({ shards: 2,
                               mongos: 1,
                               other: { rs: true,
                                        numReplicas: 2,
                                        chunkSize: 1,
                                        rsOptions: { oplogSize: 50 },
                                        enableBalancer: 1 } });

    // Reduce chunk size to split
    var config = s.getDB("config");
    config.settings.save({_id: "chunksize", value: 1});

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

function runTest(s) {
    jsTest.log("Inserting a lot of documents into test.foo");
    db = s.getDB("test");

    var idInc = 0;
    var valInc = 0;
    var str = "";

    if (db.serverBuildInfo().bits == 32) {
        // Make data ~0.5MB for 32 bit builds
        for (var i = 0; i < 512; i++) str += "a";
    }
    else {
        // Make data ~4MB
        for (var i = 0; i < 4*1024; i++) str += "a";
    }

    var bulk = db.foo.initializeUnorderedBulkOp();
    for (j=0; j<100; j++) {
        for (i=0; i<512; i++) {
            bulk.insert({ i: idInc++, val: valInc++, y:str });
        }
    }
    assert.writeOK(bulk.execute());

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

    // Collect some useful stats to figure out what happened
    if (db.foo.find().itcount() != 51200) {
        sleep(1000);

        s.printShardingStatus(true);

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

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

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

        print("PROBABLY WILL ASSERT NOW");
    }

    assert.soon(function() {
        var c = db.foo.find().itcount();
        if (c == 51200) {
            return true;
        }

        print("Count is " + c);
        return false;
    });

    s.printChunks();
    s.printChangeLog();

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

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

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

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

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

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

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

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

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

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

        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() {
        db.foo.mapReduce(map, reduce, { out: {replace: "big_out", nonAtomic: true } });
    });

    jsTest.log();

    // Add docs with dup "i"
    valInc = 0;
    for (j=0; j<100; j++) {
        print("Inserted document: " + (j * 100));
        bulk = db.foo.initializeUnorderedBulkOp();
        for (i=0; i<512; i++) {
            bulk.insert({ i: idInc++, val: valInc++, y: str });
        }
        // wait for replication to catch up
        assert.writeOK(bulk.execute({ w: 2 }));
    }

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

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

    // Test merge
    outcol = "big_out_merge";

    // M/R quarter of the docs
    jsTestLog("Test A");
    out = db.foo.mapReduce(map2, reduce2, { query: {i: {$lt: 25600} }, out: { merge: outcol } });
    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");
    out = db.foo.mapReduce(
        map2, reduce2, { query: {i: {$gte: 25600, $lt: 51200} }, out: { merge: outcol } });
    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");
    out = db.foo.mapReduce(
        map2, reduce2, { query: {i: {$gte: 51200} }, out: { merge: outcol, 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 , db[outcol].findOne().value , "Received wrong result");

    // Test reduce
    jsTestLog("Test D");
    outcol = "big_out_reduce";

    // M/R quarter of the docs
    out = db.foo.mapReduce(map2, reduce2,{ query: { i: { $lt: 25600 } }, out: { reduce: outcol } });
    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");
    out = db.foo.mapReduce(
        map2, reduce2, { query: { i: { $gte: 25600, $lt: 51200 } }, out: { reduce: outcol } });
    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");
    out = db.foo.mapReduce(
        map2, reduce2, { query: { i: {$gte: 51200} }, out: { reduce: outcol, 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 , db[outcol].findOne().value , "Received wrong result");

    // Verify that data is also on secondary
    jsTestLog("Test G");
    var primary = s._rs[0].test.liveNodes.master;
    var secondaries = s._rs[0].test.liveNodes.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._rs[0].test.awaitReplication(300 * 1000);
    assert.eq(51200 , primary.getDB("test")[outcol].count() , "Wrong count");
    for (var i = 0; i < secondaries.length; ++i) {
        assert.eq(51200 , secondaries[i].getDB("test")[outcol].count() , "Wrong count");
    }
}

var s = setupTest();

if (s.getDB("admin").runCommand("buildInfo").bits < 64) {
    print("Skipping test on 32-bit platforms");
}
else {
    runTest(s);
}

s.stop();