Execute queries de snapshots de longa duração

Nesta página

Comparação de read concerns locais e de snapshots

Exemplos
Executar queries relacionadas a partir do mesmo ponto no tempo
Ler a partir de um estado consistente dos dados de algum ponto no passado
Configurar retenção de snapshot
Espaço em disco e histórico

As queries de snapshot permitem que você leia os dados conforme apareciam em um único ponto no tempo no passado recente.

A partir do MongoDB 5.0, você pode usar a preocupação de leitura "snapshot" para consultar dados em nós secundários . Esse recurso aumenta a versatilidade e a resiliência das leituras do seu aplicativo. Você não precisa criar uma cópia estática de seus dados, movê-los para um sistema separado e isolar manualmente essas queries de longa duração para que não interfiram em seu volume de trabalho operacional. Em vez disso, você pode realizar queries de longa duração em um banco de banco de dados transacional ativo enquanto lê a partir de um estado consistente dos dados.

O uso de preocupação de leitura "snapshot" em nós secundários não impacto o volume de trabalho de gravação do aplicativo. Somente as leituras de aplicação se beneficiam do fato de queries de longa duração sendo isoladas para secundários.

Use queries de snapshots quando quiser:

Execute várias queries relacionadas e garanta que cada query leia dados do mesmo ponto.
Certifique-se de ler de um estado consistente dos dados de algum ponto no passado.

Comparação de read concerns locais e de snapshots

Quando o MongoDB executa consultas de longa duração usando a preocupação de leitura padrão "local" , os resultados da consulta podem conter dados de gravações que ocorrem ao mesmo tempo que a consulta. Como resultado, a consulta poderá retornar resultados inesperados ou inconsistentes.

Para evitar esse cenário, crie uma sessão e especifique a preocupação de leitura "snapshot". Com a preocupação de leitura "snapshot", o MongoDB executa sua query com isolamento de snapshot, o que significa que sua query lê os dados como eles apareceram em um único ponto no passado recente.

Exemplos

Os exemplos nesta página mostram como você pode usar queries de snapshots para:

Executar queries relacionadas a partir do mesmo ponto no tempo
Ler a partir de um estado consistente dos dados de algum ponto no passado

Executar queries relacionadas a partir do mesmo ponto no tempo

An ppreocupação de leitura "snapshot" permite que você execute várias consultas relacionadas dentro de uma sessão e garanta que cada consulta leia os dados a partir do mesmo ponto no tempo.

Um canil tem um banco de banco de dados pets que contém collections para cada tipo de animal de estimação. O banco de banco de dados do pets tem estas collections:

cats
dogs

Cada documento de cada coleção contém um campo adoptable , indicando se o animal de estimação está disponível para adoção. Por exemplo, um documento na coleção cats terá essa aparência:

{
   "name": "Whiskers",
   "color": "white",
   "age": 10,
   "adoptable": true
}

Você deseja executar uma query para ver o número total de animais de estimação disponíveis para adoção em todas as collections. Para fornecer uma exibição consistente dos dados, você deseja garantir que os dados retornados de cada collection sejam de um único ponto no tempo.

Para atingir esse objetivo, use a preocupação de leitura "snapshot" em uma sessão:

mongoc_client_session_t *cs = NULL;
mongoc_collection_t *cats_collection = NULL;
mongoc_collection_t *dogs_collection = NULL;
int64_t adoptable_pets_count = 0;
bson_error_t error;
mongoc_session_opt_t *session_opts;
cats_collection = mongoc_client_get_collection (client, "pets", "cats");
dogs_collection = mongoc_client_get_collection (client, "pets", "dogs");
/* Seed 'pets.cats' and 'pets.dogs' with example data */
if (!pet_setup (cats_collection, dogs_collection)) {
   goto cleanup;
}
/* start a snapshot session */
session_opts = mongoc_session_opts_new ();
mongoc_session_opts_set_snapshot (session_opts, true);
cs = mongoc_client_start_session (client, session_opts, &error);
mongoc_session_opts_destroy (session_opts);
if (!cs) {
   MONGOC_ERROR ("Could not start session: %s", error.message);
   goto cleanup;
}
/*
 * Perform the following aggregation pipeline, and accumulate the count in
 * `adoptable_pets_count`.
 *
 *  adoptablePetsCount = db.cats.aggregate(
 *      [ { "$match": { "adoptable": true } },
 *        { "$count": "adoptableCatsCount" } ], session=s
 *  ).next()["adoptableCatsCount"]
 *
 *  adoptablePetsCount += db.dogs.aggregate(
 *      [ { "$match": { "adoptable": True} },
 *        { "$count": "adoptableDogsCount" } ], session=s
 *  ).next()["adoptableDogsCount"]
 *
 * Remember in order to apply the client session to
 * this operation, you must append the client session to the options passed
 * to `mongoc_collection_aggregate`, i.e.,
 *
 * mongoc_client_session_append (cs, &opts, &error);
 * cursor = mongoc_collection_aggregate (
 *    collection, MONGOC_QUERY_NONE, pipeline, &opts, NULL);
 */
