feat(split): scan support validate_partition_hash (#702)

apache · Mar 31, 2021 · 8e35518 · 8e35518
1 parent 5d54403
commit 8e35518
Show file tree

Hide file tree

Showing 9 changed files with 206 additions and 147 deletions.
diff --git a/src/base/rrdb_types.cpp b/src/base/rrdb_types.cpp
diff --git a/src/client_lib/pegasus_client_impl.cpp b/src/client_lib/pegasus_client_impl.cpp
@@ -1179,7 +1179,7 @@ int pegasus_client_impl::get_scanner(const std::string &hash_key,
     if (c < 0 || (c == 0 && o.start_inclusive && o.stop_inclusive)) {
         v.push_back(pegasus_key_hash(start));
     }
-    scanner = new pegasus_scanner_impl(_client, std::move(v), o, start, stop);
+    scanner = new pegasus_scanner_impl(_client, std::move(v), o, start, stop, false);
 
     return PERR_OK;
 }
@@ -1223,7 +1223,7 @@ void pegasus_client_impl::async_get_unordered_scanners(
                     std::vector<uint64_t> hash(s);
                     for (int j = 0; j < s; j++)
                         hash[j] = --count;
-                    scanners[i] = new pegasus_scanner_impl(_client, std::move(hash), options);
+                    scanners[i] = new pegasus_scanner_impl(_client, std::move(hash), options, true);
                 }
             }
         }

diff --git a/src/client_lib/pegasus_client_impl.h b/src/client_lib/pegasus_client_impl.h
@@ -265,12 +265,14 @@ class pegasus_client_impl : public pegasus_client
 
         pegasus_scanner_impl(::dsn::apps::rrdb_client *client,
                              std::vector<uint64_t> &&hash,
-                             const scan_options &options);
+                             const scan_options &options,
+                             bool validate_partition_hash);
         pegasus_scanner_impl(::dsn::apps::rrdb_client *client,
                              std::vector<uint64_t> &&hash,
                              const scan_options &options,
                              const ::dsn::blob &start_key,
-                             const ::dsn::blob &stop_key);
+                             const ::dsn::blob &stop_key,
+                             bool validate_partition_hash);
 
     private:
         ::dsn::apps::rrdb_client *_client;
@@ -288,6 +290,7 @@ class pegasus_client_impl : public pegasus_client
         mutable ::dsn::zlock _lock;
         std::list<async_scan_next_callback_t> _queue;
         volatile bool _rpc_started;
+        bool _validate_partition_hash;
 
         void _async_next_internal();
         void _start_scan();

diff --git a/src/client_lib/pegasus_scanner_impl.cpp b/src/client_lib/pegasus_scanner_impl.cpp
@@ -28,8 +28,9 @@ namespace client {
 
 pegasus_client_impl::pegasus_scanner_impl::pegasus_scanner_impl(::dsn::apps::rrdb_client *client,
                                                                 std::vector<uint64_t> &&hash,
-                                                                const scan_options &options)
-    : pegasus_scanner_impl(client, std::move(hash), options, _min, _max)
+                                                                const scan_options &options,
+                                                                bool validate_partition_hash)
+    : pegasus_scanner_impl(client, std::move(hash), options, _min, _max, validate_partition_hash)
 {
     _options.start_inclusive = true;
     _options.stop_inclusive = false;
@@ -39,15 +40,17 @@ pegasus_client_impl::pegasus_scanner_impl::pegasus_scanner_impl(::dsn::apps::rrd
                                                                 std::vector<uint64_t> &&hash,
                                                                 const scan_options &options,
                                                                 const ::dsn::blob &start_key,
-                                                                const ::dsn::blob &stop_key)
+                                                                const ::dsn::blob &stop_key,
+                                                                bool validate_partition_hash)
     : _client(client),
       _start_key(start_key),
       _stop_key(stop_key),
       _options(options),
       _splits_hash(std::move(hash)),
       _p(-1),
       _context(SCAN_CONTEXT_ID_COMPLETED),
-      _rpc_started(false)
+      _rpc_started(false),
+      _validate_partition_hash(validate_partition_hash)
 {
 }
 
@@ -211,6 +214,7 @@ void pegasus_client_impl::pegasus_scanner_impl::_start_scan()
     req.sort_key_filter_pattern = ::dsn::blob(
         _options.sort_key_filter_pattern.data(), 0, _options.sort_key_filter_pattern.size());
     req.no_value = _options.no_value;
+    req.__set_validate_partition_hash(_validate_partition_hash);
 
     dassert(!_rpc_started, "");
     _rpc_started = true;

diff --git a/src/idl/rrdb.thrift b/src/idl/rrdb.thrift
@@ -252,6 +252,7 @@ struct get_scanner_request
     8:dsn.blob     hash_key_filter_pattern;
     9:filter_type  sort_key_filter_type;
     10:dsn.blob    sort_key_filter_pattern;
+    11:optional bool    validate_partition_hash;
 }
 
 struct scan_request

diff --git a/src/include/rrdb/rrdb_types.h b/src/include/rrdb/rrdb_types.h
diff --git a/src/server/pegasus_scan_context.h b/src/server/pegasus_scan_context.h
@@ -41,7 +41,8 @@ struct pegasus_scan_context
                          ::dsn::apps::filter_type::type sort_key_filter_type_,
                          const std::string &&sort_key_filter_pattern_,
                          int32_t batch_size_,
-                         bool no_value_)
+                         bool no_value_,
+                         bool validate_partition_hash_)
         : _stop_holder(std::move(stop_)),
           _hash_key_filter_pattern_holder(std::move(hash_key_filter_pattern_)),
           _sort_key_filter_pattern_holder(std::move(sort_key_filter_pattern_)),
@@ -55,7 +56,8 @@ struct pegasus_scan_context
           sort_key_filter_pattern(
               _sort_key_filter_pattern_holder.data(), 0, _sort_key_filter_pattern_holder.length()),
           batch_size(batch_size_),
-          no_value(no_value_)
+          no_value(no_value_),
+          validate_partition_hash(validate_partition_hash_)
     {
     }
 
@@ -74,6 +76,7 @@ struct pegasus_scan_context
     dsn::blob sort_key_filter_pattern;
     int32_t batch_size;
     bool no_value;
+    bool validate_partition_hash;
 };
 
 class pegasus_context_cache