diff --git a/dbms/src/Flash/Planner/PhysicalPlan.cpp b/dbms/src/Flash/Planner/PhysicalPlan.cpp
index 1384457729e..c3654c435ab 100644
--- a/dbms/src/Flash/Planner/PhysicalPlan.cpp
+++ b/dbms/src/Flash/Planner/PhysicalPlan.cpp
@@ -54,8 +54,18 @@ void fillOrderForListBasedExecutors(DAGContext & dag_context, const PhysicalPlan
     auto & list_based_executors_order = dag_context.list_based_executors_order;
     PhysicalPlanVisitor::visitPostOrder(root_node, [&](const PhysicalPlanNodePtr & plan) {
         assert(plan);
-        if (plan->isRecordProfileStreams())
-            list_based_executors_order.push_back(plan->execId());
+        if (plan->isTiDBOperator())
+        {
+            if (plan->tp() == PlanType::TableScan)
+            {
+                auto physical_table_scan = std::static_pointer_cast<PhysicalTableScan>(plan);
+                if (physical_table_scan->hasPushDownFilter())
+                    list_based_executors_order.push_back(physical_table_scan->getPushDownFilterId());
+                list_based_executors_order.push_back(physical_table_scan->execId());
+            }
+            else
+                list_based_executors_order.push_back(plan->execId());
+        }
     });
 }
 } // namespace
diff --git a/dbms/src/Flash/Planner/PhysicalPlanNode.cpp b/dbms/src/Flash/Planner/PhysicalPlanNode.cpp
index 82482548737..b16a2e544b7 100644
--- a/dbms/src/Flash/Planner/PhysicalPlanNode.cpp
+++ b/dbms/src/Flash/Planner/PhysicalPlanNode.cpp
@@ -45,10 +45,10 @@ String PhysicalPlanNode::toString()
         return buffer.toString();
     };
     return fmt::format(
-        "<{}, {}> | is_record_profile_streams: {}, schema: {}",
+        "<{}, {}> | is_tidb_operator: {}, schema: {}",
         type.toString(),
         executor_id,
-        is_record_profile_streams,
+        is_tidb_operator,
         schema_to_string());
 }
 
@@ -59,14 +59,20 @@ void PhysicalPlanNode::finalize()
 
 void PhysicalPlanNode::recordProfileStreams(DAGPipeline & pipeline, const Context & context)
 {
-    auto & profile_streams = context.getDAGContext()->getProfileStreamsMap()[executor_id];
-    pipeline.transform([&profile_streams](auto & stream) { profile_streams.push_back(stream); });
+    auto & profile_streams_map = context.getDAGContext()->getProfileStreamsMap();
+    /// The profile stream of some operators has been recorded.
+    /// For example, `DAGStorageInterpreter` records the profile streams of PhysicalTableScan.
+    if (profile_streams_map.find(executor_id) == profile_streams_map.end())
+    {
+        auto & profile_streams = profile_streams_map[executor_id];
+        pipeline.transform([&profile_streams](auto & stream) { profile_streams.push_back(stream); });
+    }
 }
 
 void PhysicalPlanNode::transform(DAGPipeline & pipeline, Context & context, size_t max_streams)
 {
     transformImpl(pipeline, context, max_streams);
-    if (is_record_profile_streams)
+    if (is_tidb_operator)
         recordProfileStreams(pipeline, context);
     if (is_restore_concurrency)
     {
diff --git a/dbms/src/Flash/Planner/PhysicalPlanNode.h b/dbms/src/Flash/Planner/PhysicalPlanNode.h
index e4437fe27ef..397f7e037f9 100644
--- a/dbms/src/Flash/Planner/PhysicalPlanNode.h
+++ b/dbms/src/Flash/Planner/PhysicalPlanNode.h
@@ -62,9 +62,9 @@ class PhysicalPlanNode
     /// Obtain a sample block that contains the names and types of result columns.
     virtual const Block & getSampleBlock() const = 0;
 
-    bool isRecordProfileStreams() const { return is_record_profile_streams; }
+    bool isTiDBOperator() const { return is_tidb_operator; }
 
-    void disableRecordProfileStreams() { is_record_profile_streams = false; }
+    void notTiDBOperator() { is_tidb_operator = false; }
 
     void disableRestoreConcurrency() { is_restore_concurrency = false; }
 
@@ -79,7 +79,7 @@ class PhysicalPlanNode
     PlanType type;
     NamesAndTypes schema;
 
-    bool is_record_profile_streams = true;
+    bool is_tidb_operator = true;
     bool is_restore_concurrency = true;
 
     LoggerPtr log;
diff --git a/dbms/src/Flash/Planner/plans/PhysicalProjection.cpp b/dbms/src/Flash/Planner/plans/PhysicalProjection.cpp
index 34b9a7688e4..f2e13417ce4 100644
--- a/dbms/src/Flash/Planner/plans/PhysicalProjection.cpp
+++ b/dbms/src/Flash/Planner/plans/PhysicalProjection.cpp
@@ -92,8 +92,8 @@ PhysicalPlanNodePtr PhysicalProjection::buildNonRootFinal(
         child,
         "final projection",
         project_actions);
-    // For final projection, no need to record profile streams.
-    physical_projection->disableRecordProfileStreams();
+    // Final Projection is not a tidb operator, so no need to record profile streams.
+    physical_projection->notTiDBOperator();
     return physical_projection;
 }
 
@@ -137,8 +137,8 @@ PhysicalPlanNodePtr PhysicalProjection::buildRootFinal(
         child,
         "final projection",
         project_actions);
-    // For final projection, no need to record profile streams.
-    physical_projection->disableRecordProfileStreams();
+    // Final Projection is not a tidb operator, so no need to record profile streams.
+    physical_projection->notTiDBOperator();
     return physical_projection;
 }
 
