NonEmptyRangesCounter (no partitions)

NonEmptyRangesCounter.scala

private[evaluator] object NonEmptyRangesCounter {

  def count(source: ImmutableRoaringBitmap, ranges: ImmutableRoaringBitmap): Int = {
    val rangesIter = ranges.getIntIterator
    val sourceIter = source.getIntIterator

    var distinct = 0
    var continue = rangesIter.hasNext

    while (continue) {
      val rangeStart = rangesIter.next()
      sourceIter.advanceIfNeeded(rangeStart)
      if (sourceIter.hasNext) {
        val el = sourceIter.next()
        distinct += (if (el >= rangeStart) 1 else 0)
        rangesIter.advanceIfNeeded(el + 1)
        continue = rangesIter.hasNext
      } else {
        continue = false
      }
    }
    distinct
  }

}