标签： mongodb

MongoDB 实现多表联查并更新数据

最近有个需求，用 SQL 描述起来大约是这样的：

# pages 表按照 projectId 统计每个 project 的页面数
# 然后更新到 projects 表的 `pageCount` 字段里
UPDATE `projects` a LEFT JOIN
  (
    SELECT COUNT('x') as `num`, `projectId`
    FROM `pages`
    WHERE `isDeleted`=false
    GROUP BY `projectId`
  ) b
  ON a.`_id`=b.`projectId`
SET `pageCount`=`num`
WHERE a.`isDeleted`=false

# 上面 SQL 凑合看，我好久不写了，有点忘记怎么写……

MongoDB Shell 其实就是封装好的 JavaScript + Node.js 16 REPL 环境，我们熟悉的箭头函数、异步函数等都可以放心使用，所以写起来大约是这样：

// 进入聚合状态
db.projects.aggregate([
  // 聚合状态会从上至下执行，所以我们先筛选出来合适的 projects
  {
    $match: {
      creatorId: '我的用户id',
      isDeleted: false,
      kind: 'Normal',
    },
  },
  // 因为我厂 pages 表里 projectId 是字符类型，所以这里要先转换一次
  {
    $addFields: {
      projectId: { '$toString': '$_id'},
    }
  },
  // 接下来就可以聚合了，用本地的 projectId 对上连表的 projectId，连起来之后的字段名为 pages，它应该是个数组
  {
    $lookup: {
      from: 'pages',
      localField: 'projectId',
      foreignField: 'projectId',
      as: 'pages',
    }
  },
  // 再添加一个字段用来存储 pages 的长度，即有多少个页面，这里要加 `$` 前缀
  {
    $addFields: {
      pageCount: {
        $size: '$pages'
      }
    }
  }
])
// 聚合完毕之后，再把内容写入数据库
.forEach(doc => {
  db.projects.updateOne({ _id: doc._id}, {
    $set: { pageCount: doc.pageCount }
  })
});

对于我来说，操作 mongosh 的难点主要在于：

很多时候要加 $ 前缀，一会儿一个一会儿两个，不熟悉经常搞错
不知道是否该用异步函数或者 .then()
不知道 $addFields 函数
没法打断点调试，需要一长串执行完才知道结果
MongoDB 作为也有不短的历史，搜索得到的内容覆盖很多版本，很难直接应用

希望将来能慢慢克服这些难点。欢迎诸位读者指教，如有问题，也欢迎提出讨论。

2022年7月29日

MongoDB 里实现多表联查

前些天遇到一个需求，不复杂，用 SQL 表现的话，大约如此：

SELECT *
FROM db1 LEFT JOIN db2 ON db1.a = db2.b
WHERE db1.userId='$me' AND db2.status=1

没想到搜了半天，我厂的代码仓库里没有这种用法，各种教程也多半只针对合并查询（即只筛选 db1，没有 db2 的条件）。所以最后只好读文档+代码尝试，终于找到答案，记录一下。

我们用 mongoose 作为连接库
联查需要用 $lookup
如果声明外键的时候用 ObjectId，就很简单：

// 假设下面两个表 db1 和 db2
export const Db1Schema = new mongoose.Schema(
  {
    userId: { type: String, index: true },
    couponId: { type: ObjectId, ref: Db2Schema },
  },
  { versionKey: false, timestamps: true }
);
export const Db2Schema = new mongoose.Schema(
  {
    status: { type: Boolean, default: 0 },
  },
  { versionKey: false, timestamps: true }
);

// 那么只要
db1Model.aggregate([
  {
    $lookup: {
      from: 'db2', // 目标表
      localField: 'couponId', // 本地字段
      foreignField: '_id', // 对应的目标字段
      as: 'source',
  },
  {
    $match: [ /* 各种条件 */ ],
  },
]);

但是我们没有用 ObjectId，而是用 string 作为外键，所以无法直接用上面的联查。必须在 pipeline 里手动转换、联合。此时，当前表（db1）的字段不能直接使用，要配合 let，然后加上 $$ 前缀；连表（db2）直接加 $ 前缀即可。

最终代码如下：

// 每次必有的条件，当前表的字段用 `$$`，连表的字段用 `$`
const filter = [{ $eq: ['$$userId', userId] }, { $eq: ['$isDeleted', false] }];
if (status === Expired) {
  dateOp = '$lte';
} else if (status === Normal) {
  dateOp = '$gte';
  filter.push({ $in: ['$$status', [Normal, Shared]] });
} else {
  dateOp = '$gte';
  filter.push({ $eq: ['$$status', status] });
}
const results = await myModel.aggregate([
  {
    $lookup: {
      from: 'coupons',
      // 当前表字段必须 `let` 之后才能用
      let: { couponId: '$couponId', userId: '$userId', status: '$status' },
      // 在 pipeline 里完成筛选
      pipeline: [
        {
          $match: {
            $expr: {
              // `$toString` 是内建方法，可以把 `ObjectId` 转换成 `string`
              $and: [{ $eq: [{ $toString: '$_id' }, '$$couponId'] }, ...filter, { [dateOp]: ['$endAt', new Date()] }],
            },
          },
        },
        // 只要某些字段，在这里筛选
        {
          $project: couponFields,
        },
      ],
      as: 'source',
    },
  },
  {
    // 这种筛选相当 LEFT JOIN，所以需要去掉没有连表内容的结果
    $match: {
      source: { $ne: [] },
    },
  },
  {
    // 为了一次查表出结果，要转换一下输出格式
    $facet: {
      results: [{ $skip: size * (page - 1) }, { $limit: size }],
      count: [
        {
          $count: 'count',
        },
      ],
    },
  },
]);

同事告诉我，这样做的效率不一定高。我觉得，考虑到实际场景，他说的可能没错，不过，早晚要迈出这样的一步。而且，未来我们也应该慢慢把外键改成 ObjectId 类型。

2022年7月2日