Dejando de leer y guardar documentos completos en MongoDB con .NET

  • mongodb
  • dotnet
  • csharp
  • bases-de-datos
  • nosql
  • backend
  • rendimiento

Cuando empezamos a trabajar con MongoDB, lo normal es apoyarse en las operaciones más sencillas. Con Find recuperamos un documento, lo modificamos en memoria y con ReplaceOneAsync lo volvemos a guardar. Es cómodo, porque trabajamos con nuestras clases de C# igual que lo haríamos con cualquier otro objeto, y durante un tiempo funciona sin problemas.

Hasta que un buen día, no es que deje de funcionar, sino que ya no cumple con nuestros requisitos. Necesitamos más velocidad y evitar colisiones, así que tenemos que buscar una solución a estos problemas. En este post os quiero contar cómo un proceso que sigue exactamente este patrón puede empezar a dar problemas de rendimiento e, incluso, a perder información. Y cómo lo puedes mejorar, paso a paso.

Dejo por aquí unos puntos como adelanto de todo lo que voy a contar en este post.

  1. Usar proyecciones para no leer el documento completo.
  2. Usar actualizaciones parciales para no reescribir el documento completo.
  3. Agrupar las escrituras con BulkWriteAsync.
  4. Añadir un campo eTag para versionar los documentos y detectar colisiones cuando dos procesos actualizan el mismo documento a la vez.

Como a día de hoy toda decisión debe estar basada en datos, al final compararemos ambas aproximaciones con un benchmark. Notese que quizás no es muy fidedigno dado que, aunque consigamos reducir el tamaño de los documentos que viajan desde/hasta la base de datos, ésta la tengo instalada en un contenedor docker en mi maquina, por lo que la latencia que tendrá es muy pequeña. En un escenario real, la latendia sería mayor y tendriamos un beneficio superior al reducir el tamaño de la información que viaja por la red.

Preparando el proyecto

Todos los ejemplos del post están en MongoDb.AdvancedOperation, una aplicación de consola en .NET que podéis descargar desde mi GitHub en este link: https://github.com/jorgediegocrespo/MongoDb.AdvancedOperations

Una vez descargado, desde la raíz del repositorio se pueden ejecutar los siguientes comandos.

docker compose up -d
dotnet run --project MongoDb.AdvancedOperation -- seed
  • docker compose up -d levanta un contenedor con MongoDB en local, en el puerto 27017. Si vuestra instancia de MongoDB no está en mongodb://localhost:27017, podéis indicar otra cadena de conexión mediante la variable de entorno MONGODB_URI.
  • dotnet run --project MongoDb.AdvancedOperation -- seed elimina la colección users si ya existe y la vuelve a crear con 6 usuarios y un índice único sobre email.

Cada usuario tiene campos en la raíz del documento, como name, email, role, status, country, loginCount o eTag, y un array sessions con sus inicios de sesión. Cada sesión es un subdocumento con su identificador, el dispositivo utilizado, la fecha de inicio, la duración, una puntuación de riesgo (riskScore) y si ha sido revocada.

El punto de partida. Leer y guardar el documento completo

Teníamos un proceso que, periódicamente, recalculaba el nivel de riesgo de cada usuario (riskLevel) a partir de la puntuación de riesgo más alta de sus sesiones.

var users = await collection.Find(FilterDefinition<User>.Empty).ToListAsync();

foreach (var user in users)
{
    var maxRiskScore = user.Sessions.Max(session => session.RiskScore);

    var updatedUser = user with
    {
        RiskLevel = RiskRules.GetRiskLevel(maxRiskScore),
        UpdatedAt = DateTime.UtcNow
    };

    await collection.ReplaceOneAsync(u => u.Id == user.Id, updatedUser);
}

