fix ut

Signed-off-by: JaySon-Huang <tshent@qq.com>

Author: JaySon-Huang

dbms/src/Operators/tests/gtest_concat_source.cpp

@@ -387,7 +387,7 @@ try
     }
 
     std::atomic<size_t> received_blocks = 0;
-        received_blocks.fetch_add(1, std::memory_order_relaxed);
+    ResultHandler h([&](const Block & /*block*/) { received_blocks.fetch_add(1, std::memory_order_relaxed); });
 
     PipelineExecGroupBuilder result_builder;
     builder_pool.generate(result_builder, exec_context, "test");

dbms/src/Storages/DeltaMerge/Index/VectorIndex/tests/gtest_dm_vector_index_utils.h

@@ -200,18 +200,21 @@ class DeltaMergeStoreVectorBase : public VectorIndexTestUtils
 
     void read(const RowKeyRange & range, const PushDownExecutorPtr & executor, const ColumnWithTypeAndName & out)
     {
+        auto read_queue = std::make_shared<ActiveSegmentReadTaskQueue>(2, Logger::get("read"));
         auto in = store->read(
             *db_context,
             db_context->getSettingsRef(),
             {cdVec()},
             {range},
+            read_queue,
             /* num_streams= */ 1,
             /* start_ts= */ std::numeric_limits<UInt64>::max(),
             executor,
             std::vector<RuntimeFilterPtr>{},
             0,
             TRACING_NAME,
             DMReadOptions{})[0];
+        read_queue->finishQueueIfEmpty();
         ASSERT_INPUTSTREAM_COLS_UR(
             in,
             Strings({vec_column_name}),

dbms/src/Storages/DeltaMerge/Index/tests/gtest_dm_minmax_index.cpp

@@ -153,18 +153,21 @@ bool checkMatch(
     store->mergeDeltaAll(context);
 
     const ColumnDefine & col_to_read = check_pk ? getExtraHandleColumnDefine(is_common_handle) : cd;
+    auto read_queue = std::make_shared<ActiveSegmentReadTaskQueue>(2, Logger::get("read"));
     auto streams = store->read(
         context,
         context.getSettingsRef(),
         {col_to_read},
         {all_range},
+        read_queue,
         1,
         std::numeric_limits<UInt64>::max(),
         std::make_shared<PushDownExecutor>(filter),
         std::vector<RuntimeFilterPtr>{},
         0,
         name,
         DMReadOptions{});
+    read_queue->finishQueueIfEmpty();
     auto rows = getInputStreamNRows(streams[0]);
     store->drop();
 

dbms/src/Storages/DeltaMerge/ReadThread/tests/gtest_segment_read_task_pool.cpp

@@ -126,15 +126,15 @@ class SegmentReadTasksPoolTest : public SegmentTestBasic
         // Submit to pending_pools
         scheduler.add(pool);
         {
-            std::lock_guard lock(scheduler.pending_mtx); // Disable TSA warnnings
+            std::lock_guard lock(scheduler.pending_mtx); // Disable TSA warnings
             ASSERT_EQ(scheduler.pending_pools.size(), 1);
         }
         ASSERT_EQ(scheduler.read_pools.size(), 0);
 
         // Reap the pending_pools
         scheduler.reapPendingPools();
         {
-            std::lock_guard lock(scheduler.pending_mtx); // Disable TSA warnnings
+            std::lock_guard lock(scheduler.pending_mtx); // Disable TSA warnings
             ASSERT_EQ(scheduler.pending_pools.size(), 0);
         }
         ASSERT_EQ(scheduler.read_pools.size(), 1);
@@ -199,9 +199,12 @@ class SegmentReadTasksPoolTest : public SegmentTestBasic
                 pool->finishSegment(merged_task->units.front().task);
             }
 
-            ASSERT_EQ(pool->q.size(), 0);
             Block blk;
-            ASSERT_FALSE(pool->q.pop(blk));
+            // popBlock can only return empty block
+            pool->shared_q->popBlock(blk);
+            ASSERT_FALSE(blk);
+            pool->shared_q->tryPopBlock(blk);
+            ASSERT_FALSE(blk);
 
             pool->decreaseUnorderedInputStreamRefCount();
             ASSERT_FALSE(pool->valid());

dbms/src/Storages/DeltaMerge/tests/gtest_dm_delta_merge_store_fast_add_peer.cpp

@@ -254,11 +254,13 @@ class DeltaMergeStoreTestFastAddPeer
     void verifyRows(const RowKeyRange & range, size_t rows)
     {
         const auto & columns = store->getTableColumns();
+        auto read_queue = std::make_shared<ActiveSegmentReadTaskQueue>(2, Logger::get("verifyRows"));
         BlockInputStreamPtr in = store->read(
             *db_context,
             db_context->getSettingsRef(),
             columns,
             {range},
+            read_queue,
             /* num_streams= */ 1,
             /* start_ts= */ std::numeric_limits<UInt64>::max(),
             EMPTY_FILTER,
@@ -267,6 +269,7 @@ class DeltaMergeStoreTestFastAddPeer
             TRACING_NAME,
             DMReadOptions{},
             /* expected_block_size= */ 1024)[0];
+        read_queue->finishQueueIfEmpty();
         ASSERT_INPUTSTREAM_NROWS(in, rows);
     }