Skip to content

GODRIVER-3522 Add support for the rawData option for time-series bucket access. #2159

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 3 commits into from
Aug 11, 2025
Merged

GODRIVER-3522 Add support for the rawData option for time-series bucket access. #2159

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
106ffd8
Add rawData option to aggregate, count/countDocuments/estimatedDocume…
qingyang-hu Aug 4, 2025
6b77cb6
Add rawData option to find/findAndModify, update/replace, and collect…
qingyang-hu Aug 4, 2025
5360234
Add rawData option to client bulkWrite, listCollections, and index op…
qingyang-hu Aug 8, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions internal/integration/unified/client_operation_execution.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@ import (
"go.mongodb.org/mongo-driver/v2/mongo"
"go.mongodb.org/mongo-driver/v2/mongo/options"
"go.mongodb.org/mongo-driver/v2/x/bsonx/bsoncore"
"go.mongodb.org/mongo-driver/v2/x/mongo/driver/xoptions"
)

// This file contains helpers to execute client operations.
Expand Down Expand Up @@ -235,6 +236,11 @@ func executeClientBulkWrite(ctx context.Context, operation *operation) (*operati
return nil, err
}
opts.SetWriteConcern(c)
case "rawData":
err = xoptions.SetInternalClientBulkWriteOptions(opts, key, val.Boolean())
if err != nil {
return nil, err
}
default:
return nil, fmt.Errorf("unrecognized bulkWrite option %q", key)
}
Expand Down
90 changes: 89 additions & 1 deletion internal/integration/unified/collection_operation_execution.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@ import (
"go.mongodb.org/mongo-driver/v2/mongo"
"go.mongodb.org/mongo-driver/v2/mongo/options"
"go.mongodb.org/mongo-driver/v2/x/bsonx/bsoncore"
"go.mongodb.org/mongo-driver/v2/x/mongo/driver/xoptions"
)

