Skip to content

Commit ad4dd5a

Browse files
DmitryLukyanovDmitryLukyanov
authored
CSHARP-3422: Change estimatedDocumentCount() to use the $collStats Agg Stage Instead of Count Command. (#469)
* CSHARP-3422: Change estimatedDocumentCount() to use the $collStats Agg Stage Instead of Count Command.
1 parent 94df8d2 commit ad4dd5a

27 files changed

+3237
-62
lines changed

‎src/MongoDB.Driver.Core/Core/Misc/Feature.cs

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -56,6 +56,7 @@ public class Feature
5656
private static readonly Feature __currentOpCommand = new Feature("CurrentOpCommand", new SemanticVersion(3, 2, 0));
5757
private static readonly Feature __documentValidation = new Feature("DocumentValidation", new SemanticVersion(3, 2, 0));
5858
private static readonly Feature __directConnectionSetting = new Feature("DirectConnectionSetting", new SemanticVersion(4, 4, 0));
59+
private static readonly Feature __estimatedDocumentCountByCollStats = new Feature("EstimatedDocumentCountByCollStats", new SemanticVersion(4, 9, 0, ""));
5960
private static readonly Feature __eval = new Feature("Eval", new SemanticVersion(0, 0, 0), new SemanticVersion(4, 1, 0, ""));
6061
private static readonly Feature __explainCommand = new Feature("ExplainCommand", new SemanticVersion(3, 0, 0));
6162
private static readonly Feature __failPoints = new Feature("FailPoints", new SemanticVersion(2, 4, 0));
@@ -267,6 +268,11 @@ public class Feature
267268
/// </summary>
268269
public static Feature DirectConnectionSetting => __directConnectionSetting;
269270

271+
/// <summary>
272+
/// Gets the estimatedDocumentCountByCollStats feature.
273+
/// </summary>
274+
public static Feature EstimatedDocumentCountByCollStats => __estimatedDocumentCountByCollStats;
275+
270276
/// <summary>
271277
/// Gets the eval feature.
272278
/// </summary>

‎src/MongoDB.Driver.Core/Core/Misc/SemanticVersion.cs

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -366,6 +366,7 @@ public static bool TryParse(string value, out SemanticVersion result)
366366
return !(b < a);
367367
}
368368

