Merge pull request #500 from s22s/fix/499-raster-join-coalesce

Fix RasterJoin coalesce issue

Author: vpipkt

core/src/main/scala/org/locationtech/rasterframes/extensions/RasterJoin.scala

@@ -104,7 +104,7 @@ object RasterJoin {
     // Assumes all LHS tiles in a row are of the same size.
     val destDims =
       if (left.tileColumns.nonEmpty)
-        rf_dimensions(coalesce(left.tileColumns.map(unresolved): _*))
+        coalesce(left.tileColumns.map(unresolved).map(rf_dimensions): _*)
       else
         serialized_literal(fallbackDimensions.getOrElse(NOMINAL_TILE_DIMS))
 

core/src/test/scala/org/locationtech/rasterframes/RasterJoinSpec.scala

@@ -172,6 +172,16 @@ class RasterJoinSpec extends TestEnvironment with TestData with RasterMatchers {
       }
     }
 
+    it("should raster join multiple times on projected raster"){
+      val df0 = Seq(one).toDF("proj_raster")
+      val result = df0.select($"proj_raster" as "t1")
+        .rasterJoin(df0.select($"proj_raster" as "t2"))
+        .rasterJoin(df0.select($"proj_raster" as "t3"))
+
+      result.tileColumns.length should be (3)
+      result.count() should be (1)
+    }
+
     it("should honor resampling options") {
       // test case. replicate existing test condition and check that resampling option results in different output
       val filterExpr = st_intersects(rf_geometry($"tile"), st_point(704940.0, 4251130.0))

core/src/test/scala/org/locationtech/rasterframes/functions/TileFunctionsSpec.scala

@@ -27,7 +27,7 @@ import geotrellis.raster._
 import geotrellis.raster.testkit.RasterMatchers
 import javax.imageio.ImageIO
 import org.apache.spark.sql.Encoders
-import org.apache.spark.sql.functions.sum
+import org.apache.spark.sql.functions.{count, sum, isnull}
 import org.locationtech.rasterframes._
 import org.locationtech.rasterframes.ref.RasterRef
 import org.locationtech.rasterframes.tiles.ProjectedRasterTile
@@ -354,6 +354,17 @@ class TileFunctionsSpec extends TestEnvironment with RasterMatchers {
       t should be(randPRT.dimensions)
       checkDocs("rf_dimensions")
     }
+
+    it("should get null for null tile dimensions") {
+      val result = (Seq(randPRT) :+ null).toDF("tile")
+        .select(rf_dimensions($"tile") as "dim")
+        .select(isnull($"dim").cast("long") as "n")
+        .agg(sum("n"), count("n"))
+        .first()
+      result.getAs[Long](0) should be (1)
+      result.getAs[Long](1) should be (2)
+    }
+
     it("should get the Extent of a ProjectedRasterTile") {
       val e = Seq(randPRT).toDF("tile").select(rf_extent($"tile")).first()
       e should be(extent)