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
|
'use strict';
load("jstests/libs/fixture_helpers.js"); // For isSharded.
/**
* agg_lookup.js
*
* Runs a $lookup aggregation simultaneously with updates.
*/
var $config = (function() {
const data = {numDocs: 100};
const isShardedAndShardedLookupDisabled = false;
const states = (function() {
function query(db, collName) {
if (this.isShardedAndShardedLookupDisabled) {
return;
}
// Run the aggregate with 'allowDiskUse' if it was configured during setup.
const aggOptions = {allowDiskUse: this.allowDiskUse};
function getQueryResults() {
let arr = null;
try {
const cursor = db[collName]
.aggregate([
{
$lookup: {
from: collName,
localField: "_id",
foreignField: "to",
as: "out",
}
},
], aggOptions);
arr = cursor.toArray();
} catch (e) {
if (TestData.runningWithShardStepdowns) {
// When running with stepdowns, we expect to sometimes see the query
// killed.
assert.eq(e.code, ErrorCodes.QueryPlanKilled);
} else {
throw e;
}
}
return arr;
}
const res = getQueryResults();
if (res) {
assertWhenOwnColl.eq(res.length, data.numDocs);
}
}
function update(db, collName) {
const index = Random.randInt(this.numDocs + 1);
const update = Random.randInt(this.numDocs + 1);
const res = db[collName].update({_id: index}, {$set: {to: update}});
assertWhenOwnColl.commandWorked(res);
}
return {query, update};
})();
const transitions = {query: {query: 0.5, update: 0.5}, update: {query: 0.5, update: 0.5}};
function setup(db, collName, cluster) {
// Do not run the rest of the tests if the foreign collection is implicitly sharded but the
// flag to allow $lookup into a sharded collection is disabled.
const getParam = db.adminCommand({
getParameter: 1,
featureFlagShardedLookup: 1,
featureFlagSBELookupPushdown: 1,
internalQueryForceClassicEngine: 1
});
const isShardedLookupEnabled = getParam.hasOwnProperty("featureFlagShardedLookup") &&
getParam.featureFlagShardedLookup.value;
if (FixtureHelpers.isSharded(db[collName]) && !isShardedLookupEnabled) {
jsTestLog(
"Skipping test because the sharded lookup feature flag is disabled and we have sharded collections");
this.isShardedAndShardedLookupDisabled = true;
return;
}
// Load example data.
const bulk = db[collName].initializeUnorderedBulkOp();
for (let i = 0; i < this.numDocs; ++i) {
bulk.insert({_id: i, to: i + 1});
}
const res = bulk.execute();
assertWhenOwnColl.commandWorked(res);
assertWhenOwnColl.eq(this.numDocs, res.nInserted);
assertWhenOwnColl.eq(this.numDocs, db[collName].find().itcount());
const isLookupPushdownEnabled = getParam.hasOwnProperty("featureFlagSBELookupPushdown") &&
getParam.hasOwnProperty("internalQueryForceClassicEngine") &&
getParam.featureFlagSBELookupPushdown.value &&
!getParam.internalQueryForceClassicEngine.value;
this.allowDiskUse = true;
// If $lookup pushdown into SBE is enabled, we select a random join algorithm to use and
// set the collection up accordingly.
if (isLookupPushdownEnabled) {
// Use a random join algorithm on each test run.
const numStrategies = 3;
const strategy = Random.randInt(numStrategies);
if (strategy === 0) {
jsTestLog("Using hash join");
} else if (strategy === 1) {
assertWhenOwnColl.commandWorked(db[collName].createIndex({to: 1}));
jsTestLog("Using index join");
this.allowDiskUse = false;
} else {
jsTestLog("Using nested loop join");
this.allowDiskUse = false;
}
}
}
function teardown(db, collName) {
// Drop indexes, if any were created.
assertWhenOwnColl.commandWorked(db[collName].dropIndexes());
}
return {
threadCount: 10,
iterations: 100,
states: states,
startState: 'query',
transitions: transitions,
data: data,
setup: setup,
teardown: teardown,
};
})();
|