Con pocos usuarios no habría nada que objetar. Pero, ¿qué pasaría si cada usuario fuese acumulando decenas de sesiones?. La colección iría creciendo y el proceso empezaría a ser cada vez más lento. Si nos fijamos, para calcular un único número estamos haciendo todos estos pasos.

  • Leyendo el documento completo, con todas sus sesiones, cuando solo necesitábamos la puntuación máxima.
  • Deserializando todo ese contenido en objetos de C#, con el consumo de memoria y CPU que conlleva.
  • Enviando el documento completo de vuelta para cambiar dos campos.
  • Haciendo un viaje de red por cada usuario.

Además, podría haber un problema más grave y menos visible. Este es la pérdida de cambios. Mientras el proceso trabaja con su copia en memoria, pueden registrarse nuevos inicios de sesión. Si reemplaza el documento con esa copia antigua, también sobrescribe el loginCount actualizado. En el ejemplo, el comando concurrency reproduce el problema haciendo que diez acciones lean el mismo documento antes de intentar guardar su incremento.

Para lanzar este ejemplo, podemos ejecutar el siguiente comando.

dotnet run --project MongoDb.AdvancedOperation -- concurrency

El cual nos deja una salida por consola análoga a la siguiente.

Replace without eTag: concurrent actions=10, initial loginCount=5, expected=15, actual=6, lost increments=9, detected conflicts=0.

Replace with eTag and retries: concurrent actions=10, initial loginCount=6, expected=16, actual=16, lost increments=0, detected conflicts=<n>.

En esa salida, podemos ver que, con la colección recién inicializada, Jorge parte de loginCount = 5. El primer escenario demuestra que diez reemplazos basados en la misma lectura solo conservan un incremento. El segundo empieza con el valor que dejó el primero y conserva los diez incrementos. <n> representa el número de conflictos detectados, que depende de cómo se intercalen los reintentos.

Vamos a ir resolviendo estos problemas uno a uno.

Paso 1. Leer solo lo que necesitamos con proyecciones

Una proyección indica a MongoDB qué campos queremos que nos devuelva. El driver nos permite definirlas de dos formas.

La primera es con Builders<User>.Projection, incluyendo o excluyendo campos. Como el resultado ya no encaja en nuestra clase User, lo recibimos como BsonDocument:

var contactProjection = Builders<User>.Projection
    .Include(user => user.Name)
    .Include(user => user.Email)
    .Exclude(user => user.Id);

var contacts = await collection
    .Find(user => user.Status == "active")
    .Project(contactProjection)
    .ToListAsync();

La segunda opción (para mi, es la más interesante) es una proyección tipada mediante una expresión. El driver traduce la expresión a una etapa de proyección que se ejecuta en el servidor, de modo que podemos pedir directamente la puntuación máxima de las sesiones en lugar de traernos el array.

public sealed record UserRiskSnapshot(ObjectId Id, string ETag, int MaxRiskScore);

var snapshots = await collection
    .Find(FilterDefinition<User>.Empty)
    .Project(user => new UserRiskSnapshot(
        user.Id,
        user.ETag,
        user.Sessions.Max(session => session.RiskScore)))
    .ToListAsync();

Con el uso de proyecciones, pasariamos de recibir un documento de varios KB a recibir un identificador, un eTag y un número entero.

Siguiente con el ejemplo, se pueden ver las dos opciones de proyecciones lanzando el comando dotnet run --project MongoDb.AdvancedOperation -- projection.

Limitaciones de las proyecciones:

  • En una misma proyección no se pueden mezclar inclusiones y exclusiones, salvo si la exclusión es del campo _id.
  • No todas las expresiones de C# se pueden traducir a MQL. Si usamos un método propio dentro de la proyección, el driver lanzará una excepción al ejecutarla.
  • Max sobre un array vacío devuelve null en el servidor. Si el campo de destino es un int, fallará la deserialización, así que debemos usar int? si puede haber usuarios sin sesiones.

Paso 2. Modificar solo lo que cambia con actualizaciones parciales

En este punto puede que estemos pensando que, si solo leemos una parte del documento, ya no podemos reemplazarlo, porque no lo tenemos completo. Pero, la verdad es que, no lo necesitamos, ya que la operacion Builders<User>.Update nos permite indicar únicamente los cambios que queremos aplicar.

