Dotazování po naplnění v Mongoose

S moderní MongoDB vyšší než 3.2 můžete použít $lookup jako alternativu k .populate() většinou. To má také tu výhodu, že se skutečně připojuje "na serveru" na rozdíl od .populate() dělá, což je ve skutečnosti „multiple queries“ to „emulovat“ připojení.

Takže .populate() není opravdu "spojení" ve smyslu toho, jak to dělá relační databáze. $lookup na druhé straně operátor ve skutečnosti vykonává práci na serveru a je víceméně analogický k "LEFT JOIN" :

Item.aggregate(
  [
    { "$lookup": {
      "from": ItemTags.collection.name,
      "localField": "tags",
      "foreignField": "_id",
      "as": "tags"
    }},
    { "$unwind": "$tags" },
    { "$match": { "tags.tagName": { "$in": [ "funny", "politics" ] } } },
    { "$group": {
      "_id": "$_id",
      "dateCreated": { "$first": "$dateCreated" },
      "title": { "$first": "$title" },
      "description": { "$first": "$description" },
      "tags": { "$push": "$tags" }
    }}
  ],
  function(err, result) {
    // "tags" is now filtered by condition and "joined"
  }
)

N.B. .collection.name zde se ve skutečnosti vyhodnotí na "řetězec", což je skutečný název kolekce MongoDB, jak je přiřazena k modelu. Protože mongoose ve výchozím nastavení „pluralizuje“ názvy kolekcí a $lookup potřebuje jako argument skutečný název kolekce MongoDB (protože se jedná o serverovou operaci), pak je to praktický trik, který lze použít v kódu mongoose, na rozdíl od přímého „pevného kódování“ názvu kolekce.

Mohli bychom také použít $filter na polích k odstranění nežádoucích položek, toto je ve skutečnosti nejúčinnější forma díky optimalizaci agregačního potrubí pro speciální podmínku jako $lookup následované oběma $unwind a $match podmínka.

To ve skutečnosti vede ke sloučení tří fází potrubí do jedné:

   { "$lookup" : {
     "from" : "itemtags",
     "as" : "tags",
     "localField" : "tags",
     "foreignField" : "_id",
     "unwinding" : {
       "preserveNullAndEmptyArrays" : false
     },
     "matching" : {
       "tagName" : {
         "$in" : [
           "funny",
           "politics"
         ]
       }
     }
   }}

To je vysoce optimální, protože skutečná operace „filtruje kolekci, aby se nejprve připojila“, poté vrátí výsledky a „rozvine“ pole. Jsou použity obě metody, takže výsledky neporušují limit BSON 16 MB, což je omezení, které klient nemá.

Jediný problém je v tom, že se to v některých ohledech zdá "neintuitivní", zvláště když chcete výsledky v poli, ale to je to, co $group je zde, protože se rekonstruuje do původní podoby dokumentu.

Je také nešťastné, že v tuto chvíli prostě nemůžeme napsat $lookup ve stejné případné syntaxi, kterou server používá. IMHO, toto je přehlédnutí, které je třeba opravit. Ale prozatím bude fungovat jednoduché použití sekvence a je to nejschůdnější možnost s nejlepším výkonem a škálovatelností.

Dodatek – MongoDB 3.6 a novější

I když je zde zobrazený vzor docela optimalizovaný kvůli tomu, jak se do $lookup vloží další fáze , má jednu chybu v tom, že "LEFT JOIN", které je normálně vlastní oběma $lookup a akce populate() je negováno "optimálním" použití $unwind zde, která nezachovává prázdná pole. Můžete přidat preserveNullAndEmptyArrays možnost, ale to neguje "optimalizováno" sekvence popsaná výše a v podstatě ponechává všechny tři fáze nedotčené, které by se normálně při optimalizaci spojily.

MongoDB 3.6 se rozšiřuje o "výraznější" ve tvaru $lookup umožňující výraz "sub-pipeline". Což nejen splňuje cíl zachovat „LEFT JOIN“, ale stále umožňuje optimální dotaz pro snížení vracených výsledků a s mnohem zjednodušenou syntaxí:

Item.aggregate([
  { "$lookup": {
    "from": ItemTags.collection.name,
    "let": { "tags": "$tags" },
    "pipeline": [
      { "$match": {
        "tags": { "$in": [ "politics", "funny" ] },
        "$expr": { "$in": [ "$_id", "$$tags" ] }
      }}
    ]
  }}
])

