MongoDB Get every Nth element of array

Question

I've been working on a small project that takes MQTT data from sensors and stores it in a MongoDB database. I'm working with nodeJS and mongoose. These are my schemas.

export const SensorSchema = new mongoose.Schema({
    name: { type: String, required: true, unique: true },
    location: { type: String, required: true },
    type: { type: String, required: true },
    unit: { type: String, required: true },
    measurements: { type: [MeasurementSchema] }
},
{ 
    toObject: { virtuals: true },
    toJSON: { virtuals: true }
});

export const MeasurementSchema = new mongoose.Schema({
    value: {type: Number, required: true},
    time: {type: Date, required: true}
});

First I wrote a function that retrieves all measurements that were made in between two timestamps.

const values = Sensor.aggregate([
                { $match: Sensor.getValuesFromPath(sensorPath) },
                { $unwind: "$measurements"},
                { $match: { "measurements.time": { $gte: startTime,  $lte: endTime} }},
                { $replaceRoot: { newRoot: "$measurements" } },
                { $project: { _id: 0}},
                { $sort: {time: 1}}
            ]).exec();

In order to draw a graph in the UI, I need to somehow sort and then limit the data that gets sent to the client. I want to send every Nth Value in a certain interval to ensure that the data somewhat resembles the course of the data. I would prefer a solution that doesn't fetch all the data from the database.

How would I go about doing this on the db? Can I somehow access the positional index of an element after sorting it? Is $arrayElemAt or $elemMatch the solution?

Answer 1

The following aggregation (i) retrieves all measurements that were made in between two timestamps, (ii) sorts by timestamp for each sensor, and (iii) gets every Nth value (specified by the variable EVERY_N).

Sample documents (with some arbitrary data for testing):

{
  name: "s-1",
  location: "123",
  type: "456",
  measurements: [ { time: 2, value: 12 }, { time: 3, value: 13 }, 
                  { time: 4, value: 15 }, { time: 5, value: 22 }, 
                  { time: 6, value: 34 }, { time: 7, value: 9 }, 
                  { time: 8, value: 5 }, { time: 9, value: 1 }, 
  ]
},
{
  name: "s-2",
  location: "789",
  type: "900",
  measurements: [ { time: 1, value: 31 }, { time: 3, value: 32 }, 
                  { time: 4, value: 35 }, { time: 6, value: 39 },
                  { time: 7, value: 6}, { time: 8, value: 70 }, 
                  { time: 9, value: 74 }, { time: 10, value: 82 }
  ]
}

The aggregation:

var startTime = 3, endTime = 10
var EVERY_N = 2   // value can be 3, etc.

db.collection.aggregate( [
  { 
      $unwind: "$measurements" 
  },
  { 
      $match: { 
          "measurements.time": { $gte: startTime, $lte: endTime } 
      } 
  },
  { 
      $sort: { name: 1, "measurements.time": 1 } 
  },
  { 
      $group: { 
          _id: "$name", 
          measurements: { $push: "$measurements" }, 
          doc: { $first: "$$ROOT" } 
      } 
  },
  { 
      $addFields: { 
          "doc.measurements": "$measurements" 
      } 
  },
  { 
      $replaceRoot: { newRoot: "$doc" } 
  },
  { 
      $addFields: { 
           measurements: { 
               $reduce: {
                    input: { $range: [ 0, { $size: "$measurements" } ] }, 
                    initialValue: [ ],
                    in: { $cond: [ { $eq: [ { $mod: [ "$$this", EVERY_N ] }, 0 ] }, 
                                   { $concatArrays: [ "$$value", [ { $arrayElemAt: [ "$measurements", "$$this" ] } ] ] }, 
                                   "$$value" 
                             ]
                    }
               }
           }
      } 
  }
] )

Answer 2

Befure you run $unwind you can use $filter to apply start/end Date filtering. This will allow you to process measurements as an array. In the next step you can get every N-th element by using $range to define a list of indexes and $arrayElemAt to retrieve elements from these indexes:

const values = Sensor.aggregate([
                { $match: Sensor.getValuesFromPath(sensorPath) },                
                { $addFields: { 
                    measurements: { 
                        $filter: { 
                            input: "$measurements", 
                            cond: { $and: [ 
                                    { $gte: [ "$$this.time", startTime ] }, 
                                    { $lte: [ "$$this.time", endTime ] } 
                                ] 
                            } 
                        } 
                    } 
                } },
                { $addFields: { 
                    measurements: { 
                        $map: { 
                            input: input: { $range: [ 0, { $size: "$measurements" }, N ] }, 
                            as: "index",
                            in: { $arrayElemAt: [ "$measurements", "$$index" ] }
                        } 
                    } 
                } },
                { $unwind: "$measurements" },
                { $replaceRoot: { newRoot: "$measurements" } },
                { $project: { _id: 0}},
                { $sort: {time: 1}}
            ]).exec();