properties: upgrade HBaseVertex to support multi-properties

The HBase graph tinkerpop implementation specifies Cardinality.single as
the default.  Since HBase does not (at the time of this writing) support
multiple values, this feature wraps the multi-values in a list and
writes the list to HBase.  As a consequence of this list wrapper,
indexed properties can still only support-single value properties.

When reading/writing from HBase, most properties will be written as
single objects.  Only when a property is specifically added with
non-single cardinality will it be written to HBase as a list.

In order to avoid java native serialization, the serialization of
non-single property values uses Kryo (with a small lightweight wrapper
class to accomplish this).

Author: Michael Rappazzo

src/main/java/io/hgraphdb/HBaseBulkLoader.java

@@ -1,6 +1,14 @@
 package io.hgraphdb;
 
-import io.hgraphdb.mutators.*;
+import io.hgraphdb.mutators.Creator;
+import io.hgraphdb.mutators.EdgeIndexRemover;
+import io.hgraphdb.mutators.EdgeIndexWriter;
+import io.hgraphdb.mutators.EdgeWriter;
+import io.hgraphdb.mutators.PropertyWriter;
+import io.hgraphdb.mutators.VertexIndexRemover;
+import io.hgraphdb.mutators.VertexIndexWriter;
+import io.hgraphdb.mutators.VertexWriter;
+
 import org.apache.hadoop.hbase.TableName;
 import org.apache.hadoop.hbase.client.BufferedMutator;
 import org.apache.hadoop.hbase.client.BufferedMutatorParams;
@@ -84,7 +92,7 @@ public Vertex addVertex(final Object... keyValues) {
             idValue = HBaseGraphUtils.generateIdIfNeeded(idValue);
             long now = System.currentTimeMillis();
             HBaseVertex vertex = new HBaseVertex(graph, idValue, label, now, now,
-                    HBaseGraphUtils.propertiesToMap(keyValues));
+                    HBaseGraphUtils.propertiesToMultimap(keyValues));
             vertex.validate();
 
             Iterator<IndexMetadata> indices = vertex.getIndices(OperationType.WRITE);
@@ -192,14 +200,14 @@ public void setProperty(Vertex vertex, String key, Object value) {
             boolean hasIndex = v.hasIndex(OperationType.WRITE, key);
             if (hasIndex) {
                 // only load old value if using index
-                oldValue = v.getProperty(key);
+                oldValue = v.getSingleProperty(key).orElse(null);
                 if (oldValue != null && !oldValue.equals(value)) {
                     VertexIndexRemover indexRemover = new VertexIndexRemover(graph, v, key, null);
                     if (vertexIndicesMutator != null) vertexIndicesMutator.mutate(getMutationList(indexRemover.constructMutations()));
                 }
             }
 
-            v.getProperties().put(key, value);
+            v.cacheProperty(key, value);
             v.updatedAt(System.currentTimeMillis());
 
             if (hasIndex) {
@@ -220,7 +228,8 @@ private List<? extends Mutation> getMutationList(Iterator<? extends Mutation> mu
                 m -> m.setDurability(skipWAL ? Durability.SKIP_WAL : Durability.USE_DEFAULT)));
     }
 
