Revert "GODRIVER-2337 Add CustomPipeline to ChangeStreamOptions (#880)"

This reverts commit 6cb4f98.

Author: prestonvasquez

mongo/change_stream.go

@@ -69,22 +69,21 @@ type ChangeStream struct {
 	// TryNext. If continued access is required, a copy must be made.
 	Current bson.Raw
 
-	aggregate       *operation.Aggregate
-	pipelineSlice   []bsoncore.Document
-	pipelineOptions map[string]bsoncore.Value
-	cursor          changeStreamCursor
-	cursorOptions   driver.CursorOptions
-	batch           []bsoncore.Document
-	resumeToken     bson.Raw
-	err             error
-	sess            *session.Client
-	client          *Client
-	registry        *bsoncodec.Registry
-	streamType      StreamType
-	options         *options.ChangeStreamOptions
-	selector        description.ServerSelector
-	operationTime   *primitive.Timestamp
-	wireVersion     *description.VersionRange
+	aggregate     *operation.Aggregate
+	pipelineSlice []bsoncore.Document
+	cursor        changeStreamCursor
+	cursorOptions driver.CursorOptions
+	batch         []bsoncore.Document
+	resumeToken   bson.Raw
+	err           error
+	sess          *session.Client
+	client        *Client
+	registry      *bsoncodec.Registry
+	streamType    StreamType
+	options       *options.ChangeStreamOptions
+	selector      description.ServerSelector
+	operationTime *primitive.Timestamp
+	wireVersion   *description.VersionRange
 }
 
 type changeStreamConfig struct {
@@ -144,11 +143,11 @@ func newChangeStream(ctx context.Context, config changeStreamConfig, pipeline in
 	if cs.options.MaxAwaitTime != nil {
 		cs.cursorOptions.MaxTimeMS = int64(*cs.options.MaxAwaitTime / time.Millisecond)
 	}
-	if cs.options.Custom != nil {
+	if cs.options.CustomOptions != nil {
 		// Marshal all custom options before passing to the initial aggregate. Return
 		// any errors from Marshaling.
 		customOptions := make(map[string]bsoncore.Value)
-		for optionName, optionValue := range cs.options.Custom {
+		for optionName, optionValue := range cs.options.CustomOptions {
 			bsonType, bsonData, err := bson.MarshalValueWithRegistry(cs.registry, optionValue)
 			if err != nil {
 				cs.err = err
@@ -160,21 +159,6 @@ func newChangeStream(ctx context.Context, config changeStreamConfig, pipeline in
 		}
 		cs.aggregate.CustomOptions(customOptions)
 	}
-	if cs.options.CustomPipeline != nil {
-		// Marshal all custom pipeline options before building pipeline slice. Return
-		// any errors from Marshaling.
-		cs.pipelineOptions = make(map[string]bsoncore.Value)
-		for optionName, optionValue := range cs.options.CustomPipeline {
-			bsonType, bsonData, err := bson.MarshalValueWithRegistry(cs.registry, optionValue)
-			if err != nil {
-				cs.err = err
-				closeImplicitSession(cs.sess)
-				return nil, cs.Err()
-			}
-			optionValueBSON := bsoncore.Value{Type: bsonType, Data: bsonData}
-			cs.pipelineOptions[optionName] = optionValueBSON
-		}
-	}
 
 	switch cs.streamType {
 	case ClientStream:
@@ -422,11 +406,6 @@ func (cs *ChangeStream) createPipelineOptionsDoc() bsoncore.Document {
 		plDoc = bsoncore.AppendTimestampElement(plDoc, "startAtOperationTime", cs.options.StartAtOperationTime.T, cs.options.StartAtOperationTime.I)
 	}
 
-	// Append custom pipeline options.
-	for optionName, optionValue := range cs.pipelineOptions {
-		plDoc = bsoncore.AppendValueElement(plDoc, optionName, optionValue)
-	}
-
 	if plDoc, cs.err = bsoncore.AppendDocumentEnd(plDoc, plDocIdx); cs.err != nil {
 		return nil
 	}

mongo/collection.go

@@ -852,11 +852,11 @@ func aggregate(a aggregateParams) (cur *Cursor, err error) {
 		}
 		op.Let(let)
 	}
-	if ao.Custom != nil {
+	if ao.CustomOptions != nil {
 		// Marshal all custom options before passing to the aggregate operation. Return
 		// any errors from Marshaling.
 		customOptions := make(map[string]bsoncore.Value)
-		for optionName, optionValue := range ao.Custom {
+		for optionName, optionValue := range ao.CustomOptions {
 			bsonType, bsonData, err := bson.MarshalValueWithRegistry(a.registry, optionValue)
 			if err != nil {
 				return nil, err

mongo/integration/change_stream_test.go

@@ -634,11 +634,11 @@ func TestChangeStream_ReplicaSet(t *testing.T) {
 		evt := mt.GetStartedEvent()
 		assert.Equal(mt, "killCursors", evt.CommandName, "expected command 'killCursors', got %q", evt.CommandName)
 	})
-	mt.Run("Custom", func(mt *mtest.T) {
+	mt.Run("CustomOptions", func(mt *mtest.T) {
 		// Custom options should be a BSON map of option names to Marshalable option values.
 		// We use "allowDiskUse" as an example.
 		customOpts := bson.M{"allowDiskUse": true}
-		opts := options.ChangeStream().SetCustom(customOpts)
+		opts := options.ChangeStream().SetCustomOptions(customOpts)
 
 		// Create change stream with custom options set.
 		mt.ClearEvents()
@@ -656,28 +656,6 @@ func TestChangeStream_ReplicaSet(t *testing.T) {
 		assert.True(mt, ok, "expected field 'allowDiskUse' to be boolean, got %v", aduVal.Type.String())
 		assert.True(mt, adu, "expected field 'allowDiskUse' to be true, got false")
 	})
-	mt.RunOpts("CustomPipeline", mtest.NewOptions().MinServerVersion("4.0"), func(mt *mtest.T) {
-		// Custom pipeline options should be a BSON map of option names to Marshalable option values.
-		// We use "allChangesForCluster" as an example.
-		customPipelineOpts := bson.M{"allChangesForCluster": false}
-		opts := options.ChangeStream().SetCustomPipeline(customPipelineOpts)
-
-		// Create change stream with custom pipeline options set.
-		mt.ClearEvents()
-		cs, err := mt.Coll.Watch(context.Background(), mongo.Pipeline{}, opts)
-		assert.Nil(mt, err, "Watch error: %v", err)
-		defer closeStream(cs)
-
-		// Assert that custom pipeline option is included in the $changeStream stage.
-		evt := mt.GetStartedEvent()
-		assert.Equal(mt, "aggregate", evt.CommandName, "expected command 'aggregate' got, %q", evt.CommandName)
-
-		acfcVal, err := evt.Command.LookupErr("pipeline", "0", "$changeStream", "allChangesForCluster")
-		assert.Nil(mt, err, "expected field 'allChangesForCluster' in $changeStream stage not found")
-		acfc, ok := acfcVal.BooleanOK()
-		assert.True(mt, ok, "expected field 'allChangesForCluster' to be boolean, got %v", acfcVal.Type.String())
-		assert.False(mt, acfc, "expected field 'allChangesForCluster' to be false, got %v", acfc)
-	})
 }
 
 func closeStream(cs *mongo.ChangeStream) {

mongo/integration/collection_test.go

@@ -803,11 +803,11 @@ func TestCollection(t *testing.T) {
 				return mt.Coll.Aggregate(context.Background(), mongo.Pipeline{}, options.Aggregate().SetBatchSize(3))
 			})
 		})
-		mt.Run("Custom", func(mt *mtest.T) {
+		mt.Run("CustomOptions", func(mt *mtest.T) {
 			// Custom options should be a BSON map of option names to Marshalable option values.
 			// We use "allowDiskUse" as an example.
 			customOpts := bson.M{"allowDiskUse": true}
-			opts := options.Aggregate().SetCustom(customOpts)
+			opts := options.Aggregate().SetCustomOptions(customOpts)
 
 			// Run aggregate with custom options set.
 			mt.ClearEvents()

mongo/options/aggregateoptions.go

@@ -58,7 +58,7 @@ type AggregateOptions struct {
 	// Custom options to be added to aggregate expression. Key-value pairs of the BSON map should correlate with desired
 	// option names and values. Values must be Marshalable. Custom options may conflict with non-custom options, and custom
 	// options bypass client-side validation. Prefer using non-custom options where possible.
-	Custom bson.M
+	CustomOptions bson.M
 }
 
 // Aggregate creates a new AggregateOptions instance.
@@ -120,12 +120,12 @@ func (ao *AggregateOptions) SetLet(let interface{}) *AggregateOptions {
 	return ao
 }
 
-// SetCustom sets the value for the Custom field. Key-value pairs of the BSON map should correlate
-// with desired option names and values. Values must be Marshalable. Custom options may conflict
-// with non-custom options, and custom options bypass client-side validation. Prefer using non-custom
-// options where possible.
-func (ao *AggregateOptions) SetCustom(c bson.M) *AggregateOptions {
-	ao.Custom = c
+// SetCustomOptions sets the value for the CustomOptions field. Key-value pairs of the BSON map
+// should correlate with desired option names and values. Values must be Marshalable. Custom options
+// may conflict with non-custom options, and custom options bypass client-side validation. Prefer
+// using non-custom options where possible.
+func (ao *AggregateOptions) SetCustomOptions(co bson.M) *AggregateOptions {
+	ao.CustomOptions = co
 	return ao
 }
 
@@ -164,8 +164,8 @@ func MergeAggregateOptions(opts ...*AggregateOptions) *AggregateOptions {
 		if ao.Let != nil {
 			aggOpts.Let = ao.Let
 		}
-		if ao.Custom != nil {
-			aggOpts.Custom = ao.Custom
+		if ao.CustomOptions != nil {
+			aggOpts.CustomOptions = ao.CustomOptions
 		}
 	}
 

mongo/options/changestreamoptions.go

@@ -51,12 +51,7 @@ type ChangeStreamOptions struct {
 	// Custom options to be added to the initial aggregate for the change stream. Key-value pairs of the BSON map should
 	// correlate with desired option names and values. Values must be Marshalable. Custom options may conflict with
 	// non-custom options, and custom options bypass client-side validation. Prefer using non-custom options where possible.
-	Custom bson.M
-
-	// Custom options to be added to the $changeStream stage in the initial aggregate. Key-value pairs of the BSON map should
-	// correlate with desired option names and values. Values must be Marshalable. Custom pipeline options bypass client-side
-	// validation. Prefer using non-custom options where possible.
-	CustomPipeline bson.M
+	CustomOptions bson.M
 }
 
 // ChangeStream creates a new ChangeStreamOptions instance.
@@ -108,20 +103,12 @@ func (cso *ChangeStreamOptions) SetStartAfter(sa interface{}) *ChangeStreamOptio
 	return cso
 }
 
-// SetCustom sets the value for the Custom field. Key-value pairs of the BSON map should correlate
-// with desired option names and values. Values must be Marshalable. Custom options may conflict
-// with non-custom options, and custom options bypass client-side validation. Prefer using non-custom
-// options where possible.
-func (cso *ChangeStreamOptions) SetCustom(c bson.M) *ChangeStreamOptions {
-	cso.Custom = c
-	return cso
-}
-
-// SetCustomPipeline sets the value for the CustomPipeline field. Key-value pairs of the BSON map
-// should correlate with desired option names and values. Values must be Marshalable. Custom pipeline
-// options bypass client-side validation. Prefer using non-custom options where possible.
-func (cso *ChangeStreamOptions) SetCustomPipeline(cp bson.M) *ChangeStreamOptions {
-	cso.CustomPipeline = cp
+// SetCustomOptions sets the value for the CustomOptions field. Key-value pairs of the BSON map
+// should correlate with desired option names and values. Values must be Marshalable. Custom options
+// may conflict with non-custom options, and custom options bypass client-side validation. Prefer
+// using non-custom options where possible.
+func (cso *ChangeStreamOptions) SetCustomOptions(co bson.M) *ChangeStreamOptions {
+	cso.CustomOptions = co
 	return cso
 }
 
@@ -154,11 +141,8 @@ func MergeChangeStreamOptions(opts ...*ChangeStreamOptions) *ChangeStreamOptions
 		if cso.StartAfter != nil {
 			csOpts.StartAfter = cso.StartAfter
 		}
-		if cso.Custom != nil {
-			csOpts.Custom = cso.Custom
-		}
-		if cso.CustomPipeline != nil {
-			csOpts.CustomPipeline = cso.CustomPipeline
+		if cso.CustomOptions != nil {
+			csOpts.CustomOptions = cso.CustomOptions
 		}
 	}