fix: fix gc coredump (#3561)

Author: dl239

src/storage/mem_table.cc

@@ -170,14 +170,18 @@ bool MemTable::Put(uint64_t time, const std::string& value, const Dimensions& di
         PDLOG(WARNING, "invalid schema version %u, tid %u pid %u", version, id_, pid_);
         return false;
     }
-    std::map<int32_t, uint64_t> ts_map;
+    std::map<uint32_t, std::map<int32_t, uint64_t>> ts_value_map;
     for (const auto& kv : inner_index_key_map) {
         auto inner_index = table_index_.GetInnerIndex(kv.first);
         if (!inner_index) {
             PDLOG(WARNING, "invalid inner index pos %d. tid %u pid %u", kv.first, id_, pid_);
             return false;
         }
+        std::map<int32_t, uint64_t> ts_map;
         for (const auto& index_def : inner_index->GetIndex()) {
+            if (!index_def->IsReady()) {
+                continue;
+            }
             auto ts_col = index_def->GetTsColumn();
             if (ts_col) {
                 int64_t ts = 0;
@@ -192,34 +196,28 @@ bool MemTable::Put(uint64_t time, const std::string& value, const Dimensions& di
                     return false;
                 }
                 ts_map.emplace(ts_col->GetId(), ts);
-            }
-            if (index_def->IsReady()) {
                 real_ref_cnt++;
             }
         }
+        if (!ts_map.empty()) {
+            ts_value_map.emplace(kv.first, std::move(ts_map));
+        }
     }
-    if (ts_map.empty()) {
+    if (ts_value_map.empty()) {
         return false;
     }
     auto* block = new DataBlock(real_ref_cnt, value.c_str(), value.length());
     for (const auto& kv : inner_index_key_map) {
-        auto inner_index = table_index_.GetInnerIndex(kv.first);
-        bool need_put = false;
-        for (const auto& index_def : inner_index->GetIndex()) {
-            if (index_def->IsReady()) {
-                // TODO(hw): if we don't find this ts(has_found_ts==false), but it's ready, will put too?
-                need_put = true;
-                break;
-            }
+        auto iter = ts_value_map.find(kv.first);
+        if (iter == ts_value_map.end()) {
+            continue;
         }
-        if (need_put) {
-            uint32_t seg_idx = 0;
-            if (seg_cnt_ > 1) {
-                seg_idx = ::openmldb::base::hash(kv.second.data(), kv.second.size(), SEED) % seg_cnt_;
-            }
-            Segment* segment = segments_[kv.first][seg_idx];
-            segment->Put(::openmldb::base::Slice(kv.second), ts_map, block);
+        uint32_t seg_idx = 0;
+        if (seg_cnt_ > 1) {
+            seg_idx = ::openmldb::base::hash(kv.second.data(), kv.second.size(), SEED) % seg_cnt_;
         }
+        Segment* segment = segments_[kv.first][seg_idx];
+        segment->Put(::openmldb::base::Slice(kv.second), iter->second, block);
     }
     record_byte_size_.fetch_add(GetRecordSize(value.length()));
     return true;

src/storage/snapshot_test.cc

@@ -718,6 +718,79 @@ TEST_F(SnapshotTest, Recover_only_snapshot) {
     ASSERT_FALSE(it->Valid());
 }
 
+TEST_F(SnapshotTest, RecoverWithDeleteIndex) {
+    uint32_t tid = 12;
+    uint32_t pid = 0;
+    ::openmldb::api::TableMeta meta;
+    meta.set_tid(tid);
+    meta.set_pid(pid);
+    SchemaCodec::SetColumnDesc(meta.add_column_desc(), "userid", ::openmldb::type::kString);
+    SchemaCodec::SetColumnDesc(meta.add_column_desc(), "ts1", ::openmldb::type::kBigInt);
+    SchemaCodec::SetColumnDesc(meta.add_column_desc(), "ts2", ::openmldb::type::kBigInt);
+    SchemaCodec::SetColumnDesc(meta.add_column_desc(), "val", ::openmldb::type::kString);
+    SchemaCodec::SetIndex(meta.add_column_key(), "index1", "userid", "ts1", ::openmldb::type::kLatestTime, 0, 1);
+    SchemaCodec::SetIndex(meta.add_column_key(), "index2", "userid", "ts2", ::openmldb::type::kLatestTime, 0, 1);
+
+    std::string snapshot_dir = absl::StrCat(FLAGS_db_root_path, "/", tid, "_", pid, "/snapshot");
+
+    ::openmldb::base::MkdirRecur(snapshot_dir);
+    std::string snapshot1 = "20231018.sdb";
+    uint64_t offset = 0;
+    {
+        if (FLAGS_snapshot_compression != "off") {
+            snapshot1.append(".");
+            snapshot1.append(FLAGS_snapshot_compression);
+        }
+        std::string full_path = snapshot_dir + "/" + snapshot1;
+        FILE* fd_w = fopen(full_path.c_str(), "ab+");
+        ASSERT_TRUE(fd_w != NULL);
+        ::openmldb::log::WritableFile* wf = ::openmldb::log::NewWritableFile(snapshot1, fd_w);
+        ::openmldb::log::Writer writer(FLAGS_snapshot_compression, wf);
+        ::openmldb::codec::SDKCodec sdk_codec(meta);
+        for (int i = 0; i < 5; i++) {
+            uint32_t ts = 100 + i;
+            for (int key_num = 0; key_num < 10; key_num++) {
+                std::string userid = absl::StrCat("userid", key_num);
+                std::string ts_str = std::to_string(ts);
+                std::vector<std::string> row = {userid, ts_str, ts_str, "aa"};
+                std::string result;
+                sdk_codec.EncodeRow(row, &result);
+                ::openmldb::api::LogEntry entry;
+                entry.set_log_index(offset++);
+                entry.set_value(result);
+                for (int k = 0; k < meta.column_key_size(); k++) {
+                    auto dimension = entry.add_dimensions();
+                    dimension->set_key(userid);
+                    dimension->set_idx(k);
+                }
+                entry.set_ts(ts);
+                entry.set_term(1);
+                std::string val;
+                bool ok = entry.SerializeToString(&val);
+                ASSERT_TRUE(ok);
+                Slice sval(val.c_str(), val.size());
+                ::openmldb::log::Status status = writer.AddRecord(sval);
+                ASSERT_TRUE(status.ok());
+            }
+        }
+        writer.EndLog();
+    }
+
+    auto index1 = meta.mutable_column_key(1);
+    index1->set_flag(1);
+    std::shared_ptr<MemTable> table = std::make_shared<MemTable>(meta);
+    table->Init();
+    LogParts* log_part = new LogParts(12, 4, scmp);
+    MemTableSnapshot snapshot(tid, pid, log_part, FLAGS_db_root_path);
+    ASSERT_TRUE(snapshot.Init());
+    int ret = snapshot.GenManifest(snapshot1, 50, offset, 1);
+    ASSERT_EQ(0, ret);
+    uint64_t r_offset = 0;
+    ASSERT_TRUE(snapshot.Recover(table, r_offset));
+    ASSERT_EQ(r_offset, offset);
+    table->SchedGc();
+}
+
 TEST_F(SnapshotTest, MakeSnapshot) {
     LogParts* log_part = new LogParts(12, 4, scmp);
     MemTableSnapshot snapshot(1, 2, log_part, FLAGS_db_root_path);