diff --git a/dbms/src/Flash/Planner/plans/PhysicalTableScan.cpp b/dbms/src/Flash/Planner/plans/PhysicalTableScan.cpp
index cb702f60bd1..a8f9bc67081 100644
--- a/dbms/src/Flash/Planner/plans/PhysicalTableScan.cpp
+++ b/dbms/src/Flash/Planner/plans/PhysicalTableScan.cpp
@@ -116,4 +116,10 @@ bool PhysicalTableScan::hasPushDownFilter() const
 {
     return push_down_filter.hasValue();
 }
+
+const String & PhysicalTableScan::getPushDownFilterId() const
+{
+    assert(hasPushDownFilter());
+    return push_down_filter.executor_id;
+}
 } // namespace DB
diff --git a/dbms/src/Flash/Planner/plans/PhysicalTableScan.h b/dbms/src/Flash/Planner/plans/PhysicalTableScan.h
index f064e69e811..947750950e8 100644
--- a/dbms/src/Flash/Planner/plans/PhysicalTableScan.h
+++ b/dbms/src/Flash/Planner/plans/PhysicalTableScan.h
@@ -40,11 +40,13 @@ class PhysicalTableScan : public PhysicalLeaf
 
     bool pushDownFilter(const String & filter_executor_id, const tipb::Selection & selection);
 
+    bool hasPushDownFilter() const;
+
+    const String & getPushDownFilterId() const;
+
 private:
     void transformImpl(DAGPipeline & pipeline, Context & context, size_t max_streams) override;
 
-    bool hasPushDownFilter() const;
-
 private:
     PushDownFilter push_down_filter;
 
diff --git a/dbms/src/Flash/Planner/tests/gtest_physical_plan.cpp b/dbms/src/Flash/Planner/tests/gtest_physical_plan.cpp
index 9dc3452de32..20ec1150276 100644
--- a/dbms/src/Flash/Planner/tests/gtest_physical_plan.cpp
+++ b/dbms/src/Flash/Planner/tests/gtest_physical_plan.cpp
@@ -133,9 +133,9 @@ try
     execute(
         request,
         /*expected_physical_plan=*/R"(
-<Projection, selection_1> | is_record_profile_streams: false, schema: <selection_1_s1, Nullable(String)>, <selection_1_s2, Nullable(String)>
- <Filter, selection_1> | is_record_profile_streams: true, schema: <s1, Nullable(String)>, <s2, Nullable(String)>
-  <MockExchangeReceiver, exchange_receiver_0> | is_record_profile_streams: true, schema: <s1, Nullable(String)>, <s2, Nullable(String)>)",
+<Projection, selection_1> | is_tidb_operator: false, schema: <selection_1_s1, Nullable(String)>, <selection_1_s2, Nullable(String)>
+ <Filter, selection_1> | is_tidb_operator: true, schema: <s1, Nullable(String)>, <s2, Nullable(String)>
+  <MockExchangeReceiver, exchange_receiver_0> | is_tidb_operator: true, schema: <s1, Nullable(String)>, <s2, Nullable(String)>)",
         /*expected_streams=*/R"(
 Expression: <final projection>
  Filter
@@ -155,9 +155,9 @@ try
     execute(
         request,
         /*expected_physical_plan=*/R"(
-<Projection, limit_1> | is_record_profile_streams: false, schema: <limit_1_s1, Nullable(String)>, <limit_1_s2, Nullable(String)>
- <Limit, limit_1> | is_record_profile_streams: true, schema: <s1, Nullable(String)>, <s2, Nullable(String)>
-  <MockExchangeReceiver, exchange_receiver_0> | is_record_profile_streams: true, schema: <s1, Nullable(String)>, <s2, Nullable(String)>)",
+<Projection, limit_1> | is_tidb_operator: false, schema: <limit_1_s1, Nullable(String)>, <limit_1_s2, Nullable(String)>
+ <Limit, limit_1> | is_tidb_operator: true, schema: <s1, Nullable(String)>, <s2, Nullable(String)>
+  <MockExchangeReceiver, exchange_receiver_0> | is_tidb_operator: true, schema: <s1, Nullable(String)>, <s2, Nullable(String)>)",
         /*expected_streams=*/R"(
 Expression: <final projection>
  Limit, limit = 1
@@ -177,9 +177,9 @@ try
     execute(
         request,
         /*expected_physical_plan=*/R"(
-<Projection, topn_1> | is_record_profile_streams: false, schema: <topn_1_s1, Nullable(String)>, <topn_1_s2, Nullable(String)>
- <TopN, topn_1> | is_record_profile_streams: true, schema: <s1, Nullable(String)>, <s2, Nullable(String)>
-  <MockExchangeReceiver, exchange_receiver_0> | is_record_profile_streams: true, schema: <s1, Nullable(String)>, <s2, Nullable(String)>)",
+<Projection, topn_1> | is_tidb_operator: false, schema: <topn_1_s1, Nullable(String)>, <topn_1_s2, Nullable(String)>
+ <TopN, topn_1> | is_tidb_operator: true, schema: <s1, Nullable(String)>, <s2, Nullable(String)>
+  <MockExchangeReceiver, exchange_receiver_0> | is_tidb_operator: true, schema: <s1, Nullable(String)>, <s2, Nullable(String)>)",
         /*expected_streams=*/R"(
 Expression: <final projection>
  MergeSorting, limit = 1
