Allow partial results flag to be set

harness/daemons/.env

@@ -41,10 +41,12 @@ COMMONSOPTS="
 CFG1OPTS=""
 CFG2OPTS=""
 CFG3OPTS=""
+CFGXOPTS=""
 
 MONGOS1OPTS="--configdb 127.0.0.1:40101"
 MONGOS2OPTS="--configdb 127.0.0.1:40102"
 MONGOS3OPTS="--configdb 127.0.0.1:40103"
+MONGOSXOPTS="--configdb 127.0.0.1:40109"
 
 
 
@@ -72,10 +74,12 @@ if versionAtLeast 3 2; then
       CFG1OPTS="--replSet conf1"
       CFG2OPTS="--replSet conf2"
       CFG3OPTS="--replSet conf3"
+      CFGXOPTS="--replSet confx"
 
       MONGOS1OPTS="--configdb conf1/127.0.0.1:40101"
       MONGOS2OPTS="--configdb conf2/127.0.0.1:40102"
       MONGOS3OPTS="--configdb conf3/127.0.0.1:40103" 
+      MONGOSXOPTS="--configdb confx/127.0.0.1:40109" 
 	fi
 fi
 

harness/daemons/cfgx/log/run

@@ -0,0 +1,3 @@
+#!/bin/sh
+
+exec cat - > log.txt

harness/daemons/cfgx/run

@@ -0,0 +1,8 @@
+#!/bin/sh
+
+. ../.env
+
+exec mongod $COMMONCOPTS \
+	--port 40109 \
+	--configsvr \
+	$CFGXOPTS

harness/daemons/dbx1/log/run

@@ -0,0 +1,3 @@
+#!/bin/sh
+
+exec cat - > log.txt

harness/daemons/dbx1/run

@@ -0,0 +1,7 @@
+#!/bin/sh
+
+. ../.env
+
+exec mongod $COMMONDOPTS \
+	--port 40091 \
+    --shardsvr

harness/daemons/dbx2/log/run

@@ -0,0 +1,3 @@
+#!/bin/sh
+
+exec cat - > log.txt

harness/daemons/dbx2/run

@@ -0,0 +1,7 @@
+#!/bin/sh
+
+. ../.env
+
+exec mongod $COMMONDOPTS \
+	--port 40092 \
+    --shardsvr

harness/daemons/dbx3/log/run

@@ -0,0 +1,3 @@
+#!/bin/sh
+
+exec cat - > log.txt

harness/daemons/dbx3/run

@@ -0,0 +1,8 @@
+#!/bin/sh
+
+. ../.env
+
+exec mongod $COMMONDOPTS \
+	--port 40093 \
+	--shardsvr
+

harness/daemons/sx/log/run

@@ -0,0 +1,3 @@
+#!/bin/sh
+
+exec cat - > log.txt

harness/daemons/sx/run

@@ -0,0 +1,7 @@
+#!/bin/sh
+
+. ../.env
+
+exec mongos $COMMONSOPTS \
+	--port 40209 \
+	$MONGOSXOPTS

harness/mongojs/init.js

