Skip to content

Commit f62067a

Browse files
JaySon-HuangJaySon-Huang
committed
Output rs_operator to scan_details
Signed-off-by: JaySon-Huang <tshent@qq.com>
1 parent 43f464e commit f62067a

7 files changed

Lines changed: 68 additions & 6 deletions

File tree

‎dbms/src/Storages/DeltaMerge/Filter/In.h‎

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -49,6 +49,20 @@ class In : public RSOperator
4949
return buf.toString();
5050
}
5151

52+
Poco::JSON::Object::Ptr toJSONObject() override
53+
{
54+
Poco::JSON::Object::Ptr obj = new Poco::JSON::Object();
55+
obj->set("op", name());
56+
obj->set("col", attr.col_name);
57+
Poco::JSON::Array arr;
58+
for (const auto & v : values)
59+
{
60+
arr.add(applyVisitor(FieldVisitorToDebugString(), v));
61+
}
62+
obj->set("value", arr);
63+
return obj;
64+
}
65+
5266
RSResults roughCheck(size_t start_pack, size_t pack_count, const RSCheckParam & param) override
5367
{
5468
// If values is empty (for example where a in ()), all packs will not match.

‎dbms/src/Storages/DeltaMerge/Filter/IsNull.h‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -34,6 +34,14 @@ class IsNull : public RSOperator
3434

3535
String toDebugString() override { return fmt::format(R"({{"op":"{}","col":"{}"}})", name(), attr.col_name); }
3636

37+
Poco::JSON::Object::Ptr toJSONObject() override
38+
{
39+
Poco::JSON::Object::Ptr obj = new Poco::JSON::Object();
40+
obj->set("op", name());
41+
obj->set("col", attr.col_name);
42+
return obj;
43+
}
44+
3745
RSResults roughCheck(size_t start_pack, size_t pack_count, const RSCheckParam & param) override
3846
{
3947
auto rs_index = getRSIndex(param, attr);

‎dbms/src/Storages/DeltaMerge/Filter/PushDownExecutor.cpp‎

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -18,7 +18,6 @@
1818
#include <Flash/Coprocessor/DAGUtils.h>
1919
#include <Flash/Coprocessor/InterpreterUtils.h>
2020
#include <Interpreters/Context.h>
21-
#include <Poco/JSON/Object.h>
2221
#include <Storages/DeltaMerge/Filter/PushDownExecutor.h>
2322
#include <Storages/SelectQueryInfo.h>
2423
#include <TiDB/Decode/TypeMapping.h>

‎dbms/src/Storages/DeltaMerge/Filter/PushDownExecutor.h‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -14,9 +14,13 @@
1414

1515
#pragma once
1616

17+
#pragma GCC diagnostic push
18+
#pragma GCC diagnostic ignored "-Wunused-parameter"
19+
#include <Poco/JSON/Object.h>
20+
#pragma GCC diagnostic pop
21+
1722
#include <Flash/Coprocessor/TiDBTableScan.h>
1823
#include <Interpreters/ExpressionActions.h>
19-
#include <Poco/JSON/Object.h>
2024
#include <Storages/DeltaMerge/Filter/RSOperator.h>
2125
#include <Storages/DeltaMerge/Index/FullTextIndex/Reader_fwd.h>
2226
#include <Storages/DeltaMerge/Index/VectorIndex/Reader_fwd.h>

‎dbms/src/Storages/DeltaMerge/Filter/RSOperator.h‎

Lines changed: 26 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,11 @@
1414

1515
#pragma once
1616

17+
#pragma GCC diagnostic push
18+
#pragma GCC diagnostic ignored "-Wunused-parameter"
19+
#include <Poco/JSON/Object.h>
20+
#pragma GCC diagnostic pop
21+
1722
#include <Common/FieldVisitors.h>
1823
#include <Storages/DeltaMerge/DeltaMergeDefines.h>
1924
#include <Storages/DeltaMerge/Filter/ColumnRange.h>
@@ -49,6 +54,7 @@ class RSOperator
4954

5055
virtual String name() = 0;
5156
virtual String toDebugString() = 0;
57+
virtual Poco::JSON::Object::Ptr toJSONObject() = 0;
5258

5359
virtual RSResults roughCheck(size_t start_pack, size_t pack_count, const RSCheckParam & param) = 0;
5460

@@ -87,6 +93,14 @@ class ColCmpVal : public RSOperator
8793
attr.col_name,
8894
applyVisitor(FieldVisitorToDebugString(), value));
8995
}
96+
Poco::JSON::Object::Ptr toJSONObject() override
97+
{
98+
Poco::JSON::Object::Ptr obj = new Poco::JSON::Object();
99+
obj->set("op", name());
100+
obj->set("col", attr.col_name);
101+
obj->set("value", applyVisitor(FieldVisitorToDebugString(), value));
102+
return obj;
103+
}
90104
};
91105

