Add some logging for locating bug

dbms/src/Flash/Disaggregated/WNEstablishDisaggTaskHandler.cpp

@@ -80,15 +80,16 @@ void WNEstablishDisaggTaskHandler::prepare(const disaggregated::EstablishDisaggT
 
 void WNEstablishDisaggTaskHandler::execute(disaggregated::EstablishDisaggTaskResponse * response)
 {
-    // run into DAGStorageInterpreter and build the segment snapshots
-    query_executor_holder.set(queryExecute(*context));
-
+    // Set the store_id to response before executing query
     auto & tmt = context->getTMTContext();
     {
         const auto & kvstore = tmt.getKVStore();
         response->set_store_id(kvstore->getStoreID());
     }
 
+    // run into DAGStorageInterpreter and build the segment snapshots
+    query_executor_holder.set(queryExecute(*context));
+
     auto snaps = context->getSharedContextDisagg()->wn_snapshot_manager;
     const auto & task_id = *dag_context->getDisaggTaskId();
     auto snap = snaps->getSnapshot(task_id);

dbms/src/Server/StorageConfigParser.h

@@ -100,7 +100,7 @@ struct StorageS3Config
     // verbose logging for http requests. Use for debugging
     bool verbose = false;
 
-    bool enable_http_pool = true; // will be removed after testing
+    bool enable_http_pool = false; // will be removed after testing
     bool enable_poco_client = true; // will be removed after testing
 
     String endpoint;

dbms/src/Storages/DeltaMerge/DMVersionFilterBlockInputStream.h

@@ -71,7 +71,7 @@ class DMVersionFilterBlockInputStream : public IBlockInputStream
                   "Total rows: {}, pass: {:.2f}%"
                   ", complete pass: {:.2f}%, complete not pass: {:.2f}%"
                   ", not clean: {:.2f}%, is deleted: {:.2f}%, effective: {:.2f}%"
-                  ", read tso: {}",
+                  ", start_ts: {}",
                   total_rows,
                   passed_rows * 100.0 / total_rows,
                   complete_passed * 100.0 / total_blocks,

dbms/src/Storages/DeltaMerge/Remote/RNReadTask.cpp

@@ -25,7 +25,7 @@ namespace DB::DM::Remote
 {
 
 RNReadSegmentTaskPtr RNReadSegmentTask::buildFromEstablishResp(
-    const LoggerPtr & log,
+    const LoggerPtr & table_log,
     const Context & db_context,
     const ScanContextPtr & scan_context,
     const RemotePb::RemoteSegment & proto,
@@ -47,6 +47,7 @@ RNReadSegmentTaskPtr RNReadSegmentTask::buildFromEstablishResp(
         read_ranges[i] = RowKeyRange::deserialize(rb);
     }
 
+    auto tracing_id = fmt::format("{} segment_id={}", table_log->identifier(), proto.segment_id());
     auto dm_context = std::make_shared<DMContext>(
         db_context,
         /* path_pool */ nullptr,
@@ -57,10 +58,11 @@ RNReadSegmentTaskPtr RNReadSegmentTask::buildFromEstablishResp(
         /* is_common_handle */ segment_range.is_common_handle,
         /* rowkey_column_size */ segment_range.rowkey_column_size,
         db_context.getSettingsRef(),
-        scan_context);
+        scan_context,
+        tracing_id);
 
     auto segment = std::make_shared<Segment>(
-        log,
+        Logger::get(),
         /*epoch*/ 0,
         segment_range,
         proto.segment_id(),
@@ -94,12 +96,9 @@ RNReadSegmentTaskPtr RNReadSegmentTask::buildFromEstablishResp(
     }
 
     LOG_DEBUG(
-        log,
-        "Build RNReadSegmentTask, segment_id={} store_id={} keyspace_id={} table_id={} memtable_cfs={} persisted_cfs={}",
+        table_log,
+        "Build RNReadSegmentTask, segment_id={} memtable_cfs={} persisted_cfs={}",
         proto.segment_id(),
-        store_id,
-        keyspace_id,
-        physical_table_id,
         segment_snap->delta->getMemTableSetSnapshot()->getColumnFileCount(),
         segment_snap->delta->getPersistedFileSetSnapshot()->getColumnFileCount());
 

dbms/src/Storages/DeltaMerge/Remote/RNReadTask.h

@@ -89,10 +89,11 @@ class RNReadSegmentTask : boost::noncopyable
     String info() const
     {
         return fmt::format(
-            "ReadSegmentTask<store_id={} seg_id={} table_id={}>",
+            "ReadSegmentTask<store_id={} keyspace_id={} table_id={} segment_id={}>",
             meta.store_id,
-            meta.segment_id,
-            meta.physical_table_id);
+            meta.keyspace_id,
+            meta.physical_table_id,
+            meta.segment_id);
     }
 
     /// Called from WorkerFetchPages.

dbms/src/Storages/DeltaMerge/Segment.cpp

@@ -2505,7 +2505,7 @@ BitmapFilterPtr Segment::buildBitmapFilterNormal(const DMContext & dm_context,
     auto bitmap_filter = std::make_shared<BitmapFilter>(total_rows, /*default_value*/ false);
     bitmap_filter->set(stream);
     bitmap_filter->runOptimize();
-    LOG_DEBUG(log, "buildBitmapFilterNormal total_rows={} cost={}ms", total_rows, sw_total.elapsedMilliseconds());
+    LOG_DEBUG(log->getChild(dm_context.tracing_id), "buildBitmapFilterNormal total_rows={} cost={}ms", total_rows, sw_total.elapsedMilliseconds());
     return bitmap_filter;
 }
 

dbms/src/Storages/DeltaMerge/StableValueSpace.cpp

@@ -433,7 +433,7 @@ StableValueSpace::Snapshot::getInputStream(
     const std::vector<IdSetPtr> & read_packs,
     bool need_row_id)
 {
-    LOG_DEBUG(log, "max_data_version: {}, enable_handle_clean_read: {}, is_fast_mode: {}, enable_del_clean_read: {}", max_data_version, enable_handle_clean_read, is_fast_scan, enable_del_clean_read);
+    LOG_DEBUG(log, "start_ts: {}, enable_handle_clean_read: {}, is_fast_mode: {}, enable_del_clean_read: {}", max_data_version, enable_handle_clean_read, is_fast_scan, enable_del_clean_read);
     SkippableBlockInputStreams streams;
     std::vector<size_t> rows;
     streams.reserve(stable->files.size());

dbms/src/Storages/StorageDisaggregatedRemote.cpp

@@ -58,6 +58,7 @@
 #include <tipb/select.pb.h>
 
 #include <atomic>
+#include <magic_enum.hpp>
 #include <numeric>
 #include <unordered_set>
 
@@ -332,6 +333,8 @@ void StorageDisaggregated::buildReadTaskForWriteNodeTable(
 
     auto thread_manager = newThreadManager();
     auto n = static_cast<size_t>(table.segments().size());
+
+    auto table_tracing_logger = log->getChild(fmt::format("store_id={} keyspace_id={} table_id={}", store_id, table.keyspace_id(), table.table_id()));
     for (size_t idx = 0; idx < n; ++idx)
     {
         const auto & remote_seg = table.segments(idx);
@@ -341,7 +344,7 @@ void StorageDisaggregated::buildReadTaskForWriteNodeTable(
             "buildRNReadSegmentTask",
             [&] {
                 auto seg_read_task = DM::Remote::RNReadSegmentTask::buildFromEstablishResp(
-                    log,
+                    table_tracing_logger,
                     db_context,
                     scan_context,
                     remote_seg,
@@ -473,8 +476,9 @@ void StorageDisaggregated::buildRemoteSegmentInputStreams(
         *column_defines,
         db_context,
         log);
-    auto read_mode = DM::DeltaMergeStore::getReadMode(db_context, table_scan.isFastScan(), table_scan.keepOrder(), push_down_filter);
+    const auto read_mode = DM::DeltaMergeStore::getReadMode(db_context, table_scan.isFastScan(), table_scan.keepOrder(), push_down_filter);
     const UInt64 read_tso = sender_target_mpp_task_id.query_id.start_ts;
+    LOG_DEBUG(log, "Building segment input streams, read_mode={} is_fast_scan={} keep_order={}", magic_enum::enum_name(read_mode), table_scan.isFastScan(), table_scan.keepOrder());
 
     auto workers = DM::Remote::RNWorkers::create({
         .log = log->getChild(executor_id),