diff --git a/mongo/change_stream.go b/mongo/change_stream.go index be8b71ccaa..ab5590d69f 100644 --- a/mongo/change_stream.go +++ b/mongo/change_stream.go @@ -24,7 +24,6 @@ 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/description" - "go.mongodb.org/mongo-driver/v2/x/mongo/driver/operation" "go.mongodb.org/mongo-driver/v2/x/mongo/driver/session" ) @@ -70,7 +69,7 @@ type ChangeStream struct { // TryNext. If continued access is required, a copy must be made. Current bson.Raw - aggregate *operation.Aggregate + aggregate *aggregateOp pipelineSlice []bsoncore.Document pipelineOptions map[string]bsoncore.Value cursor changeStreamCursor @@ -141,17 +140,26 @@ func newChangeStream(ctx context.Context, config changeStreamConfig, pipeline an return nil, cs.Err() } - cs.aggregate = operation.NewAggregate(nil). - ReadPreference(config.readPreference).ReadConcern(config.readConcern). - Deployment(cs.client.deployment).ClusterClock(cs.client.clock). - CommandMonitor(cs.client.monitor).Session(cs.sess).ServerSelector(cs.selector). - Retry(driver.RetryNone).MaxAdaptiveRetries(cursorOpts.MaxAdaptiveRetries). - EnableOverloadRetargeting(cursorOpts.EnableOverloadRetargeting). - ServerAPI(cs.client.serverAPI).Crypt(config.crypt).Timeout(cs.client.timeout). - Authenticator(cs.client.authenticator) + retry := driver.RetryNone + cs.aggregate = &aggregateOp{ + readPreference: config.readPreference, + readConcern: config.readConcern, + deployment: cs.client.deployment, + clock: cs.client.clock, + monitor: cs.client.monitor, + session: cs.sess, + selector: cs.selector, + retry: &retry, + maxAdaptiveRetries: cursorOpts.MaxAdaptiveRetries, + enableOverloadRetargeting: cursorOpts.EnableOverloadRetargeting, + serverAPI: cs.client.serverAPI, + crypt: config.crypt, + timeout: cs.client.timeout, + authenticator: cs.client.authenticator, + } if cs.options.Collation != nil { - cs.aggregate.Collation(bsoncore.Document(toDocument(cs.options.Collation))) + cs.aggregate.collation = bsoncore.Document(toDocument(cs.options.Collation)) } if cs.options.Comment != nil { comment, err := marshalValue(cs.options.Comment, cs.bsonOpts, cs.registry) @@ -159,11 +167,11 @@ func newChangeStream(ctx context.Context, config changeStreamConfig, pipeline an return nil, err } - cs.aggregate.Comment(comment) + cs.aggregate.comment = comment cs.cursorOptions.Comment = comment } if cs.options.BatchSize != nil { - cs.aggregate.BatchSize(*cs.options.BatchSize) + cs.aggregate.batchSize = cs.options.BatchSize cs.cursorOptions.BatchSize = *cs.options.BatchSize } if cs.options.MaxAwaitTime != nil { @@ -182,7 +190,7 @@ func newChangeStream(ctx context.Context, config changeStreamConfig, pipeline an } customOptions[optionName] = optionValueBSON } - cs.aggregate.CustomOptions(customOptions) + cs.aggregate.customOptions = customOptions } if cs.options.CustomPipeline != nil { // Marshal all custom pipeline options before building pipeline slice. Return @@ -201,11 +209,12 @@ func newChangeStream(ctx context.Context, config changeStreamConfig, pipeline an switch cs.streamType { case ClientStream: - cs.aggregate.Database("admin") + cs.aggregate.database = "admin" case DatabaseStream: - cs.aggregate.Database(config.databaseName) + cs.aggregate.database = config.databaseName case CollectionStream: - cs.aggregate.Collection(config.collectionName).Database(config.databaseName) + cs.aggregate.collection = config.collectionName + cs.aggregate.database = config.databaseName default: closeImplicitSession(cs.sess) return nil, fmt.Errorf("must supply a valid StreamType in config, instead of %v", cs.streamType) @@ -232,7 +241,7 @@ func newChangeStream(ctx context.Context, config changeStreamConfig, pipeline an } var pipelineArr bsoncore.Document pipelineArr, cs.err = cs.pipelineToBSON() - cs.aggregate.Pipeline(pipelineArr) + cs.aggregate.pipeline = pipelineArr if cs.err = cs.executeOperation(ctx, false); cs.err != nil { closeImplicitSession(cs.sess) @@ -298,7 +307,7 @@ func (cs *ChangeStream) executeOperation(ctx context.Context, resuming bool) err _ = deployment.close() }() - cs.aggregate.Deployment(deployment) + cs.aggregate.deployment = deployment if resuming { cs.replaceOptions(cs.wireVersion) @@ -318,7 +327,7 @@ func (cs *ChangeStream) executeOperation(ctx context.Context, resuming bool) err if plArr, cs.err = cs.pipelineToBSON(); cs.err != nil { return cs.Err() } - cs.aggregate.Pipeline(plArr) + cs.aggregate.pipeline = plArr } // Execute the aggregate, retrying on retryable errors once (1) if retryable reads are enabled and diff --git a/mongo/collection.go b/mongo/collection.go index 3f1df6dbdf..516809e2cb 100644 --- a/mongo/collection.go +++ b/mongo/collection.go @@ -1044,44 +1044,46 @@ func aggregate(a aggregateParams, opts ...options.Lister[options.AggregateOption cursorOpts.MarshalValueEncoderFn = newEncoderFn(a.bsonOpts, a.registry) - op := operation.NewAggregate(pipelineArr). - Session(sess). - WriteConcern(wc). - ReadConcern(rc). - ReadPreference(a.readPreference). - CommandMonitor(a.client.monitor). - Retry(retry). - MaxAdaptiveRetries(cursorOpts.MaxAdaptiveRetries). - EnableOverloadRetargeting(cursorOpts.EnableOverloadRetargeting). - ServerSelector(selector). - ClusterClock(a.client.clock). - Database(a.db). - Collection(a.col). - Deployment(a.client.deployment). - Crypt(a.client.cryptFLE). - ServerAPI(a.client.serverAPI). - HasOutputStage(hasOutputStage). - Timeout(a.client.timeout). - Authenticator(a.client.authenticator). + op := aggregateOp{ + pipeline: pipelineArr, + session: sess, + writeConcern: wc, + readConcern: rc, + readPreference: a.readPreference, + monitor: a.client.monitor, + retry: &retry, + maxAdaptiveRetries: cursorOpts.MaxAdaptiveRetries, + enableOverloadRetargeting: cursorOpts.EnableOverloadRetargeting, + selector: selector, + clock: a.client.clock, + database: a.db, + collection: a.col, + deployment: a.client.deployment, + crypt: a.client.cryptFLE, + serverAPI: a.client.serverAPI, + hasOutputStage: hasOutputStage, + timeout: a.client.timeout, + authenticator: a.client.authenticator, // Omit "maxTimeMS" from operations that return a user-managed cursor to // prevent confusing "cursor not found" errors. // // See DRIVERS-2722 for more detail. - OmitMaxTimeMS(true) + omitMaxTimeMS: true, + } if args.AllowDiskUse != nil { - op.AllowDiskUse(*args.AllowDiskUse) + op.allowDiskUse = args.AllowDiskUse } // ignore batchSize of 0 with $out if args.BatchSize != nil && (*args.BatchSize != 0 || !hasOutputStage) { - op.BatchSize(*args.BatchSize) + op.batchSize = args.BatchSize cursorOpts.BatchSize = *args.BatchSize } if args.BypassDocumentValidation != nil && *args.BypassDocumentValidation { - op.BypassDocumentValidation(*args.BypassDocumentValidation) + op.bypassDocumentValidation = args.BypassDocumentValidation } if args.Collation != nil { - op.Collation(bsoncore.Document(toDocument(args.Collation))) + op.collation = bsoncore.Document(toDocument(args.Collation)) } if args.MaxAwaitTime != nil { cursorOpts.SetMaxAwaitTime(*args.MaxAwaitTime) @@ -1092,7 +1094,7 @@ func aggregate(a aggregateParams, opts ...options.Lister[options.AggregateOption return nil, err } - op.Comment(comment) + op.comment = comment cursorOpts.Comment = comment } if args.Hint != nil { @@ -1103,14 +1105,14 @@ func aggregate(a aggregateParams, opts ...options.Lister[options.AggregateOption if err != nil { return nil, err } - op.Hint(hintVal) + op.hint = hintVal } if args.Let != nil { let, err := marshal(args.Let, a.bsonOpts, a.registry) if err != nil { return nil, err } - op.Let(let) + op.let = let } if args.Custom != nil { // Marshal all custom options before passing to the aggregate operation. Return @@ -1123,10 +1125,10 @@ func aggregate(a aggregateParams, opts ...options.Lister[options.AggregateOption } customOptions[optionName] = optionValueBSON } - op.CustomOptions(customOptions) + op.customOptions = customOptions } if rawData, ok := optionsutil.Value(args.Internal, "rawData").(bool); ok { - op = op.RawData(rawData) + op.rawData = &rawData } err = op.Execute(a.ctx) @@ -1199,14 +1201,27 @@ func (coll *Collection) CountDocuments(ctx context.Context, filter any, maxAdaptiveRetries := coll.client.effectiveAdaptiveRetries(coll.client.retryReads) selector := makeReadPrefSelector(sess, coll.readSelector, coll.client.localThreshold) - op := operation.NewAggregate(pipelineArr).Session(sess).ReadConcern(rc).ReadPreference(coll.readPreference). - Retry(retry).MaxAdaptiveRetries(maxAdaptiveRetries). - EnableOverloadRetargeting(coll.client.enableOverloadRetargeting). - CommandMonitor(coll.client.monitor).ServerSelector(selector).ClusterClock(coll.client.clock).Database(coll.db.name). - Collection(coll.name).Deployment(coll.client.deployment).Crypt(coll.client.cryptFLE).ServerAPI(coll.client.serverAPI). - Timeout(coll.client.timeout).Authenticator(coll.client.authenticator) + op := aggregateOp{ + pipeline: pipelineArr, + session: sess, + readConcern: rc, + readPreference: coll.readPreference, + retry: &retry, + maxAdaptiveRetries: maxAdaptiveRetries, + enableOverloadRetargeting: coll.client.enableOverloadRetargeting, + monitor: coll.client.monitor, + selector: selector, + clock: coll.client.clock, + database: coll.db.name, + collection: coll.name, + deployment: coll.client.deployment, + crypt: coll.client.cryptFLE, + serverAPI: coll.client.serverAPI, + timeout: coll.client.timeout, + authenticator: coll.client.authenticator, + } if args.Collation != nil { - op.Collation(bsoncore.Document(toDocument(args.Collation))) + op.collation = bsoncore.Document(toDocument(args.Collation)) } if args.Comment != nil { comment, err := marshalValue(args.Comment, coll.bsonOpts, coll.registry) @@ -1214,7 +1229,7 @@ func (coll *Collection) CountDocuments(ctx context.Context, filter any, return 0, err } - op.Comment(comment) + op.comment = comment } if args.Hint != nil { if isUnorderedMap(args.Hint) { @@ -1224,10 +1239,10 @@ func (coll *Collection) CountDocuments(ctx context.Context, filter any, if err != nil { return 0, err } - op.Hint(hintVal) + op.hint = hintVal } if rawData, ok := optionsutil.Value(args.Internal, "rawData").(bool); ok { - op = op.RawData(rawData) + op.rawData = &rawData } err = op.Execute(ctx) diff --git a/mongo/op_aggregate.go b/mongo/op_aggregate.go new file mode 100644 index 0000000000..750262336b --- /dev/null +++ b/mongo/op_aggregate.go @@ -0,0 +1,164 @@ +// Copyright (C) MongoDB, Inc. 2019-present. +// +// Licensed under the Apache License, Version 2.0 (the "License"); you may +// not use this file except in compliance with the License. You may obtain +// a copy of the License at http://www.apache.org/licenses/LICENSE-2.0 + +package mongo + +import ( + "context" + "errors" + "time" + + "go.mongodb.org/mongo-driver/v2/event" + "go.mongodb.org/mongo-driver/v2/internal/driverutil" + "go.mongodb.org/mongo-driver/v2/mongo/readconcern" + "go.mongodb.org/mongo-driver/v2/mongo/readpref" + "go.mongodb.org/mongo-driver/v2/mongo/writeconcern" + "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/description" + "go.mongodb.org/mongo-driver/v2/x/mongo/driver/session" +) + +// aggregateOp represents an aggregate operation. +type aggregateOp struct { + authenticator driver.Authenticator + allowDiskUse *bool + batchSize *int32 + bypassDocumentValidation *bool + collation bsoncore.Document + comment bsoncore.Value + hint bsoncore.Value + pipeline bsoncore.Document + session *session.Client + clock *session.ClusterClock + collection string + monitor *event.CommandMonitor + database string + deployment driver.Deployment + readConcern *readconcern.ReadConcern + readPreference *readpref.ReadPref + retry *driver.RetryMode + maxAdaptiveRetries uint + enableOverloadRetargeting bool + selector description.ServerSelector + writeConcern *writeconcern.WriteConcern + crypt driver.Crypt + serverAPI *driver.ServerAPIOptions + let bsoncore.Document + hasOutputStage bool + customOptions map[string]bsoncore.Value + timeout *time.Duration + omitMaxTimeMS bool + rawData *bool + + result driver.CursorResponse +} + +// Result returns the result of executing this operation. +func (a *aggregateOp) Result(opts driver.CursorOptions) (*driver.BatchCursor, error) { + clientSession := a.session + + clock := a.clock + opts.ServerAPI = a.serverAPI + return driver.NewBatchCursor(a.result, clientSession, clock, opts) +} + +// ResultCursorResponse returns the underlying CursorResponse result of executing this +// operation. +func (a *aggregateOp) ResultCursorResponse() driver.CursorResponse { + return a.result +} + +func (a *aggregateOp) processResponse(_ context.Context, resp bsoncore.Document, info driver.ResponseInfo) error { + curDoc, err := driver.ExtractCursorDocument(resp) + if err != nil { + return err + } + a.result, err = driver.NewCursorResponse(curDoc, info) + return err +} + +// Execute runs this operation and returns an error if the operation did not execute successfully. +func (a *aggregateOp) Execute(ctx context.Context) error { + if a.deployment == nil { + return errors.New("the Aggregate operation must have a Deployment set before Execute can be called") + } + + return driver.Operation{ + CommandFn: a.command, + ProcessResponseFn: a.processResponse, + + Client: a.session, + Clock: a.clock, + CommandMonitor: a.monitor, + Database: a.database, + Deployment: a.deployment, + ReadConcern: a.readConcern, + ReadPreference: a.readPreference, + Type: driver.Read, + RetryMode: a.retry, + MaxAdaptiveRetries: a.maxAdaptiveRetries, + EnableOverloadRetargeting: a.enableOverloadRetargeting, + Selector: a.selector, + WriteConcern: a.writeConcern, + Crypt: a.crypt, + MinimumWriteConcernWireVersion: 5, + ServerAPI: a.serverAPI, + IsOutputAggregate: a.hasOutputStage, + Timeout: a.timeout, + Name: driverutil.AggregateOp, + Authenticator: a.authenticator, + OmitMaxTimeMS: a.omitMaxTimeMS, + }.Execute(ctx) +} + +func (a *aggregateOp) command(dst []byte, desc description.SelectedServer) ([]byte, error) { + header := bsoncore.Value{Type: bsoncore.TypeString, Data: bsoncore.AppendString(nil, a.collection)} + if a.collection == "" { + header = bsoncore.Value{Type: bsoncore.TypeInt32, Data: []byte{0x01, 0x00, 0x00, 0x00}} + } + dst = bsoncore.AppendValueElement(dst, "aggregate", header) + + cursorIdx, cursorDoc := bsoncore.AppendDocumentStart(nil) + if a.allowDiskUse != nil { + dst = bsoncore.AppendBooleanElement(dst, "allowDiskUse", *a.allowDiskUse) + } + if a.batchSize != nil { + cursorDoc = bsoncore.AppendInt32Element(cursorDoc, "batchSize", *a.batchSize) + } + if a.bypassDocumentValidation != nil { + dst = bsoncore.AppendBooleanElement(dst, "bypassDocumentValidation", *a.bypassDocumentValidation) + } + if a.collation != nil { + if desc.WireVersion == nil || !driverutil.VersionRangeIncludes(*desc.WireVersion, 5) { + return nil, errors.New("the 'collation' command parameter requires a minimum server wire version of 5") + } + dst = bsoncore.AppendDocumentElement(dst, "collation", a.collation) + } + if a.comment.Type != bsoncore.Type(0) { + dst = bsoncore.AppendValueElement(dst, "comment", a.comment) + } + if a.hint.Type != bsoncore.Type(0) { + dst = bsoncore.AppendValueElement(dst, "hint", a.hint) + } + if a.pipeline != nil { + dst = bsoncore.AppendArrayElement(dst, "pipeline", a.pipeline) + } + if a.let != nil { + dst = bsoncore.AppendDocumentElement(dst, "let", a.let) + } + // Set rawData for 8.2+ servers. + if a.rawData != nil && desc.WireVersion != nil && driverutil.VersionRangeIncludes(*desc.WireVersion, 27) { + dst = bsoncore.AppendBooleanElement(dst, "rawData", *a.rawData) + } + for optionName, optionValue := range a.customOptions { + dst = bsoncore.AppendValueElement(dst, optionName, optionValue) + } + cursorDoc, _ = bsoncore.AppendDocumentEnd(cursorDoc, cursorIdx) + dst = bsoncore.AppendDocumentElement(dst, "cursor", cursorDoc) + + return dst, nil +} diff --git a/x/mongo/driver/integration/aggregate_test.go b/x/mongo/driver/integration/aggregate_test.go deleted file mode 100644 index 21367bfe20..0000000000 --- a/x/mongo/driver/integration/aggregate_test.go +++ /dev/null @@ -1,94 +0,0 @@ -// Copyright (C) MongoDB, Inc. 2022-present. -// -// Licensed under the Apache License, Version 2.0 (the "License"); you may -// not use this file except in compliance with the License. You may obtain -// a copy of the License at http://www.apache.org/licenses/LICENSE-2.0 - -package integration - -import ( - "bytes" - "context" - "testing" - - "go.mongodb.org/mongo-driver/v2/bson" - "go.mongodb.org/mongo-driver/v2/internal/integtest" - "go.mongodb.org/mongo-driver/v2/internal/serverselector" - "go.mongodb.org/mongo-driver/v2/mongo/writeconcern" - "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/operation" -) - -func TestAggregate(t *testing.T) { - if testing.Short() { - t.Skip("skipping integration test in short mode") - } - - t.Run("Multiple Batches", func(t *testing.T) { - ds := []bsoncore.Document{ - bsoncore.BuildDocument(nil, bsoncore.AppendInt32Element(nil, "_id", 1)), - bsoncore.BuildDocument(nil, bsoncore.AppendInt32Element(nil, "_id", 2)), - bsoncore.BuildDocument(nil, bsoncore.AppendInt32Element(nil, "_id", 3)), - bsoncore.BuildDocument(nil, bsoncore.AppendInt32Element(nil, "_id", 4)), - bsoncore.BuildDocument(nil, bsoncore.AppendInt32Element(nil, "_id", 5)), - } - wc := writeconcern.Majority() - autoInsertDocs(t, wc, ds...) - - op := operation.NewAggregate(bsoncore.BuildArray(nil, - bsoncore.BuildDocumentValue( - bsoncore.BuildDocumentElement(nil, - "$match", bsoncore.BuildDocumentElement(nil, - "_id", bsoncore.AppendInt32Element(nil, "$gt", 2), - ), - ), - ), - bsoncore.BuildDocumentValue( - bsoncore.BuildDocumentElement(nil, - "$sort", bsoncore.AppendInt32Element(nil, "_id", 1), - ), - ), - )).Collection(integtest.ColName(t)).Database(dbName).Deployment(integtest.Topology(t)). - ServerSelector(&serverselector.Write{}).BatchSize(2) - err := op.Execute(context.Background()) - noerr(t, err) - cursor, err := op.Result(driver.CursorOptions{BatchSize: 2}) - noerr(t, err) - - var got []bsoncore.Document - for i := 0; i < 2; i++ { - if !cursor.Next(context.Background()) { - t.Error("Cursor should have results, but does not have a next result") - } - docs, err := cursor.Batch().Documents() - noerr(t, err) - got = append(got, docs...) - } - readers := ds[2:] - for i, g := range got { - if !bytes.Equal(g[:len(readers[i])], readers[i]) { - t.Errorf("Did not get expected document. got %v; want %v", bson.Raw(g[:len(readers[i])]), readers[i]) - } - } - - if cursor.Next(context.Background()) { - t.Error("Cursor should be exhausted but has more results") - } - }) - t.Run("AllowDiskUse", func(t *testing.T) { - ds := []bsoncore.Document{ - bsoncore.BuildDocument(nil, bsoncore.AppendInt32Element(nil, "_id", 1)), - bsoncore.BuildDocument(nil, bsoncore.AppendInt32Element(nil, "_id", 2)), - } - wc := writeconcern.Majority() - autoInsertDocs(t, wc, ds...) - - op := operation.NewAggregate(bsoncore.BuildArray(nil)).Collection(integtest.ColName(t)).Database(dbName). - Deployment(integtest.Topology(t)).ServerSelector(&serverselector.Write{}).AllowDiskUse(true) - err := op.Execute(context.Background()) - if err != nil { - t.Errorf("Expected no error from allowing disk use, but got %v", err) - } - }) -} diff --git a/x/mongo/driver/integration/main_test.go b/x/mongo/driver/integration/main_test.go index 7ca6090e98..0fed3b95a1 100644 --- a/x/mongo/driver/integration/main_test.go +++ b/x/mongo/driver/integration/main_test.go @@ -142,11 +142,6 @@ func dropCollection(t *testing.T, dbname, colname string) { } } -// autoInsertDocs inserts the docs into the test cluster. -func autoInsertDocs(t *testing.T, writeConcern *writeconcern.WriteConcern, docs ...bsoncore.Document) { - insertDocs(t, integtest.DBName(t), integtest.ColName(t), writeConcern, docs...) -} - // insertDocs inserts the docs into the test cluster. func insertDocs(t *testing.T, dbname, colname string, writeConcern *writeconcern.WriteConcern, docs ...bsoncore.Document) { t.Helper() diff --git a/x/mongo/driver/operation/aggregate.go b/x/mongo/driver/operation/aggregate.go deleted file mode 100644 index 22dbb8087f..0000000000 --- a/x/mongo/driver/operation/aggregate.go +++ /dev/null @@ -1,467 +0,0 @@ -// Copyright (C) MongoDB, Inc. 2019-present. -// -// Licensed under the Apache License, Version 2.0 (the "License"); you may -// not use this file except in compliance with the License. You may obtain -// a copy of the License at http://www.apache.org/licenses/LICENSE-2.0 - -package operation - -import ( - "context" - "errors" - "time" - - "go.mongodb.org/mongo-driver/v2/event" - "go.mongodb.org/mongo-driver/v2/internal/driverutil" - "go.mongodb.org/mongo-driver/v2/mongo/readconcern" - "go.mongodb.org/mongo-driver/v2/mongo/readpref" - "go.mongodb.org/mongo-driver/v2/mongo/writeconcern" - "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/description" - "go.mongodb.org/mongo-driver/v2/x/mongo/driver/session" -) - -// Aggregate represents an aggregate operation. -type Aggregate struct { - authenticator driver.Authenticator - allowDiskUse *bool - batchSize *int32 - bypassDocumentValidation *bool - collation bsoncore.Document - comment bsoncore.Value - hint bsoncore.Value - pipeline bsoncore.Document - session *session.Client - clock *session.ClusterClock - collection string - monitor *event.CommandMonitor - database string - deployment driver.Deployment - readConcern *readconcern.ReadConcern - readPreference *readpref.ReadPref - retry *driver.RetryMode - maxAdaptiveRetries uint - enableOverloadRetargeting bool - selector description.ServerSelector - writeConcern *writeconcern.WriteConcern - crypt driver.Crypt - serverAPI *driver.ServerAPIOptions - let bsoncore.Document - hasOutputStage bool - customOptions map[string]bsoncore.Value - timeout *time.Duration - omitMaxTimeMS bool - rawData *bool - - result driver.CursorResponse -} - -// NewAggregate constructs and returns a new Aggregate. -func NewAggregate(pipeline bsoncore.Document) *Aggregate { - return &Aggregate{ - pipeline: pipeline, - } -} - -// Result returns the result of executing this operation. -func (a *Aggregate) Result(opts driver.CursorOptions) (*driver.BatchCursor, error) { - clientSession := a.session - - clock := a.clock - opts.ServerAPI = a.serverAPI - return driver.NewBatchCursor(a.result, clientSession, clock, opts) -} - -// ResultCursorResponse returns the underlying CursorResponse result of executing this -// operation. -func (a *Aggregate) ResultCursorResponse() driver.CursorResponse { - return a.result -} - -func (a *Aggregate) processResponse(_ context.Context, resp bsoncore.Document, info driver.ResponseInfo) error { - curDoc, err := driver.ExtractCursorDocument(resp) - if err != nil { - return err - } - a.result, err = driver.NewCursorResponse(curDoc, info) - return err -} - -// Execute runs this operations and returns an error if the operation did not execute successfully. -func (a *Aggregate) Execute(ctx context.Context) error { - if a.deployment == nil { - return errors.New("the Aggregate operation must have a Deployment set before Execute can be called") - } - - return driver.Operation{ - CommandFn: a.command, - ProcessResponseFn: a.processResponse, - - Client: a.session, - Clock: a.clock, - CommandMonitor: a.monitor, - Database: a.database, - Deployment: a.deployment, - ReadConcern: a.readConcern, - ReadPreference: a.readPreference, - Type: driver.Read, - RetryMode: a.retry, - MaxAdaptiveRetries: a.maxAdaptiveRetries, - EnableOverloadRetargeting: a.enableOverloadRetargeting, - Selector: a.selector, - WriteConcern: a.writeConcern, - Crypt: a.crypt, - MinimumWriteConcernWireVersion: 5, - ServerAPI: a.serverAPI, - IsOutputAggregate: a.hasOutputStage, - Timeout: a.timeout, - Name: driverutil.AggregateOp, - Authenticator: a.authenticator, - OmitMaxTimeMS: a.omitMaxTimeMS, - }.Execute(ctx) -} - -func (a *Aggregate) command(dst []byte, desc description.SelectedServer) ([]byte, error) { - header := bsoncore.Value{Type: bsoncore.TypeString, Data: bsoncore.AppendString(nil, a.collection)} - if a.collection == "" { - header = bsoncore.Value{Type: bsoncore.TypeInt32, Data: []byte{0x01, 0x00, 0x00, 0x00}} - } - dst = bsoncore.AppendValueElement(dst, "aggregate", header) - - cursorIdx, cursorDoc := bsoncore.AppendDocumentStart(nil) - if a.allowDiskUse != nil { - dst = bsoncore.AppendBooleanElement(dst, "allowDiskUse", *a.allowDiskUse) - } - if a.batchSize != nil { - cursorDoc = bsoncore.AppendInt32Element(cursorDoc, "batchSize", *a.batchSize) - } - if a.bypassDocumentValidation != nil { - dst = bsoncore.AppendBooleanElement(dst, "bypassDocumentValidation", *a.bypassDocumentValidation) - } - if a.collation != nil { - if desc.WireVersion == nil || !driverutil.VersionRangeIncludes(*desc.WireVersion, 5) { - return nil, errors.New("the 'collation' command parameter requires a minimum server wire version of 5") - } - dst = bsoncore.AppendDocumentElement(dst, "collation", a.collation) - } - if a.comment.Type != bsoncore.Type(0) { - dst = bsoncore.AppendValueElement(dst, "comment", a.comment) - } - if a.hint.Type != bsoncore.Type(0) { - dst = bsoncore.AppendValueElement(dst, "hint", a.hint) - } - if a.pipeline != nil { - dst = bsoncore.AppendArrayElement(dst, "pipeline", a.pipeline) - } - if a.let != nil { - dst = bsoncore.AppendDocumentElement(dst, "let", a.let) - } - // Set rawData for 8.2+ servers. - if a.rawData != nil && desc.WireVersion != nil && driverutil.VersionRangeIncludes(*desc.WireVersion, 27) { - dst = bsoncore.AppendBooleanElement(dst, "rawData", *a.rawData) - } - for optionName, optionValue := range a.customOptions { - dst = bsoncore.AppendValueElement(dst, optionName, optionValue) - } - cursorDoc, _ = bsoncore.AppendDocumentEnd(cursorDoc, cursorIdx) - dst = bsoncore.AppendDocumentElement(dst, "cursor", cursorDoc) - - return dst, nil -} - -// AllowDiskUse enables writing to temporary files. When true, aggregation stages can write to the dbPath/_tmp directory. -func (a *Aggregate) AllowDiskUse(allowDiskUse bool) *Aggregate { - if a == nil { - a = new(Aggregate) - } - - a.allowDiskUse = &allowDiskUse - return a -} - -// BatchSize specifies the number of documents to return in every batch. -func (a *Aggregate) BatchSize(batchSize int32) *Aggregate { - if a == nil { - a = new(Aggregate) - } - - a.batchSize = &batchSize - return a -} - -// BypassDocumentValidation allows the write to opt-out of document level validation. This only applies when the $out stage is specified. -func (a *Aggregate) BypassDocumentValidation(bypassDocumentValidation bool) *Aggregate { - if a == nil { - a = new(Aggregate) - } - - a.bypassDocumentValidation = &bypassDocumentValidation - return a -} - -// Collation specifies a collation. -func (a *Aggregate) Collation(collation bsoncore.Document) *Aggregate { - if a == nil { - a = new(Aggregate) - } - - a.collation = collation - return a -} - -// Comment sets a value to help trace an operation. -func (a *Aggregate) Comment(comment bsoncore.Value) *Aggregate { - if a == nil { - a = new(Aggregate) - } - - a.comment = comment - return a -} - -// Hint specifies the index to use. -func (a *Aggregate) Hint(hint bsoncore.Value) *Aggregate { - if a == nil { - a = new(Aggregate) - } - - a.hint = hint - return a -} - -// Pipeline determines how data is transformed for an aggregation. -func (a *Aggregate) Pipeline(pipeline bsoncore.Document) *Aggregate { - if a == nil { - a = new(Aggregate) - } - - a.pipeline = pipeline - return a -} - -// Session sets the session for this operation. -func (a *Aggregate) Session(session *session.Client) *Aggregate { - if a == nil { - a = new(Aggregate) - } - - a.session = session - return a -} - -// ClusterClock sets the cluster clock for this operation. -func (a *Aggregate) ClusterClock(clock *session.ClusterClock) *Aggregate { - if a == nil { - a = new(Aggregate) - } - - a.clock = clock - return a -} - -// Collection sets the collection that this command will run against. -func (a *Aggregate) Collection(collection string) *Aggregate { - if a == nil { - a = new(Aggregate) - } - - a.collection = collection - return a -} - -// CommandMonitor sets the monitor to use for APM events. -func (a *Aggregate) CommandMonitor(monitor *event.CommandMonitor) *Aggregate { - if a == nil { - a = new(Aggregate) - } - - a.monitor = monitor - return a -} - -// Database sets the database to run this operation against. -func (a *Aggregate) Database(database string) *Aggregate { - if a == nil { - a = new(Aggregate) - } - - a.database = database - return a -} - -// Deployment sets the deployment to use for this operation. -func (a *Aggregate) Deployment(deployment driver.Deployment) *Aggregate { - if a == nil { - a = new(Aggregate) - } - - a.deployment = deployment - return a -} - -// ReadConcern specifies the read concern for this operation. -func (a *Aggregate) ReadConcern(readConcern *readconcern.ReadConcern) *Aggregate { - if a == nil { - a = new(Aggregate) - } - - a.readConcern = readConcern - return a -} - -// ReadPreference set the read preference used with this operation. -func (a *Aggregate) ReadPreference(readPreference *readpref.ReadPref) *Aggregate { - if a == nil { - a = new(Aggregate) - } - - a.readPreference = readPreference - return a -} - -// ServerSelector sets the selector used to retrieve a server. -func (a *Aggregate) ServerSelector(selector description.ServerSelector) *Aggregate { - if a == nil { - a = new(Aggregate) - } - - a.selector = selector - return a -} - -// WriteConcern sets the write concern for this operation. -func (a *Aggregate) WriteConcern(writeConcern *writeconcern.WriteConcern) *Aggregate { - if a == nil { - a = new(Aggregate) - } - - a.writeConcern = writeConcern - return a -} - -// Retry enables retryable writes for this operation. Retries are not handled automatically, -// instead a boolean is returned from Execute and SelectAndExecute that indicates if the -// operation can be retried. Retrying is handled by calling RetryExecute. -func (a *Aggregate) Retry(retry driver.RetryMode) *Aggregate { - if a == nil { - a = new(Aggregate) - } - - a.retry = &retry - return a -} - -// MaxAdaptiveRetries specifies the maximum number of times the driver should retry operations -// that fail with a server side overload error. -func (a *Aggregate) MaxAdaptiveRetries(maxAdaptiveRetries uint) *Aggregate { - if a == nil { - a = new(Aggregate) - } - - a.maxAdaptiveRetries = maxAdaptiveRetries - return a -} - -// EnableOverloadRetargeting specifies whether the driver adds the previously failed server's address -// to the list of deprioritized server addresses -func (a *Aggregate) EnableOverloadRetargeting(enabled bool) *Aggregate { - if a == nil { - a = new(Aggregate) - } - - a.enableOverloadRetargeting = enabled - return a -} - -// Crypt sets the Crypt object to use for automatic encryption and decryption. -func (a *Aggregate) Crypt(crypt driver.Crypt) *Aggregate { - if a == nil { - a = new(Aggregate) - } - - a.crypt = crypt - return a -} - -// ServerAPI sets the server API version for this operation. -func (a *Aggregate) ServerAPI(serverAPI *driver.ServerAPIOptions) *Aggregate { - if a == nil { - a = new(Aggregate) - } - - a.serverAPI = serverAPI - return a -} - -// Let specifies the let document to use. This option is only valid for server versions 5.0 and above. -func (a *Aggregate) Let(let bsoncore.Document) *Aggregate { - if a == nil { - a = new(Aggregate) - } - - a.let = let - return a -} - -// HasOutputStage specifies whether the aggregate contains an output stage. Used in determining when to -// append read preference at the operation level. -func (a *Aggregate) HasOutputStage(hos bool) *Aggregate { - if a == nil { - a = new(Aggregate) - } - - a.hasOutputStage = hos - return a -} - -// CustomOptions specifies extra options to use in the aggregate command. -func (a *Aggregate) CustomOptions(co map[string]bsoncore.Value) *Aggregate { - if a == nil { - a = new(Aggregate) - } - - a.customOptions = co - return a -} - -// Timeout sets the timeout for this operation. -func (a *Aggregate) Timeout(timeout *time.Duration) *Aggregate { - if a == nil { - a = new(Aggregate) - } - - a.timeout = timeout - return a -} - -// Authenticator sets the authenticator to use for this operation. -func (a *Aggregate) Authenticator(authenticator driver.Authenticator) *Aggregate { - if a == nil { - a = new(Aggregate) - } - - a.authenticator = authenticator - return a -} - -// OmitMaxTimeMS omits the automatically-calculated "maxTimeMS" from the -// command. -func (a *Aggregate) OmitMaxTimeMS(omit bool) *Aggregate { - if a == nil { - a = new(Aggregate) - } - - a.omitMaxTimeMS = omit - return a -} - -// RawData sets the rawData to access timeseries data in the compressed format. -func (a *Aggregate) RawData(rawData bool) *Aggregate { - if a == nil { - a = new(Aggregate) - } - - a.rawData = &rawData - return a -}