point_rdd: cleanup

Author: zouzias

…rdd/spatial/commons/MaxPointMonoid.scala …lucenerdd/aggregate/MaxPointMonoid.scalasrc/main/scala/org/zouzias/spark/lucenerdd/spatial/commons/MaxPointMonoid.scala renamed to src/main/scala/org/zouzias/spark/lucenerdd/aggregate/MaxPointMonoid.scala src/main/scala/org/zouzias/spark/lucenerdd/spatial/commons/MaxPointMonoid.scala renamed to src/main/scala/org/zouzias/spark/lucenerdd/aggregate/MaxPointMonoid.scala

@@ -14,12 +14,18 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.zouzias.spark.lucenerdd.spatial.commons
+package org.zouzias.spark.lucenerdd.aggregate
 
 import com.twitter.algebird.Monoid
 import org.zouzias.spark.lucenerdd.spatial.point.PointLuceneRDD.PointType
 
+/**
+  * Maximum point [[Monoid]] used for spatial linkage
+  *
+  * Keeps the maximum value per coordinate
+  */
 object MaxPointMonoid extends Monoid[PointType] {
+
   override def zero: PointType = (Double.MaxValue, Double.MaxValue)
 
   override def plus(x: PointType, y: PointType): PointType = {

…rdd/spatial/commons/MinPointMonoid.scala …lucenerdd/aggregate/MinPointMonoid.scalasrc/main/scala/org/zouzias/spark/lucenerdd/spatial/commons/MinPointMonoid.scala renamed to src/main/scala/org/zouzias/spark/lucenerdd/aggregate/MinPointMonoid.scala src/main/scala/org/zouzias/spark/lucenerdd/spatial/commons/MinPointMonoid.scala renamed to src/main/scala/org/zouzias/spark/lucenerdd/aggregate/MinPointMonoid.scala

@@ -14,12 +14,18 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.zouzias.spark.lucenerdd.spatial.commons
+package org.zouzias.spark.lucenerdd.aggregate
 
 import com.twitter.algebird.Monoid
 import org.zouzias.spark.lucenerdd.spatial.point.PointLuceneRDD.PointType
 
+/**
+  * Minimum point Monoid used for spatial linkage
+  *
+  * Keeps the minimum value per coordinate
+  */
 object MinPointMonoid extends Monoid[PointType] {
+
   override def zero: PointType = (Double.MinValue, Double.MinValue)
 
   override def plus(x: PointType, y: PointType): PointType = {

src/main/scala/org/zouzias/spark/lucenerdd/spatial/commons/SpatialByXPartitioner.scala

@@ -0,0 +1,40 @@
+/*
+ * 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.commons
+
+import org.apache.spark.Partitioner
+import org.zouzias.spark.lucenerdd.spatial.point.PointLuceneRDD.PointType
+
+import scala.util.Random
+
+/**
+  * Spark RDD [[Partitioner]] based on the x-axis and bounds per partition
+  * @param boundsPerPart Bounds of x-axis (minX, maxX) per RDD's partition
+  */
+case class SpatialByXPartitioner(boundsPerPart: Array[(Double, Double)]) extends Partitioner {
+  override def numPartitions: Int = boundsPerPart.length
+
+  override def getPartition(key: Any): Int = {
+    val keyPoint = key.asInstanceOf[PointType]
+    val indexOpt = boundsPerPart.indexWhere{ case (minX, maxX) =>
+      minX <= keyPoint._1 && keyPoint._1 <= maxX
+    }
+
+    // If key is not assigned to a partition, randomly assign the key
+    if (indexOpt == -1) Random.nextInt(numPartitions) else indexOpt
+  }
+}

src/main/scala/org/zouzias/spark/lucenerdd/spatial/point/PointLuceneRDD.scala

@@ -19,16 +19,17 @@ package org.zouzias.spark.lucenerdd.spatial.point
 import com.twitter.algebird._
 import org.apache.lucene.document.Document
 import org.apache.lucene.spatial.query.SpatialOperation
-import org.apache.spark.{OneToOneDependency, Partition, TaskContext}
+import org.apache.spark.{OneToOneDependency, Partition, Partitioner, TaskContext}
 import org.apache.spark.rdd.RDD
 import org.apache.spark.sql.{DataFrame, Dataset, Row}
 import org.apache.spark.storage.StorageLevel
+import org.zouzias.spark.lucenerdd.aggregate.{MaxPointMonoid, MinPointMonoid}
 import org.zouzias.spark.lucenerdd.analyzers.AnalyzerConfigurable
 import org.zouzias.spark.lucenerdd.config.ShapeLuceneRDDConfigurable
 import org.zouzias.spark.lucenerdd.models.SparkScoreDoc
 import org.zouzias.spark.lucenerdd.query.{LuceneQueryHelpers, SimilarityConfigurable}
 import org.zouzias.spark.lucenerdd.response.{LuceneRDDResponse, LuceneRDDResponsePartition}
-import org.zouzias.spark.lucenerdd.spatial.commons.{MaxPointMonoid, MinPointMonoid}
+import org.zouzias.spark.lucenerdd.spatial.commons.SpatialByXPartitioner
 import org.zouzias.spark.lucenerdd.spatial.point.PointLuceneRDD.PointType
 import org.zouzias.spark.lucenerdd.spatial.point.partition.{AbstractPointLuceneRDDPartition, PointLuceneRDDPartition}
 import org.zouzias.spark.lucenerdd.versioning.Versionable
@@ -198,7 +199,7 @@ class PointLuceneRDD[V: ClassTag]
    *
    * @param shapeWKT Shape in WKT format
    * @param k Number of element to return
-   * @param operationName
+   * @param operationName Spatial operation name, i.e., intersection, contained, etc.
    * @return
    */
   def spatialSearch(shapeWKT: String, k: Int,
@@ -213,7 +214,7 @@ class PointLuceneRDD[V: ClassTag]
    * @param center given as (x, y)
    * @param radius in kilometers (KM)
    * @param k
-   * @param operationName
+   * @param operationName Spatial operation name, i.e., intersection, contained, etc.
    * @return
    */
   def bboxSearch(center: PointType, radius: Double, k: Int,
@@ -228,7 +229,7 @@ class PointLuceneRDD[V: ClassTag]
    * @param lowerLeft Lower left corner
    * @param upperRight Upper right corner
    * @param k Number of results
-   * @param operationName Intersect, contained, etc.
+   * @param operationName Spatial operation name, i.e., intersection, contained, etc.
    * @return
    */
   def bboxSearch(lowerLeft: PointType,
@@ -257,6 +258,15 @@ class PointLuceneRDD[V: ClassTag]
       }
   }
 
+  /**
+    * Returns an x-axis spatial [[Partitioner]] that is used in link related methods
+    * @return Spark [[Partitioner]]
+    */
+  def spatialPartitioner(): Partitioner = {
+    val bpp = boundsPerPartition().map(x => (x._1._1, x._2._1)).collect()
+    SpatialByXPartitioner(bpp)
+  }
+
   override def count(): Long = {
     logInfo("Count requested")
     partitionsRDD.map(_.size).reduce(_ + _)