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
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
|
/**
* Tests the optimization of "lastpoint"-type queries on time-series collections.
*
* @tags: [
* does_not_support_stepdowns,
* does_not_support_transactions,
* requires_timeseries,
* requires_pipeline_optimization,
* requires_fcv_53,
* # TODO (SERVER-63590): Investigate presence of getmore tag in timeseries jstests.
* requires_getmore
* ]
*/
(function() {
"use strict";
load("jstests/aggregation/extras/utils.js");
load("jstests/core/timeseries/libs/timeseries_agg_helpers.js");
load('jstests/libs/analyze_plan.js');
const testDB = TimeseriesAggTests.getTestDb();
assert.commandWorked(testDB.dropDatabase());
// Do not run the rest of the tests if the lastpoint optimization is disabled.
const getLastpointParam = db.adminCommand({getParameter: 1, featureFlagLastPointQuery: 1});
const isLastpointEnabled = getLastpointParam.hasOwnProperty("featureFlagLastPointQuery") &&
getLastpointParam.featureFlagLastPointQuery.value;
if (!isLastpointEnabled) {
return;
}
// Timeseries test parameters.
const numHosts = 10;
const numIterations = 20;
function verifyTsResults({pipeline, precedingFilter, expectStage, prepareTest}) {
// Prepare collections. Note: we test without idle measurements (all meta subfields are
// non-null). If we allow the insertion of idle measurements, we will obtain multiple lastpoints
// per bucket, and may have different results on the observer and timeseries collections.
const [tsColl, observerColl] = TimeseriesAggTests.prepareInputCollections(
numHosts, numIterations, false /* includeIdleMeasurements */);
// Additional preparation before running the test.
if (prepareTest) {
prepareTest(tsColl, observerColl);
}
// Verify lastpoint optmization.
const explain = tsColl.explain().aggregate(pipeline);
expectStage({explain, precedingFilter});
// Assert that the time-series aggregation results match that of the observer collection.
const expected = observerColl.aggregate(pipeline).toArray();
const actual = tsColl.aggregate(pipeline).toArray();
assertArrayEq({actual, expected});
// Drop collections.
tsColl.drop();
observerColl.drop();
}
function verifyTsResultsWithAndWithoutIndex(
{pipeline, index, bucketsIndex, precedingFilter, expectStage, prePrepareTest}) {
verifyTsResults(
{pipeline, precedingFilter, expectStage: expectCollScan, prepareTest: prePrepareTest});
verifyTsResults({
pipeline,
precedingFilter,
expectStage,
prepareTest: (testColl, observerColl) => {
// Optionally do extra test preparation.
if (prePrepareTest) {
prePrepareTest(testColl, observerColl);
}
// Create index on the timeseries collection.
testColl.createIndex(index);
// Create an additional secondary index directly on the buckets collection so that we
// can test the DISTINCT_SCAN optimization when time is sorted in ascending order.
if (bucketsIndex) {
const bucketsColl = testDB["system.buckets.in"];
bucketsColl.createIndex(bucketsIndex);
}
}
});
}
function expectDistinctScan({explain}) {
// The query can utilize DISTINCT_SCAN.
assert.neq(getAggPlanStage(explain, "DISTINCT_SCAN"), null, explain);
// Pipelines that use the DISTINCT_SCAN optimization should not also have a blocking sort.
assert.eq(getAggPlanStage(explain, "SORT"), null, explain);
}
function expectCollScan({explain, precedingFilter}) {
// $sort can be pushed into the cursor layer.
assert.neq(getAggPlanStage(explain, "SORT"), null, explain);
// At the bottom, there should be a COLLSCAN.
const collScanStage = getAggPlanStage(explain, "COLLSCAN");
assert.neq(collScanStage, null, explain);
if (precedingFilter) {
assert.eq(precedingFilter, collScanStage.filter, collScanStage);
}
}
function expectIxscan({explain}) {
// $sort can be pushed into the cursor layer.
assert.neq(getAggPlanStage(explain, "SORT"), null, explain);
// At the bottom, there should be a IXSCAN.
assert.neq(getAggPlanStage(explain, "IXSCAN"), null, explain);
}
function getGroupStage(accumulator) {
return {
$group: {
_id: "$tags.hostid",
usage_user: {[accumulator]: "$usage_user"},
usage_guest: {[accumulator]: "$usage_guest"},
usage_idle: {[accumulator]: "$usage_idle"}
}
};
}
/**
Test cases:
1. Lastpoint queries on indexes with descending time and $first (DISTINCT_SCAN).
2. Lastpoint queries on indexes with ascending time and $last (no DISTINCT_SCAN).
3. Lastpoint queries on indexes with ascending time and $last and an additional secondary
index so that we can use the DISTINCT_SCAN optimization.
*/
const testCases = [
{time: -1},
{time: 1},
{time: -1, bucketsIndex: {"meta.hostid": -1, "control.max.time": 1, "control.min.time": 1}}
];
for (const {time, bucketsIndex} of testCases) {
const isTimeDescending = time < 0;
const canUseDistinct = isTimeDescending || bucketsIndex;
const groupStage = isTimeDescending ? getGroupStage("$first") : getGroupStage("$last");
// Test both directions of the metaField sort for each direction of time.
for (const index of [{"tags.hostid": 1, time}, {"tags.hostid": -1, time}]) {
// Test pipeline without a preceding $match stage.
verifyTsResultsWithAndWithoutIndex({
pipeline: [{$sort: index}, groupStage],
index,
bucketsIndex,
expectStage: (canUseDistinct ? expectDistinctScan : expectCollScan)
});
// Test pipeline without a preceding $match stage which has an extra idle measurement. This
// verifies that the query rewrite correctly returns missing fields.
verifyTsResultsWithAndWithoutIndex({
pipeline: [{$sort: index}, groupStage],
index,
bucketsIndex,
expectStage: (canUseDistinct ? expectDistinctScan : expectCollScan),
prePrepareTest: (testColl, observerColl) => {
const currTime = new Date();
for (const host of TimeseriesTest.generateHosts(numHosts)) {
const idleMeasurement = {
tags: host.tags,
time: new Date(currTime + numIterations), // Ensure this is the lastpoint.
idle_user: 100 - TimeseriesTest.getRandomUsage()
};
assert.commandWorked(testColl.insert(idleMeasurement));
assert.commandWorked(observerColl.insert(idleMeasurement));
}
}
});
// Test pipeline with a preceding $match stage.
function testWithMatch(matchStage, precedingFilter) {
verifyTsResultsWithAndWithoutIndex({
pipeline: [matchStage, {$sort: index}, groupStage],
index,
bucketsIndex,
expectStage: canUseDistinct ? expectDistinctScan : expectIxscan,
precedingFilter
});
}
// Test pipeline with an equality $match stage.
testWithMatch({$match: {"tags.hostid": 0}}, {"meta.hostid": {$eq: 0}});
// Test pipeline with an inequality $match stage.
testWithMatch({$match: {"tags.hostid": {$ne: 0}}}, {"meta.hostid": {$not: {$eq: 0}}});
// Test pipeline with a $match stage that uses a $gt query.
testWithMatch({$match: {"tags.hostid": {$gt: 5}}}, {"meta.hostid": {$gt: 5}});
// Test pipeline with a $match stage that uses a $lt query.
testWithMatch({$match: {"tags.hostid": {$lt: 5}}}, {"meta.hostid": {$lt: 5}});
}
}
})();
|
