Merge pull request #236 from s22s/fix/62

Fixed `Add` expression to use GT behavior

Author: metasim

core/src/main/scala/org/locationtech/rasterframes/expressions/aggregates/LocalMeanAggregate.scala

@@ -22,7 +22,7 @@
 package org.locationtech.rasterframes.expressions.aggregates
 
 import org.locationtech.rasterframes.expressions.UnaryRasterAggregate
-import org.locationtech.rasterframes.expressions.localops.{Add => AddTiles, Divide => DivideTiles}
+import org.locationtech.rasterframes.expressions.localops.{BiasedAdd, Divide => DivideTiles}
 import org.locationtech.rasterframes.expressions.transformers.SetCellType
 import geotrellis.raster.Tile
 import geotrellis.raster.mapalgebra.local
@@ -59,16 +59,16 @@ case class LocalMeanAggregate(child: Expression) extends UnaryRasterAggregate {
   override lazy val updateExpressions: Seq[Expression] = Seq(
     If(IsNull(count),
       SetCellType(Defined(child), Literal("int32")),
-      If(IsNull(child), count, AddTiles(count, Defined(child)))
+      If(IsNull(child), count, BiasedAdd(count, Defined(child)))
     ),
     If(IsNull(sum),
       SetCellType(child, Literal("float64")),
-      If(IsNull(child), sum, AddTiles(sum, child))
+      If(IsNull(child), sum, BiasedAdd(sum, child))
     )
   )
   override val mergeExpressions: Seq[Expression] = Seq(
-    AddTiles(count.left, count.right),
-    AddTiles(sum.left, sum.right)
+    BiasedAdd(count.left, count.right),
+    BiasedAdd(sum.left, sum.right)
   )
   override lazy val evaluateExpression: Expression = DivideTiles(sum, count)
 }

core/src/main/scala/org/locationtech/rasterframes/expressions/localops/Add.scala

@@ -30,7 +30,6 @@ import org.apache.spark.sql.{Column, TypedColumn}
 import org.locationtech.rasterframes._
 import org.locationtech.rasterframes.expressions.BinaryLocalRasterOp
 import org.locationtech.rasterframes.expressions.DynamicExtractors.tileExtractor
