Merge pull request #2034 from metasim/fix/2033

Fix/2033
locationtech · Mar 9, 2017 · 94dce0c · 94dce0c
2 parents 147c6a2 + 73bbe2a
commit 94dce0c
Show file tree

Hide file tree

Showing 8 changed files with 123 additions and 36 deletions.
diff --git a/.sbtopts b/.sbtopts
@@ -1,2 +1,3 @@
+-J-Xmx2g
 -Djava.awt.headless=true
 -Dsun.io.serialization.extendedDebugInfo=true
diff --git a/spark/src/main/scala/geotrellis/spark/io/avro/codecs/CRSCodec.scala b/spark/src/main/scala/geotrellis/spark/io/avro/codecs/CRSCodec.scala
@@ -0,0 +1,53 @@
+/*
+ * Copyright 2017 Azavea
+ *
+ * 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.
+ */
+
+package geotrellis.spark.io.avro.codecs
+
+import geotrellis.proj4.CRS
+import geotrellis.spark.io.avro.{AvroRecordCodec, _}
+import org.apache.avro.generic.GenericRecord
+import org.apache.avro.util.Utf8
+import org.apache.avro.{Schema, SchemaBuilder}
+
+trait CRSCodec {
+  implicit def crsCodec = new AvroRecordCodec[CRS] {
+    override def schema: Schema = SchemaBuilder
+      .record("CRS").namespace("geotrellis.spark")
+      .fields()
+      .name("epsg").`type`().optional().intType()
+      .name("proj4").`type`().optional().stringType()
+      .endRecord()
+
+    override def encode(crs: CRS, rec: GenericRecord): Unit = {
+      if(crs.epsgCode.isDefined) {
+        rec.put("epsg", crs.epsgCode.get)
+      }
+      else {
+        rec.put("proj4", crs.toProj4String)
+      }
+    }
+    override def decode(rec: GenericRecord): CRS = {
+      if(rec[AnyRef]("epsg") != null) {
+        val epsg = rec[Int]("epsg")
+        CRS.fromEpsgCode(epsg)
+      }
+      else {
+        val proj4 = rec[Utf8]("proj4")
+        CRS.fromString(proj4.toString)
+      }
+    }
+  }
+}
diff --git a/spark/src/main/scala/geotrellis/spark/io/avro/codecs/Implicits.scala b/spark/src/main/scala/geotrellis/spark/io/avro/codecs/Implicits.scala
@@ -26,6 +26,7 @@ trait Implicits
     with ConstantTileCodecs
     with TileFeatureCodec
     with VectorTileCodec
+    with CRSCodec
     with ExtentCodec
     with ProjectedExtentCodec
     with TemporalProjectedExtentCodec

diff --git a/spark/src/main/scala/geotrellis/spark/io/avro/codecs/ProjectedExtentCodec.scala b/spark/src/main/scala/geotrellis/spark/io/avro/codecs/ProjectedExtentCodec.scala
@@ -16,36 +16,33 @@
 
 package geotrellis.spark.io.avro.codecs
 
-import geotrellis.proj4.CRS
-import geotrellis.spark._
 import geotrellis.spark.io.avro._
-import geotrellis.spark.io.avro.codecs._
 import geotrellis.spark.io.avro.codecs.Implicits._
 import geotrellis.vector._
