Skip to content

Replace map-reduce based QuerySet.sum/average with aggregation-based implementations #1336

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 3 commits into from
Jul 12, 2016
Merged

Replace map-reduce based QuerySet.sum/average with aggregation-based implementations #1336

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
16fea6f
replace QuerySet.sum/average implementations with aggregate_sum/avera…
wojcikstefan Jul 10, 2016
3fa9e70
prefer tuples over lists for immutable structures
wojcikstefan Jul 11, 2016
6148a60
update the changelog
wojcikstefan Jul 11, 2016
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions docs/changelog.rst
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,7 @@ Changes in 0.10.7 - DEV
- ListField now handles negative indicies correctly. #1270
- Fixed AttributeError when initializing EmbeddedDocument with positional args. #681
- Fixed no_cursor_timeout error with pymongo 3.0+ #1304
- Replaced map-reduce based QuerySet.sum/average with aggregation-based implementations #1336

Changes in 0.10.6
=================
Expand Down
145 changes: 30 additions & 115 deletions mongoengine/queryset/base.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1237,140 +1237,55 @@ def where(self, where_clause):
def sum(self, field):
"""Sum over the values of the specified field.

:param field: the field to sum over; use dot-notation to refer to
:param field: the field to sum over; use dot notation to refer to
embedded document fields

.. versionchanged:: 0.5 - updated to map_reduce as db.eval doesnt work
with sharding.
"""
map_func = """
function() {
var path = '{{~%(field)s}}'.split('.'),
field = this;

for (p in path) {
if (typeof field != 'undefined')
field = field[path[p]];
else
break;
}

if (field && field.constructor == Array) {
field.forEach(function(item) {
emit(1, item||0);
});
} else if (typeof field != 'undefined') {
emit(1, field||0);
}
}
""" % dict(field=field)

reduce_func = Code("""
function(key, values) {
var sum = 0;
for (var i in values) {
sum += values[i];
}
return sum;
}
""")

for result in self.map_reduce(map_func, reduce_func, output='inline'):
return result.value
else:
return 0

def aggregate_sum(self, field):
"""Sum over the values of the specified field.

:param field: the field to sum over; use dot-notation to refer to
embedded document fields

This method is more performant than the regular `sum`, because it uses
the aggregation framework instead of map-reduce.
"""
result = self._document._get_collection().aggregate([
pipeline = [
{'$match': self._query},
{'$group': {'_id': 'sum', 'total': {'$sum': '$' + field}}}
])
]

# if we're performing a sum over a list field, we sum up all the
# elements in the list, hence we need to $unwind the arrays first
ListField = _import_class('ListField')
field_parts = field.split('.')
field_instances = self._document._lookup_field(field_parts)
if isinstance(field_instances[-1], ListField):
pipeline.insert(1, {'$unwind': '$' + field})

result = self._document._get_collection().aggregate(pipeline)
if IS_PYMONGO_3:
result = list(result)
result = tuple(result)
else:
result = result.get('result')

if result:
return result[0]['total']
return 0

def average(self, field):
"""Average over the values of the specified field.

:param field: the field to average over; use dot-notation to refer to
:param field: the field to average over; use dot notation to refer to
embedded document fields

.. versionchanged:: 0.5 - updated to map_reduce as db.eval doesnt work
with sharding.
"""
map_func = """
function() {
var path = '{{~%(field)s}}'.split('.'),
field = this;

for (p in path) {
if (typeof field != 'undefined')
field = field[path[p]];
else
break;
}

if (field && field.constructor == Array) {
field.forEach(function(item) {
emit(1, {t: item||0, c: 1});
});
} else if (typeof field != 'undefined') {
emit(1, {t: field||0, c: 1});
}
}
""" % dict(field=field)

reduce_func = Code("""
function(key, values) {
var out = {t: 0, c: 0};
for (var i in values) {
var value = values[i];
out.t += value.t;
out.c += value.c;
}
return out;
}
""")

finalize_func = Code("""
function(key, value) {
return value.t / value.c;
}
""")

for result in self.map_reduce(map_func, reduce_func,
finalize_f=finalize_func, output='inline'):
return result.value
else:
return 0

def aggregate_average(self, field):
"""Average over the values of the specified field.

:param field: the field to average over; use dot-notation to refer to
embedded document fields

This method is more performant than the regular `average`, because it
uses the aggregation framework instead of map-reduce.
"""
result = self._document._get_collection().aggregate([
pipeline = [
{'$match': self._query},
{'$group': {'_id': 'avg', 'total': {'$avg': '$' + field}}}
])
]

# if we're performing an average over a list field, we average out
# all the elements in the list, hence we need to $unwind the arrays
# first
ListField = _import_class('ListField')
field_parts = field.split('.')
field_instances = self._document._lookup_field(field_parts)
if isinstance(field_instances[-1], ListField):
pipeline.insert(1, {'$unwind': '$' + field})

result = self._document._get_collection().aggregate(pipeline)
if IS_PYMONGO_3:
result = list(result)
result = tuple(result)
else:
result = result.get('result')
if result:
Expand Down
Loading