GEOMESA-449 adding support for lists and maps in simple feature attributes

* Using SimpleFeatureTypes.createType(), you can specify a list or map:
    * names:List[String]:index=true
    * weights:Map[String][Double]:index=false

* Attribute values will be instances of java.util.List or java.util.Map when retrieved

* Lists can be queried through CQL which will execute against the individual items in the list
    * Can use standard comparison operators
    * Given the attribute names=List('bill', 'bob', 'henry') =>
        * (names = 'bob') will match
        * (names < 'joe') will match
        * (names > 'joe') will match
        * (names > 'mark') will not match

* Maps can not be queried, but can be stored and retrieved

Author: elahrvivaz

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

@@ -43,6 +43,7 @@ import org.locationtech.geomesa.core.index._
 import org.locationtech.geomesa.core.security.AuthorizationsProvider
 import org.locationtech.geomesa.utils.geotools.SimpleFeatureTypes
 import org.locationtech.geomesa.utils.geotools.SimpleFeatureTypes.{AttributeSpec, NonGeomAttributeSpec}
+import org.opengis.feature.`type`.AttributeDescriptor
 import org.opengis.feature.simple.SimpleFeatureType
 import org.opengis.filter.Filter
 import org.opengis.referencing.crs.CoordinateReferenceSystem
