Merge pull request brianfrankcooper#7 from smkniazi/RONDB-363

RONDB-363 & RONDB-336
smkniazi · May 22, 2023 · aea0333 · aea0333
2 parents 82e1473 + be4da6b
commit aea0333
Show file tree

Hide file tree

Showing 72 changed files with 863 additions and 103 deletions.
diff --git a/accumulo1.9/src/main/java/site/ycsb/db/accumulo/AccumuloClient.java b/accumulo1.9/src/main/java/site/ycsb/db/accumulo/AccumuloClient.java
@@ -223,6 +223,12 @@ public Status batchRead(String table, List<String> keys, List<Set<String>> field
     throw  new UnsupportedOperationException("Batch reads are not yet supported");
   }
 
+  @Override
+  public Status batchUpdate(String table, List<String> keys,
+                            List<Map<String, ByteIterator>>  values) {
+    throw  new UnsupportedOperationException("Batch updates are not yet supported");
+  }
+
   @Override
   public Status scan(String table, String startkey, int recordcount,
       Set<String> fields, Vector<HashMap<String, ByteIterator>> result) {

diff --git a/aerospike/src/main/java/site/ycsb/db/AerospikeClient.java b/aerospike/src/main/java/site/ycsb/db/AerospikeClient.java
@@ -129,6 +129,12 @@ public Status batchRead(String table, List<String> keys, List<Set<String>> field
     throw  new UnsupportedOperationException("Batch reads are not yet supported");
   }
 
+  @Override
+  public Status batchUpdate(String table, List<String> keys,
+                            List<Map<String, ByteIterator>>  values) {
+    throw  new UnsupportedOperationException("Batch updates are not yet supported");
+  }
+
   @Override
   public Status scan(String table, String start, int count, Set<String> fields,
       Vector<HashMap<String, ByteIterator>> result) {

diff --git a/arangodb/src/main/java/site/ycsb/db/arangodb/ArangoDBClient.java b/arangodb/src/main/java/site/ycsb/db/arangodb/ArangoDBClient.java
@@ -225,6 +225,12 @@ public Status batchRead(String table, List<String> keys, List<Set<String>> field
     throw  new UnsupportedOperationException("Batch reads are not yet supported");
   }
 
+  @Override
+  public Status batchUpdate(String table, List<String> keys,
+                            List<Map<String, ByteIterator>>  values) {
+    throw  new UnsupportedOperationException("Batch updates are not yet supported");
+  }
+
   /**
    * Update a record in the database. Any field/value pairs in the specified
    * values HashMap will be written into the record with the specified record

diff --git a/asynchbase/src/main/java/site/ycsb/db/AsyncHBaseClient.java b/asynchbase/src/main/java/site/ycsb/db/AsyncHBaseClient.java
@@ -246,6 +246,12 @@ public Status batchRead(String table, List<String> keys, List<Set<String>> field
     throw  new UnsupportedOperationException("Batch reads are not yet supported");
   }
 
+  @Override
+  public Status batchUpdate(String table, List<String> keys,
+                            List<Map<String, ByteIterator>>  values) {
+    throw  new UnsupportedOperationException("Batch updates are not yet supported");
+  }
+
   @Override
   public Status scan(String table, String startkey, int recordcount,
       Set<String> fields, Vector<HashMap<String, ByteIterator>> result) {

diff --git a/azurecosmos/src/main/java/site/ycsb/db/AzureCosmosClient.java b/azurecosmos/src/main/java/site/ycsb/db/AzureCosmosClient.java
@@ -319,6 +319,12 @@ public Status batchRead(String table, List<String> keys, List<Set<String>> field
     throw  new UnsupportedOperationException("Batch reads are not yet supported");
   }
 
+  @Override
+  public Status batchUpdate(String table, List<String> keys,
+                            List<Map<String, ByteIterator>>  values) {
+    throw  new UnsupportedOperationException("Batch updates are not yet supported");
+  }
+
   /**
    * Perform a range scan for a set of records in the database. Each field/value
    * pair from the result will be stored in a HashMap.

diff --git a/azuretablestorage/src/main/java/site/ycsb/db/azuretablestorage/AzureClient.java b/azuretablestorage/src/main/java/site/ycsb/db/azuretablestorage/AzureClient.java
@@ -116,6 +116,12 @@ public Status batchRead(String table, List<String> keys, List<Set<String>> field
     throw  new UnsupportedOperationException("Batch reads are not yet supported");
   }
 
+  @Override
+  public Status batchUpdate(String table, List<String> keys,
+                            List<Map<String, ByteIterator>>  values) {
+    throw  new UnsupportedOperationException("Batch updates are not yet supported");
+  }
+
   @Override
   public Status scan(String table, String startkey, int recordcount, 
       Set<String> fields, Vector<HashMap<String, ByteIterator>> result) {

diff --git a/cassandra/src/main/java/site/ycsb/db/CassandraCQLClient.java b/cassandra/src/main/java/site/ycsb/db/CassandraCQLClient.java
@@ -349,6 +349,12 @@ public Status batchRead(String table, List<String> keys, List<Set<String>> field
     throw  new UnsupportedOperationException("Batch reads are not yet supported");
   }
 
+  @Override
+  public Status batchUpdate(String table, List<String> keys,
+                            List<Map<String, ByteIterator>>  values) {
+    throw  new UnsupportedOperationException("Batch updates are not yet supported");
+  }
+
   /**
    * Perform a range scan for a set of records in the database. Each field/value
    * pair from the result will be stored in a HashMap.

diff --git a/cloudspanner/src/main/java/site/ycsb/db/cloudspanner/CloudSpannerClient.java b/cloudspanner/src/main/java/site/ycsb/db/cloudspanner/CloudSpannerClient.java
@@ -271,6 +271,12 @@ public Status batchRead(String table, List<String> keys, List<Set<String>> field
     throw  new UnsupportedOperationException("Batch reads are not yet supported");
   }
 
+  @Override
+  public Status batchUpdate(String table, List<String> keys,
+                            List<Map<String, ByteIterator>>  values) {
+    throw  new UnsupportedOperationException("Batch updates are not yet supported");
+  }
+
   private Status scanUsingQuery(
       String table, String startKey, int recordCount, Set<String> fields,
       Vector<HashMap<String, ByteIterator>> result) {

diff --git a/core/src/main/java/site/ycsb/BasicDB.java b/core/src/main/java/site/ycsb/BasicDB.java
@@ -166,6 +166,13 @@ public Status batchRead(String table, List<String> keys, List<Set<String>> field
     throw  new UnsupportedOperationException("Batch reads are not yet supported");
   }
 
+
+  @Override
+  public Status batchUpdate(String table, List<String> keys,
+                            List<Map<String, ByteIterator>>  values) {
+    throw  new UnsupportedOperationException("Batch updates are not yet supported");
+  }
+
   /**
    * Perform a range scan for a set of records in the database. Each field/value pair from the result will be stored
    * in a HashMap.

diff --git a/core/src/main/java/site/ycsb/DB.java b/core/src/main/java/site/ycsb/DB.java
@@ -99,6 +99,16 @@ public void cleanup() throws DBException {
   public abstract Status batchRead(String table, List<String> keys, List<Set<String>> fields,
                                    HashMap<String /*key*/, HashMap<String/*field*/, ByteIterator>> result);
 
+  /**
+   * Batch update records.
+   *
+   * @param table The name of the table
+   * @param keys record keys
+   * @param values A list of HashMaps of field/value pairs to update in the record
+   * @return The result of the operation.
+   */
+  public abstract Status batchUpdate(String table, List<String> keys, List<Map<String, ByteIterator>>  values);
+
   /**
    * Perform a range scan for a set of records in the database. Each field/value pair from the result will be stored
    * in a HashMap.

diff --git a/core/src/main/java/site/ycsb/DBWrapper.java b/core/src/main/java/site/ycsb/DBWrapper.java
@@ -161,6 +161,20 @@ public Status batchRead(String table, List<String> keys, List<Set<String>> field
     }
   }
 
+  @Override
+  public Status batchUpdate(String table, List<String> keys,
+                            List<Map<String, ByteIterator>>  values) {
+    try (final TraceScope span = tracer.newScope(scopeStringRead)) {
+      long ist = measurements.getIntendedStartTimeNs();
+      long st = System.nanoTime();
+      Status res = db.batchUpdate(table, keys, values);
+      long en = System.nanoTime();
+      measure("BATCH_UPDATE", res, ist, st, en);
+      measurements.reportStatus("BATCH_UPDATE", res);
+      return res;
+    }
+  }
+
   /**
    * Perform a range scan for a set of records in the database.
    * Each field/value pair from the result will be stored in a HashMap.

diff --git a/core/src/main/java/site/ycsb/GoodBadUglyDB.java b/core/src/main/java/site/ycsb/GoodBadUglyDB.java
@@ -103,6 +103,12 @@ public Status batchRead(String table, List<String> keys, List<Set<String>> field
     throw  new UnsupportedOperationException("Batch reads are not yet supported");
   }
 
+  @Override
+  public Status batchUpdate(String table, List<String> keys,
+                            List<Map<String, ByteIterator>>  values) {
+    throw  new UnsupportedOperationException("Batch updates are not yet supported");
+  }
+
   /**
    * Perform a range scan for a set of records in the database. Each field/value pair from the result will be stored
    * in a HashMap.

diff --git a/core/src/main/java/site/ycsb/measurements/OneMeasurementHdrHistogram.java b/core/src/main/java/site/ycsb/measurements/OneMeasurementHdrHistogram.java
@@ -71,13 +71,16 @@ public class OneMeasurementHdrHistogram extends OneMeasurement {
   private final List<Double> percentiles;
 
   private final int readBatchSize;
+  private final int updateBatchSize;
 
   public OneMeasurementHdrHistogram(String name, Properties props) {
     super(name);
     percentiles = getPercentileValues(props.getProperty(PERCENTILES_PROPERTY, PERCENTILES_PROPERTY_DEFAULT));
     verbose = Boolean.valueOf(props.getProperty(VERBOSE_PROPERTY, String.valueOf(false)));
     readBatchSize =Integer.valueOf(props.getProperty(CoreWorkload.READ_BATCH_SIZE_PROPERTY,
           CoreWorkload.READ_BATCH_SIZE_PROPERTY_DEFAULT));
+    updateBatchSize =Integer.valueOf(props.getProperty(CoreWorkload.UPDATE_BATCH_SIZE_PROPERTY,
+        CoreWorkload.UPDATE_BATCH_SIZE_PROPERTY_DEFAULT));
     boolean shouldLog = Boolean.parseBoolean(props.getProperty("hdrhistogram.fileoutput", "false"));
     if (!shouldLog) {
       log = null;
@@ -127,6 +130,11 @@ public void exportMeasurements(MeasurementsExporter exporter) throws IOException
       prepend = "Batch";
     }
 
+    if (getName().compareTo("BATCH_UPDATE") == 0) {
+      exporter.write(getName(), "BatchSize", updateBatchSize);
+      prepend = "Batch";
+    }
+
     exporter.write(getName(), prepend+"Operations", totalHistogram.getTotalCount());
     exporter.write(getName(), prepend+"AverageLatency(us)", totalHistogram.getMean());
     exporter.write(getName(), prepend+"MinLatency(us)", totalHistogram.getMinValue());

diff --git a/core/src/main/java/site/ycsb/measurements/OneMeasurementHistogram.java b/core/src/main/java/site/ycsb/measurements/OneMeasurementHistogram.java
@@ -80,13 +80,16 @@ public class OneMeasurementHistogram extends OneMeasurement {
   private int max;
 
   private final int readBatchSize;
+  private final int updateBatchSize;
 
   public OneMeasurementHistogram(String name, Properties props) {
     super(name);
     buckets = Integer.parseInt(props.getProperty(BUCKETS, BUCKETS_DEFAULT));
     verbose = Boolean.valueOf(props.getProperty(VERBOSE_PROPERTY, String.valueOf(false)));
     readBatchSize = Integer.valueOf(props.getProperty(CoreWorkload.READ_BATCH_SIZE_PROPERTY,
           CoreWorkload.READ_BATCH_SIZE_PROPERTY_DEFAULT));
+    updateBatchSize = Integer.valueOf(props.getProperty(CoreWorkload.UPDATE_BATCH_SIZE_PROPERTY,
+        CoreWorkload.UPDATE_BATCH_SIZE_PROPERTY_DEFAULT));
     histogram = new long[buckets];
     histogramoverflow = 0;
     operations = 0;
@@ -131,6 +134,9 @@ public void exportMeasurements(MeasurementsExporter exporter) throws IOException
     if (getName().compareTo("BATCH_READ") == 0) {
       exporter.write(getName(), "BatchSize", readBatchSize);
     }
+    if (getName().compareTo("BATCH_UPDATE") == 0) {
+      exporter.write(getName(), "BatchSize", updateBatchSize);
+    }
     exporter.write(getName(), "Operations", operations);
     exporter.write(getName(), "AverageLatency(us)", mean);
     exporter.write(getName(), "LatencyVariance(us)", variance);

diff --git a/core/src/main/java/site/ycsb/measurements/OneMeasurementTimeSeries.java b/core/src/main/java/site/ycsb/measurements/OneMeasurementTimeSeries.java
@@ -68,13 +68,16 @@ public class OneMeasurementTimeSeries extends OneMeasurement {
   private int min = -1;
   private int max = -1;
   private final int readBatchSize;
+  private final int updateBatchSize;
 
   public OneMeasurementTimeSeries(String name, Properties props) {
     super(name);
     granularity = Integer.parseInt(props.getProperty(GRANULARITY, GRANULARITY_DEFAULT));
     measurements = new Vector<>();
     readBatchSize = Integer.valueOf(props.getProperty(CoreWorkload.READ_BATCH_SIZE_PROPERTY,
           CoreWorkload.READ_BATCH_SIZE_PROPERTY_DEFAULT));
+    updateBatchSize = Integer.valueOf(props.getProperty(CoreWorkload.UPDATE_BATCH_SIZE_PROPERTY,
+        CoreWorkload.UPDATE_BATCH_SIZE_PROPERTY_DEFAULT));
   }
 
   private synchronized void checkEndOfUnit(boolean forceend) {
@@ -128,6 +131,11 @@ public void exportMeasurements(MeasurementsExporter exporter) throws IOException
       exporter.write(getName(), "BatchSize", readBatchSize);
       prepend = "Batch";
     }
+
+    if (getName().compareTo("BATCH_UPDATE") == 0) {
+      exporter.write(getName(), "BatchSize", updateBatchSize);
+      prepend = "Batch";
+    }
     exporter.write(getName(), prepend+"Operations", operations);
     exporter.write(getName(), prepend+"AverageLatency(us)", (((double) totallatency) / ((double) operations)));
     exporter.write(getName(), prepend+"MinLatency(us)", min);

diff --git a/core/src/main/java/site/ycsb/workloads/CoreWorkload.java b/core/src/main/java/site/ycsb/workloads/CoreWorkload.java
@@ -221,6 +221,16 @@ public class CoreWorkload extends Workload {
    */
   public static final String READ_BATCH_SIZE_PROPERTY_DEFAULT = "1";
 
+  /**
+   * The name of the property for the size of update batches.
+   */
+  public static final String UPDATE_BATCH_SIZE_PROPERTY = "updateBatchSize";
+
+  /**
+   * The default update batch.
+   */
+  public static final String UPDATE_BATCH_SIZE_PROPERTY_DEFAULT = "1";
+
   /**
    * The name of the property for the proportion of transactions that are updates.
    */
@@ -380,6 +390,7 @@ public class CoreWorkload extends Workload {
   protected int insertionRetryLimit;
   protected int insertionRetryInterval;
   protected int readBatchSize;
+  protected int updateBatchSize;
 
   private Measurements measurements = Measurements.getMeasurements();
 
@@ -562,6 +573,11 @@ public void init(Properties p) throws WorkloadException {
     if (readBatchSize <= 0) {
       throw new WorkloadException("Invalid read batch size \"" + readBatchSize + "\"");
     }
+    updateBatchSize = Integer.parseInt(p.getProperty(
+        UPDATE_BATCH_SIZE_PROPERTY, UPDATE_BATCH_SIZE_PROPERTY_DEFAULT));
+    if (updateBatchSize <= 0) {
+      throw new WorkloadException("Invalid update batch size \"" + updateBatchSize + "\"");
+    }
   }
 
   /**
@@ -685,7 +701,11 @@ public boolean doTransaction(DB db, Object threadstate) {
       }
       break;
     case "UPDATE":
-      doTransactionUpdate(db);
+      if (updateBatchSize == 1){
+        doTransactionUpdate(db);
+      } else {
+        doTransactionBatchUpdate(db);
+      }
       break;
     case "INSERT":
       doTransactionInsert(db);
@@ -893,6 +913,33 @@ public void doTransactionUpdate(DB db) {
     db.update(table, keyname, values);
   }
 
+  public void doTransactionBatchUpdate(DB db) {
+    LinkedList<String> keys = new LinkedList<>();
+    LinkedList<Map<String, ByteIterator>> allValues = new LinkedList<>();
+
+    for (int i = 0; i < updateBatchSize; i++) {
+      // choose a random key
+      long keynum = nextKeynum();
+
+      String keyname = CoreWorkload.buildKeyName(keynum, zeropadding, orderedinserts);
+
+      HashMap<String, ByteIterator> values;
+
+      if (writeallfields) {
+        // new data for all the fields
+        values = buildValues(keyname);
+      } else {
+        // update a random field
+        values = buildSingleValue(keyname);
+      }
+
+      keys.add(keyname);
+      allValues.add(values);
+    }
+
+    db.batchUpdate(table, keys, allValues);
+  }
+
   public void doTransactionInsert(DB db) {
     // choose the next key
     long keynum = transactioninsertkeysequence.nextValue();

diff --git a/core/src/main/java/site/ycsb/workloads/RestWorkload.java b/core/src/main/java/site/ycsb/workloads/RestWorkload.java
@@ -152,6 +152,11 @@ public void init(Properties p) throws WorkloadException {
     if (readBatchSize <= 0) {
       throw new WorkloadException("Invalid read batch size \"" + readBatchSize + "\"");
     }
+    updateBatchSize = Integer.parseInt(p.getProperty(
+        UPDATE_BATCH_SIZE_PROPERTY, UPDATE_BATCH_SIZE_PROPERTY_DEFAULT));
+    if (updateBatchSize <= 0) {
+      throw new WorkloadException("Invalid update batch size \"" + updateBatchSize + "\"");
+    }
   }
 
   public static DiscreteGenerator createOperationGenerator(final Properties p) {

diff --git a/core/src/test/java/site/ycsb/workloads/TestTimeSeriesWorkload.java b/core/src/test/java/site/ycsb/workloads/TestTimeSeriesWorkload.java
@@ -531,6 +531,12 @@ public Status batchRead(String table, List<String> keys, List<Set<String>> field
       return Status.OK;
     }
 
+    @Override
+    public Status batchUpdate(String table, List<String> keys,
+                              List<Map<String, ByteIterator>>  values) {
+      return Status.OK;
+    }
+
     @Override
     public Status scan(String table, String startkey, int recordcount,
         Set<String> fields, Vector<HashMap<String, ByteIterator>> result) {

diff --git a/couchbase/src/main/java/site/ycsb/db/CouchbaseClient.java b/couchbase/src/main/java/site/ycsb/db/CouchbaseClient.java
@@ -201,6 +201,12 @@ public Status batchRead(String table, List<String> keys, List<Set<String>> field
     throw  new UnsupportedOperationException("Batch reads are not yet supported");
   }
 
+  @Override
+  public Status batchUpdate(String table, List<String> keys,
+                            List<Map<String, ByteIterator>>  values) {
+    throw  new UnsupportedOperationException("Batch updates are not yet supported");
+  }
+
   @Override
   public Status scan(final String table, final String startkey, final int recordcount, final Set<String> fields,
                      final Vector<HashMap<String, ByteIterator>> result) {

diff --git a/couchbase2/src/main/java/site/ycsb/db/couchbase2/Couchbase2Client.java b/couchbase2/src/main/java/site/ycsb/db/couchbase2/Couchbase2Client.java
@@ -262,6 +262,12 @@ public Status batchRead(String table, List<String> keys, List<Set<String>> field
     throw  new UnsupportedOperationException("Batch reads are not yet supported");
   }
 
+  @Override
+  public Status batchUpdate(String table, List<String> keys,
+                            List<Map<String, ByteIterator>>  values) {
+    throw  new UnsupportedOperationException("Batch updates are not yet supported");
+  }
+
   /**
    * Performs the {@link #read(String, String, Set, Map)} operation via Key/Value ("get").
    *