Merge branch 'upstream_emulator' into sample_row_keys

Author: marete

google/cloud/bigtable/emulator/column_family.cc

@@ -29,20 +29,6 @@ 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>(
@@ -56,10 +42,7 @@ StatusOr<ReadModifyWriteCellResult> ColumnRow::ReadModifyWrite(
                                      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 latest_it = cells_.begin();
 
   auto maybe_old_value =
       google::cloud::internal::DecodeBigEndian<std::int64_t>(latest_it->second);
@@ -97,10 +80,7 @@ ReadModifyWriteCellResult ColumnRow::ReadModifyWrite(
                                      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 latest_it = cells_.begin();
 
   auto value = latest_it->second + append_value;
 
@@ -159,14 +139,21 @@ StatusOr<absl::optional<std::string>> ColumnRow::UpdateCell(
 std::vector<Cell> ColumnRow::DeleteTimeRange(
     ::google::bigtable::v2::TimestampRange const& time_range) {
   std::vector<Cell> deleted_cells;
-  for (auto cell_it = cells_.lower_bound(
-           std::chrono::duration_cast<std::chrono::milliseconds>(
-               std::chrono::microseconds(time_range.start_timestamp_micros())));
+  absl::optional<std::int64_t> maybe_end_micros =
+      time_range.end_timestamp_micros();
+  if (maybe_end_micros.value_or(0) == 0) {
+    maybe_end_micros.reset();
+  }
+  for (auto cell_it =
+           maybe_end_micros
+               ? upper_bound(
+                     std::chrono::duration_cast<std::chrono::milliseconds>(
+                         std::chrono::microseconds(*maybe_end_micros)))
+               : begin();
        cell_it != cells_.end() &&
-       (time_range.end_timestamp_micros() == 0 ||
-        cell_it->first < std::chrono::duration_cast<std::chrono::milliseconds>(
+       cell_it->first >= std::chrono::duration_cast<std::chrono::milliseconds>(
                              std::chrono::microseconds(
-                                 time_range.end_timestamp_micros())));) {
+                                 time_range.start_timestamp_micros()));) {
     Cell cell = {std::move(cell_it->first), std::move(cell_it->second)};
     deleted_cells.emplace_back(std::move(cell));
     cells_.erase(cell_it++);
@@ -346,9 +333,9 @@ FilteredColumnFamilyStream::FilteredColumnFamilyStream(
       row_ranges_(std::move(row_set)),
       column_ranges_(StringRangeSet::All()),
       timestamp_ranges_(TimestampRangeSet::All()),
-      rows_(RangeFilteredMapView<ColumnFamily, StringRangeSet>(column_family,
-                                                               *row_ranges_),
-            std::cref(row_regexes_)) {}
+      rows_(
+          StringRangeFilteredMapView<ColumnFamily>(column_family, *row_ranges_),
+          std::cref(row_regexes_)) {}
 
 bool FilteredColumnFamilyStream::ApplyFilter(
     InternalFilter const& internal_filter) {
@@ -404,7 +391,7 @@ void FilteredColumnFamilyStream::InitializeIfNeeded() const {
 
 bool FilteredColumnFamilyStream::PointToFirstCellAfterColumnChange() const {
   for (; column_it_.value() != columns_.value().end(); ++(column_it_.value())) {
-    cells_ = RangeFilteredMapView<ColumnRow, TimestampRangeSet>(
+    cells_ = TimestampRangeFilteredMapView<ColumnRow>(
         column_it_.value()->second, timestamp_ranges_);
     cell_it_ = cells_.value().begin();
     if (cell_it_.value() != cells_.value().end()) {
@@ -416,10 +403,9 @@ bool FilteredColumnFamilyStream::PointToFirstCellAfterColumnChange() const {
 
 bool FilteredColumnFamilyStream::PointToFirstCellAfterRowChange() const {
   for (; (*row_it_) != rows_.end(); ++(*row_it_)) {
-    columns_ = RegexFiteredMapView<
-        RangeFilteredMapView<ColumnFamilyRow, StringRangeSet>>(
-        RangeFilteredMapView<ColumnFamilyRow, StringRangeSet>(
-            (*row_it_)->second, column_ranges_),
+    columns_ = RegexFiteredMapView<StringRangeFilteredMapView<ColumnFamilyRow>>(
+        StringRangeFilteredMapView<ColumnFamilyRow>((*row_it_)->second,
+                                                    column_ranges_),
         column_regexes_);
     column_it_ = columns_.value().begin();
     if (PointToFirstCellAfterColumnChange()) {

google/cloud/bigtable/emulator/column_family.h

@@ -85,8 +85,7 @@ class ColumnRow {
    * Insert or update and existing cell at a given timestamp.
    *
    * @param timestamp the time stamp at which the value will be inserted or
-   *     updated. If it equals zero then number of milliseconds since epoch will
-   *     be used instead.
+   *     updated.
    * @param value the value to insert/update.
    *
    * @return no value if the timestamp had no value before, otherwise
@@ -121,32 +120,28 @@ class ColumnRow {
   absl::optional<Cell> DeleteTimeStamp(std::chrono::milliseconds timestamp);
 
   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;
+
+  using const_iterator = std::map<std::chrono::milliseconds, std::string,
+                                  std::greater<>>::const_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);
   }
   const_iterator upper_bound(std::chrono::milliseconds timestamp) const {
     return cells_.upper_bound(timestamp);
   }
 
-  std::map<std::chrono::milliseconds, std::string>::iterator find(
-      std::chrono::milliseconds const& timestamp) {
+  const_iterator find(std::chrono::milliseconds const& timestamp) {
     return cells_.find(timestamp);
   }
 
-  void erase(
-      std::map<std::chrono::milliseconds, std::string>::iterator timestamp_it) {
-    cells_.erase(timestamp_it);
-  }
+  void erase(const_iterator timestamp_it) { cells_.erase(timestamp_it); }
 
  private:
-  std::map<std::chrono::milliseconds, std::string> cells_;
+  // Note the order - the iterator return the freshest cells first.
+  std::map<std::chrono::milliseconds, std::string, std::greater<>> cells_;
 };
 
 /**
@@ -174,8 +169,7 @@ class ColumnFamilyRow {
    *
    * @param column_qualifier the column qualifier at which to update the value.
    * @param timestamp the time stamp at which the value will be inserted or
-   *     updated. If it equals zero then number of milliseconds since epoch will
-   *     be used instead.
+   *     updated.
    * @param value the value to insert/update.
    *
    * @return no value if the timestamp had no value before, otherwise
@@ -288,8 +282,7 @@ class ColumnFamily {
    * @param row_key the row key at which to update the value.
    * @param column_qualifier the column qualifier at which to update the value.
    * @param timestamp the time stamp at which the value will be inserted or
-   *     updated. If it equals zero then number of milliseconds since epoch will
-   *     be used instead.
+   *     updated.
    * @param value the value to insert/update.
    *
    * @return no value if the timestamp had no value before, otherwise
@@ -432,8 +425,8 @@ class ColumnFamily {
     if (!existing_int) {
       return existing_int.status();
     }
-    auto new_int = google::cloud::internal::DecodeBigEndian<std::int64_t>(
-        std::move(new_value));
+    auto new_int =
+        google::cloud::internal::DecodeBigEndian<std::int64_t>(new_value);
     if (!new_int) {
       return new_int.status();
     }
@@ -452,8 +445,8 @@ class ColumnFamily {
     if (!existing_int) {
       return existing_int.status();
     }
-    auto new_int = google::cloud::internal::DecodeBigEndian<std::int64_t>(
-        std::move(new_value));
+    auto new_int =
+        google::cloud::internal::DecodeBigEndian<std::int64_t>(new_value);
     if (!new_int) {
       return new_int.status();
     }
@@ -540,25 +533,24 @@ class FilteredColumnFamilyStream : public AbstractCellStreamImpl {
   std::vector<std::shared_ptr<re2::RE2 const>> column_regexes_;
   mutable TimestampRangeSet timestamp_ranges_;
 
-  RegexFiteredMapView<RangeFilteredMapView<ColumnFamily, StringRangeSet>> rows_;
-  mutable absl::optional<RegexFiteredMapView<
-      RangeFilteredMapView<ColumnFamilyRow, StringRangeSet>>>
+  RegexFiteredMapView<StringRangeFilteredMapView<ColumnFamily>> rows_;
+  mutable absl::optional<
+      RegexFiteredMapView<StringRangeFilteredMapView<ColumnFamilyRow>>>
       columns_;
-  mutable absl::optional<RangeFilteredMapView<ColumnRow, TimestampRangeSet>>
-      cells_;
+  mutable absl::optional<TimestampRangeFilteredMapView<ColumnRow>> cells_;
 
   // If row_it_ == rows_.end() we've reached the end.
   // We maintain the following invariant:
   //   if (row_it_ != rows_.end()) then
   //   cell_it_ != cells.end() && column_it_ != columns_.end().
   mutable absl::optional<RegexFiteredMapView<
-      RangeFilteredMapView<ColumnFamily, StringRangeSet>>::const_iterator>
+      StringRangeFilteredMapView<ColumnFamily>>::const_iterator>
       row_it_;
   mutable absl::optional<RegexFiteredMapView<
-      RangeFilteredMapView<ColumnFamilyRow, StringRangeSet>>::const_iterator>
+      StringRangeFilteredMapView<ColumnFamilyRow>>::const_iterator>
       column_it_;
   mutable absl::optional<
-      RangeFilteredMapView<ColumnRow, TimestampRangeSet>::const_iterator>
+      TimestampRangeFilteredMapView<ColumnRow>::const_iterator>
       cell_it_;
   mutable absl::optional<CellView> cur_value_;
   mutable bool initialized_{false};

google/cloud/bigtable/emulator/column_family_test.cc

@@ -13,9 +13,7 @@
 // limitations under the License.
 
 #include "google/cloud/bigtable/emulator/column_family.h"
-#include "google/cloud/bigtable/row_range.h"
 #include "google/cloud/testing_util/chrono_literals.h"
-#include "google/cloud/testing_util/is_proto_equal.h"
 #include <google/protobuf/text_format.h>
 #include <gmock/gmock.h>
 #include <array>
@@ -69,8 +67,14 @@ TEST(ColumnRow, Trivial) {
 
   col_row.SetCell(0_ms, "baz");
   col_row.SetCell(20_ms, "qux");
+  EXPECT_EQ("qux", col_row.lower_bound(30_ms)->second);
+  EXPECT_EQ("qux", col_row.lower_bound(20_ms)->second);
   EXPECT_EQ("bar", col_row.lower_bound(10_ms)->second);
-  EXPECT_EQ("qux", col_row.upper_bound(10_ms)->second);
+  EXPECT_EQ("baz", col_row.lower_bound(0_ms)->second);
+  EXPECT_EQ("qux", col_row.upper_bound(30_ms)->second);
+  EXPECT_EQ("bar", col_row.upper_bound(20_ms)->second);
+  EXPECT_EQ("baz", col_row.upper_bound(10_ms)->second);
+  EXPECT_EQ(col_row.end(), col_row.upper_bound(0_ms));
 }
 
 TEST(ColumnRow, DeleteTimeRangeFinite) {
@@ -225,14 +229,14 @@ TEST(FilteredColumnFamilyStream, Unfiltered) {
   FilteredColumnFamilyStream filtered_stream(fam, "cf1", included_rows);
   EXPECT_EQ(R"""(
 row0 cf1:col0 @10ms: foo
-row0 cf1:col1 @20ms: bar
 row0 cf1:col1 @30ms: baz
+row0 cf1:col1 @20ms: bar
 row1 cf1:col0 @10ms: foo
-row1 cf1:col1 @20ms: foo
 row1 cf1:col1 @30ms: foo
+row1 cf1:col1 @20ms: foo
 row2 cf1:col0 @10ms: qux
-row2 cf1:col2 @40ms: qux
 row2 cf1:col2 @50ms: qux
+row2 cf1:col2 @40ms: qux
 )""",
             "\n" + DumpFilteredColumnFamilyStream(filtered_stream));
 }
@@ -272,12 +276,12 @@ TEST(FilteredColumnFamilyStream, FilterByTimestampRange) {
       TimestampRange{TimestampRangeSet::Range(100_ms, 200_ms)});
   EXPECT_EQ(R"""(
 row0 cf1:col0 @100ms: foo
-row0 cf1:col2 @100ms: foo
-row0 cf1:col2 @120ms: foo
 row0 cf1:col2 @140ms: foo
-row1 cf1:col2 @100ms: foo
-row1 cf1:col2 @120ms: foo
+row0 cf1:col2 @120ms: foo
+row0 cf1:col2 @100ms: foo
 row1 cf1:col2 @140ms: foo
+row1 cf1:col2 @120ms: foo
+row1 cf1:col2 @100ms: foo
 )""",
             "\n" + DumpFilteredColumnFamilyStream(filtered_stream));
 }

google/cloud/bigtable/emulator/drop_row_range_test.cc

@@ -13,13 +13,11 @@
 // limitations under the License.
 
 #include "google/cloud/bigtable/emulator/column_family.h"
-#include "google/cloud/bigtable/emulator/row_streamer.h"
 #include "google/cloud/bigtable/emulator/table.h"
 #include "google/cloud/internal/make_status.h"
 #include "google/cloud/status.h"
 #include "google/cloud/status_or.h"
 #include "google/cloud/testing_util/status_matchers.h"
-#include "gmock/gmock.h"
 #include <google/bigtable/admin/v2/bigtable_table_admin.grpc.pb.h>
 #include <google/bigtable/admin/v2/bigtable_table_admin.pb.h>
 #include <google/bigtable/admin/v2/table.pb.h>

google/cloud/bigtable/emulator/filter.cc

@@ -19,7 +19,6 @@
 #include "google/cloud/internal/make_status.h"
 #include "google/cloud/status_or.h"
 #include <re2/re2.h>
-#include <queue>
 #include <random>
 
 namespace google {
@@ -325,8 +324,8 @@ class TrivialFilter : public AbstractCellStreamImpl {
  * @param filter_filter a functor which given an `InternalFilter` decides
  *     whether filtering this cell stream's results and then applying the
  *     `InternalFilter` would yield the same results as applying
- *     `InternalFilter` to the underlying stream and the perform this stream's
- *     filtering.
+ *     `InternalFilter` to the underlying stream and then performing this
+ *     stream's filtering.
  */
 template <typename Filter>
 CellStream MakeTrivialFilter(
@@ -539,14 +538,14 @@ class ConditionStream : public AbstractCellStreamImpl {
     if (condition_true_) {
       true_stream_.Next(mode);
       if (!true_stream_ ||
-          !internal::CompareRowKey(current_row_, true_stream_->row_key())) {
+          internal::CompareRowKey(current_row_, true_stream_->row_key()) != 0) {
         source_.Next(NextMode::kRow);
         OnNewRow();
       }
     } else {
       false_stream_.Next(mode);
-      if (!false_stream_ ||
-          !internal::CompareRowKey(current_row_, false_stream_->row_key())) {
+      if (!false_stream_ || internal::CompareRowKey(
+                                current_row_, false_stream_->row_key()) != 0) {
         source_.Next(NextMode::kRow);
         OnNewRow();
       }
@@ -870,7 +869,7 @@ StatusOr<CellStreamConstructor> CreateFilterImpl(
             if (per_row_state-- <= 0) {
               return {};
             }
-            return NextMode::kRow;
+            return NextMode::kCell;
           },
           [cells_per_row_offset]() { return cells_per_row_offset; },
           [](InternalFilter const& internal_filter) {
@@ -971,10 +970,14 @@ StatusOr<CellStreamConstructor> CreateFilterImpl(
     }
     CellStreamConstructor res = [source_ctor = std::move(source_ctor)] {
       auto source = source_ctor();
-      return MakeTrivialTransformer(std::move(source), [](CellView cell_view) {
-        cell_view.SetValue("");
-        return cell_view;
-      });
+      // We need to ensure that the value outlives the reference.
+      std::string const stripped_value;
+      return MakeTrivialTransformer(
+          std::move(source),
+          [stripped_value = std::move(stripped_value)](CellView cell_view) {
+            cell_view.SetValue(stripped_value);
+            return cell_view;
+          });
     };
     return res;
   }
@@ -1046,6 +1049,16 @@ StatusOr<CellStreamConstructor> CreateFilterImpl(
           "`condition` must have a `predicate_filter` set.",
           GCP_ERROR_INFO().WithMetadata("filter", filter.DebugString()));
     }
+    if (!filter.condition().has_true_filter() &&
+        !filter.condition().has_false_filter()) {
+      return InvalidArgumentError(
+          "`condition` must have `true_filter` or `false_filter` set.",
+          GCP_ERROR_INFO().WithMetadata("filter", filter.DebugString()));
+    }
+    // FIXME: validate that `sink` is not present in condition's predicate.
+    // Expected error:
+    //  INVALID_ARGUMENT: Error in field 'condition filter predicate' : sink
+    //  cannot be nested in a condition filter
 
     auto maybe_predicate_stream_ctor = CreateFilterImpl(
         filter.condition().predicate_filter(), source_ctor, direct_sinks);