-import org.locationtech.rasterframes.util.DataBiasedOp.BiasedAdd
 
 @ExpressionDescription(
   usage = "_FUNC_(tile, rhs) - Performs cell-wise addition between two tiles or a tile and a scalar.",
@@ -48,9 +47,9 @@ import org.locationtech.rasterframes.util.DataBiasedOp.BiasedAdd
 case class Add(left: Expression, right: Expression) extends BinaryLocalRasterOp
   with CodegenFallback {
   override val nodeName: String = "rf_local_add"
-  override protected def op(left: Tile, right: Tile): Tile = BiasedAdd(left, right)
-  override protected def op(left: Tile, right: Double): Tile = BiasedAdd(left, right)
-  override protected def op(left: Tile, right: Int): Tile = BiasedAdd(left, right)
+  override protected def op(left: Tile, right: Tile): Tile = left.localAdd(right)
+  override protected def op(left: Tile, right: Double): Tile = left.localAdd(right)
+  override protected def op(left: Tile, right: Int): Tile = left.localAdd(right)
 
   override def eval(input: InternalRow): Any = {
     if(input == null) null

core/src/main/scala/org/locationtech/rasterframes/expressions/localops/BiasedAdd.scala

@@ -0,0 +1,74 @@
+/*
+ * This software is licensed under the Apache 2 license, quoted below.
+ *
+ * Copyright 2019 Astraea, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not
+ * use this file except in compliance with the License. You may obtain a copy of
+ * the License at
+ *
+ *     [http://www.apache.org/licenses/LICENSE-2.0]
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations under
+ * the License.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ */
+
+package org.locationtech.rasterframes.expressions.localops
+import geotrellis.raster.Tile
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.codegen.CodegenFallback
+import org.apache.spark.sql.catalyst.expressions.{Expression, ExpressionDescription}
+import org.apache.spark.sql.functions.lit
+import org.apache.spark.sql.{Column, TypedColumn}
+import org.locationtech.rasterframes._
+import org.locationtech.rasterframes.expressions.BinaryLocalRasterOp
+import org.locationtech.rasterframes.expressions.DynamicExtractors.tileExtractor
+import org.locationtech.rasterframes.util.DataBiasedOp
+
+@ExpressionDescription(
+  usage = "_FUNC_(tile, rhs) - Performs cell-wise addition between two tiles or a tile and a scalar. " +
+    "Unlike a regular 'add', this considers `<data> + <nodata> = <data>.",
+  arguments = """
+  Arguments:
+    * tile - left-hand-side tile
+    * rhs  - a tile or scalar value to add to each cell""",
+  examples = """
+  Examples:
+    > SELECT _FUNC_(tile, 1.5);
+       ...
+    > SELECT _FUNC_(tile1, tile2);
+       ..."""
+)
+case class BiasedAdd(left: Expression, right: Expression) extends BinaryLocalRasterOp
+  with CodegenFallback {
+  override val nodeName: String = "rf_local_biased_add"
+  override protected def op(left: Tile, right: Tile): Tile = DataBiasedOp.BiasedAdd(left, right)
+  override protected def op(left: Tile, right: Double): Tile = DataBiasedOp.BiasedAdd(left, right)
+  override protected def op(left: Tile, right: Int): Tile = DataBiasedOp.BiasedAdd(left, right)
+
+  override def eval(input: InternalRow): Any = {
+    if(input == null) null
+    else {
+      val l = left.eval(input)
+      val r = right.eval(input)
+      if (l == null && r == null) null
+      else if (l == null) r
+      else if (r == null && tileExtractor.isDefinedAt(right.dataType)) l
+      else if (r == null) null
+      else nullSafeEval(l, r)
+    }
+  }
+}
+object BiasedAdd {
+  def apply(left: Column, right: Column): TypedColumn[Any, Tile] =
+    new Column(BiasedAdd(left.expr, right.expr)).as[Tile]
+
+  def apply[N: Numeric](tile: Column, value: N): TypedColumn[Any, Tile] =
+    new Column(BiasedAdd(tile.expr, lit(value).expr)).as[Tile]
+}

pyrasterframes/src/main/python/docs/reference.pymd

@@ -238,6 +238,8 @@ If input `tile` had a NoData value already, the behaviour depends on if its cell
 
 [Local map algebra](https://gisgeography.com/map-algebra-global-zonal-focal-local/) raster operations are element-wise operations on a single tile (unary), between a `tile` and a scalar, between two `tile`s, or across many `tile`s.
 
+When these operations encounter a NoData value in either operand, the cell in the resulting `tile` will have a NoData.
+
 The binary local map algebra functions have similar variations in the Python API depending on the left hand side type:
 
  - `rf_local_op`: applies `op` to two columns; the right hand side can be a `tile` or a numeric column.
@@ -536,6 +538,8 @@ Aggregates over all of the rows in DataFrame of `tile` and returns a count of ea
 
 Local statistics compute the element-wise statistics across a DataFrame or group of `tile`s, resulting in a `tile` that has the same dimension.
 
+When these functions encounter a NoData in a cell location, it will be ignored. 
+
 ### rf_agg_local_max
 
     Tile rf_agg_local_max(Tile tile)

pyrasterframes/src/main/python/setup.py

@@ -144,15 +144,16 @@ def initialize_options(self):
         'setuptools>=0.8',
         'ipython==6.2.1',
         "ipykernel==4.8.0",
-        'Pweave==0.30.3'
+        'Pweave==0.30.3',
     ],
     tests_require=[
         'pytest==3.4.2',
         'pypandoc',
         'numpy>=1.7',
         'shapely',
         'pandas',
-        'rasterio'
+        'rasterio',
+        'boto3'
     ],
     packages=[
         'pyrasterframes',

pyrasterframes/src/main/python/tests/GeoTiffWriterTests.py

@@ -37,7 +37,7 @@ def test_identity_write(self):
         dest = self._tmpfile()
         rf.write.geotiff(dest)
 
-        rf2 = self.spark.read.geotiff(dest)
+        rf2 = self.spark.read.geotiff('file://' + dest)
         self.assertEqual(rf2.count(), rf.count())
 
         os.remove(dest)
@@ -47,7 +47,7 @@ def test_unstructured_write(self):
         dest = self._tmpfile()
         rf.write.geotiff(dest, crs='EPSG:32616')
 
-        rf2 = self.spark.read.raster(dest)
+        rf2 = self.spark.read.raster('file://' + dest)
         self.assertEqual(rf2.count(), rf.count())
 
         os.remove(dest)

pyrasterframes/src/main/python/tests/PyRasterFramesTests.py

@@ -28,7 +28,6 @@
 from . import TestEnvironment
 
 
-
 class CellTypeHandling(unittest.TestCase):
 
     def test_is_raw(self):
@@ -237,11 +236,16 @@ def less_pi(t):
 class TileOps(TestEnvironment):
 
     def setUp(self):
+        from pyspark.sql import Row
         # convenience so we can assert around Tile() == Tile()
         self.t1 = Tile(np.array([[1, 2],
                                  [3, 4]]), CellType.int8().with_no_data_value(3))
         self.t2 = Tile(np.array([[1, 2],
                                  [3, 4]]), CellType.int8().with_no_data_value(1))
+        self.t3 = Tile(np.array([[1,  2],
+                                 [-3, 4]]), CellType.int8().with_no_data_value(3))
+
+        self.df = self.spark.createDataFrame([Row(t1=self.t1, t2=self.t2, t3=self.t3)])
 
     def test_addition(self):
         e1 = np.ma.masked_equal(np.array([[5, 6],
@@ -253,6 +257,9 @@ def test_addition(self):
         r2 = (self.t1 + self.t2).cells
         self.assertTrue(np.ma.allequal(r2, e2))
 
+        col_result = self.df.select(rf_local_add('t1', 't3').alias('sum')).first()
+        self.assertEqual(col_result.sum, self.t1 + self.t3)
+
     def test_multiplication(self):
         e1 = np.ma.masked_equal(np.array([[4, 8],
                                           [12, 16]]), 12)
@@ -263,6 +270,9 @@ def test_multiplication(self):
         r2 = (self.t1 * self.t2).cells
         self.assertTrue(np.ma.allequal(r2, e2))
 
+        r3 = self.df.select(rf_local_multiply('t1', 't3').alias('r3')).first().r3
+        self.assertEqual(r3, self.t1 * self.t3)
+
     def test_subtraction(self):
         t3 = self.t1 * 4
         r1 = t3 - self.t1
@@ -541,7 +551,6 @@ def path(scene, band):
         self.assertTrue(df2.select('b1_path').distinct().count() == 3)
 
 
-
 def suite():
     function_tests = unittest.TestSuite()
     return function_tests

pyrasterframes/src/main/python/tests/RasterFunctionsTests.py

@@ -97,6 +97,21 @@ def test_agg_mean(self):
         mean = self.rf.agg(rf_agg_mean('tile')).first()['rf_agg_mean(tile)']
         self.assertTrue(self.rounded_compare(mean, 10160))
 
+    def test_agg_local_mean(self):
+        from pyspark.sql import Row
+
+        # this is really testing the nodata propagation in the agg  local summation
+        ct = CellType.int8().with_no_data_value(4)
+        df = self.spark.createDataFrame([
+            Row(tile=Tile(np.array([[1, 2, 3, 4, 5, 6]]), ct)),
+            Row(tile=Tile(np.array([[1, 2, 4, 3, 5, 6]]), ct)),
+        ])
+
+        result = df.agg(rf_agg_local_mean('tile').alias('mean')).first().mean
+
+        expected = Tile(np.array([[1.0, 2.0, 3.0, 3.0, 5.0, 6.0]]), CellType.float64())
+        self.assertEqual(result, expected)
+
     def test_aggregations(self):
         aggs = self.rf.agg(
             rf_agg_data_cells('tile'),
@@ -112,28 +127,59 @@ def test_aggregations(self):
         self.assertEqual(row['rf_agg_stats(tile)'].data_cells, row['rf_agg_data_cells(tile)'])
 
     def test_sql(self):
+
         self.rf.createOrReplaceTempView("rf_test_sql")
 
-        self.spark.sql("""SELECT tile, 
-                            rf_local_add(tile, 1) AS and_one, 
-                            rf_local_subtract(tile, 1) AS less_one, 
-                            rf_local_multiply(tile, 2) AS times_two, 
-                            rf_local_divide(tile, 2) AS over_two 
-                        FROM rf_test_sql""").createOrReplaceTempView('rf_test_sql_1')
-
-        statsRow = self.spark.sql("""
-        SELECT rf_tile_mean(tile) as base,
-            rf_tile_mean(and_one) as plus_one,
-            rf_tile_mean(less_one) as minus_one,
-            rf_tile_mean(times_two) as double,
-            rf_tile_mean(over_two) as half
-        FROM rf_test_sql_1
-        """).first()
-
-        self.assertTrue(self.rounded_compare(statsRow.base, statsRow.plus_one - 1))
-        self.assertTrue(self.rounded_compare(statsRow.base, statsRow.minus_one + 1))
-        self.assertTrue(self.rounded_compare(statsRow.base, statsRow.double / 2))
-        self.assertTrue(self.rounded_compare(statsRow.base, statsRow.half * 2))
+        arith = self.spark.sql("""SELECT tile, 
+                                rf_local_add(tile, 1) AS add_one, 
+                                rf_local_subtract(tile, 1) AS less_one, 
+                                rf_local_multiply(tile, 2) AS times_two, 
+                                rf_local_divide(
+                                    rf_convert_cell_type(tile, "float32"), 
+                                    2) AS over_two 
+                            FROM rf_test_sql""")
+
+        arith.createOrReplaceTempView('rf_test_sql_1')
+        arith.show(truncate=False)
+        stats = self.spark.sql("""
+            SELECT rf_tile_mean(tile) as base,
+                rf_tile_mean(add_one) as plus_one,
+                rf_tile_mean(less_one) as minus_one,
+                rf_tile_mean(times_two) as double,
+                rf_tile_mean(over_two) as half,
+                rf_no_data_cells(tile) as nd
+                
+            FROM rf_test_sql_1
+            ORDER BY rf_no_data_cells(tile)
+            """)
+        stats.show(truncate=False)
+        stats.createOrReplaceTempView('rf_test_sql_stats')
+
+        compare = self.spark.sql("""
+            SELECT 
+                plus_one - 1.0 = base as add,
+                minus_one + 1.0 = base as subtract,
+                double / 2.0 = base as multiply,
+                half * 2.0 = base as divide,
+                nd
+            FROM rf_test_sql_stats
+            """)
+
+        expect_row1 = compare.orderBy('nd').first()
+
+        self.assertTrue(expect_row1.subtract)
+        self.assertTrue(expect_row1.multiply)
+        self.assertTrue(expect_row1.divide)
+        self.assertEqual(expect_row1.nd, 0)
+        self.assertTrue(expect_row1.add)
+
+        expect_row2 = compare.orderBy('nd', ascending=False).first()
+
+        self.assertTrue(expect_row2.subtract)
+        self.assertTrue(expect_row2.multiply)
+        self.assertTrue(expect_row2.divide)
+        self.assertTrue(expect_row2.nd > 0)
+        self.assertTrue(expect_row2.add)  # <-- Would fail in a case where ND + 1 = 1
 
     def test_explode(self):
         import pyspark.sql.functions as F