Collections API

accumulo/src/main/scala/geotrellis/spark/io/accumulo/AccumuloAttributeStore.scala

@@ -12,7 +12,6 @@ import org.apache.accumulo.core.security.Authorizations
 import org.apache.accumulo.core.data._
 import org.apache.hadoop.io.Text
 
-
 import scala.collection.JavaConversions._
 
 object AccumuloAttributeStore {

accumulo/src/main/scala/geotrellis/spark/io/accumulo/AccumuloCollectionReader.scala

@@ -0,0 +1,43 @@
+package geotrellis.spark.io.accumulo
+
+import geotrellis.spark.io.avro.codecs.KeyValueRecordCodec
+import geotrellis.spark.io.avro.{AvroEncoder, AvroRecordCodec}
+import geotrellis.spark.{Boundable, KeyBounds}
+
+import org.apache.accumulo.core.data.{Range => AccumuloRange}
+import org.apache.accumulo.core.security.Authorizations
+import org.apache.avro.Schema
+import org.apache.hadoop.io.Text
+
+import scala.collection.JavaConversions._
+import scala.reflect.ClassTag
+
+object AccumuloCollectionReader {
+  def read[K: Boundable: AvroRecordCodec: ClassTag, V: AvroRecordCodec: ClassTag](
+    table: String,
+    columnFamily: Text,
+    queryKeyBounds: Seq[KeyBounds[K]],
+    decomposeBounds: KeyBounds[K] => Seq[AccumuloRange],
+    filterIndexOnly: Boolean,
+    writerSchema: Option[Schema] = None
+  )(implicit instance: AccumuloInstance): Seq[(K, V)] = {
+    if(queryKeyBounds.isEmpty) return Seq.empty[(K, V)]
+
+    val codec = KeyValueRecordCodec[K, V]
+    val includeKey = (key: K) => queryKeyBounds.includeKey(key)
+
+    val ranges = queryKeyBounds.flatMap(decomposeBounds)
+
+    ranges flatMap { range: AccumuloRange =>
+      val scanner = instance.connector.createScanner(table, new Authorizations())
+      scanner.setRange(range)
+      scanner.fetchColumnFamily(columnFamily)
+      scanner.iterator.map { case entry =>
+        AvroEncoder.fromBinary(writerSchema.getOrElse(codec.schema), entry.getValue.get)(codec)
+      }.flatMap { pairs: Vector[(K, V)] =>
+        if(filterIndexOnly) pairs
+        else pairs.filter { pair => includeKey(pair._1) }
+      }
+    }
+  }
+}

accumulo/src/main/scala/geotrellis/spark/io/accumulo/AccumuloLayerCollectionReader.scala

@@ -0,0 +1,47 @@
+package geotrellis.spark.io.accumulo
+
+import geotrellis.spark._
+import geotrellis.spark.io._
+import geotrellis.spark.io.avro._
+import geotrellis.util._
+
+import org.apache.accumulo.core.data.{Range => AccumuloRange}
+import org.apache.hadoop.io.Text
+import spray.json._
+
+import scala.reflect._
+
+class AccumuloLayerCollectionReader(val attributeStore: AttributeStore)(implicit instance: AccumuloInstance) extends CollectionLayerReader[LayerId] {
+
+  def read[
+    K: AvroRecordCodec: Boundable: JsonFormat: ClassTag,
+    V: AvroRecordCodec: ClassTag,
+    M: JsonFormat: GetComponent[?, Bounds[K]]
+  ](id: LayerId, rasterQuery: LayerQuery[K, M], numPartitions: Int, filterIndexOnly: Boolean) = {
+    if (!attributeStore.layerExists(id)) throw new LayerNotFoundError(id)
+
+    val LayerAttributes(header, metadata, keyIndex, writerSchema) = try {
+      attributeStore.readLayerAttributes[AccumuloLayerHeader, M, K](id)
+    } catch {
+      case e: AttributeNotFoundError => throw new LayerReadError(id).initCause(e)
+    }
+
+    val queryKeyBounds = rasterQuery(metadata)
+
+    val decompose = (bounds: KeyBounds[K]) =>
+      keyIndex.indexRanges(bounds).map { case (min, max) =>
+        new AccumuloRange(new Text(AccumuloKeyEncoder.long2Bytes(min)), new Text(AccumuloKeyEncoder.long2Bytes(max)))
+      }
+
+    val seq = AccumuloCollectionReader.read[K, V](header.tileTable, columnFamily(id), queryKeyBounds, decompose, filterIndexOnly, Some(writerSchema))
+    new ContextCollection(seq, metadata)
+  }
+}
+
+object AccumuloLayerCollectionReader {
+  def apply(attributeStore: AccumuloAttributeStore)(implicit instance: AccumuloInstance): AccumuloLayerCollectionReader =
+    new AccumuloLayerCollectionReader(attributeStore)
+
+  def apply(implicit instance: AccumuloInstance): AccumuloLayerCollectionReader =
+    new AccumuloLayerCollectionReader(AccumuloAttributeStore(instance.connector))
+}

accumulo/src/test/scala/geotrellis/spark/io/accumulo/AccumuloSpaceTimeSpec.scala

@@ -20,12 +20,13 @@ class AccumuloSpaceTimeSpec
   implicit lazy val instance = MockAccumuloInstance()
 
   lazy val reader    = AccumuloLayerReader(instance)
-  lazy val writer = AccumuloLayerWriter(instance, "tiles", SocketWriteStrategy())
+  lazy val creader   = AccumuloLayerCollectionReader(instance)
+  lazy val writer    = AccumuloLayerWriter(instance, "tiles", SocketWriteStrategy())
   lazy val deleter   = AccumuloLayerDeleter(instance)
   lazy val reindexer = AccumuloLayerReindexer(instance, SocketWriteStrategy())
   lazy val updater   = AccumuloLayerUpdater(instance, SocketWriteStrategy())
   lazy val tiles     = AccumuloValueReader(instance)
   lazy val sample    = CoordinateSpaceTime
-  lazy val copier = AccumuloLayerCopier(instance, reader, writer)
-  lazy val mover  = AccumuloLayerMover(copier, deleter)
+  lazy val copier    = AccumuloLayerCopier(instance, reader, writer)
+  lazy val mover     = AccumuloLayerMover(copier, deleter)
 }

