mongodb - 在 Golang 和 MongoDB 中使用 $lookup 和 $unwind 请求缓慢

Question

我的模型:一个事件可以被一个具有一个或多个提升的事件提升。

我正在尝试通过下面的请求获取所有提升的事件。

// Boosted ...
func (dao *campaignDAO) Boosted() ([]*models.Event, error) {
    // Clone the session
    session := dao.session.Clone()
    defer session.Close()

    // Get the time
    now := time.Now()

    // Create the pipe
    pipe := session.DB(shared.DatabaseNamespace).C("events").Pipe([]bson.M{
        {
            "$lookup": bson.M{
                "from":         "event_boosts",
                "localField":   "_id",
                "foreignField": "_event_id",
                "as":           "boost",
            },
        },
        {"$unwind": "$boost"},
        {
            "$match": bson.M{
                "boost.is_published": true,               // Boost is active
                "boost.start_date":   bson.M{"$lt": now}, // now is between start and end
                "boost.end_date":     bson.M{"$gt": now}, // now is between start and end
            },
        },
        {
            "$lookup": bson.M{
                "from":         "campaigns",
                "localField":   "boost._campaign_id",
                "foreignField": "_id",
                "as":           "campaign",
            },
        },
        {"$unwind": "$campaign"},
        {
            "$match": bson.M{
                "campaign.is_published": true, // Attached campaign is active
            },
        },
    })

    var result []*models.Event
    err := pipe.All(&result)
    if err != nil {
        return nil, err
    }

    return result, nil
}

但是这个请求需要3秒。以下是我在事件中的索引:

// NewCampaignDAO returns a new CampaignDAO
func NewCampaignDAO(session *mgo.Session) dao.CampaignDAO {
    // Set the collection
    col := session.DB(shared.DatabaseNamespace).C("campaigns")

    // Set the indexes
    col.EnsureIndexKey("start_date")
    col.EnsureIndexKey("end_date")
    col.EnsureIndexKey("created_by")
    col.EnsureIndexKey("is_published")

    return &campaignDAO{
        session:    session,
        collection: "campaigns",
    }
}

事件索引:

// NewEventDAO returns a new EventDAO
func NewEventDAO(session *mgo.Session) dao.EventDAO {
    // Set the collection
    col := session.DB(shared.DatabaseNamespace).C("events")

    // Set the indexes
    col.EnsureIndexKey("old_id")
    col.EnsureIndexKey("_parent_id")
    col.EnsureIndexKey("_location_id")
    col.EnsureIndexKey("price")
    col.EnsureIndexKey("name")
    col.EnsureIndexKey("category")
    col.EnsureIndexKey("start_date")
    col.EnsureIndexKey("end_date")
    col.EnsureIndexKey("is_recurrent")
    col.EnsureIndexKey("is_published")
    col.EnsureIndexKey("is_proposed")
    col.EnsureIndexKey("tags")
    col.EnsureIndexKey("price", "date", "name")

    return &eventDAO{
        session:    session,
        collection: "events",
    }
}

以及 MongoDB 的日志:

2018-06-19T13:22:53.465+0000 I COMMAND  [conn506] command clutch.event_boosts command: aggregate { aggregate: "events", pipeline: [ { $lookup: { as: "boost", from: "event_boosts", localField: "_id", foreignField: "_event_id" } }, { $unwind: "$boost" }, { $match: { boost.is_published: true, boost.start_date: { $lt: new Date(1529414570196) }, boost.end_date: { $gt: new Date(1529414570196) } } }, { $lookup: { from: "campaigns", localField: "boost._campaign_id", foreignField: "_id", as: "campaign" } }, { $unwind: "$campaign" }, { $match: { campaign.is_published: true } } ], cursor: {} } planSummary: COLLSCAN keysExamined:0 docsExamined:12936 cursorExhausted:1 numYields:121 nreturned:1 reslen:1149 locks:{ Global: { acquireCount: { r: 52018 } }, Database: { acquireCount: { r: 26009 } }, Collection: { acquireCount: { r: 26008 } } } protocol:op_query 3268ms

我没有找到可以改进的地方。

编辑:另外，我想知道是否可以通过在 event_boosts 集合上启动请求然后在事件中查找来改进请求。

EDIT2:添加 Mongo 版本。

MongoDB shell version v3.4.6
git version: c55eb86ef46ee7aede3b1e2a5d184a7df4bfb5b5
OpenSSL version: OpenSSL 1.0.1t  3 May 2016
allocator: tcmalloc
modules: none
build environment:
    distmod: debian81
    distarch: x86_64
    target_arch: x86_64

Answer 1

这大概就是我认为应该有所帮助的内容。它未经测试，因为我没有样本数据。加上它的 Go 语法可能有点不稳定，因为我不知道 Go。 ;) 但是，我相对确定 $lookup pipeline 中的 $match 语句将利用可用索引，而在您的查询中您在 $matches 之前有 $unwinds，这有效地使索引无用。

pipe := session.DB(shared.DatabaseNamespace).C("events").Pipe([]bson.M{
    {
        "$lookup": bson.M{
            "from": "event_boosts",
            "let": bson.M{ "e_id": "$_id" },
            "pipeline": []bson.M{
                "$match": bson.M{
                    "$expr": bson.M{
                       "$and": []interface{}{
                            bson.M{ "$eq": []string{ "$_event_id", "$$e_id" } },
                            bson.M{ "$eq": []string{ "$is_published", true } }, // Boost is active
                            bson.M{ "$lt": []string{ "$start_date", now } }, // now is between start and end
                            bson.M{ "$gt": []string{ "$end_date", now } },  // now is between start and end
                        },
                    },
                },
            },
            "as": "boost",
        },
    },
    { "$unwind": "$boost" },
    {
        "$lookup": bson.M{
            "from":         "campaigns",
            "let": bson.M{ "c_id": "$boost._campaign_id" },
            "pipeline": []bson.M{
                "$match": bson.M{
                    "$expr": bson.M{
                       "$and": []interface{}{
                            bson.M{ "$eq": []string{ "$id", "$$c_id" } },
                            bson.M{ "$eq": []string{ "$is_published", true } }, // Attached campaign is active
                        },
                    },
                },
            },
            "as": "campaign",
        },
    },
    { "$unwind": "$campaign" },
})