Por ejemplo, para activar al usuario Jorge, guardar la fecha de modificación, renovar su eTag e incrementar su contador de accesos, podemos hacerlo con el código que se muestra a continuación.

var filter = Builders<User>.Filter.Eq(user => user.Email, "jorge@dicres.dev");

var update = Builders<User>.Update
    .Set(user => user.Status, "active")
    .Set(user => user.UpdatedAt, DateTime.UtcNow)
    .Set(user => user.ETag, User.NewETag())
    .Inc(user => user.LoginCount, 1);

var result = await collection.UpdateOneAsync(filter, update);

Console.WriteLine(
    $"Partial update: matched={result.MatchedCount}, modified={result.ModifiedCount}.");

MongoDB aplica todos los cambios de forma atómica sobre el documento y el resto de campos no se tocan. Además, el incremento lo calcula el servidor, por lo que dos procesos que incrementen loginCount a la vez no se pisan. Con esto, el problema del loginCount perdido desaparecería.

Este ejemplo lo podríamos ejecutar con el comando dotnet run --project MongoDb.AdvancedOperation -- partial-update.

Limitaciones de UpdateOneAsync, Set e Inc:

  • UpdateOneAsync actualiza como máximo un documento y, si el filtro no encuentra ninguno, no inserta uno nuevo salvo que activemos IsUpsert en las UpdateOptions.
  • Set sustituye el valor completo del campo, así que ojo con esto, ya que, si lo aplicamos a sessions, se reemplazaría todo el array. Y aplicado a un subdocumento, también lo reemplazaría entero.
  • Inc crea el campo si no existe y no impone límites por sí solo. Para ello habría que combinarlo con Min y Max.

Otros operadores de actualización

Una vez que dejamos de reemplazar documentos, el builder nos ofrece muchos más operadores para modificar documentos sin leerlos antes. No voy a entrar en detalle en todos en este post, pero si que hay algunos que aparecen en el comando update-operators de la aplicación de ejemplo.

var update = Builders<User>.Update;

// Root fields
update.Unset(user => user.LegacyCode);                    // Removes the field
update.Mul("sessions.$[].riskScore", 0.5);                // Multiplies across all sessions
update.CurrentDate(user => user.UpdatedAt);               // Uses the server date
update.Rename(user => user.LegacyCode, "legacyId");       // Renames the field
update.Min(user => user.FastestSessionSeconds, 120);      // Applies only if 120 is lower
update.Max("sessions.$[mobile].riskScore", 75);           // Applies only if 75 is higher

// Arrays
update.Push(user => user.Sessions, newSession);           // Adds one item
update.PushEach(user => user.Sessions, newSessions);      // Adds multiple items
update.AddToSet(user => user.Sessions, existingSession);  // Adds only if it does not exist
update.PullFilter(user => user.Sessions, revokedSession); // Removes items matching the filter

El $[mobile] del ejemplo de Max es un operador posicional de filtrado, que se completa con ArrayFilters para decidir a qué elementos del array se aplica el cambio.

var result = await collection.UpdateOneAsync(
    filter,
    Builders<User>.Update.Max("sessions.$[mobile].riskScore", 75),
    new UpdateOptions
    {
        ArrayFilters =
        [
            new BsonDocumentArrayFilterDefinition<LoginSession>(
                new BsonDocument("mobile.device", "mobile"))
        ]
    });

En cambio, $[] se aplica a todos los elementos del array.