accumulo/src/test/scala/geotrellis/spark/io/accumulo/AccumuloSpatialSpec.scala

@@ -16,14 +16,14 @@ class AccumuloSpatialSpec
 
   implicit lazy val instance = MockAccumuloInstance()
 
-  lazy val reader = AccumuloLayerReader(instance)
-  lazy val writer = AccumuloLayerWriter(instance, "tiles", SocketWriteStrategy())
-  lazy val deleter = AccumuloLayerDeleter(instance)
+  lazy val reader    = AccumuloLayerReader(instance)
+  lazy val creader   = AccumuloLayerCollectionReader(instance)
+  lazy val writer    = AccumuloLayerWriter(instance, "tiles", SocketWriteStrategy())
+  lazy val deleter   = AccumuloLayerDeleter(instance)
   lazy val reindexer = AccumuloLayerReindexer(instance, SocketWriteStrategy())
   lazy val updater   = AccumuloLayerUpdater(instance, SocketWriteStrategy())
-  lazy val tiles = AccumuloValueReader(instance)
-  lazy val sample = AllOnesTestFile
-
-  lazy val copier = AccumuloLayerCopier(instance, reader, writer)
-  lazy val mover  = AccumuloLayerMover(copier, deleter)
+  lazy val tiles     = AccumuloValueReader(instance)
+  lazy val sample    = AllOnesTestFile
+  lazy val copier    = AccumuloLayerCopier(instance, reader, writer)
+  lazy val mover     = AccumuloLayerMover(copier, deleter)
 }

accumulo/src/test/scala/geotrellis/spark/io/accumulo/AccumuloTileFeatureSpaceTimeSpec.scala