accumulate_adoptable_count (cs, cats_collection, &adoptable_pets_count);
accumulate_adoptable_count (cs, dogs_collection, &adoptable_pets_count);
printf ("there are %" PRId64 " adoptable pets\n", adoptable_pets_count);

using namespace mongocxx;
using bsoncxx::builder::basic::kvp;
using bsoncxx::builder::basic::make_document;
auto db = client["pets"];
int64_t adoptable_pets_count = 0;
auto opts = mongocxx::options::client_session{};
opts.snapshot(true);
auto session = client.start_session(opts);
{
    pipeline p;
    p.match(make_document(kvp("adoptable", true))).count("adoptableCatsCount");
    auto cursor = db["cats"].aggregate(session, p);
    for (auto doc : cursor) {
        adoptable_pets_count += doc.find("adoptableCatsCount")->get_int32();
    }
}
{
    pipeline p;
    p.match(make_document(kvp("adoptable", true))).count("adoptableDogsCount");
    auto cursor = db["dogs"].aggregate(session, p);
    for (auto doc : cursor) {
        adoptable_pets_count += doc.find("adoptableDogsCount")->get_int32();
    }
}

ctx := context.TODO()
sess, err := client.StartSession(options.Session().SetSnapshot(true))
if err != nil {
	return err
}
defer sess.EndSession(ctx)
var adoptablePetsCount int32
err = mongo.WithSession(ctx, sess, func(ctx context.Context) error {
	// Count the adoptable cats
	const adoptableCatsOutput = "adoptableCatsCount"
	cursor, err := db.Collection("cats").Aggregate(ctx, mongo.Pipeline{
		bson.D{{"$match", bson.D{{"adoptable", true}}}},
		bson.D{{"$count", adoptableCatsOutput}},
	})
	if err != nil {
		return err
	}
	if !cursor.Next(ctx) {
		return fmt.Errorf("expected aggregate to return a document, but got none")
	}
	resp := cursor.Current.Lookup(adoptableCatsOutput)
	adoptableCatsCount, ok := resp.Int32OK()
	if !ok {
		return fmt.Errorf("failed to find int32 field %q in document %v", adoptableCatsOutput, cursor.Current)
	}
	adoptablePetsCount += adoptableCatsCount
	// Count the adoptable dogs
	const adoptableDogsOutput = "adoptableDogsCount"
	cursor, err = db.Collection("dogs").Aggregate(ctx, mongo.Pipeline{
		bson.D{{"$match", bson.D{{"adoptable", true}}}},
		bson.D{{"$count", adoptableDogsOutput}},
	})
	if err != nil {
		return err
	}
	if !cursor.Next(ctx) {
		return fmt.Errorf("expected aggregate to return a document, but got none")
	}
	resp = cursor.Current.Lookup(adoptableDogsOutput)
	adoptableDogsCount, ok := resp.Int32OK()
	if !ok {
		return fmt.Errorf("failed to find int32 field %q in document %v", adoptableDogsOutput, cursor.Current)
	}
	adoptablePetsCount += adoptableDogsCount
	return nil
})
if err != nil {
	return err
}