-
 import org.apache.avro._
 import org.apache.avro.generic._
 
 // --- //
 
 trait ProjectedExtentCodec {
   implicit def projectedExtentCodec = new AvroRecordCodec[ProjectedExtent] {
-    def schema: Schema = SchemaBuilder
-      .record("ProjectedExtent").namespace("geotrellis.vector")
-      .fields()
-      .name("extent").`type`(extentCodec.schema).noDefault()
-      .name("epsg").`type`().intType().noDefault()
-      .endRecord()
+    def schema: Schema = {
+      val base = SchemaBuilder
+        .record("ProjectedExtent").namespace("geotrellis.vector")
+        .fields()
+        .name("extent").`type`(extentCodec.schema).noDefault()
+
+      injectFields(crsCodec.schema, base)
+        .endRecord()
+    }
 
     def encode(projectedExtent: ProjectedExtent, rec: GenericRecord): Unit = {
       rec.put("extent", extentCodec.encode(projectedExtent.extent))
-      rec.put("epsg", projectedExtent.crs.epsgCode.get)
+      crsCodec.encode(projectedExtent.crs, rec)
     }
 
     def decode(rec: GenericRecord): ProjectedExtent = {
-      val epsg = rec[Int]("epsg")
-      val crs = CRS.fromEpsgCode(epsg)
-
+      val crs = crsCodec.decode(rec)
       val extent = extentCodec.decode(rec[GenericRecord]("extent"))
 
       ProjectedExtent(extent, crs)

diff --git a/spark/src/main/scala/geotrellis/spark/io/avro/codecs/TemporalProjectedExtentCodec.scala b/spark/src/main/scala/geotrellis/spark/io/avro/codecs/TemporalProjectedExtentCodec.scala
@@ -16,38 +16,35 @@
 
 package geotrellis.spark.io.avro.codecs
 
-import geotrellis.proj4.CRS
 import geotrellis.spark._
 import geotrellis.spark.io.avro._
-import geotrellis.spark.io.avro.codecs._
 import geotrellis.spark.io.avro.codecs.Implicits._
-import geotrellis.vector._
-
 import org.apache.avro._
 import org.apache.avro.generic._
 
 
 trait TemporalProjectedExtentCodec {
   implicit def temporalProjectedExtentCodec = new AvroRecordCodec[TemporalProjectedExtent] {
-    def schema: Schema = SchemaBuilder
-      .record("TemporalProjectedExtent").namespace("geotrellis.spark")
-      .fields()
-      .name("extent").`type`(extentCodec.schema).noDefault()
-      .name("epsg").`type`().intType().noDefault()
-      .name("instant").`type`().longType().noDefault()
-      .endRecord()
+    def schema: Schema = {
+      val base = SchemaBuilder
+        .record("TemporalProjectedExtent").namespace("geotrellis.spark")
+        .fields()
+        .name("extent").`type`(extentCodec.schema).noDefault()
+
+      injectFields(crsCodec.schema, base)
+        .name("instant").`type`().longType().noDefault()
+        .endRecord()
+    }
 
     def encode(temporalProjectedExtent: TemporalProjectedExtent, rec: GenericRecord): Unit = {
       rec.put("extent", extentCodec.encode(temporalProjectedExtent.extent))
-      rec.put("epsg", temporalProjectedExtent.crs.epsgCode.get)
+      crsCodec.encode(temporalProjectedExtent.crs, rec)
       rec.put("instant", temporalProjectedExtent.instant)
     }
 
     def decode(rec: GenericRecord): TemporalProjectedExtent = {
       val instant = rec[Long]("instant")
-      val epsg = rec[Int]("epsg")
-      val crs = CRS.fromEpsgCode(epsg)
-
+      val crs = crsCodec.decode(rec)
       val extent = extentCodec.decode(rec[GenericRecord]("extent"))
 
       TemporalProjectedExtent(extent, crs, instant)

diff --git a/spark/src/main/scala/geotrellis/spark/io/avro/codecs/package.scala b/spark/src/main/scala/geotrellis/spark/io/avro/codecs/package.scala
@@ -0,0 +1,31 @@
+/*
+ * Copyright 2017 Azavea
+ *
+ * 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.
+ */
+
+package geotrellis.spark.io.avro
+
+import org.apache.avro.Schema
+import org.apache.avro.SchemaBuilder.FieldAssembler
+
+import scala.collection.JavaConversions._
+
+package object codecs {
+  private[codecs] def injectFields(from: Schema, to: FieldAssembler[Schema]): FieldAssembler[Schema] = {
+    from.getFields.foldLeft(to){
+      case (builder, field) ⇒
+        builder.name(field.name()).`type`(field.schema()).noDefault()
+    }
+  }
+}
diff --git a/spark/src/main/scala/geotrellis/spark/io/avro/package.scala b/spark/src/main/scala/geotrellis/spark/io/avro/package.scala
@@ -17,12 +17,9 @@
 package geotrellis.spark.io
 
 import geotrellis.util._
-
 import org.apache.avro._
 import org.apache.avro.generic._
 
-import scala.reflect.ClassTag
-
 package object avro {
   implicit class withGenericRecordMethods(val self: GenericRecord) extends MethodExtensions[GenericRecord] {
     def apply[X](name: String) = self.get(name).asInstanceOf[X]

diff --git a/spark/src/test/scala/geotrellis/spark/io/avro/ExtentCodecsSpec.scala b/spark/src/test/scala/geotrellis/spark/io/avro/ExtentCodecsSpec.scala
@@ -16,23 +16,33 @@
 
 package geotrellis.spark.io.avro
 
-import geotrellis.spark.io.avro.codecs._
 import org.scalatest._
 import geotrellis.proj4._
 import geotrellis.vector._
 import geotrellis.spark._
-import geotrellis.spark.io.avro.AvroTools._
 
 class ExtentCodecsSpec extends FunSpec with Matchers with AvroTools  {
+  val fortyThreeTwentyFour = CRS.fromEpsgCode(4324)
+  // A proj4 CRS known to not have a CRS code.
+  val nonEpsgCrs = CRS.fromString("+proj=sinu +lon_0=0.0 +x_0=0.0 +y_0=0.0 +a=6371007.181 +b=6371007.181 +units=m")
+
   describe("ExtentCodecs") {
     it("encodes Extent"){
       roundTrip(Extent(0, 1, 2, 3))
     }
     it("encodes ProjectedExtent") {
-      roundTrip(ProjectedExtent(Extent(0, 1, 2, 3), CRS.fromEpsgCode(4324)))
+      roundTrip(ProjectedExtent(Extent(0, 1, 2, 3), fortyThreeTwentyFour))
+    }
+    it("encodes ProjectedExtent with missing EPSG code"){
+      assert(nonEpsgCrs.epsgCode.isEmpty)
+      roundTrip(ProjectedExtent(Extent(0, 1, 2, 3), nonEpsgCrs))
     }
     it("encodes TemporalProjectedExtent"){
-      roundTrip(TemporalProjectedExtent(Extent(0, 1, 2, 3), CRS.fromEpsgCode(4324), 1.toLong))
+      roundTrip(TemporalProjectedExtent(Extent(0, 1, 2, 3), fortyThreeTwentyFour, 1.toLong))
+    }
+    it("encodes TemporalProjectedExtent with missing EPSG code"){
+      assert(nonEpsgCrs.epsgCode.isEmpty)
+      roundTrip(TemporalProjectedExtent(Extent(0, 1, 2, 3), nonEpsgCrs, 1.toLong))
     }
   }
 }