code polish

Author: qingyang-hu

internal/integration/client_test.go

@@ -719,6 +719,11 @@ func TestClient(t *testing.T) {
 			})
 		}
 	})
+}
+
+func TestClient_BulkWrite(t *testing.T) {
+	mt := mtest.New(t, noClientOpts)
+
 	mtBulkWriteOpts := mtest.NewOptions().MinServerVersion("8.0").AtlasDataLake(false).ClientType(mtest.Pinned)
 	mt.RunOpts("bulk write with nil filter", mtBulkWriteOpts, func(mt *mtest.T) {
 		mt.Parallel()
@@ -808,6 +813,29 @@ func TestClient(t *testing.T) {
 			})
 		}
 	})
+	var bulkWrites int
+	cmdMonitor := &event.CommandMonitor{
+		Started: func(_ context.Context, evt *event.CommandStartedEvent) {
+			if evt.CommandName == "bulkWrite" {
+				bulkWrites++
+			}
+		},
+	}
+	clientOpts := options.Client().SetMonitor(cmdMonitor)
+	mt.RunOpts("bulk write with large messages", mtBulkWriteOpts.ClientOptions(clientOpts), func(mt *mtest.T) {
+		mt.Parallel()
+
+		document := bson.D{{"largeField", strings.Repeat("a", 16777216-100)}} // Adjust size to account for BSON overhead
+		writes := []mongo.ClientBulkWrite{
+			{"db", "x", mongo.NewClientInsertOneModel().SetDocument(document)},
+			{"db", "x", mongo.NewClientInsertOneModel().SetDocument(document)},
+			{"db", "x", mongo.NewClientInsertOneModel().SetDocument(document)},
+		}
+
+		_, err := mt.Client.BulkWrite(context.Background(), writes)
+		require.NoError(t, err)
+		assert.Equal(t, 2, bulkWrites, "expected %d bulkWrites, got %d", 2, bulkWrites)
+	})
 }
 
 func TestClient_BSONOptions(t *testing.T) {

internal/integration/crud_prose_test.go

@@ -775,8 +775,8 @@ func TestClientBulkWrite(t *testing.T) {
 		assert.Error(mt, err, "expected a BulkWrite error")
 		bwe, ok := err.(mongo.ClientBulkWriteException)
 		require.True(mt, ok, "expected a BulkWriteException, got %T: %v", err, err)
-		require.NotNil(mt, bwe.TopLevelError)
-		assert.Equal(mt, 8, bwe.TopLevelError.Code, "expected top level error code: %d, got; %d", 8, bwe.TopLevelError.Code)
+		require.NotNil(mt, bwe.WriteError)
+		assert.Equal(mt, 8, bwe.WriteError.Code, "expected top level error code: %d, got; %d", 8, bwe.WriteError.Code)
 		require.NotNil(mt, bwe.PartialResult)
 		assert.Equal(mt, int64(2), bwe.PartialResult.UpsertedCount, "expected UpsertedCount: %d, got: %d", 2, bwe.PartialResult.UpsertedCount)
 		assert.Len(mt, bwe.PartialResult.UpdateResults, 1, "expected %d UpdateResults, got: %d", 1, len(bwe.PartialResult.UpdateResults))

internal/integration/unified/error.go

@@ -218,9 +218,9 @@ func extractErrorDetails(err error) (errorDetails, bool) {
 		}
 		details.labels = converted.Labels
 	case mongo.ClientBulkWriteException:
-		if converted.TopLevelError != nil {
-			details.raw = converted.TopLevelError.Raw
-			details.codes = append(details.codes, int32(converted.TopLevelError.Code))
+		if converted.WriteError != nil {
+			details.raw = converted.WriteError.Raw
+			details.codes = append(details.codes, int32(converted.WriteError.Code))
 		}
 	default:
 		return errorDetails{}, false

mongo/bulk_write.go

@@ -295,11 +295,11 @@ func createDeleteDoc(
 	bsonOpts *options.BSONOptions,
 	registry *bson.Registry,
 ) (bsoncore.Document, error) {
+	if filter == nil {
+		return nil, fmt.Errorf("delete filter cannot be nil")
+	}
 	f, err := marshal(filter, bsonOpts, registry)
 	if err != nil {
-		if filter == nil {
-			return nil, fmt.Errorf("%w: filter is required", err)
-		}
 		return nil, err
 	}
 
@@ -430,11 +430,11 @@ type updateDoc struct {
 }
 
 func (doc updateDoc) marshal(bsonOpts *options.BSONOptions, registry *bson.Registry) (bsoncore.Document, error) {
+	if doc.filter == nil {
+		return nil, fmt.Errorf("update filter cannot be nil")
+	}
 	f, err := marshal(doc.filter, bsonOpts, registry)
 	if err != nil {
-		if doc.filter == nil {
-			return nil, fmt.Errorf("%w: filter is required", err)
-		}
 		return nil, err
 	}
 

mongo/client_bulk_write.go

@@ -91,7 +91,7 @@ func (bw *clientBulkWrite) execute(ctx context.Context) error {
 	var ce CommandError
 	if errors.As(err, &ce) {
 		exception = &ClientBulkWriteException{
-			TopLevelError: &WriteError{
+			WriteError: &WriteError{
 				Code:    int(ce.Code),
 				Message: ce.Message,
 				Raw:     ce.Raw,
@@ -435,7 +435,7 @@ func (mb *modelBatches) processResponse(ctx context.Context, resp bsoncore.Docum
 	}
 	if !res.Ok {
 		return ClientBulkWriteException{
-			TopLevelError: &WriteError{
+			WriteError: &WriteError{
 				Code:    int(res.Code),
 				Message: res.Errmsg,
 				Raw:     bson.Raw(resp),
@@ -616,11 +616,11 @@ func (d *clientUpdateDoc) marshal(bsonOpts *options.BSONOptions, registry *bson.
 
 	doc = bsoncore.AppendInt32Element(doc, "update", int32(d.namespace))
 
+	if d.filter == nil {
+		return nil, fmt.Errorf("update filter cannot be nil")
+	}
 	f, err := marshal(d.filter, bsonOpts, registry)
 	if err != nil {
-		if d.filter == nil {
-			return nil, fmt.Errorf("%w: filter is required", err)
-		}
 		return nil, err
 	}
 	doc = bsoncore.AppendDocumentElement(doc, "filter", f)
@@ -676,11 +676,11 @@ func (d *clientDeleteDoc) marshal(bsonOpts *options.BSONOptions, registry *bson.
 
 	doc = bsoncore.AppendInt32Element(doc, "delete", int32(d.namespace))
 
+	if d.filter == nil {
+		return nil, fmt.Errorf("delete filter cannot be nil")
+	}
 	f, err := marshal(d.filter, bsonOpts, registry)
 	if err != nil {
-		if d.filter == nil {
-			return nil, fmt.Errorf("%w: filter is required", err)
-		}
 		return nil, err
 	}
 	doc = bsoncore.AppendDocumentElement(doc, "filter", f)

mongo/client_test.go

@@ -11,7 +11,6 @@ import (
 	"errors"
 	"math"
 	"os"
-	"strings"
 	"testing"
 	"time"
 
@@ -515,39 +514,4 @@ func TestClient(t *testing.T) {
 		errmsg := `invalid value "-1s" for "Timeout": value must be positive`
 		assert.Equal(t, errmsg, err.Error(), "expected error %v, got %v", errmsg, err.Error())
 	})
-	t.Run("bulkWrite with large messages", func(t *testing.T) {
-		var bulkWrites int
-		cmdMonitor := &event.CommandMonitor{
-			Started: func(_ context.Context, evt *event.CommandStartedEvent) {
-				if evt.CommandName == "bulkWrite" {
-					bulkWrites++
-				}
-			},
-		}
-		cs := integtest.ConnString(t)
-		clientOpts := options.Client().ApplyURI(cs.Original).SetMonitor(cmdMonitor)
-		integtest.AddTestServerAPIVersion(clientOpts)
-		client, err := Connect(clientOpts)
-		assert.Nil(t, err, "Connect error: %v", err)
-		defer func() {
-			_ = client.Disconnect(bgCtx)
-		}()
-
-		serverVersion, err := getServerVersion(client.Database("admin"))
-		require.NoError(t, err)
-		if compareVersions(serverVersion, "8.0.0") < 1 {
-			t.Skip("skipping server version < 8.0")
-		}
-
-		document := bson.D{{"largeField", strings.Repeat("a", 16777216-100)}} // Adjust size to account for BSON overhead
-		writes := []ClientBulkWrite{
-			{"db", "x", NewClientInsertOneModel().SetDocument(document)},
-			{"db", "x", NewClientInsertOneModel().SetDocument(document)},
-			{"db", "x", NewClientInsertOneModel().SetDocument(document)},
-		}
-
-		_, err = client.BulkWrite(context.Background(), writes)
-		require.NoError(t, err)
-		assert.Equal(t, 2, bulkWrites, "expected %d bulkWrites, got %d", 2, bulkWrites)
-	})
 }

mongo/errors.go

@@ -619,7 +619,7 @@ type ClientBulkWriteException struct {
 	// A top-level error that occurred when attempting to communicate with the server
 	// or execute the bulk write. This value may not be populated if the exception was
 	// thrown due to errors occurring on individual writes.
-	TopLevelError *WriteError
+	WriteError *WriteError
 
 	// The write concern errors that occurred.
 	WriteConcernErrors []WriteConcernError
@@ -636,8 +636,8 @@ type ClientBulkWriteException struct {
 // Error implements the error interface.
 func (bwe ClientBulkWriteException) Error() string {
 	causes := make([]string, 0, 4)
-	if bwe.TopLevelError != nil {
-		causes = append(causes, "top level error: "+bwe.TopLevelError.Error())
+	if bwe.WriteError != nil {
+		causes = append(causes, "top level error: "+bwe.WriteError.Error())
 	}
 	if len(bwe.WriteConcernErrors) > 0 {
 		errs := make([]error, len(bwe.WriteConcernErrors))