import quincy beta 17.1.0

[ceph.git] / ceph / src / rocksdb / table / cuckoo / cuckoo_table_builder.cc

diff --git a/ceph/src/rocksdb/table/cuckoo/cuckoo_table_builder.cc b/ceph/src/rocksdb/table/cuckoo/cuckoo_table_builder.cc
index 3ddce0b6ff88eb7e251b66be54a8ab150b496095..77df906c9940e455e4b6bd103dbaea13d5e2cc0b 100644 (file)
--- a/ceph/src/rocksdb/table/cuckoo/cuckoo_table_builder.cc
+++ b/ceph/src/rocksdb/table/cuckoo/cuckoo_table_builder.cc
@@ -53,7 +53,8 @@ CuckooTableBuilder::CuckooTableBuilder(
     const Comparator* user_comparator, uint32_t cuckoo_block_size,
     bool use_module_hash, bool identity_as_first_hash,
     uint64_t (*get_slice_hash)(const Slice&, uint32_t, uint64_t),
-    uint32_t column_family_id, const std::string& column_family_name)
+    uint32_t column_family_id, const std::string& column_family_name,
+    const std::string& db_id, const std::string& db_session_id)
     : num_hash_func_(2),
       file_(file),
       max_hash_table_ratio_(max_hash_table_ratio),
@@ -79,6 +80,8 @@ CuckooTableBuilder::CuckooTableBuilder(
   properties_.filter_size = 0;
   properties_.column_family_id = column_family_id;
   properties_.column_family_name = column_family_name;
+  properties_.db_id = db_id;
+  properties_.db_session_id = db_session_id;
 }
 
 void CuckooTableBuilder::Add(const Slice& key, const Slice& value) {
@@ -87,8 +90,11 @@ void CuckooTableBuilder::Add(const Slice& key, const Slice& value) {
     return;
   }
   ParsedInternalKey ikey;
-  if (!ParseInternalKey(key, &ikey)) {
-    status_ = Status::Corruption("Unable to parse key into inernal key.");
+  Status pik_status =
+      ParseInternalKey(key, &ikey, false /* log_err_key */);  // TODO
+  if (!pik_status.ok()) {
+    status_ = Status::Corruption("Unable to parse key into internal key. ",
+                                 pik_status.getState());
     return;
   }
   if (ikey.type != kTypeDeletion && ikey.type != kTypeValue) {
@@ -252,9 +258,9 @@ Status CuckooTableBuilder::Finish() {
       hash_table_size_ =
         static_cast<uint64_t>(num_entries_ / max_hash_table_ratio_);
     }
-    s = MakeHashTable(&buckets);
-    if (!s.ok()) {
-      return s;
+    status_ = MakeHashTable(&buckets);
+    if (!status_.ok()) {
+      return status_;
     }
     // Determine unused_user_key to fill empty buckets.
     std::string unused_user_key = smallest_user_key_;
@@ -301,18 +307,19 @@ Status CuckooTableBuilder::Finish() {
   uint32_t num_added = 0;
   for (auto& bucket : buckets) {
     if (bucket.vector_idx == kMaxVectorIdx) {
-      s = file_->Append(Slice(unused_bucket));
+      io_status_ = file_->Append(Slice(unused_bucket));
     } else {
       ++num_added;
-      s = file_->Append(GetKey(bucket.vector_idx));
-      if (s.ok()) {
+      io_status_ = file_->Append(GetKey(bucket.vector_idx));
+      if (io_status_.ok()) {
         if (value_size_ > 0) {
-          s = file_->Append(GetValue(bucket.vector_idx));
+          io_status_ = file_->Append(GetValue(bucket.vector_idx));
         }
       }
     }
-    if (!s.ok()) {
-      return s;
+    if (!io_status_.ok()) {
+      status_ = io_status_;
+      return status_;
     }
   }
   assert(num_added == NumEntries());
@@ -364,10 +371,11 @@ Status CuckooTableBuilder::Finish() {
   BlockHandle property_block_handle;
   property_block_handle.set_offset(offset);
   property_block_handle.set_size(property_block.size());
-  s = file_->Append(property_block);
+  io_status_ = file_->Append(property_block);
   offset += property_block.size();
-  if (!s.ok()) {
-    return s;
+  if (!io_status_.ok()) {
+    status_ = io_status_;
+    return status_;
   }
 
   meta_index_builder.Add(kPropertiesBlock, property_block_handle);
@@ -376,9 +384,10 @@ Status CuckooTableBuilder::Finish() {
   BlockHandle meta_index_block_handle;
   meta_index_block_handle.set_offset(offset);
   meta_index_block_handle.set_size(meta_index_block.size());
-  s = file_->Append(meta_index_block);
-  if (!s.ok()) {
-    return s;
+  io_status_ = file_->Append(meta_index_block);
+  if (!io_status_.ok()) {
+    status_ = io_status_;
+    return status_;
   }
 
   Footer footer(kCuckooTableMagicNumber, 1);
@@ -386,12 +395,9 @@ Status CuckooTableBuilder::Finish() {
   footer.set_index_handle(BlockHandle::NullBlockHandle());
   std::string footer_encoding;
   footer.EncodeTo(&footer_encoding);
-  s = file_->Append(footer_encoding);
-
-  if (file_ != nullptr) {
-    file_checksum_ = file_->GetFileChecksum();
-  }
-  return s;
+  io_status_ = file_->Append(footer_encoding);
+  status_ = io_status_;
+  return status_;
 }
 
 void CuckooTableBuilder::Abandon() {
@@ -516,11 +522,19 @@ bool CuckooTableBuilder::MakeSpaceForKey(
   return null_found;
 }
 
+std::string CuckooTableBuilder::GetFileChecksum() const {
+  if (file_ != nullptr) {
+    return file_->GetFileChecksum();
+  } else {
+    return kUnknownFileChecksum;
+  }
+}
+
 const char* CuckooTableBuilder::GetFileChecksumFuncName() const {
   if (file_ != nullptr) {
     return file_->GetFileChecksumFuncName();
   } else {
-    return kUnknownFileChecksumFuncName.c_str();
+    return kUnknownFileChecksumFuncName;
   }
 }