Refactor index and bug fix (infiniflow#794)

### What problem does this PR solve?

Remove redundant codes
Fix several codec errors
Add NoCompressed codec for debug usage
Have the MemoryIndexer unit test passed under single thread

Issue link:infiniflow#551

### Type of change

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

Author: yingfeng

src/storage/invertedindex/format/buffered_byte_slice.cppm

@@ -57,7 +57,7 @@ public:
     virtual SizeT EstimateDumpSize() const { return posting_writer_.GetSize(); }
 
 protected:
-    virtual SizeT DoFlush();
+    SizeT DoFlush();
 
 protected:
     FlushInfo flush_info_;

src/storage/invertedindex/format/buffered_skiplist_writer.cpp

@@ -24,7 +24,6 @@ void BufferedSkipListWriter::AddItem(u32 key, u32 value1, u32 value2) {
     last_value1_ = value1;
     PushBack(2, value2);
     EndPushBack();
-
     if (NeedFlush(SKIP_LIST_BUFFER_SIZE)) {
         Flush();
     }
@@ -52,74 +51,7 @@ void BufferedSkipListWriter::AddItem(u32 value_delta) {
     }
 }
 
-SizeT BufferedSkipListWriter::DoFlush() {
-    u32 flush_size = 0;
-    u8 size = buffer_.Size();
-
-    const PostingValues *posting_values = GetPostingValues();
-    for (SizeT i = 0; i < posting_values->GetSize(); ++i) {
-        PostingValue *posting_value = posting_values->GetValue(i);
-        u8 *buffer = buffer_.GetRow(posting_value->location_);
-        flush_size += posting_value->Encode(posting_writer_, buffer, size * posting_value->GetSize());
-    }
-    return flush_size;
-}
-
-void BufferedSkipListWriter::Dump(const SharedPtr<FileWriter> &file) {
-    posting_writer_.Dump(file);
-    return;
-
-    u32 skiplist_size = GetTotalCount();
-    if (skiplist_size == 0) {
-        return;
-    }
-    if (GetPostingValues()->GetSize() != 3) {
-        posting_writer_.Dump(file);
-        return;
-    }
-
-    const ByteSliceList *skipList = posting_writer_.GetByteSliceList();
-    ByteSliceListIterator iter(skipList);
-    u32 start = 0;
-    const PostingValues *posting_values = GetPostingValues();
-    for (SizeT i = 0; i < posting_values->GetSize(); ++i) {
-        PostingValue *posting_value = posting_values->GetValue(i);
-        u32 len = posting_value->GetSize() * skiplist_size;
-        if (i > 0) {
-            // not encode last value after first row
-            len -= posting_value->GetSize();
-        }
-        if (len == 0) {
-            break;
-        }
-        bool ret = iter.SeekSlice(start);
-        assert(ret);
-        (void)ret;
-        while (iter.HasNext(start + len)) {
-            void *data = nullptr;
-            SizeT size = 0;
-            iter.Next(data, size);
-            assert(data);
-            assert(size);
-            file->Write((char *)data, size);
-        }
-        start += posting_value->GetSize() * skiplist_size;
-    }
-}
-
-SizeT BufferedSkipListWriter::EstimateDumpSize() const {
-    u32 skiplist_size = GetTotalCount();
-    if (skiplist_size == 0) {
-        return 0;
-    }
-    if (GetPostingValues()->GetSize() != 3) {
-        return posting_writer_.GetSize();
-    }
+void BufferedSkipListWriter::Dump(const SharedPtr<FileWriter> &file) { posting_writer_.Dump(file); }
 
-    const PostingValues *posting_values = GetPostingValues();
-    assert(posting_values->GetSize() == 3);
-
-    SizeT opt_size = posting_values->GetValue(1)->GetSize() + posting_values->GetValue(1)->GetSize();
-    return posting_writer_.GetSize() - opt_size;
-}
+SizeT BufferedSkipListWriter::EstimateDumpSize() const { return posting_writer_.GetSize(); }
 } // namespace infinity

