Skip to content

Commit 2a32ac3

Browse files
jyeminjyemin
committed
Support aggregation with $merge as a string (#768)
JAVA-4258
1 parent 81cc1a7 commit 2a32ac3

File tree

4 files changed

+84
-16
lines changed

4 files changed

+84
-16
lines changed

‎driver-async/src/main/com/mongodb/async/client/AggregateIterableImpl.java‎

Lines changed: 13 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -186,13 +186,19 @@ private MongoNamespace getOutNamespace() {
186186
if (lastStageDocument.containsKey("$out")) {
187187
return new MongoNamespace(namespace.getDatabaseName(), lastStageDocument.getString("$out").getValue());
188188
} else if (lastStageDocument.containsKey("$merge")) {
189-
BsonDocument mergeDocument = lastStageDocument.getDocument("$merge");
190-
if (mergeDocument.isDocument("into")) {
191-
BsonDocument intoDocument = mergeDocument.getDocument("into");
192-
return new MongoNamespace(intoDocument.getString("db", new BsonString(namespace.getDatabaseName())).getValue(),
193-
intoDocument.getString("coll").getValue());
194-
} else if (mergeDocument.isString("into")) {
195-
return new MongoNamespace(namespace.getDatabaseName(), mergeDocument.getString("into").getValue());
189+
if (lastStageDocument.isString("$merge")) {
190+
return new MongoNamespace(namespace.getDatabaseName(), lastStageDocument.getString("$merge").getValue());
191+
} else if (lastStageDocument.isDocument("$merge")) {
192+
BsonDocument mergeDocument = lastStageDocument.getDocument("$merge");
193+
if (mergeDocument.isDocument("into")) {
194+
BsonDocument intoDocument = mergeDocument.getDocument("into");
195+
return new MongoNamespace(intoDocument.getString("db", new BsonString(namespace.getDatabaseName())).getValue(),
196+
intoDocument.getString("coll").getValue());
197+
} else if (mergeDocument.isString("into")) {
198+
return new MongoNamespace(namespace.getDatabaseName(), mergeDocument.getString("into").getValue());
199+
}
200+
} else {
201+
throw new IllegalStateException("Cannot return a cursor when the value for $merge stage is not a string or a document");
196202
}
197203
}
198204

‎driver-async/src/test/unit/com/mongodb/async/client/AggregateIterableSpecification.groovy‎

Lines changed: 33 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -209,7 +209,7 @@ class AggregateIterableSpecification extends Specification {
209209
.comment('this is a comment'))
210210
}
211211

212-
def 'should build the expected AggregateToCollectionOperation for $merge'() {
212+
def 'should build the expected AggregateToCollectionOperation for $merge document'() {
213213
given:
214214
def cursor = Stub(AsyncBatchCursor) {
215215
next(_) >> {
@@ -358,6 +358,38 @@ class AggregateIterableSpecification extends Specification {
358358
.comment('this is a comment'))
359359
}
360360

361+
def 'should build the expected AggregateToCollectionOperation for $merge string'() {
362+
given:
363+
def cursor = Stub(AsyncBatchCursor) {
364+
next(_) >> {
365+
it[0].onResult(null, null)
366+
}
367+
}
368+
def executor = new TestOperationExecutor([cursor, cursor, cursor, cursor, cursor, cursor, cursor]);
369+
def collectionName = 'collectionName'
370+
def collectionNamespace = new MongoNamespace(namespace.getDatabaseName(), collectionName)
371+
def pipeline = [new Document('$match', 1), new Document('$merge', new Document('into', collectionName))]
372+
373+
when: 'aggregation includes $merge'
374+
new AggregateIterableImpl(null, namespace, Document, Document, codecRegistry, readPreference, readConcern, writeConcern, executor,
375+
pipeline, AggregationLevel.COLLECTION, true)
376+
.into([]) { result, t -> }
377+
378+
def operation = executor.getReadOperation() as WriteOperationThenCursorReadOperation
379+
380+
then:
381+
expect operation.getAggregateToCollectionOperation(), isTheSameAs(new AggregateToCollectionOperation(namespace,
382+
[new BsonDocument('$match', new BsonInt32(1)),
383+
new BsonDocument('$merge', new BsonDocument('into', new BsonString(collectionName)))],
384+
readConcern, writeConcern))
385+
386+
when:
387+
operation = operation.getReadOperation() as FindOperation
388+
389+
then:
390+
operation.getNamespace() == collectionNamespace
391+
}
392+
361393
def 'should handle exceptions correctly'() {
362394
given:
363395
def codecRegistry = fromProviders([new ValueCodecProvider(), new BsonValueCodecProvider()])

‎driver-sync/src/main/com/mongodb/client/internal/AggregateIterableImpl.java‎

Lines changed: 13 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -187,13 +187,19 @@ private MongoNamespace getOutNamespace() {
187187
if (lastStageDocument.containsKey("$out")) {
188188
return new MongoNamespace(namespace.getDatabaseName(), lastStageDocument.getString("$out").getValue());
189189
} else if (lastStageDocument.containsKey("$merge")) {
190-
BsonDocument mergeDocument = lastStageDocument.getDocument("$merge");
191-
if (mergeDocument.isDocument("into")) {
192-
BsonDocument intoDocument = mergeDocument.getDocument("into");
193-
return new MongoNamespace(intoDocument.getString("db", new BsonString(namespace.getDatabaseName())).getValue(),
194-
intoDocument.getString("coll").getValue());
195-
} else if (mergeDocument.isString("into")) {
196-
return new MongoNamespace(namespace.getDatabaseName(), mergeDocument.getString("into").getValue());
190+
if (lastStageDocument.isString("$merge")) {
191+
return new MongoNamespace(namespace.getDatabaseName(), lastStageDocument.getString("$merge").getValue());
192+
} else if (lastStageDocument.isDocument("$merge")) {
193+
BsonDocument mergeDocument = lastStageDocument.getDocument("$merge");
194+
if (mergeDocument.isDocument("into")) {
195+
BsonDocument intoDocument = mergeDocument.getDocument("into");
196+
return new MongoNamespace(intoDocument.getString("db", new BsonString(namespace.getDatabaseName())).getValue(),
197+
intoDocument.getString("coll").getValue());
198+
} else if (mergeDocument.isString("into")) {
199+
return new MongoNamespace(namespace.getDatabaseName(), mergeDocument.getString("into").getValue());
200+
}
201+
} else {
202+
throw new IllegalStateException("Cannot return a cursor when the value for $merge stage is not a string or a document");
197203
}
198204
}
199205

‎driver-sync/src/test/unit/com/mongodb/client/internal/AggregateIterableSpecification.groovy‎

Lines changed: 25 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -195,7 +195,7 @@ class AggregateIterableSpecification extends Specification {
195195
.comment('this is a comment'))
196196
}
197197

198-
def 'should build the expected AggregateToCollectionOperation for $merge'() {
198+
def 'should build the expected AggregateToCollectionOperation for $merge document'() {
199199
given:
200200
def executor = new TestOperationExecutor([null, null, null, null, null, null, null])
201201
def collectionName = 'collectionName'
@@ -335,6 +335,30 @@ class AggregateIterableSpecification extends Specification {
335335
.comment('this is a comment'))
336336
}
337337

338+
def 'should build the expected AggregateToCollectionOperation for $merge string'() {
339+
given:
340+
def executor = new TestOperationExecutor([null, null, null, null, null, null, null])
341+
def collectionName = 'collectionName'
342+
def collectionNamespace = new MongoNamespace(namespace.getDatabaseName(), collectionName)
343+
def pipeline = [new BsonDocument('$match', new BsonDocument()), new BsonDocument('$merge', new BsonString(collectionName))]
344+
345+
when:
346+
new AggregateIterableImpl(null, namespace, Document, Document, codecRegistry, readPreference, readConcern, writeConcern, executor,
347+
pipeline, AggregationLevel.COLLECTION, false)
348+
.iterator()
349+
350+
def operation = executor.getWriteOperation() as AggregateToCollectionOperation
351+
352+
then:
353+
expect operation, isTheSameAs(new AggregateToCollectionOperation(namespace, pipeline, readConcern, writeConcern,
354+
AggregationLevel.COLLECTION))
355+
356+
when:
357+
operation = executor.getReadOperation() as FindOperation<Document>
358+
359+
then:
360+
operation.getNamespace() == collectionNamespace
361+
}
338362

339363
def 'should use ClientSession for AggregationOperation'() {
340364
given:

0 commit comments

Comments
 (0)