Counts ranges with at least one value inside. Split results per partitions.

NonEmptyRangesCounter.scala

object NonEmptyRangesCounter {

  def count(source: ImmutableRoaringBitmap, ranges: ImmutableRoaringBitmap, partitionsEnds: Seq[Long]): Seq[Int] = {
    val rangesIter = ranges.getIntIterator
    val sourceIter = source.getIntIterator
    var partIdx = 0
    val results = Array.ofDim[Int](partitionsEnds.length)

    var distinct = 0
    var continue = rangesIter.hasNext
    var partEnd = partitionsEnds(partIdx)

    while (continue) {
      val rangeStart = Integer.toUnsignedLong(rangesIter.next())
      if (rangeStart == partEnd + 1) {
        results(partIdx) = distinct
        distinct = 0
        partIdx += 1
        partEnd = partitionsEnds(partIdx)
      }
      sourceIter.advanceIfNeeded(Math.min(rangeStart, partEnd + 1).toInt)
      if (sourceIter.hasNext) {
        val el = Integer.toUnsignedLong(sourceIter.next())
        if (el >= partEnd + 1) {
          results(partIdx) = distinct
          distinct = 0
          while (el >= partEnd + 1) {
            partIdx += 1
            partEnd = partitionsEnds(partIdx)
          }
        }
        distinct += (if (el >= rangeStart) 1 else 0)
        rangesIter.advanceIfNeeded(Math.min(el + 1, partEnd + 1).toInt)
        continue = rangesIter.hasNext
      } else {
        continue = false
      }
    }
    results(partIdx) = distinct //last one
    results.toSeq
  }