TODO: Merge raster to existing layer

[vsimko]

[aklink]

ManyLayersToMultibandLayer should solve this for both Tile and MultibandTile layers.

[vsimko]

e.g. merging single-band raster to existing single-band layer (NA values ignored, existing values overwritten)

[aklink]

OK. We have used the term "Merge" for two different things in the past.
ManyLayersToMultibandLayer is LayerStacking (e.g. 3 bands + 3 bands = 6 bands).
What we need here is an overlapping Mosaicing (e.g. 3 bands + 3bands := 3bands ).

[aklink]

WIP: merge (mosaicing / update)

• added biggis.landuse.api.mergeRddIntoLayer[K, V, M]
• added biggis.landuse.spark.examples.LayerUpdaterExample

Some ToDo's:

• check zoom levels (and/or projection), probably enable ZoomResample (Pyramiding)
• reduce updating/merge to changed area only (currently merge applies to whole layer) - maybe this is no issue, but maybe runtime concerns
• error handling in case merge/update fails (assert same zoom level etc.)

[aklink]

An other approach from Gitter (probably solving "unnecessary untouched tiles updates"), not tested yet:

  // initial steps skipped...
    val (zoom, reprojected): (Int, RDD[(SpatialKey, Tile)] with Metadata[TileLayerMetadata[SpatialKey]]) =
      TileLayerRDD(tiled, rasterMetaData)
        .reproject(WebMercator, layoutScheme, Bilinear)
    val attributeStore = HadoopAttributeStore(hdfsPath)

    val updater = HadoopLayerUpdater(hdfsPath)
    val writer = HadoopLayerWriter(hdfsPath)

    val keyIndex: KeyIndexMethod[SpatialKey] = ZCurveKeyIndexMethod
    val layerId = LayerId("hrdem", zoom)

    if(attributeStore.layerExists(layerId)) {
      // update existing layer using a custom merge function 
      updater.update(layerId, reprojected, (v1:Tile, v2:Tile) => v1.combineDouble(v2) { (z1, z2) => {
        if (isData(z1) && isData(z2)) (z1 + z2) / 2
        else if (isData(z1)) z1
        else if (isData(z2)) z2
        else {
          Double.NaN // TODO: investigate why returning NODATA is not producing the expected output 
        }
      }
      })
    } else {
      // get computed keybounds
      val initial = reprojected.metadata.bounds.get
      // arbitrary enlarge keybounds to ensure updates would be possible anywhere
      val enlarged = KeyBounds(
        initial.minKey.copy(
          col = 0,
          row = 0
        ),
        initial.maxKey.copy(
          col = Int.MaxValue,
          row = Int.MaxValue
        ))
      writer.write(layerId, reprojected, keyIndex.createIndex(enlarged))
    }