@@ -201,9 +201,9 @@ try
     execute(
         request,
         /*expected_physical_plan=*/R"(
-<Projection, aggregation_1> | is_record_profile_streams: false, schema: <aggregation_1_max(s2)_collator_46 , Nullable(String)>, <aggregation_1_any(s1)_collator_46 , Nullable(String)>
- <Aggregation, aggregation_1> | is_record_profile_streams: true, schema: <max(s2)_collator_46 , Nullable(String)>, <any(s1)_collator_46 , Nullable(String)>
-  <MockExchangeReceiver, exchange_receiver_0> | is_record_profile_streams: true, schema: <s1, Nullable(String)>, <s2, Nullable(String)>)",
+<Projection, aggregation_1> | is_tidb_operator: false, schema: <aggregation_1_max(s2)_collator_46 , Nullable(String)>, <aggregation_1_any(s1)_collator_46 , Nullable(String)>
+ <Aggregation, aggregation_1> | is_tidb_operator: true, schema: <max(s2)_collator_46 , Nullable(String)>, <any(s1)_collator_46 , Nullable(String)>
+  <MockExchangeReceiver, exchange_receiver_0> | is_tidb_operator: true, schema: <s1, Nullable(String)>, <s2, Nullable(String)>)",
         /*expected_streams=*/R"(
 Expression: <final projection>
  Expression: <expr after aggregation>
@@ -225,9 +225,9 @@ try
     execute(
         request,
         /*expected_physical_plan=*/R"(
-<Projection, project_1> | is_record_profile_streams: false, schema: <project_1_tidbConcat(s1, s2)_collator_46 , Nullable(String)>
- <Projection, project_1> | is_record_profile_streams: true, schema: <tidbConcat(s1, s2)_collator_46 , Nullable(String)>
-  <MockExchangeReceiver, exchange_receiver_0> | is_record_profile_streams: true, schema: <s1, Nullable(String)>, <s2, Nullable(String)>)",
+<Projection, project_1> | is_tidb_operator: false, schema: <project_1_tidbConcat(s1, s2)_collator_46 , Nullable(String)>
+ <Projection, project_1> | is_tidb_operator: true, schema: <tidbConcat(s1, s2)_collator_46 , Nullable(String)>
+  <MockExchangeReceiver, exchange_receiver_0> | is_tidb_operator: true, schema: <s1, Nullable(String)>, <s2, Nullable(String)>)",
         /*expected_streams=*/R"(
 Expression: <final projection>
  Expression: <projection>
@@ -246,9 +246,9 @@ try
     execute(
         request,
         /*expected_physical_plan=*/R"(
-<MockExchangeSender, exchange_sender_1> | is_record_profile_streams: true, schema: <exchange_sender_1_s1, Nullable(String)>, <exchange_sender_1_s2, Nullable(String)>
- <Projection, exchange_receiver_0> | is_record_profile_streams: false, schema: <exchange_sender_1_s1, Nullable(String)>, <exchange_sender_1_s2, Nullable(String)>
-  <MockExchangeReceiver, exchange_receiver_0> | is_record_profile_streams: true, schema: <s1, Nullable(String)>, <s2, Nullable(String)>)",
+<MockExchangeSender, exchange_sender_1> | is_tidb_operator: true, schema: <exchange_sender_1_s1, Nullable(String)>, <exchange_sender_1_s2, Nullable(String)>
+ <Projection, exchange_receiver_0> | is_tidb_operator: false, schema: <exchange_sender_1_s1, Nullable(String)>, <exchange_sender_1_s2, Nullable(String)>
+  <MockExchangeReceiver, exchange_receiver_0> | is_tidb_operator: true, schema: <s1, Nullable(String)>, <s2, Nullable(String)>)",
         /*expected_streams=*/R"(
 MockExchangeSender
  Expression: <final projection>
@@ -267,8 +267,8 @@ try
     execute(
         request,
         /*expected_physical_plan=*/R"(
-<Projection, exchange_receiver_0> | is_record_profile_streams: false, schema: <exchange_receiver_0_s1, Nullable(String)>, <exchange_receiver_0_s2, Nullable(String)>
- <MockExchangeReceiver, exchange_receiver_0> | is_record_profile_streams: true, schema: <s1, Nullable(String)>, <s2, Nullable(String)>)",
+<Projection, exchange_receiver_0> | is_tidb_operator: false, schema: <exchange_receiver_0_s1, Nullable(String)>, <exchange_receiver_0_s2, Nullable(String)>
+ <MockExchangeReceiver, exchange_receiver_0> | is_tidb_operator: true, schema: <s1, Nullable(String)>, <s2, Nullable(String)>)",
         /*expected_streams=*/R"(
 Expression: <final projection>
  MockExchangeReceiver)",
@@ -295,10 +295,10 @@ try
     execute(
         request,
         /*expected_physical_plan=*/R"(
-<Projection, window_2> | is_record_profile_streams: false, schema: <window_2_partition, Nullable(Int64)>, <window_2_order, Nullable(Int64)>, <window_2_CAST(row_number()_collator , Nullable(Int64)_String)_collator_0 , Nullable(Int64)>
- <Window, window_2> | is_record_profile_streams: true, schema: <partition, Nullable(Int64)>, <order, Nullable(Int64)>, <CAST(row_number()_collator , Nullable(Int64)_String)_collator_0 , Nullable(Int64)>
-  <WindowSort, sort_1> | is_record_profile_streams: true, schema: <partition, Nullable(Int64)>, <order, Nullable(Int64)>
-   <MockExchangeReceiver, exchange_receiver_0> | is_record_profile_streams: true, schema: <partition, Nullable(Int64)>, <order, Nullable(Int64)>)",
+<Projection, window_2> | is_tidb_operator: false, schema: <window_2_partition, Nullable(Int64)>, <window_2_order, Nullable(Int64)>, <window_2_CAST(row_number()_collator , Nullable(Int64)_String)_collator_0 , Nullable(Int64)>
+ <Window, window_2> | is_tidb_operator: true, schema: <partition, Nullable(Int64)>, <order, Nullable(Int64)>, <CAST(row_number()_collator , Nullable(Int64)_String)_collator_0 , Nullable(Int64)>
+  <WindowSort, sort_1> | is_tidb_operator: true, schema: <partition, Nullable(Int64)>, <order, Nullable(Int64)>
+   <MockExchangeReceiver, exchange_receiver_0> | is_tidb_operator: true, schema: <partition, Nullable(Int64)>, <order, Nullable(Int64)>)",
         /*expected_streams=*/R"(
 Expression: <final projection>
  Expression: <cast after window>