Algunos matices a tener en cuenta

  • Si Unset se aplica a un campo, éste no se pone a null, sino que se elimina dicho campo, por lo que, si el modelo de C# lo espera, al leerlo tendrá su valor por defecto. Por contra, si se aplica a una posición de un array (sessions.0) deja un null en vez de eliminar el elemento. Para eliminarlo, se pueden usar las operaciones Pull y PullFilter.
  • Mul, Min y Max crean el campo si no existe. Además, Mul puede cambiar el tipo BSON si, por ejemplo, se multiplica un entero por 0.5, en cuyo caso se guardaría un double.
  • CurrentDate usa el reloj del servidor de MongoDB, lo que evita discrepancias entre los relojes de distintas instancias de nuestra aplicación.
  • Rename sobrescribe el campo de destino si ya existe, así que hay que usarlo con cuidado. Si renombramos un campo en la base de datos, tenemos que reflejarlo también en nuestra clase de C# o el driver no sabrá deserializarlo. En el ejemplo, User tiene mapeados tanto legacyCode como legacyId.
  • Push y PushEach no comprueban duplicados y el campo de destino debe ser un array. Como los documentos tienen un límite de 16 MiB, conviene acotar los arrays que crecen sin parar usando Slice dentro de PushEach.
  • AddToSet compara el subdocumento completo, incluido el orden de sus campos. Dos sesiones con los mismos datos en distinto orden se consideran distintas.
  • PullFilter elimina todas las coincidencias, no solo la primera.
  • Al indicar rutas como cadenas de texto ("sessions.$[mobile].riskScore") perdemos la comprobación de tipos en tiempo de compilación.

Paso 3. Agrupar las escrituras con BulkWriteAsync

Tras el paso 2, ya leemos y escribimos mucha menos información, pero seguimos haciendo un viaje de red por cada usuario que estamos modificando. Para poder hacer todas las actualizaciones a la vez, podemos utilizar BulkWriteAsync, que nos permite agrupar inserciones, actualizaciones y borrados en una sola llamada, tal y como se muestra a continuación.

var operations = new WriteModel<User>[]
{
    new InsertOneModel<User>(
        new User
        {
            Name = "Marta",
            Email = "marta@example.com",
            Role = "admin",
            Status = "active",
            Country = "ES",
            LoginCount = 0
        }),

    new UpdateOneModel<User>(
        Builders<User>.Filter.Eq(user => user.Email, "pepe@example.com"),
        Builders<User>.Update
            .Set(user => user.Status, "inactive")
            .Set(user => user.UpdatedAt, DateTime.UtcNow)
            .Set(user => user.ETag, User.NewETag())),

    new DeleteManyModel<User>(
        Builders<User>.Filter.Eq(user => user.Status, "banned"))
};

var result = await collection.BulkWriteAsync(
    operations,
    new BulkWriteOptions { IsOrdered = false });

En nuestro proceso, bastaría con transformar cada operación en un UpdateOneModel<User> y enviarlas todas juntas.

Un parámetro importante es IsOrdered. Si tiene un valor false, MongoDB puede ejecutar las operaciones en cualquier orden y no detiene el lote si una falla. Por el contrario, si queremos mantener el orden de las operaciones y que se pare la ejecución si una de ellas falla, debemos usar IsOrdered = true.

Limitaciones de BulkWriteAsync

  • Un lote no es una transacción. Aunque usemos IsOrdered = true, si una operación falla, las que ya se han aplicado no se revierten.
  • El driver divide automáticamente el lote en varios comandos si supera el número máximo de operaciones o el tamaño máximo de mensaje (48 MB).
  • Para lotes muy grandes conviene enviarlos en bloques, por ejemplo de 1.000 operaciones, para no acumular todos los modelos en memoria.

Paso 4. Controlar la concurrencia con un eTag

Con las actualizaciones parciales conseguimos resolver el problema de concurrencia en loginCount, porque nuestro proceso ya no escribe ese campo, dado que usamos la operación Inc. No obstante, sí que podríamos tener un problema de concurrencia si el proceso actualiza el campo riskLevel a partir de unos datos que leyó antes. Si entre la lectura y la escritura alguien añade una sesión de riesgo, guardaríamos un riskLevel calculado con datos obsoletos.

Un primer acercamiento son las actualizaciones condicionales, es decir, incluir en el filtro el estado que esperamos encontrar. Por ejemplo, solo activamos a Jorge si sigue inactivo.

var filter = Builders<User>.Filter.And(
    Builders<User>.Filter.Eq(user => user.Email, "jorge@dicres.dev"),
    Builders<User>.Filter.Eq(user => user.Status, "inactive"));