369+
// private methods
369370
private ServerVersion AsServerVersion()
370371
{
371372
return new ServerVersion(_major, _minor, _patch, _preRelease);

‎src/MongoDB.Driver.Core/Core/Operations/CountOperation.cs

Lines changed: 20 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -29,7 +29,7 @@ namespace MongoDB.Driver.Core.Operations
2929
/// <summary>
3030
/// Represents a count operation.
3131
/// </summary>
32-
public class CountOperation : IReadOperation<long>
32+
public class CountOperation : IReadOperation<long>, IExecutableInRetryableReadContext<long>
3333
{
3434
// fields
3535
private Collation _collation;
@@ -198,25 +198,38 @@ public long Execute(IReadBinding binding, CancellationToken cancellationToken)
198198

199199
using (var context = RetryableReadContext.Create(binding, _retryRequested, cancellationToken))
200200
{
201-
var operation = CreateOperation(context);
202-
var document = operation.Execute(context, cancellationToken);
203-
return document["n"].ToInt64();
201+
return Execute(context, cancellationToken);
204202
}
205203
}
206204

205+
/// <inheritdoc/>
206+
public long Execute(RetryableReadContext context, CancellationToken cancellationToken)
207+
{
208+
var operation = CreateOperation(context);
209+
var document = operation.Execute(context, cancellationToken);
210+
return document["n"].ToInt64();
211+
}
212+
207213
/// <inheritdoc/>
208214
public async Task<long> ExecuteAsync(IReadBinding binding, CancellationToken cancellationToken)
209215
{
210216
Ensure.IsNotNull(binding, nameof(binding));
211217

212218
using (var context = await RetryableReadContext.CreateAsync(binding, _retryRequested, cancellationToken).ConfigureAwait(false))
213219
{
214-
var operation = CreateOperation(context);
215-
var document = await operation.ExecuteAsync(context, cancellationToken).ConfigureAwait(false);
216-
return document["n"].ToInt64();
220+
return await ExecuteAsync(context, cancellationToken).ConfigureAwait(false);
217221
}
218222
}
219223

224+
/// <inheritdoc/>
225+
public async Task<long> ExecuteAsync(RetryableReadContext context, CancellationToken cancellationToken)
226+
{
227+
var operation = CreateOperation(context);
228+
var document = await operation.ExecuteAsync(context, cancellationToken).ConfigureAwait(false);
229+
return document["n"].ToInt64();
230+
}
231+
232+
// private methods
220233
private ReadCommandOperation<BsonDocument> CreateOperation(RetryableReadContext context)
221234
{
222235
var command = CreateCommand(context.Channel.ConnectionDescription, context.Binding.Session);

‎src/MongoDB.Driver.Core/Core/Operations/EstimatedDocumentCountOperation.cs

Lines changed: 203 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,203 @@
1+
/* Copyright 2021-present MongoDB Inc.
2+
*
3+
* Licensed under the Apache License, Version 2.0 (the "License");
4+
* you may not use this file except in compliance with the License.
5+
* You may obtain a copy of the License at
6+
*
7+
* http://www.apache.org/licenses/LICENSE-2.0
8+
*
9+
* Unless required by applicable law or agreed to in writing, software
10+
* distributed under the License is distributed on an "AS IS" BASIS,
11+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12+
* See the License for the specific language governing permissions and
13+
* limitations under the License.
14+
*/
15+
16+
using System;
17+
using System.Collections.Generic;
18+
using System.Threading;
19+
using System.Threading.Tasks;
20+
using MongoDB.Bson;
21+
using MongoDB.Bson.Serialization.Serializers;
22+
using MongoDB.Driver.Core.Bindings;
23+
using MongoDB.Driver.Core.Misc;
24+
using MongoDB.Driver.Core.WireProtocol.Messages.Encoders;
25+
26+
namespace MongoDB.Driver.Core.Operations
27+
{
28+
/// <summary>
29+
/// Represents an estimated document count operation.
30+
/// </summary>
31+
public class EstimatedDocumentCountOperation : IReadOperation<long>
32+
{
33+
// private fields
34+
private readonly CollectionNamespace _collectionNamespace;
35+
private TimeSpan? _maxTime;
36+
private readonly MessageEncoderSettings _messageEncoderSettings;
37+
private ReadConcern _readConcern = ReadConcern.Default;
38+
private bool _retryRequested;
39+
40+
// constructors
41+
/// <summary>
42+
/// Initializes a new instance of the <see cref="EstimatedDocumentCountOperation"/> class.
43+
/// </summary>
44+
/// <param name="collectionNamespace">The collection namespace.</param>
45+
/// <param name="messageEncoderSettings">The message encoder settings.</param>
46+
public EstimatedDocumentCountOperation(CollectionNamespace collectionNamespace, MessageEncoderSettings messageEncoderSettings)
47+
{
48+
_collectionNamespace = Ensure.IsNotNull(collectionNamespace, nameof(collectionNamespace));
49+
_messageEncoderSettings = Ensure.IsNotNull(messageEncoderSettings, nameof(messageEncoderSettings));
50+
}
51+
52+
// public properties
53+
/// <summary>
54+
/// Gets the collection namespace.
55+
/// </summary>
56+
public CollectionNamespace CollectionNamespace => _collectionNamespace;
57+
58+
/// <summary>
59+
/// Gets or sets the maximum time the server should spend on this operation.
60+
/// </summary>
61+
public TimeSpan? MaxTime
62+
{
63+
get => _maxTime;
64+
set => _maxTime = Ensure.IsNullOrInfiniteOrGreaterThanOrEqualToZero(value, nameof(value));
65+
}
66+
67+
/// <summary>
68+
/// Gets the message encoder settings.
69+
/// </summary>
70+
public MessageEncoderSettings MessageEncoderSettings => _messageEncoderSettings;
71+
72+
/// <summary>
73+
/// Gets or sets the read concern.
74+
/// </summary>
75+
public ReadConcern ReadConcern
76+
{
77+
get => _readConcern;
78+
set => _readConcern = Ensure.IsNotNull(value, nameof(value));
79+
}
80+
81+
/// <summary>
82+
/// Gets or sets a value indicating whether to retry.
83+
/// </summary>
84+
public bool RetryRequested
85+
{
86+
get => _retryRequested;
87+
set => _retryRequested = value;
88+
}
89+
90+
/// <inheritdoc/>
91+
public long Execute(IReadBinding binding, CancellationToken cancellationToken)
92+
{
93+
Ensure.IsNotNull(binding, nameof(binding));
94+
95+
using (var context = RetryableReadContext.Create(binding, _retryRequested, cancellationToken))
96+
{
97+
if (Feature.EstimatedDocumentCountByCollStats.IsSupported(context.Channel.ConnectionDescription.ServerVersion))
98+
{
99+
var operation = CreateAggregationOperation(context.Channel.ConnectionDescription.ServerVersion);
100+
IAsyncCursor<BsonDocument> cursor;
101+
try
102+
{
103+
cursor = operation.Execute(context, cancellationToken);
104+
}
105+
catch (MongoCommandException ex) when (ex.Code == (int)ServerErrorCode.NamespaceNotFound)
106+
{
107+
// In the event this aggregation is run against a non-existent namespace, a NamespaceNotFound(26) error will be returned during execution.
108+
return 0;
109+
}
110+
var results = cursor.ToList(cancellationToken);
111+
112+
return ExtractCountFromAggregationResults(results);
113+
}
114+
else
115+
{
116+
var operation = CreateCountOperation();
117+
118+
return operation.Execute(context, cancellationToken);
119+
}
120+
}
121+
}
122+
123+
/// <inheritdoc/>
124+
public async Task<long> ExecuteAsync(IReadBinding binding, CancellationToken cancellationToken)
125+
{
126+
Ensure.IsNotNull(binding, nameof(binding));
127+
128+
using (var context = RetryableReadContext.Create(binding, _retryRequested, cancellationToken))
129+
{
130+
if (Feature.EstimatedDocumentCountByCollStats.IsSupported(context.Channel.ConnectionDescription.ServerVersion))
131+
{
132+
var operation = CreateAggregationOperation(context.Channel.ConnectionDescription.ServerVersion);
133+
IAsyncCursor<BsonDocument> cursor;
134+
try
135+
{
136+
cursor = await operation.ExecuteAsync(context, cancellationToken).ConfigureAwait(false);
137+
}
138+
catch (MongoCommandException ex) when (ex.Code == (int)ServerErrorCode.NamespaceNotFound)
139+
{
140+
// In the event this aggregation is run against a non-existent namespace, a NamespaceNotFound(26) error will be returned during execution.
141+
return 0;
142+
}
143+
var results = await cursor.ToListAsync(cancellationToken).ConfigureAwait(false);
144+
145+
return ExtractCountFromAggregationResults(results);
146+
}
147+
else
148+
{
149+
var operation = CreateCountOperation();
150+
151+
return await operation.ExecuteAsync(context, cancellationToken).ConfigureAwait(false);
152+
}
153+
}
154+
}
155+
156+
// private methods
157+
private IExecutableInRetryableReadContext<IAsyncCursor<BsonDocument>> CreateAggregationOperation(SemanticVersion serverVersion)
158+
{
159+
Feature.ReadConcern.ThrowIfNotSupported(serverVersion, _readConcern);
160+
161+
var pipeline = CreateAggregationPipeline();
162+
var aggregateOperation = new AggregateOperation<BsonDocument>(_collectionNamespace, pipeline, BsonDocumentSerializer.Instance, _messageEncoderSettings)
163+
{
164+
MaxTime = _maxTime,
165+
ReadConcern = _readConcern,
166+
RetryRequested = _retryRequested
167+
};
168+
return aggregateOperation;
169+
170+
IEnumerable<BsonDocument> CreateAggregationPipeline() =>
171+
new BsonDocument[]
172+
{
173+
new BsonDocument("$collStats", new BsonDocument("count", new BsonDocument())),
174+
new BsonDocument(
175+
"$group",
176+
new BsonDocument
177+
{
178+
{ "_id", 1 },
179+
{ "n", new BsonDocument("$sum", "$count") }
180+
})
181+
};
182+
}
183+
184+
private IExecutableInRetryableReadContext<long> CreateCountOperation()
185+
{
186+
var countOperation = new CountOperation(_collectionNamespace, _messageEncoderSettings)
187+
{
188+
MaxTime = _maxTime,
189+
ReadConcern = _readConcern,
190+
RetryRequested = _retryRequested
191+
};
192+
return countOperation;
193+
}
194+
195+
private long ExtractCountFromAggregationResults(List<BsonDocument> results) =>
196+
results.Count switch
197+
{
198+
0 => 0,
199+
1 => results[0]["n"].ToInt64(),
200+
_ => throw new MongoClientException($"Expected aggregate command for {nameof(EstimatedDocumentCountOperation)} to return 1 document, but got {results.Count}."),
201+
};
202+
}
203+
}

‎src/MongoDB.Driver/MongoCollectionImpl.cs

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -951,9 +951,9 @@ private DistinctOperation<TField> CreateDistinctOperation<TField>(FieldDefinitio
951951
};
952952
}
953953

954-
private CountOperation CreateEstimatedDocumentCountOperation(EstimatedDocumentCountOptions options)
954+
private EstimatedDocumentCountOperation CreateEstimatedDocumentCountOperation(EstimatedDocumentCountOptions options)
955955
{
956-
return new CountOperation(_collectionNamespace, _messageEncoderSettings)
956+
return new EstimatedDocumentCountOperation(_collectionNamespace, _messageEncoderSettings)
957957
{
958958
MaxTime = options?.MaxTime,
959959
RetryRequested = _database.Client.Settings.RetryReads

‎tests/MongoDB.Driver.Core.TestHelpers/XunitExtensions/RequireServer.cs

Lines changed: 9 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -226,7 +226,7 @@ private bool CanRunOn(ICluster cluster, BsonDocument requirement)
226226
{
227227
var actualVersion = CoreTestConfiguration.ServerVersion;
228228
var minServerVersion = SemanticVersion.Parse(item.Value.AsString);
229-
if (actualVersion < minServerVersion)
229+
if (SemanticVersionCompareToAsReleased(actualVersion, minServerVersion) < 0)
230230
{
231231
return false;
232232
}
@@ -236,7 +236,7 @@ private bool CanRunOn(ICluster cluster, BsonDocument requirement)
236236
{
237237
var actualVersion = CoreTestConfiguration.ServerVersion;
238238
var maxServerVersion = SemanticVersion.Parse(item.Value.AsString);
239-
if (actualVersion > maxServerVersion)
239+
if (SemanticVersionCompareToAsReleased(actualVersion, maxServerVersion) > 0)
240240
{
241241
return false;
242242
}
@@ -284,5 +284,12 @@ private ClusterType MapTopologyToClusterType(string topology)
284284
default: throw new ArgumentException($"Invalid topology: \"{topology}\".", nameof(topology));
285285
}
286286
}
287+
288+
private int SemanticVersionCompareToAsReleased(SemanticVersion a, SemanticVersion b)
289+
{
290+
var aReleased = new SemanticVersion(a.Major, a.Minor, a.Patch);
291+
var bReleased = new SemanticVersion(b.Major, b.Minor, b.Patch);
292+
return aReleased.CompareTo(bReleased);
293+
}
287294
}
288295
}

0 commit comments

Comments
 (0)