@@ -314,10 +314,10 @@ Expression: <final projection>
     execute(
         request,
         /*expected_physical_plan=*/R"(
-<Projection, window_2> | is_record_profile_streams: false, schema: <window_2_partition, Nullable(Int64)>, <window_2_order, Nullable(Int64)>, <window_2_CAST(row_number()_collator , Nullable(Int64)_String)_collator_0 , Nullable(Int64)>
- <Window, window_2> | is_record_profile_streams: true, schema: <partition, Nullable(Int64)>, <order, Nullable(Int64)>, <CAST(row_number()_collator , Nullable(Int64)_String)_collator_0 , Nullable(Int64)>
-  <WindowSort, sort_1> | is_record_profile_streams: true, schema: <partition, Nullable(Int64)>, <order, Nullable(Int64)>
-   <MockExchangeReceiver, exchange_receiver_0> | is_record_profile_streams: true, schema: <partition, Nullable(Int64)>, <order, Nullable(Int64)>)",
+<Projection, window_2> | is_tidb_operator: false, schema: <window_2_partition, Nullable(Int64)>, <window_2_order, Nullable(Int64)>, <window_2_CAST(row_number()_collator , Nullable(Int64)_String)_collator_0 , Nullable(Int64)>
+ <Window, window_2> | is_tidb_operator: true, schema: <partition, Nullable(Int64)>, <order, Nullable(Int64)>, <CAST(row_number()_collator , Nullable(Int64)_String)_collator_0 , Nullable(Int64)>
+  <WindowSort, sort_1> | is_tidb_operator: true, schema: <partition, Nullable(Int64)>, <order, Nullable(Int64)>
+   <MockExchangeReceiver, exchange_receiver_0> | is_tidb_operator: true, schema: <partition, Nullable(Int64)>, <order, Nullable(Int64)>)",
         /*expected_streams=*/R"(
 Expression: <final projection>
  Expression: <cast after window>
@@ -340,8 +340,8 @@ try
     execute(
         request,
         /*expected_physical_plan=*/R"(
-<Projection, table_scan_0> | is_record_profile_streams: false, schema: <table_scan_0_s1, Nullable(String)>, <table_scan_0_s2, Nullable(String)>
- <MockTableScan, table_scan_0> | is_record_profile_streams: true, schema: <s1, Nullable(String)>, <s2, Nullable(String)>)",
+<Projection, table_scan_0> | is_tidb_operator: false, schema: <table_scan_0_s1, Nullable(String)>, <table_scan_0_s2, Nullable(String)>
+ <MockTableScan, table_scan_0> | is_tidb_operator: true, schema: <s1, Nullable(String)>, <s2, Nullable(String)>)",
         /*expected_streams=*/R"(
 Expression: <final projection>
  MockTableScan)",
@@ -366,12 +366,12 @@ try
         execute(
             request,
             /*expected_physical_plan=*/R"(
-<Projection, Join_2> | is_record_profile_streams: false, schema: <Join_2_Join_2_l_s, Nullable(String)>, <Join_2_Join_2_l_join_c, Nullable(String)>, <Join_2_Join_2_r_s, Nullable(String)>, <Join_2_Join_2_r_join_c, Nullable(String)>
- <Join, Join_2> | is_record_profile_streams: true, schema: <Join_2_l_s, Nullable(String)>, <Join_2_l_join_c, Nullable(String)>, <Join_2_r_s, Nullable(String)>, <Join_2_r_join_c, Nullable(String)>
-  <Projection, exchange_receiver_0> | is_record_profile_streams: false, schema: <Join_2_l_s, Nullable(String)>, <Join_2_l_join_c, Nullable(String)>
-   <MockExchangeReceiver, exchange_receiver_0> | is_record_profile_streams: true, schema: <s, Nullable(String)>, <join_c, Nullable(String)>
-  <Projection, exchange_receiver_1> | is_record_profile_streams: false, schema: <Join_2_r_s, Nullable(String)>, <Join_2_r_join_c, Nullable(String)>
-   <MockExchangeReceiver, exchange_receiver_1> | is_record_profile_streams: true, schema: <s, Nullable(String)>, <join_c, Nullable(String)>)",
+<Projection, Join_2> | is_tidb_operator: false, schema: <Join_2_Join_2_l_s, Nullable(String)>, <Join_2_Join_2_l_join_c, Nullable(String)>, <Join_2_Join_2_r_s, Nullable(String)>, <Join_2_Join_2_r_join_c, Nullable(String)>
+ <Join, Join_2> | is_tidb_operator: true, schema: <Join_2_l_s, Nullable(String)>, <Join_2_l_join_c, Nullable(String)>, <Join_2_r_s, Nullable(String)>, <Join_2_r_join_c, Nullable(String)>
+  <Projection, exchange_receiver_0> | is_tidb_operator: false, schema: <Join_2_l_s, Nullable(String)>, <Join_2_l_join_c, Nullable(String)>
+   <MockExchangeReceiver, exchange_receiver_0> | is_tidb_operator: true, schema: <s, Nullable(String)>, <join_c, Nullable(String)>
+  <Projection, exchange_receiver_1> | is_tidb_operator: false, schema: <Join_2_r_s, Nullable(String)>, <Join_2_r_join_c, Nullable(String)>
+   <MockExchangeReceiver, exchange_receiver_1> | is_tidb_operator: true, schema: <s, Nullable(String)>, <join_c, Nullable(String)>)",
             /*expected_streams=*/R"(
 CreatingSets
  HashJoinBuildBlockInputStream: <join build, build_side_root_executor_id = exchange_receiver_1>, join_kind = Inner