src/storage/invertedindex/format/buffered_skiplist_writer.cppm

@@ -24,9 +24,6 @@ public:
 
     SizeT EstimateDumpSize() const override;
 
-protected:
-    SizeT DoFlush() override;
-
 private:
     static const u32 INVALID_LAST_KEY = (u32)-1;
 

src/storage/invertedindex/format/doc_list_format_option.cppm

@@ -64,6 +64,7 @@ public:
         u8 row_count = 0;
         u32 offset = 0;
         {
+            // NoCompressPostingValue<u32> *doc_id_value = new NoCompressPostingValue<u32>;
             TypedPostingValue<u32> *doc_id_value = new TypedPostingValue<u32>;
             doc_id_value->location_ = row_count++;
             doc_id_value->offset_ = offset;

src/storage/invertedindex/format/inmem_doc_list_decoder.cpp

@@ -34,10 +34,10 @@ void InMemDocListDecoder::Init(df_t df, SkipListReader *skiplist_reader, Buffere
 
 bool InMemDocListDecoder::DecodeDocBuffer(docid_t start_doc_id,
                                           docid_t *doc_buffer,
-                                          docid_t &first_doc_id_,
-                                          docid_t &last_doc_id_,
+                                          docid_t &first_doc_id,
+                                          docid_t &last_doc_id,
                                           ttf_t &current_ttf) {
-    DocBufferInfo doc_buffer_info(doc_buffer, first_doc_id_, last_doc_id_, current_ttf);
+    DocBufferInfo doc_buffer_info(doc_buffer, first_doc_id, last_doc_id, current_ttf);
     if (skiplist_reader_ == nullptr) {
         current_ttf = 0;
         return DecodeDocBufferWithoutSkipList(0, 0, start_doc_id, doc_buffer_info);
@@ -47,7 +47,7 @@ bool InMemDocListDecoder::DecodeDocBuffer(docid_t start_doc_id,
     u32 record_len;
     u32 last_doc_id_in_prev_record;
 
-    auto ret = skiplist_reader_->SkipTo((u32)start_doc_id, (u32 &)last_doc_id_, last_doc_id_in_prev_record, offset, record_len);
+    auto ret = skiplist_reader_->SkipTo((u32)start_doc_id, (u32 &)last_doc_id, last_doc_id_in_prev_record, offset, record_len);
     if (!ret) {
         // we should decode buffer
         last_doc_id_in_prev_record = skiplist_reader_->GetLastKeyInBuffer();
@@ -64,23 +64,20 @@ bool InMemDocListDecoder::DecodeDocBuffer(docid_t start_doc_id,
     SizeT acutal_decode_count = 0;
     doc_list_reader_.Decode(doc_buffer, MAX_DOC_PER_RECORD, acutal_decode_count);
 
-    first_doc_id_ = doc_buffer[0] + last_doc_id_in_prev_record;
+    first_doc_id = doc_buffer[0] + last_doc_id_in_prev_record;
     return true;
 }
 
 bool InMemDocListDecoder::DecodeDocBufferWithoutSkipList(docid_t last_doc_id_in_prev_record,
                                                          u32 offset,
                                                          docid_t start_doc_id,
                                                          DocBufferInfo &doc_buffer_info) {
-    // only decode one time
     if (finish_decoded_) {
         return false;
     }
 
-    // TODO: seek return value
     doc_list_reader_.Seek(offset);
 
-    // short list when no skip
     SizeT decode_count;
     if (!doc_list_reader_.Decode(doc_buffer_info.doc_buffer_, MAX_DOC_PER_RECORD, decode_count)) {
         return false;

src/storage/invertedindex/format/inmem_pair_value_skiplist_reader.cpp

@@ -6,6 +6,8 @@ import buffered_byte_slice;
 import buffered_byte_slice_reader;
 import pair_value_skiplist_reader;
 import index_defines;
+import logger;
+import third_party;
 
 module inmem_pair_value_skiplist_reader;
 
@@ -51,7 +53,7 @@ Pair<int, bool> InMemPairValueSkipListReader::LoadBuffer() {
         return MakePair(0, false);
     }
     if (key_num != value_num) {
-        // LOG_ERROR(fmt::format("SKipList decode error, doc_num = {} offset_num = {}", key_num, value_num));
+        LOG_ERROR(fmt::format("SKipList decode error, doc_num = {} offset_num = {}", key_num, value_num));
         return MakePair(-1, false);
     }
     num_in_buffer_ = key_num;

src/storage/invertedindex/format/inmem_pair_value_skiplist_reader.cppm

@@ -12,13 +12,13 @@ namespace infinity {
 
 export class InMemPairValueSkipListReader : public PairValueSkipListReader {
 public:
-    InMemPairValueSkipListReader(MemoryPool *sessionPool);
+    InMemPairValueSkipListReader(MemoryPool *session_pool);
 
     ~InMemPairValueSkipListReader();
 
     InMemPairValueSkipListReader(const InMemPairValueSkipListReader &other) = delete;
 
-    void Load(BufferedByteSlice *postingBuffer);
+    void Load(BufferedByteSlice *posting_buffer);
 
 private:
     Pair<int, bool> LoadBuffer() override;

src/storage/invertedindex/format/no_compress_encoder.cppm

@@ -0,0 +1,38 @@
+module;
+
+import stl;
+import byte_slice_reader;
+import byte_slice_writer;
+
+export module no_compress_encoder;
+
+namespace infinity {
+
+export template <typename T>
+class NoCompressIntEncoder {
+public:
+    NoCompressIntEncoder() = default;
+    ~NoCompressIntEncoder() = default;
+
+    inline u32 Encode(ByteSliceWriter &slice_writer, const T *src, u32 src_len) const;
+
+    inline u32 Decode(T *dest, u32 dest_len, ByteSliceReader &slice_reader) const;
+};
+
+template <typename T>
+u32 NoCompressIntEncoder<T>::Encode(ByteSliceWriter &slice_writer, const T *src, u32 src_len) const {
+    u32 encode_len = 0;
+    slice_writer.WriteByte((u8)src_len);
+    encode_len += sizeof(u8);
+    slice_writer.Write((const void *)src, src_len * sizeof(T));
+    return encode_len + src_len * sizeof(T);
+}
+
+template <typename T>
+u32 NoCompressIntEncoder<T>::Decode(T *dest, u32 dest_len, ByteSliceReader &slice_reader) const {
+    u32 read_count = slice_reader.ReadByte();
+    slice_reader.Read((void *)dest, read_count * sizeof(T));
+    return read_count;
+}
+
+} // namespace infinity

src/storage/invertedindex/format/pair_value_skiplist_reader.cppm

@@ -31,6 +31,7 @@ public:
     u32 GetCurrentKey() const { return current_key_; }
 
     u32 GetLastValueInBuffer() const override;
+
     u32 GetLastKeyInBuffer() const override;
 
 private:

src/storage/invertedindex/format/pos_list_decoder.cpp

@@ -225,7 +225,7 @@ bool PositionListDecoder::LocateRecord(const InDocPositionState *state, u32 &tf)
     record_offset_ = record_offset;
     offset_in_record_ = offset_in_record;
 
-    ByteSliceList *positionList = pos_list_reader_.GetByteSliceList();
+    ByteSliceList *position_list = pos_list_reader_.GetByteSliceList();
     if (!need_reopen_ && last_decode_offset_ == record_offset_) {
         // no need to relocate
         return false;
@@ -235,7 +235,7 @@ bool PositionListDecoder::LocateRecord(const InDocPositionState *state, u32 &tf)
         if (pos_single_slice_) {
             pos_list_reader_.Open(pos_single_slice_);
         } else {
-            pos_list_reader_.Open(positionList);
+            pos_list_reader_.Open(position_list);
         }
         last_decode_offset_ = 0;
     }

src/storage/invertedindex/format/pos_list_decoder.cppm

@@ -28,7 +28,6 @@ public:
 
     void SetNeedReopen(bool need_reopen) { need_reopen_ = need_reopen; }
 
-    // TODO: separate TfBitmap CalculateRecordOffset from LocateRecord
     virtual bool LocateRecord(const InDocPositionState *state, u32 &tf);
 
     virtual u32 DecodeRecord(pos_t *pos_buffer, u32 pos_buffer_len);

src/storage/invertedindex/format/pos_list_encoder.cpp

@@ -105,7 +105,6 @@ InMemPositionListDecoder *PositionListEncoder::GetInMemPositionListDecoder(Memor
     // doclist -> ttf -> pos skiplist -> poslist
     ttf_t ttf = total_pos_count_;
 
-    // TODO: delete buffer in MemoryPool
     InMemPairValueSkipListReader *in_mem_skiplist_reader = nullptr;
     if (pos_skiplist_writer_) {
         // not support tf bitmap in realtime segment

src/storage/invertedindex/format/posting_value.cpp

@@ -2,6 +2,7 @@ module;
 
 import stl;
 import int_encoder;
+import no_compress_encoder;
 
 module posting_value;
 
@@ -10,6 +11,7 @@ namespace infinity {
 struct EncoderProvider {
     UniquePtr<Int32Encoder> int32_encoder_;
     UniquePtr<Int16Encoder> int16_encoder_;
+    UniquePtr<NoCompressIntEncoder<u32>> no_compress_encoder_;
 
     static EncoderProvider *GetInstance() {
         static EncoderProvider instance;
@@ -27,6 +29,12 @@ struct EncoderProvider {
             int16_encoder_.reset(new Int16Encoder);
         return int16_encoder_.get();
     }
+
+    NoCompressEncoder *GetNoCompressEncoder() {
+        if (!no_compress_encoder_.get())
+            no_compress_encoder_.reset(new NoCompressEncoder);
+        return no_compress_encoder_.get();
+    }
 };
 
 const Int32Encoder *GetDocIDEncoder() { return EncoderProvider::GetInstance()->GetInt32Encoder(); }

src/storage/invertedindex/format/posting_value.cppm

@@ -4,6 +4,7 @@ import stl;
 import int_encoder;
 import byte_slice_reader;
 import byte_slice_writer;
+import no_compress_encoder;
 
 export module posting_value;
 
@@ -42,6 +43,7 @@ struct ValueTypeTraits<u32> {
 
 export typedef IntEncoder<u32, NewPForDeltaCompressor> Int32Encoder;
 export typedef IntEncoder<u16, NewPForDeltaCompressor> Int16Encoder;
+export typedef NoCompressIntEncoder<u32> NoCompressEncoder;
 
 template <typename T>
 struct EncoderTypeTraits {
@@ -84,6 +86,22 @@ struct TypedPostingValue : public PostingValue {
     const Encoder *encoder_;
 };
 
+export template <typename T>
+struct NoCompressPostingValue : public PostingValue {
+
+    SizeT GetSize() const override { return sizeof(T); }
+
+    u32 Encode(ByteSliceWriter &slice_writer, const u8 *src, u32 len) const override {
+        return encoder_->Encode(slice_writer, (const T *)src, len / sizeof(T));
+    }
+
+    u32 Decode(u8 *dest, u32 dest_len, ByteSliceReader &slice_reader) const override {
+        return encoder_->Decode((T *)dest, dest_len / sizeof(T), slice_reader);
+    }
+
+    const NoCompressEncoder *encoder_;
+};
+
 export struct PostingValues {
     ~PostingValues() {
         for (SizeT i = 0; i < values_.size(); ++i) {