Merge branch 'upstream_emulator' into sample_row_keys

Author: marete

google/cloud/bigtable/emulator/column_family.cc

@@ -13,15 +13,108 @@
 // limitations under the License.
 
 #include "google/cloud/bigtable/emulator/column_family.h"
+#include "google/cloud/internal/big_endian.h"
 #include <absl/types/optional.h>
 #include <chrono>
+#include <cstdint>
 #include <map>
 
 namespace google {
 namespace cloud {
 namespace bigtable {
 namespace emulator {
 
+// FIXME: Workaround our current incorrect ordering of
+// timestamps. Remove when that is fixed and they are in decreasing
+// order, at which point we can just pick the first element.
+std::map<std::chrono::milliseconds, std::string>::iterator latest(
+    std::map<std::chrono::milliseconds, std::string>& cells_not_empty) {
+  assert(!cells_not_empty.empty());
+
+  auto first_it = cells_not_empty.begin();
+  auto last_it = std::prev(cells_not_empty.end());
+  auto latest_it = first_it->first >= last_it->first ? first_it : last_it;
+
+  return latest_it;
+}
+
+StatusOr<ReadModifyWriteCellResult> ColumnRow::ReadModifyWrite(
+    std::int64_t inc_value) {
+  auto system_ms = std::chrono::duration_cast<std::chrono::milliseconds>(
+      std::chrono::system_clock::now().time_since_epoch());
+
+  if (cells_.empty()) {
+    std::string value = google::cloud::internal::EncodeBigEndian(inc_value);
+    cells_[system_ms] = value;
+
+    return ReadModifyWriteCellResult{system_ms, std::move(value),
+                                     absl::nullopt};
+  }
+
+  // FIXME: Workaround our current incorrect ordering of
+  // timestamps. Remove when that is fixed and they are in decreasing
+  // order, at which point we can just pick the first element.
+  auto latest_it = latest(cells_);
+
+  auto maybe_old_value =
+      google::cloud::internal::DecodeBigEndian<std::int64_t>(latest_it->second);
+  if (!maybe_old_value) {
+    return maybe_old_value.status();
+  }
+
+  auto value = google::cloud::internal::EncodeBigEndian(
+      inc_value + maybe_old_value.value());
+
+  if (latest_it->first < system_ms) {
+    // We need to add a cell with the current system timestamp
+    cells_[system_ms] = value;
+
+    return ReadModifyWriteCellResult{system_ms, std::move(value),
+                                     absl::nullopt};
+  }
+
+  // Latest timestamp is >= system time. Overwrite latest timestamp
+  auto old_value = std::move(latest_it->second);
+  latest_it->second = value;
+
+  return ReadModifyWriteCellResult{latest_it->first, std::move(value),
+                                   std::move(old_value)};
+}
+
+ReadModifyWriteCellResult ColumnRow::ReadModifyWrite(
+    std::string const& append_value) {
+  auto system_ms = std::chrono::duration_cast<std::chrono::milliseconds>(
+      std::chrono::system_clock::now().time_since_epoch());
+  if (cells_.empty()) {
+    cells_[system_ms] = append_value;
+
+    return ReadModifyWriteCellResult{system_ms, std::move(append_value),
+                                     absl::nullopt};
+  }
+
+  // FIXME: Workaround our current incorrect ordering of
+  // timestamps. Remove when that is fixed and they are in decreasing
+  // order, at which point we can just pick the first element.
+  auto latest_it = latest(cells_);
+
+  auto value = latest_it->second + append_value;
+
+  if (latest_it->first < system_ms) {
+    // We need to add a cell with the current system timestamp
+    cells_[system_ms] = value;
+
+    return ReadModifyWriteCellResult{system_ms, std::move(value),
+                                     absl::nullopt};
+  }
+
+  // Latest timestamp is >= system time. Overwrite latest timestamp
+  auto old_value = std::move(latest_it->second);
+  latest_it->second = value;
+
+  return ReadModifyWriteCellResult{latest_it->first, value,
+                                   std::move(old_value)};
+}
+
 absl::optional<std::string> ColumnRow::SetCell(
     std::chrono::milliseconds timestamp, std::string const& value) {
   absl::optional<std::string> ret = absl::nullopt;

google/cloud/bigtable/emulator/column_family.h

@@ -19,12 +19,18 @@
 #include "google/cloud/bigtable/emulator/filter.h"
 #include "google/cloud/bigtable/emulator/filtered_map.h"
 #include "google/cloud/bigtable/emulator/range_set.h"
+#include "google/cloud/bigtable/read_modify_write_rule.h"
+#include "google/cloud/internal/big_endian.h"
+#include "google/cloud/internal/make_status.h"
 #include "absl/types/optional.h"
 #include <google/bigtable/admin/v2/table.pb.h>
 #include <google/bigtable/v2/data.pb.h>
 #include <chrono>
 #include <cstddef>
 #include <map>
+#include <memory>
+#include <optional>
+#include <sstream>
 
 namespace google {
 namespace cloud {
@@ -36,6 +42,24 @@ struct Cell {
   std::string value;
 };
 
+// ReadModifyWriteCellResult supports undo and return value
+// construction for the ReadModifyWrite RPC.
+//
+// The timestamp and value written are always returned in timestamp
+// and value and will be used to construct the Row returned by the
+// RPC.
+//
+// If maybe_old_value has a value, then a timestamp was overwritten
+// and the ReadModifyWriteCellResult will be used to create a
+// RestoreValue for undo log. Otherwise, a new cell was added and the
+// ReadmodifyWriteCellResult will be used to create a DeleteValue for
+// the undo log.
+struct ReadModifyWriteCellResult {
+  std::chrono::milliseconds timestamp;
+  std::string value;
+  absl::optional<std::string> maybe_old_value;
+};
+
 /**
  * Objects of this class hold contents of a specific column in a specific row.
  *
@@ -48,6 +72,10 @@ class ColumnRow {
   ColumnRow(ColumnRow const&) = delete;
   ColumnRow& operator=(ColumnRow const&) = delete;
 
+  StatusOr<ReadModifyWriteCellResult> ReadModifyWrite(std::int64_t inc_value);
+
+  ReadModifyWriteCellResult ReadModifyWrite(std::string const& append_value);
+
   /**
    * Insert or update and existing cell at a given timestamp.
    *
@@ -84,8 +112,11 @@ class ColumnRow {
   bool HasCells() const { return !cells_.empty(); }
   using const_iterator =
       std::map<std::chrono::milliseconds, std::string>::const_iterator;
+  using iterator = std::map<std::chrono::milliseconds, std::string>::iterator;
   const_iterator begin() const { return cells_.begin(); }
   const_iterator end() const { return cells_.end(); }
+  iterator begin() { return cells_.begin(); }
+  iterator end() { return cells_.end(); }
   const_iterator lower_bound(std::chrono::milliseconds timestamp) const {
     return cells_.lower_bound(timestamp);
   }
@@ -117,6 +148,16 @@ class ColumnRow {
  */
 class ColumnFamilyRow {
  public:
+  StatusOr<ReadModifyWriteCellResult> ReadModifyWrite(
+      std::string const& column_qualifier, std::int64_t inc_value) {
+    return columns_[column_qualifier].ReadModifyWrite(inc_value);
+  };
+
+  ReadModifyWriteCellResult ReadModifyWrite(std::string const& column_qualifier,
+                                            std::string const& append_value) {
+    return columns_[column_qualifier].ReadModifyWrite(append_value);
+  }
+
   /**
    * Insert or update and existing cell at a given column and timestamp.
    *
@@ -204,6 +245,18 @@ class ColumnFamily {
   using const_iterator = std::map<std::string, ColumnFamilyRow>::const_iterator;
   using iterator = std::map<std::string, ColumnFamilyRow>::iterator;
 
+  StatusOr<ReadModifyWriteCellResult> ReadModifyWrite(
+      std::string const& row_key, std::string const& column_qualifier,
+      std::int64_t inc_value) {
+    return rows_[row_key].ReadModifyWrite(column_qualifier, inc_value);
+  };
+
+  ReadModifyWriteCellResult ReadModifyWrite(std::string const& row_key,
+                                            std::string const& column_qualifier,
+                                            std::string const& append_value) {
+    return rows_[row_key].ReadModifyWrite(column_qualifier, append_value);
+  };
+
   /**
    * Insert or update and existing cell at a given row, column and timestamp.
    *

google/cloud/bigtable/emulator/rollback_test.cc

@@ -273,7 +273,7 @@ Status HasRow(std::shared_ptr<google::cloud::bigtable::emulator::Table>& table,
 //
 // Supplied with a timestamp < -1, it should return an error and fail the entire
 // mutation chain.
-TEST(TransactonRollback, ZeroOrNegativeTimestampHandling) {
+TEST(TransactionRollback, ZeroOrNegativeTimestampHandling) {
   ::google::bigtable::admin::v2::Table schema;
   ::google::bigtable::admin::v2::ColumnFamily column_family;
 
@@ -346,12 +346,13 @@ TEST(TransactonRollback, ZeroOrNegativeTimestampHandling) {
       {column_family_name, column_qualifier, -1, data},
   };
   auto const* const row_key_5 = "4";
-  auto system_time_ms_before = std::chrono::duration_cast<std::chrono::milliseconds>(
-      std::chrono::system_clock::now().time_since_epoch());
+  auto system_time_ms_before =
+      std::chrono::duration_cast<std::chrono::milliseconds>(
+          std::chrono::system_clock::now().time_since_epoch());
   status = SetCells(table, table_name, row_key_5, v);
   ASSERT_STATUS_OK(status);
-  auto column_or = GetColumn(
-      table, v[0].column_family_name, row_key_5, v[0].column_qualifier);
+  auto column_or = GetColumn(table, v[0].column_family_name, row_key_5,
+                             v[0].column_qualifier);
   ASSERT_STATUS_OK(column_or.status());
   auto col = column_or.value();
   ASSERT_EQ(col.size(), 1);
@@ -362,7 +363,7 @@ TEST(TransactonRollback, ZeroOrNegativeTimestampHandling) {
 }
 
 // Does the SetCell mutation work to set a cell to a specific value?
-TEST(TransactonRollback, SetCellBasicFunction) {
+TEST(TransactionRollback, SetCellBasicFunction) {
   ::google::bigtable::admin::v2::Table schema;
   ::google::bigtable::admin::v2::ColumnFamily column_family;
 
@@ -395,7 +396,7 @@ TEST(TransactonRollback, SetCellBasicFunction) {
 // Test that an old value is correctly restored in a pre-populated
 // cell, when one of a set of SetCell mutations fails after the cell
 // had been updated with a new value.
-TEST(TransactonRollback, TestRestoreValue) {
+TEST(TransactionRollback, TestRestoreValue) {
   ::google::bigtable::admin::v2::Table schema;
   ::google::bigtable::admin::v2::ColumnFamily column_family;
 
@@ -458,7 +459,7 @@ TEST(TransactonRollback, TestRestoreValue) {
 
 // Test that a new cell introduced in a chain of SetCell mutations is
 // deleted on rollback if a subsequent mutation fails.
-TEST(TransactonRollback, DeleteValue) {
+TEST(TransactionRollback, DeleteValue) {
   ::google::bigtable::admin::v2::Table schema;
   ::google::bigtable::admin::v2::ColumnFamily column_family;
 
@@ -513,7 +514,7 @@ TEST(TransactonRollback, DeleteValue) {
 // column family name that is not in the table schema) then the column
 // and any of the cells introduced is deleted in the rollback, but
 // that any pre-transaction-attemot data in the row is unaffected.
-TEST(TransactonRollback, DeleteColumn) {
+TEST(TransactionRollback, DeleteColumn) {
   ::google::bigtable::admin::v2::Table schema;
   ::google::bigtable::admin::v2::ColumnFamily column_family;
 
@@ -562,7 +563,7 @@ TEST(TransactonRollback, DeleteColumn) {
 // Test that a chain of SetCell mutations that initially introduces a
 // new row, but one of which eventually fails, will end with the whole
 // row rolled back.
-TEST(TransactonRollback, DeleteRow) {
+TEST(TransactionRollback, DeleteRow) {
   ::google::bigtable::admin::v2::Table schema;
   ::google::bigtable::admin::v2::ColumnFamily column_family;
 
@@ -599,7 +600,7 @@ TEST(TransactonRollback, DeleteRow) {
 // Does the DeleteFromfamily mutation work to delete a row from a
 // specific family and does it rows with the same row key in other
 // column families alone?
-TEST(TransactonRollback, DeleteFromFamilyBasicFunction) {
+TEST(TransactionRollback, DeleteFromFamilyBasicFunction) {
   ::google::bigtable::admin::v2::Table schema;
   ::google::bigtable::admin::v2::ColumnFamily column_family;
 
@@ -648,7 +649,7 @@ TEST(TransactonRollback, DeleteFromFamilyBasicFunction) {
 
 // Test that DeleteFromfamily can be rolled back in case a subsequent
 // mutation fails.
-TEST(TransactonRollback, DeleteFromFamilyRollback) {
+TEST(TransactionRollback, DeleteFromFamilyRollback) {
   ::google::bigtable::admin::v2::Table schema;
   ::google::bigtable::admin::v2::ColumnFamily column_family;
 
@@ -715,7 +716,7 @@ ::google::bigtable::v2::TimestampRange* NewTimestampRange(int64_t start,
 }
 
 // Does DeleteFromColumn basically work?
-TEST(TransactonRollback, DeleteFromColumnBasicFunction) {
+TEST(TransactionRollback, DeleteFromColumnBasicFunction) {
   ::google::bigtable::admin::v2::Table schema;
   ::google::bigtable::admin::v2::ColumnFamily column_family;
 
@@ -757,7 +758,7 @@ TEST(TransactonRollback, DeleteFromColumnBasicFunction) {
 }
 
 // Does DeleteFromColumn rollback work?
-TEST(TransactonRollback, DeleteFromColumnRollback) {
+TEST(TransactionRollback, DeleteFromColumnRollback) {
   ::google::bigtable::admin::v2::Table schema;
   ::google::bigtable::admin::v2::ColumnFamily column_family;
 
@@ -816,7 +817,7 @@ TEST(TransactonRollback, DeleteFromColumnRollback) {
 }
 
 // Can we delete a row from all column families?
-TEST(TransactonRollback, DeleteFromRowBasicFunction) {
+TEST(TransactionRollback, DeleteFromRowBasicFunction) {
   ::google::bigtable::admin::v2::Table schema;
   ::google::bigtable::admin::v2::ColumnFamily column_family;
 

google/cloud/bigtable/emulator/server.cc

@@ -143,8 +143,20 @@ class EmulatorService final : public btproto::Bigtable::Service {
 
   grpc::Status ReadModifyWriteRow(
       grpc::ServerContext* /* context */,
-      btproto::ReadModifyWriteRowRequest const* /* request */,
-      btproto::ReadModifyWriteRowResponse* /* response */) override {
+      btproto::ReadModifyWriteRowRequest const* request,
+      btproto::ReadModifyWriteRowResponse* response) override {
+    auto maybe_table = cluster_->FindTable(request->table_name());
+    if (!maybe_table) {
+      return ToGrpcStatus(maybe_table.status());
+    }
+
+    auto maybe_response = (*maybe_table)->ReadModifyWriteRow(*request);
+    if (!maybe_response) {
+      return ToGrpcStatus(maybe_response.status());
+    }
+
+    *response = std::move(maybe_response.value());
+
     return grpc::Status::OK;
   }