@@ -393,12 +393,12 @@ CreatingSets
         execute(
             request,
             /*expected_physical_plan=*/R"(
-<Projection, Join_2> | is_record_profile_streams: false, schema: <Join_2_Join_2_l_s, Nullable(String)>, <Join_2_Join_2_l_join_c, Nullable(String)>, <Join_2_Join_2_r_s, Nullable(String)>, <Join_2_Join_2_r_join_c, Nullable(String)>
- <Join, Join_2> | is_record_profile_streams: true, schema: <Join_2_l_s, Nullable(String)>, <Join_2_l_join_c, Nullable(String)>, <Join_2_r_s, Nullable(String)>, <Join_2_r_join_c, Nullable(String)>
-  <Projection, exchange_receiver_0> | is_record_profile_streams: false, schema: <Join_2_l_s, Nullable(String)>, <Join_2_l_join_c, Nullable(String)>
-   <MockExchangeReceiver, exchange_receiver_0> | is_record_profile_streams: true, schema: <s, Nullable(String)>, <join_c, Nullable(String)>
-  <Projection, exchange_receiver_1> | is_record_profile_streams: false, schema: <Join_2_r_s, Nullable(String)>, <Join_2_r_join_c, Nullable(String)>
-   <MockExchangeReceiver, exchange_receiver_1> | is_record_profile_streams: true, schema: <s, Nullable(String)>, <join_c, Nullable(String)>)",
+<Projection, Join_2> | is_tidb_operator: false, schema: <Join_2_Join_2_l_s, Nullable(String)>, <Join_2_Join_2_l_join_c, Nullable(String)>, <Join_2_Join_2_r_s, Nullable(String)>, <Join_2_Join_2_r_join_c, Nullable(String)>
+ <Join, Join_2> | is_tidb_operator: true, schema: <Join_2_l_s, Nullable(String)>, <Join_2_l_join_c, Nullable(String)>, <Join_2_r_s, Nullable(String)>, <Join_2_r_join_c, Nullable(String)>
+  <Projection, exchange_receiver_0> | is_tidb_operator: false, schema: <Join_2_l_s, Nullable(String)>, <Join_2_l_join_c, Nullable(String)>
+   <MockExchangeReceiver, exchange_receiver_0> | is_tidb_operator: true, schema: <s, Nullable(String)>, <join_c, Nullable(String)>
+  <Projection, exchange_receiver_1> | is_tidb_operator: false, schema: <Join_2_r_s, Nullable(String)>, <Join_2_r_join_c, Nullable(String)>
+   <MockExchangeReceiver, exchange_receiver_1> | is_tidb_operator: true, schema: <s, Nullable(String)>, <join_c, Nullable(String)>)",
             /*expected_streams=*/R"(
 CreatingSets
  HashJoinBuildBlockInputStream: <join build, build_side_root_executor_id = exchange_receiver_1>, join_kind = Left
@@ -420,12 +420,12 @@ CreatingSets
         execute(
             request,
             /*expected_physical_plan=*/R"(
-<Projection, Join_2> | is_record_profile_streams: false, schema: <Join_2_Join_2_l_s, Nullable(String)>, <Join_2_Join_2_l_join_c, Nullable(String)>, <Join_2_Join_2_r_s, Nullable(String)>, <Join_2_Join_2_r_join_c, Nullable(String)>
- <Join, Join_2> | is_record_profile_streams: true, schema: <Join_2_l_s, Nullable(String)>, <Join_2_l_join_c, Nullable(String)>, <Join_2_r_s, Nullable(String)>, <Join_2_r_join_c, Nullable(String)>
-  <Projection, exchange_receiver_0> | is_record_profile_streams: false, schema: <Join_2_l_s, Nullable(String)>, <Join_2_l_join_c, Nullable(String)>
-   <MockExchangeReceiver, exchange_receiver_0> | is_record_profile_streams: true, schema: <s, Nullable(String)>, <join_c, Nullable(String)>
-  <Projection, exchange_receiver_1> | is_record_profile_streams: false, schema: <Join_2_r_s, Nullable(String)>, <Join_2_r_join_c, Nullable(String)>
-   <MockExchangeReceiver, exchange_receiver_1> | is_record_profile_streams: true, schema: <s, Nullable(String)>, <join_c, Nullable(String)>)",
+<Projection, Join_2> | is_tidb_operator: false, schema: <Join_2_Join_2_l_s, Nullable(String)>, <Join_2_Join_2_l_join_c, Nullable(String)>, <Join_2_Join_2_r_s, Nullable(String)>, <Join_2_Join_2_r_join_c, Nullable(String)>
+ <Join, Join_2> | is_tidb_operator: true, schema: <Join_2_l_s, Nullable(String)>, <Join_2_l_join_c, Nullable(String)>, <Join_2_r_s, Nullable(String)>, <Join_2_r_join_c, Nullable(String)>
+  <Projection, exchange_receiver_0> | is_tidb_operator: false, schema: <Join_2_l_s, Nullable(String)>, <Join_2_l_join_c, Nullable(String)>
+   <MockExchangeReceiver, exchange_receiver_0> | is_tidb_operator: true, schema: <s, Nullable(String)>, <join_c, Nullable(String)>
+  <Projection, exchange_receiver_1> | is_tidb_operator: false, schema: <Join_2_r_s, Nullable(String)>, <Join_2_r_join_c, Nullable(String)>
+   <MockExchangeReceiver, exchange_receiver_1> | is_tidb_operator: true, schema: <s, Nullable(String)>, <join_c, Nullable(String)>)",
             /*expected_streams=*/R"(
 CreatingSets
  HashJoinBuildBlockInputStream: <join build, build_side_root_executor_id = exchange_receiver_1>, join_kind = Right
