Storages: Add statistical data of TableScanning in ScanContext (pingc…

…ap#8842)

ref pingcap#8675

dbms/src/Flash/Coprocessor/DAGStorageInterpreter.cpp

@@ -1184,7 +1184,7 @@ void DAGStorageInterpreter::buildLocalStreams(DAGPipeline & pipeline, size_t max
     size_t total_local_region_num = mvcc_query_info->regions_query_info.size();
     if (total_local_region_num == 0)
         return;
-    mvcc_query_info->scan_context->total_local_region_num = total_local_region_num;
+    mvcc_query_info->scan_context->setRegionNumOfCurrentInstance(total_local_region_num);
     const auto table_query_infos = generateSelectQueryInfos();
     bool has_multiple_partitions = table_query_infos.size() > 1;
     // MultiPartitionStreamPool will be disabled in no partition mode or single-partition case
@@ -1250,7 +1250,7 @@ void DAGStorageInterpreter::buildLocalExec(
     size_t total_local_region_num = mvcc_query_info->regions_query_info.size();
     if (total_local_region_num == 0)
         return;
-    mvcc_query_info->scan_context->total_local_region_num = total_local_region_num;
+    mvcc_query_info->scan_context->setRegionNumOfCurrentInstance(total_local_region_num);
     const auto table_query_infos = generateSelectQueryInfos();
     bool has_multiple_partitions = table_query_infos.size() > 1;
     ConcatBuilderPool builder_pool{max_streams};

dbms/src/Flash/Coprocessor/tests/gtest_ti_remote_block_inputstream.cpp

@@ -53,8 +53,8 @@ bool equalSummaries(const ExecutionSummary & left, const ExecutionSummary & righ
         (left.num_iterations == right.num_iterations) && //
         (left.num_produced_rows == right.num_produced_rows) && //
         (left.time_processed_ns == right.time_processed_ns) && //
-        (left.scan_context->total_dmfile_scanned_rows == right.scan_context->total_dmfile_scanned_rows) && //
-        (left.scan_context->total_dmfile_skipped_rows == right.scan_context->total_dmfile_skipped_rows);
+        (left.scan_context->dmfile_data_scanned_rows == right.scan_context->dmfile_data_scanned_rows) && //
+        (left.scan_context->dmfile_data_skipped_rows == right.scan_context->dmfile_data_skipped_rows);
 }
 
 struct MockWriter
@@ -73,10 +73,12 @@ struct MockWriter
         summary.concurrency = 1;
         summary.scan_context = std::make_unique<DM::ScanContext>();
 
-        summary.scan_context->total_dmfile_scanned_packs = 1;
-        summary.scan_context->total_dmfile_skipped_packs = 2;
-        summary.scan_context->total_dmfile_scanned_rows = 8000;
-        summary.scan_context->total_dmfile_skipped_rows = 15000;
+        summary.scan_context->dmfile_data_scanned_rows = 8000;
+        summary.scan_context->dmfile_data_skipped_rows = 15000;
+        summary.scan_context->dmfile_mvcc_scanned_rows = 8000;
+        summary.scan_context->dmfile_mvcc_skipped_rows = 15000;
+        summary.scan_context->dmfile_lm_filter_scanned_rows = 8000;
+        summary.scan_context->dmfile_lm_filter_skipped_rows = 15000;
         summary.scan_context->total_dmfile_rough_set_index_check_time_ns = 10;
         summary.scan_context->total_dmfile_read_time_ns = 200;
         summary.scan_context->create_snapshot_time_ns = 5;

dbms/src/Flash/Statistics/TableScanImpl.cpp

@@ -114,6 +114,15 @@ void TableScanStatistics::collectExtraRuntimeDetail()
         });
         break;
     }
+
+    if (auto it = dag_context.scan_context_map.find(executor_id); it != dag_context.scan_context_map.end())
+    {
+        it->second->setStreamCost(
+            std::max(local_table_scan_detail.min_stream_cost_ns, 0.0),
+            std::max(local_table_scan_detail.max_stream_cost_ns, 0.0),
+            std::max(remote_table_scan_detail.min_stream_cost_ns, 0.0),
+            std::max(remote_table_scan_detail.max_stream_cost_ns, 0.0));
+    }
 }
 
 TableScanStatistics::TableScanStatistics(const tipb::Executor * executor, DAGContext & dag_context_)

