Revisit #9118 for compaction outputs (#9480)

Summary:
Crash test recently started showing failures as in https://github.com/facebook/rocksdb/issues/9118 but
for files created by compaction. This change applies a similar fix.

Pull Request resolved: https://github.com/facebook/rocksdb/pull/9480

Test Plan:
Updated / extended unit test. (Some re-arranging to do the
simpler compaction testing before this special case.)

Reviewed By: ltamasi

Differential Revision: D33909835

Pulled By: pdillinger

fbshipit-source-id: 58e4b44e4ecc2d21e4df2c2d8440ec0633aa1f6c

HISTORY.md

@@ -2,6 +2,7 @@
 ## Unreleased
 ### Bug Fixes
 * Fixed a major bug in which batched MultiGet could return old values for keys deleted by DeleteRange when memtable Bloom filter is enabled (memtable_prefix_bloom_size_ratio > 0). (The fix includes a substantial MultiGet performance improvement in the unusual case of both memtable_whole_key_filtering and prefix_extractor.)
+* Fixed more cases of EventListener::OnTableFileCreated called with OK status, file_size==0, and no SST file kept. Now the status is Aborted.
 
 ### Public API changes
 * Remove HDFS support from main repo.

db/compaction/compaction_job.cc

@@ -1944,17 +1944,21 @@ Status CompactionJob::FinishCompactionOutputFile(
   std::string fname;
   FileDescriptor output_fd;
   uint64_t oldest_blob_file_number = kInvalidBlobFileNumber;
+  Status status_for_listener = s;
   if (meta != nullptr) {
     fname = GetTableFileName(meta->fd.GetNumber());
     output_fd = meta->fd;
     oldest_blob_file_number = meta->oldest_blob_file_number;
   } else {
     fname = "(nil)";
+    if (s.ok()) {
+      status_for_listener = Status::Aborted("Empty SST file not kept");
+    }
   }
   EventHelpers::LogAndNotifyTableFileCreationFinished(
       event_logger_, cfd->ioptions()->listeners, dbname_, cfd->GetName(), fname,
       job_id_, output_fd, oldest_blob_file_number, tp,
-      TableFileCreationReason::kCompaction, s, file_checksum,
+      TableFileCreationReason::kCompaction, status_for_listener, file_checksum,
       file_checksum_func_name);
 
 #ifndef ROCKSDB_LITE

db/listener_test.cc

@@ -772,11 +772,13 @@ class TableFileCreationListener : public EventListener {
     ASSERT_EQ(info.file_checksum, kUnknownFileChecksum);
     ASSERT_EQ(info.file_checksum_func_name, kUnknownFileChecksumFuncName);
     if (info.status.ok()) {
+      if (info.table_properties.num_range_deletions == 0U) {
         ASSERT_GT(info.table_properties.data_size, 0U);
         ASSERT_GT(info.table_properties.raw_key_size, 0U);
         ASSERT_GT(info.table_properties.raw_value_size, 0U);
         ASSERT_GT(info.table_properties.num_data_blocks, 0U);
         ASSERT_GT(info.table_properties.num_entries, 0U);
+      }
     } else {
       if (idx >= 0) {
         failure_[idx]++;
@@ -828,14 +830,6 @@ TEST_F(EventListenerTest, TableFileCreationListenersTest) {
   ASSERT_OK(dbfull()->TEST_WaitForCompact());
   listener->CheckAndResetCounters(0, 0, 0, 1, 1, 0);
 
-  // Verify that an empty table file that is immediately deleted gives Aborted
-  // status to listener.
-  ASSERT_OK(Put("baz", "z"));
-  ASSERT_OK(SingleDelete("baz"));
-  ASSERT_OK(Flush());
-  listener->CheckAndResetCounters(1, 1, 1, 0, 0, 0);
-  ASSERT_TRUE(listener->last_failure_.IsAborted());
-
   ASSERT_OK(Put("foo", "aaa3"));
   ASSERT_OK(Put("bar", "bbb3"));
   ASSERT_OK(Flush());
@@ -845,7 +839,31 @@ TEST_F(EventListenerTest, TableFileCreationListenersTest) {
   ASSERT_NOK(dbfull()->TEST_WaitForCompact());
   listener->CheckAndResetCounters(1, 1, 0, 1, 1, 1);
   ASSERT_TRUE(listener->last_failure_.IsNotSupported());
-  Close();
+
+  // Reset
+  test_env->SetStatus(Status::OK());
+  DestroyAndReopen(options);
+
+  // Verify that an empty table file that is immediately deleted gives Aborted
+  // status to listener.
+  ASSERT_OK(Put("baz", "z"));
+  ASSERT_OK(SingleDelete("baz"));
+  ASSERT_OK(Flush());
+  listener->CheckAndResetCounters(1, 1, 1, 0, 0, 0);
+  ASSERT_TRUE(listener->last_failure_.IsAborted());
+
+  // Also in compaction
+  ASSERT_OK(Put("baz", "z"));
+  ASSERT_OK(Flush());
+  ASSERT_OK(db_->DeleteRange(WriteOptions(), db_->DefaultColumnFamily(),
+                             kRangeStart, kRangeEnd));
+  ASSERT_OK(Flush());
+  ASSERT_OK(db_->CompactRange(CompactRangeOptions(), nullptr, nullptr));
+  ASSERT_OK(dbfull()->TEST_WaitForCompact());
+  listener->CheckAndResetCounters(2, 2, 0, 1, 1, 1);
+  ASSERT_TRUE(listener->last_failure_.IsAborted());
+
+  Close();  // Avoid UAF on listener
 }
 
 class MemTableSealedListener : public EventListener {