db = client.pets
async with await client.start_session(snapshot=True) as s:
    adoptablePetsCount = 0
    docs = await db.cats.aggregate(
        [{"$match": {"adoptable": True}}, {"$count": "adoptableCatsCount"}], session=s
    ).to_list(None)
    adoptablePetsCount = docs[0]["adoptableCatsCount"]
    docs = await db.dogs.aggregate(
        [{"$match": {"adoptable": True}}, {"$count": "adoptableDogsCount"}], session=s
    ).to_list(None)
    adoptablePetsCount += docs[0]["adoptableDogsCount"]
print(adoptablePetsCount)

$catsCollection = $client->selectCollection('pets', 'cats');
$dogsCollection = $client->selectCollection('pets', 'dogs');
$session = $client->startSession(['snapshot' => true]);
$adoptablePetsCount = $catsCollection->aggregate(
    [
        ['$match' => ['adoptable' => true]],
        ['$count' => 'adoptableCatsCount'],
    ],
    ['session' => $session],
)->toArray()[0]->adoptableCatsCount;
$adoptablePetsCount += $dogsCollection->aggregate(
    [
        ['$match' => ['adoptable' => true]],
        ['$count' => 'adoptableDogsCount'],
    ],
    ['session' => $session],
)->toArray()[0]->adoptableDogsCount;
var_dump($adoptablePetsCount);

db = client.pets
with client.start_session(snapshot=True) as s:
    adoptablePetsCount = (
        (
            db.cats.aggregate(
                [{"$match": {"adoptable": True}}, {"$count": "adoptableCatsCount"}],
                session=s,
            )
        ).next()
    )["adoptableCatsCount"]
    adoptablePetsCount += (
        (
            db.dogs.aggregate(
                [{"$match": {"adoptable": True}}, {"$count": "adoptableDogsCount"}],
                session=s,
            )
        ).next()
    )["adoptableDogsCount"]
print(adoptablePetsCount)

client = Mongo::Client.new(uri_string, database: "pets")
client.start_session(snapshot: true) do |session|
  adoptable_pets_count = client['cats'].aggregate([
    { "$match": { "adoptable": true } },
    { "$count": "adoptable_cats_count" }
  ], session: session).first["adoptable_cats_count"]
  adoptable_pets_count += client['dogs'].aggregate([
    { "$match": { "adoptable": true } },
    { "$count": "adoptable_dogs_count" }
  ], session: session).first["adoptable_dogs_count"]
  puts adoptable_pets_count
end

A série anterior de comandos:

Usa MongoClient() para estabelecer uma conexão com a implantação do MongoDB.
Passa para o banco de dados pets.
Estabelece uma sessão. O comando especifica snapshot=True, portanto a sessão usa preocupação de leitura "snapshot".
Executa estas ação para cada collection no banco de dados pets :
- Usa $match para filtrar documentos onde o campo adoptable é True.
- Utiliza $count para retornar uma contagem dos documentos filtrados.
- Aumenta a variável adoptablePetsCount com a contagem do banco de banco de dados.
Imprime a variável adoptablePetsCount.

Todas as queries da sessão leem os dados conforme eles apareceram no mesmo ponto no tempo. Como resultado, a contagem final reflete uma visão consistente dos dados.

Observação

Se a sessão durar mais do que o período de retenção do histórico do WiredTiger (300 segundos, por padrão), a query apresentará um erro SnapshotTooOld . Para saber como configurar a retenção de snapshots e habilitar queries de longa duração, consulte Configurar retenção de snapshots.

Ler a partir de um estado consistente dos dados de algum ponto no passado

A preocupacão de leitura "snapshot" garante que sua query leia os dados conforme apareciam em algum ponto específico no passado recente.

Uma loja de sapatos online tem uma coleção de sales que contém dados para cada item vendido na loja. Por exemplo, um documento na coleção sales se parece com isto:

{
   "shoeType": "boot",
   "price": 30,
   "saleDate": ISODate("2022-02-02T06:01:17.171Z")
}

Todos os dias, à meia-noite, é executada uma query para ver quantos pares de sapatos foram vendidos naquele dia. A query diária de vendas é parecida com a seguinte:

mongoc_client_session_t *cs = NULL;
mongoc_collection_t *sales_collection = NULL;
bson_error_t error;
mongoc_session_opt_t *session_opts;
bson_t *pipeline = NULL;
bson_t opts = BSON_INITIALIZER;
mongoc_cursor_t *cursor = NULL;
const bson_t *doc = NULL;
bool ok = true;
bson_iter_t iter;
int64_t total_sales = 0;
sales_collection = mongoc_client_get_collection (client, "retail", "sales");
/* seed 'retail.sales' with example data */
if (!retail_setup (sales_collection)) {
   goto cleanup;
}
/* start a snapshot session */
session_opts = mongoc_session_opts_new ();
mongoc_session_opts_set_snapshot (session_opts, true);
cs = mongoc_client_start_session (client, session_opts, &error);
mongoc_session_opts_destroy (session_opts);
if (!cs) {
   MONGOC_ERROR ("Could not start session: %s", error.message);
   goto cleanup;
}
if (!mongoc_client_session_append (cs, &opts, &error)) {
   MONGOC_ERROR ("could not apply session options: %s", error.message);
   goto cleanup;
}
pipeline = BCON_NEW ("pipeline",
                     "[",
                     "{",
                     "$match",
                     "{",
                     "$expr",
                     "{",
                     "$gt",
                     "[",
                     "$saleDate",
                     "{",
                     "$dateSubtract",
                     "{",
                     "startDate",
                     "$$NOW",
                     "unit",
                     BCON_UTF8 ("day"),
                     "amount",
                     BCON_INT64 (1),
                     "}",
                     "}",
                     "]",
                     "}",
                     "}",
                     "}",
                     "{",
                     "$count",
                     BCON_UTF8 ("totalDailySales"),
                     "}",
                     "]");
cursor = mongoc_collection_aggregate (sales_collection, MONGOC_QUERY_NONE, pipeline, &opts, NULL);
bson_destroy (&opts);
ok = mongoc_cursor_next (cursor, &doc);
if (mongoc_cursor_error (cursor, &error)) {
   MONGOC_ERROR ("could not get totalDailySales: %s", error.message);
   goto cleanup;
}
if (!ok) {
   MONGOC_ERROR ("%s", "cursor has no results");
   goto cleanup;
}
ok = bson_iter_init_find (&iter, doc, "totalDailySales");
if (ok) {
   total_sales = bson_iter_as_int64 (&iter);
} else {
   MONGOC_ERROR ("%s", "missing key: 'totalDailySales'");
   goto cleanup;
}

ctx := context.TODO()
sess, err := client.StartSession(options.Session().SetSnapshot(true))
if err != nil {
	return err
}
defer sess.EndSession(ctx)
var totalDailySales int32
err = mongo.WithSession(ctx, sess, func(ctx context.Context) error {
	// Count the total daily sales
	const totalDailySalesOutput = "totalDailySales"
	cursor, err := db.Collection("sales").Aggregate(ctx, mongo.Pipeline{
		bson.D{{"$match",
			bson.D{{"$expr",
				bson.D{{"$gt",
					bson.A{"$saleDate",
						bson.D{{"$dateSubtract",
							bson.D{
								{"startDate", "$$NOW"},
								{"unit", "day"},
								{"amount", 1},
							},
						}},
					},
				}},
			}},
		}},
		bson.D{{"$count", totalDailySalesOutput}},
	})
	if err != nil {
		return err
	}
	if !cursor.Next(ctx) {
		return fmt.Errorf("expected aggregate to return a document, but got none")
	}
	resp := cursor.Current.Lookup(totalDailySalesOutput)
	var ok bool
	totalDailySales, ok = resp.Int32OK()
	if !ok {
		return fmt.Errorf("failed to find int32 field %q in document %v", totalDailySalesOutput, cursor.Current)
	}
	return nil
})
if err != nil {
	return err
}

db = client.retail
async with await client.start_session(snapshot=True) as s:
    docs = await db.sales.aggregate(
        [
            {
                "$match": {
                    "$expr": {
                        "$gt": [
                            "$saleDate",
                            {
                                "$dateSubtract": {
                                    "startDate": "$$NOW",
                                    "unit": "day",
                                    "amount": 1,
                                }
                            },
                        ]
                    }
                }
            },
            {"$count": "totalDailySales"},
        ],
        session=s,
    ).to_list(None)
    total = docs[0]["totalDailySales"]
    print(total)