@@ -21,12 +21,13 @@ class AccumuloTileFeatureSpaceTimeSpec
   implicit lazy val instance = MockAccumuloInstance()
 
   lazy val reader    = AccumuloLayerReader(instance)
-  lazy val writer = AccumuloLayerWriter(instance, "tiles", SocketWriteStrategy())
+  lazy val creader   = AccumuloLayerCollectionReader(instance)
+  lazy val writer    = AccumuloLayerWriter(instance, "tiles", SocketWriteStrategy())
   lazy val deleter   = AccumuloLayerDeleter(instance)
   lazy val reindexer = AccumuloLayerReindexer(instance, SocketWriteStrategy())
   lazy val updater   = AccumuloLayerUpdater(instance, SocketWriteStrategy())
   lazy val tiles     = AccumuloValueReader(instance)
   lazy val sample    = CoordinateSpaceTime
-  lazy val copier = AccumuloLayerCopier(instance, reader, writer)
-  lazy val mover  = AccumuloLayerMover(copier, deleter)
+  lazy val copier    = AccumuloLayerCopier(instance, reader, writer)
+  lazy val mover     = AccumuloLayerMover(copier, deleter)
 }

accumulo/src/test/scala/geotrellis/spark/io/accumulo/AccumuloTileFeatureSpatialSpec.scala

@@ -17,14 +17,14 @@ class AccumuloTileFeatureSpatialSpec
 
   implicit lazy val instance = MockAccumuloInstance()
 
-  lazy val reader = AccumuloLayerReader(instance)
-  lazy val writer = AccumuloLayerWriter(instance, "tiles", SocketWriteStrategy())
-  lazy val deleter = AccumuloLayerDeleter(instance)
+  lazy val reader    = AccumuloLayerReader(instance)
+  lazy val creader   = AccumuloLayerCollectionReader(instance)
+  lazy val writer    = AccumuloLayerWriter(instance, "tiles", SocketWriteStrategy())
+  lazy val deleter   = AccumuloLayerDeleter(instance)
   lazy val reindexer = AccumuloLayerReindexer(instance, SocketWriteStrategy())
   lazy val updater   = AccumuloLayerUpdater(instance, SocketWriteStrategy())
-  lazy val tiles = AccumuloValueReader(instance)
-  lazy val sample = AllOnesTestFile
-
-  lazy val copier = AccumuloLayerCopier(instance, reader, writer)
-  lazy val mover  = AccumuloLayerMover(copier, deleter)
+  lazy val tiles     = AccumuloValueReader(instance)
+  lazy val sample    = AllOnesTestFile
+  lazy val copier    = AccumuloLayerCopier(instance, reader, writer)
+  lazy val mover     = AccumuloLayerMover(copier, deleter)
 }

cassandra/src/main/resources/reference.conf

@@ -6,4 +6,7 @@ geotrellis.cassandra {
   localDc              = "datacenter1"
   usedHostsPerRemoteDc = 0
   allowRemoteDCsForLocalConsistencyLevel = false
+  threads = {
+    collection.read = 32
+  }
 }

cassandra/src/main/scala/geotrellis/spark/io/cassandra/CassandraCollectionReader.scala

