Fix issues caused by background index optimize (#1141)

Fix PostingWriter::Dump
Serialize ChunkIndexEntry::deprecate_ts_
Fix MemIndexRecover crash

- [x] Bug Fix (non-breaking change which fixes an issue)

Author: yuzhichang

benchmark/local_infinity/fulltext/fulltext_benchmark.cpp

@@ -301,9 +301,7 @@ void BenchmarkMoreQuery(SharedPtr<Infinity> infinity, const String &db_name, con
     profiler.End();
 }
 
-int main(int argc, char *argn[]) {
-    argc = 3;
-    const char* argv[] = {"fulltext", "--mode", "query"};
+int main(int argc, char *argv[]) {
     CLI::App app{"fulltext_benchmark"};
     // https://github.com/CLIUtils/CLI11/blob/main/examples/enum.cpp
     // Using enumerations in an option

python/hello_infinity.py

@@ -12,7 +12,6 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-import json
 import infinity
 import infinity.index as index
 from infinity.common import REMOTE_HOST
@@ -75,12 +74,6 @@ def test_english():
     except Exception as e:
         print(str(e))
 
-    # Print the JSON data
-    # print("------json-------")
-    # print(json_data)
-    # print("------tabular - querybuilder-------")
-    # print(qb_result)
-
 
 def test_chinese():
     """

src/storage/buffer/file_worker/raw_file_worker.cpp

@@ -25,12 +25,8 @@ import status;
 
 namespace infinity {
 
-RawFileWorker::RawFileWorker(SharedPtr<String> file_dir, SharedPtr<String> file_name)
-    : FileWorker(std::move(file_dir), std::move(file_name)), buffer_size_(0) {
-    LocalFileSystem fs;
-    String full_path = fs.ConcatenateFilePath(*file_dir_, *file_name_);
-    buffer_size_ = fs.GetFileSizeByPath(full_path);
-}
+RawFileWorker::RawFileWorker(SharedPtr<String> file_dir, SharedPtr<String> file_name, u32 file_size)
+    : FileWorker(std::move(file_dir), std::move(file_name)), buffer_size_(file_size) {}
 
 RawFileWorker::~RawFileWorker() {
     if (data_ != nullptr) {

src/storage/buffer/file_worker/raw_file_worker.cppm

@@ -27,7 +27,7 @@ namespace infinity {
 // - The file size is consistant since creation.
 export class RawFileWorker : public FileWorker {
 public:
-    explicit RawFileWorker(SharedPtr<String> file_dir, SharedPtr<String> file_name);
+    explicit RawFileWorker(SharedPtr<String> file_dir, SharedPtr<String> file_name, u32 file_size);
 
     virtual ~RawFileWorker() override;
 

src/storage/invertedindex/column_inverter.cpp

@@ -108,7 +108,7 @@ void ColumnInverter::MergePrepare() {
 }
 
 void ColumnInverter::Merge(ColumnInverter &rhs) {
-    assert(begin_doc_id_ + doc_count_ == rhs.begin_doc_id_);
+    assert(begin_doc_id_ + doc_count_ <= rhs.begin_doc_id_);
     MergePrepare();
     for (auto &doc_terms : rhs.terms_per_doc_) {
         u32 doc_id = doc_terms.first;

src/storage/invertedindex/memory_indexer.cpp

@@ -144,6 +144,14 @@ void MemoryIndexer::Insert(SharedPtr<ColumnVector> column_vector, u32 row_offset
     }
 }
 
+void MemoryIndexer::InsertGap(u32 row_count) {
+    if (is_spilled_)
+        Load();
+
+    std::unique_lock<std::mutex> lock(mutex_);
+    doc_count_ += row_count;
+}
+
 void MemoryIndexer::Commit(bool offline) {
     if (offline) {
         commiting_thread_pool_.push([this](int id) { this->CommitOffline(); });

src/storage/invertedindex/memory_indexer.cppm

@@ -65,6 +65,9 @@ public:
     // Insert is non-blocking. Caller must ensure there's no RowID gap between each call.
     void Insert(SharedPtr<ColumnVector> column_vector, u32 row_offset, u32 row_count, bool offline = false);
 
+    // InsertGap insert some empty documents. This is for abnormal case.
+    void InsertGap(u32 row_count);
+
     // Commit is non-blocking and thread-safe. There shall be a background thread which call this method regularly.
     void Commit(bool offline = false);
 

src/storage/invertedindex/posting_writer.cpp

@@ -60,6 +60,9 @@ tf_t PostingWriter::GetCurrentTF() const { return doc_list_encoder_->GetCurrentT
 void PostingWriter::SetCurrentTF(tf_t tf) { doc_list_encoder_->SetCurrentTF(tf); }
 
 void PostingWriter::Dump(const SharedPtr<FileWriter> &file_writer, TermMeta &term_meta, bool spill) {
+    term_meta.doc_freq_ = GetDF();
+    term_meta.total_tf_ = GetTotalTF();
+    term_meta.payload_ = 0;
     term_meta.doc_start_ = file_writer->TotalWrittenBytes();
     doc_list_encoder_->Dump(file_writer, spill);
     if (position_list_encoder_) {

src/storage/meta/catalog.cpp

@@ -331,8 +331,13 @@ void Catalog::CommitCreateIndex(TxnIndexStore *txn_index_store, TxnTimeStamp com
 //     table_index_entry->RollbackPopulateIndex(txn_index_store, txn);
 // }
 
-void Catalog::Append(TableEntry *table_entry, TransactionID txn_id, void *txn_store, TxnTimeStamp commit_ts, BufferManager *buffer_mgr) {
-    return table_entry->AppendData(txn_id, txn_store, commit_ts, buffer_mgr);
+void Catalog::Append(TableEntry *table_entry,
+                     TransactionID txn_id,
+                     void *txn_store,
+                     TxnTimeStamp commit_ts,
+                     BufferManager *buffer_mgr,
+                     bool is_replay) {
+    return table_entry->AppendData(txn_id, txn_store, commit_ts, buffer_mgr, is_replay);
 }
 
 void Catalog::RollbackAppend(TableEntry *table_entry, TransactionID txn_id, TxnTimeStamp commit_ts, void *txn_store) {

src/storage/meta/catalog.cppm

@@ -198,7 +198,8 @@ public:
     // static void RollbackPopulateIndex(TxnIndexStore *txn_index_store, Txn *txn);
 
     // Append related functions
-    static void Append(TableEntry *table_entry, TransactionID txn_id, void *txn_store, TxnTimeStamp commit_ts, BufferManager *buffer_mgr);
+    static void
+    Append(TableEntry *table_entry, TransactionID txn_id, void *txn_store, TxnTimeStamp commit_ts, BufferManager *buffer_mgr, bool is_replay = false);
 
     static void RollbackAppend(TableEntry *table_entry, TransactionID txn_id, TxnTimeStamp commit_ts, void *txn_store);
 

src/storage/meta/entry/block_entry.cppm

@@ -14,6 +14,8 @@
 
 module;
 
+#include "type/complex/row_id.h"
+
 export module block_entry;
 
 import stl;
@@ -124,6 +126,10 @@ public:
 
     u32 segment_id() const;
 
+    u32 segment_offset() const { return u32(block_id() * row_capacity()); }
+
+    RowID base_row_id() const { return RowID(segment_id(), segment_offset()); }
+
     const SharedPtr<String> &base_dir() const { return block_dir_; }
 
     BlockColumnEntry *GetColumnBlockEntry(SizeT column_id) const { return columns_[column_id].get(); }

src/storage/meta/entry/chunk_index_entry.cpp

@@ -106,7 +106,7 @@ SharedPtr<ChunkIndexEntry> ChunkIndexEntry::NewFtChunkIndexEntry(SegmentIndexEnt
     assert(index_dir.get() != nullptr);
     if (buffer_mgr != nullptr) {
         auto column_length_file_name = MakeShared<String>(base_name + LENGTH_SUFFIX);
-        auto file_worker = MakeUnique<RawFileWorker>(index_dir, column_length_file_name);
+        auto file_worker = MakeUnique<RawFileWorker>(index_dir, column_length_file_name, row_count * sizeof(u32));
         chunk_index_entry->buffer_obj_ = buffer_mgr->GetBufferObject(std::move(file_worker));
     }
     return chunk_index_entry;
@@ -123,7 +123,7 @@ SharedPtr<ChunkIndexEntry> ChunkIndexEntry::NewReplayChunkIndexEntry(ChunkID chu
     if (param->index_base_->index_type_ == IndexType::kFullText) {
         const auto &index_dir = segment_index_entry->index_dir();
         auto column_length_file_name = MakeShared<String>(base_name + LENGTH_SUFFIX);
-        auto file_worker = MakeUnique<RawFileWorker>(index_dir, column_length_file_name);
+        auto file_worker = MakeUnique<RawFileWorker>(index_dir, column_length_file_name, row_count * sizeof(u32));
         chunk_index_entry->buffer_obj_ = buffer_mgr->GetBufferObject(std::move(file_worker));
     } else {
         const auto &index_dir = segment_index_entry->index_dir();
@@ -145,6 +145,7 @@ nlohmann::json ChunkIndexEntry::Serialize() {
     index_entry_json["base_rowid"] = this->base_rowid_.ToUint64();
     index_entry_json["row_count"] = this->row_count_;
     index_entry_json["commit_ts"] = this->commit_ts_.load();
+    index_entry_json["deprecate_ts_"] = this->deprecate_ts_.load();
     return index_entry_json;
 }
 
@@ -158,6 +159,7 @@ SharedPtr<ChunkIndexEntry> ChunkIndexEntry::Deserialize(const nlohmann::json &in
     u32 row_count = index_entry_json["row_count"];
     auto ret = NewReplayChunkIndexEntry(chunk_id, segment_index_entry, param, base_name, base_rowid, row_count, buffer_mgr);
     ret->commit_ts_.store(index_entry_json["commit_ts"]);
+    ret->deprecate_ts_.store(index_entry_json["deprecate_ts_"]);
     return ret;
 }
 

src/storage/meta/entry/chunk_index_entry.cppm

@@ -15,6 +15,7 @@
 module;
 
 #include "type/complex/row_id.h"
+#include <cassert>
 
 export module chunk_index_entry;
 
@@ -89,11 +90,15 @@ public:
 
     BufferObj *GetBufferObj() { return buffer_obj_; }
 
-    void DeprecateChunk(TxnTimeStamp commit_ts) { deprecate_ts_.store(commit_ts); }
+    void DeprecateChunk(TxnTimeStamp commit_ts) {
+        assert(commit_ts_.load() < commit_ts);
+        deprecate_ts_.store(commit_ts);
+    }
 
     bool CheckVisible(TxnTimeStamp ts) {
         TxnTimeStamp deprecate_ts = deprecate_ts_.load();
         TxnTimeStamp commit_ts = commit_ts_.load();
+        assert(commit_ts == UNCOMMIT_TS || commit_ts < deprecate_ts);
         return ts >= commit_ts && ts <= deprecate_ts;
     }
 

src/storage/meta/entry/segment_index_entry.cpp

@@ -241,10 +241,8 @@ void SegmentIndexEntry::MemIndexInsert(SharedPtr<BlockEntry> block_entry,
                                        u32 row_count,
                                        TxnTimeStamp commit_ts,
                                        BufferManager *buffer_manager) {
-    u32 seg_id = block_entry->segment_id();
-    u16 block_id = block_entry->block_id();
-    SegmentOffset block_offset = u32(block_id) * block_entry->row_capacity();
-    RowID begin_row_id(seg_id, row_offset + block_offset);
+    SegmentOffset block_offset = block_entry->segment_offset();
+    RowID begin_row_id = block_entry->base_row_id() + row_offset;
 
     const SharedPtr<IndexBase> &index_base = table_index_entry_->table_index_def();
     const SharedPtr<ColumnDef> &column_def = table_index_entry_->column_def();
@@ -267,7 +265,15 @@ void SegmentIndexEntry::MemIndexInsert(SharedPtr<BlockEntry> block_entry,
                                                             table_index_entry_->GetFulltextCommitingThreadPool());
                 table_index_entry_->UpdateFulltextSegmentTs(commit_ts);
             } else {
-                assert(begin_row_id == memory_indexer_->GetBaseRowId() + memory_indexer_->GetDocCount());
+                RowID exp_begin_row_id = memory_indexer_->GetBaseRowId() + memory_indexer_->GetDocCount();
+                assert(begin_row_id >= exp_begin_row_id);
+                if (begin_row_id > exp_begin_row_id) {
+                    LOG_WARN(fmt::format("Begin row id: {}, expect begin row id: {}, insert gap: {}",
+                                         begin_row_id.ToUint64(),
+                                         exp_begin_row_id.ToUint64(),
+                                         begin_row_id - exp_begin_row_id));
+                    memory_indexer_->InsertGap(begin_row_id - exp_begin_row_id);
+                }
             }
             BlockColumnEntry *block_column_entry = block_entry->GetColumnBlockEntry(column_id);
             SharedPtr<ColumnVector> column_vector = MakeShared<ColumnVector>(block_column_entry->GetColumnVector(buffer_manager));
@@ -408,6 +414,17 @@ void SegmentIndexEntry::PopulateEntirely(const SegmentEntry *segment_entry, Txn
             auto block_entry_iter = BlockEntryIter(segment_entry);
             for (const auto *block_entry = block_entry_iter.Next(); block_entry != nullptr; block_entry = block_entry_iter.Next()) {
                 BlockColumnEntry *block_column_entry = block_entry->GetColumnBlockEntry(column_id);
+                RowID begin_row_id = block_entry->base_row_id();
+                RowID exp_begin_row_id = memory_indexer_->GetBaseRowId() + memory_indexer_->GetDocCount();
+                assert(begin_row_id >= exp_begin_row_id);
+                if (begin_row_id > exp_begin_row_id) {
+                    LOG_WARN(fmt::format("Begin row id: {}, expect begin row id: {}, insert gap: {}",
+                                         begin_row_id.ToUint64(),
+                                         exp_begin_row_id.ToUint64(),
+                                         begin_row_id - exp_begin_row_id));
+                    memory_indexer_->InsertGap(begin_row_id - exp_begin_row_id);
+                }
+
                 SharedPtr<ColumnVector> column_vector = MakeShared<ColumnVector>(block_column_entry->GetColumnVector(buffer_mgr));
                 memory_indexer_->Insert(column_vector, 0, block_entry->row_count(), true);
                 memory_indexer_->Commit(true);

src/storage/meta/entry/segment_index_entry.cppm

@@ -145,7 +145,7 @@ public:
         std::sort(std::begin(chunk_index_entries),
                   std::end(chunk_index_entries),
                   [](const SharedPtr<ChunkIndexEntry> &lhs, const SharedPtr<ChunkIndexEntry> &rhs) noexcept {
-                      return lhs->base_rowid_ < rhs->base_rowid_;
+                      return (lhs->base_rowid_ < rhs->base_rowid_ || (lhs->base_rowid_ == rhs->base_rowid_ && lhs->row_count_ < rhs->row_count_));
                   });
     }
 

src/storage/meta/entry/table_entry.cpp

@@ -355,7 +355,7 @@ void TableEntry::AddCompactNew(SharedPtr<SegmentEntry> segment_entry) {
     }
 }
 
-void TableEntry::AppendData(TransactionID txn_id, void *txn_store, TxnTimeStamp commit_ts, BufferManager *buffer_mgr) {
+void TableEntry::AppendData(TransactionID txn_id, void *txn_store, TxnTimeStamp commit_ts, BufferManager *buffer_mgr, bool is_replay) {
     SizeT row_count = 0;
 
     // Read-only no lock needed.
@@ -402,8 +402,11 @@ void TableEntry::AppendData(TransactionID txn_id, void *txn_store, TxnTimeStamp
         row_count += actual_appended;
     }
 
-    // Realtime index insertion
-    MemIndexInsert(txn, append_state_ptr->append_ranges_);
+    // Needn't inserting into MemIndex since MemIndexRecover is responsible for recovering MemIndex
+    if (!is_replay) {
+        // Realtime index insertion.
+        MemIndexInsert(txn, append_state_ptr->append_ranges_);
+    }
 
     this->row_count_ += row_count;
 }
@@ -665,6 +668,7 @@ void TableEntry::MemIndexInsertInner(TableIndexEntry *table_index_entry, Txn *tx
         if (i == dump_idx && segment_index_entry->MemIndexRowCount() >= 1000000) {
             SharedPtr<ChunkIndexEntry> chunk_index_entry = segment_index_entry->MemIndexDump();
             if (chunk_index_entry.get() != nullptr) {
+                chunk_index_entry->Commit(txn->CommitTS());
                 txn_table_store->AddChunkIndexStore(table_index_entry, chunk_index_entry.get());
 
                 if (index_base->index_type_ == IndexType::kFullText) {
@@ -685,6 +689,7 @@ void TableEntry::MemIndexDump(Txn *txn, bool spill) {
         TxnIndexStore *txn_index_store = txn_table_store->GetIndexStore(table_index_entry);
         SharedPtr<ChunkIndexEntry> chunk_index_entry = table_index_entry->MemIndexDump(txn_index_store, spill);
         if (chunk_index_entry.get() != nullptr) {
+            chunk_index_entry->Commit(txn->CommitTS());
             txn_table_store->AddChunkIndexStore(table_index_entry, chunk_index_entry.get());
             table_index_entry->UpdateFulltextSegmentTs(txn->CommitTS());
         }
@@ -737,7 +742,12 @@ void TableEntry::MemIndexRecover(BufferManager *buffer_manager) {
                 append_ranges.emplace_back(segment_id, block_id, start_offset, row_count);
                 for (SizeT i = block_id + 1; i < block_entries.size(); i++) {
                     assert(block_entries[i - 1]->row_capacity() == block_capacity);
-                    assert(block_entries[i - 1]->GetAvailableCapacity() <= 0);
+                    if (block_entries[i - 1]->GetAvailableCapacity() > 0) {
+                        LOG_ERROR(fmt::format("MemIndexRecover got a non-full BlockEntry. block_id {}, row_count {}, block_entries {}",
+                                              block_entries[i - 1]->block_id(),
+                                              block_entries[i - 1]->row_count(),
+                                              block_entries.size()));
+                    }
                     append_ranges.emplace_back(segment_id, i, 0, block_entries[i]->row_count());
                 }
             }

src/storage/meta/entry/table_entry.cppm

@@ -150,7 +150,7 @@ public:
 
     void AddCompactNew(SharedPtr<SegmentEntry> segment_entry);
 
-    void AppendData(TransactionID txn_id, void *txn_store, TxnTimeStamp commit_ts, BufferManager *buffer_mgr);
+    void AppendData(TransactionID txn_id, void *txn_store, TxnTimeStamp commit_ts, BufferManager *buffer_mgr, bool is_replay = false);
 
     void RollbackAppend(TransactionID txn_id, TxnTimeStamp commit_ts, void *txn_store);
 

src/storage/txn/txn_store.cpp

@@ -100,7 +100,11 @@ void TxnIndexStore::Commit(TransactionID txn_id, TxnTimeStamp commit_ts) const {
         segment_index_entry->CommitSegmentIndex(txn_id, commit_ts);
     }
     for (auto *chunk_index_entry : chunk_index_entries_) {
-        chunk_index_entry->Commit(commit_ts);
+        // uncommitted: insert or import, create index
+        // committed: create index, insert or import
+        if (!chunk_index_entry->Committed()) {
+            chunk_index_entry->Commit(commit_ts);
+        }
     }
 }
 

src/storage/wal/wal_manager.cpp

@@ -779,7 +779,7 @@ void WalManager::WalCmdAppendReplay(const WalCmdAppend &cmd, TransactionID txn_i
     table_store->append_state_ = std::move(append_state);
 
     fake_txn->FakeCommit(commit_ts);
-    Catalog::Append(table_store->table_entry_, fake_txn->TxnID(), table_store, commit_ts, storage_->buffer_manager());
+    Catalog::Append(table_store->table_entry_, fake_txn->TxnID(), table_store, commit_ts, storage_->buffer_manager(), true);
     Catalog::CommitWrite(table_store->table_entry_, fake_txn->TxnID(), commit_ts, table_store->txn_segments());
 }