@@ -459,20 +459,20 @@ CreatingSets
         execute(
             request,
             /*expected_physical_plan=*/R"(
-<Projection, Join_6> | is_record_profile_streams: false, schema: <Join_6_Join_6_l_Join_4_l_a, Nullable(Int32)>, <Join_6_Join_6_l_Join_4_l_b, Nullable(Int32)>, <Join_6_Join_6_l_Join_4_l_c, Nullable(Int32)>, <Join_6_CAST(Join_6_l_Join_4_r_a, Nullable(Int32)_String)_collator_0 , Nullable(Int32)>, <Join_6_CAST(Join_6_l_Join_4_r_b, Nullable(Int32)_String)_collator_0 , Nullable(Int32)>, <Join_6_CAST(Join_6_l_Join_4_r_c, Nullable(Int32)_String)_collator_0 , Nullable(Int32)>, <Join_6_Join_6_r_Join_5_l_a, Nullable(Int32)>, <Join_6_Join_6_r_Join_5_l_b, Nullable(Int32)>, <Join_6_CAST(Join_6_r_Join_5_r_a, Nullable(Int32)_String)_collator_0 , Nullable(Int32)>, <Join_6_CAST(Join_6_r_Join_5_r_b, Nullable(Int32)_String)_collator_0 , Nullable(Int32)>
- <Join, Join_6> | is_record_profile_streams: true, schema: <Join_6_l_Join_4_l_a, Nullable(Int32)>, <Join_6_l_Join_4_l_b, Nullable(Int32)>, <Join_6_l_Join_4_l_c, Nullable(Int32)>, <Join_6_l_Join_4_r_a, Int32>, <Join_6_l_Join_4_r_b, Int32>, <Join_6_l_Join_4_r_c, Int32>, <Join_6_r_Join_5_l_a, Nullable(Int32)>, <Join_6_r_Join_5_l_b, Nullable(Int32)>, <Join_6_r_Join_5_r_a, Int32>, <Join_6_r_Join_5_r_b, Int32>
-  <Projection, Join_4> | is_record_profile_streams: false, schema: <Join_6_l_Join_4_l_a, Nullable(Int32)>, <Join_6_l_Join_4_l_b, Nullable(Int32)>, <Join_6_l_Join_4_l_c, Nullable(Int32)>, <Join_6_l_Join_4_r_a, Int32>, <Join_6_l_Join_4_r_b, Int32>, <Join_6_l_Join_4_r_c, Int32>
-   <Join, Join_4> | is_record_profile_streams: true, schema: <Join_4_l_a, Nullable(Int32)>, <Join_4_l_b, Nullable(Int32)>, <Join_4_l_c, Nullable(Int32)>, <Join_4_r_a, Int32>, <Join_4_r_b, Int32>, <Join_4_r_c, Int32>
-    <Projection, table_scan_0> | is_record_profile_streams: false, schema: <Join_4_l_a, Int32>, <Join_4_l_b, Int32>, <Join_4_l_c, Int32>
-     <MockTableScan, table_scan_0> | is_record_profile_streams: true, schema: <a, Int32>, <b, Int32>, <c, Int32>
-    <Projection, table_scan_1> | is_record_profile_streams: false, schema: <Join_4_r_a, Int32>, <Join_4_r_b, Int32>, <Join_4_r_c, Int32>
-     <MockTableScan, table_scan_1> | is_record_profile_streams: true, schema: <a, Int32>, <b, Int32>, <c, Int32>
-  <Projection, Join_5> | is_record_profile_streams: false, schema: <Join_6_r_Join_5_l_a, Nullable(Int32)>, <Join_6_r_Join_5_l_b, Nullable(Int32)>, <Join_6_r_Join_5_r_a, Int32>, <Join_6_r_Join_5_r_b, Int32>
-   <Join, Join_5> | is_record_profile_streams: true, schema: <Join_5_l_a, Nullable(Int32)>, <Join_5_l_b, Nullable(Int32)>, <Join_5_r_a, Int32>, <Join_5_r_b, Int32>
-    <Projection, table_scan_2> | is_record_profile_streams: false, schema: <Join_5_l_a, Int32>, <Join_5_l_b, Int32>
-     <MockTableScan, table_scan_2> | is_record_profile_streams: true, schema: <a, Int32>, <b, Int32>
-    <Projection, table_scan_3> | is_record_profile_streams: false, schema: <Join_5_r_a, Int32>, <Join_5_r_b, Int32>
-     <MockTableScan, table_scan_3> | is_record_profile_streams: true, schema: <a, Int32>, <b, Int32>)",
+<Projection, Join_6> | is_tidb_operator: false, schema: <Join_6_Join_6_l_Join_4_l_a, Nullable(Int32)>, <Join_6_Join_6_l_Join_4_l_b, Nullable(Int32)>, <Join_6_Join_6_l_Join_4_l_c, Nullable(Int32)>, <Join_6_CAST(Join_6_l_Join_4_r_a, Nullable(Int32)_String)_collator_0 , Nullable(Int32)>, <Join_6_CAST(Join_6_l_Join_4_r_b, Nullable(Int32)_String)_collator_0 , Nullable(Int32)>, <Join_6_CAST(Join_6_l_Join_4_r_c, Nullable(Int32)_String)_collator_0 , Nullable(Int32)>, <Join_6_Join_6_r_Join_5_l_a, Nullable(Int32)>, <Join_6_Join_6_r_Join_5_l_b, Nullable(Int32)>, <Join_6_CAST(Join_6_r_Join_5_r_a, Nullable(Int32)_String)_collator_0 , Nullable(Int32)>, <Join_6_CAST(Join_6_r_Join_5_r_b, Nullable(Int32)_String)_collator_0 , Nullable(Int32)>
+ <Join, Join_6> | is_tidb_operator: true, schema: <Join_6_l_Join_4_l_a, Nullable(Int32)>, <Join_6_l_Join_4_l_b, Nullable(Int32)>, <Join_6_l_Join_4_l_c, Nullable(Int32)>, <Join_6_l_Join_4_r_a, Int32>, <Join_6_l_Join_4_r_b, Int32>, <Join_6_l_Join_4_r_c, Int32>, <Join_6_r_Join_5_l_a, Nullable(Int32)>, <Join_6_r_Join_5_l_b, Nullable(Int32)>, <Join_6_r_Join_5_r_a, Int32>, <Join_6_r_Join_5_r_b, Int32>
+  <Projection, Join_4> | is_tidb_operator: false, schema: <Join_6_l_Join_4_l_a, Nullable(Int32)>, <Join_6_l_Join_4_l_b, Nullable(Int32)>, <Join_6_l_Join_4_l_c, Nullable(Int32)>, <Join_6_l_Join_4_r_a, Int32>, <Join_6_l_Join_4_r_b, Int32>, <Join_6_l_Join_4_r_c, Int32>
+   <Join, Join_4> | is_tidb_operator: true, schema: <Join_4_l_a, Nullable(Int32)>, <Join_4_l_b, Nullable(Int32)>, <Join_4_l_c, Nullable(Int32)>, <Join_4_r_a, Int32>, <Join_4_r_b, Int32>, <Join_4_r_c, Int32>
+    <Projection, table_scan_0> | is_tidb_operator: false, schema: <Join_4_l_a, Int32>, <Join_4_l_b, Int32>, <Join_4_l_c, Int32>
+     <MockTableScan, table_scan_0> | is_tidb_operator: true, schema: <a, Int32>, <b, Int32>, <c, Int32>
+    <Projection, table_scan_1> | is_tidb_operator: false, schema: <Join_4_r_a, Int32>, <Join_4_r_b, Int32>, <Join_4_r_c, Int32>
+     <MockTableScan, table_scan_1> | is_tidb_operator: true, schema: <a, Int32>, <b, Int32>, <c, Int32>
+  <Projection, Join_5> | is_tidb_operator: false, schema: <Join_6_r_Join_5_l_a, Nullable(Int32)>, <Join_6_r_Join_5_l_b, Nullable(Int32)>, <Join_6_r_Join_5_r_a, Int32>, <Join_6_r_Join_5_r_b, Int32>
+   <Join, Join_5> | is_tidb_operator: true, schema: <Join_5_l_a, Nullable(Int32)>, <Join_5_l_b, Nullable(Int32)>, <Join_5_r_a, Int32>, <Join_5_r_b, Int32>
+    <Projection, table_scan_2> | is_tidb_operator: false, schema: <Join_5_l_a, Int32>, <Join_5_l_b, Int32>
+     <MockTableScan, table_scan_2> | is_tidb_operator: true, schema: <a, Int32>, <b, Int32>
+    <Projection, table_scan_3> | is_tidb_operator: false, schema: <Join_5_r_a, Int32>, <Join_5_r_b, Int32>
+     <MockTableScan, table_scan_3> | is_tidb_operator: true, schema: <a, Int32>, <b, Int32>)",
             /*expected_streams=*/R"(
 CreatingSets
  HashJoinBuildBlockInputStream x 2: <join build, build_side_root_executor_id = table_scan_3>, join_kind = Right
