1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
|
// Tests that aggregation is supported in transactions.
// @tags: [uses_transactions, uses_snapshot_read_concern]
(function() {
"use strict";
load("jstests/libs/fixture_helpers.js"); // For isSharded.
// TODO (SERVER-39704): Remove the following load after SERVER-397074 is completed
// For withTxnAndAutoRetryOnMongos.
load('jstests/libs/auto_retry_transaction_in_sharding.js');
const session = db.getMongo().startSession({causalConsistency: false});
const testDB = session.getDatabase("test");
const coll = testDB.getCollection("aggregation_in_transaction");
const foreignColl = testDB.getCollection("aggregation_in_transaction_lookup");
[coll, foreignColl].forEach(col => {
const reply = col.runCommand("drop", {writeConcern: {w: "majority"}});
if (reply.ok !== 1) {
assert.commandFailedWithCode(reply, ErrorCodes.NamespaceNotFound);
}
});
// Populate the collections.
const testDoc = {
_id: 0,
foreignKey: "orange"
};
assert.commandWorked(coll.insert(testDoc, {writeConcern: {w: "majority"}}));
const foreignDoc = {
_id: "orange",
val: 9
};
assert.commandWorked(foreignColl.insert(foreignDoc, {writeConcern: {w: "majority"}}));
const isForeignSharded = FixtureHelpers.isSharded(foreignColl);
const txnOptions = {
readConcern: {level: "snapshot"}
};
// TODO (SERVER-39704): We use the withTxnAndAutoRetryOnMongos
// function to handle how MongoS will propagate a StaleShardVersion error as a
// TransientTransactionError. After SERVER-39704 is completed the
// withTxnAndAutoRetryOnMongos function can be removed
withTxnAndAutoRetryOnMongos(session, () => {
// Cleaning collection in case the transaction is retried
db.getSiblingDB(testDB.getName()).getCollection(coll.getName()).remove({
_id: "not_visible_in_transaction"
});
// Run a dummy find to start the transaction.
jsTestLog("Transaction started.");
let cursor = coll.find();
cursor.next();
// Insert a document outside of the transaction. Subsequent aggregations should not see this
// document.
jsTestLog("Inserting document outside of transaction.");
assert.commandWorked(db.getSiblingDB(testDB.getName()).getCollection(coll.getName()).insert({
_id: "not_visible_in_transaction",
foreignKey: "orange",
}));
// Perform an aggregation that is fed by a cursor on the underlying collection. Only the
// majority-committed document present at the start of the transaction should be found.
jsTestLog("Starting aggregations inside of the transaction.");
cursor = coll.aggregate({$match: {}});
assert.docEq(testDoc, cursor.next());
assert(!cursor.hasNext());
// Perform aggregations that look at other collections.
// TODO: SERVER-39162 Sharded $lookup is not supported in transactions.
if (!isForeignSharded) {
jsTestLog("Testing $lookup within a transaction.");
const lookupDoc = Object.merge(testDoc, {lookup: [foreignDoc]});
cursor = coll.aggregate({
$lookup: {
from: foreignColl.getName(),
localField: "foreignKey",
foreignField: "_id",
as: "lookup",
}
});
assert.docEq(cursor.next(), lookupDoc);
assert(!cursor.hasNext());
jsTestLog("Testing $graphLookup within a transaction.");
cursor = coll.aggregate({
$graphLookup: {
from: foreignColl.getName(),
startWith: "$foreignKey",
connectFromField: "foreignKey",
connectToField: "_id",
as: "lookup"
}
});
assert.docEq(cursor.next(), lookupDoc);
assert(!cursor.hasNext());
}
jsTestLog("Testing $count within a transaction.");
let countRes = coll.aggregate([{$count: "count"}]).toArray();
assert.eq(countRes.length, 1, tojson(countRes));
assert.eq(countRes[0].count, 1, tojson(countRes));
assert.commandWorked(coll.insert({a: 2}));
countRes = coll.aggregate([{$count: "count"}]).toArray();
assert.eq(countRes.length, 1, tojson(countRes));
assert.eq(countRes[0].count, 2, tojson(countRes));
assert.commandWorked(
db.getSiblingDB(testDB.getName()).getCollection(coll.getName()).insert({a: 3}));
countRes = coll.aggregate([{$count: "count"}]).toArray();
assert.eq(countRes.length, 1, tojson(countRes));
assert.eq(countRes[0].count, 2, tojson(countRes));
}, txnOptions);
jsTestLog("Transaction committed.");
// Perform aggregations with non-cursor initial sources and assert that they are not supported
// in transactions.
jsTestLog("Running aggregations in transactions that are expected to throw and fail.");
session.startTransaction({readConcern: {level: "snapshot"}});
assert.throws(() => coll.aggregate({$currentOp: {allUsers: true, localOps: true}}).next());
assert.commandFailedWithCode(session.abortTransaction_forTesting(), ErrorCodes.NoSuchTransaction);
session.startTransaction({readConcern: {level: "snapshot"}});
assert.throws(
() =>
coll.aggregate({$collStats: {latencyStats: {histograms: true}, storageStats: {}}}).next());
assert.commandFailedWithCode(session.abortTransaction_forTesting(), ErrorCodes.NoSuchTransaction);
session.startTransaction({readConcern: {level: "snapshot"}});
assert.throws(() => coll.aggregate({$indexStats: {}}).next());
assert.commandFailedWithCode(session.abortTransaction_forTesting(), ErrorCodes.NoSuchTransaction);
}());
|
