[feature] travse all sst in dumpstatus thread, mark the sst and try to trigger...

[feature] travse all sst in dumpstatus thread, mark the sst and try to trigger Compaction ; add unit test to prove the mark logic is correct #5

db/builder.cc

@@ -414,8 +414,8 @@ Status BuildTable(
                                     : TablePropertyCache::kNoRangeDeletions;
       sst_meta()->prop.flags |=
           tp.snapshots.empty() ? 0 : TablePropertyCache::kHasSnapshots;
-      sst_meta()->prop.ratio_expire_time = tp.ratio_expire_time;
-      sst_meta()->prop.scan_gap_expire_time = tp.scan_gap_expire_time;
+      sst_meta()->prop.ratio_expire_time = builder->GetTableProperties().ratio_expire_time;
+      sst_meta()->prop.scan_gap_expire_time = builder->GetTableProperties().scan_gap_expire_time;
     }
 
     if (s.ok() && !empty) {

db/db_impl.cc

@@ -123,6 +123,7 @@
 #ifdef __GNUC__
 #pragma GCC diagnostic pop
 #endif
+#include <iostream>
 
 #include "rocksdb/terark_namespace.h"
 namespace TERARKDB_NAMESPACE {
@@ -772,26 +773,31 @@ void DBImpl::StartTimedTasks() {
   }
 }
 void DBImpl::ScheduleGCTTL() {
+  TEST_SYNC_POINT("DBImpl:ScheduleGCTTL");
   uint64_t mark_count = 0;
   uint64_t marked_count = 0;
   uint64_t nowSeconds = env_->NowMicros() / 1000000;
   auto should_marked_for_compacted = [](uint64_t ratio_expire_time,
                                         uint64_t scan_gap_expire_time,
                                         uint64_t now) {
-    return (std::min(ratio_expire_time, scan_gap_expire_time) < now);
+    return (std::min(ratio_expire_time, scan_gap_expire_time) <= now);
   };
   ROCKS_LOG_INFO(immutable_db_options_.info_log, "Start ScheduleGCTTL");
   for (auto cfd : *versions_->GetColumnFamilySet()) {
+//    if(cfd->GetLatestCFOptions().ttl_extractor_factory == nullptr) continue;
     if (cfd->initialized()) {
       VersionStorageInfo* vsi = cfd->current()->storage_info();
       for (int l = 0; l < vsi->num_levels(); l++) {
         for (auto sst : vsi->LevelFiles(l)) {
+
           if (sst->marked_for_compaction) marked_count++;
           if (!sst->marked_for_compaction)
             sst->marked_for_compaction = should_marked_for_compacted(
                 sst->prop.ratio_expire_time, sst->prop.scan_gap_expire_time,
                 nowSeconds);
           if (sst->marked_for_compaction) {
+            std::cout << sst->marked_for_compaction << std::endl ;
+            TEST_SYNC_POINT("DBImpl:ScheduleGCTTL-mark");
             mark_count++;
           }
         }
@@ -801,7 +807,7 @@ void DBImpl::ScheduleGCTTL() {
   ROCKS_LOG_INFO(immutable_db_options_.info_log, "marked for compact SST: %d,%d",
                  marked_count,mark_count);
   if (mark_count > 0) {
-    MaybeScheduleFlushOrCompaction();
+//    MaybeScheduleFlushOrCompaction();
   }
 }
 void DBImpl::DumpStats() {

db/db_impl_gc_ttl_test.cc

 // Copyright (c) 2020-present, Bytedance Inc.  All rights reserved.
 // This source code is licensed under Apache 2.0 License.
 
-#include "table/block_based_table_builder.h"
-
-#include "include/rocksdb/ttl_extractor.h"
 #include "util/testharness.h"
 #include "util/testutil.h"
+#include "util/string_util.h"
+#include "db/db_impl.h"
+#include "util/sync_point.h"
+#include "db/db_test_util.h"
+
+namespace rocksdb {
 
-namespace rocksdb{
+class DBImplGCTTL_Test : public DBTestBase {
+ public:
+  DBImplGCTTL_Test() : DBTestBase("/db_GC_ttl_test"){
+  }
+  void init() {
+    rocksdb::SyncPoint::GetInstance()->SetCallBack(
+        "DBImpl:ScheduleGCTTL",
+        [&](void* /*arg*/) { mark = 0;flag = true;});
+    rocksdb::SyncPoint::GetInstance()->SetCallBack(
+        "DBImpl:ScheduleGCTTL-mark",
+        [&](void* /*arg*/) { mark++;});
+    SyncPoint::GetInstance()->EnableProcessing();
 
-class DBImplGCTTL_Test : public testing::Test {
-  //
+    dbname = test::PerThreadDBPath("ttl_gc_test");
+    ASSERT_OK(DestroyDB(dbname, options));
 
- private:
+    options.create_if_missing = true;
+    options.ttl_garbage_collection_percentage = 50.0;
+    options.ttl_scan_gap = 10;
+    options.ttl_extractor_factory.reset(new test::TestTtlExtractorFactory());
+    options.level0_file_num_compaction_trigger = 8;
+    options.stats_dump_period_sec = 10;
+    options.table_factory.reset(new BlockBasedTableFactory(BlockBasedTableOptions()));
 
+  }
+
+ protected:
+  Options options;
+  std::string dbname;
+  bool flag = false;
+  int mark = 0;
 };
 
-TEST_F(DBImplGCTTL_Test,OpenTest) {
+TEST_F(DBImplGCTTL_Test, L0FileExpiredTest) {
+  init();
+  std::unique_ptr<rocksdb::MockTimeEnv> mock_env;
+  mock_env.reset(new rocksdb::MockTimeEnv(env_));
+  mock_env->set_current_time(0); // in seconds
+  options.env = mock_env.get();
+  Reopen(options);
+
+  int L0FilesNums = 4;
+  uint64_t ttl = 200;
+  char ts_string[8];
+  EncodeFixed64(ts_string, ttl);
+  int KeyEntrys = 800;
+  for(int i = 0; i < L0FilesNums; i++){
+    for(int j = 0 ; j < KeyEntrys;j++ ){
+      std::string key = "key";
+      std::string value = "value";
+      AppendNumberTo(&key,j);
+      AppendNumberTo(&value,j);
+      value.append(ts_string,8);
+      dbfull()->Put(WriteOptions(),key,value);
+    }
+    dbfull()->Flush(FlushOptions());
+  }
+
+  ASSERT_EQ(10, dbfull()->GetDBOptions().stats_dump_period_sec);
+  dbfull()->TEST_WaitForTimedTaskRun([&] { mock_env->set_current_time(ttl); });
+  ASSERT_TRUE(flag);
+  ASSERT_EQ(L0FilesNums,mark);
+
 
 }
-}
+}  // namespace rocksdb
 int main(int argc, char** argv) {
   ::testing::InitGoogleTest(&argc, argv);
   return RUN_ALL_TESTS();