Merge branch 'master' into feature/godriver-3544-metadata-on-demand

Author: matthewdale

.evergreen/config.yml

@@ -1741,10 +1741,6 @@ axes:
         display_name: "4.2"
         variables:
           VERSION: "4.2"
-      - id: "4.0"
-        display_name: "4.0"
-        variables:
-          VERSION: "4.0"
       - id: "rapid"
         display_name: "rapid"
         variables:
@@ -2170,19 +2166,6 @@ buildvariants:
     tasks:
       - name: ".goleak"
 
-  - matrix_name: "tests-rhel-40-with-snappy-support"
-    tags: ["pullrequest"]
-    matrix_spec: { version: ["4.0"], os-ssl-40: ["rhel87-64"] }
-    display_name: "${version} ${os-ssl-40}"
-    tasks:
-      - name: ".test !.enterprise-auth !.zlib !.zstd"
-
-  - matrix_name: "tests-windows-40-with-snappy-support"
-    matrix_spec: { version: ["4.0"], os-ssl-40: ["windows-64"] }
-    display_name: "${version} ${os-ssl-40}"
-    tasks:
-      - name: ".test !.enterprise-auth !.zlib !.zstd"
-
   - matrix_name: "tests-rhel-44-plus-zlib-zstd-support"
     tags: ["pullrequest"]
     matrix_spec: { version: ["4.2", "4.4", "5.0", "6.0", "7.0", "8.0"], os-ssl-40: ["rhel87-64"] }

README.md

@@ -24,7 +24,7 @@ The MongoDB Go driver follows [semantic versioning](https://semver.org/) for its
 
 - Go 1.19 or higher. We aim to support the latest versions of Go.
 - Go 1.23 or higher is required to run the driver test suite.
-- MongoDB 4.0 and higher.
+- MongoDB 4.2 and higher.
 
 ## Installation
 

internal/docexamples/examples_test.go

@@ -131,14 +131,8 @@ func TestDocumentationExamples(t *testing.T) {
 		docexamples.ChangeStreamExamples(mt.T, csdb)
 	})
 
-	// Causal consistency examples cannot run on 4.0.
-	// TODO(GODRIVER-2238): Remove version filtering once failures on 4.0 sharded clusters are fixed.
-	mtOpts = mtest.NewOptions().MinServerVersion("4.2").Topologies(mtest.ReplicaSet)
-	mt.RunOpts("CausalConsistencyExamples/post4.2", mtOpts, func(mt *mtest.T) {
-		docexamples.CausalConsistencyExamples(mt.Client)
-	})
-	mtOpts = mtest.NewOptions().MaxServerVersion("4.0").Topologies(mtest.ReplicaSet)
-	mt.RunOpts("CausalConsistencyExamples/pre4.0", mtOpts, func(mt *mtest.T) {
+	mtOpts = mtest.NewOptions().Topologies(mtest.ReplicaSet)
+	mt.RunOpts("CausalConsistencyExamples", mtOpts, func(mt *mtest.T) {
 		docexamples.CausalConsistencyExamples(mt.Client)
 	})
 }

internal/driverutil/description.go

@@ -21,7 +21,7 @@ import (
 )
 
 const (
-	MinWireVersion = 7
+	MinWireVersion = 8
 	MaxWireVersion = 25
 )
 

internal/integration/change_stream_test.go

@@ -259,34 +259,6 @@ func TestChangeStream_ReplicaSet(t *testing.T) {
 		assert.Nil(mt, cs.Err(), "change stream error: %v", cs.Err())
 	})
 
