Add AddFile() InternalStats for Total files/L0 files/total keys ingested

Summary:
Report more information about the ingested files in CF InternalStats
- Total files
- Total L0 files
- Total keys

There was also noticed that we were reporting files that failed to ingest, fix this bug

Test Plan: print stats in tests

Reviewers: sdong, andrewkr, lightmark

Reviewed By: lightmark

Subscribers: jkedgar, andrewkr, dhruba, yoshinorim

Differential Revision: https://reviews.facebook.net/D63039

db/db_impl_add_file.cc

@@ -303,16 +303,33 @@ Status DBImpl::AddFile(ColumnFamilyHandle* column_family,
     if (status.ok()) {
       delete InstallSuperVersionAndScheduleWork(cfd, nullptr,
                                                 mutable_cf_options);
+
+      // Update internal stats for new ingested files
+      uint64_t total_keys = 0;
+      uint64_t total_l0_files = 0;
+      for (size_t i = 0; i < num_files; i++) {
+        InternalStats::CompactionStats stats(1);
+        stats.micros = micro_list[i];
+        stats.bytes_written = meta_list[i].fd.GetFileSize();
+        stats.num_output_files = 1;
+        cfd->internal_stats()->AddCompactionStats(target_level_list[i], stats);
+        cfd->internal_stats()->AddCFStats(
+            InternalStats::BYTES_INGESTED_ADD_FILE,
+            meta_list[i].fd.GetFileSize());
+        total_keys += file_info_list[i].num_entries;
+        if (target_level_list[i] == 0) {
+          total_l0_files += 1;
+        }
+      }
+      cfd->internal_stats()->AddCFStats(InternalStats::INGESTED_NUM_KEYS_TOTAL,
+                                        total_keys);
+      cfd->internal_stats()->AddCFStats(InternalStats::INGESTED_NUM_FILES_TOTAL,
+                                        num_files);
+      cfd->internal_stats()->AddCFStats(
+          InternalStats::INGESTED_LEVEL0_NUM_FILES_TOTAL, total_l0_files);
     }
+
     for (size_t i = 0; i < num_files; i++) {
-      // Update internal stats
-      InternalStats::CompactionStats stats(1);
-      stats.micros = micro_list[i];
-      stats.bytes_written = meta_list[i].fd.GetFileSize();
-      stats.num_output_files = 1;
-      cfd->internal_stats()->AddCompactionStats(target_level_list[i], stats);
-      cfd->internal_stats()->AddCFStats(InternalStats::BYTES_INGESTED_ADD_FILE,
-                                        meta_list[i].fd.GetFileSize());
       ReleaseFileNumberFromPendingOutputs(
           pending_outputs_inserted_elem_list[i]);
     }

db/internal_stats.cc

@@ -769,10 +769,13 @@ void InternalStats::DumpCFStats(std::string* value) {
       value->append(buf);
     }
   }
-
   uint64_t flush_ingest = cf_stats_value_[BYTES_FLUSHED];
   uint64_t add_file_ingest = cf_stats_value_[BYTES_INGESTED_ADD_FILE];
   uint64_t curr_ingest = flush_ingest + add_file_ingest;
+  uint64_t ingest_files_addfile = cf_stats_value_[INGESTED_NUM_FILES_TOTAL];
+  uint64_t ingest_l0_files_addfile =
+      cf_stats_value_[INGESTED_LEVEL0_NUM_FILES_TOTAL];
+  uint64_t ingest_keys_addfile = cf_stats_value_[INGESTED_NUM_KEYS_TOTAL];
   // Cumulative summary
   double w_amp = stats_sum.bytes_written / static_cast<double>(curr_ingest + 1);
   uint64_t total_stall_count =