dbms/src/Flash/Statistics/TableScanImpl.h

@@ -22,7 +22,9 @@ namespace DB
 {
 struct TableScanDetail : public ConnectionProfileInfo
 {
-    bool is_local;
+    const bool is_local;
+    double min_stream_cost_ns = -1.0;
+    double max_stream_cost_ns = -1.0;
 
     explicit TableScanDetail(bool is_local_)
         : is_local(is_local_)

dbms/src/Interpreters/Settings.h

@@ -39,7 +39,7 @@ struct Settings
       *  but we are not going to do it, because settings is used everywhere as static struct fields.
       */
 
-// clang-format off
+    // clang-format off
 #define APPLY_FOR_SETTINGS(M)                                                                                                                                                                                                           \
     M(SettingString, regions, "", "Deprecated. the region need to be read.")                                                                                                                                                            \
     M(SettingBool, resolve_locks, false, "resolve locks for TiDB transaction")                                                                                                                                                          \

dbms/src/Server/Server.cpp

@@ -951,6 +951,8 @@ int Server::main(const std::vector<std::string> & /*args*/)
 
     TiFlashErrorRegistry::instance(); // This invocation is for initializing
 
+    DM::ScanContext::initCurrentInstanceId(config(), log);
+
     const auto disaggregated_mode = getDisaggregatedMode(config());
     const auto use_autoscaler = useAutoScaler(config());
     const bool use_autoscaler_without_s3 = useAutoScalerWithoutS3(config());

dbms/src/Storages/DeltaMerge/ColumnFile/ColumnFile.h

@@ -18,7 +18,9 @@
 #include <Core/Block.h>
 #include <IO/WriteHelpers.h>
 #include <Storages/DeltaMerge/ColumnFile/ColumnFileDataProvider_fwd.h>
+#include <Storages/DeltaMerge/DMContext.h>
 #include <Storages/DeltaMerge/File/DMFile.h>
+#include <Storages/DeltaMerge/ReadMode.h>
 #include <Storages/DeltaMerge/RowKeyRange.h>
 #include <Storages/DeltaMerge/StoragePool/StoragePool_fwd.h>
 #include <Storages/Page/PageDefinesBase.h>
@@ -178,6 +180,8 @@ class ColumnFileReader
 
     /// Create a new reader from current reader with different columns to read.
     virtual ColumnFileReaderPtr createNewReader(const ColumnDefinesPtr & col_defs) = 0;
+
+    virtual void setReadTag(ReadTag /*read_tag*/) {}
 };
 
 std::pair<size_t, size_t> copyColumnsData(

dbms/src/Storages/DeltaMerge/ColumnFile/ColumnFileBig.cpp

@@ -175,6 +175,7 @@ void ColumnFileBigReader::initStream()
     DMFileBlockInputStreamBuilder builder(context.db_context);
     file_stream
         = builder.setTracingID(context.tracing_id)
+              .setReadTag(read_tag)
               .build(column_file.getFile(), *col_defs, RowKeyRanges{column_file.segment_range}, context.scan_context);
 
     header = file_stream->getHeader();
@@ -386,5 +387,12 @@ ColumnFileReaderPtr ColumnFileBigReader::createNewReader(const ColumnDefinesPtr
     return std::make_shared<ColumnFileBigReader>(context, column_file, new_col_defs);
 }
 
+void ColumnFileBigReader::setReadTag(ReadTag read_tag_)
+{
+    // `read_tag` should be set before `file_stream` is initialized.
+    RUNTIME_CHECK(file_stream == nullptr);
+    read_tag = read_tag_;
+}
+
 } // namespace DM
 } // namespace DB

dbms/src/Storages/DeltaMerge/ColumnFile/ColumnFileBig.h

@@ -137,6 +137,8 @@ class ColumnFileBigReader : public ColumnFileReader
     Block cur_block;
     Columns cur_block_data; // The references to columns in cur_block, for faster access.
 
+    ReadTag read_tag;
+
 private:
     void initStream();
     std::pair<size_t, size_t> readRowsRepeatedly(
@@ -190,6 +192,8 @@ class ColumnFileBigReader : public ColumnFileReader
     size_t skipNextBlock() override;
 
     ColumnFileReaderPtr createNewReader(const ColumnDefinesPtr & new_col_defs) override;
+
+    void setReadTag(ReadTag read_tag_) override;
 };
 
 } // namespace DM

