Feature/introduce shape rdd

scripts/spatial/loadSolrSpatialData.scala

@@ -6,9 +6,9 @@ import com.spatial4j.core.context.jts.JtsSpatialContext
 import com.spatial4j.core.io.ShapeIO
 import org.apache.spark.rdd.RDD
 import org.zouzias.spark.lucenerdd.spatial.shape._
-import org.zouzias.spark.lucenerdd.spatial.shape.ShapeLuceneRDD
 import org.zouzias.spark.lucenerdd._
 import org.zouzias.spark.lucenerdd.LuceneRDD
+import org.zouzias.spark.lucenerdd.spatial.shape.rdds.ShapeLuceneRDD
 
 import scala.reflect.ClassTag
 

scripts/spatial/loadSwissCities.scala

@@ -16,9 +16,9 @@
  */
 
 import org.zouzias.spark.lucenerdd.spatial.shape._
-import org.zouzias.spark.lucenerdd.spatial.shape.ShapeLuceneRDD
 import org.zouzias.spark.lucenerdd._
 import org.zouzias.spark.lucenerdd.LuceneRDD
+import org.zouzias.spark.lucenerdd.spatial.shape.rdds.ShapeLuceneRDD
 val df = spark.read.format("com.databricks.spark.csv").option("header", "false").option("inferSchema", "true").option("delimiter", "\t").load("src/test/resources/spatial/CH.txt")
 val swissCities = df.select("_c0", "_c1", "_c5", "_c4").map(row => ((row.getDouble(2), row.getDouble(3)), row.getString(1).toLowerCase()))
 val shapes = ShapeLuceneRDD(swissCities.rdd)

src/main/scala/org/zouzias/spark/lucenerdd/spatial/shape/ShapeLuceneRDDKryoRegistrator.scala

@@ -16,13 +16,14 @@
  */
 package org.zouzias.spark.lucenerdd.spatial.shape
 
+import com.esotericsoftware.kryo.Kryo
 import com.twitter.algebird.TopK
-import com.twitter.chill._
 import org.apache.spark.SparkConf
 import org.apache.spark.serializer.{KryoRegistrator, KryoSerializer}
 import org.zouzias.spark.lucenerdd.models.{SparkDoc, SparkFacetResult, SparkScoreDoc}
 import org.zouzias.spark.lucenerdd.response.{LuceneRDDResponse, LuceneRDDResponsePartition}
-import org.zouzias.spark.lucenerdd.spatial.shape.partition.ShapeLuceneRDDPartition
+import org.zouzias.spark.lucenerdd.spatial.shape.partition.impl.ShapeLuceneRDDPartition
+import org.zouzias.spark.lucenerdd.spatial.shape.rdds.ShapeLuceneRDD
 
 
 class ShapeLuceneRDDKryoRegistrator extends KryoRegistrator {
@@ -35,6 +36,32 @@ class ShapeLuceneRDDKryoRegistrator extends KryoRegistrator {
     kryo.register(classOf[LuceneRDDResponsePartition])
     kryo.register(classOf[SparkScoreDoc])
     kryo.register(classOf[TopK[_]])
+    kryo.register(classOf[SparkDoc])
+    kryo.register(classOf[Number])
+    kryo.register(classOf[java.lang.Double])
+    kryo.register(classOf[java.lang.Float])
+    kryo.register(classOf[java.lang.Integer])
+    kryo.register(classOf[java.lang.Long])
+    kryo.register(classOf[java.lang.Short])
+    kryo.register(classOf[scala.collection.mutable.WrappedArray$ofRef])
+    kryo.register(classOf[scala.collection.mutable.WrappedArray$ofFloat])
+    kryo.register(classOf[scala.collection.mutable.WrappedArray$ofDouble])
+    kryo.register(classOf[scala.collection.mutable.WrappedArray$ofInt])
+    kryo.register(classOf[scala.collection.mutable.WrappedArray$ofLong])
+    kryo.register(classOf[Array[String]])
+    kryo.register(classOf[Array[Number]])
+    kryo.register(classOf[Array[Float]])
+    kryo.register(classOf[Array[Int]])
+    kryo.register(classOf[Array[Long]])
+    kryo.register(classOf[Array[Double]])
+    kryo.register(classOf[Array[Boolean]])
+    kryo.register(classOf[Range])
+    kryo.register(classOf[scala.collection.immutable.Map[String, String]])
+    kryo.register(classOf[scala.collection.immutable.Map[String, Number]])
+    kryo.register(classOf[scala.collection.immutable.Map$EmptyMap$])
+    kryo.register(classOf[scala.collection.immutable.Set$EmptySet$])
+    kryo.register(classOf[scala.collection.immutable.Map[_, _]])
+    kryo.register(classOf[Array[scala.collection.immutable.Map[_, _]]])
     ()
   }
 }

src/main/scala/org/zouzias/spark/lucenerdd/spatial/shape/partition/AbstractShapeLuceneRDDPartition.scala

@@ -17,10 +17,17 @@
 package org.zouzias.spark.lucenerdd.spatial.shape.partition
 
 import org.zouzias.spark.lucenerdd.response.LuceneRDDResponsePartition
-import org.zouzias.spark.lucenerdd.spatial.shape.ShapeLuceneRDD.PointType
+import org.zouzias.spark.lucenerdd.spatial.shape.rdds.ShapeLuceneRDD
+import org.zouzias.spark.lucenerdd.spatial.shape.rdds.ShapeLuceneRDD.PointType
 
 import scala.reflect.ClassTag
 
