WIP: Point rdd

build.sbt

@@ -17,8 +17,8 @@
 
 name := "spark-lucenerdd"
 organization := "org.zouzias"
-scalaVersion := "2.11.11"
-crossScalaVersions := Seq("2.11.11")
+scalaVersion := "2.11.12"
+crossScalaVersions := Seq("2.11.12")
 licenses := Seq("Apache-2.0" -> url("http://www.apache.org/licenses/LICENSE-2.0.html"))
 homepage := Some(url("https://github.com/zouzias/spark-lucenerdd"))
 
@@ -79,10 +79,10 @@ pomExtra := (
   </developers>
 )
 
-val luceneV = "7.1.0"
+val luceneV = "7.3.0"
 
 spName := "zouzias/spark-lucenerdd"
-sparkVersion := "2.2.0"
+sparkVersion := "2.2.1"
 spShortDescription := "Spark RDD with Lucene's query capabilities"
 sparkComponents ++= Seq("core", "sql", "mllib")
 spAppendScalaVersion := true
@@ -96,15 +96,15 @@ testSparkVersion := sys.props.get("spark.testVersion").getOrElse(sparkVersion.va
 
 
 // scalastyle:off
-val scalactic                 = "org.scalactic"                  %% "scalactic" % "3.0.4"
-val scalatest                 = "org.scalatest"                  %% "scalatest"                % "3.0.4" % "test"
+val scalactic                 = "org.scalactic"                  %% "scalactic"                % "3.0.5"
+val scalatest                 = "org.scalatest"                  %% "scalatest"                % "3.0.5" % "test"
 
 val joda_time                 = "joda-time"                      % "joda-time"                 % "2.9.9"
-val algebird                  = "com.twitter"                    %% "algebird-core"            % "0.13.3"
+val algebird                  = "com.twitter"                    %% "algebird-core"            % "0.13.4"
 val joda_convert              = "org.joda"                       % "joda-convert"              % "1.9.2"
-val spatial4j                 = "org.locationtech.spatial4j"     % "spatial4j"                 % "0.6"
+val spatial4j                 = "org.locationtech.spatial4j"     % "spatial4j"                 % "0.7"
 
-val typesafe_config           = "com.typesafe"                   % "config"                    % "1.3.1"
+val typesafe_config           = "com.typesafe"                   % "config"                    % "1.3.3"
 
 val lucene_facet              = "org.apache.lucene"              % "lucene-facet"              % luceneV
 val lucene_analyzers          = "org.apache.lucene"              % "lucene-analyzers-common"   % luceneV
@@ -113,7 +113,7 @@ val lucene_expressions        = "org.apache.lucene"              % "lucene-expre
 val lucene_spatial            = "org.apache.lucene"              % "lucene-spatial"            % luceneV
 val lucene_spatial_extras     = "org.apache.lucene"              % "lucene-spatial-extras"     % luceneV
 
-val jts                       = "com.vividsolutions"             % "jts"                       % "1.13"
+val jts                       = "org.locationtech.jts"           % "jts-core"                  % "1.15.0"
 // scalastyle:on
 
 
@@ -137,7 +137,7 @@ libraryDependencies ++= Seq(
 libraryDependencies ++= Seq(
   "org.apache.spark" %% "spark-core" % testSparkVersion.value % "test" force(),
   "org.apache.spark" %% "spark-sql" % testSparkVersion.value % "test" force(),
-  "com.holdenkarau"  %% "spark-testing-base" % s"${sparkVersion.value}_0.7.4" % "test" intransitive(),
+  "com.holdenkarau"  %% "spark-testing-base" % s"2.2.1_0.9.0" % "test" intransitive(),
   "org.scala-lang"    % "scala-library" % scalaVersion.value % "compile"
 )
 

project/plugins.sbt

@@ -17,9 +17,9 @@
 
 resolvers += "bintray-spark-packages" at "https://dl.bintray.com/spark-packages/maven/"
 
-addSbtPlugin("com.eed3si9n" % "sbt-assembly" % "0.14.4")
+addSbtPlugin("com.eed3si9n" % "sbt-assembly" % "0.14.6")
 
-addSbtPlugin("com.github.gseitz" % "sbt-release" % "1.0.5")
+addSbtPlugin("com.github.gseitz" % "sbt-release" % "1.0.6")
 
 addSbtPlugin("org.scalastyle" %% "scalastyle-sbt-plugin" % "0.8.0")
 
@@ -31,4 +31,4 @@ addSbtPlugin("org.xerial.sbt" % "sbt-sonatype" % "1.1")
 
 addSbtPlugin("com.jsuereth" % "sbt-pgp" % "1.0.0")
 
-addSbtPlugin("org.spark-packages" % "sbt-spark-package" % "0.2.5")
+addSbtPlugin("org.spark-packages" % "sbt-spark-package" % "0.2.6")

src/main/scala/org/zouzias/spark/lucenerdd/aggregate/BoundingBoxMonoid.scala

@@ -0,0 +1,38 @@
+/*
+ * 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.aggregate
+
+import com.twitter.algebird.Monoid
+import org.zouzias.spark.lucenerdd.models.BoundingBox
+import org.zouzias.spark.lucenerdd.spatial.point.PointLuceneRDD.PointType
+
+
+/**
+  * Bounding box monoid based on [[BoundingBox]]
+  */
+object BoundingBoxMonoid extends Monoid[BoundingBox] {
+
+  override def zero: BoundingBox = {
+    BoundingBox((Double.MaxValue, Double.MaxValue), (Double.MinValue, Double.MinValue))
+  }
+
+  override def plus(x: BoundingBox,
+                    y: BoundingBox): BoundingBox = {
+    BoundingBox(MinPointMonoid.plus(x.lowerLeft, y.lowerLeft),
+      MaxPointMonoid.plus(x.upperRight, y.upperRight))
+  }
+}

src/main/scala/org/zouzias/spark/lucenerdd/aggregate/MaxPointMonoid.scala

@@ -0,0 +1,34 @@
+/*
+ * 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.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.MinValue, Double.MinValue)
+
+  override def plus(a: PointType, b: PointType): PointType = {
+    (Math.max(a._1, b._1), Math.max(a._2, b._2))
+  }
+}

src/main/scala/org/zouzias/spark/lucenerdd/aggregate/MinPointMonoid.scala

@@ -0,0 +1,34 @@
+/*
+ * 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.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.MaxValue, Double.MaxValue)
+
+  override def plus(x: PointType, y: PointType): PointType = {
+    (Math.min(x._1, y._1), Math.min(x._2, y._2))
+  }
+}

src/main/scala/org/zouzias/spark/lucenerdd/models/BoundingBox.scala

@@ -0,0 +1,27 @@
+/*
+ * 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.models
+
+import org.zouzias.spark.lucenerdd.spatial.point.PointLuceneRDD.PointType
+
+/**
+  * Bounding box described as lower-left and upper right point
+  *
+  * @param lowerLeft The lowest-left point, i.e., (min_X, min_Y)
+  * @param upperRight The upper-right point, i.e., (max_X, max_Y)
+  */
+case class BoundingBox(lowerLeft: PointType, upperRight: PointType)

src/main/scala/org/zouzias/spark/lucenerdd/response/LuceneRDDResponse.scala

@@ -76,9 +76,14 @@ private[lucenerdd] class LuceneRDDResponse
 
   override def collect(): Array[SparkScoreDoc] = {
     val sz = partitionsRDD.map(_.size).sum().toInt
-    val monoid = new TopKMonoid[SparkScoreDoc](sz)(ordering)
-    partitionsRDD.map(monoid.build(_))
-      .reduce(monoid.plus).items.toArray
+    sz match {
+      case 0 => Array.empty[SparkScoreDoc]
+      case s =>
+        // Total size must be positive for [[TopKMonoid]]
+        val monoid = new TopKMonoid[SparkScoreDoc](s)(ordering)
+        partitionsRDD.map(monoid.build(_))
+          .reduce(monoid.plus).items.toArray
+    }
   }
 
   /**

src/main/scala/org/zouzias/spark/lucenerdd/response/LuceneRDDResponsePartition.scala

@@ -23,3 +23,9 @@ case class LuceneRDDResponsePartition(results: Iterator[SparkScoreDoc])
   extends Iterable[SparkScoreDoc] {
   override def iterator(): Iterator[SparkScoreDoc] = results
 }
+
+object LuceneRDDResponsePartition {
+  def empty(): LuceneRDDResponsePartition = {
+    LuceneRDDResponsePartition(Iterator.empty)
+  }
+}

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[PointType]) 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/shape/context/ContextLoader.scala → src/main/scala/org/zouzias/spark/lucenerdd/spatial/commons/context/ContextLoader.scala

@@ -14,11 +14,13 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.zouzias.spark.lucenerdd.spatial.shape.context
+package org.zouzias.spark.lucenerdd.spatial.commons.context
 
 import java.io.{StringReader, StringWriter}
 
+import com.vividsolutions.jts.io.ParseException
 import org.locationtech.spatial4j.context.jts.JtsSpatialContext
+import org.locationtech.spatial4j.exception.InvalidShapeException
 import org.locationtech.spatial4j.io.{ShapeReader, ShapeWriter}
 import org.locationtech.spatial4j.shape.Shape
 import org.zouzias.spark.lucenerdd.config.ShapeLuceneRDDConfigurable
@@ -37,8 +39,13 @@ trait ContextLoader extends ShapeLuceneRDDConfigurable{
     writer.toString
   }
 
-  protected def stringToShape(shapeAsString: String): Shape = {
-   shapeReader.read(new StringReader(shapeAsString))
+  protected def stringToShape(shapeAsString: String): Option[Shape] = {
+   try {
+     Some(shapeReader.read(new StringReader(shapeAsString)))
+   } catch {
+     case _: ParseException => None
+     case _ : InvalidShapeException => None
+   }
   }
 
   /**

src/main/scala/org/zouzias/spark/lucenerdd/spatial/shape/grids/PrefixTreeLoader.scala → src/main/scala/org/zouzias/spark/lucenerdd/spatial/commons/grids/PrefixTreeLoader.scala

@@ -14,11 +14,11 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.zouzias.spark.lucenerdd.spatial.shape.grids
+package org.zouzias.spark.lucenerdd.spatial.commons.grids
 
 import org.apache.lucene.spatial.prefix.tree.{SpatialPrefixTree, SpatialPrefixTreeFactory}
 import org.zouzias.spark.lucenerdd.config.ShapeLuceneRDDConfigurable
-import org.zouzias.spark.lucenerdd.spatial.shape.context.ContextLoader
+import org.zouzias.spark.lucenerdd.spatial.commons.context.ContextLoader
 
 import scala.collection.JavaConverters._
 

src/main/scala/org/zouzias/spark/lucenerdd/spatial/shape/strategies/SpatialStrategy.scala → src/main/scala/org/zouzias/spark/lucenerdd/spatial/commons/strategies/SpatialStrategy.scala

@@ -14,10 +14,10 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.zouzias.spark.lucenerdd.spatial.shape.strategies
+package org.zouzias.spark.lucenerdd.spatial.commons.strategies
 
 import org.apache.lucene.spatial.prefix.{PrefixTreeStrategy, RecursivePrefixTreeStrategy}
-import org.zouzias.spark.lucenerdd.spatial.shape.grids.PrefixTreeLoader
+import org.zouzias.spark.lucenerdd.spatial.commons.grids.PrefixTreeLoader
 
 trait SpatialStrategy extends PrefixTreeLoader {