$merge (aggregation stage)

On-Demand Materialized View: Initial Creation

If the output collection does not exist, the $merge creates the collection.

You can use the $group and $merge stages to create a movieRatingSummary collection that summarizes critically acclaimed films by release year and content rating:

db.movies.aggregate( [
   { $match: { metacritic: 100, rated: { $ne: null },
     year: { $lte: 1972 } } },
   { $group: { _id: { year: "$year", rated: "$rated" },
     count: { $sum: 1 } } },
   { $merge : { into: "movieRatingSummary", on: "_id",
     whenMatched: "replace", whenNotMatched: "insert" } }
] )

The pipeline uses the following stages:

• $match stage to filter for critically acclaimed films released through 1972 with a content rating

• $group stage to group the films by year and rated

• $merge stage to write the output of the preceding $group stage to the movieRatingSummary collection in the sample_mflix database

To view the documents in the new movieRatingSummary collection:

db.movieRatingSummary.find().sort(
   { _id: 1 } )

[
  { _id: { year: 1939, rated: 'PASSED' }, count: 1 },
  { _id: { year: 1962, rated: 'PG' }, count: 1 },
  { _id: { year: 1963, rated: 'PG' }, count: 1 },
  { _id: { year: 1970, rated: 'R' }, count: 1 },
  { _id: { year: 1972, rated: 'R' }, count: 1 }
]

On-Demand Materialized View: Update/Replace Data

To update the movieRatingSummary collection from the previous example to include critically acclaimed films from 1963 onward, this aggregation pipeline uses the following stages:

• $match stage to find all films with metacritic: 100, a content rating, and a release year greater than or equal to 1963.

• $group stage to group the films by year and rated.

• $merge to write the result set to the movieRatingSummary collection, replacing documents with the same _id value. For documents that do not have matches in the collection, $merge inserts the new documents.

db.movies.aggregate( [
   { $match: { metacritic: 100, rated: { $ne: null },
     year: { $gte: 1963 } } },
   { $group: { _id: { year: "$year", rated: "$rated" },
     count: { $sum: 1 } } },
   { $merge : { into: "movieRatingSummary", on: "_id",
     whenMatched: "replace", whenNotMatched: "insert" } }
] )

After the aggregation runs, view the documents in the movieRatingSummary collection:

db.movieRatingSummary.find().sort(
   { _id: 1 } )

[
  { _id: { year: 1939, rated: 'PASSED' }, count: 1 },
  { _id: { year: 1962, rated: 'PG' }, count: 1 },
  { _id: { year: 1963, rated: 'PG' }, count: 1 },
  { _id: { year: 1970, rated: 'R' }, count: 1 },
  { _id: { year: 1972, rated: 'R' }, count: 1 },
  { _id: { year: 1982, rated: 'R' }, count: 1 },
  { _id: { year: 2014, rated: 'R' }, count: 1 }
]

Only Insert New Data

To ensure that the $merge does not overwrite existing data in the collection, set whenMatched to keepExisting or fail.

A collection movieArchive in the sample_mflix database contains historical records of critically acclaimed films per release year.

The movieArchive collection has a unique index on the year field. At most one record should exist per release year:

db.movieArchive.createIndex(
   { year: 1 }, { unique: true } )

This aggregation pipeline updates the movieArchive collection with data from the movies collection to include critically acclaimed films from 1963 onward. The pipeline uses the following stages:

• $match stage to find all films with metacritic: 100, a content rating, and year >= 1963.

• $group stage to group the film titles by year.

• $project stage to suppress the _id field and promote year to a top-level field. When the documents are passed to $merge, $merge automatically generates a new _id field for the documents.

• $merge to write the result set to movieArchive.

  The $merge stage matches documents on the year field and fails when matched. That is, if a document already exists for that release year, the $merge errors.

db.movies.aggregate( [
   { $match: { metacritic: 100, rated: { $ne: null },
     year: { $gte: 1963 } } },
   { $group: { _id: "$year",
     titles: { $push: "$title" } } },
   { $project: { _id: 0, year: "$_id", titles: 1 } },
   { $merge : { into: "movieArchive", on: "year",
     whenMatched: "fail" } }
] )

If the movieArchive collection already contains a document for any year in the range 1963–2014, the aggregation fails because of the duplicate key error. However, the pipeline does not roll back any documents that it inserted before the error.

If you specify keepExisting for the matching document, the aggregation does not affect the matching document and does not error with a duplicate key error. Similarly, if you specify replace, the operation does not fail; however, the operation replaces the existing document.

Merge Results from Multiple Collections

By default, if a document in the aggregation results matches a document in the collection, the $merge stage merges the documents.

You can use $merge to merge results from the movies collection and the comments collection to create a new collection yearlyStats.

To create the yearlyStats collection, run the following pipeline:

db.movies.aggregate( [
   { $match: { metacritic: 100, rated: { $ne: null },
     year: { $gte: 1970, $lte: 1972 } } },
   { $group: { _id: "$year", movieCount: { $sum: 1 } } },
   { $merge : { into: "yearlyStats", on: "_id",
     whenMatched: "merge", whenNotMatched: "insert" } }
])

First stage:

The $match stage filters for critically acclaimed films (metacritic: 100) with a content rating, released between 1970 and 1972.

Second stage:

The $group stage groups by year and counts films into a new movieCount field.

Third stage:

The $merge stage writes the documents to the yearlyStats collection in the same database. If the stage finds an existing document in the collection that matches on the _id field, the stage merges the matching documents. Otherwise, the stage inserts the document. For the initial creation, no documents match.

To view the documents in the collection, run the following operation:

