docs: @Override
    public long scanAndAggregate(
        HistoricalDimensionSelector 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();
        double metric = metricSelector.getDouble(rowNum);
        final IndexedInts dimValues = dimensionSelector.getRow(rowNum);
        final int dimSize = dimValues.size();
        for (int i = 0; i < dimSize; i++) {
          int dimIndex = dimValues.get(i);
          int position = positions[dimIndex];
          if (position >= 0) {
            aggregator.aggregate(resultsBuffer, position, metric);
          } else if (position == TopNAlgorithm.INIT_POSITION_VALUE) {
            positions[dimIndex] = positionToAllocate;
            aggregator.putFirst(resultsBuffer, positionToAllocate, metric);
            positionToAllocate += aggregatorSize;
          }
        }
        processedRows++;
        offset.increment();
      }
      return processedRows;
    }