@@ -337,8 +338,8 @@ class AccumuloDataStore(val connector: Connector,
     val indexedAttrs = SimpleFeatureTypes.getIndexedAttributes(featureType)
     if (!indexedAttrs.isEmpty) {
       val prefix = index.getTableSharingPrefix(featureType)
-      val prefixFn = AttributeTable.getAttributeIndexRowPrefix(prefix, _: String)
-      val names = indexedAttrs.map(_.getLocalName).map(prefixFn).map(new Text(_))
+      val prefixFn = AttributeTable.getAttributeIndexRowPrefix(prefix, _: AttributeDescriptor)
+      val names = indexedAttrs.map(prefixFn).map(new Text(_))
       val splits = ImmutableSortedSet.copyOf(names.toArray)
       tableOps.addSplits(attributeIndexTable, splits)
     }

geomesa-core/src/main/scala/org/locationtech/geomesa/core/data/tables/AttributeTable.scala

@@ -17,7 +17,7 @@
 
 package org.locationtech.geomesa.core.data.tables
 
-import java.util.{Date, Locale}
+import java.util.{Date, Locale, Collection => JCollection, Map => JMap}
 
 import com.typesafe.scalalogging.slf4j.Logging
 import org.apache.accumulo.core.client.BatchWriter
@@ -27,11 +27,14 @@ import org.apache.hadoop.io.Text
 import org.calrissian.mango.types.{LexiTypeEncoders, SimpleTypeEncoders, TypeEncoder}
 import org.joda.time.format.ISODateTimeFormat
 import org.locationtech.geomesa.core.data._
-import org.locationtech.geomesa.core.index._
+import org.locationtech.geomesa.core.index
+import org.locationtech.geomesa.core.index.IndexEntry
+import org.locationtech.geomesa.utils.geotools.Conversions.RichAttributeDescriptor
 import org.opengis.feature.`type`.AttributeDescriptor
 import org.opengis.feature.simple.SimpleFeature
 
 import scala.collection.JavaConversions._
+import scala.collection.JavaConverters._
 import scala.util.{Failure, Success, Try}
 
 /**
@@ -85,70 +88,107 @@ object AttributeTable extends GeoMesaTable with Logging {
                                  delete: Boolean = false): Seq[Mutation] = {
     val cq = new Text(feature.getID)
     lazy val value = IndexEntry.encodeIndexValue(feature)
-    indexedAttributes.map { descriptor =>
+    indexedAttributes.flatMap { descriptor =>
       val attribute = Option(feature.getAttribute(descriptor.getName))
-      val m = new Mutation(getAttributeIndexRow(rowIdPrefix, descriptor.getLocalName, attribute))
+      val mutations = getAttributeIndexRows(rowIdPrefix, descriptor, attribute).map(new Mutation(_))
       if (delete) {
-        m.putDelete(EMPTY_COLF, cq, visibility)
+        mutations.foreach(_.putDelete(EMPTY_COLF, cq, visibility))
       } else {
-        m.put(EMPTY_COLF, cq, visibility, value)
+        mutations.foreach(_.put(EMPTY_COLF, cq, visibility, value))
       }
-      m
+      mutations
     }
   }
 
   /**
-   * Gets a row key for the attribute index
+   * Gets row keys for the attribute index. Usually a single row will be returned, but collections
+   * will result in multiple rows.
    *
-   * @param attributeName
-   * @param attributeValue
+   * @param rowIdPrefix
+   * @param descriptor
+   * @param value
    * @return
    */
-  def getAttributeIndexRow(rowIdPrefix: String, attributeName: String, attributeValue: Option[Any]): String =
-    getAttributeIndexRowPrefix(rowIdPrefix, attributeName) ++ encode(attributeValue)
+  def getAttributeIndexRows(rowIdPrefix: String,
+                            descriptor: AttributeDescriptor,
+                            value: Option[Any]): Seq[String] = {
+    val prefix = getAttributeIndexRowPrefix(rowIdPrefix, descriptor)
+    encode(value, descriptor).map(prefix + _)
+  }
 
   /**
    * Gets a prefix for an attribute row - useful for ranges over a particular attribute
    *
-   * @param attributeName
+   * @param rowIdPrefix
+   * @param descriptor
    * @return
    */
-  def getAttributeIndexRowPrefix(rowIdPrefix: String, attributeName: String): String =
-    rowIdPrefix ++ attributeName ++ NULLBYTE
+  def getAttributeIndexRowPrefix(rowIdPrefix: String, descriptor: AttributeDescriptor): String =
+    rowIdPrefix ++ descriptor.getLocalName ++ NULLBYTE
 
   /**
    * Decodes an attribute value out of row string
    *
    * @param rowIdPrefix table sharing prefix
-   * @param attributeType class of the attribute we're decoding
+   * @param descriptor the attribute we're decoding
    * @param row
    * @return
    */
-  def decodeAttributeIndexRow(rowIdPrefix: String, attributeType: Class[_], row: String): Try[AttributeIndexRow] =
+  def decodeAttributeIndexRow(rowIdPrefix: String,
+                              descriptor: AttributeDescriptor,
+                              row: String): Try[AttributeIndexRow] =
     for {
       suffix <- Try(row.substring(rowIdPrefix.length))
       separator = suffix.indexOf(NULLBYTE)
-      name <- Try(suffix.substring(0, separator))
-      encodedValue <- Try(suffix.substring(separator + 1))
-      decodedValue <- Try(decode(encodedValue, attributeType))
+      name = suffix.substring(0, separator)
+      encodedValue = suffix.substring(separator + 1)
+      decodedValue = decode(encodedValue, descriptor)
     } yield {
       AttributeIndexRow(name, decodedValue)
     }
 
   /**
-   * Lexicographically encode the value
+   * Lexicographically encode the value. Collections will return multiple rows, one for each entry.
    *
    * @param valueOption
+   * @param descriptor
    * @return
    */
-  def encode(valueOption: Option[Any]): String = {
+  def encode(valueOption: Option[Any], descriptor: AttributeDescriptor): Seq[String] = {
     val value = valueOption.getOrElse(nullString)
-    Try(typeRegistry.encode(value)).getOrElse(value.toString)
+    if (descriptor.isCollection) {
+      // encode each value into a separate row
+      value.asInstanceOf[JCollection[_]].asScala.toSeq.map(Option(_).getOrElse(nullString)).map(typeEncode)
+    } else if (descriptor.isMap) {
+      // TODO GEOMESA-454 - support querying against map attributes
+      Seq.empty
+    } else {
+      Seq(typeEncode(value))
+    }
   }
 
-  def decode(encoded: String, attributeType: Class[_]): Any = {
-    val alias = attributeType.getSimpleName.toLowerCase(Locale.US)
-    typeRegistry.decode(alias, encoded)
+  private def typeEncode(value: Any): String = Try(typeRegistry.encode(value)).getOrElse(value.toString)
+
+  /**
+   * Decode an encoded value. Note that for collection types, only a single entry of the collection
+   * will be decoded - this is because the collection entries have been broken up into multiple rows.
+   *
+   * @param encoded
+   * @param descriptor
+   * @return
+   */
+  def decode(encoded: String, descriptor: AttributeDescriptor): Any = {
+    if (descriptor.isCollection) {
+      // get the alias from the type of values in the collection
+      val alias = index.getCollectionType(descriptor).map(_.getSimpleName.toLowerCase(Locale.US)).head
+      Seq(typeRegistry.decode(alias, encoded)).asJava
+    } else if (descriptor.isMap) {
+      // TODO GEOMESA-454 - support querying against map attributes
+      Map.empty.asJava
+    } else {
+      val alias = descriptor.getType.getBinding.getSimpleName.toLowerCase(Locale.US)
+      typeRegistry.decode(alias, encoded)
+    }
   }
 
   private val dateFormat = ISODateTimeFormat.dateTime()

geomesa-core/src/main/scala/org/locationtech/geomesa/core/index/AttributeIdxStrategy.scala

@@ -18,14 +18,15 @@ package org.locationtech.geomesa.core.index
 
 
 import java.util.Map.Entry
+import java.util.{Collection => JCollection, Map => JMap}
 
+import com.typesafe.scalalogging.slf4j.Logging
 import org.apache.accumulo.core.client.{BatchScanner, IteratorSetting, Scanner}
 import org.apache.accumulo.core.data.{Key, Value, Range => AccRange}
 import org.apache.hadoop.io.Text
 import org.geotools.data.Query
 import org.geotools.filter.text.ecql.ECQL
 import org.geotools.temporal.`object`.DefaultPeriod
-import org.locationtech.geomesa.core.DEFAULT_FILTER_PROPERTY_NAME
 import org.locationtech.geomesa.core.data.FeatureEncoding.FeatureEncoding
 import org.locationtech.geomesa.core.data._
 import org.locationtech.geomesa.core.data.tables.{AttributeTable, RecordTable}
@@ -34,14 +35,17 @@ import org.locationtech.geomesa.core.index.FilterHelper._
 import org.locationtech.geomesa.core.index.QueryPlanner._
 import org.locationtech.geomesa.core.iterators._
 import org.locationtech.geomesa.core.util.{BatchMultiScanner, SelfClosingIterator}
+import org.locationtech.geomesa.core.{DEFAULT_FILTER_PROPERTY_NAME, index}
+import org.locationtech.geomesa.utils.geotools.Conversions.RichAttributeDescriptor
 import org.opengis.feature.simple.SimpleFeatureType
 import org.opengis.filter.expression.{Expression, Literal, PropertyName}
 import org.opengis.filter.temporal.{After, Before, During, TEquals}
 import org.opengis.filter.{Filter, PropertyIsEqualTo, PropertyIsLike, _}
 
 import scala.collection.JavaConversions._
+import scala.collection.JavaConverters._
 
-trait AttributeIdxStrategy extends Strategy {
+trait AttributeIdxStrategy extends Strategy with Logging {
 
   /**
    * Perform scan against the Attribute Index Table and get an iterator returning records from the Record table
@@ -77,7 +81,7 @@ trait AttributeIdxStrategy extends Strategy {
 
     val opts = oFilter.map(f => DEFAULT_FILTER_PROPERTY_NAME -> ECQL.toCQL(f)).toMap
 
-    iteratorChoice.iterator match {
+    val iter = iteratorChoice.iterator match {
       case IndexOnlyIterator =>
         indexOnlyIterator(attrScanner,
                           featureType,
@@ -100,6 +104,19 @@ trait AttributeIdxStrategy extends Strategy {
                            opts,
                            output)
     }
+
+    // wrap with a de-duplicator if the attribute could have multiple values, and it won't be
+    // de-duped by the query planner
+    if (!IndexSchema.mayContainDuplicates(featureType)
+        && featureType.getDescriptor(attributeName).isMultiValued) {
+      val returnSft = Option(query.getHints.get(TRANSFORM_SCHEMA).asInstanceOf[SimpleFeatureType])
+          .getOrElse(featureType)
+      val decoder = SimpleFeatureDecoder(returnSft, iqp.featureEncoding)
+      val deduper = new DeDuplicatingIterator(iter, (_: Key, value: Value) => decoder.extractFeatureId(value))
+      SelfClosingIterator(deduper)
+    } else {
+      iter
+    }
   }
 
   /**
@@ -242,21 +259,34 @@ trait AttributeIdxStrategy extends Strategy {
    * @return
    */
   def getEncodedAttrIdxRow(sft: SimpleFeatureType, prop: String, value: Any): String = {
+    val descriptor = sft.getDescriptor(prop)
     // the class type as defined in the SFT
-    val expectedBinding = sft.getDescriptor(prop).getType.getBinding
+    val expectedBinding = descriptor.getType.getBinding
     // the class type of the literal pulled from the query
     val actualBinding = value.getClass
-    val typedValue =
-      if (expectedBinding.equals(actualBinding)) {
-        value
+    val typedValue = if (expectedBinding == actualBinding) {
+      value
+    } else if (descriptor.isCollection) {
+      // we need to encode with the collection type
+      val collectionType = index.getCollectionType(descriptor).head
+      if (collectionType == actualBinding) {
+        Seq(value).asJava
       } else {
-        // type mismatch, encoding won't work b/c class is stored as part of the row
-        // try to convert to the appropriate class
-        AttributeTable.convertType(value, actualBinding, expectedBinding)
+        Seq(AttributeTable.convertType(value, actualBinding, collectionType)).asJava
       }
+    } else if (descriptor.isMap) {
+      // TODO GEOMESA-454 - support querying against map attributes
+      Map.empty.asJava
+    } else {
+      // type mismatch, encoding won't work b/c value is wrong class
+      // try to convert to the appropriate class
+      AttributeTable.convertType(value, actualBinding, expectedBinding)
+    }
 
     val rowIdPrefix = org.locationtech.geomesa.core.index.getTableSharingPrefix(sft)
-    AttributeTable.getAttributeIndexRow(rowIdPrefix, prop, Some(typedValue))
+    // grab the first encoded row - right now there will only ever be a single item in the seq
+    // eventually we may support searching a whole collection at once
+    AttributeTable.getAttributeIndexRows(rowIdPrefix, descriptor, Some(typedValue)).head
   }
 
   /**
@@ -305,36 +335,38 @@ class AttributeIdxEqualsStrategy extends AttributeIdxStrategy {
 
   override def execute(acc: AccumuloConnectorCreator,
                        iqp: QueryPlanner,
-                       featureType: SimpleFeatureType,
+                       sft: SimpleFeatureType,
                        query: Query,
                        output: ExplainerOutputType): SelfClosingIterator[Entry[Key, Value]] = {
-    val (strippedQuery, filter) = partitionFilter(query, featureType)
+    val (strippedQuery, filter) = partitionFilter(query, sft)
     val (prop, range) =
       filter match {
         case f: PropertyIsEqualTo =>
           val (prop, lit, _) = checkOrder(f.getExpression1, f.getExpression2)
-          (prop, AccRange.exact(getEncodedAttrIdxRow(featureType, prop, lit)))
+          (prop, AccRange.exact(getEncodedAttrIdxRow(sft, prop, lit)))
 
         case f: TEquals =>
           val (prop, lit, _) = checkOrder(f.getExpression1, f.getExpression2)
-          (prop, AccRange.exact(getEncodedAttrIdxRow(featureType, prop, lit)))
+          (prop, AccRange.exact(getEncodedAttrIdxRow(sft, prop, lit)))
 
         case f: PropertyIsNil =>
           val prop = f.getExpression.asInstanceOf[PropertyName].getPropertyName
-          val rowIdPrefix = org.locationtech.geomesa.core.index.getTableSharingPrefix(featureType)
-          (prop, AccRange.exact(AttributeTable.getAttributeIndexRow(rowIdPrefix, prop, None)))
+          val rowIdPrefix = org.locationtech.geomesa.core.index.getTableSharingPrefix(sft)
+          val exact = AttributeTable.getAttributeIndexRows(rowIdPrefix, sft.getDescriptor(prop), None).head
+          (prop, AccRange.exact(exact))
 
         case f: PropertyIsNull =>
           val prop = f.getExpression.asInstanceOf[PropertyName].getPropertyName
-          val rowIdPrefix = org.locationtech.geomesa.core.index.getTableSharingPrefix(featureType)
-          (prop, AccRange.exact(AttributeTable.getAttributeIndexRow(rowIdPrefix, prop, None)))
+          val rowIdPrefix = org.locationtech.geomesa.core.index.getTableSharingPrefix(sft)
+          val exact = AttributeTable.getAttributeIndexRows(rowIdPrefix, sft.getDescriptor(prop), None).head
+          (prop, AccRange.exact(exact))
 
         case _ =>
           val msg = s"Unhandled filter type in equals strategy: ${filter.getClass.getName}"
           throw new RuntimeException(msg)
       }
 
-    attrIdxQuery(acc, strippedQuery, iqp, featureType, prop, range, output)
+    attrIdxQuery(acc, strippedQuery, iqp, sft, prop, range, output)
   }
 }
 
@@ -415,31 +447,33 @@ class AttributeIdxRangeStrategy extends AttributeIdxStrategy {
     attrIdxQuery(acc, strippedQuery, iqp, featureType, prop, range, output)
   }
 
-  private def greaterThanRange(featureType: SimpleFeatureType, prop: String, lit: AnyRef): AccRange = {
-    val rowIdPrefix = org.locationtech.geomesa.core.index.getTableSharingPrefix(featureType)
-    val start = new Text(getEncodedAttrIdxRow(featureType, prop, lit))
-    val end = AccRange.followingPrefix(new Text(AttributeTable.getAttributeIndexRowPrefix(rowIdPrefix, prop)))
+  private def greaterThanRange(sft: SimpleFeatureType, prop: String, lit: AnyRef): AccRange = {
+    val rowIdPrefix = getTableSharingPrefix(sft)
+    val start = new Text(getEncodedAttrIdxRow(sft, prop, lit))
+    val endPrefix = AttributeTable.getAttributeIndexRowPrefix(rowIdPrefix, sft.getDescriptor(prop))
+    val end = AccRange.followingPrefix(new Text(endPrefix))
     new AccRange(start, false, end, false)
   }
 
-  private def greaterThanOrEqualRange(featureType: SimpleFeatureType, prop: String, lit: AnyRef): AccRange = {
-    val rowIdPrefix = org.locationtech.geomesa.core.index.getTableSharingPrefix(featureType)
-    val start = new Text(getEncodedAttrIdxRow(featureType, prop, lit))
-    val end = AccRange.followingPrefix(new Text(AttributeTable.getAttributeIndexRowPrefix(rowIdPrefix, prop)))
+  private def greaterThanOrEqualRange(sft: SimpleFeatureType, prop: String, lit: AnyRef): AccRange = {
+    val rowIdPrefix = getTableSharingPrefix(sft)
+    val start = new Text(getEncodedAttrIdxRow(sft, prop, lit))
+    val endPrefix = AttributeTable.getAttributeIndexRowPrefix(rowIdPrefix, sft.getDescriptor(prop))
+    val end = AccRange.followingPrefix(new Text(endPrefix))
     new AccRange(start, true, end, false)
   }
 
-  private def lessThanRange(featureType: SimpleFeatureType, prop: String, lit: AnyRef): AccRange = {
-    val rowIdPrefix = org.locationtech.geomesa.core.index.getTableSharingPrefix(featureType)
-    val start = AttributeTable.getAttributeIndexRowPrefix(rowIdPrefix, prop)
-    val end = getEncodedAttrIdxRow(featureType, prop, lit)
+  private def lessThanRange(sft: SimpleFeatureType, prop: String, lit: AnyRef): AccRange = {
+    val rowIdPrefix = getTableSharingPrefix(sft)
+    val start = AttributeTable.getAttributeIndexRowPrefix(rowIdPrefix, sft.getDescriptor(prop))
+    val end = getEncodedAttrIdxRow(sft, prop, lit)
     new AccRange(start, false, end, false)
   }
 
-  private def lessThanOrEqualRange(featureType: SimpleFeatureType, prop: String, lit: AnyRef): AccRange = {
-    val rowIdPrefix = org.locationtech.geomesa.core.index.getTableSharingPrefix(featureType)
-    val start = AttributeTable.getAttributeIndexRowPrefix(rowIdPrefix, prop)
-    val end = getEncodedAttrIdxRow(featureType, prop, lit)
+  private def lessThanOrEqualRange(sft: SimpleFeatureType, prop: String, lit: AnyRef): AccRange = {
+    val rowIdPrefix = getTableSharingPrefix(sft)
+    val start = AttributeTable.getAttributeIndexRowPrefix(rowIdPrefix, sft.getDescriptor(prop))
+    val end = getEncodedAttrIdxRow(sft, prop, lit)
     new AccRange(start, false, end, true)
   }
 }