node.js - 使用 "dotted field" '.' 查询和过滤 ChangeStream 文档

Question

我有一个看起来像这样的文档

{
  components: { weapon: { type: "Sword" }, health: { value: 10 } }
  type: "Monster"
}

我正在使用 change stream正在返回

{
  operationType: 'update',
  updateDescription: {
    updatedFields: { 'components.weapon': [Object] }
  }
}

我想向聚合管道添加一个查询，以过滤掉任何不组件的更新，即，如果类型字段已更新，我不希望获得更新。

我的查询看起来像

{ $match: { "updateDescription.updatedFields.components.weapon": { $exists: true } }

但是这不起作用，因为updatedFields上的字段是“components.weapon”而不是组件:{武器:..}。

如果我可以使用括号符号，我会这样做

{ $match: { "updateDescription.updatedFields['components.weapon']": { $exists: true } }

然而，这在 MongoDB 语法中是不允许的(或者至少它不起作用)。

有没有办法解决这个问题？

Answer 1

这是一个很难掌握的问题，但是这里有一些坚实的原因，并且正常 MongoDB 文档情况不会包含这样的“点式字段”。所以确实需要一些特殊的处理。

这里的基本情况是，您本质上需要将包含“点分字段”的文档中的“key”转换为实际上的“字符串”，然后只需查找该字符串中存在“组件”。

简单的情况是您基本上想要一个 pipeline 表达式来表示 watch()像这样:

const pipeline = [
  { "$match": {
    "$expr": {
      "$ne": [
        { "$size": {
          "$filter": {
            "input": {
              "$objectToArray": "$updateDescription.updatedFields"
            },
            "cond": {
              "$eq": [{ "$indexOfCP": [ "$$this.k", "component" ], }, -1]
            }
          }
        }},
        0
      ]
    }
  }}
];

它的作用是使用 $objectToArray将 updatedFields 对象转换为 k 和 v 属性数组，而不是命名键。此时，结果值将如下所示:

[
  {
    "k": "coponents.weapon",
    "v": "Sword"
  }
]

这允许 now 数组 与 $filter 一起使用使用 $indexOfCP 中的表达式进行操作它测试 k 属性值中是否存在字符串。如果它不是 -1 (表示 not find )，则该值中包含 "component" 的任何元素都将是唯一的内容保留，并匹配结果数组中的元素。

NOTE If you have MongoDB 4.2, you might want to look at the $regexMatch operator, in place of $indexOfCP. It should not be necessary for simply testing for the "presence" of a string within a string, but Regular expressions can of course do a bit more if your use case requires it.

Appropriate here, you would probably search from the beginning of string up to the included "dot", replacing inside cond for the $filter:

     "$not": { "$regexMatch": { "input": "$$this.k", "regex": /^components\./ } }

由于它作为数组返回，因此您可以测试 $size以便查看在删除“component”值后，现在过滤数组中是否确实还剩下任何内容。如果不是，并且大小确实为 0，则结果将通过 $expr 被丢弃。 ，这也是允许在 $match 中使用聚合表达式的主要运算符。也是如此。

当然，所做的只是选择实际上有效返回的文档。如果您在变更流文档中指示的更新结果中可能有其他已更改的字段，那么您实际上需要使用相同类型的 $filter对其进行操作，以便实际从结果中删除 "component" 字段:

  { "$addFields": {
      "updateDescription": {
        "updatedFields": {
          "$arrayToObject": {
            "$filter": {
              "input": {
                "$objectToArray": "$updateDescription.updatedFields"
              },
              "cond": {
                "$eq": [{ "$indexOfCP": [ "$$this.k", "component" ] }, -1 ]
              }
            }
          }
        }
      }
    }
  }

请注意此处添加了 $arrayToObject这本质上反转所使用的操作过程，然后实际上以原始形式返回updatedFields内容，而没有不需要的键。

<小时/>

为了进行演示，这里是一个实际的完整列表，它重现了对具有此类结构的集合所做的更改，并包含观察者管道，以便静音不需要的内容变化:

const { MongoClient } = require('mongodb');

const uri = 'mongodb://localhost:27017';
const options = { useNewUrlParser: true, useUnifiedTopology: true };

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

(async function() {


  try {

    let client = await MongoClient.connect(uri, options);

    let db = client.db('test');


    // Insert some starting data
    await db.collection('things').deleteMany();
    await db.collection('things').insertOne({
      components: {
        weapon: { type: "Sword" },
        health: { value: 10 }
      },
      type: "Monster"
    });

    // Set up the changeStream
    const pipeline = [
      // Filters documents
      { "$match": {
        "$expr": {
          "$ne": [
            { "$size": {
              "$filter": {
                "input": {
                  "$objectToArray": "$updateDescription.updatedFields"
                },
                "cond": {
                  "$eq": [{ "$indexOfCP": [ "$$this.k", "component" ], }, -1]
                  /* Alternate MongoDB 4.2 syntax
                  "$not": {
                    "$regexMatch": {
                      "input": "$$this.k",
                      "regex": /^components\./
                    }
                  }
                  */
                }
              }
            }},
            0
          ]
        }
      }},
      /* -- Uncomment just to see the k and v structure
      { "$project": {
        "update": { "$objectToArray": "$updateDescription.updatedFields" }
      }}
      */

      // Actually removes the keys and returns only non filtered
      { "$addFields": {
          "updateDescription": {
            "updatedFields": {
              "$arrayToObject": {
                "$filter": {
                  "input": {
                    "$objectToArray": "$updateDescription.updatedFields"
                  },
                  "cond": {
                    "$eq": [{ "$indexOfCP": [ "$$this.k", "component" ] }, -1 ]
                    /* Alternate MongoDB 4.2 syntax
                    "$not": {
                      "$regexMatch": {
                        "input": "$$this.k",
                        "regex": /^components\./
                      }
                    }
                    */
                  }
                }
              }
            }
          }
        }
      }
    ];

    const changeStream = db.collection('things').watch(pipeline);

    changeStream.on('change', next => log({ changeDocument: next }));


    // Loop some changes


    await new Promise(async (resolve, reject) =>  {

      let tick = true;

      setInterval(async () => {
        try {
          let { value }= await db.collection('things')
            .findOneAndUpdate(
              {},
              { $set: { 'components.weapon': (tick) ? 'Knife' : 'Sword' }},
              { returnOriginalDocument: false }
            );
          tick = !tick; // flip the boolean
          log({ currentDoc: value });
        } catch (e) {
          reject(e);
        }
      },2000)

    });



  } catch (e) {
    console.error(e)
  }


})()