Si otro proceso lo ha cambiado antes, MatchedCount será 0 y no se modificará nada. El problema es que tendríamos que añadir al filtro todos los campos en los que se basa nuestra decisión. Para simplificar todo esto podemos usar el versionado del documento completo.

Para eso añadimos a User un campo eTag, que se genera al crear el documento y que cualquier escritura debe renovar.

[BsonElement("eTag")]
public string ETag { get; init; } = NewETag();

public static string NewETag() => Guid.NewGuid().ToString("N");

Al leer, la proyección incluye el eTag. Al escribir, filtramos por el _id y por el eTag que hemos leído previamente. Además, en la misma actualización, generamos uno nuevo.

private async Task<bool> TryUpdateRiskLevelAsync(UserRiskSnapshot snapshot)
{
    var filter = Builders<User>.Filter.And(
        Builders<User>.Filter.Eq(user => user.Id, snapshot.Id),
        Builders<User>.Filter.Eq(user => user.ETag, snapshot.ETag));

    var update = Builders<User>.Update
        .Set(user => user.RiskLevel, RiskRules.GetRiskLevel(snapshot.MaxRiskScore))
        .Set(user => user.ETag, User.NewETag())
        .CurrentDate(user => user.UpdatedAt);

    var result = await collection.UpdateOneAsync(filter, update);
    return result.MatchedCount == 1;
}

Si alguien ha modificado el documento entre nuestra lectura y nuestra escritura, su eTag ya no coincidirá, el filtro no encontrará nada y sabremos que ha habido una colisión. En ese caso, volvemos a leer el documento, recalculamos el cambio a partir del valor actual y lo intentamos de nuevo.

private async Task<int> RegisterLoginWithRetryAsync(User snapshot, int maxAttempts)
{
    var idFilter = Builders<User>.Filter.Eq(user => user.Id, snapshot.Id);
    for (var attempt = 1; attempt <= maxAttempts; attempt++)
    {
        var writeFilter = Builders<User>.Filter.And(
            idFilter,
            Builders<User>.Filter.Eq(user => user.ETag, snapshot.ETag));
        var updatedUser = snapshot with
        {
            LoginCount = checked(snapshot.LoginCount + 1),
            UpdatedAt = DateTime.UtcNow,
            ETag = User.NewETag()
        };

        var result = await collection.ReplaceOneAsync(writeFilter, updatedUser);
        if (result.MatchedCount == 1)
        {
            return attempt - 1;
        }

        if (attempt < maxAttempts)
        {
            snapshot = await collection.Find(idFilter).FirstAsync();
        }
    }

    throw new InvalidOperationException(
        $"Could not register the login after {maxAttempts} attempts.");
}

Es lo que se conoce como concurrencia optimista. Esto es, no bloqueamos el documento, simplemente detectamos el conflicto al escribir. La implementación del ejemplo, limita los reintentos a diez. Y, si no consigue registrar el login en todos esos reintentos, informa del error en lugar de darlo por aplicado.

El comando concurrency de nuestro ejemplo, compara dos escenarios con diez acciones concurrentes. Primero lee diez copias del documento y espera a que todas las lecturas terminen antes de empezar las escrituras. Así cada acción parte del mismo loginCount y, en la variante con eTag, se detectan al menos nueve conflictos.

Replace without eTag: concurrent actions=10, initial loginCount=5, expected=15, actual=6, lost increments=9, detected conflicts=0.
Replace with eTag and retries: concurrent actions=10, initial loginCount=6, expected=16, actual=16, lost increments=0, detected conflicts=<n>.

Sin eTag, las diez acciones reemplazan el documento con loginCount = 6 y solo el primer incremento sobrevive (los otros nueve se pierden). Con eTag, cada reemplazo filtra también por la versión que leyó y guarda un eTag nuevo. Cuando detecta que otra acción ya escribió, vuelve a leer el documento y recalcula el incremento. <n> representa el total de conflictos. Para nuestro ejemplo, siempre es al menos nueve y puede variar según el orden de los reintentos. Cada escenario calcula su resultado esperado desde su propio valor inicial, por eso el segundo parte del valor que dejó el primero.