-	startAtOpTimeOpts := mtest.NewOptions().MinServerVersion("4.0").MaxServerVersion("4.0.6")
-	mt.RunOpts("include startAtOperationTime", startAtOpTimeOpts, func(mt *mtest.T) {
-		// $changeStream stage for ChangeStream against a server >=4.0 and <4.0.7 that has not received any results yet
-		// MUST include a startAtOperationTime option when resuming a changestream.
-
-		cs, err := mt.Coll.Watch(context.Background(), mongo.Pipeline{})
-		assert.Nil(mt, err, "Watch error: %v", err)
-		defer closeStream(cs)
-
-		generateEvents(mt, 1)
-		// kill cursor to force resumable error
-		killChangeStreamCursor(mt, cs)
-
-		mt.ClearEvents()
-		// change stream should resume once and get new change
-		assert.True(mt, cs.Next(context.Background()), "expected Next to return true, got false")
-		// Next should cause getMore, killCursors, and aggregate to run
-		assert.NotNil(mt, mt.GetStartedEvent(), "expected getMore event, got nil")
-		assert.NotNil(mt, mt.GetStartedEvent(), "expected killCursors event, got nil")
-		aggEvent := mt.GetStartedEvent()
-		assert.NotNil(mt, aggEvent, "expected aggregate event, got nil")
-		assert.Equal(mt, "aggregate", aggEvent.CommandName, "expected command name 'aggregate', got '%v'", aggEvent.CommandName)
-
-		// check for startAtOperationTime in pipeline
-		csStage := aggEvent.Command.Lookup("pipeline").Array().Index(0).Document() // $changeStream stage
-		_, err = csStage.Lookup("$changeStream").Document().LookupErr("startAtOperationTime")
-		assert.Nil(mt, err, "startAtOperationTime not included in aggregate command")
-	})
 	mt.RunOpts("decode does not panic", noClientOpts, func(mt *mtest.T) {
 		testCases := []struct {
 			name             string
@@ -418,53 +390,6 @@ func TestChangeStream_ReplicaSet(t *testing.T) {
 					})
 				}
 			})
-
-			noPbrtOpts := mtest.NewOptions().MaxServerVersion("4.0.6")
-			mt.RunOpts("without PBRT support", noPbrtOpts, func(mt *mtest.T) {
-				collName := mt.Coll.Name()
-				dbName := mt.Coll.Database().Name()
-				cs, err := mt.Coll.Watch(context.Background(), mongo.Pipeline{})
-				assert.Nil(mt, err, "Watch error: %v", err)
-				defer closeStream(cs)
-
-				compareResumeTokens(mt, cs, nil) // should be no resume token because no PBRT
-				numEvents := 5
-				generateEvents(mt, numEvents)
-				// iterate once to get a resume token
-				assert.True(mt, cs.Next(context.Background()), "expected Next to return true, got false")
-				token := cs.ResumeToken()
-				assert.NotNil(mt, token, "expected resume token, got nil")
-
-				testCases := []struct {
-					name            string
-					opts            *options.ChangeStreamOptionsBuilder
-					iterateStream   bool // whether or not resulting change stream should be iterated
-					initialToken    bson.Raw
-					numDocsExpected int
-				}{
-					{"resumeAfter", options.ChangeStream().SetResumeAfter(token), true, token, numEvents - 1},
-					{"no options", nil, false, nil, 0},
-				}
-				for _, tc := range testCases {
-					mt.Run(tc.name, func(mt *mtest.T) {
-						coll := mt.Client.Database(dbName).Collection(collName)
-						cs, err := coll.Watch(context.Background(), mongo.Pipeline{}, tc.opts)
-						assert.Nil(mt, err, "Watch error: %v", err)
-						defer closeStream(cs)
-
-						compareResumeTokens(mt, cs, tc.initialToken)
-						if !tc.iterateStream {
-							return
-						}
-
-						for i := 0; i < tc.numDocsExpected; i++ {
-							assert.True(mt, cs.Next(context.Background()), "expected Next to return true, got false")
-							// current resume token should always equal _id of current document
-							compareResumeTokens(mt, cs, cs.Current.Lookup("_id").Document())
-						}
-					})
-				}
-			})
 		})
 	})
 	mt.RunOpts("try next", noClientOpts, func(mt *mtest.T) {
@@ -826,17 +751,6 @@ func generateEvents(mt *mtest.T, numEvents int) {
 	}
 }
 
-func killChangeStreamCursor(mt *mtest.T, cs *mongo.ChangeStream) {
-	mt.Helper()
-
-	db := mt.Coll.Database().Client().Database("admin")
-	err := db.RunCommand(context.Background(), bson.D{
-		{"killCursors", mt.Coll.Name()},
-		{"cursors", bson.A{cs.ID()}},
-	}).Err()
-	assert.Nil(mt, err, "killCursors error: %v", err)
-}
-
 // returns pbrt, operationTime from aggregate command response
 func getAggregateResponseInfo(mt *mtest.T) (bson.Raw, bson.Timestamp) {
 	mt.Helper()

internal/integration/client_test.go

@@ -34,6 +34,7 @@ import (
 	"go.mongodb.org/mongo-driver/v2/x/bsonx/bsoncore"
 	"go.mongodb.org/mongo-driver/v2/x/mongo/driver"
 	"go.mongodb.org/mongo-driver/v2/x/mongo/driver/wiremessage"
+	"go.mongodb.org/mongo-driver/v2/x/mongo/driver/xoptions"
 	"golang.org/x/sync/errgroup"
 )
 
@@ -849,11 +850,96 @@ func TestClient_BulkWrite(t *testing.T) {
 		}
 
 		_, err := mt.Client.BulkWrite(context.Background(), writes)
-		require.NoError(t, err)
-		assert.Equal(t, 2, bulkWrites, "expected %d bulkWrites, got %d", 2, bulkWrites)
+		require.NoError(mt, err)
+		assert.Equal(mt, 2, bulkWrites, "expected %d bulkWrites, got %d", 2, bulkWrites)
 	})
 }
 