$expr používá se k porovnání deklarované „místní“ hodnoty s „cizí“ hodnotou je ve skutečnosti to, co MongoDB dělá „interně“ nyní s původním $lookup syntax. Vyjádřením v této formě můžeme přizpůsobit počáteční $match výraz v rámci "sub-pipeline" sami.

Ve skutečnosti, jako skutečný „potrubí agregace“, můžete s agregačním potrubím v rámci tohoto výrazu „sub-pipeline“ dělat téměř cokoli, včetně „vnořování“ úrovní $lookup do dalších souvisejících sbírek.

Další použití je trochu nad rámec toho, co zde klade otázka, ale ve vztahu k dokonce "vnořené populaci" pak nový vzor použití $lookup umožňuje, aby to bylo téměř stejné a "hodně" výkonnější při plném využití.

Pracovní příklad

Níže je uveden příklad použití statické metody na modelu. Jakmile je tato statická metoda implementována, volání se jednoduše změní na:

  Item.lookup(
    {
      path: 'tags',
      query: { 'tags.tagName' : { '$in': [ 'funny', 'politics' ] } }
    },
    callback
  )

Nebo se vylepšení, které je o něco modernější, dokonce stává:

  let results = await Item.lookup({
    path: 'tags',
    query: { 'tagName' : { '$in': [ 'funny', 'politics' ] } }
  })

Díky tomu je velmi podobný .populate() ve struktuře, ale ve skutečnosti místo toho provádí připojení na serveru. Pro úplnost, zde uvedené použití přenese vrácená data zpět do instancí dokumentu mongoose v jak v nadřazeném, tak v podřízeném případě.

Je to docela triviální a snadno se přizpůsobí nebo prostě použije ve většině běžných případů.

N.B Použití async je zde pouze pro stručnost spuštění přiloženého příkladu. Vlastní implementace je bez této závislosti.

const async = require('async'),
      mongoose = require('mongoose'),
      Schema = mongoose.Schema;

mongoose.Promise = global.Promise;
mongoose.set('debug', true);
mongoose.connect('mongodb://localhost/looktest');

const itemTagSchema = new Schema({
  tagName: String
});

const itemSchema = new Schema({
  dateCreated: { type: Date, default: Date.now },
  title: String,
  description: String,
  tags: [{ type: Schema.Types.ObjectId, ref: 'ItemTag' }]
});

itemSchema.statics.lookup = function(opt,callback) {
  let rel =
    mongoose.model(this.schema.path(opt.path).caster.options.ref);

  let group = { "$group": { } };
  this.schema.eachPath(p =>
    group.$group[p] = (p === "_id") ? "$_id" :
      (p === opt.path) ? { "$push": `$${p}` } : { "$first": `$${p}` });

  let pipeline = [
    { "$lookup": {
      "from": rel.collection.name,
      "as": opt.path,
      "localField": opt.path,
      "foreignField": "_id"
    }},
    { "$unwind": `$${opt.path}` },
    { "$match": opt.query },
    group
  ];

  this.aggregate(pipeline,(err,result) => {
    if (err) callback(err);
    result = result.map(m => {
      m[opt.path] = m[opt.path].map(r => rel(r));
      return this(m);
    });
    callback(err,result);
  });
}

const Item = mongoose.model('Item', itemSchema);
const ItemTag = mongoose.model('ItemTag', itemTagSchema);

function log(body) {
  console.log(JSON.stringify(body, undefined, 2))
}
async.series(
  [
    // Clean data
    (callback) => async.each(mongoose.models,(model,callback) =>
      model.remove({},callback),callback),

    // Create tags and items
    (callback) =>
      async.waterfall(
        [
          (callback) =>
            ItemTag.create([{ "tagName": "movies" }, { "tagName": "funny" }],
              callback),

          (tags, callback) =>
            Item.create({ "title": "Something","description": "An item",
              "tags": tags },callback)
        ],
        callback
      ),

    // Query with our static
    (callback) =>
      Item.lookup(
        {
          path: 'tags',
          query: { 'tags.tagName' : { '$in': [ 'funny', 'politics' ] } }
        },
        callback
      )
  ],
  (err,results) => {
    if (err) throw err;
    let result = results.pop();
    log(result);
    mongoose.disconnect();
  }
)

