SERVER-72687 support $out to time-series collections

2 files changed, 120 insertions, 16 deletions

diff --git a/jstests/core/timeseries/libs/timeseries_agg_helpers.js b/jstests/core/timeseries/libs/timeseries_agg_helpers.js
index a2a2f74393f..becad2575ae 100644
--- a/jstests/core/timeseries/libs/timeseries_agg_helpers.js
+++ b/jstests/core/timeseries/libs/timeseries_agg_helpers.js
@@ -78,11 +78,13 @@ var TimeseriesAggTests = class {
     /**
      * Gets an output collection object with the name 'outCollname'.
      */
-    static getOutputCollection(outCollName) {
+    static getOutputCollection(outCollName, shouldDrop) {
         const testDB = TimeseriesAggTests.getTestDb();
 
         let outColl = testDB.getCollection(outCollName);
-        outColl.drop();
+        if (shouldDrop) {
+            outColl.drop();
+        }
 
         return outColl;
     }
@@ -96,21 +98,29 @@ var TimeseriesAggTests = class {
      * Executes 'prepareAction' before executing 'pipeline'. 'prepareAction' takes a collection
      * parameter and returns nothing.
      *
+     * If 'shouldDrop' is set to false, the output collection will not be dropped before executing
+     * 'pipeline'.
+     *
      * Returns sorted data by "time" field. The sorted result data will help simplify comparison
      * logic.
      */
-    static getOutputAggregateResults(inColl, pipeline, prepareAction = null) {
+    static getOutputAggregateResults(inColl, pipeline, prepareAction = null, shouldDrop = true) {
         // Figures out the output collection name from the last pipeline stage.
         var outCollName = "out";
         if (pipeline[pipeline.length - 1]["$out"] != undefined) {
-            // If the last stage is "$out", gets the output collection name from it.
-            outCollName = pipeline[pipeline.length - 1]["$out"];
+            // If the last stage is "$out", gets the output collection name from the string or
+            // object input.
+            if (typeof pipeline[pipeline.length - 1]["$out"] == 'string') {
+                outCollName = pipeline[pipeline.length - 1]["$out"];
+            } else {
+                outCollName = pipeline[pipeline.length - 1]["$out"]["coll"];
+            }
         } else if (pipeline[pipeline.length - 1]["$merge"] != undefined) {
             // If the last stage is "$merge", gets the output collection name from it.
             outCollName = pipeline[pipeline.length - 1]["$merge"].into;
         }
 
-        let outColl = TimeseriesAggTests.getOutputCollection(outCollName);
+        let outColl = TimeseriesAggTests.getOutputCollection(outCollName, shouldDrop);
         if (prepareAction != null) {
             prepareAction(outColl);
         }
@@ -122,4 +132,14 @@ var TimeseriesAggTests = class {
             .sort({"time": 1})
             .toArray();
     }
+
+    static verifyResults(actualResults, expectedResults) {
+        // Verifies that the number of measurements is same as expected.
+        assert.eq(actualResults.length, expectedResults.length, actualResults);
+
+        // Verifies that every measurement is same as expected.
+        for (var i = 0; i < expectedResults.length; ++i) {
+            assert.eq(actualResults[i], expectedResults[i], actualResults);
+        }
+    }
 };
diff --git a/jstests/core/timeseries/timeseries_out.js b/jstests/core/timeseries/timeseries_out.js
index 46beccd99ed..a75b9a26b35 100644
--- a/jstests/core/timeseries/timeseries_out.js
+++ b/jstests/core/timeseries/timeseries_out.js
@@ -7,6 +7,11 @@
  *   does_not_support_stepdowns,
  *   # We need a timeseries collection.
  *   requires_timeseries,
+ *   # TODO SERVER-74601 remove tag after support for secondaries.
+ *   assumes_read_preference_unchanged,
+ *   # TODO SERVER-74601 remove tag after support for sharded clusters.
+ *   assumes_against_mongod_not_mongos,
+ *   requires_fcv_71
  * ]
  */
 (function() {
@@ -21,18 +26,97 @@ const numIterations = 20;
 
 let [inColl, observerInColl] = TimeseriesAggTests.prepareInputCollections(numHosts, numIterations);
 
-// Gets the expected results from non time-series observer input collection.
-let expectedResults =
-    TimeseriesAggTests.getOutputAggregateResults(observerInColl, [{$out: "observer_out"}]);
+function generateOutPipeline(collName, options, aggStage = null) {
+    let outStage = {$out: {db: testDB.getName(), coll: collName, timeseries: options}};
+    if (aggStage) {
+        return [aggStage, outStage];
+    }
+    return [outStage];
+}
 
-// Gets the actual results from time-series input collection.
-let actualResults = TimeseriesAggTests.getOutputAggregateResults(inColl, [{$out: "out"}]);
+function runTest(observerPipeline, actualPipeline, shouldDrop = true, valueToCheck = null) {
+    // Gets the expected results from a non time-series observer input collection.
+    const expectedResults = TimeseriesAggTests.getOutputAggregateResults(
+        observerInColl, observerPipeline, null, shouldDrop);
 
-// Verifies that the number of measurements is same as expected.
-assert.eq(actualResults.length, expectedResults.length, actualResults);
+    // Gets the actual results from a time-series input collection.
+    const actualResults =
+        TimeseriesAggTests.getOutputAggregateResults(inColl, actualPipeline, null, shouldDrop);
 
-// Verifies that every measurement is same as expected.
-for (var i = 0; i < expectedResults.length; ++i) {
-    assert.eq(actualResults[i], expectedResults[i], actualResults);
+    // Verifies that the number of measurements is same as expected.
+    TimeseriesAggTests.verifyResults(actualResults, expectedResults);
+    if (valueToCheck) {
+        for (var i = 0; i < expectedResults.length; ++i) {
+            assert.eq(actualResults[i], {"time": valueToCheck}, actualResults);
+        }
+    }
 }
+
+//  Tests that $out works with time-series collections writing to a non-timeseries collection.
+runTest([{$out: "observer_out"}], [{$out: "out"}]);
+
+//  Tests that $out creates a time-series collection when the collection does not exist.
+let actualPipeline = generateOutPipeline("out_time", {timeField: "time", metaField: "tags"});
+runTest([{$out: "observer_out"}], actualPipeline);
+
+// Tests that $out creates a time-series collection with more time-series options.
+actualPipeline = generateOutPipeline(
+    "out_time",
+    {timeField: "time", metaField: "tags", bucketMaxSpanSeconds: 100, bucketRoundingSeconds: 100});
+runTest([{$out: "observer_out"}], actualPipeline);
+
+// Change an option in the existing time-series collections.
+assert.commandWorked(testDB.runCommand({collMod: "out_time", expireAfterSeconds: 360}));
+assert.commandWorked(testDB.runCommand({collMod: "observer_out", validationLevel: "moderate"}));
+
+// Tests that a time-series collection is replaced when a time-series collection does exist.
+let newDate = new Date('1999-09-30T03:24:00');
+let observerPipeline = [{$set: {"time": newDate}}, {$out: "observer_out"}];
+actualPipeline = generateOutPipeline("out_time", {timeField: "time"}, {$set: {"time": newDate}});
+
+// Confirms that all the documents have the 'newDate' value.
+runTest(observerPipeline, actualPipeline, false, newDate);
+
+// Confirms that the original time-series collection options were preserved by $out.
+let collections = assert.commandWorked(testDB.runCommand({listCollections: 1})).cursor.firstBatch;
+let coll = collections.find(entry => entry.name === "out_time");
+assert.eq(coll["options"]["expireAfterSeconds"], 360);
+coll = collections.find(entry => entry.name === "observer_out");
+assert.eq(coll["options"]["validationLevel"], "moderate");
+
+// Tests that an error is raised when trying to create a time-series collection from a non
+// time-series collection.
+let pipeline = generateOutPipeline("observer_out", {timeField: "time"});
+assert.throwsWithCode(() => inColl.aggregate(pipeline), 7268700);
+assert.throwsWithCode(() => observerInColl.aggregate(pipeline), 7268700);
+
+// Tests that an error is raised for invalid timeseries options.
+pipeline = generateOutPipeline("out_time", {timeField: "time", invalidField: "invalid"});
+assert.throwsWithCode(() => inColl.aggregate(pipeline), 40415);
+assert.throwsWithCode(() => observerInColl.aggregate(pipeline), 40415);
+
+// Tests that an error is raised if the time-series specification changes.
+pipeline = generateOutPipeline("out_time", {timeField: "usage_guest_nice"});
+assert.throwsWithCode(() => inColl.aggregate(pipeline), 7268701);
+assert.throwsWithCode(() => observerInColl.aggregate(pipeline), 7268701);
+
+pipeline = generateOutPipeline("out_time", {timeField: "time", metaField: "usage_guest_nice"});
+assert.throwsWithCode(() => inColl.aggregate(pipeline), 7268702);
+assert.throwsWithCode(() => observerInColl.aggregate(pipeline), 7268702);
+
+// Tests that a time-series collection can be replaced with a non-timeseries collection.
+runTest([{"$out": "observer_out_time"}], [{"$out": "out_time"}]);
+
+// Tests that an error is raised if a conflicting view exists.
+assert.commandWorked(testDB.createCollection("view_out", {viewOn: "out"}));
+
+pipeline = generateOutPipeline("view_out", {timeField: "time"});
+assert.throwsWithCode(() => inColl.aggregate(pipeline), 7268703);
+assert.throwsWithCode(() => observerInColl.aggregate(pipeline), 7268703);
+
+// Test $out for time-series works with a non-existent database.
+const destDB = testDB.getSiblingDB("outDifferentDB");
+assert.commandWorked(destDB.dropDatabase());
+inColl.aggregate({$out: {db: destDB.getName(), coll: "out_time", timeseries: {timeField: "time"}}});
+assert.eq(300, destDB["out_time"].find().itcount());
 })();