+/**
+ * Partition class for [[ShapeLuceneRDD]]
+ *
+ * @tparam K Spatial type
+ * @tparam V Document type
+ */
 private[shape] abstract class AbstractShapeLuceneRDDPartition[K, V] extends Serializable {
 
   protected implicit def kTag: ClassTag[K]

src/main/scala/org/zouzias/spark/lucenerdd/spatial/shape/partition/AbstractShapeRDDPartition.scala

@@ -0,0 +1,117 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.zouzias.spark.lucenerdd.spatial.shape.partition
+
+import org.zouzias.spark.lucenerdd.spatial.shape.rdds.ShapeLuceneRDD.PointType
+import org.zouzias.spark.lucenerdd.spatial.shape.rdds.ShapeRDD
+import org.zouzias.spark.lucenerdd.spatial.shape.response.ShapeRDDResponsePartition
+
+import scala.reflect.ClassTag
+
+/**
+ * Partition class for [[ShapeRDD]]
+ * @tparam K Spatial parameter (geo-shape)
+ * @tparam V Additional information associated with shape
+ */
+private[shape] abstract class AbstractShapeRDDPartition[K, V] extends Serializable {
+
+  protected implicit def kTag: ClassTag[K]
+  protected implicit def vTag: ClassTag[V]
+
+  def size: Long
+
+  def iterator: Iterator[(Long, (K, V))]
+
+  def isDefined(key: K): Boolean
+
+  def close(): Unit
+
+  /**
+   * Nearest neighbour search
+   *
+   * @param point query point
+   * @param k number of neighbors to return
+   * @param searchString Lucene Query string
+   * @return
+   */
+  def knnSearch(point: PointType, k: Int, searchString: String): ShapeRDDResponsePartition
+
+  /**
+   * Search for points within a circle
+   *
+   * @param center center of circle
+   * @param radius radius of circle in kilometers (KM)
+   * @param k number of points to return
+   * @return
+   */
+  def circleSearch(center: PointType, radius: Double, k: Int, operationName: String)
+  : ShapeRDDResponsePartition
+
+  /**
+   * Spatial search with arbitrary shape
+   *
+   * @param shapeAsString Shape object represented as String
+   * @param k Number of results to return
+   * @param operationName Operation name, i.e., intersect, within, etc
+   * @return
+   */
+  def spatialSearch(shapeAsString: String, k: Int, operationName: String)
+  : ShapeRDDResponsePartition
+
+  /**
+   * Spatial search with point
+   *
+   * @param point Query point
+   * @param k Number of result to return
+   * @param operationName Operation name, i.e., intersect, within, etc
+   * @return
+   */
+  def spatialSearch(point: PointType, k: Int, operationName: String)
+  : ShapeRDDResponsePartition
+
+  /**
+   * Bounding box search with point and radius
+   *
+   * @param center given as (x, y)
+   * @param radius distance from center in kilometers (KM)
+   * @param k Number of results to return
+   * @param operationName Operation name, i.e., intersect, within, etc
+   * @return
+   */
+  def bboxSearch(center: PointType, radius: Double, k: Int, operationName: String)
+  : ShapeRDDResponsePartition
+
+  /**
+   * Bounding box search with lower left and upper right corners
+   *
+   * @param lowerLeft Lower left point
+   * @param upperRight Upper left point
+   * @param k Number of results
+   * @param operationName Operation name, i.e., intersect, within, etc
+   * @return
+   */
+  def bboxSearch(lowerLeft: PointType, upperRight: PointType, k: Int, operationName: String)
+  : ShapeRDDResponsePartition
+
+  /**
+   * Restricts the entries to those satisfying a predicate
+   *
+   * @param pred Predicate to filter on
+   * @return
+   */
+  def filter(pred: (K, V) => Boolean): AbstractShapeRDDPartition[K, V]
+}

src/main/scala/org/zouzias/spark/lucenerdd/spatial/shape/partition/ShapeLuceneRDDPartition.scala → src/main/scala/org/zouzias/spark/lucenerdd/spatial/shape/partition/impl/ShapeLuceneRDDPartition.scala

@@ -14,21 +14,20 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.zouzias.spark.lucenerdd.spatial.shape.partition
+package org.zouzias.spark.lucenerdd.spatial.shape.partition.impl
 
 import com.spatial4j.core.distance.DistanceUtils
 import com.spatial4j.core.shape.Shape
 import org.apache.lucene.document.{Document, StoredField}
-import org.apache.lucene.facet.taxonomy.directory.DirectoryTaxonomyReader
 import org.apache.lucene.index.DirectoryReader
 import org.apache.lucene.search.{IndexSearcher, ScoreDoc, Sort}
 import org.apache.lucene.spatial.query.{SpatialArgs, SpatialOperation}
 import org.joda.time.DateTime
 import org.zouzias.spark.lucenerdd.models.SparkScoreDoc
 import org.zouzias.spark.lucenerdd.query.LuceneQueryHelpers
 import org.zouzias.spark.lucenerdd.response.LuceneRDDResponsePartition
-import org.zouzias.spark.lucenerdd.spatial.shape.ShapeLuceneRDD.PointType
-import org.zouzias.spark.lucenerdd.spatial.shape.grids.PrefixTreeLoader
+import org.zouzias.spark.lucenerdd.spatial.shape.rdds.ShapeLuceneRDD.PointType
+import org.zouzias.spark.lucenerdd.spatial.shape.partition.AbstractShapeLuceneRDDPartition
 import org.zouzias.spark.lucenerdd.spatial.shape.strategies.SpatialStrategy
 import org.zouzias.spark.lucenerdd.store.IndexWithTaxonomyWriter