-    public void close() {
+    @Override
+	public void close() {
         try {
             if (edgesMutator != null) edgesMutator.close();
             if (edgeIndicesMutator != null) edgeIndicesMutator.close();

src/main/java/io/hgraphdb/HBaseEdge.java

@@ -2,6 +2,9 @@
 
 import io.hgraphdb.models.EdgeIndexModel;
 import io.hgraphdb.models.EdgeModel;
+import io.hgraphdb.mutators.Mutator;
+import io.hgraphdb.mutators.Mutators;
+
 import org.apache.tinkerpop.gremlin.structure.Direction;
 import org.apache.tinkerpop.gremlin.structure.Edge;
 import org.apache.tinkerpop.gremlin.structure.Property;
@@ -12,15 +15,22 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.util.Collections;
 import java.util.Iterator;
 import java.util.Map;
+import java.util.Set;
+import java.util.Map.Entry;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.stream.Stream;
 
 public class HBaseEdge extends HBaseElement implements Edge {
 
     private static final Logger LOGGER = LoggerFactory.getLogger(HBaseEdge.class);
 
     private Vertex inVertex;
     private Vertex outVertex;
+    protected Map<String, Object> properties;
+    protected transient boolean propertiesFullyLoaded;
 
     public HBaseEdge(HBaseGraph graph, Object id) {
         this(graph, id, null, null, null, null, null, null);
@@ -37,10 +47,12 @@ public HBaseEdge(HBaseGraph graph, Object id, String label, Long createdAt, Long
 
     public HBaseEdge(HBaseGraph graph, Object id, String label, Long createdAt, Long updatedAt, Map<String, Object> properties,
                      boolean propertiesFullyLoaded, Vertex inVertex, Vertex outVertex) {
-        super(graph, id, label, createdAt, updatedAt, properties, propertiesFullyLoaded);
+        super(graph, id, label, createdAt, updatedAt);
 
         this.inVertex = inVertex;
         this.outVertex = outVertex;
+        this.properties = properties;
+        this.propertiesFullyLoaded = propertiesFullyLoaded;
     }
 
     @Override
@@ -55,13 +67,26 @@ public ElementType getElementType() {
         return ElementType.EDGE;
     }
 
+    public Map<String, Object> getProperties() {
+        if (properties == null || !propertiesFullyLoaded) {
+            load();
+            propertiesFullyLoaded = true;
+        }
+        return properties;
+    }
+
     @Override
     public void copyFrom(HBaseElement element) {
         super.copyFrom(element);
         if (element instanceof HBaseEdge) {
             HBaseEdge copy = (HBaseEdge) element;
             if (copy.inVertex != null) this.inVertex = copy.inVertex;
             if (copy.outVertex != null) this.outVertex = copy.outVertex;
+            if (copy.properties != null
+                    && (copy.propertiesFullyLoaded || this.properties == null)) {
+                this.properties = new ConcurrentHashMap<>(copy.properties);
+                this.propertiesFullyLoaded = copy.propertiesFullyLoaded;
+            }
         }
     }
 
@@ -100,6 +125,101 @@ public Vertex getVertex(Direction direction) throws IllegalArgumentException {
         return Direction.IN.equals(direction) ? inVertex : outVertex;
     }
 
+    public void setProperty(String key, Object value) {
+        ElementHelper.validateProperty(key, value);
+
+        graph.validateProperty(getElementType(), label, key, value);
+
+        // delete from index model before setting property
+        Object oldValue = null;
+        boolean hasIndex = hasIndex(OperationType.WRITE, key);
+        if (hasIndex) {
+            // only load old value if using index
+            oldValue = getProperty(key);
+            if (oldValue != null && !oldValue.equals(value)) {
+                deleteFromIndexModel(key, null);
+            }
+        }
+
+        getProperties().put(key, value);
+        updatedAt(System.currentTimeMillis());
+
+        if (hasIndex) {
+            if (oldValue == null || !oldValue.equals(value)) {
+                writeToIndexModel(key);
+            }
+        }
+        Mutator writer = getModel().writeProperty(this, key, value);
+        Mutators.write(getTable(), writer);
+    }
+
+    public void incrementProperty(String key, long value) {
+        if (!graph.configuration().getUseSchema()) {
+            throw new HBaseGraphNoSchemaException("Schema not enabled");
+        }
+        ElementHelper.validateProperty(key, value);
+
+        graph.validateProperty(getElementType(), label, key, value);
+
+        updatedAt(System.currentTimeMillis());
+
+        Mutator writer = getModel().incrementProperty(this, key, value);
+        long newValue = Mutators.increment(getTable(), writer, key);
+        getProperties().put(key, newValue);
+    }
+
+    @Override
+    public Set<String> keys() {
+        return Collections.unmodifiableSet(properties.keySet());
+    }
+
+    public void removeProperty(String key) {
+        Object value = getProperty(key);
+        if (value != null) {
+            // delete from index model before removing property
+            boolean hasIndex = hasIndex(OperationType.WRITE, key);
+            if (hasIndex) {
+                deleteFromIndexModel(key, null);
+            }
+
+            getProperties().remove(key);
+            updatedAt(System.currentTimeMillis());
+
+            Mutator writer = getModel().clearProperty(this, key);
+            Mutators.write(getTable(), writer);
+        }
+    }
+
+    @Override
+    public Stream<Entry<String, Object>> propertyEntriesStream() {
+        return properties.entrySet().stream();
+    }
+
+    @Override
+    public int propertySize() {
+        return this.properties.size();
+    }
+
+    @SuppressWarnings("unchecked")
+    public <V> V getProperty(String key) {
+        if (properties != null) {
+            // optimization for partially loaded properties
+            V val = (V) properties.get(key);
+            if (val != null) return val;
+        }
+        return (V) getProperties().get(key);
+    }
+
+    @Override
+	public boolean hasProperty(String key) {
+        if (properties != null) {
+            // optimization for partially loaded properties
+            Object val = properties.get(key);
+            if (val != null) return true;
+        }
+        return keys().contains(key);
+    }
+
     @Override
     public void remove() {
         // Get rid of the endpoints and edge themselves.
@@ -115,7 +235,7 @@ public void remove() {
 
     @Override
     public <V> Iterator<Property<V>> properties(final String... propertyKeys) {
-        Iterable<String> keys = getPropertyKeys();
+        Iterable<String> keys = keys();
         Iterator<String> filter = IteratorUtils.filter(keys.iterator(),
                 key -> ElementHelper.keyExists(key, propertyKeys));
         return IteratorUtils.map(filter,