Scratch

partionWindows.scala

  def partitionWindows(
    segmentLayout: GeoTiffSegmentLayout, 
    windows: IndexedSeq[GridBounds], 
    maxPartitionSize: Long
  ): Array[Array[GridBounds]] = {
  
    val partition = mutable.ArrayBuilder.make[GridBounds]
    val partitions = mutable.ArrayBuilder.make[Array[GridBounds]]
    var partitionSize: Long = 0l
    var windowsProcessed: Int = 0

    type WindowIndex = Int
    val InvalidPriority = Short.MinValue
    val priorities = Array.fill[Short](windows.length)(Short.MinValue + 1)

    // mapping of segment index to overlapping window indecies
    val segmentWindows: mutable.Map[Int, mutable.Set[WindowIndex]] = {
      val res = mutable.Map.empty[Int, mutable.Set[WindowIndex]]
      for { 
        wi <- 0 until windows.length
        si <- segmentLayout.intersectingSegments(windows(wi))  
      } {
        val windowSet = res.getOrElseUpdate(si, mutable.Set.empty[WindowIndex])
        windowSet += wi
      }
      res
    }

    def incrementPrioirty(i: WindowIndex) = {
      if (priorities(i) != InvalidPriority) priorities(i) = (priorities(i) + 1).toShort
    }
    
    def popMostConnectedWindow(): GridBounds = {
      // find the index with max prioirt
      // TODO: use window position to perform the ranking
      val maxIndex = priorities.zipWithIndex.maxBy(_._1)._2
      priorities(maxIndex) = InvalidPriority
      windowsProcessed += 1
      windows(maxIndex)
    }

    def addToPartition (window: GridBounds) {
      partition += window
      partitionSize += window.sizeLong
    }

    def finalizePartition() {
      val res = partition.result
      if (res.nonEmpty) partitions += res
      partition.clear()
      partitionSize = 0l
    }

    while (windowsProcessed < windows.length) {
      val window = popMostConnectedWindow()
      for {
        si <- segmentLayout.intersectingSegments(window)
        wi <- segmentWindows(si)
      } incrementPrioirty(wi)

      if (partitionSize + window.sizeLong <= maxPartitionSize) {
        addToPartition(window)
      } else {
        finalizePartition()
        addToPartition(window)
      }
    }

    finalizePartition()
    partitions.result
  }

partitionWindowsByQuad.scala

 import com.vividsolutions.jts.index.quadtree.Quadtree
  import com.vividsolutions.jts.geom.Envelope
  import com.vividsolutions.jts.index.ItemVisitor
  import scala.collection.JavaConverters._


  def partitionWindowsByQuad(
    segmentLayout: GeoTiffSegmentLayout, 
    windows: IndexedSeq[GridBounds], 
    maxPartitionSize: Long
  ): Array[Array[GridBounds]] = {
    type SegmentIndex = Int
    def envelope(gb: GridBounds) = new Envelope(gb.colMin, gb.colMax, gb.rowMin, gb.rowMax)

    val windowSet = mutable.Set.empty[GridBounds]
    windowSet ++= windows
  
    val qt = new Quadtree()
    windows.foreach { gb => qt.insert(envelope(gb), gb) }
  
    val partitionSegments = mutable.Set.empty[Int]
    val addQueue = new mutable.PriorityQueue[(GridBounds, Array[Int], Int)]()(Ordering.by(_._3))
    val partitionSegmentEnvelope = new Envelope()

    val partition = mutable.ArrayBuilder.make[GridBounds]
    val partitions = mutable.ArrayBuilder.make[Array[GridBounds]]
    var partitionSize: Long = 0l
    

    def nextWindow(): (GridBounds, Array[Int], Int) = {
      if (addQueue.isEmpty) {
        qt.query(partitionSegmentEnvelope, new ItemVisitor {
          def visitItem(o: Any) {
            val window = o.asInstanceOf[GridBounds]
            val windowSegments = segmentLayout.intersectingSegments(window)
            val priority = windowSegments.filter(partitionSegments.contains).length
            addQueue.enqueue((window, windowSegments, priority)) //partitionSegments.intersect(windowSegments).size)
          }
        })
      }
      if (addQueue.nonEmpty)
        addQueue.dequeue()
      else {
        val window = windowSet.head
        val windowSegments = segmentLayout.intersectingSegments(window)
        (window, windowSegments, 1)
      }
    }


    def finalizePartition() {
      val res = partition.result
      if (res.nonEmpty) partitions += res
      partition.clear()
      partitionSize = 0l
      partitionSegments.clear()
      partitionSegmentEnvelope.setToNull()
      // do not disturb the addQueue 
      // next partitions can be started from previous point
    }

    def addToPartition (window: GridBounds, segments: Array[Int]) {
      qt.remove(envelope(window), window)
      windowSet.remove(window)
      partition += window
      partitionSize += window.sizeLong

      if (partitionSize >= maxPartitionSize) {
        finalizePartition()
      } else {
        segments.foreach { index =>
          partitionSegments += index
          val segmentBounds = segmentLayout.getGridBounds(index)
          partitionSegmentEnvelope.expandToInclude(envelope(segmentBounds))
        }
      }
    }

    while (windowSet.nonEmpty) {
      val (window, segments, _) = nextWindow()
      addToPartition(window, segments)
    }

    finalizePartition()
    partitions.result
  }

partitionWindowsBySegments.scala

 def partitionWindowsBySegments(
    segmentLayout: GeoTiffSegmentLayout, 
    windows: IndexedSeq[GridBounds], 
    maxPartitionSize: Long
  ): Array[Array[GridBounds]] = {
    // Because GeoTiff segment indecies are enumorated in row-major order
    // sorting windows by the min index also provides spatial order
    val sorted = windows.map { window =>
      window -> segmentLayout.getSegmentIndex(window.colMin, window.rowMin)
    }.sortBy(_._2).map(_._1)
            
    val partition = mutable.ArrayBuilder.make[GridBounds]
    partition.sizeHintBounded(128, windows)
    val partitions = mutable.ArrayBuilder.make[Array[GridBounds]]
    var partitionSize: Long = 0l
    var partitionCount: Long = 0l

    def finalizePartition() {
      val res = partition.result
      if (res.nonEmpty) partitions += res
      partition.clear()
      partitionSize = 0l
      partitionCount = 0l
    }

    for (window <- sorted) {
      if ((partitionCount == 0) || (partitionSize + window.sizeLong) < maxPartitionSize) {
        partition += window
        partitionSize += window.sizeLong
        partitionCount += 1
      } else {
        finalizePartition()
      }
    }

    finalizePartition()
    partitions.result
  }