import quincy beta 17.1.0

[ceph.git] / ceph / src / rocksdb / table / plain / plain_table_key_coding.cc

diff --git a/ceph/src/rocksdb/table/plain/plain_table_key_coding.cc b/ceph/src/rocksdb/table/plain/plain_table_key_coding.cc
index d82b969ba25fe5c463a039c7bdb859cd930a7e87..e3a76f89ea26d96effb50c853e753babab2125b1 100644 (file)
--- a/ceph/src/rocksdb/table/plain/plain_table_key_coding.cc
+++ b/ceph/src/rocksdb/table/plain/plain_table_key_coding.cc
@@ -80,13 +80,15 @@ inline Status PlainTableKeyDecoder::DecodeSize(uint32_t start_offset,
   }
 }
 
-Status PlainTableKeyEncoder::AppendKey(const Slice& key,
-                                       WritableFileWriter* file,
-                                       uint64_t* offset, char* meta_bytes_buf,
-                                       size_t* meta_bytes_buf_size) {
+IOStatus PlainTableKeyEncoder::AppendKey(const Slice& key,
+                                         WritableFileWriter* file,
+                                         uint64_t* offset, char* meta_bytes_buf,
+                                         size_t* meta_bytes_buf_size) {
   ParsedInternalKey parsed_key;
-  if (!ParseInternalKey(key, &parsed_key)) {
-    return Status::Corruption(Slice());
+  Status pik_status =
+      ParseInternalKey(key, &parsed_key, false /* log_err_key */);  // TODO
+  if (!pik_status.ok()) {
+    return IOStatus::Corruption(pik_status.getState());
   }
 
   Slice key_to_write = key;  // Portion of internal key to write out.
@@ -99,9 +101,9 @@ Status PlainTableKeyEncoder::AppendKey(const Slice& key,
       char* ptr = EncodeVarint32(key_size_buf, user_key_size);
       assert(ptr <= key_size_buf + sizeof(key_size_buf));
       auto len = ptr - key_size_buf;
-      Status s = file->Append(Slice(key_size_buf, len));
-      if (!s.ok()) {
-        return s;
+      IOStatus io_s = file->Append(Slice(key_size_buf, len));
+      if (!io_s.ok()) {
+        return io_s;
       }
       *offset += len;
     }
@@ -117,9 +119,9 @@ Status PlainTableKeyEncoder::AppendKey(const Slice& key,
       key_count_for_prefix_ = 1;
       pre_prefix_.SetUserKey(prefix);
       size_bytes_pos += EncodeSize(kFullKey, user_key_size, size_bytes);
-      Status s = file->Append(Slice(size_bytes, size_bytes_pos));
-      if (!s.ok()) {
-        return s;
+      IOStatus io_s = file->Append(Slice(size_bytes, size_bytes_pos));
+      if (!io_s.ok()) {
+        return io_s;
       }
       *offset += size_bytes_pos;
     } else {
@@ -135,9 +137,9 @@ Status PlainTableKeyEncoder::AppendKey(const Slice& key,
           static_cast<uint32_t>(pre_prefix_.GetUserKey().size());
       size_bytes_pos += EncodeSize(kKeySuffix, user_key_size - prefix_len,
                                    size_bytes + size_bytes_pos);
-      Status s = file->Append(Slice(size_bytes, size_bytes_pos));
-      if (!s.ok()) {
-        return s;
+      IOStatus io_s = file->Append(Slice(size_bytes, size_bytes_pos));
+      if (!io_s.ok()) {
+        return io_s;
       }
       *offset += size_bytes_pos;
       key_to_write = Slice(key.data() + prefix_len, key.size() - prefix_len);
@@ -149,20 +151,23 @@ Status PlainTableKeyEncoder::AppendKey(const Slice& key,
   // If the row is of value type with seqId 0, flush the special flag together
   // in this buffer to safe one file append call, which takes 1 byte.
   if (parsed_key.sequence == 0 && parsed_key.type == kTypeValue) {
-    Status s =
+    IOStatus io_s =
         file->Append(Slice(key_to_write.data(), key_to_write.size() - 8));
-    if (!s.ok()) {
-      return s;
+    if (!io_s.ok()) {
+      return io_s;
     }
     *offset += key_to_write.size() - 8;
     meta_bytes_buf[*meta_bytes_buf_size] = PlainTableFactory::kValueTypeSeqId0;
     *meta_bytes_buf_size += 1;
   } else {
-    file->Append(key_to_write);
+    IOStatus io_s = file->Append(key_to_write);
+    if (!io_s.ok()) {
+      return io_s;
+    }
     *offset += key_to_write.size();
   }
 
-  return Status::OK();
+  return IOStatus::OK();
 }
 
 Slice PlainTableFileReader::GetFromBuffer(Buffer* buffer, uint32_t file_offset,
@@ -207,8 +212,9 @@ bool PlainTableFileReader::ReadNonMmap(uint32_t file_offset, uint32_t len,
     new_buffer->buf_len = 0;
   }
   Slice read_result;
-  Status s = file_info_->file->Read(file_offset, size_to_read, &read_result,
-                                    new_buffer->buf.get());
+  Status s =
+      file_info_->file->Read(IOOptions(), file_offset, size_to_read,
+                             &read_result, new_buffer->buf.get(), nullptr);
   if (!s.ok()) {
     status_ = s;
     return false;
@@ -275,9 +281,12 @@ Status PlainTableKeyDecoder::ReadInternalKey(
       return file_reader_.status();
     }
     *internal_key_valid = true;
-    if (!ParseInternalKey(*internal_key, parsed_key)) {
+    Status pik_status = ParseInternalKey(*internal_key, parsed_key,
+                                         false /* log_err_key */);  // TODO
+    if (!pik_status.ok()) {
       return Status::Corruption(
-          Slice("Incorrect value type found when reading the next key"));
+          Slice("Corrupted key found during next key read. "),
+          pik_status.getState());
     }
     *bytes_read += user_key_size + 8;
   }
@@ -483,7 +492,6 @@ Status PlainTableKeyDecoder::NextKeyNoValue(uint32_t start_offset,
   if (seekable != nullptr) {
     *seekable = true;
   }
-  Status s;
   if (encoding_type_ == kPlain) {
     return NextPlainEncodingKey(start_offset, parsed_key, internal_key,
                                 bytes_read, seekable);