// This file contains helpers to execute collection operations.
Expand Down Expand Up @@ -75,6 +76,11 @@ func executeAggregate(ctx context.Context, operation *operation) (*operationResu
pipeline = bsonutil.RawToInterfaces(bsonutil.RawArrayToDocuments(val.Array())...)
case "let":
opts.SetLet(val.Document())
case "rawData":
err = xoptions.SetInternalAggregateOptions(opts, key, val.Boolean())
if err != nil {
return nil, err
}
default:
return nil, fmt.Errorf("unrecognized aggregate option %q", key)
}
Expand Down Expand Up @@ -125,6 +131,11 @@ func executeBulkWrite(ctx context.Context, operation *operation) (*operationResu
}
case "let":
opts.SetLet(val.Document())
case "rawData":
err = xoptions.SetInternalBulkWriteOptions(opts, key, val.Boolean())
if err != nil {
return nil, err
}
default:
return nil, fmt.Errorf("unrecognized bulkWrite option %q", key)
}
Expand Down Expand Up @@ -202,6 +213,11 @@ func executeCountDocuments(ctx context.Context, operation *operation) (*operatio
return nil, fmt.Errorf("the maxTimeMS collection option is not supported")
case "skip":
opts.SetSkip(int64(val.Int32()))
case "rawData":
err = xoptions.SetInternalCountOptions(opts, key, val.Boolean())
if err != nil {
return nil, err
}
default:
return nil, fmt.Errorf("unrecognized countDocuments option %q", key)
}
Expand All @@ -225,6 +241,7 @@ func executeCreateIndex(ctx context.Context, operation *operation) (*operationRe

var keys bson.Raw
indexOpts := options.Index()
opts := options.CreateIndexes()

elems, err := operation.Arguments.Elements()
if err != nil {
Expand Down Expand Up @@ -279,6 +296,11 @@ func executeCreateIndex(ctx context.Context, operation *operation) (*operationRe
indexOpts.SetWeights(val.Document())
case "wildcardProjection":
indexOpts.SetWildcardProjection(val.Document())
case "rawData":
err = xoptions.SetInternalCreateIndexesOptions(opts, key, val.Boolean())
if err != nil {
return nil, err
}
default:
return nil, fmt.Errorf("unrecognized createIndex option %q", key)
}
Expand All @@ -291,7 +313,8 @@ func executeCreateIndex(ctx context.Context, operation *operation) (*operationRe
Keys: keys,
Options: indexOpts,
}
name, err := coll.Indexes().CreateOne(ctx, model)

name, err := coll.Indexes().CreateOne(ctx, model, opts)
return newValueResult(bson.TypeString, bsoncore.AppendString(nil, name), err), nil
}

Expand Down Expand Up @@ -433,6 +456,11 @@ func executeDeleteOne(ctx context.Context, operation *operation) (*operationResu
opts.SetHint(hint)
case "let":
opts.SetLet(val.Document())
case "rawData":
err = xoptions.SetInternalDeleteOneOptions(opts, key, val.Boolean())
if err != nil {
return nil, err
}
default:
return nil, fmt.Errorf("unrecognized deleteOne option %q", key)
}
Expand Down Expand Up @@ -487,6 +515,11 @@ func executeDeleteMany(ctx context.Context, operation *operation) (*operationRes
opts.SetHint(hint)
case "let":
opts.SetLet(val.Document())
case "rawData":
err = xoptions.SetInternalDeleteManyOptions(opts, key, val.Boolean())
if err != nil {
return nil, err
}
default:
return nil, fmt.Errorf("unrecognized deleteMany option %q", key)
}
Expand Down Expand Up @@ -545,6 +578,11 @@ func executeDistinct(ctx context.Context, operation *operation) (*operationResul
// ensured an analogue exists, extend "skippedTestDescriptions" to avoid
// this error.
return nil, fmt.Errorf("the maxTimeMS collection option is not supported")
case "rawData":
err = xoptions.SetInternalDistinctOptions(opts, key, val.Boolean())
if err != nil {
return nil, err
}
default:
return nil, fmt.Errorf("unrecognized distinct option %q", key)
}
Expand Down Expand Up @@ -593,6 +631,11 @@ func executeDropIndex(ctx context.Context, operation *operation) (*operationResu
// ensured an analogue exists, extend "skippedTestDescriptions" to avoid
// this error.
return nil, fmt.Errorf("the maxTimeMS collection option is not supported")
case "rawData":
err = xoptions.SetInternalDropIndexesOptions(dropIndexOpts, key, val.Boolean())
if err != nil {
return nil, err
}
default:
return nil, fmt.Errorf("unrecognized dropIndex option %q", key)
}
Expand Down Expand Up @@ -690,6 +733,11 @@ func executeEstimatedDocumentCount(ctx context.Context, operation *operation) (*
// ensured an analogue exists, extend "skippedTestDescriptions" to avoid
// this error.
return nil, fmt.Errorf("the maxTimeMS collection option is not supported")
case "rawData":
err = xoptions.SetInternalEstimatedDocumentCountOptions(opts, key, val.Boolean())
if err != nil {
return nil, err
}
default:
return nil, fmt.Errorf("unrecognized estimatedDocumentCount option %q", key)
}
Expand Down Expand Up @@ -842,6 +890,11 @@ func executeFindOneAndDelete(ctx context.Context, operation *operation) (*operat
opts.SetSort(val.Document())
case "let":
opts.SetLet(val.Document())
case "rawData":
err = xoptions.SetInternalFindOneAndDeleteOptions(opts, key, val.Boolean())
if err != nil {
return nil, err
}
default:
return nil, fmt.Errorf("unrecognized findOneAndDelete option %q", key)
}
Expand Down Expand Up @@ -924,6 +977,11 @@ func executeFindOneAndReplace(ctx context.Context, operation *operation) (*opera
opts.SetSort(val.Document())
case "upsert":
opts.SetUpsert(val.Boolean())
case "rawData":
err = xoptions.SetInternalFindOneAndReplaceOptions(opts, key, val.Boolean())
if err != nil {
return nil, err
}
default:
return nil, fmt.Errorf("unrecognized findOneAndReplace option %q", key)
}
Expand Down Expand Up @@ -1016,6 +1074,11 @@ func executeFindOneAndUpdate(ctx context.Context, operation *operation) (*operat
}
case "upsert":
opts.SetUpsert(val.Boolean())
case "rawData":
err = xoptions.SetInternalFindOneAndUpdateOptions(opts, key, val.Boolean())
if err != nil {
return nil, err
}
default:
return nil, fmt.Errorf("unrecognized findOneAndUpdate option %q", key)
}
Expand Down Expand Up @@ -1062,6 +1125,11 @@ func executeInsertMany(ctx context.Context, operation *operation) (*operationRes
documents = bsonutil.RawToInterfaces(bsonutil.RawArrayToDocuments(val.Array())...)
case "ordered":
opts.SetOrdered(val.Boolean())
case "rawData":
err = xoptions.SetInternalInsertManyOptions(opts, key, val.Boolean())
if err != nil {
return nil, err
}
default:
return nil, fmt.Errorf("unrecognized insertMany option %q", key)
}
Expand Down Expand Up @@ -1112,6 +1180,11 @@ func executeInsertOne(ctx context.Context, operation *operation) (*operationResu
opts.SetBypassDocumentValidation(val.Boolean())
case "comment":
opts.SetComment(val)
case "rawData":
err = xoptions.SetInternalInsertOneOptions(opts, key, val.Boolean())
if err != nil {
return nil, err
}
default:
return nil, fmt.Errorf("unrecognized insertOne option %q", key)
}
Expand Down Expand Up @@ -1156,6 +1229,11 @@ func executeListIndexes(ctx context.Context, operation *operation) (*operationRe
switch key {
case "batchSize":
opts.SetBatchSize(val.Int32())
case "rawData":
err = xoptions.SetInternalListIndexesOptions(opts, key, val.Boolean())
if err != nil {
return nil, err
}
default:
return nil, fmt.Errorf("unrecognized listIndexes option: %q", key)
}
Expand Down Expand Up @@ -1302,6 +1380,11 @@ func executeReplaceOne(ctx context.Context, operation *operation) (*operationRes
opts.SetUpsert(val.Boolean())
case "let":
opts.SetLet(val.Document())
case "rawData":
err = xoptions.SetInternalReplaceOptions(opts, key, val.Boolean())
if err != nil {
return nil, err
}
default:
return nil, fmt.Errorf("unrecognized replaceOne option %q", key)
}
Expand Down Expand Up @@ -1500,6 +1583,11 @@ func createFindCursor(ctx context.Context, operation *operation) (*cursorResult,
case "maxAwaitTimeMS":
maxAwaitTimeMS := time.Duration(val.Int32()) * time.Millisecond
opts.SetMaxAwaitTime(maxAwaitTimeMS)
case "rawData":
err = xoptions.SetInternalFindOptions(opts, key, val.Boolean())
if err != nil {
return nil, err
}
default:
return nil, fmt.Errorf("unrecognized find option %q", key)
}
Expand Down
16 changes: 16 additions & 0 deletions internal/integration/unified/crud_helpers.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@ import (
"go.mongodb.org/mongo-driver/v2/bson"
"go.mongodb.org/mongo-driver/v2/internal/bsonutil"
"go.mongodb.org/mongo-driver/v2/mongo/options"
"go.mongodb.org/mongo-driver/v2/x/mongo/driver/xoptions"
)

// newMissingArgumentError creates an error to convey that an argument that is required to run an operation is missing
Expand Down Expand Up @@ -67,6 +68,11 @@ func createUpdateManyArguments(args bson.Raw) (*updateArguments, *options.Update
}
case "upsert":
opts.SetUpsert(val.Boolean())
case "rawData":
err := xoptions.SetInternalUpdateManyOptions(opts, key, val.Boolean())
if err != nil {
return nil, nil, err
}
default:
return nil, nil, fmt.Errorf("unrecognized update option %q", key)
}
Expand Down Expand Up @@ -125,6 +131,11 @@ func createUpdateOneArguments(args bson.Raw) (*updateArguments, *options.UpdateO
opts.SetUpsert(val.Boolean())
case "sort":
opts.SetSort(val.Document())
case "rawData":
err := xoptions.SetInternalUpdateOneOptions(opts, key, val.Boolean())
if err != nil {
return nil, nil, err
}
default:
return nil, nil, fmt.Errorf("unrecognized update option %q", key)
}
Expand Down Expand Up @@ -162,6 +173,11 @@ func createListCollectionsArguments(args bson.Raw) (*listCollectionsArguments, e
lca.filter = val.Document()
case "nameOnly":
lca.opts.SetNameOnly(val.Boolean())
case "rawData":
err := xoptions.SetInternalListCollectionsOptions(lca.opts, key, val.Boolean())
if err != nil {
return nil, err
}
default:
return nil, fmt.Errorf("unrecognized listCollections option %q", key)
}
Expand Down
13 changes: 13 additions & 0 deletions mongo/bulk_write.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,7 @@ type bulkWrite struct {
writeConcern *writeconcern.WriteConcern
result BulkWriteResult
let interface{}
rawData *bool
}

