docs: @Override
    public long scanAndAggregate(
        SingleValueHistoricalDimensionSelector dimensionSelector,
        HistoricalColumnSelector metricSelector,
        SimpleDoubleBufferAggregator aggregator,
        int aggregatorSize,
        HistoricalCursor cursor,
        int[] positions,
        ByteBuffer resultsBuffer
    )
    {
      // See TopNUtils.copyOffset() for explanation
      Offset offset = (Offset) TopNUtils.copyOffset(cursor);
      long processedRows = 0;
      int positionToAllocate = 0;
      while (offset.withinBounds() && !Thread.currentThread().isInterrupted()) {
        int rowNum = offset.getOffset();
        int dimIndex = dimensionSelector.getRowValue(rowNum);
        int position = positions[dimIndex];
        if (position >= 0) {
          aggregator.aggregate(resultsBuffer, position, metricSelector.getDouble(rowNum));
        } else if (position == TopNAlgorithm.INIT_POSITION_VALUE) {
          positions[dimIndex] = positionToAllocate;
          aggregator.putFirst(resultsBuffer, positionToAllocate, metricSelector.getDouble(rowNum));
          positionToAllocate += aggregatorSize;
        }
        processedRows++;
        offset.increment();
      }
      return processedRows;
    }