[cloudspanner] Add binding for Google's Cloud Spanner. (#939)

bin/bindings.properties

@@ -34,6 +34,7 @@ azuretablestorage:com.yahoo.ycsb.db.azuretablestorage.AzureClient
 basic:com.yahoo.ycsb.BasicDB
 cassandra-cql:com.yahoo.ycsb.db.CassandraCQLClient
 cassandra2-cql:com.yahoo.ycsb.db.CassandraCQLClient
+cloudspanner:com.yahoo.ycsb.db.cloudspanner.CloudSpannerClient
 couchbase:com.yahoo.ycsb.db.CouchbaseClient
 couchbase2:com.yahoo.ycsb.db.couchbase2.Couchbase2Client
 azuredocumentdb:com.yahoo.ycsb.db.azuredocumentdb.AzureDocumentDBClient

bin/ycsb

@@ -54,14 +54,15 @@ DATABASES = {
     "accumulo"     : "com.yahoo.ycsb.db.accumulo.AccumuloClient",
     "aerospike"    : "com.yahoo.ycsb.db.AerospikeClient",
     "arangodb"     : "com.yahoo.ycsb.db.ArangoDBClient",
-    "arangodb3"     : "com.yahoo.ycsb.db.arangodb.ArangoDB3Client",
+    "arangodb3"    : "com.yahoo.ycsb.db.arangodb.ArangoDB3Client",
     "asynchbase"   : "com.yahoo.ycsb.db.AsyncHBaseClient",
     "basic"        : "com.yahoo.ycsb.BasicDB",
     "cassandra-cql": "com.yahoo.ycsb.db.CassandraCQLClient",
     "cassandra2-cql": "com.yahoo.ycsb.db.CassandraCQLClient",
+    "cloudspanner" : "com.yahoo.ycsb.db.cloudspanner.CloudSpannerClient",
     "couchbase"    : "com.yahoo.ycsb.db.CouchbaseClient",
     "couchbase2"   : "com.yahoo.ycsb.db.couchbase2.Couchbase2Client",
-    "azuredocumentdb"   : "com.yahoo.ycsb.db.azuredocumentdb.AzureDocumentDBClient",
+    "azuredocumentdb" : "com.yahoo.ycsb.db.azuredocumentdb.AzureDocumentDBClient",
     "dynamodb"     : "com.yahoo.ycsb.db.DynamoDBClient",
     "elasticsearch": "com.yahoo.ycsb.db.ElasticsearchClient",
     "geode"        : "com.yahoo.ycsb.db.GeodeClient",

cloudspanner/README.md

+<!--
+Copyright (c) 2017 YCSB contributors. All rights reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License"); you
+may not use this file except in compliance with the License. You
+may obtain a copy of the License at
+
+http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+implied. See the License for the specific language governing
+permissions and limitations under the License. See accompanying
+LICENSE file.
+-->
+
+# Cloud Spanner Driver for YCSB
+
+This driver provides a YCSB workload binding for Google's Cloud Spanner database, the first relational database service that is both strongly consistent and horizontally scalable. This binding is implemented using the official Java client library for Cloud Spanner which uses GRPC for making calls.
+
+For best results, we strongly recommend running the benchmark from a Google Compute Engine (GCE) VM.
+
+## Running a Workload
+
+We recommend reading the [general guidelines](https://github.com/brianfrankcooper/YCSB/wiki/Running-a-Workload) in the YCSB documentation, and following the Cloud Spanner specific steps below.
+
+### 1. Set up Cloud Spanner with the Expected Schema
+
+Follow the [Quickstart instructions](https://cloud.google.com/spanner/docs/quickstart-console) in the Cloud Spanner documentation to set up a Cloud Spanner instance, and create a database with the following schema:
+
+```
+CREATE TABLE usertable (
+  id STRING(MAX),
+  field0 STRING(MAX),
+  field1 STRING(MAX),
+  field2 STRING(MAX),
+  field3 STRING(MAX),
+  field4 STRING(MAX),
+  field5 STRING(MAX),
+  field6 STRING(MAX),
+  field7 STRING(MAX),
+  field8 STRING(MAX),
+  field9 STRING(MAX),
+) PRIMARY KEY(id);
+```
+Make note of your project ID, instance ID, and database name.
+
+### 2. Set Up Your Environment and Auth
+
+Follow the [set up instructions](https://cloud.google.com/spanner/docs/getting-started/set-up) in the Cloud Spanner documentation to set up your environment and authentication. When not running on a GCE VM, make sure you run `gcloud auth application-default login`.
+
+### 3. Edit Properties
+
+In your YCSB root directory, edit `cloudspanner/conf/cloudspanner.properties` and specify your project ID, instance ID, and database name.
+
+### 4. Run the YCSB Shell
+
+Start the YCBS shell connected to Cloud Spanner using the following command:
+
+```
+./bin/ycsb shell cloudspanner -P cloudspanner/conf/cloudspanner.properties
+```
+
+You can use the `insert`, `read`, `update`, `scan`, and `delete` commands in the shell to experiment with your database and make sure the connection works. For example, try the following:
+
+```
+insert name field0=adam
+read name field0
+delete name
+```
+
+### 5. Load the Data
+
+You can load, say, 10 GB of data into your YCSB database using the following command:
+
+```
+./bin/ycsb load cloudspanner -P cloudspanner/conf/cloudspanner.properties -P workloads/workloada -p recordcount=10000000 -p cloudspanner.batchinserts=1000 -threads 10 -s
+```
+
+We recommend batching insertions so as to reach ~1 MB of data per commit request; this is controlled via the `cloudspanner.batchinserts` parameter which we recommend setting to `1000` during data load.
+
+If you wish to load a large database, you can run YCSB on multiple client VMs in parallel and use the `insertstart` and `insertcount` parameters to distribute the load as described [here](https://github.com/brianfrankcooper/YCSB/wiki/Running-a-Workload-in-Parallel). In this case, we recommend the following:
+
+* Use ordered inserts via specifying the YCSB parameter `insertorder=ordered`;
+* Use zero-padding so that ordered inserts are actually lexicographically ordered; the option `zeropadding = 12` is set in the default `cloudspanner.properties` file;
+* Split the key range evenly between client VMs;
+* Use few threads on each client VM, so that each individual commit request contains keys which are (close to) consecutive, and would thus likely address a single split; this also helps avoid overloading the servers.
+
+The idea is that we have a number of 'write heads' which are all writing to different parts of the database (and thus talking to different servers), but each individual head is writing its own data (more or less) in order. See the [best practices page](https://cloud.google.com/spanner/docs/best-practices#loading_data) for further details.
+
+### 6. Run a Workload
+
+After data load, you can a run a workload, say, workload B, using the following command:
+
+```
+./bin/ycsb run cloudspanner -P cloudspanner/conf/cloudspanner.properties -P workloads/workloadb -p recordcount=10000000 -p operationcount=1000000 -threads 10 -s 
+```
+
+Make sure that you use the same `insertorder` (i.e. `ordered` or `hashed`) and `zeropadding` as specified during the data load. Further details about running workloads are given in the [YCSB wiki pages](https://github.com/brianfrankcooper/YCSB/wiki/Running-a-Workload).
+
+## Configuration Options
+
+In addition to the standard YCSB parameters, the following Cloud Spanner specific options can be configured using the `-p` parameter or in `cloudspanner/conf/cloudspanner.properties`.
+
+* `cloudspanner.database`: (Required) The name of the database created in the instance, e.g. `ycsb-database`.
+* `cloudspanner.instance`: (Required) The ID of the Cloud Spanner instance, e.g. `ycsb-instance`.
+* `cloudspanner.project`: The ID of the project containing the Cloud Spanner instance, e.g. `myproject`. This is not strictly required and can often be automatically inferred from the environment.
+* `cloudspanner.readmode`: Allows choosing between the `read` and `query` interface of Cloud Spanner. The default is `query`.
+* `cloudspanner.batchinserts`: The number of inserts to batch into a single commit request. The default value is 1 which means no batching is done. Recommended value during data load is 1000.
+* `cloudspanner.boundedstaleness`: Number of seconds we allow reads to be stale for. Set to 0 for strong reads (default). For performance gains, this should be set to 10 seconds.

cloudspanner/conf/cloudspanner.properties

+# Copyright (c) 2017 YCSB contributors. All rights reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License"); you
+# may not use this file except in compliance with the License. You
+# may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+# implied. See the License for the specific language governing
+# permissions and limitations under the License. See accompanying
+# LICENSE file.
+
+# Core YCSB properties.
+table = usertable
+zeropadding = 12
+
+# Cloud Spanner properties
+cloudspanner.instance = ycsb-instance
+cloudspanner.database = ycsb-database
+
+cloudspanner.readmode = query
+cloudspanner.boundedstaleness = 0
+cloudspanner.batchinserts = 1

cloudspanner/pom.xml

+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+Copyright (c) 2017 YCSB contributors. All rights reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License"); you
+may not use this file except in compliance with the License. You
+may obtain a copy of the License at
+
+http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+implied. See the License for the specific language governing
+permissions and limitations under the License. See accompanying
+LICENSE file.
+-->
+
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+  <modelVersion>4.0.0</modelVersion>
+  <parent>
+    <groupId>com.yahoo.ycsb</groupId>
+    <artifactId>binding-parent</artifactId>
+    <version>0.13.0-SNAPSHOT</version>
+    <relativePath>../binding-parent/</relativePath>
+  </parent>
+
+  <artifactId>cloudspanner-binding</artifactId>
+  <name>Cloud Spanner DB Binding</name>
+  <packaging>jar</packaging>
+
+  <dependencies>
+    <dependency>
+      <groupId>com.google.cloud</groupId>
+      <artifactId>google-cloud-spanner</artifactId>
+      <version>${cloudspanner.version}</version>
+      <exclusions>
+        <exclusion> <!-- exclude an old version of Guava -->
+          <groupId>com.google.guava</groupId>
+          <artifactId>guava-jdk5</artifactId>
+        </exclusion>
+      </exclusions>
+    </dependency>
+    
+    <dependency>
+      <groupId>com.yahoo.ycsb</groupId>
+      <artifactId>core</artifactId>
+      <version>${project.version}</version>
+      <scope>provided</scope>
+    </dependency>
+    
+  </dependencies>
+</project>

cloudspanner/src/main/java/com/yahoo/ycsb/db/cloudspanner/CloudSpannerClient.java

+/**
+ * Copyright (c) 2017 YCSB contributors. All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you
+ * may not use this file except in compliance with the License. You
+ * may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+ * implied. See the License for the specific language governing
+ * permissions and limitations under the License. See accompanying
+ * LICENSE file.
+ */
+package com.yahoo.ycsb.db.cloudspanner;
+
+import com.google.common.base.Joiner;
+import com.google.cloud.spanner.DatabaseId;
+import com.google.cloud.spanner.DatabaseClient;
+import com.google.cloud.spanner.Key;
+import com.google.cloud.spanner.KeySet;
+import com.google.cloud.spanner.KeyRange;
+import com.google.cloud.spanner.Mutation;
+import com.google.cloud.spanner.Options;
+import com.google.cloud.spanner.ResultSet;
+import com.google.cloud.spanner.SessionPoolOptions;
+import com.google.cloud.spanner.Spanner;
+import com.google.cloud.spanner.SpannerOptions;
+import com.google.cloud.spanner.Statement;
+import com.google.cloud.spanner.Struct;
+import com.google.cloud.spanner.StructReader;
+import com.google.cloud.spanner.TimestampBound;
+import com.yahoo.ycsb.ByteIterator;
+import com.yahoo.ycsb.Client;
+import com.yahoo.ycsb.DB;
+import com.yahoo.ycsb.DBException;
+import com.yahoo.ycsb.Status;
+import com.yahoo.ycsb.StringByteIterator;
+import com.yahoo.ycsb.workloads.CoreWorkload;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+import java.util.Vector;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * YCSB Client for Google's Cloud Spanner.
+ */
+public class CloudSpannerClient extends DB {
+
+  /**
+   * The names of properties which can be specified in the config files and flags.
+   */
+  public static final class CloudSpannerProperties {
+    private CloudSpannerProperties() {}
+
+    /**
+     * The Cloud Spanner database name to use when running the YCSB benchmark, e.g. 'ycsb-database'.
+     */
+    static final String DATABASE = "cloudspanner.database";
+    /**
+     * The Cloud Spanner instance ID to use when running the YCSB benchmark, e.g. 'ycsb-instance'.
+     */
+    static final String INSTANCE = "cloudspanner.instance";
+    /**
+     * Choose between 'read' and 'query'. Affects both read() and scan() operations.
+     */
+    static final String READ_MODE = "cloudspanner.readmode";
+    /**
+     * The number of inserts to batch during the bulk loading phase. The default value is 1, which means no batching
+     * is done. Recommended value during data load is 1000.
+     */
+    static final String BATCH_INSERTS = "cloudspanner.batchinserts";
+    /**
+     * Number of seconds we allow reads to be stale for. Set to 0 for strong reads (default).
+     * For performance gains, this should be set to 10 seconds.
+     */
+    static final String BOUNDED_STALENESS = "cloudspanner.boundedstaleness";
+
+    // The properties below usually do not need to be set explicitly.
+
+    /**
+     * The Cloud Spanner project ID to use when running the YCSB benchmark, e.g. 'myproject'. This is not strictly
+     * necessary and can often be inferred from the environment.
+     */
+    static final String PROJECT = "cloudspanner.project";
+    /**
+     * The Cloud Spanner host name to use in the YCSB run.
+     */
+    static final String HOST = "cloudspanner.host";
+    /**
+     * Number of Cloud Spanner client channels to use. It's recommended to leave this to be the default value.
+     */
+    static final String NUM_CHANNELS = "cloudspanner.channels";
+  }
+
+  private static int fieldCount;
+
+  private static boolean queriesForReads;
+
+  private static int batchInserts;
+
+  private static TimestampBound timestampBound;
+
+  private static String standardQuery;
+
+  private static String standardScan;
+
+  private static final ArrayList<String> STANDARD_FIELDS = new ArrayList<>();
+
+  private static final String PRIMARY_KEY_COLUMN = "id";
+
+  private static final Logger LOGGER = Logger.getLogger(CloudSpannerClient.class.getName());
+
+  // Static lock for the class.
+  private static final Object CLASS_LOCK = new Object();
+
+  // Single Spanner client per process.
+  private static Spanner spanner = null;
+
+  // Single database client per process.
+  private static DatabaseClient dbClient = null;
+
+  // Buffered mutations on a per object/thread basis for batch inserts.
+  // Note that we have a separate CloudSpannerClient object per thread.
+  private final ArrayList<Mutation> bufferedMutations = new ArrayList<>();
+
+  private static void constructStandardQueriesAndFields(Properties properties) {
+    String table = properties.getProperty(CoreWorkload.TABLENAME_PROPERTY, CoreWorkload.TABLENAME_PROPERTY_DEFAULT);
+    standardQuery = new StringBuilder()
+        .append("SELECT * FROM ").append(table).append(" WHERE id=@key").toString();
+    standardScan = new StringBuilder()
+        .append("SELECT * FROM ").append(table).append(" WHERE id>=@startKey LIMIT @count").toString();
+    for (int i = 0; i < fieldCount; i++) {
+      STANDARD_FIELDS.add("field" + i);
+    }
+  }
+
+  private static Spanner getSpanner(Properties properties, String host, String project) {
+    if (spanner != null) {
+      return spanner;
+    }
+    String numChannels = properties.getProperty(CloudSpannerProperties.NUM_CHANNELS);
+    int numThreads = Integer.parseInt(properties.getProperty(Client.THREAD_COUNT_PROPERTY, "1"));
+    SpannerOptions.Builder optionsBuilder = SpannerOptions.newBuilder()
+        .setSessionPoolOption(SessionPoolOptions.newBuilder()
+            .setMinSessions(numThreads)
+            // Since we have no read-write transactions, we can set the write session fraction to 0.
+            .setWriteSessionsFraction(0)
+            .build());
+    if (host != null) {
+      optionsBuilder.setHost(host);
+    }
+    if (project != null) {
+      optionsBuilder.setProjectId(project);
+    }
+    if (numChannels != null) {
+      optionsBuilder.setNumChannels(Integer.parseInt(numChannels));
+    }
+    spanner = optionsBuilder.build().getService();
+    Runtime.getRuntime().addShutdownHook(new Thread("spannerShutdown") {
+        @Override
+        public void run() {
+          spanner.closeAsync();
+        }
+      });
+    return spanner;
+  }
+
+  @Override
+  public void init() throws DBException {
+    synchronized (CLASS_LOCK) {
+      if (dbClient != null) {
+        return;
+      }
+      Properties properties = getProperties();
+      String host = properties.getProperty(CloudSpannerProperties.HOST);
+      String project = properties.getProperty(CloudSpannerProperties.PROJECT);
+      String instance = properties.getProperty(CloudSpannerProperties.INSTANCE, "ycsb-instance");
+      String database = properties.getProperty(CloudSpannerProperties.DATABASE, "ycsb-database");
+
+      fieldCount = Integer.parseInt(properties.getProperty(
+          CoreWorkload.FIELD_COUNT_PROPERTY, CoreWorkload.FIELD_COUNT_PROPERTY_DEFAULT));
+      queriesForReads = properties.getProperty(CloudSpannerProperties.READ_MODE, "query").equals("query");
+      batchInserts = Integer.parseInt(properties.getProperty(CloudSpannerProperties.BATCH_INSERTS, "1"));
+      constructStandardQueriesAndFields(properties);
+
+      int boundedStalenessSeconds = Integer.parseInt(properties.getProperty(
+          CloudSpannerProperties.BOUNDED_STALENESS, "0"));
+      timestampBound = (boundedStalenessSeconds <= 0) ?
+          TimestampBound.strong() : TimestampBound.ofMaxStaleness(boundedStalenessSeconds, TimeUnit.SECONDS);
+
+      try {
+        spanner = getSpanner(properties, host, project);
+        if (project == null) {
+          project = spanner.getOptions().getProjectId();
+        }
+        dbClient = spanner.getDatabaseClient(DatabaseId.of(project, instance, database));
+      } catch (Exception e) {
+        LOGGER.log(Level.SEVERE, "init()", e);
+        throw new DBException(e);
+      }
+
+      LOGGER.log(Level.INFO, new StringBuilder()
+          .append("\nHost: ").append(spanner.getOptions().getHost())
+          .append("\nProject: ").append(project)
+          .append("\nInstance: ").append(instance)
+          .append("\nDatabase: ").append(database)
+          .append("\nUsing queries for reads: ").append(queriesForReads)
+          .append("\nBatching inserts: ").append(batchInserts)
+          .append("\nBounded staleness seconds: ").append(boundedStalenessSeconds)
+          .toString());
+    }
+  }
+
+  private Status readUsingQuery(
+      String table, String key, Set<String> fields, HashMap<String, ByteIterator> result) {
+    Statement query;
+    Iterable<String> columns = fields == null ? STANDARD_FIELDS : fields;
+    if (fields == null || fields.size() == fieldCount) {
+      query = Statement.newBuilder(standardQuery).bind("key").to(key).build();
+    } else {
+      Joiner joiner = Joiner.on(',');
+      query = Statement.newBuilder("SELECT ")
+          .append(joiner.join(fields))
+          .append(" FROM ")
+          .append(table)
+          .append(" WHERE id=@key")
+          .bind("key").to(key)
+          .build();
+    }
+    try (ResultSet resultSet = dbClient.singleUse(timestampBound).executeQuery(query)) {
+      resultSet.next();
+      decodeStruct(columns, resultSet, result);
+      if (resultSet.next()) {
+        throw new Exception("Expected exactly one row for each read.");
+      }
+
+      return Status.OK;
+    } catch (Exception e) {
+      LOGGER.log(Level.INFO, "readUsingQuery()", e);
+      return Status.ERROR;
+    }
+  }
+
+  @Override
+  public Status read(
+      String table, String key, Set<String> fields, HashMap<String, ByteIterator> result) {
+    if (queriesForReads) {
+      return readUsingQuery(table, key, fields, result);
+    }
+    Iterable<String> columns = fields == null ? STANDARD_FIELDS : fields;
+    try {
+      Struct row = dbClient.singleUse(timestampBound).readRow(table, Key.of(key), columns);
+      decodeStruct(columns, row, result);
+      return Status.OK;
+    } catch (Exception e) {
+      LOGGER.log(Level.INFO, "read()", e);
+      return Status.ERROR;
+    }
+  }
+
+  private Status scanUsingQuery(
+      String table, String startKey, int recordCount, Set<String> fields,
+      Vector<HashMap<String, ByteIterator>> result) {
+    Iterable<String> columns = fields == null ? STANDARD_FIELDS : fields;
+    Statement query;
+    if (fields == null || fields.size() == fieldCount) {
+      query = Statement.newBuilder(standardScan).bind("startKey").to(startKey).bind("count").to(recordCount).build();
+    } else {
+      Joiner joiner = Joiner.on(',');
+      query = Statement.newBuilder("SELECT ")
+          .append(joiner.join(fields))
+          .append(" FROM ")
+          .append(table)
+          .append(" WHERE id>=@startKey LIMIT @count")
+          .bind("startKey").to(startKey)
+          .bind("count").to(recordCount)
+          .build();
+    }
+    try (ResultSet resultSet = dbClient.singleUse(timestampBound).executeQuery(query)) {
+      while (resultSet.next()) {
+        HashMap<String, ByteIterator> row = new HashMap<>();
+        decodeStruct(columns, resultSet, row);
+        result.add(row);
+      }
+      return Status.OK;
+    } catch (Exception e) {
+      LOGGER.log(Level.INFO, "scanUsingQuery()", e);
+      return Status.ERROR;
+    }
+  }
+
+  @Override
+  public Status scan(
+      String table, String startKey, int recordCount, Set<String> fields,
+      Vector<HashMap<String, ByteIterator>> result) {
+    if (queriesForReads) {
+      return scanUsingQuery(table, startKey, recordCount, fields, result);
+    }
+    Iterable<String> columns = fields == null ? STANDARD_FIELDS : fields;
+    KeySet keySet =
+        KeySet.newBuilder().addRange(KeyRange.closedClosed(Key.of(startKey), Key.of())).build();
+    try (ResultSet resultSet = dbClient.singleUse(timestampBound)
+                                       .read(table, keySet, columns, Options.limit(recordCount))) {
+      while (resultSet.next()) {
+        HashMap<String, ByteIterator> row = new HashMap<>();
+        decodeStruct(columns, resultSet, row);
+        result.add(row);
+      }
+      return Status.OK;
+    } catch (Exception e) {
+      LOGGER.log(Level.INFO, "scan()", e);
+      return Status.ERROR;
+    }
+  }
+
+  @Override
+  public Status update(String table, String key, HashMap<String, ByteIterator> values) {
+    Mutation.WriteBuilder m = Mutation.newInsertOrUpdateBuilder(table);
+    m.set(PRIMARY_KEY_COLUMN).to(key);
+    for (Map.Entry<String, ByteIterator> e : values.entrySet()) {
+      m.set(e.getKey()).to(e.getValue().toString());
+    }
+    try {
+      dbClient.writeAtLeastOnce(Arrays.asList(m.build()));
+    } catch (Exception e) {
+      LOGGER.log(Level.INFO, "update()", e);
+      return Status.ERROR;
+    }
+    return Status.OK;
+  }
+
+  @Override
+  public Status insert(String table, String key, HashMap<String, ByteIterator> values) {
+    if (bufferedMutations.size() < batchInserts) {
+      Mutation.WriteBuilder m = Mutation.newInsertOrUpdateBuilder(table);
+      m.set(PRIMARY_KEY_COLUMN).to(key);
+      for (Map.Entry<String, ByteIterator> e : values.entrySet()) {
+        m.set(e.getKey()).to(e.getValue().toString());
+      }
+      bufferedMutations.add(m.build());
+    } else {
+      LOGGER.log(Level.INFO, "Limit of cached mutations reached. The given mutation with key " + key +
+          " is ignored. Is this a retry?");
+    }
+    if (bufferedMutations.size() < batchInserts) {
+      return Status.BATCHED_OK;
+    }
+    try {
+      dbClient.writeAtLeastOnce(bufferedMutations);
+      bufferedMutations.clear();
+    } catch (Exception e) {
+      LOGGER.log(Level.INFO, "insert()", e);
+      return Status.ERROR;
+    }
+    return Status.OK;
+  }
+
+  @Override
+  public void cleanup() {
+    try {
+      if (bufferedMutations.size() > 0) {
+        dbClient.writeAtLeastOnce(bufferedMutations);
+        bufferedMutations.clear();
+      }
+    } catch (Exception e) {
+      LOGGER.log(Level.INFO, "cleanup()", e);
+    }
+  }
+
+  @Override
+  public Status delete(String table, String key) {
+    try {
+      dbClient.writeAtLeastOnce(Arrays.asList(Mutation.delete(table, Key.of(key))));
+    } catch (Exception e) {
+      LOGGER.log(Level.INFO, "delete()", e);
+      return Status.ERROR;
+    }
+    return Status.OK;
+  }
+
+  private static void decodeStruct(
+      Iterable<String> columns, StructReader structReader, HashMap<String, ByteIterator> result) {
+    for (String col : columns) {
+      result.put(col, new StringByteIterator(structReader.getString(col)));
+    }
+  }
+}

cloudspanner/src/main/java/com/yahoo/ycsb/db/cloudspanner/package-info.java

+/*
+ * Copyright (c) 2017 YCSB contributors. All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you
+ * may not use this file except in compliance with the License. You
+ * may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+ * implied. See the License for the specific language governing
+ * permissions and limitations under the License. See accompanying
+ * LICENSE file.
+ */
+
+/**
+ * The YCSB binding for Google's <a href="https://cloud.google.com/spanner/">
+ * Cloud Spanner</a>.
+ */
+package com.yahoo.ycsb.db.cloudspanner;

distribution/pom.xml

@@ -69,6 +69,11 @@ LICENSE file.
       <artifactId>cassandra-binding</artifactId>
       <version>${project.version}</version>
     </dependency>
+    <dependency>
+      <groupId>com.yahoo.ycsb</groupId>
+      <artifactId>cloudspanner-binding</artifactId>
+      <version>${project.version}</version>
+    </dependency>
     <dependency>
       <groupId>com.yahoo.ycsb</groupId>
       <artifactId>couchbase-binding</artifactId>

pom.xml

@@ -100,6 +100,7 @@ LICENSE file.
     <arangodb.version>2.7.3</arangodb.version>
     <arangodb3.version>4.1.7</arangodb3.version>
     <azurestorage.version>4.0.0</azurestorage.version>
+    <cloudspanner.version>0.9.3-beta</cloudspanner.version>
   </properties>
 
   <modules>
@@ -114,6 +115,7 @@ LICENSE file.
     <module>asynchbase</module>
     <module>azuretablestorage</module>
     <module>cassandra</module>
+    <module>cloudspanner</module>
     <module>couchbase</module>
     <module>couchbase2</module>
     <module>distribution</module>