@@ -529,20 +529,20 @@ CreatingSets
         execute(
             request,
             /*expected_physical_plan=*/R"(
-<Projection, Join_6> | is_record_profile_streams: false, schema: <Join_6_Join_6_l_Join_4_l_a, Nullable(Int32)>, <Join_6_Join_6_l_Join_4_l_b, Nullable(Int32)>, <Join_6_Join_6_l_Join_4_l_c, Nullable(Int32)>, <Join_6_CAST(Join_6_l_Join_4_r_a, Nullable(Int32)_String)_collator_0 , Nullable(Int32)>, <Join_6_CAST(Join_6_l_Join_4_r_b, Nullable(Int32)_String)_collator_0 , Nullable(Int32)>, <Join_6_CAST(Join_6_l_Join_4_r_c, Nullable(Int32)_String)_collator_0 , Nullable(Int32)>, <Join_6_Join_6_r_Join_5_l_a, Nullable(Int32)>, <Join_6_Join_6_r_Join_5_l_b, Nullable(Int32)>, <Join_6_Join_6_r_Join_5_r_a, Nullable(Int32)>, <Join_6_Join_6_r_Join_5_r_b, Nullable(Int32)>
- <Join, Join_6> | is_record_profile_streams: true, schema: <Join_6_l_Join_4_l_a, Nullable(Int32)>, <Join_6_l_Join_4_l_b, Nullable(Int32)>, <Join_6_l_Join_4_l_c, Nullable(Int32)>, <Join_6_l_Join_4_r_a, Int32>, <Join_6_l_Join_4_r_b, Int32>, <Join_6_l_Join_4_r_c, Int32>, <Join_6_r_Join_5_l_a, Nullable(Int32)>, <Join_6_r_Join_5_l_b, Nullable(Int32)>, <Join_6_r_Join_5_r_a, Nullable(Int32)>, <Join_6_r_Join_5_r_b, Nullable(Int32)>
-  <Projection, Join_4> | is_record_profile_streams: false, schema: <Join_6_l_Join_4_l_a, Nullable(Int32)>, <Join_6_l_Join_4_l_b, Nullable(Int32)>, <Join_6_l_Join_4_l_c, Nullable(Int32)>, <Join_6_l_Join_4_r_a, Int32>, <Join_6_l_Join_4_r_b, Int32>, <Join_6_l_Join_4_r_c, Int32>
-   <Join, Join_4> | is_record_profile_streams: true, schema: <Join_4_l_a, Nullable(Int32)>, <Join_4_l_b, Nullable(Int32)>, <Join_4_l_c, Nullable(Int32)>, <Join_4_r_a, Int32>, <Join_4_r_b, Int32>, <Join_4_r_c, Int32>
-    <Projection, table_scan_0> | is_record_profile_streams: false, schema: <Join_4_l_a, Int32>, <Join_4_l_b, Int32>, <Join_4_l_c, Int32>
-     <MockTableScan, table_scan_0> | is_record_profile_streams: true, schema: <a, Int32>, <b, Int32>, <c, Int32>
-    <Projection, table_scan_1> | is_record_profile_streams: false, schema: <Join_4_r_a, Int32>, <Join_4_r_b, Int32>, <Join_4_r_c, Int32>
-     <MockTableScan, table_scan_1> | is_record_profile_streams: true, schema: <a, Int32>, <b, Int32>, <c, Int32>
-  <Projection, Join_5> | is_record_profile_streams: false, schema: <Join_6_r_Join_5_l_a, Nullable(Int32)>, <Join_6_r_Join_5_l_b, Nullable(Int32)>, <Join_6_r_Join_5_r_a, Int32>, <Join_6_r_Join_5_r_b, Int32>
-   <Join, Join_5> | is_record_profile_streams: true, schema: <Join_5_l_a, Nullable(Int32)>, <Join_5_l_b, Nullable(Int32)>, <Join_5_r_a, Int32>, <Join_5_r_b, Int32>
-    <Projection, table_scan_2> | is_record_profile_streams: false, schema: <Join_5_l_a, Int32>, <Join_5_l_b, Int32>
-     <MockTableScan, table_scan_2> | is_record_profile_streams: true, schema: <a, Int32>, <b, Int32>
-    <Projection, table_scan_3> | is_record_profile_streams: false, schema: <Join_5_r_a, Int32>, <Join_5_r_b, Int32>
-     <MockTableScan, table_scan_3> | is_record_profile_streams: true, schema: <a, Int32>, <b, Int32>)",
+<Projection, Join_6> | is_tidb_operator: false, schema: <Join_6_Join_6_l_Join_4_l_a, Nullable(Int32)>, <Join_6_Join_6_l_Join_4_l_b, Nullable(Int32)>, <Join_6_Join_6_l_Join_4_l_c, Nullable(Int32)>, <Join_6_CAST(Join_6_l_Join_4_r_a, Nullable(Int32)_String)_collator_0 , Nullable(Int32)>, <Join_6_CAST(Join_6_l_Join_4_r_b, Nullable(Int32)_String)_collator_0 , Nullable(Int32)>, <Join_6_CAST(Join_6_l_Join_4_r_c, Nullable(Int32)_String)_collator_0 , Nullable(Int32)>, <Join_6_Join_6_r_Join_5_l_a, Nullable(Int32)>, <Join_6_Join_6_r_Join_5_l_b, Nullable(Int32)>, <Join_6_Join_6_r_Join_5_r_a, Nullable(Int32)>, <Join_6_Join_6_r_Join_5_r_b, Nullable(Int32)>
+ <Join, Join_6> | is_tidb_operator: true, schema: <Join_6_l_Join_4_l_a, Nullable(Int32)>, <Join_6_l_Join_4_l_b, Nullable(Int32)>, <Join_6_l_Join_4_l_c, Nullable(Int32)>, <Join_6_l_Join_4_r_a, Int32>, <Join_6_l_Join_4_r_b, Int32>, <Join_6_l_Join_4_r_c, Int32>, <Join_6_r_Join_5_l_a, Nullable(Int32)>, <Join_6_r_Join_5_l_b, Nullable(Int32)>, <Join_6_r_Join_5_r_a, Nullable(Int32)>, <Join_6_r_Join_5_r_b, Nullable(Int32)>
+  <Projection, Join_4> | is_tidb_operator: false, schema: <Join_6_l_Join_4_l_a, Nullable(Int32)>, <Join_6_l_Join_4_l_b, Nullable(Int32)>, <Join_6_l_Join_4_l_c, Nullable(Int32)>, <Join_6_l_Join_4_r_a, Int32>, <Join_6_l_Join_4_r_b, Int32>, <Join_6_l_Join_4_r_c, Int32>
+   <Join, Join_4> | is_tidb_operator: true, schema: <Join_4_l_a, Nullable(Int32)>, <Join_4_l_b, Nullable(Int32)>, <Join_4_l_c, Nullable(Int32)>, <Join_4_r_a, Int32>, <Join_4_r_b, Int32>, <Join_4_r_c, Int32>
+    <Projection, table_scan_0> | is_tidb_operator: false, schema: <Join_4_l_a, Int32>, <Join_4_l_b, Int32>, <Join_4_l_c, Int32>
+     <MockTableScan, table_scan_0> | is_tidb_operator: true, schema: <a, Int32>, <b, Int32>, <c, Int32>
+    <Projection, table_scan_1> | is_tidb_operator: false, schema: <Join_4_r_a, Int32>, <Join_4_r_b, Int32>, <Join_4_r_c, Int32>
+     <MockTableScan, table_scan_1> | is_tidb_operator: true, schema: <a, Int32>, <b, Int32>, <c, Int32>
+  <Projection, Join_5> | is_tidb_operator: false, schema: <Join_6_r_Join_5_l_a, Nullable(Int32)>, <Join_6_r_Join_5_l_b, Nullable(Int32)>, <Join_6_r_Join_5_r_a, Int32>, <Join_6_r_Join_5_r_b, Int32>
+   <Join, Join_5> | is_tidb_operator: true, schema: <Join_5_l_a, Nullable(Int32)>, <Join_5_l_b, Nullable(Int32)>, <Join_5_r_a, Int32>, <Join_5_r_b, Int32>
+    <Projection, table_scan_2> | is_tidb_operator: false, schema: <Join_5_l_a, Int32>, <Join_5_l_b, Int32>
+     <MockTableScan, table_scan_2> | is_tidb_operator: true, schema: <a, Int32>, <b, Int32>
+    <Projection, table_scan_3> | is_tidb_operator: false, schema: <Join_5_r_a, Int32>, <Join_5_r_b, Int32>
+     <MockTableScan, table_scan_3> | is_tidb_operator: true, schema: <a, Int32>, <b, Int32>)",
             /*expected_streams=*/R"(
 CreatingSets
  HashJoinBuildBlockInputStream x 2: <join build, build_side_root_executor_id = table_scan_3>, join_kind = Right