Fix link error reported in issue 9272 (#9278)

Summary:
As title, Closes https://github.com/facebook/rocksdb/issues/9272
Since TimestampAssigner-related classes needs to access
`WriteBatch::ProtectionInfo` objects which is for internal use only,
it's difficult to make `AssignTimestamp` methods a template and put them
in the same public header, `include/rocksdb/write_batch.h`.

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

Test Plan:
```
make check
# Also manually test following the repro-steps in issue 9272
```

Reviewed By: ltamasi

Differential Revision: D33012686

Pulled By: riversand963

fbshipit-source-id: 89f24a86a1170125bd0b94ef3b32e69aa08bd949

db/write_batch.cc

@@ -1223,6 +1223,20 @@ Status WriteBatch::PopSavePoint() {
   return Status::OK();
 }
 
+Status WriteBatch::AssignTimestamp(
+    const Slice& ts, std::function<Status(uint32_t, size_t&)> checker) {
+  TimestampAssigner ts_assigner(prot_info_.get(), std::move(checker), ts);
+  return Iterate(&ts_assigner);
+}
+
+Status WriteBatch::AssignTimestamps(
+    const std::vector<Slice>& ts_list,
+    std::function<Status(uint32_t, size_t&)> checker) {
+  SimpleListTimestampAssigner ts_assigner(prot_info_.get(), std::move(checker),
+                                          ts_list);
+  return Iterate(&ts_assigner);
+}
+
 class MemTableInserter : public WriteBatch::Handler {
 
   SequenceNumber sequence_;

db/write_batch_internal.h

@@ -265,11 +265,12 @@ class LocalSavePoint {
 #endif
 };
 
-template <typename Derived, typename Checker>
+template <typename Derived>
 class TimestampAssignerBase : public WriteBatch::Handler {
  public:
-  explicit TimestampAssignerBase(WriteBatch::ProtectionInfo* prot_info,
+  explicit TimestampAssignerBase(
-                                 Checker&& checker)
+      WriteBatch::ProtectionInfo* prot_info,
+      std::function<Status(uint32_t, size_t&)>&& checker)
       : prot_info_(prot_info), checker_(std::move(checker)) {}
 
   ~TimestampAssignerBase() override {}
@@ -360,27 +361,25 @@ class TimestampAssignerBase : public WriteBatch::Handler {
   TimestampAssignerBase& operator=(TimestampAssignerBase&&) = delete;
 
   WriteBatch::ProtectionInfo* const prot_info_ = nullptr;
-  const Checker checker_{};
+  const std::function<Status(uint32_t, size_t&)> checker_{};
   size_t idx_ = 0;
 };
 
-template <typename Checker>
 class SimpleListTimestampAssigner
-    : public TimestampAssignerBase<SimpleListTimestampAssigner<Checker>,
+    : public TimestampAssignerBase<SimpleListTimestampAssigner> {
-                                   Checker> {
  public:
-  explicit SimpleListTimestampAssigner(WriteBatch::ProtectionInfo* prot_info,
+  explicit SimpleListTimestampAssigner(
-                                       Checker checker,
+      WriteBatch::ProtectionInfo* prot_info,
+      std::function<Status(uint32_t, size_t&)>&& checker,
       const std::vector<Slice>& timestamps)
-      : TimestampAssignerBase<SimpleListTimestampAssigner<Checker>, Checker>(
+      : TimestampAssignerBase<SimpleListTimestampAssigner>(prot_info,
-            prot_info, std::move(checker)),
+                                                           std::move(checker)),
         timestamps_(timestamps) {}
 
   ~SimpleListTimestampAssigner() override {}
 
  private:
-  friend class TimestampAssignerBase<SimpleListTimestampAssigner<Checker>,
+  friend class TimestampAssignerBase<SimpleListTimestampAssigner>;
-                                     Checker>;
 
   Status AssignTimestampImpl(uint32_t cf, const Slice& key, size_t idx) {
     if (idx >= timestamps_.size()) {
@@ -398,21 +397,19 @@ class SimpleListTimestampAssigner
   const std::vector<Slice>& timestamps_;
 };
 
-template <typename Checker>
+class TimestampAssigner : public TimestampAssignerBase<TimestampAssigner> {
-class TimestampAssigner
-    : public TimestampAssignerBase<TimestampAssigner<Checker>, Checker> {
  public:
   explicit TimestampAssigner(WriteBatch::ProtectionInfo* prot_info,
-                             Checker checker, const Slice& ts)
+                             std::function<Status(uint32_t, size_t&)>&& checker,
-      : TimestampAssignerBase<TimestampAssigner<Checker>, Checker>(
+                             const Slice& ts)
-            prot_info, std::move(checker)),
+      : TimestampAssignerBase<TimestampAssigner>(prot_info, std::move(checker)),
         timestamp_(ts) {
     assert(!timestamp_.empty());
   }
   ~TimestampAssigner() override {}
 
  private:
-  friend class TimestampAssignerBase<TimestampAssigner<Checker>, Checker>;
+  friend class TimestampAssignerBase<TimestampAssigner>;
 
   Status AssignTimestampImpl(uint32_t cf, const Slice& key, size_t /*idx*/) {
     if (timestamp_.empty()) {
@@ -429,18 +426,4 @@ class TimestampAssigner
   const Slice timestamp_;
 };
 
-template <typename Checker>
-Status WriteBatch::AssignTimestamp(const Slice& ts, Checker checker) {
-  TimestampAssigner<Checker> ts_assigner(prot_info_.get(), checker, ts);
-  return Iterate(&ts_assigner);
-}
-
-template <typename Checker>
-Status WriteBatch::AssignTimestamps(const std::vector<Slice>& ts_list,
-                                    Checker checker) {
-  SimpleListTimestampAssigner<Checker> ts_assigner(prot_info_.get(), checker,
-                                                   ts_list);
-  return Iterate(&ts_assigner);
-}
-
 }  // namespace ROCKSDB_NAMESPACE

include/rocksdb/write_batch.h

@@ -342,25 +342,29 @@ class WriteBatch : public WriteBatchBase {
   // Returns true if MarkRollback will be called during Iterate
   bool HasRollback() const;
 
-  struct TimestampChecker final {
-    Status operator()(uint32_t /*cf*/, size_t& /*ts_sz*/) const {
-      return Status::OK();
-    }
-  };
-
   // Experimental.
   // Assign timestamp to write batch.
   // This requires that all keys, if enable timestamp, (possibly from multiple
   // column families) in the write batch have timestamps of the same format.
+  //
   // checker: callable object to check the timestamp sizes of column families.
+  //
+  // in: cf, the column family id.
+  // in/out: ts_sz. Input as the expected timestamp size of the column
+  //         family, output as the actual timestamp size of the column family.
+  // ret: OK if assignment succeeds.
+  // Status checker(uint32_t cf, size_t& ts_sz);
+  //
   // User can call checker(uint32_t cf, size_t& ts_sz) which does the
   // following:
-  // 1. find out the timestamp size of cf.
+  // 1. find out the timestamp size of the column family whose id equals `cf`.
   // 2. if cf's timestamp size is 0, then set ts_sz to 0 and return OK.
   // 3. otherwise, compare ts_sz with cf's timestamp size and return
   //    Status::InvalidArgument() if different.
-  template <typename Checker = TimestampChecker>
+  Status AssignTimestamp(
-  Status AssignTimestamp(const Slice& ts, Checker checker = Checker());
+      const Slice& ts,
+      std::function<Status(uint32_t, size_t&)> checker =
+          [](uint32_t /*cf*/, size_t& /*ts_sz*/) { return Status::OK(); });
 
   // Experimental.
   // Assign timestamps to write batch.
@@ -369,16 +373,25 @@ class WriteBatch : public WriteBatchBase {
   // families can enable timestamp, while others disable the feature.
   // If key does not have timestamp, then put an empty Slice in ts_list as
   // a placeholder.
+  //
   // checker: callable object specified by caller to check the timestamp sizes
   // of column families.
+  //
+  // in: cf, the column family id.
+  // in/out: ts_sz. Input as the expected timestamp size of the column
+  //         family, output as the actual timestamp size of the column family.
+  // ret: OK if assignment succeeds.
+  // Status checker(uint32_t cf, size_t& ts_sz);
+  //
   // User can call checker(uint32_t cf, size_t& ts_sz) which does the
   // following:
-  // 1. find out the timestamp size of cf.
+  // 1. find out the timestamp size of the column family whose id equals `cf`.
   // 2. compare ts_sz with cf's timestamp size and return
   //    Status::InvalidArgument() if different.
-  template <typename Checker = TimestampChecker>
+  Status AssignTimestamps(
-  Status AssignTimestamps(const std::vector<Slice>& ts_list,
+      const std::vector<Slice>& ts_list,
-                          Checker checker = Checker());
+      std::function<Status(uint32_t, size_t&)> checker =
+          [](uint32_t /*cf*/, size_t& /*ts_sz*/) { return Status::OK(); });
 
   using WriteBatchBase::GetWriteBatch;
   WriteBatch* GetWriteBatch() override { return this; }