Generalize and move the wide-column Find method to facilitate reuse (#13168)

Summary:
Pull Request resolved: https://github.com/facebook/rocksdb/pull/13168

The patch moves `WideColumnSerialization::Find` to `WideColumnsHelper` to facilitate reuse in non-serialization-related contexts. It also generalizes the method to take a range of iterators, and templatizes it on the iterator type to enable using it with both `const` and non-`const` iterators. Finally, it adds an assertion to ensure the method is called with a properly sorted range, which is a precondition for binary search.

Reviewed By: jaykorean

Differential Revision: D66602558

fbshipit-source-id: 841a885af31e183edeb7e3314167c55f8ed53ff1

db/wide/wide_column_serialization.cc

@@ -129,21 +129,6 @@ Status WideColumnSerialization::Deserialize(Slice& input,
   return Status::OK();
 }
 
-WideColumns::const_iterator WideColumnSerialization::Find(
-    const WideColumns& columns, const Slice& column_name) {
-  const auto it =
-      std::lower_bound(columns.cbegin(), columns.cend(), column_name,
-                       [](const WideColumn& lhs, const Slice& rhs) {
-                         return lhs.name().compare(rhs) < 0;
-                       });
-
-  if (it == columns.cend() || it->name() != column_name) {
-    return columns.cend();
-  }
-
-  return it;
-}
-
 Status WideColumnSerialization::GetValueOfDefaultColumn(Slice& input,
                                                         Slice& value) {
   WideColumns columns;

db/wide/wide_column_serialization.h

@@ -47,8 +47,6 @@ class WideColumnSerialization {
 
   static Status Deserialize(Slice& input, WideColumns& columns);
 
-  static WideColumns::const_iterator Find(const WideColumns& columns,
-                                          const Slice& column_name);
   static Status GetValueOfDefaultColumn(Slice& input, Slice& value);
 
   static constexpr uint32_t kCurrentVersion = 1;

db/wide/wide_column_serialization_test.cc

@@ -5,6 +5,7 @@
 
 #include "db/wide/wide_column_serialization.h"
 
+#include "db/wide/wide_columns_helper.h"
 #include "test_util/testharness.h"
 #include "util/coding.h"
 
@@ -99,27 +100,28 @@ TEST(WideColumnSerializationTest, SerializeDeserialize) {
   ASSERT_EQ(columns, deserialized_columns);
 
   {
-    const auto it = WideColumnSerialization::Find(deserialized_columns, "foo");
+    const auto it = WideColumnsHelper::Find(deserialized_columns.cbegin(),
+                                            deserialized_columns.cend(), "foo");
     ASSERT_NE(it, deserialized_columns.cend());
     ASSERT_EQ(*it, deserialized_columns.front());
   }
 
   {
-    const auto it =
-        WideColumnSerialization::Find(deserialized_columns, "hello");
+    const auto it = WideColumnsHelper::Find(
+        deserialized_columns.cbegin(), deserialized_columns.cend(), "hello");
     ASSERT_NE(it, deserialized_columns.cend());
     ASSERT_EQ(*it, deserialized_columns.back());
   }
 
   {
-    const auto it =
-        WideColumnSerialization::Find(deserialized_columns, "fubar");
+    const auto it = WideColumnsHelper::Find(
+        deserialized_columns.cbegin(), deserialized_columns.cend(), "fubar");
     ASSERT_EQ(it, deserialized_columns.cend());
   }
 
   {
-    const auto it =
-        WideColumnSerialization::Find(deserialized_columns, "snafu");
+    const auto it = WideColumnsHelper::Find(
+        deserialized_columns.cbegin(), deserialized_columns.cend(), "snafu");
     ASSERT_EQ(it, deserialized_columns.cend());
   }
 }

db/wide/wide_columns_helper.cc

@@ -5,7 +5,6 @@
 
 #include "db/wide/wide_columns_helper.h"
 
-#include <algorithm>
 #include <ios>
 
 #include "db/wide/wide_column_serialization.h"
@@ -42,11 +41,4 @@ Status WideColumnsHelper::DumpSliceAsWideColumns(const Slice& value,
   return s;
 }
 
-void WideColumnsHelper::SortColumns(WideColumns& columns) {
-  std::sort(columns.begin(), columns.end(),
-            [](const WideColumn& lhs, const WideColumn& rhs) {
-              return lhs.name().compare(rhs.name()) < 0;
-            });
-}
-
 }  // namespace ROCKSDB_NAMESPACE

db/wide/wide_columns_helper.h

@@ -4,8 +4,10 @@
 //  (found in the LICENSE.Apache file in the root directory).
 
 #pragma once
+
+#include <algorithm>
+#include <cassert>
 #include <ostream>
-#include <string>
 
 #include "rocksdb/rocksdb_namespace.h"
 #include "rocksdb/wide_columns.h"
@@ -34,7 +36,31 @@ class WideColumnsHelper {
     return columns.front().value();
   }
 
-  static void SortColumns(WideColumns& columns);
+  static void SortColumns(WideColumns& columns) {
+    std::sort(columns.begin(), columns.end(),
+              [](const WideColumn& lhs, const WideColumn& rhs) {
+                return lhs.name().compare(rhs.name()) < 0;
+              });
+  }
+
+  template <typename Iterator>
+  static Iterator Find(Iterator begin, Iterator end, const Slice& column_name) {
+    assert(std::is_sorted(begin, end,
+                          [](const WideColumn& lhs, const WideColumn& rhs) {
+                            return lhs.name().compare(rhs.name()) < 0;
+                          }));
+
+    auto it = std::lower_bound(begin, end, column_name,
+                               [](const WideColumn& lhs, const Slice& rhs) {
+                                 return lhs.name().compare(rhs) < 0;
+                               });
+
+    if (it == end || it->name() != column_name) {
+      return end;
+    }
+
+    return it;
+  }
 };
 
 }  // namespace ROCKSDB_NAMESPACE