dbms/src/Storages/DeltaMerge/ColumnFile/ColumnFileSetReader.cpp

@@ -84,7 +84,8 @@ ColumnFileSetReader::ColumnFileSetReader(
     const DMContext & context_,
     const ColumnFileSetSnapshotPtr & snapshot_,
     const ColumnDefinesPtr & col_defs_,
-    const RowKeyRange & segment_range_)
+    const RowKeyRange & segment_range_,
+    ReadTag read_tag_)
     : context(context_)
     , snapshot(snapshot_)
     , col_defs(col_defs_)
@@ -98,10 +99,11 @@ ColumnFileSetReader::ColumnFileSetReader(
         column_file_rows.push_back(f->getRows());
         column_file_rows_end.push_back(total_rows);
         column_file_readers.push_back(f->getReader(context, snapshot->getDataProvider(), col_defs));
+        column_file_readers.back()->setReadTag(read_tag_);
     }
 }
 
-ColumnFileSetReaderPtr ColumnFileSetReader::createNewReader(const ColumnDefinesPtr & new_col_defs)
+ColumnFileSetReaderPtr ColumnFileSetReader::createNewReader(const ColumnDefinesPtr & new_col_defs, ReadTag read_tag)
 {
     auto * new_reader = new ColumnFileSetReader(context);
     new_reader->snapshot = snapshot;
@@ -111,7 +113,10 @@ ColumnFileSetReaderPtr ColumnFileSetReader::createNewReader(const ColumnDefinesP
     new_reader->column_file_rows_end = column_file_rows_end;
 
     for (auto & fr : column_file_readers)
+    {
         new_reader->column_file_readers.push_back(fr->createNewReader(new_col_defs));
+        new_reader->column_file_readers.back()->setReadTag(read_tag);
+    }
 
     return std::shared_ptr<ColumnFileSetReader>(new_reader);
 }

dbms/src/Storages/DeltaMerge/ColumnFile/ColumnFileSetReader.h

@@ -57,11 +57,12 @@ class ColumnFileSetReader
         const DMContext & context_,
         const ColumnFileSetSnapshotPtr & snapshot_,
         const ColumnDefinesPtr & col_defs_,
-        const RowKeyRange & segment_range_);
+        const RowKeyRange & segment_range_,
+        ReadTag read_tag_);
 
     // If we need to read columns besides pk and version, a ColumnFileSetReader can NOT be used more than once.
     // This method create a new reader based on the current one. It will reuse some caches in the current reader.
-    ColumnFileSetReaderPtr createNewReader(const ColumnDefinesPtr & new_col_defs);
+    ColumnFileSetReaderPtr createNewReader(const ColumnDefinesPtr & new_col_defs, ReadTag read_tag);
 
     // Use for DeltaMergeBlockInputStream to read rows from MemTableSet to do full compaction with other layer.
     // This method will check whether offset and limit are valid. It only return those valid rows.
@@ -104,8 +105,9 @@ class ColumnFileSetInputStream : public SkippableBlockInputStream
         const DMContext & context_,
         const ColumnFileSetSnapshotPtr & delta_snap_,
         const ColumnDefinesPtr & col_defs_,
-        const RowKeyRange & segment_range_)
-        : reader(context_, delta_snap_, col_defs_, segment_range_)
+        const RowKeyRange & segment_range_,
+        ReadTag read_tag_)
+        : reader(context_, delta_snap_, col_defs_, segment_range_, read_tag_)
         , column_files(reader.snapshot->getColumnFiles())
         , column_files_count(column_files.size())
     {}

dbms/src/Storages/DeltaMerge/Delta/DeltaValueSpace.h

@@ -427,11 +427,12 @@ class DeltaValueReader
         const DMContext & context_,
         const DeltaSnapshotPtr & delta_snap_,
         const ColumnDefinesPtr & col_defs_,
-        const RowKeyRange & segment_range_);
+        const RowKeyRange & segment_range_,
+        ReadTag read_tag_);
 
     // If we need to read columns besides pk and version, a DeltaValueReader can NOT be used more than once.
     // This method create a new reader based on then current one. It will reuse some caches in the current reader.
-    DeltaValueReaderPtr createNewReader(const ColumnDefinesPtr & new_col_defs);
+    DeltaValueReaderPtr createNewReader(const ColumnDefinesPtr & new_col_defs, ReadTag read_tag);
 
     void setDeltaIndex(const DeltaIndexCompactedPtr & delta_index_) { compacted_delta_index = delta_index_; }
 