@@ -0,0 +1,92 @@
+package geotrellis.spark.io.cassandra
+
+import geotrellis.spark.{Boundable, KeyBounds, LayerId}
+import geotrellis.spark.io.CollectionLayerReader
+import geotrellis.spark.io.avro.codecs.KeyValueRecordCodec
+import geotrellis.spark.io.avro.{AvroEncoder, AvroRecordCodec}
+import geotrellis.spark.io.index.{IndexRanges, MergeQueue}
+import geotrellis.spark.util.KryoWrapper
+
+import org.apache.avro.Schema
+import com.datastax.driver.core.querybuilder.QueryBuilder
+import com.datastax.driver.core.querybuilder.QueryBuilder.{eq => eqs}
+import com.typesafe.config.ConfigFactory
+import scalaz.std.vector._
+import scalaz.concurrent.{Strategy, Task}
+import scalaz.stream.{Process, nondeterminism}
+
+import java.util.concurrent.Executors
+import scala.collection.JavaConversions._
+import scala.reflect.ClassTag
+
+object CassandraCollectionReader {
+  def read[K: Boundable : AvroRecordCodec : ClassTag, V: AvroRecordCodec : ClassTag](
+    instance: CassandraInstance,
+    keyspace: String,
+    table: String,
+    layerId: LayerId,
+    queryKeyBounds: Seq[KeyBounds[K]],
+    decomposeBounds: KeyBounds[K] => Seq[(Long, Long)],
+    filterIndexOnly: Boolean,
+    writerSchema: Option[Schema] = None,
+    numPartitions: Option[Int] = None,
+    threads: Int = ConfigFactory.load().getInt("geotrellis.cassandra.threads.collection.read")
+  ): Seq[(K, V)] = {
+    if (queryKeyBounds.isEmpty) return Seq.empty[(K, V)]
+
+    val includeKey = (key: K) => queryKeyBounds.includeKey(key)
+    val _recordCodec = KeyValueRecordCodec[K, V]
+    val kwWriterSchema = KryoWrapper(writerSchema) //Avro Schema is not Serializable
+
+    val ranges = if (queryKeyBounds.length > 1)
+      MergeQueue(queryKeyBounds.flatMap(decomposeBounds))
+    else
+      queryKeyBounds.flatMap(decomposeBounds)
+
+    val bins = IndexRanges.bin(ranges, numPartitions.getOrElse(CollectionLayerReader.defaultNumPartitions)).toVector.map(_.toIterator)
+
+    val query = QueryBuilder.select("value")
+      .from(keyspace, table)
+      .where(eqs("key", QueryBuilder.bindMarker()))
+      .and(eqs("name", layerId.name))
+      .and(eqs("zoom", layerId.zoom))
+      .toString
+
+    val pool = Executors.newFixedThreadPool(threads)
+
+    val result = instance.withSessionDo { session =>
+      val statement = session.prepare(query)
+
+      bins flatMap { partition =>
+        val range: Process[Task, Iterator[Long]] = Process.unfold(partition) { iter =>
+          if (iter.hasNext) {
+            val (start, end) = iter.next()
+            Some((start to end).toIterator, iter)
+          }
+          else None
+        }
+
+        val read: Iterator[Long] => Process[Task, Vector[(K, V)]] = { iterator =>
+          Process.unfold(iterator) { iter =>
+            if (iter.hasNext) {
+              val index = iter.next()
+              val row = session.execute(statement.bind(index.asInstanceOf[java.lang.Long]))
+              if (row.nonEmpty) {
+                val bytes = row.one().getBytes("value").array()
+                val recs = AvroEncoder.fromBinary(kwWriterSchema.value.getOrElse(_recordCodec.schema), bytes)(_recordCodec)
+                if (filterIndexOnly) Some(recs, iter)
+                else Some(recs.filter { row => includeKey(row._1) }, iter)
+              } else Some(Vector.empty, iter)
+            } else {
+              None
+            }
+          }
+        }
+
+        nondeterminism.njoin(maxOpen = threads, maxQueued = threads) { range map read }(Strategy.Executor(pool)).runFoldMap(identity).unsafePerformSync
+      }
+    }
+
+    pool.shutdown(); result
+  }
+}

cassandra/src/main/scala/geotrellis/spark/io/cassandra/CassandraLayerCollectionReader.scala

