This method uses locks because it can be used during indexing,
and Druid can call aggregate() and get() concurrently
See https://github.com/druid-io/druid/pull/3956
516387
@Override
  public void aggregate(final ByteBuffer buf, final int position)
  {
    final Object value = selector.getObject();
    if (value == null) {
      return;
    }
    final Lock lock = stripedLock.getAt(lockIndex(position)).writeLock();
    lock.lock();
    try {
      final HllSketch sketch = sketchCache.get(buf).get(position);
      HllSketchBuildAggregator.updateSketch(sketch, value);
    }
    finally {
      lock.unlock();
    }
  }
108528
public static Sequence<Row> postAggregate(final GroupByQuery query, IncrementalIndex index)
  {
    return Sequences.map(
        Sequences.simple(index.iterableWithPostAggregations(query.getPostAggregatorSpecs(), query.isDescending())),
        new Function<Row, Row>()
        {
          @Override
          public Row apply(Row input)
          {
            final MapBasedRow row = (MapBasedRow) input;
            return new MapBasedRow(
                query.getGranularity()
                     .toDateTime(row.getTimestampFromEpoch()),
                row.getEvent()
            );
          }
        }
    );
  }
600587
private ILockableEntityGroup ifindLockableGroup(String key, String lockOwner)
            throws GroupsException {
        return compositeGroupService.findGroupWithLock(key, lockOwner);
    }
46222
def aggregate_sum(aggr)
      sum = {}
      aggr.each do |ts, counterVals|
        sum[ts] = {} unless sum.has_key?ts
        counterVals.each do |obj, count|
          if obj.respond_to?(:enterprise_id)
            eid = obj.public_send(:enterprise_id).to_s
            sum[ts][eid] = sum[ts].fetch(eid, 0) + count
          end
        end
      end
      sum
    end
399225
def get(self, *args, **kwargs):
        
        clone = self.filter(*args, **kwargs)

        if self.query.can_filter():
            clone = clone.order_by()

        num = len(clone)

        if num == 1:
            return clone._result_cache[0]
        if not num:
            raise self.resource.DoesNotExist(
                "%s matching query does not exist. "
                "Lookup parameters were %s" %
                (self.resource._meta.resource_name, kwargs))

        raise self.resource.MultipleObjectsReturned(
            "get() returned more than one %s -- it returned %s! "
            "Lookup parameters were %s" %
            (self.resource._meta.resource_name, num, kwargs))
571067
def aggregate(self, query: Optional[dict] = None,
                  group: Optional[dict] = None,
                  order_by: Union[None, list, tuple] = None) -> list:
        """"""
        raise NotImplementedError
245869
synchronized Collection<KeyValue> getAll(long index) {
        incrementLastAppliedIndex(index);

        Collection<KeyValue> copiedEntries = Lists.newArrayListWithCapacity(entries.size());
        for (Map.Entry<String, String> entry : entries.entrySet()) {
            copiedEntries.add(new KeyValue(entry.getKey(), entry.getValue()));
        }

        return copiedEntries;
    }
970257
boolean updateMoreAggInfo(Table aggTable, AppAggregationKey appAggKey,
      JobDetails jobDetails) throws IOException {

    int attempts = 0;
    boolean status = false;
    // update number of runs
    while ((!status) && (attempts < AggregationConstants.RETRY_COUNT)) {
      status = updateNumberRuns(appAggKey, aggTable, jobDetails);
      attempts++;
    }

    // update queue
    attempts = 0;
    status = false;
    while ((!status) && (attempts < AggregationConstants.RETRY_COUNT)) {
      status = updateQueue(appAggKey, aggTable, jobDetails);
      attempts++;
    }

    // update cost
    attempts = 0;
    status = false;
    while ((!status) && (attempts < AggregationConstants.RETRY_COUNT)) {
      status = updateCost(appAggKey, aggTable, jobDetails);
      attempts++;
    }

    // create a put for username, appid
    byte[] rowKey = aggConv.toBytes(appAggKey);
    Put p = new Put(rowKey);
    p.addColumn(Constants.INFO_FAM_BYTES, AggregationConstants.USER_BYTES,
        Bytes.toBytes(appAggKey.getUserName()));
    p.addColumn(Constants.INFO_FAM_BYTES, AggregationConstants.APP_ID_COL_BYTES,
        Bytes.toBytes(appAggKey.getAppId()));
    aggTable.put(p);
    return true;
  }
