fix: support dump for external objects

Before: the code crashed on check-fail if a dump of the external string was requested.
This PR fixes it.

We call SerializerBase::DumpObject also in CmdSerializer::SerializeRestore,
and there we also do not handle the external strings, which will be fixed
in the future.

Signed-off-by: Roman Gershman <[email protected]>

Author: romange

src/server/generic_family.cc

@@ -31,7 +31,9 @@ extern "C" {
 #include "server/rdb_save.h"
 #include "server/search/doc_index.h"
 #include "server/set_family.h"
+#include "server/tiered_storage.h"
 #include "server/transaction.h"
+#include "util/fibers/future.h"
 #include "util/varz.h"
 
 ABSL_FLAG(uint32_t, dbnum, 16, "Number of databases");
@@ -524,9 +526,21 @@ OpResult<std::string> OpDump(const OpArgs& op_args, string_view key) {
   if (IsValid(it)) {
     DVLOG(1) << "Dump: key '" << key << "' successfully found, going to dump it";
     io::StringSink sink;
-    SerializerBase::DumpObject(it->second, &sink);
-    return sink.str();  // TODO: Add rvalue overload to str()
+    const PrimeValue& pv = it->second;
+
+    if (pv.IsExternal() && !pv.IsCool()) {
+      util::fb2::Future<string> future =
+          op_args.shard->tiered_storage()->Read(op_args.db_cntx.db_index, key, pv);
+
+      CompactObj co(future.Get());
+      SerializerBase::DumpObject(co, &sink);
+    } else {
+      SerializerBase::DumpObject(it->second, &sink);
+    }
+
+    return std::move(sink).str();
   }
+
   // fallback
   DVLOG(1) << "Dump: '" << key << "' Not found";
   return OpStatus::KEY_NOTFOUND;

src/server/rdb_save.cc

@@ -243,7 +243,14 @@ std::error_code RdbSerializer::SaveValue(const PrimeValue& pv) {
     if (opt_int) {
       ec = SaveLongLongAsString(*opt_int);
     } else {
-      ec = SaveString(pv.GetSlice(&tmp_str_));
+      if (pv.IsExternal()) {
+        if (pv.IsCool()) {
+          return SaveValue(pv.GetCool().record->value);
+        }
+        LOG(FATAL) << "External string not supported yet";
+      } else {
+        ec = SaveString(pv.GetSlice(&tmp_str_));
+      }
     }
   } else {
     ec = SaveObject(pv);

src/server/snapshot.cc

@@ -389,9 +389,8 @@ void SliceSnapshot::SerializeExternal(DbIndex db_index, PrimeKey key, const Prim
                                       time_t expire_time, uint32_t mc_flags) {
   // We prefer avoid blocking, so we just schedule a tiered read and append
   // it to the delayed entries.
-  util::fb2::Future<string> future;
-  auto cb = [future](const std::string& v) mutable { future.Resolve(v); };
-  EngineShard::tlocal()->tiered_storage()->Read(db_index, key.ToString(), pv, std::move(cb));
+  util::fb2::Future<string> future =
+      EngineShard::tlocal()->tiered_storage()->Read(db_index, key.ToString(), pv);
 
   delayed_entries_.push_back({db_index, std::move(key), std::move(future), expire_time, mc_flags});
   ++type_freq_map_[RDB_TYPE_STRING];

src/server/tiered_storage_test.cc

@@ -355,4 +355,21 @@ TEST_F(TieredStorageTest, SetExistingExpire) {
   }
 }
 
+TEST_F(TieredStorageTest, Dump) {
+  absl::FlagSaver saver;
+  SetFlag(&FLAGS_tiered_offload_threshold, 0.0f);  // offload all values
+
+  // we want to test without cooling to trigger disk I/O on reads.
+  SetFlag(&FLAGS_tiered_experimental_cooling, false);
+
+  const int kNum = 10;
+  for (size_t i = 0; i < kNum; i++) {
+    Run({"SET", absl::StrCat("k", i), BuildString(3000)});  // big enough to trigger offloading.
+  }
+
+  ExpectConditionWithinTimeout([&] { return GetMetrics().tiered_stats.total_stashes == kNum; });
+
+  Run({"DUMP", "k0"});
+}
+
 }  // namespace dfly