@@ -0,0 +1,42 @@
+package geotrellis.spark.io.cassandra
+
+import geotrellis.spark._
+import geotrellis.spark.io._
+import geotrellis.spark.io.avro._
+import geotrellis.util._
+
+import spray.json._
+
+import scala.reflect._
+
+class CassandraLayerCollectionReader(val attributeStore: AttributeStore, instance: CassandraInstance) extends CollectionLayerReader[LayerId] {
+
+  def read[
+    K: AvroRecordCodec: Boundable: JsonFormat: ClassTag,
+    V: AvroRecordCodec: ClassTag,
+    M: JsonFormat: GetComponent[?, Bounds[K]]
+  ](id: LayerId, rasterQuery: LayerQuery[K, M], numPartitions: Int, filterIndexOnly: Boolean) = {
+    if (!attributeStore.layerExists(id)) throw new LayerNotFoundError(id)
+
+    val LayerAttributes(header, metadata, keyIndex, writerSchema) = try {
+      attributeStore.readLayerAttributes[CassandraLayerHeader, M, K](id)
+    } catch {
+      case e: AttributeNotFoundError => throw new LayerReadError(id).initCause(e)
+    }
+
+    val queryKeyBounds = rasterQuery(metadata)
+
+    val decompose = (bounds: KeyBounds[K]) => keyIndex.indexRanges(bounds)
+
+    val seq = CassandraCollectionReader.read[K, V](instance, header.keyspace, header.tileTable, id, queryKeyBounds, decompose, filterIndexOnly, Some(writerSchema))
+    new ContextCollection(seq, metadata)
+  }
+}
+
+object CassandraLayerCollectionReader {
+  def apply(instance: CassandraInstance): CassandraLayerCollectionReader =
+    new CassandraLayerCollectionReader(CassandraAttributeStore(instance), instance)
+
+  def apply(attributeStore: CassandraAttributeStore): CassandraLayerCollectionReader =
+    new CassandraLayerCollectionReader(attributeStore, attributeStore.instance)
+}

cassandra/src/test/scala/geotrellis/spark/io/cassandra/CassandraSpaceTimeSpec.scala

@@ -18,6 +18,7 @@ class CassandraSpaceTimeSpec
   lazy val attributeStore = CassandraAttributeStore(instance)
 
   lazy val reader    = CassandraLayerReader(attributeStore)
+  lazy val creader   = CassandraLayerCollectionReader(attributeStore)
   lazy val writer    = CassandraLayerWriter(attributeStore, "geotrellis", "tiles")
   lazy val deleter   = CassandraLayerDeleter(attributeStore)
   lazy val updater   = CassandraLayerUpdater(attributeStore)

cassandra/src/test/scala/geotrellis/spark/io/cassandra/CassandraSpatialSpec.scala

@@ -17,6 +17,7 @@ class CassandraSpatialSpec
   lazy val attributeStore = CassandraAttributeStore(instance)
 
   lazy val reader    = CassandraLayerReader(attributeStore)
+  lazy val creader   = CassandraLayerCollectionReader(attributeStore)
   lazy val writer    = CassandraLayerWriter(attributeStore, "geotrellis", "tiles")
   lazy val deleter   = CassandraLayerDeleter(attributeStore)
   lazy val updater   = CassandraLayerUpdater(attributeStore)

cassandra/src/test/scala/geotrellis/spark/io/cassandra/CassandraTileFeatureSpaceTimeSpec.scala

@@ -18,6 +18,7 @@ class CassandraTileFeatureSpaceTimeSpec
   lazy val attributeStore = CassandraAttributeStore(instance, "geotrellis_tf", "metadata")
 
   lazy val reader    = CassandraLayerReader(attributeStore)
+  lazy val creader   = CassandraLayerCollectionReader(attributeStore)
   lazy val writer    = CassandraLayerWriter(attributeStore, "geotrellis_tf", "tiles")
   lazy val deleter   = CassandraLayerDeleter(attributeStore)
   lazy val updater   = CassandraLayerUpdater(attributeStore)

cassandra/src/test/scala/geotrellis/spark/io/cassandra/CassandraTileFeatureSpatialSpec.scala

@@ -17,6 +17,7 @@ class CassandraTileFeatureSpatialSpec
   lazy val attributeStore = CassandraAttributeStore(instance, "geotrellis_tf", "metadata")
 
   lazy val reader    = CassandraLayerReader(attributeStore)
+  lazy val creader   = CassandraLayerCollectionReader(attributeStore)
   lazy val writer    = CassandraLayerWriter(attributeStore, "geotrellis_tf", "tiles")
   lazy val deleter   = CassandraLayerDeleter(attributeStore)
   lazy val updater   = CassandraLayerUpdater(attributeStore)