En el proceso de recálculo de riesgo, el mismo mecanismo se aplica con actualizaciones parciales y BulkWriteAsync, tal y como se muestra a continuación.

var snapshots = await collection
    .Find(FilterDefinition<User>.Empty)
    .Project(user => new UserRiskSnapshot(
        user.Id,
        user.ETag,
        user.Sessions.Max(session => session.RiskScore)))
    .ToListAsync();

var operations = snapshots
    .Select(snapshot => new UpdateOneModel<User>(
        Builders<User>.Filter.And(
            Builders<User>.Filter.Eq(user => user.Id, snapshot.Id),
            Builders<User>.Filter.Eq(user => user.ETag, snapshot.ETag)),
        Builders<User>.Update
            .Set(user => user.RiskLevel, RiskRules.GetRiskLevel(snapshot.MaxRiskScore))
            .Set(user => user.ETag, User.NewETag())
            .CurrentDate(user => user.UpdatedAt)))
    .ToList();

var result = await collection.BulkWriteAsync(
    operations,
    new BulkWriteOptions { IsOrdered = false });

var conflicts = operations.Count - result.MatchedCount;

Limitaciones del eTag

  • Solo funciona si todas las escrituras sobre la colección renuevan el eTag. Basta con que un proceso lo olvide para que sus cambios pasen desapercibidos. En la aplicación de ejemplo, para no complicar los ejemplos, los operadores de la sección anterior no lo renuevan.
  • BulkWriteAsync nos dice cuántas operaciones no encontraron su documento, pero no cuáles. Para reintentarlas tenemos que volver a consultar esos usuarios y comparar su eTag con el que teníamos.
  • Si hay muchas colisiones sobre los mismos documentos, los reintentos se acumulan. La concurrencia optimista funciona bien cuando los conflictos son la excepción.
  • No siempre lo necesitamos. Si la actualización no depende de lo que hemos leído, como el Inc de loginCount, los operadores atómicos ya son suficientes.

El mismo eTag se puede exponer en una API REST con la cabecera ETag y recibirlo de vuelta en If-Match, devolviendo un 412 Precondition Failed si no coincide.

Benchmark. Antes y después

Para medir la mejora, la aplicación de ejemplo incluye el comando benchmark, el cual crea una colección independiente (users_benchmark) con 10.000 usuarios y 50 sesiones por usuario (valores por defecto que se pueden modificar al lanzar el comando), unos 6,6 KB por documento y 64 MB en total, y recalcula el riskLevel de todos ellos de cuatro formas distintas, tal y como se indica a continuación.

  1. Leyendo el documento completo y reemplazándolo uno a uno (el punto de partida).
  2. Leyendo el documento completo y reemplazándolo con BulkWriteAsync.
  3. Usando una proyección y actualizaciones parciales una a una.
  4. Usando una proyección y actualizaciones parciales con BulkWriteAsync (la versión final, con eTag).

Antes de cada ejecución la colección se vuelve a crear para que todos los escenarios partan de los mismos datos. Cada escenario se ejecuta tres veces y se muestra la mediana. Además del tiempo, se mide el tráfico de red a partir de serverStatus y la memoria asignada en .NET con GC.GetTotalAllocatedBytes.

Para ejecutar este benchmark, podemos lanzar el siguiente comando.

dotnet run -c Release --project MongoDb.AdvancedOperation -- benchmark 10000 50 3

Estos son los resultados con MongoDB 8 en Docker en local.

Escenario Tiempo (ms) Recibido por MongoDB (MB) Enviado por MongoDB (MB) Memoria asignada en .NET (MB)
Documento completo + ReplaceOne uno a uno 2.804 66,0 64,6 497,3
Documento completo + BulkWrite de ReplaceOne 1.441 64,6 64,1 254,8
Proyección + UpdateOne parcial uno a uno 1.924 3,1 1,5 370,3
Proyección + BulkWrite de UpdateOne parciales 571 1,8 0,9 153,7