$salesCollection = $client->selectCollection('retail', 'sales');
$session = $client->startSession(['snapshot' => true]);
$totalDailySales = $salesCollection->aggregate(
    [
        [
            '$match' => [
                '$expr' => [
                    '$gt' => ['$saleDate', [
                        '$dateSubtract' => [
                            'startDate' => '$$NOW',
                            'unit' => 'day',
                            'amount' => 1,
                        ],
                    ],
                    ],
                ],
            ],
        ],
        ['$count' => 'totalDailySales'],
    ],
    ['session' => $session],
)->toArray()[0]->totalDailySales;

db = client.retail
with client.start_session(snapshot=True) as s:
    _ = (
        (
            db.sales.aggregate(
                [
                    {
                        "$match": {
                            "$expr": {
                                "$gt": [
                                    "$saleDate",
                                    {
                                        "$dateSubtract": {
                                            "startDate": "$$NOW",
                                            "unit": "day",
                                            "amount": 1,
                                        }
                                    },
                                ]
                            }
                        }
                    },
                    {"$count": "totalDailySales"},
                ],
                session=s,
            )
        ).next()
    )["totalDailySales"]

A query anterior:

Utiliza $match com $expr para especificar um filtro no campo saleDate .
- $expr permite o uso de expressões de agregação (como NOW) no estágio $match .
Usa o operador $gt e a expressão $dateSubtract para retornar documentos em que saleDate é maior que um dia antes da hora em que a query é executada.
Utiliza $count para retornar uma contagem dos documentos correspondentes. A contagem é armazenada na variável totalDailySales .
Especifica a preocupação de leitura "snapshot" para garantir que a query leia a partir de um único ponto no tempo.

A collection sales é bem grande e, como resultado, essa query pode levar alguns minutos para ser executada. Como a loja é online, as vendas podem ocorrer a qualquer hora do dia.

Por exemplo, considere se:

A query começa a ser executada em 12:00 AM.
Um cliente compra três pares de sapatos em 12:02 AM.
A query termina a execução em 12:04 AM.

Se a consulta não usar preocupação de leitura "snapshot", as vendas que ocorrem entre o início e o término da query podem ser incluídas na contagem de queries, apesar de não ocorrerem no dia do relatório. Isso pode resultar em relatórios imprecisos, com algumas vendas sendo contadas duas vezes.

Ao especificar a preocupação de leitura "snapshot", a query retorna apenas dados que estavam presentes no banco de banco de dados em um ponto no tempo pouco antes de a query começar a ser executada.

Observação

Se a query demorar mais do que o período de retenção do histórico do WiredTiger (300 segundos, por padrão), a query apresenta um erro SnapshotTooOld . Para saber como configurar a retenção de snapshots e habilitar queries de longa duração, consulte Configurar retenção de snapshots.

Configurar retenção de snapshot

Por padrão, o mecanismo WiredTiger storage engine mantém o histórico por 300 segundos. Você pode usar uma sessão com snapshot=true por um total de 300 segundos, desde o momento da primeira operação na sessão até a última. Se você usar a sessão por um longo período de tempo, a sessão falhará com um erro SnapshotTooOld . Da mesma forma, se você executar query de dados usando read concern "snapshot" e sua query durar mais de 300 segundos, a query falhará.

Se sua query ou sessão for executada por mais de 300 segundos, considere aumentar o período de retenção de snapshots. Para aumentar o período de retenção, modifique o parâmetro minSnapshotHistoryWindowInSeconds .

Por exemplo, este comando define o valor de minSnapshotHistoryWindowInSeconds para 600 segundos:

db.adminCommand( { setParameter: 1, minSnapshotHistoryWindowInSeconds: 600 } )

Importante

Para modificar minSnapshotHistoryWindowInSeconds para um cluster MongoDB Atlas , você deve entrar em contato com o suporte do Atlas.

Espaço em disco e histórico

Aumentar o valor de minSnapshotHistoryWindowInSeconds aumenta o uso do disco porque o servidor precisa manter o histórico dos valores modificados mais antigos dentro da janela de tempo especificada. A quantidade de espaço em disco usada depende da sua carga de trabalho, com cargas de trabalho de maior volume exigindo mais espaço em disco.

Voltar

Tempo limite

Iteração de um cursor no mongosh