/**
 * 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();
})();