PYTHON / DATABASES WITH PYTHON
Aggregation pipelines and indexes
Build MongoDB aggregation pipelines from PyMongo and create indexes that the early stages of those pipelines can actually use.
What you will learn
- Chain $match, $group, $sort, $unwind stages with collection.aggregate()
- Reference document fields inside stages with the "$field" path syntax
- Create single-field and compound indexes with create_index()
- Read explain() output to tell an IXSCAN apart from a COLLSCAN
Understanding Aggregation pipelines and indexes
An aggregation pipeline is a list of stage dictionaries handed to collection.aggregate(). Each stage receives the stream of documents produced by the previous one and emits a new stream, so the order of the list is the order of computation. $match keeps documents, $group collapses them into one document per _id key, $sort orders them, and $unwind turns each element of an array field into its own document. Inside a stage, a bare string is a literal while "$total" is a field path meaning "the value of the total field of the current document".
$group is where most of the mental model lives. Its _id is the grouping key and every other key is an accumulator expression evaluated over all documents in the group, which is why {"$sum": 1} counts documents while {"$sum": "$total"} adds a field. After $group the documents no longer look like the stored ones: they only have _id and the accumulator fields, so any later stage must refer to those names, not the original field names.
Indexes matter because only the leading stages of a pipeline can touch them. MongoDB can push a $match, and sometimes a following $sort, down to the collection scan and satisfy it from an index; once a $group or $project has reshaped the stream, the data is intermediate and no index exists for it. A compound index on (status ascending, total descending) can serve a query on status alone, or on status plus a sort by descending total, because index keys are ordered left to right — but it cannot serve a query on total alone, since the first key is not constrained.
from pymongo import MongoClient, ASCENDING, DESCENDING
client = MongoClient("mongodb://localhost:27017/")
db = client.shop
db.orders.drop()
db.orders.insert_many([
{"customer": "ana", "city": "lisbon", "status": "paid", "total": 40.0},
{"customer": "ana", "city": "lisbon", "status": "paid", "total": 15.5},
{"customer": "bo", "city": "porto", "status": "paid", "total": 99.0},
{"customer": "bo", "city": "porto", "status": "pending", "total": 12.0},
{"customer": "cleo", "city": "lisbon", "status": "paid", "total": 60.5},
])
pipeline = [
{"$match": {"status": "paid"}},
{"$group": {"_id": "$city",
"revenue": {"$sum": "$total"},
"orders": {"$sum": 1}}},
{"$sort": {"revenue": -1}},
]
for doc in db.orders.aggregate(pipeline):
print(f"{doc['_id']:>6} revenue={doc['revenue']:.2f} orders={doc['orders']}")
name = db.orders.create_index([("status", ASCENDING), ("total", DESCENDING)])
print("created:", name)
print(sorted(db.orders.index_information()))Stage order decides everything: filter and sort before you group, because only pre-group stages can be served by an index.
Worked examples
Proving an index is used
Compares the query plan for the same filter before and after creating an index.
from pymongo import MongoClient
client = MongoClient("mongodb://localhost:27017/")
db = client.shop
db.readings.drop()
db.readings.insert_many([{"sensor": i % 50, "value": i} for i in range(5000)])
def plan(query):
report = str(db.readings.find(query).explain())
return "IXSCAN" if "IXSCAN" in report else "COLLSCAN"
print("before index:", plan({"sensor": 7}))
db.readings.create_index("sensor")
print("after index: ", plan({"sensor": 7}))
print("docs matched:", db.readings.count_documents({"sensor": 7}))Example explained
Line 1find(...).explain() asks the server for the chosen plan instead of the documents.
Line 2With no index on sensor the only option is COLLSCAN: read all 5000 documents and test each.
Line 3create_index("sensor") builds an ascending single-field index, so the planner can jump straight to the 100 matching keys.
Line 4The same reasoning applies to a leading $match in a pipeline, which is pushed down to this level.
$unwind and filtering after $group
Expands an array field, aggregates per tag, then filters the groups themselves.
from pymongo import MongoClient
client = MongoClient("mongodb://localhost:27017/")
db = client.blog
db.posts.drop()
db.posts.insert_many([
{"title": "a", "tags": ["python", "mongo"], "views": 10},
{"title": "b", "tags": ["python"], "views": 30},
{"title": "c", "tags": ["mongo", "index"], "views": 5},
])
pipeline = [
{"$unwind": "$tags"},
{"$group": {"_id": "$tags", "views": {"$sum": "$views"}}},
{"$match": {"views": {"$gte": 10}}},
{"$sort": {"views": -1, "_id": 1}},
]
for doc in db.posts.aggregate(pipeline):
print(doc["_id"], doc["views"])Example explained
Line 1$unwind turns the three posts into six documents, one per tag, each carrying a copy of views.
Line 2$group sums views per tag: python gets 10 + 30, mongo gets 10 + 5, index gets 5.
Line 3The second $match filters grouped results, like SQL HAVING, and cannot use any index because views here is computed.
Line 4$sort takes two keys so ties would break on _id, keeping the output deterministic.
Important notes
create_index() is idempotent and cheap to call again on an existing index, but do it once at startup, not inside a request handler.
Each pipeline stage is limited to about 100 MB of memory; a large $group or $sort fails unless you pass allowDiskUse=True to aggregate().
Common mistakes
Writing {"$sum": "total"} instead of {"$sum": "$total"}: without the dollar prefix it is a literal string, which is not numeric, so every group reports 0.
Placing $match after $group to filter raw fields: the field no longer exists at that point, so the stage silently matches nothing and the pipeline returns an empty result.
Referring to original field names after $group: only _id and the accumulator names survive, so a later $sort on "city" sorts on a missing field and the order looks random.
Try it yourself
Change, predict, then run
Insert five order documents with city, status and total, then write a pipeline that keeps only paid orders, computes the average total per city with $avg, and sorts descending. Add a compound index on status and city and confirm with explain() that the filter uses IXSCAN.
Open the Python workspaceCheck your understanding
A pipeline is [$group by city, $match status == "paid", $sort by revenue]. Why does adding an index on status not speed it up?
- By the time $match runs the documents are grouped intermediate results, so no collection index applies
- Indexes are never used by aggregate(), only by find()
- $sort invalidates any index used earlier in the pipeline
- An index on a string field cannot be used for equality matching
Show answer
Only stages before the first reshaping stage can be pushed down to an indexed collection scan; after $group the stream is computed data with no index behind it, and status has already been dropped. The claim that aggregate() never uses indexes is wrong: move the $match to the front and the same index is used.