Viendo los datos anteriores, podemos extraer las siguientes conclusiones.

  • La versión final es casi 5 veces más rápida que la inicial y asigna algo más de 3 veces menos memoria.
  • El tráfico de red pasa de unos 130 MB a menos de 3 MB. Con proyecciones y actualizaciones parciales ya no movemos los documentos, solo los datos que necesitamos.
  • BulkWriteAsync por sí solo reduce el tiempo a la mitad, pero sigue moviendo los mismos 130 MB. Las proyecciones y las actualizaciones parciales reducen el tráfico, pero los 10.000 viajes de red siguen penalizando. Las mejoras se complementan.
  • Estas pruebas se han hecho con MongoDB en la misma máquina, con una latencia de red prácticamente nula. Con una base de datos remota, cada viaje de red cuesta milisegundos y la diferencia entre escribir uno a uno o en bloque es mucho mayor. Lo mismo ocurre con el volumen de datos si el ancho de banda es limitado.

Los números exactos dependerán de vuestra máquina, pero la tendencia debería ser la misma.

Agregaciones. Cuando ni siquiera necesitamos los documentos.

Llevando la idea de las proyecciones un poco más allá, hay casos en los que no necesitamos los documentos en la aplicación, sino un resultado calculado a partir de ellos. Los pipelines de agregación permiten filtrar, agrupar y transformar los datos en el servidor y devolver solo el resultado.

Por ejemplo, para obtener el número de usuarios activos y de accesos por país.

var results = await collection.Aggregate()
    .Match(user => user.Status == "active")
    .Group(
        user => user.Country,
        group => new
        {
            Country = group.Key,
            TotalUsers = group.Count(),
            TotalLogins = group.Sum(user => user.LoginCount)
        })
    .Match(result => result.TotalLogins > 10)
    .SortByDescending(result => result.TotalUsers)
    .ThenByDescending(result => result.TotalLogins)
    .Project(result => new CountryStats
    {
        Country = result.Country,
        TotalUsers = result.TotalUsers,
        TotalLogins = result.TotalLogins
    })
    .ToListAsync();

El driver traduce cada método a su etapa de MongoDB ($match, $group, $sort, $project). Además, podemos usar otras como Limit ($limit), Unwind ($unwind) para convertir cada elemento de un array en un documento, Lookup ($lookup) para cruzar colecciones o Count ($count).

Para analizar cómo ejecuta MongoDB un pipeline, el comando explain de la aplicación lo lanza con la verbosidad executionStats, que nos muestra, entre otras cosas, si se ha usado un índice (IXSCAN) o se ha recorrido la colección completa (COLLSCAN) y cuántos documentos se han examinado.

Limitaciones de las agregaciones:

  • Igual que con las proyecciones, no todas las expresiones de C# se pueden traducir a MQL.
  • Las etapas $group y $sort tienen un límite de 100 MB de memoria. Si se supera, la agregación fallará salvo que activemos AllowDiskUse = true.
  • Si el resultado es muy grande, es mejor recorrerlo con ToCursorAsync() que cargarlo entero con ToListAsync().

Conclusiones

Leer un documento completo, modificarlo en memoria y reemplazarlo es la forma más sencilla de trabajar con MongoDB, pero no escala bien, dado que se mueven muchos más datos de los necesarios y se pueden sobrescribir los cambios de otros procesos.

Por lo tanto, la solución pasa por ir dando los pasos antes expuestos.

  • Proyecciones para leer solo los campos necesarios, o incluso calcularlos en el servidor.
  • Actualizaciones parciales con operadores atómicos para modificar solo lo que cambia.
  • BulkWriteAsync para reducir los viajes de red.
  • Un eTag para detectar colisiones cuando la escritura depende de lo que hemos leído.

En nuestro ejemplo, el resultado, tras aplicar todas estas mejoras, fue un proceso casi 5 veces más rápido, que mueve un 98 % menos de datos y que ya no pierde cambios.

Fuentes