Nebo trochu modernější pro Node 8.xa vyšší s async/await a žádné další závislosti:

const { Schema } = mongoose = require('mongoose');
const uri = 'mongodb://localhost/looktest';

mongoose.Promise = global.Promise;
mongoose.set('debug', true);

const itemTagSchema = new Schema({
  tagName: String
});

const itemSchema = new Schema({
  dateCreated: { type: Date, default: Date.now },
  title: String,
  description: String,
  tags: [{ type: Schema.Types.ObjectId, ref: 'ItemTag' }]
});

itemSchema.statics.lookup = function(opt) {
  let rel =
    mongoose.model(this.schema.path(opt.path).caster.options.ref);

  let group = { "$group": { } };
  this.schema.eachPath(p =>
    group.$group[p] = (p === "_id") ? "$_id" :
      (p === opt.path) ? { "$push": `$${p}` } : { "$first": `$${p}` });

  let pipeline = [
    { "$lookup": {
      "from": rel.collection.name,
      "as": opt.path,
      "localField": opt.path,
      "foreignField": "_id"
    }},
    { "$unwind": `$${opt.path}` },
    { "$match": opt.query },
    group
  ];

  return this.aggregate(pipeline).exec().then(r => r.map(m => 
    this({ ...m, [opt.path]: m[opt.path].map(r => rel(r)) })
  ));
}

const Item = mongoose.model('Item', itemSchema);
const ItemTag = mongoose.model('ItemTag', itemTagSchema);

const log = body => console.log(JSON.stringify(body, undefined, 2));

(async function() {
  try {

    const conn = await mongoose.connect(uri);

    // Clean data
    await Promise.all(Object.entries(conn.models).map(([k,m]) => m.remove()));

    // Create tags and items
    const tags = await ItemTag.create(
      ["movies", "funny"].map(tagName =>({ tagName }))
    );
    const item = await Item.create({ 
      "title": "Something",
      "description": "An item",
      tags 
    });

    // Query with our static
    const result = (await Item.lookup({
      path: 'tags',
      query: { 'tags.tagName' : { '$in': [ 'funny', 'politics' ] } }
    })).pop();
    log(result);

    mongoose.disconnect();

  } catch (e) {
    console.error(e);
  } finally {
    process.exit()
  }
})()

A od MongoDB 3.6 a vyšší, dokonce i bez $unwind a $group budova:

const { Schema, Types: { ObjectId } } = mongoose = require('mongoose');

const uri = 'mongodb://localhost/looktest';

mongoose.Promise = global.Promise;
mongoose.set('debug', true);

const itemTagSchema = new Schema({
  tagName: String
});

const itemSchema = new Schema({
  title: String,
  description: String,
  tags: [{ type: Schema.Types.ObjectId, ref: 'ItemTag' }]
},{ timestamps: true });

itemSchema.statics.lookup = function({ path, query }) {
  let rel =
    mongoose.model(this.schema.path(path).caster.options.ref);

  // MongoDB 3.6 and up $lookup with sub-pipeline
  let pipeline = [
    { "$lookup": {
      "from": rel.collection.name,
      "as": path,
      "let": { [path]: `$${path}` },
      "pipeline": [
        { "$match": {
          ...query,
          "$expr": { "$in": [ "$_id", `$$${path}` ] }
        }}
      ]
    }}
  ];

  return this.aggregate(pipeline).exec().then(r => r.map(m =>
    this({ ...m, [path]: m[path].map(r => rel(r)) })
  ));
};

const Item = mongoose.model('Item', itemSchema);
const ItemTag = mongoose.model('ItemTag', itemTagSchema);

const log = body => console.log(JSON.stringify(body, undefined, 2));

(async function() {

  try {

    const conn = await mongoose.connect(uri);

    // Clean data
    await Promise.all(Object.entries(conn.models).map(([k,m]) => m.remove()));

    // Create tags and items
    const tags = await ItemTag.insertMany(
      ["movies", "funny"].map(tagName => ({ tagName }))
    );

    const item = await Item.create({
      "title": "Something",
      "description": "An item",
      tags
    });

    // Query with our static
    let result = (await Item.lookup({
      path: 'tags',
      query: { 'tagName': { '$in': [ 'funny', 'politics' ] } }
    })).pop();
    log(result);


    await mongoose.disconnect();

  } catch(e) {
    console.error(e)
  } finally {
    process.exit()
  }

})()