Fix bug in 0.1.x (#1195)

### What problem does this PR solve?

1. Fix: visible check of entry created by committing txn.
2. Fix: dead lock in `SegmentIndexEntry::MemIndexInsert`
3. Fix: cleanup processor get visit_ts from txn_manager, visit_ts is
Min(first_uncommitted_begin_ts, last_checkpoint_ts)
4. Fix: config file typo.

Issue link:#1172
### Type of change

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

conf/infinity_conf.toml

@@ -43,31 +43,31 @@ storage_capacity        = "64GB"
 # s means seconds, for example "60s", 60 seconds
 # m means minutes, for example "60m", 60 minutes
 # h means hours, for example "1h", 1 hour
-garbage_collection_interval = "60s"
+cleanup_interval = "60s"
 
 # storage ratio activates garbage collection:
 # 0 means disable,
 # 0.1 means, once the storage reach 10% storage capacity, GC is triggered.
 garbage_collection_storage_ratio = 0.1
 
 # dump memory index entry when it reachs the capacity
-memindex_capacity       = 1048576
+mem_index_capacity       = 1048576
 
 [buffer]
-buffer_pool_size        = "4GB"
+buffer_manager_size        = "4GB"
 temp_dir                = "/var/infinity/tmp"
 
 [wal]
 wal_dir                 = "/var/infinity/wal"
-full_checkpoint_interval_sec      = 86400
-delta_checkpoint_interval_sec     = 60
-delta_checkpoint_interval_wal_bytes = 1000000000
-wal_file_size_threshold            = "1GB"
+full_checkpoint_interval      = "86400s"
+delta_checkpoint_interval     = "60s"
+# delta_checkpoint_threshold = 1000000000
+wal_compact_threshold            = "1GB"
 
 # flush_at_once: write and flush log each commit
 # only_write: write log, OS control when to flush the log, default
 # flush_per_second: logs are written after each commit and flushed to disk per second.
-flush_at_commit                   = "only_write"
+flush_option                   = "only_write"
 
 [resource]
 dictionary_dir                = "/var/infinity/resource"

src/executor/operator/physical_index_scan.cpp

@@ -452,14 +452,13 @@ struct FilterResult {
     }
 
     template <typename ColumnValueType>
-    inline void
-    ExecuteSingleRangeT(const FilterIntervalRangeT<ColumnValueType> &interval_range, SegmentIndexEntry &index_entry, const TxnTimeStamp ts) {
+    inline void ExecuteSingleRangeT(const FilterIntervalRangeT<ColumnValueType> &interval_range, SegmentIndexEntry &index_entry, Txn *txn) {
         Vector<UniquePtr<TrunkReader<ColumnValueType>>> trunk_readers;
         Tuple<Vector<SharedPtr<ChunkIndexEntry>>, SharedPtr<SecondaryIndexInMem>> chunks_snapshot = index_entry.GetSecondaryIndexSnapshot();
         const u32 segment_row_count = SegmentRowCount();
         auto &[chunk_index_entries, memory_secondary_index] = chunks_snapshot;
         for (const auto &chunk_index_entry : chunk_index_entries) {
-            if (chunk_index_entry->CheckVisible(ts)) {
+            if (chunk_index_entry->CheckVisible(txn)) {
                 trunk_readers.emplace_back(MakeUnique<TrunkReaderT<ColumnValueType>>(segment_row_count, chunk_index_entry));
             }
         }
@@ -492,7 +491,7 @@ struct FilterResult {
     inline void ExecuteSingleRange(const HashMap<ColumnID, TableIndexEntry *> &column_index_map,
                                    const FilterExecuteSingleRange &single_range,
                                    SegmentID segment_id,
-                                   const TxnTimeStamp ts) {
+                                   Txn *txn) {
         // step 1. check if range is empty
         if (single_range.IsEmpty()) {
             return SetEmptyResult();
@@ -504,7 +503,7 @@ struct FilterResult {
         // step 3. search index
         auto &interval_range_variant = single_range.GetIntervalRange();
         std::visit(Overload{[&]<typename ColumnValueType>(const FilterIntervalRangeT<ColumnValueType> &interval_range) {
-                                ExecuteSingleRangeT(interval_range, index_entry, ts);
+                                ExecuteSingleRangeT(interval_range, index_entry, txn);
                             },
                             [](const std::monostate &empty) {
                                 UnrecoverableError("FilterResult::ExecuteSingleRange(): class member interval_range_ not initialized!");
@@ -598,7 +597,7 @@ FilterResult SolveSecondaryIndexFilterInner(const Vector<FilterExecuteElem> &fil
                                             const SegmentID segment_id,
                                             const u32 segment_row_count,
                                             const u32 segment_row_actual_count,
-                                            const TxnTimeStamp ts) {
+                                            Txn *txn) {
     Vector<FilterResult> result_stack;
     // execute filter_execute_command_ (Reverse Polish notation)
     for (auto const &elem : filter_execute_command) {
@@ -628,7 +627,7 @@ FilterResult SolveSecondaryIndexFilterInner(const Vector<FilterExecuteElem> &fil
                             },
                             [&](const FilterExecuteSingleRange &single_range) {
                                 result_stack.emplace_back(segment_row_count, segment_row_actual_count);
-                                result_stack.back().ExecuteSingleRange(column_index_map, single_range, segment_id, ts);
+                                result_stack.back().ExecuteSingleRange(column_index_map, single_range, segment_id, txn);
                             }},
                    elem);
     }
@@ -644,13 +643,13 @@ std::variant<Vector<u32>, Bitmask> SolveSecondaryIndexFilter(const Vector<Filter
                                                              const SegmentID segment_id,
                                                              const u32 segment_row_count,
                                                              const u32 segment_row_actual_count,
-                                                             const TxnTimeStamp ts) {
+                                                             Txn *txn) {
     if (filter_execute_command.empty()) {
         // return all true
         return std::variant<Vector<u32>, Bitmask>(std::in_place_type<Bitmask>);
     }
     auto result =
-        SolveSecondaryIndexFilterInner(filter_execute_command, column_index_map, segment_id, segment_row_count, segment_row_actual_count, ts);
+        SolveSecondaryIndexFilterInner(filter_execute_command, column_index_map, segment_id, segment_row_count, segment_row_actual_count, txn);
     return std::move(result.selected_rows_);
 }
 
@@ -709,7 +708,7 @@ void PhysicalIndexScan::ExecuteInternal(QueryContext *query_context, IndexScanOp
     DeleteFilter delete_filter(segment_entry, begin_ts, segment_entry->row_count(begin_ts));
     // output
     const auto result =
-        SolveSecondaryIndexFilterInner(filter_execute_command_, column_index_map_, segment_id, segment_row_count, segment_row_actual_count, begin_ts);
+        SolveSecondaryIndexFilterInner(filter_execute_command_, column_index_map_, segment_id, segment_row_count, segment_row_actual_count, txn);
     result.Output(output_data_blocks, segment_id, delete_filter);
 
     LOG_TRACE(fmt::format("IndexScan: job number: {}, segment_ids.size(): {}, finished", next_idx, segment_ids.size()));

src/executor/operator/physical_index_scan.cppm

@@ -39,6 +39,8 @@ import bitmask;
 
 namespace infinity {
 
+class Txn;
+
 // for int range filter, x > n is equivalent to x >= n + 1
 // for float range filter, x > f is equivalent to x >= std::nextafter(f, INFINITY)
 // we can use this to simplify the filter
@@ -110,6 +112,6 @@ export std::variant<Vector<u32>, Bitmask> SolveSecondaryIndexFilter(const Vector
                                                                     const SegmentID segment_id,
                                                                     const u32 segment_row_count,
                                                                     const u32 segment_row_actual_count,
-                                                                    const TxnTimeStamp ts);
+                                                                    Txn *txn);
 
 } // namespace infinity

src/executor/operator/physical_knn_scan.cpp

@@ -241,9 +241,10 @@ SizeT PhysicalKnnScan::BlockEntryCount() const { return base_table_ref_->block_i
 
 template <typename DataType, template <typename, typename> typename C>
 void PhysicalKnnScan::ExecuteInternal(QueryContext *query_context, KnnScanOperatorState *operator_state) {
-    TxnTimeStamp begin_ts = query_context->GetTxn()->BeginTS();
+    Txn *txn = query_context->GetTxn();
+    TxnTimeStamp begin_ts = txn->BeginTS();
 
-    if (!common_query_filter_->TryFinishBuild(begin_ts, query_context->GetTxn()->buffer_mgr())) {
+    if (!common_query_filter_->TryFinishBuild(txn)) {
         // not ready, abort and wait for next time
         return;
     }
@@ -484,7 +485,7 @@ void PhysicalKnnScan::ExecuteInternal(QueryContext *query_context, KnnScanOperat
                                 if (block_entry == nullptr) {
                                     UnrecoverableError(
                                         fmt::format("Cannot find segment id: {}, block id: {}, index chunk is {}", segment_id, block_id, chunk_id));
-                                }
+                                } // this is for debug
                             }
                             merge_heap->Search(0, d_ptr.get(), row_ids.get(), result_n);
                         }
@@ -493,7 +494,7 @@ void PhysicalKnnScan::ExecuteInternal(QueryContext *query_context, KnnScanOperat
                     auto [chunk_index_entries, memory_index_entry] = segment_index_entry->GetHnswIndexSnapshot();
                     int i = 0;
                     for (auto &chunk_index_entry : chunk_index_entries) {
-                        if (chunk_index_entry->CheckVisible(begin_ts)) {
+                        if (chunk_index_entry->CheckVisible(txn)) {
                             BufferHandle index_handle = chunk_index_entry->GetIndex();
                             hnsw_search(index_handle, false, i++);
                         }

src/executor/operator/physical_match.cpp

@@ -23,7 +23,7 @@ module;
 module physical_match;
 
 import stl;
-
+import txn;
 import query_context;
 import operator_state;
 import physical_operator;
@@ -556,9 +556,8 @@ bool PhysicalMatch::ExecuteInnerHomebrewed(QueryContext *query_context, Operator
     auto execute_start_time = std::chrono::high_resolution_clock::now();
     // 1. build QueryNode tree
     // 1.1 populate column2analyzer
-    TransactionID txn_id = query_context->GetTxn()->TxnID();
-    TxnTimeStamp begin_ts = query_context->GetTxn()->BeginTS();
-    QueryBuilder query_builder(txn_id, begin_ts, base_table_ref_);
+    Txn *txn = query_context->GetTxn();
+    QueryBuilder query_builder(txn, base_table_ref_);
     auto finish_init_query_builder_time = std::chrono::high_resolution_clock::now();
     TimeDurationType query_builder_init_duration = finish_init_query_builder_time - execute_start_time;
     LOG_TRACE(fmt::format("PhysicalMatch Part 0.1: Init QueryBuilder time: {} ms", query_builder_init_duration.count()));
@@ -849,7 +848,8 @@ bool PhysicalMatch::Execute(QueryContext *query_context, OperatorState *operator
     auto start_time = std::chrono::high_resolution_clock::now();
     assert(common_query_filter_);
     {
-        bool try_result = common_query_filter_->TryFinishBuild(query_context->GetTxn()->BeginTS(), query_context->GetTxn()->buffer_mgr());
+        Txn *txn = query_context->GetTxn();
+        bool try_result = common_query_filter_->TryFinishBuild(txn);
         auto finish_filter_time = std::chrono::high_resolution_clock::now();
         std::chrono::duration<float, std::milli> filter_duration = finish_filter_time - start_time;
         LOG_TRACE(fmt::format("PhysicalMatch Prepare: Filter time: {} ms", filter_duration.count()));

src/main/query_context.cpp

@@ -50,6 +50,7 @@ import session_manager;
 import base_statement;
 import parser_result;
 import parser_assert;
+import plan_fragment;
 
 namespace infinity {
 
@@ -106,6 +107,10 @@ QueryResult QueryContext::Query(const String &query) {
 
 QueryResult QueryContext::QueryStatement(const BaseStatement *statement) {
     QueryResult query_result;
+    SharedPtr<LogicalNode> logical_plan = nullptr;
+    UniquePtr<PlanFragment> plan_fragment = nullptr;
+    UniquePtr<PhysicalOperator> physical_plan = nullptr;
+
 //    ProfilerStart("Query");
 //    BaseProfiler profiler;
 //    profiler.Begin();
@@ -127,7 +132,7 @@ QueryResult QueryContext::QueryStatement(const BaseStatement *statement) {
         }
 
         current_max_node_id_ = bind_context->GetNewLogicalNodeId();
-        SharedPtr<LogicalNode> logical_plan = logical_planner_->LogicalPlan();
+        logical_plan = logical_planner_->LogicalPlan();
         StopProfile(QueryPhase::kLogicalPlan);
 //        LOG_WARN(fmt::format("Before optimizer cost: {}", profiler.ElapsedToString()));
         // Apply optimized rule to the logical plan
@@ -137,13 +142,13 @@ QueryResult QueryContext::QueryStatement(const BaseStatement *statement) {
 
         // Build physical plan
         StartProfile(QueryPhase::kPhysicalPlan);
-        UniquePtr<PhysicalOperator> physical_plan = physical_planner_->BuildPhysicalOperator(logical_plan);
+        physical_plan = physical_planner_->BuildPhysicalOperator(logical_plan);
         StopProfile(QueryPhase::kPhysicalPlan);
 //        LOG_WARN(fmt::format("Before pipeline cost: {}", profiler.ElapsedToString()));
         StartProfile(QueryPhase::kPipelineBuild);
         // Fragment Builder, only for test now.
         // SharedPtr<PlanFragment> plan_fragment = fragment_builder.Build(physical_plan);
-        auto plan_fragment = fragment_builder_->BuildFragment(physical_plan.get());
+        plan_fragment = fragment_builder_->BuildFragment(physical_plan.get());
         StopProfile(QueryPhase::kPipelineBuild);
 
         auto notifier = MakeUnique<Notifier>();
@@ -159,14 +164,7 @@ QueryResult QueryContext::QueryStatement(const BaseStatement *statement) {
         StopProfile(QueryPhase::kExecution);
 //        LOG_WARN(fmt::format("Before commit cost: {}", profiler.ElapsedToString()));
         StartProfile(QueryPhase::kCommit);
-        try {
-            this->CommitTxn();
-        } catch (RecoverableException &e) {
-            StopProfile();
-            this->RollbackTxn();
-            query_result.result_table_ = nullptr;
-            query_result.status_.Init(e.ErrorCode(), e.what());
-        }
+        this->CommitTxn();
         StopProfile(QueryPhase::kCommit);
 
     } catch (RecoverableException &e) {

src/planner/bound/base_table_ref.cppm

@@ -21,7 +21,7 @@ export module base_table_ref;
 import stl;
 import table_ref;
 import table_entry;
-
+import txn;
 import table_function;
 import block_index;
 import internal_types;
@@ -48,8 +48,8 @@ public:
     explicit BaseTableRef(TableEntry *table_entry, SharedPtr<BlockIndex> block_index)
         : TableRef(TableRefType::kTable, ""), table_entry_ptr_(table_entry), block_index_(std::move(block_index)) {}
 
-    static SharedPtr<BaseTableRef> FakeTableRef(TableEntry *table_entry, TxnTimeStamp ts) {
-        SharedPtr<BlockIndex> block_index = table_entry->GetBlockIndex(ts);
+    static SharedPtr<BaseTableRef> FakeTableRef(TableEntry *table_entry, Txn *txn) {
+        SharedPtr<BlockIndex> block_index = table_entry->GetBlockIndex(txn);
         return MakeShared<BaseTableRef>(table_entry, std::move(block_index));
     }
 

src/planner/query_binder.cpp

@@ -71,6 +71,7 @@ import data_type;
 import logical_type;
 import base_entry;
 import view_entry;
+import txn;
 
 namespace infinity {
 
@@ -425,10 +426,9 @@ SharedPtr<BaseTableRef> QueryBinder::BuildBaseTable(QueryContext *query_context,
         columns.emplace_back(idx);
     }
 
-//    TransactionID txn_id = query_context->GetTxn()->TxnID();
-    TxnTimeStamp begin_ts = query_context->GetTxn()->BeginTS();
+    Txn *txn = query_context->GetTxn();
 
-    SharedPtr<BlockIndex> block_index = table_entry->GetBlockIndex(begin_ts);
+    SharedPtr<BlockIndex> block_index = table_entry->GetBlockIndex(txn);
 
     u64 table_index = bind_context_ptr_->GenerateTableIndex();
     auto table_ref = MakeShared<BaseTableRef>(table_entry, std::move(columns), block_index, alias, table_index, names_ptr, types_ptr);

src/storage/bg_task/compact_segments_task.cpp

@@ -148,7 +148,7 @@ void CompactSegmentsTask::CompactSegments(CompactSegmentsTaskState &state) {
         }
 
         auto new_segment = CompactSegmentsToOne(state, to_compact_segments);
-        block_index->Insert(new_segment.get(), UNCOMMIT_TS, false);
+        block_index->Insert(new_segment.get(), txn_);
 
         segment_data.emplace_back(new_segment, std::move(to_compact_segments));
         old_segments.insert(old_segments.end(), to_compact_segments.begin(), to_compact_segments.end());

src/storage/bg_task/periodic_trigger.cpp

@@ -28,7 +28,7 @@ import third_party;
 namespace infinity {
 
 void CleanupPeriodicTrigger::Trigger() {
-    TxnTimeStamp visible_ts = txn_mgr_->GetMinUnflushedTS();
+    TxnTimeStamp visible_ts = txn_mgr_->GetCleanupScanTS();
     // if (visible_ts == last_visible_ts_) {
     //     LOG_TRACE(fmt::format("Skip cleanup. visible timestamp: {}", visible_ts));
     //     return;
@@ -49,8 +49,7 @@ void CheckpointPeriodicTrigger::Trigger() {
     auto checkpoint_task = MakeShared<CheckpointTask>(is_full_checkpoint_);
     LOG_TRACE(fmt::format("Trigger {} periodic checkpoint.", is_full_checkpoint_ ? "FULL" : "DELTA"));
     if (!wal_mgr_->TrySubmitCheckpointTask(std::move(checkpoint_task))) {
-        LOG_TRACE(
-            fmt::format("Skip {} checkpoint(time) because there is already a checkpoint task running.", is_full_checkpoint_ ? "FULL" : "DELTA"));
+        LOG_TRACE(fmt::format("Skip {} checkpoint(time) because there is already a checkpoint task running.", is_full_checkpoint_ ? "FULL" : "DELTA"));
     }
 }
 

src/storage/buffer/buffer_obj.cppm

@@ -48,6 +48,8 @@ export String BufferStatusToString(BufferStatus status) {
             return "Freed";
         case BufferStatus::kNew:
             return "New";
+        case BufferStatus::kClean:
+            return "Clean";
         default:
             return "Invalid";
     }
@@ -77,7 +79,7 @@ public:
 
     void CleanupFile() const;
 
-    void CleanupTempFile() const ;
+    void CleanupTempFile() const;
 
     SizeT GetBufferSize() const { return file_worker_->GetMemoryCost(); }
 

src/storage/common/block_index.cpp

@@ -14,32 +14,36 @@
 
 module;
 
+#include <vector>
+
 module block_index;
 
 import stl;
 import segment_entry;
 import global_block_id;
 import block_iter;
-import segment_iter;
+import txn;
 
 namespace infinity {
 
-void BlockIndex::Insert(SegmentEntry *segment_entry, TxnTimeStamp timestamp, bool check_ts) {
-    if (!check_ts || segment_entry->CheckVisible(timestamp)) {
+void BlockIndex::Insert(SegmentEntry *segment_entry, Txn *txn) {
+    if (segment_entry->CheckVisible(txn)) {
         u32 segment_id = segment_entry->segment_id();
         segments_.emplace_back(segment_entry);
         segment_index_.emplace(segment_id, segment_entry);
         BlocksInfo blocks_info;
 
-        auto block_entry_iter = BlockEntryIter(segment_entry);
-        for (auto *block_entry = block_entry_iter.Next(); block_entry != nullptr; block_entry = block_entry_iter.Next()) {
-            if (timestamp >= block_entry->min_row_ts()) {
-                blocks_info.block_map_.emplace(block_entry->block_id(), block_entry);
-                global_blocks_.emplace_back(GlobalBlockID{segment_id, block_entry->block_id()});
+        {
+            auto block_guard = segment_entry->GetBlocksGuard();
+            for (const auto &block_entry : block_guard.block_entries_) {
+                if (block_entry->CheckVisible(txn)) {
+                    blocks_info.block_map_.emplace(block_entry->block_id(), block_entry.get());
+                    global_blocks_.emplace_back(GlobalBlockID{segment_id, block_entry->block_id()});
+                }
             }
         }
-        blocks_info.segment_offset_ = segment_entry->row_count(timestamp);
-        // blocks_info.segment_offset_ = segment_entry->row_count(); // use false row count to pass benchmark
+        TxnTimeStamp begin_ts = txn->BeginTS();
+        blocks_info.segment_offset_ = segment_entry->row_count(begin_ts);
 
         segment_block_index_.emplace(segment_id, std::move(blocks_info));
     }