@@ -790,7 +793,7 @@ void InternalStats::DumpCFStats(std::string* value) {
   uint64_t interval_flush_ingest =
       flush_ingest - cf_stats_snapshot_.ingest_bytes_flush;
   uint64_t interval_add_file_inget =
-      add_file_ingest - cf_stats_snapshot_.ingest_bytes_add_file;
+      add_file_ingest - cf_stats_snapshot_.ingest_bytes_addfile;
   uint64_t interval_ingest =
       interval_flush_ingest + interval_add_file_inget + 1;
   CompactionStats interval_stats(stats_sum);
@@ -804,13 +807,33 @@ void InternalStats::DumpCFStats(std::string* value) {
   snprintf(buf, sizeof(buf), "Uptime(secs): %.1f total, %.1f interval\n",
            seconds_up, interval_seconds_up);
   value->append(buf);
-
   snprintf(buf, sizeof(buf), "Flush(GB): cumulative %.3f, interval %.3f\n",
            flush_ingest / kGB, interval_flush_ingest / kGB);
+  value->append(buf);
   snprintf(buf, sizeof(buf), "AddFile(GB): cumulative %.3f, interval %.3f\n",
            add_file_ingest / kGB, interval_add_file_inget / kGB);
   value->append(buf);
 
+  uint64_t interval_ingest_files_addfile =
+      ingest_files_addfile - cf_stats_snapshot_.ingest_files_addfile;
+  snprintf(buf, sizeof(buf),
+           "AddFile(Total Files): cumulative %lu, interval %lu\n",
+           ingest_files_addfile, interval_ingest_files_addfile);
+  value->append(buf);
+
+  uint64_t interval_ingest_l0_files_addfile =
+      ingest_l0_files_addfile - cf_stats_snapshot_.ingest_l0_files_addfile;
+  snprintf(buf, sizeof(buf),
+           "AddFile(L0 Files): cumulative %lu, interval %lu\n",
+           ingest_l0_files_addfile, interval_ingest_l0_files_addfile);
+  value->append(buf);
+
+  uint64_t interval_ingest_keys_addfile =
+      ingest_keys_addfile - cf_stats_snapshot_.ingest_keys_addfile;
+  snprintf(buf, sizeof(buf), "AddFile(Keys): cumulative %lu, interval %lu\n",
+           ingest_keys_addfile, interval_ingest_keys_addfile);
+  value->append(buf);
+
   // Compact
   uint64_t compact_bytes_read = 0;
   uint64_t compact_bytes_write = 0;
@@ -881,7 +904,10 @@ void InternalStats::DumpCFStats(std::string* value) {
   value->append(buf);
 
   cf_stats_snapshot_.ingest_bytes_flush = flush_ingest;
-  cf_stats_snapshot_.ingest_bytes_add_file = add_file_ingest;
+  cf_stats_snapshot_.ingest_bytes_addfile = add_file_ingest;
+  cf_stats_snapshot_.ingest_files_addfile = ingest_files_addfile;
+  cf_stats_snapshot_.ingest_l0_files_addfile = ingest_l0_files_addfile;
+  cf_stats_snapshot_.ingest_keys_addfile = ingest_keys_addfile;
   cf_stats_snapshot_.comp_stats = stats_sum;
   cf_stats_snapshot_.stall_count = total_stall_count;
 }

db/internal_stats.h

@@ -60,6 +60,9 @@ class InternalStats {
     WRITE_STALLS_ENUM_MAX,
     BYTES_FLUSHED,
     BYTES_INGESTED_ADD_FILE,
+    INGESTED_NUM_FILES_TOTAL,
+    INGESTED_LEVEL0_NUM_FILES_TOTAL,
+    INGESTED_NUM_KEYS_TOTAL,
     INTERNAL_CF_STATS_ENUM_MAX,
   };
 
@@ -246,7 +249,6 @@ class InternalStats {
     // ColumnFamily-level stats
     CompactionStats comp_stats;
     uint64_t ingest_bytes_flush;      // Bytes written to L0 (Flush)
-    uint64_t ingest_bytes_add_file;   // Bytes written to L0 (AddFile)
     uint64_t stall_count;             // Stall count
     // Stats from compaction jobs - bytes written, bytes read, duration.
     uint64_t compact_bytes_write;
@@ -254,15 +256,24 @@ class InternalStats {
     uint64_t compact_micros;
     double seconds_up;
 
+    // AddFile specific stats
+    uint64_t ingest_bytes_addfile;     // Total Bytes ingested
+    uint64_t ingest_files_addfile;     // Total number of files ingested
+    uint64_t ingest_l0_files_addfile;  // Total number of files ingested to L0
+    uint64_t ingest_keys_addfile;      // Total number of keys ingested
+
     CFStatsSnapshot()
         : comp_stats(0),
           ingest_bytes_flush(0),
-          ingest_bytes_add_file(0),
           stall_count(0),
           compact_bytes_write(0),
           compact_bytes_read(0),
           compact_micros(0),
-          seconds_up(0) {}
+          seconds_up(0),
+          ingest_bytes_addfile(0),
+          ingest_files_addfile(0),
+          ingest_l0_files_addfile(0),
+          ingest_keys_addfile(0) {}
   } cf_stats_snapshot_;
 
   struct DBStatsSnapshot {
@@ -376,6 +387,9 @@ class InternalStats {
     WRITE_STALLS_ENUM_MAX,
     BYTES_FLUSHED,
     BYTES_INGESTED_ADD_FILE,
+    INGESTED_NUM_FILES_TOTAL,
+    INGESTED_LEVEL0_NUM_FILES_TOTAL,
+    INGESTED_NUM_KEYS_TOTAL,
     INTERNAL_CF_STATS_ENUM_MAX,
   };