92106

@@ -123,6 +137,18 @@ class LogicalOp : public RSOperator
123137
buf.append("]}");
124138
return buf.toString();
125139
}
140+
Poco::JSON::Object::Ptr toJSONObject() override
141+
{
142+
Poco::JSON::Object::Ptr obj = new Poco::JSON::Object();
143+
obj->set("op", name());
144+
Poco::JSON::Array arr;
145+
for (const auto & child : children)
146+
{
147+
arr.add(child->toJSONObject());
148+
}
149+
obj->set("children", arr);
150+
return obj;
151+
}
126152
};
127153

128154
inline std::optional<RSIndex> getRSIndex(const RSCheckParam & param, const Attr & attr)

‎dbms/src/Storages/DeltaMerge/Filter/Unsupported.h‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -34,6 +34,14 @@ class Unsupported : public RSOperator
3434

3535
String toDebugString() override { return fmt::format(R"({{"op":"{}","reason":"{}"}})", name(), reason); }
3636

37+
Poco::JSON::Object::Ptr toJSONObject() override
38+
{
39+
Poco::JSON::Object::Ptr obj = new Poco::JSON::Object();
40+
obj->set("op", name());
41+
obj->set("reason", reason);
42+
return obj;
43+
}
44+
3745
RSResults roughCheck(size_t /*start_pack*/, size_t pack_count, const RSCheckParam & /*param*/) override
3846
{
3947
return RSResults(pack_count, RSResult::Some);

‎dbms/src/Storages/StorageDeltaMerge.cpp‎

Lines changed: 7 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -48,6 +48,7 @@
4848
#include <Storages/DeltaMerge/Index/LocalIndexInfo.h>
4949
#include <Storages/DeltaMerge/Index/VectorIndex/Stream/Ctx.h>
5050
#include <Storages/DeltaMerge/Remote/DisaggSnapshot.h>
51+
#include <Storages/DeltaMerge/ScanContext.h>
5152
#include <Storages/KVStore/Region.h>
5253
#include <Storages/KVStore/TMTContext.h>
5354
#include <Storages/KVStore/TiKVHelpers/TiKVRecordFormat.h>
@@ -863,7 +864,7 @@ BlockInputStreams StorageDeltaMerge::read(
863864
query_info.req_id,
864865
tracing_logger);
865866

866-
auto filter = PushDownExecutor::build(
867+
auto pushdown_executor = PushDownExecutor::build(
867868
query_info,
868869
columns_to_read,
869870
store->getTableColumns(),
@@ -875,6 +876,7 @@ BlockInputStreams StorageDeltaMerge::read(
875876
auto runtime_filter_list = parseRuntimeFilterList(query_info, store->getTableColumns(), context, tracing_logger);
876877

877878
const auto & scan_context = mvcc_query_info.scan_context;
879+
scan_context->pushdown_executor = pushdown_executor;
878880

879881
auto streams = store->read(
880882
context,
@@ -883,7 +885,7 @@ BlockInputStreams StorageDeltaMerge::read(
883885
ranges,
884886
num_streams,
885887
/*start_ts=*/mvcc_query_info.start_ts,
886-
filter,
888+
pushdown_executor,
887889
runtime_filter_list,
888890
query_info.dag_query ? query_info.dag_query->rf_max_wait_time_ms : 0,
889891
query_info.req_id,
@@ -953,7 +955,7 @@ void StorageDeltaMerge::read(
953955
query_info.req_id,
954956
tracing_logger);
955957

956-
auto filter = PushDownExecutor::build(
958+
auto pushdown_executor = PushDownExecutor::build(
957959
query_info,
958960
columns_to_read,
959961
store->getTableColumns(),
@@ -965,6 +967,7 @@ void StorageDeltaMerge::read(
965967
auto runtime_filter_list = parseRuntimeFilterList(query_info, store->getTableColumns(), context, tracing_logger);
966968

967969
const auto & scan_context = mvcc_query_info.scan_context;
970+
scan_context->pushdown_executor = pushdown_executor;
968971

969972
store->read(
970973
exec_context_,
@@ -975,7 +978,7 @@ void StorageDeltaMerge::read(
975978
ranges,
976979
num_streams,
977980
/*start_ts=*/mvcc_query_info.start_ts,
978-
filter,
981+
pushdown_executor,
979982
runtime_filter_list,
980983
query_info.dag_query ? query_info.dag_query->rf_max_wait_time_ms : 0,
981984
query_info.req_id,

0 commit comments

Comments
 (0)