+func TestClient_BulkWrite_AddCommandFields(t *testing.T) {
+	newOpts := func(option bson.D) *options.ClientBulkWriteOptionsBuilder {
+		opts := options.ClientBulkWrite()
+		err := xoptions.SetInternalClientBulkWriteOptions(opts, "addCommandFields", option)
+		require.NoError(t, err, "unexpected error: %v", err)
+		return opts
+	}
+
+	marshalValue := func(val interface{}) bson.RawValue {
+		t.Helper()
+
+		valType, data, err := bson.MarshalValue(val)
+		require.NoError(t, err, "MarshalValue error: %v", err)
+		return bson.RawValue{
+			Type:  valType,
+			Value: data,
+		}
+	}
+
+	models := []struct {
+		name  string
+		model mongo.ClientWriteModel
+	}{
+		{
+			name:  "insert one",
+			model: mongo.NewClientInsertOneModel().SetDocument(bson.D{{"x", 1}}),
+		},
+		{
+			name:  "update one",
+			model: mongo.NewClientUpdateOneModel().SetFilter(bson.D{{"x", 1}}).SetUpdate(bson.D{{"$set", bson.D{{"x", 3.14159}}}}),
+		},
+		{
+			name:  "update many",
+			model: mongo.NewClientUpdateManyModel().SetFilter(bson.D{{"x", 1}}).SetUpdate(bson.D{{"$set", bson.D{{"x", 3.14159}}}}),
+		},
+		{
+			name:  "replace one",
+			model: mongo.NewClientReplaceOneModel().SetFilter(bson.D{{"x", 1}}).SetReplacement(bson.D{{"x", 3.14159}}),
+		},
+	}
+
+	testCases := []struct {
+		name     string
+		opts     *options.ClientBulkWriteOptionsBuilder
+		expected bson.RawValue
+	}{
+		{
+			name:     "empty",
+			opts:     options.ClientBulkWrite(),
+			expected: bson.RawValue{},
+		},
+		{
+			name:     "false",
+			opts:     newOpts(bson.D{{"bypassEmptyTsReplacement", false}}),
+			expected: marshalValue(false),
+		},
+		{
+			name:     "true",
+			opts:     newOpts(bson.D{{"bypassEmptyTsReplacement", true}}),
+			expected: marshalValue(true),
+		},
+	}
+
+	mtBulkWriteOpts := mtest.NewOptions().MinServerVersion("8.0").ClientType(mtest.Pinned)
+	mt := mtest.New(t, noClientOpts)
+	for _, m := range models {
+		for _, tc := range testCases {
+			mt.RunOpts(fmt.Sprintf("%s %s", m.name, tc.name), mtBulkWriteOpts, func(mt *mtest.T) {
+				mt.Parallel()
+
+				writes := []mongo.ClientBulkWrite{{
+					Database:   "foo",
+					Collection: "bar",
+					Model:      m.model,
+				}}
+				_, err := mt.Client.BulkWrite(context.Background(), writes, tc.opts)
+				require.NoError(mt, err, "BulkWrite error: %v", err)
+				evt := mt.GetStartedEvent()
+				val := evt.Command.Lookup("bypassEmptyTsReplacement")
+				assert.Equal(mt, tc.expected, val, "expected bypassEmptyTsReplacement to be %s", tc.expected.String())
+			})
+		}
+	}
+}
+
 func TestClient_BSONOptions(t *testing.T) {
 	t.Parallel()