[geode] Update to Apache Geode version 1.0.0-incubating.M3 (#828)

[geode] Update to Apache Geode version 1.0.0-incubating.M3

geode/README.md

@@ -42,6 +42,7 @@ Start a locator and two servers:
 gfsh> start locator --name=locator1
 gfsh> start server --name=server1 --server-port=40404
 gfsh> start server --name=server2 --server-port=40405
+gfsh> configure pdx --read-serialized=true
 ```
 
 Create the "usertable" region required by YCSB driver:

geode/src/main/java/com/yahoo/ycsb/db/GeodeClient.java

@@ -23,6 +23,10 @@ import com.gemstone.gemfire.cache.client.ClientCacheFactory;
 import com.gemstone.gemfire.cache.client.ClientRegionFactory;
 import com.gemstone.gemfire.cache.client.ClientRegionShortcut;
 import com.gemstone.gemfire.internal.admin.remote.DistributionLocatorId;
+import com.gemstone.gemfire.internal.cache.GemFireCacheImpl;
+import com.gemstone.gemfire.pdx.JSONFormatter;
+import com.gemstone.gemfire.pdx.PdxInstance;
+import com.gemstone.gemfire.pdx.PdxInstanceFactory;
 import com.yahoo.ycsb.*;
 
 import java.util.*;
@@ -36,42 +40,55 @@ import java.util.*;
  * geode.serverhost=host</code> properties on YCSB command line.
  * A locator may also be used for discovering a cacheServer
  * by using the property <code>geode.locator=host[port]</code></p>
- *
+ * <p>
  * <p>To run this client in a peer-to-peer topology with other Geode
  * nodes, use the property <code>geode.topology=p2p</code>. Running
  * in p2p mode will enable embedded caching in this client.</p>
- *
+ * <p>
  * <p>YCSB by default does its operations against "usertable". When running
  * as a client this is a <code>ClientRegionShortcut.PROXY</code> region,
  * when running in p2p mode it is a <code>RegionShortcut.PARTITION</code>
  * region. A cache.xml defining "usertable" region can be placed in the
  * working directory to override these region definitions.</p>
- *
  */
 public class GeodeClient extends DB {
-  /** property name of the port where Geode server is listening for connections. */
+  /**
+   * property name of the port where Geode server is listening for connections.
+   */
   private static final String SERVERPORT_PROPERTY_NAME = "geode.serverport";
 
-  /** property name of the host where Geode server is running. */
+  /**
+   * property name of the host where Geode server is running.
+   */
   private static final String SERVERHOST_PROPERTY_NAME = "geode.serverhost";
 
-  /** default value of {@link #SERVERHOST_PROPERTY_NAME}. */
+  /**
+   * default value of {@link #SERVERHOST_PROPERTY_NAME}.
+   */
   private static final String SERVERHOST_PROPERTY_DEFAULT = "localhost";
 
-  /** property name to specify a Geode locator. This property can be used in both
-   * client server and p2p topology */
+  /**
+   * property name to specify a Geode locator. This property can be used in both
+   * client server and p2p topology
+   */
   private static final String LOCATOR_PROPERTY_NAME = "geode.locator";
 
-  /** property name to specify Geode topology. */
+  /**
+   * property name to specify Geode topology.
+   */
   private static final String TOPOLOGY_PROPERTY_NAME = "geode.topology";
 
-  /** value of {@value #TOPOLOGY_PROPERTY_NAME} when peer to peer topology should be used.
-   *  (client-server topology is default) */
+  /**
+   * value of {@value #TOPOLOGY_PROPERTY_NAME} when peer to peer topology should be used.
+   * (client-server topology is default)
+   */
   private static final String TOPOLOGY_P2P_VALUE = "p2p";
 
   private GemFireCache cache;
 
-  /** true if ycsb client runs as a client to a Geode cache server. */
+  /**
+   * true if ycsb client runs as a client to a Geode cache server.
+   */
   private boolean isClient;
 
   @Override
@@ -119,16 +136,16 @@ public class GeodeClient extends DB {
   @Override
   public Status read(String table, String key, Set<String> fields,
                      HashMap<String, ByteIterator> result) {
-    Region<String, Map<String, byte[]>> r = getRegion(table);
-    Map<String, byte[]> val = r.get(key);
+    Region<String, PdxInstance> r = getRegion(table);
+    PdxInstance val = r.get(key);
     if (val != null) {
       if (fields == null) {
-        for (Map.Entry<String, byte[]> entry : val.entrySet()) {
-          result.put(entry.getKey(), new ByteArrayByteIterator(entry.getValue()));
+        for (String fieldName : val.getFieldNames()) {
+          result.put(fieldName, new ByteArrayByteIterator((byte[]) val.getField(fieldName)));
         }
       } else {
         for (String field : fields) {
-          result.put(field, new ByteArrayByteIterator(val.get(field)));
+          result.put(field, new ByteArrayByteIterator((byte[]) val.getField(field)));
         }
       }
       return Status.OK;
@@ -161,24 +178,26 @@ public class GeodeClient extends DB {
     return Status.OK;
   }
 
-  private Map<String, byte[]> convertToBytearrayMap(Map<String, ByteIterator> values) {
-    Map<String, byte[]> retVal = new HashMap<String, byte[]>();
+  private PdxInstance convertToBytearrayMap(Map<String, ByteIterator> values) {
+    GemFireCacheImpl gci = (GemFireCacheImpl) CacheFactory.getAnyInstance();
+    PdxInstanceFactory pdxInstanceFactory = gci.createPdxInstanceFactory(JSONFormatter.JSON_CLASSNAME);
+
     for (Map.Entry<String, ByteIterator> entry : values.entrySet()) {
-      retVal.put(entry.getKey(), entry.getValue().toArray());
+      pdxInstanceFactory.writeByteArray(entry.getKey(), entry.getValue().toArray());
     }
-    return retVal;
+    return pdxInstanceFactory.create();
   }
 
-  private Region<String, Map<String, byte[]>> getRegion(String table) {
-    Region<String, Map<String, byte[]>> r = cache.getRegion(table);
+  private Region<String, PdxInstance> getRegion(String table) {
+    Region<String, PdxInstance> r = cache.getRegion(table);
     if (r == null) {
       try {
         if (isClient) {
-          ClientRegionFactory<String, Map<String, byte[]>> crf =
+          ClientRegionFactory<String, PdxInstance> crf =
               ((ClientCache) cache).createClientRegionFactory(ClientRegionShortcut.PROXY);
           r = crf.create(table);
         } else {
-          RegionFactory<String, Map<String, byte[]>> rf = ((Cache) cache).createRegionFactory(RegionShortcut.PARTITION);
+          RegionFactory<String, PdxInstance> rf = ((Cache) cache).createRegionFactory(RegionShortcut.PARTITION);
           r = rf.create(table);
         }
       } catch (RegionExistsException e) {

pom.xml

@@ -74,7 +74,7 @@ LICENSE file.
     <hbase10.version>1.0.2</hbase10.version>
     <accumulo.version>1.6.0</accumulo.version>
     <cassandra.cql.version>3.0.0</cassandra.cql.version>
-    <geode.version>1.0.0-incubating.M2</geode.version>
+    <geode.version>1.0.0-incubating.M3</geode.version>
     <googlebigtable.version>0.2.3</googlebigtable.version>
     <infinispan.version>7.2.2.Final</infinispan.version>
     <kudu.version>0.9.0</kudu.version>