GEOMESA-498 adding filters to modify feature writer

Author: elahrvivaz

geomesa-core/src/main/scala/org/locationtech/geomesa/core/data/AccumuloDataStore.scala

@@ -731,14 +731,14 @@ class AccumuloDataStore(val connector: Connector,
   }
 
   /* create a general purpose writer that is capable of insert, deletes, and updates */
-  override def createFeatureWriter(typeName: String, transaction: Transaction): SFFeatureWriter = {
+  override def getFeatureWriter(typeName: String, filter: Filter, transaction: Transaction): SFFeatureWriter = {
     validateMetadata(typeName)
     checkWritePermissions(typeName)
     val sft = getSchema(typeName)
     val indexSchemaFmt = getIndexSchemaFmt(typeName)
     val fe = SimpleFeatureEncoder(sft, getFeatureEncoding(sft))
     val encoder = IndexSchema.buildKeyEncoder(indexSchemaFmt, fe)
-    new ModifyAccumuloFeatureWriter(sft, encoder, connector, fe, writeVisibilities, this)
+    new ModifyAccumuloFeatureWriter(sft, encoder, connector, fe, writeVisibilities, filter, this)
   }
 
   /* optimized for GeoTools API to return writer ONLY for appending (aka don't scan table) */

geomesa-core/src/main/scala/org/locationtech/geomesa/core/data/AccumuloFeatureWriter.scala

@@ -23,15 +23,16 @@ import org.apache.accumulo.core.client.{BatchWriterConfig, Connector}
 import org.apache.accumulo.core.data.{Key, Mutation, Value}
 import org.apache.hadoop.mapred.{RecordWriter, Reporter}
 import org.apache.hadoop.mapreduce.TaskInputOutputContext
-import org.geotools.data.DataUtilities
 import org.geotools.data.simple.SimpleFeatureWriter
+import org.geotools.data.{DataUtilities, Query}
 import org.geotools.factory.Hints
 import org.geotools.filter.identity.FeatureIdImpl
 import org.locationtech.geomesa.core.data.tables.{AttributeTable, RecordTable, SpatioTemporalTable}
 import org.locationtech.geomesa.core.index._
 import org.locationtech.geomesa.feature.{AvroSimpleFeature, AvroSimpleFeatureFactory}
 import org.locationtech.geomesa.utils.geotools.SimpleFeatureTypes
 import org.opengis.feature.simple.{SimpleFeature, SimpleFeatureType}
+import org.opengis.filter.Filter
 
 object AccumuloFeatureWriter {
 
@@ -159,10 +160,12 @@ class ModifyAccumuloFeatureWriter(featureType: SimpleFeatureType,
                                   connector: Connector,
                                   encoder: SimpleFeatureEncoder,
                                   visibility: String,
+                                  filter: Filter,
                                   dataStore: AccumuloDataStore)
   extends AccumuloFeatureWriter(featureType, indexEncoder, encoder, dataStore, visibility) {
 
-  val reader = dataStore.getFeatureReader(featureType.getName.toString)
+  val reader = dataStore.getFeatureReader(featureType.getTypeName, new Query(featureType.getTypeName, filter))
+
   var live: SimpleFeature = null      /* feature to let user modify   */
   var original: SimpleFeature = null  /* feature returned from reader */
 

geomesa-core/src/main/scala/org/locationtech/geomesa/core/data/mapreduce/FeatureIngestMapper.scala

@@ -48,7 +48,7 @@ object FeatureIngestMapper
           context.getConfiguration)).asInstanceOf[AccumuloDataStore]
 
       featureType = ds.getSchema(featureName)
-      fw = ds.createFeatureWriter(featureName, Transaction.AUTO_COMMIT)
+      fw = ds.getFeatureWriterAppend(featureName, Transaction.AUTO_COMMIT)
     }
 
     override def map(key: LongWritable, value: Text, context: Mapper#Context) {

geomesa-core/src/test/scala/org/locationtech/geomesa/core/data/AccumuloDataStoreTest.scala

@@ -1010,6 +1010,38 @@ class AccumuloDataStoreTest extends Specification {
         retrievedSchema mustEqual originalSchema
       }
     }
+
+    "Provide a feature update implementation" >> {
+      val sftName = "featureUpdateTest"
+      val sft = SimpleFeatureTypes.createType(sftName, "name:String,dtg:Date,*geom:Point:srid=4326")
+      val ds = DataStoreFinder.getDataStore(Map(
+        "instanceId"        -> "mycloud",
+        "zookeepers"        -> "zoo1:2181,zoo2:2181,zoo3:2181",
+        "user"              -> "myuser",
+        "password"          -> "mypassword",
+        "tableName"         -> sftName,
+        "useMock"           -> "true"))
+      ds.createSchema(sft)
+      val builder = AvroSimpleFeatureFactory.featureBuilder(ds.getSchema(sftName))
+      val features = (0 until 6).map { i =>
+        builder.reset()
+        builder.set("geom", WKTUtils.read("POINT(45.0 45.0)"))
+        builder.set("dtg", "2012-01-02T05:06:07.000Z")
+        builder.set("name",i.toString)
+        val sf = builder.buildFeature(i.toString)
+        sf.getUserData()(Hints.USE_PROVIDED_FID) = java.lang.Boolean.TRUE
+        sf
+      }
+      val fs = ds.getFeatureSource(sftName).asInstanceOf[AccumuloFeatureStore]
+      fs.addFeatures(new ListFeatureCollection(sft, features))
+
+      val filter = ff.id(ff.featureId("2"))
+      val writer = ds.getFeatureWriter(sftName, filter, Transaction.AUTO_COMMIT)
+      writer must beAnInstanceOf[ModifyAccumuloFeatureWriter]
+      writer.hasNext must beTrue
+      writer.next.getID mustEqual "2"
+      writer.hasNext must beFalse
+    }
   }
 
   "AccumuloFeatureStore" should {