db.yearlyStats.find().sort( { _id: 1 } )

[
  { _id: 1970, movieCount: 1 },
  { _id: 1972, movieCount: 1 }
]

Similarly, run the following aggregation pipeline against the comments collection to merge the comment counts into the yearlyStats collection.

db.comments.aggregate( [
   { $match: { date: { $gte: new Date("1970-01-01"),
     $lt: new Date("1973-01-01") } } },
   { $group: { _id: { $year: "$date" },
     commentCount: { $sum: 1 } } },
   { $merge : { into: "yearlyStats", on: "_id",
     whenMatched: "merge", whenNotMatched: "insert" } }
])

First stage:

The $match stage filters for comments posted between 1970 and 1972.

Second stage:

The $group stage groups by the year extracted from the comment date and counts comments into a new commentCount field.

Third stage:

The $merge stage writes the documents to the yearlyStats collection in the same database. If the stage finds an existing document in the collection that matches on the _id field (the year), the stage merges the matching documents. Otherwise, the stage inserts the document.

To view the documents in the yearlyStats collection after the data has been merged, run the following operation:

db.yearlyStats.find().sort( { _id: 1 } )

[
  { _id: 1970, movieCount: 1, commentCount: 889 },
  { _id: 1971, commentCount: 825 },
  { _id: 1972, movieCount: 1, commentCount: 863 }
]

Use the Pipeline to Customize the Merge

The $merge can use a custom update pipeline when documents match. The whenMatched pipeline can have the following stages:

• $addFields and its alias $set

• $project and its alias $unset

• $replaceRoot and its alias $replaceWith

A collection monthlyCommentTotals tracks the running count of comments for each month.

Each day, new comments arrive in the sample_mflix.comments collection. The following aggregation pipeline updates the monthly total with that day's comment count:

db.comments.aggregate([
   { $match: { date: { $gte: new Date("1970-01-15"),
     $lt: new Date("1970-01-16") } } },
   { $group: { _id: { $dateToString: { format: "%Y-%m",
     date: "$date" } }, count: { $sum: 1 } } },
   { $merge: {
         into: "monthlyCommentTotals",
         on: "_id",
         whenMatched: [
            { $addFields: {
               count: { $add: [ "$count", "$$new.count" ] }
            } }
         ],
         whenNotMatched: "insert"
   } }
])

First stage:

The $match stage finds all comments posted on January 15, 1970.

Second stage:

The $group stage groups the matching comments by year-month and counts them.

Third stage:

The $merge stage writes the documents to the monthlyCommentTotals collection. If the stage finds an existing document in the collection that matches on the _id field, the stage uses a pipeline to add the day's count to the existing monthly total.

• This pipeline cannot directly access the fields from the results document. To access the count field in the results document, the pipeline uses the $$new variable; i.e. $$new.count.

• This pipeline can directly access the count field in the existing document in the collection; i.e. $count.

The resulting document replaces the existing document.

To view documents in the monthlyCommentTotals collection after the merge operation, run the following operation:

db.monthlyCommentTotals.find()

[ { _id: '1970-01', count: 71 } ]

Use Variables to Customize the Merge

You can use variables in the $merge stage whenMatched field. Variables must be defined before they can be used.

Define variables in one or both of the following:

• $merge stage let

• aggregate command let (starting in MongoDB 5.0)

To use variables in whenMatched:

Specify the double dollar sign ($$) prefix together with the variable name in the form $$<variable_name>. For example, $$year. If the variable is set to a document, you can also include a document field in the form $$<variable_name>.<field>. For example, $$year.month.

The tabs below demonstrate behavior when variables are defined in the merge stage, the aggregate command, or both.

db.movies.aggregate( [
   { $match: { title: "The Godfather" } },
   { $limit: 1 },
   { $project: { title: 1 } },
   { $merge: { into: "movieDetails", whenNotMatched: "insert" } }
] )
db.runCommand( {
   aggregate: db.movieDetails.getName(),
   pipeline: [ {
      $merge: {
         into: db.movieDetails.getName(),
         let : { year: "2023" },
         whenMatched: [ {
            $addFields: { "addedYear": "$$year" }
         } ]
      }
   } ],
   cursor: {}
} )
db.movieDetails.find()

[ { _id: ..., title: 'The Godfather', addedYear: '2023' } ]

db.movies.aggregate( [
   { $match: { title: "The Godfather" } },
   { $limit: 1 },
   { $project: { title: 1 } },
   { $merge: { into: "movieDetails", whenNotMatched: "insert" } }
] )
db.runCommand( {
   aggregate: db.movieDetails.getName(),
   pipeline: [ {
      $merge: {
         into: db.movieDetails.getName(),
         whenMatched: [ {
            $addFields: { "addedYear": "$$year" }
         } ]
      }
   } ],
   cursor: {},
   let : { year: "2023" }
} )
db.movieDetails.find()

[ { _id: ..., title: 'The Godfather', addedYear: '2023' } ]

db.movies.aggregate( [
   { $match: { title: "The Godfather" } },
   { $limit: 1 },
   { $project: { title: 1 } },
   { $merge: { into: "movieDetails", whenNotMatched: "insert" } }
] )
db.runCommand( {
   aggregate: db.movieDetails.getName(),
   pipeline: [ {
      $merge: {
         into: db.movieDetails.getName(),
         let : { year: "2023" },
         whenMatched: [ {
            $addFields: { "addedYear": "$$year" }
         } ]
      }
   } ],
   cursor: {},
   let : { year: "2019" }
} )
db.movieDetails.find()

[ { _id: ..., title: 'The Godfather', addedYear: '2023' } ]