func (bw *bulkWrite) execute(ctx context.Context) error {
Expand Down Expand Up @@ -209,6 +210,10 @@ func (bw *bulkWrite) runInsert(ctx context.Context, batch bulkWriteBatch) (opera
}
op = op.Retry(retry)

if bw.rawData != nil {
op.RawData(*bw.rawData)
Copy link
Preview

Copilot AI Aug 8, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The return value of op.RawData() is not being assigned back to op. This should be op = op.RawData(*bw.rawData) to properly chain the method call.

Suggested change
op.RawData(*bw.rawData)
op = op.RawData(*bw.rawData)

Copilot uses AI. Check for mistakes.

}

err := op.Execute(ctx)

return op.Result(), err
Expand Down Expand Up @@ -282,6 +287,10 @@ func (bw *bulkWrite) runDelete(ctx context.Context, batch bulkWriteBatch) (opera
}
op = op.Retry(retry)

if bw.rawData != nil {
op.RawData(*bw.rawData)
Copy link
Preview

Copilot AI Aug 8, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The return value of op.RawData() is not being assigned back to op. This should be op = op.RawData(*bw.rawData) to properly chain the method call.

Suggested change
op.RawData(*bw.rawData)
op = op.RawData(*bw.rawData)

Copilot uses AI. Check for mistakes.

}

err := op.Execute(ctx)

return op.Result(), err
Expand Down Expand Up @@ -415,6 +424,10 @@ func (bw *bulkWrite) runUpdate(ctx context.Context, batch bulkWriteBatch) (opera
}
op = op.Retry(retry)

if bw.rawData != nil {
op.RawData(*bw.rawData)
Copy link
Preview

Copilot AI Aug 8, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The return value of op.RawData() is not being assigned back to op. This should be op = op.RawData(*bw.rawData) to properly chain the method call.

Suggested change
op.RawData(*bw.rawData)
op = op.RawData(*bw.rawData)

Copilot uses AI. Check for mistakes.

}

err := op.Execute(ctx)

return op.Result(), err
Expand Down
6 changes: 6 additions & 0 deletions mongo/client.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@ import (
"go.mongodb.org/mongo-driver/v2/internal/httputil"
"go.mongodb.org/mongo-driver/v2/internal/logger"
"go.mongodb.org/mongo-driver/v2/internal/mongoutil"
"go.mongodb.org/mongo-driver/v2/internal/optionsutil"
"go.mongodb.org/mongo-driver/v2/internal/ptrutil"
"go.mongodb.org/mongo-driver/v2/internal/serverselector"
"go.mongodb.org/mongo-driver/v2/internal/uuid"
Expand Down Expand Up @@ -957,6 +958,11 @@ func (c *Client) BulkWrite(ctx context.Context, writes []ClientBulkWrite,
selector: selector,
writeConcern: wc,
}
if rawDataOpt := optionsutil.Value(bwo.Internal, "rawData"); rawDataOpt != nil {
if rawData, ok := rawDataOpt.(bool); ok {
op.rawData = &rawData
}
}
if bwo.VerboseResults == nil || !(*bwo.VerboseResults) {
op.errorsOnly = true
} else if !acknowledged {
Expand Down
5 changes: 5 additions & 0 deletions mongo/client_bulk_write.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -44,6 +44,7 @@ type clientBulkWrite struct {
client *Client
selector description.ServerSelector
writeConcern *writeconcern.WriteConcern
rawData *bool

result ClientBulkWriteResult
}
Expand Down Expand Up @@ -143,6 +144,10 @@ func (bw *clientBulkWrite) newCommand() func([]byte, description.SelectedServer)
}
dst = bsoncore.AppendDocumentElement(dst, "let", let)
}
// Set rawData for 8.2+ servers.
if bw.rawData != nil && desc.WireVersion != nil && driverutil.VersionRangeIncludes(*desc.WireVersion, 27) {
dst = bsoncore.AppendBooleanElement(dst, "rawData", *bw.rawData)
}
return dst, nil
}
}
Expand Down
Loading
Loading