@@ -33,6 +33,10 @@ for (var i = 0; i != 60; i++) {
         cfg1 = new Mongo("127.0.0.1:40101").getDB("admin")
         cfg2 = new Mongo("127.0.0.1:40102").getDB("admin")
         cfg3 = new Mongo("127.0.0.1:40103").getDB("admin")
+        cfgx = new Mongo("127.0.0.1:40109").getDB("admin")
+        dbx1 = new Mongo("127.0.0.1:40091").getDB("admin")
+        dbx2 = new Mongo("127.0.0.1:40092").getDB("admin")
+        dbx3 = new Mongo("127.0.0.1:40093").getDB("admin")
         break
     } catch (err) {
         print("Can't connect yet...")
@@ -63,6 +67,7 @@ if (versionAtLeast(3, 4)) {
     cfg1.runCommand({ replSetInitiate: { _id: "conf1", configsvr: true, members: [{ "_id": 1, "host": "localhost:40101" }] } })
     cfg2.runCommand({ replSetInitiate: { _id: "conf2", configsvr: true, members: [{ "_id": 1, "host": "localhost:40102" }] } })
     cfg3.runCommand({ replSetInitiate: { _id: "conf3", configsvr: true, members: [{ "_id": 1, "host": "localhost:40103" }] } })
+    cfgx.runCommand({ replSetInitiate: { _id: "confx", configsvr: true, members: [{ "_id": 1, "host": "localhost:40109" }] } })
 }
 
 sleep(3000)
@@ -85,12 +90,12 @@ function configAuth() {
             try {
                 db.createUser({ user: "root", pwd: "rapadura", roles: ["root"] })
             } catch (err) {
-                // 3.2 consistently fails replication of creds on 40031 (config server) 
+                // 3.2 consistently fails replication of creds on 40031 (config server)
                 print("createUser command returned an error: " + err)
                 if (String(err).indexOf("timed out") >= 0) {
                     timedOut = true;
                 }
-                // on 3.6 cluster with keyFile, we sometimes get this error 
+                // on 3.6 cluster with keyFile, we sometimes get this error
                 if (String(err).indexOf("Cache Reader No keys found for HMAC that is valid for time")) {
                     sleep(500)
                     continue createUser;
@@ -145,6 +150,20 @@ function configShards() {
     addShard(s3, ["rs3/127.0.0.1:40031"])
 }
 
+function configShardedCluster() {
+    sx = new Mongo("127.0.0.1:40209")
+    addShard(sx.getDB("admin"), ["127.0.0.1:40091"])
+    addShard(sx.getDB("admin"), ["127.0.0.1:40092"])
+    addShard(sx.getDB("admin"), ["127.0.0.1:40093"])
+
+    sx.getDB("partial").getCollection("partial").find({})
+    sx.getDB("admin").runCommand({enableSharding:"partial"})
+    sx.getDB("admin").runCommand({shardCollection:"partial.partial", key : {_id : "hashed"}, numInitialChunks : 3})
+    for (var i = 0; i < 10; i++) {
+        sx.getDB("partial").getCollection("partial").insertOne({_id : i})
+    }
+}
+
 function countHealthy(rs) {
     var status = rs.runCommand({ replSetGetStatus: 1 })
     var count = 0
@@ -166,6 +185,9 @@ function countHealthy(rs) {
     return count
 }
 
+sleep(5000)
+configShardedCluster()
+
 var totalRSMembers = rs1cfg.members.length + rs2cfg.members.length + rs3cfg.members.length
 
 for (var i = 0; i != 60; i++) {

harness/setup.sh

@@ -32,6 +32,7 @@ start() {
         if [ x$COUNT = x$UP ]; then
             echo "Running init.js with mongo..."
             mongo --nodb ../harness/mongojs/init.js
+            svc -d daemons/dbx1/
             exit 0
         fi
         sleep 1

session.go

@@ -3732,6 +3732,13 @@ func (q *Query) SetMaxTime(d time.Duration) *Query {
 	return q
 }
 
+func (q *Query) AllowPartial() *Query {
+	q.m.Lock()
+	q.op.AllowPartial()
+	q.m.Unlock()
+	return q
+}
+
 // Snapshot will force the performed query to make use of an available
 // index on the _id field to prevent the same document from being returned
 // more than once in a single iteration. This might happen without this
@@ -3908,25 +3915,26 @@ func prepareFindOp(socket *mongoSocket, op *queryOp, limit int32) bool {
 	}
 
 	find := findCmd{
-		Collection:      op.collection[nameDot+1:],
-		Filter:          op.query,
-		Projection:      op.selector,
-		Sort:            op.options.OrderBy,
-		Skip:            op.skip,
-		Limit:           limit,
-		MaxTimeMS:       op.options.MaxTimeMS,
-		MaxScan:         op.options.MaxScan,
-		Hint:            op.options.Hint,
-		Min:             op.options.Min,
-		Max:             op.options.Max,
-		Comment:         op.options.Comment,
-		Snapshot:        op.options.Snapshot,
-		Collation:       op.options.Collation,
-		Tailable:        op.flags&flagTailable != 0,
-		AwaitData:       op.flags&flagAwaitData != 0,
-		OplogReplay:     op.flags&flagLogReplay != 0,
-		NoCursorTimeout: op.flags&flagNoCursorTimeout != 0,
-		ReadConcern:     readLevel{level: op.readConcern},
+		Collection:          op.collection[nameDot+1:],
+		Filter:              op.query,
+		Projection:          op.selector,
+		Sort:                op.options.OrderBy,
+		Skip:                op.skip,
+		Limit:               limit,
+		MaxTimeMS:           op.options.MaxTimeMS,
+		MaxScan:             op.options.MaxScan,
+		Hint:                op.options.Hint,
+		Min:                 op.options.Min,
+		Max:                 op.options.Max,
+		Comment:             op.options.Comment,
+		Snapshot:            op.options.Snapshot,
+		Collation:           op.options.Collation,
+		Tailable:            op.flags&flagTailable != 0,
+		AwaitData:           op.flags&flagAwaitData != 0,
+		OplogReplay:         op.flags&flagLogReplay != 0,
+		NoCursorTimeout:     op.flags&flagNoCursorTimeout != 0,
+		ReadConcern:         readLevel{level: op.readConcern},
+		AllowPartialResults: op.flags&flagPartial != 0,
 	}
 
 	if op.limit < 0 {

session_test.go

@@ -2045,6 +2045,34 @@ func (s *S) TestQueryComment(c *C) {
 	c.Assert(err, IsNil)
 }
 
+func (s *S) TestPartial(c *C) {
+	session, err := mgo.Dial("localhost:40209")
+	c.Assert(err, IsNil)
+	defer session.Close()
+
+	coll := session.DB("partial").C("partial")
+
+	expectedCount := 0
+	// These docs were inserted as a part of the testing setup process
+	expectedDocs := map[int]bool{0: true, 1: true, 2: true, 3: true, 4: true, 5: true, 7: true, 9: true}
+	iter := coll.Find(M{}).AllowPartial().Iter()
+	result := M{}
+	for {
+		gotDoc := iter.Next(&result)
+		if !gotDoc {
+			break
+		}
+		expectedCount += 1
+		_, ok := expectedDocs[int(result["_id"].(float64))]
+		c.Assert(ok, Equals, true)
+	}
+	c.Assert(iter.Err(), IsNil)
+	c.Assert(expectedCount, Equals, len(expectedDocs))
+
+	err = coll.Find(M{}).One(&result)
+	c.Assert(err, Not(IsNil))
+}
+
 func (s *S) TestFindOneNotFound(c *C) {
 	session, err := mgo.Dial("localhost:40001")
 	c.Assert(err, IsNil)

socket.go

@@ -68,6 +68,8 @@ const (
 	flagLogReplay
 	flagNoCursorTimeout
 	flagAwaitData
+	flagExhaust
+	flagPartial
 )
 
 type queryOp struct {
@@ -100,6 +102,10 @@ type queryWrapper struct {
 	Collation      *Collation  `bson:"$collation,omitempty"`
 }
 
+func (op *queryOp) AllowPartial() {
+	op.flags |= flagPartial
+}
+
 func (op *queryOp) finalQuery(socket *mongoSocket) interface{} {
 	if op.flags&flagSlaveOk != 0 && socket.ServerInfo().Mongos {
 		var modeName string