978155
function preCacheQueries(queries) {
  var groupBy = {};
  _.each(queries, function (query) {
    if (!query.queryKey) {
      return;
    }
    var hasAllValues = _.reduce(_.values(query.queryParams), function (acc, val) {
      return acc && (val !== null);
    }, true);
    if (hasAllValues) {
      var cacheKey = _executionCache.getKey(query);
      groupBy[cacheKey] = groupBy[cacheKey] || [];
      groupBy[cacheKey].push(query);
    }
  });

  _.each(groupBy, function (groupedQueries, key) {
    var aggregateShape = _.reduce(_.map(groupedQueries, 'shape'), _.assign, {});
    var query = _.cloneDeep(groupedQueries[0]);
    query.shape = aggregateShape;
    _executionCache.execute(query);
  });
}
10683
function record(name, num, statistics, aggregations, timestamp, callback) {
        timestamp = timestamp || Date.now();
        function recordStats(key) {
            statistics && statistics.forEach(function(stat) {
                    if(stat == 'sum') {
                        multi.hincrby(key, 'sum', num)
                    } else if(scripts[stat]){
                        evalScript(multi, stat, [key, num],[],function(err, result) {
                                if(err) throw err;
                                if(result && result.indexOf('ERR') == 0) {
                                    // console.log(scripts[stat]);
                                    throw new Error(result);
                                }
                        })
                    }
            });
        }
        if(!statistics) statistics = ['sum'];
        if(num == undefined) num = 1;

        var multi = redis.multi();

        granularities.forEach(function(gran) {
                var key = getGranKey(name, gran, timestamp);
                recordStats(key);
                var unitPeriod = gran[0];
                multi.expire(key, points * unitPeriod / 1000);
        });

        if(aggregations) for(var i = 0; i < aggregations.length; i++) {
            var aggr = aggregations[i];
            var key = getAggrKey(name, aggr, timestamp);
            recordStats(key);
        }

        multi.exec(callback || log);
    }
1152514
def aggregate(input, **params):
    
    PARAM_CFG_EXTRACT = 'extract'
    PARAM_CFG_SUBSTITUTE = 'substitute'
    PARAM_CFG_AGGREGATE = 'aggregate'
    AGGR_FIELD = 'field'
    AGGR_FUNC = 'func'

    extract_params = params.get(PARAM_CFG_EXTRACT)
    extract_params.update({AccessParams.KEY_TYPE: AccessParams.TYPE_MULTI})
    dataset = __extract(input, extract_params)
    if PARAM_CFG_SUBSTITUTE in params:
        dataset = __substitute(input, dataset, params.get(PARAM_CFG_SUBSTITUTE))
    cfg = params.get(PARAM_CFG_AGGREGATE)
    res = Aggregator.agg_single_func(dataset, cfg[AGGR_FIELD], cfg[AGGR_FUNC])
    return res
371259
def get_aggregation(self, name):
        '''
        
        '''
        agg = self.aggregations[name]
        if 'buckets' in agg:
            return agg['buckets']
        else:
            return agg
208277
def acquire_lock(self):
		
		# first ensure that a record exists for this session id
		try:
			self.collection.insert_one(dict(_id=self.id))
		except pymongo.errors.DuplicateKeyError:
			pass
		unlocked_spec = dict(_id=self.id, locked=None)
		lock_timer = (
			timers.Timer.after(self.lock_timeout)
			if self.lock_timeout
			else timers.NeverExpires()
		)
		while not lock_timer.expired():
			locked_spec = {'$set': dict(locked=datetime.datetime.utcnow())}
			res = self.collection.update_one(unlocked_spec, locked_spec)
			if res.raw_result['updatedExisting']:
				# we have the lock
				break
			time.sleep(0.1)
		else:
			raise LockTimeout(f"Timeout acquiring lock for {self.id}")
		self.locked = True
919025
def get(cls, id=None, condition=None, fields=None, cache=False, engine_name=None, **kwargs):
        
        
        if id is None and condition is None:
            return None
        
        can_cacheable = (cache or getattr(cls, '__cacheable__', None)) and \
            isinstance(id, (int, long, str, unicode))
        if can_cacheable:
            #send 'get_object' topic to get cached object
            obj = dispatch.get(cls, 'get_object', id)
            if obj:
                return obj

        if condition is not None:
            _cond = condition
        else:
            if is_condition(id):
                _cond = id
            else:
                _cond = cls.c[cls._primary_field] == id
            #todo
            # if isinstance(id, (int, long)):
            #     _cond = cls.c.id==id
            # elif isinstance(id, (str, unicode)) and id.isdigit():
            #     _cond = cls.c.id==int(id)
            # else:
            #     _cond = id

        #if there is no cached object, then just fetch from database
        obj = cls.filter(_cond, **kwargs).fields(*(fields or [])).one()
        
        if obj and cache or getattr(cls, '__cacheable__', None):
            dispatch.call(cls, 'set_object', instance=obj)

        return obj
906986
@SuppressWarnings("unchecked")
	public static <T> List<T> getAny(Collection<T> collection, int... indexes) {
		final int size = collection.size();
		final ArrayList<T> result = new ArrayList<>();
		if (collection instanceof List) {
			final List<T> list = ((List<T>) collection);
			for (int index : indexes) {
				if (index < 0) {
					index += size;
				}
				result.add(list.get(index));
			}
		} else {
			Object[] array = ((Collection<T>) collection).toArray();
			for (int index : indexes) {
				if (index < 0) {
					index += size;
				}
				result.add((T) array[index]);
			}
		}
		return result;
	}
106470
function AggregateByKey(sc, dependencies, reducer, combiner, init, args) {
  Dataset.call(this, sc, dependencies);
  this.combiner = combiner;
  this.reducer = reducer;
  this.init = init;
  this.args = args;
  this.shuffling = true;
  this.executed = false;
  this.buffer = {};
  this.type = 'AggregateByKey';
}
1071645
def action_aggregate(reader, *args):
    """"""
    all_aggregated = aggregated_records(reader)
    first_row = next(all_aggregated)
    keys = sorted(first_row.keys())
    print(*keys, sep='\t')

    # Join the first row with the rest of the rows and print them
    iterable = chain([first_row], all_aggregated)
    for item in iterable:
        print(*[item[k] for k in keys], sep='\t')