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
|
/**
* Test that time-series bucket collections work as expected with $lookup.
*
* @tags: [
* does_not_support_transactions,
* requires_timeseries,
* requires_fcv_51,
* ]
*/
(function() {
"use strict";
load("jstests/core/timeseries/libs/timeseries.js");
load("jstests/libs/sbe_util.js"); // For checkSBEEnabled.
TimeseriesTest.run((insert) => {
const testDB = db.getSiblingDB(jsTestName());
assert.commandWorked(testDB.dropDatabase());
const timeFieldName = "time";
const tagFieldName = "tag";
const collOptions = [null, {timeseries: {timeField: timeFieldName, metaField: tagFieldName}}];
const numHosts = 10;
const numDocs = 200;
Random.setRandomSeed();
const hosts = TimeseriesTest.generateHosts(numHosts);
let testFunc = function(collAOption, collBOption) {
// Prepare two time-series collections.
const collA = testDB.getCollection("a");
const collB = testDB.getCollection("b");
collA.drop();
collB.drop();
assert.commandWorked(testDB.createCollection(collA.getName(), collAOption));
assert.commandWorked(testDB.createCollection(collB.getName(), collBOption));
let entryCountPerHost = new Array(numHosts).fill(0);
// Insert into collA, one entry per host.
for (let i = 0; i < numHosts; i++) {
let host = hosts[i];
assert.commandWorked(insert(collA, {
measurement: "cpu",
time: ISODate(),
tags: host.tags,
}));
}
// Insert some random documents to collB.
for (let i = 0; i < numDocs; i++) {
let host = TimeseriesTest.getRandomElem(hosts);
assert.commandWorked(insert(collB, {
measurement: "cpu",
time: ISODate(),
tags: host.tags,
}));
// Here we extract the hostId from "host.tags.hostname". It is expected that the
// "host.tags.hostname" is in the form of 'host_<hostNum>'.
entryCountPerHost[parseInt(
host.tags.hostname.substring(5, host.tags.hostname.length))]++;
}
// Equality Match
let results = collA.aggregate([
{
$lookup: {
from: collB.getName(),
localField: "tags.hostname",
foreignField: "tags.hostname",
as: "matchedB"
}
}, {
$project: {
_id: 0,
host: "$tags.hostname",
matchedB: {
$size: "$matchedB"
}
}
},
{$sort: {host: 1}}
]).toArray();
assert.eq(numHosts, results.length, results);
for (let i = 0; i < numHosts; i++) {
assert.eq({host: "host_" + i, matchedB: entryCountPerHost[i]}, results[i], results);
}
// Unequal joins
results = collA.aggregate([
{
$lookup: {
from: collB.getName(),
let: {
hostId: {$toInt: {$substr: ["$tags.hostname", 5, -1]}}
},
pipeline: [
{
$match: {
$expr: {
$lte: [{$toInt: {$substr: ["$tags.hostname", 5, -1]}}, "$$hostId"]
}
},
}
],
as: "matchedB"
}
}, {
$project: {
_id: 0,
host: "$tags.hostname",
matchedB: {
$size: "$matchedB"
}
}
},
{$sort: {host: 1}}
]).toArray();
assert.eq(numHosts, results.length, results);
let expectedCount = 0;
for (let i = 0; i < numHosts; i++) {
expectedCount += entryCountPerHost[i];
assert.eq({host: "host_" + i, matchedB: expectedCount}, results[i], entryCountPerHost);
}
};
// Exhaust the combinations of non-time-series and time-series collections for inner and outer
// $lookup collections.
collOptions.forEach(function(collAOption) {
collOptions.forEach(function(collBOption) {
if (collAOption == null && collBOption == null) {
// Normal $lookup call, both inner and outer collections are non-time-series
// collections.
return;
}
testFunc(collAOption, collBOption);
});
});
});
})();
|
