enable config reader/writer interval by env variable (#4529)

* support change reader/writer interval by env variable
Signed-off-by: Npeng.xu <peng.xu@zilliz.com>

* optimize db query
Signed-off-by: Npeng.xu <peng.xu@zilliz.com>

core/src/db/meta/MetaAdapter.h

@@ -54,6 +54,14 @@ class MetaAdapter {
         return session->Select<ResourceT, U>(field, values, {}, resources);
     }
 
+    template <typename ResourceT, typename U>
+    Status
+    SelectByAttrs(const std::string& field, const std::vector<U>& values, const std::vector<std::string>& target_attrs,
+                  std::vector<typename ResourceT::Ptr>& resources) {
+        auto session = CreateSession();
+        return session->Select<ResourceT, U>(field, values, target_attrs, resources);
+    }
+
     template <typename ResourceT, typename U>
     Status
     SelectResourceIDs(std::vector<int64_t>& ids, const std::string& filter_field, const std::vector<U>& filter_values) {

core/src/db/snapshot/CompoundOperations.cpp

@@ -18,6 +18,7 @@
 #include <vector>
 
 #include "db/meta/MetaAdapter.h"
+#include "db/meta/MetaResourceAttrs.h"
 #include "db/snapshot/IterateHandler.h"
 #include "db/snapshot/OperationExecutor.h"
 #include "db/snapshot/ResourceContext.h"
@@ -942,7 +943,8 @@ GetAllActiveSnapshotIDsOperation::GetAllActiveSnapshotIDsOperation()
 Status
 GetAllActiveSnapshotIDsOperation::DoExecute(StorePtr store) {
     std::vector<CollectionCommitPtr> ccs;
-    STATUS_CHECK(store->GetActiveResources<CollectionCommit>(ccs));
+    STATUS_CHECK(store->GetActiveResourcesByAttrs<CollectionCommit>(ccs, {meta::F_ID, meta::F_COLLECTON_ID}));
+    /* STATUS_CHECK(store->GetActiveResources<CollectionCommit>(ccs)); */
     for (auto& cc : ccs) {
         auto cid = cc->GetCollectionId();
         auto it = cid_ccid_.find(cid);

core/src/db/snapshot/Snapshots.cpp

@@ -380,12 +380,36 @@ Snapshots::RegisterTimers(TimerManager* mgr) {
     auto role = config.cluster.role();
     if (is_cluster && (role == ClusterRole::RO)) {
         TimerContext::Context ctx;
-        ctx.interval_us = DEFAULT_READER_TIMER_INTERVAL_US;
+        if (const char* env_interval_us = std::getenv("READER_TIMER_INTERVAL_US")) {
+            LOG_SERVER_INFO_ << "Environment READER_TIMER_INTERVAL_US=" << env_interval_us;
+            ctx.interval_us = atoi(env_interval_us);
+        } else {
+            ctx.interval_us = DEFAULT_READER_TIMER_INTERVAL_US;
+        }
+        auto low_limit = DEFAULT_READER_TIMER_INTERVAL_US * 0.6;
+        if (ctx.interval_us < low_limit) {
+            LOG_SERVER_WARNING_ << "Environment READER_TIMER_INTERVAL_US is too small, set it to: " << low_limit;
+            ctx.interval_us = low_limit;
+        }
+        LOG_SERVER_INFO_ << "OnReaderTimer INTERVAL: " << ctx.interval_us << " US";
         ctx.handler = std::bind(&Snapshots::OnReaderTimer, this, std::placeholders::_1);
         mgr->AddTimer(ctx);
     } else {
         TimerContext::Context ctx;
-        ctx.interval_us = DEFAULT_WRITER_TIMER_INTERVAL_US;
+
+        if (const char* env_interval_us = std::getenv("WRITER_TIMER_INTERVAL_US")) {
+            LOG_SERVER_INFO_ << "Environment WRITER_TIMER_INTERVAL_US=" << env_interval_us;
+            ctx.interval_us = atoi(env_interval_us);
+        } else {
+            ctx.interval_us = DEFAULT_WRITER_TIMER_INTERVAL_US;
+        }
+        auto low_limit = DEFAULT_WRITER_TIMER_INTERVAL_US * 0.4;
+        if (ctx.interval_us < low_limit) {
+            LOG_SERVER_WARNING_ << "Environment WRITER_TIMER_INTERVAL_US is too small, set it to: " << low_limit;
+            ctx.interval_us = low_limit;
+        }
+        LOG_SERVER_INFO_ << "OnWriterTimer INTERVAL: " << ctx.interval_us << " US";
+
         ctx.handler = std::bind(&Snapshots::OnWriterTimer, this, std::placeholders::_1);
         mgr->AddTimer(ctx);
     }

core/src/db/snapshot/Store.h

@@ -178,6 +178,14 @@ class Store : public std::enable_shared_from_this<Store> {
         return adapter_->SelectBy<ResourceT>(StateField::Name, filter_states, return_vs);
     }
 
+    template <typename ResourceT>
+    Status
+    GetActiveResourcesByAttrs(std::vector<typename ResourceT::Ptr>& return_vs,
+                              const std::vector<std::string>& target_attrs) {
+        std::vector<State> filter_states = {State::ACTIVE};
+        return adapter_->SelectByAttrs<ResourceT>(StateField::Name, filter_states, target_attrs, return_vs);
+    }
+
     template <typename ResourceT>
     Status
     GetActiveResources(std::vector<typename ResourceT::Ptr>& return_vs) {