Merge 80d6c09fa4 into ac691084fd

db/db_impl.cc

@@ -1003,7 +1003,12 @@ Status DBImpl::DoCompactionWork(CompactionState* compact) {
         compact->current_output()->smallest.DecodeFrom(key);
       }
       compact->current_output()->largest.DecodeFrom(key);
-      compact->builder->Add(key, input->value());
+
+      status = compact->builder->Add(key, input->value());
+      if (!status.ok()) {
+        FinishCompactionOutputFile(compact, input);
+        break;
+      }
 
       // Close output file if it is big enough
       if (compact->builder->FileSize() >=

include/leveldb/table_builder.h

@@ -49,7 +49,7 @@ class LEVELDB_EXPORT TableBuilder {
   // Add key,value to the table being constructed.
   // REQUIRES: key is after any previously added key according to comparator.
   // REQUIRES: Finish(), Abandon() have not been called
-  void Add(const Slice& key, const Slice& value);
+  Status Add(const Slice& key, const Slice& value);
 
   // Advanced operation: flush any buffered key/value pairs to file.
   // Can be used to ensure that two adjacent entries never live in

table/table_builder.cc

@@ -91,10 +91,12 @@ Status TableBuilder::ChangeOptions(const Options& options) {
   return Status::OK();
 }
 
-void TableBuilder::Add(const Slice& key, const Slice& value) {
+Status TableBuilder::Add(const Slice& key, const Slice& value) {
   Rep* r = rep_;
   assert(!r->closed);
-  if (!ok()) return;
+  if (!ok()) {
+    return status();
+  }
   if (r->num_entries > 0) {
     assert(r->options.comparator->Compare(key, Slice(r->last_key)) > 0);
   }
@@ -120,6 +122,7 @@ void TableBuilder::Add(const Slice& key, const Slice& value) {
   if (estimated_block_size >= r->options.block_size) {
     Flush();
   }
+  return status();
 }
 
 void TableBuilder::Flush() {