@@ -473,9 +474,15 @@ class DeltaValueInputStream : public SkippableBlockInputStream
         const DMContext & context_,
         const DeltaSnapshotPtr & delta_snap_,
         const ColumnDefinesPtr & col_defs_,
-        const RowKeyRange & segment_range_)
-        : mem_table_input_stream(context_, delta_snap_->getMemTableSetSnapshot(), col_defs_, segment_range_)
-        , persisted_files_input_stream(context_, delta_snap_->getPersistedFileSetSnapshot(), col_defs_, segment_range_)
+        const RowKeyRange & segment_range_,
+        ReadTag read_tag_)
+        : mem_table_input_stream(context_, delta_snap_->getMemTableSetSnapshot(), col_defs_, segment_range_, read_tag_)
+        , persisted_files_input_stream(
+              context_,
+              delta_snap_->getPersistedFileSetSnapshot(),
+              col_defs_,
+              segment_range_,
+              read_tag_)
     {}
 
     String getName() const override { return "DeltaValue"; }

dbms/src/Storages/DeltaMerge/Delta/Snapshot.cpp

@@ -71,29 +71,32 @@ DeltaValueReader::DeltaValueReader(
     const DMContext & context,
     const DeltaSnapshotPtr & delta_snap_,
     const ColumnDefinesPtr & col_defs_,
-    const RowKeyRange & segment_range_)
+    const RowKeyRange & segment_range_,
+    ReadTag read_tag_)
     : delta_snap(delta_snap_)
     , mem_table_reader(std::make_shared<ColumnFileSetReader>(
           context,
           delta_snap_->getMemTableSetSnapshot(),
           col_defs_,
-          segment_range_))
+          segment_range_,
+          read_tag_))
     , persisted_files_reader(std::make_shared<ColumnFileSetReader>(
           context,
           delta_snap_->getPersistedFileSetSnapshot(),
           col_defs_,
-          segment_range_))
+          segment_range_,
+          read_tag_))
     , col_defs(col_defs_)
     , segment_range(segment_range_)
 {}
 
-DeltaValueReaderPtr DeltaValueReader::createNewReader(const ColumnDefinesPtr & new_col_defs)
+DeltaValueReaderPtr DeltaValueReader::createNewReader(const ColumnDefinesPtr & new_col_defs, ReadTag read_tag)
 {
     auto * new_reader = new DeltaValueReader();
     new_reader->delta_snap = delta_snap;
     new_reader->compacted_delta_index = compacted_delta_index;
-    new_reader->persisted_files_reader = persisted_files_reader->createNewReader(new_col_defs);
-    new_reader->mem_table_reader = mem_table_reader->createNewReader(new_col_defs);
+    new_reader->persisted_files_reader = persisted_files_reader->createNewReader(new_col_defs, read_tag);
+    new_reader->mem_table_reader = mem_table_reader->createNewReader(new_col_defs, read_tag);
     new_reader->col_defs = new_col_defs;
     new_reader->segment_range = segment_range;
 

dbms/src/Storages/DeltaMerge/File/DMFileBlockInputStream.cpp

@@ -89,7 +89,8 @@ DMFileBlockInputStreamPtr DMFileBlockInputStreamBuilder::build(
         read_one_pack_every_time,
         tracing_id,
         max_sharing_column_bytes_for_all,
-        scan_context);
+        scan_context,
+        read_tag);
 
     return std::make_shared<DMFileBlockInputStream>(std::move(reader), max_sharing_column_bytes_for_all > 0);
 }

dbms/src/Storages/DeltaMerge/File/DMFileBlockInputStream.h

@@ -151,6 +151,12 @@ class DMFileBlockInputStreamBuilder
         return *this;
     }
 
+    DMFileBlockInputStreamBuilder & setReadTag(ReadTag read_tag_)
+    {
+        read_tag = read_tag_;
+        return *this;
+    }
+
 private:
     // These methods are called by the ctor
 
@@ -196,6 +202,7 @@ class DMFileBlockInputStreamBuilder
     bool read_one_pack_every_time = false;
     size_t max_sharing_column_bytes_for_all = 0;
     String tracing_id;
+    ReadTag read_tag = ReadTag::Internal;
 };
 
 /**