Skip to content

branch-4.1: [Profile][exec] The profile hitcache not correct in force refresh and sink id not correct #62645 #62818

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
yiguolei merged 1 commit into from
May 1, 2026
Merged

branch-4.1: [Profile][exec] The profile hitcache not correct in force refresh and sink id not correct #62645 #62818

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 4 additions & 2 deletions be/src/exec/operator/cache_source_operator.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -72,9 +72,11 @@ Status CacheSourceLocalState::init(RuntimeState* state, LocalStateInfo& info) {
custom_profile()->add_info_string("CacheTabletId", tablet_ids_str);

// 3. lookup the cache and find proper slot order
hit_cache = _global_cache->lookup(_cache_key, _version, &_query_cache_handle);
if (!cache_param.force_refresh_query_cache) {
hit_cache = _global_cache->lookup(_cache_key, _version, &_query_cache_handle);
}
custom_profile()->add_info_string("HitCache", std::to_string(hit_cache));
if (hit_cache && !cache_param.force_refresh_query_cache) {
if (hit_cache) {
_hit_cache_results = _query_cache_handle.get_cache_result();
auto hit_cache_slot_orders = _query_cache_handle.get_cache_slot_orders();

Expand Down
6 changes: 3 additions & 3 deletions be/src/exec/operator/olap_table_sink_operator.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -38,9 +38,9 @@ class OlapTableSinkLocalState final
class OlapTableSinkOperatorX final : public DataSinkOperatorX<OlapTableSinkLocalState> {
public:
using Base = DataSinkOperatorX<OlapTableSinkLocalState>;
OlapTableSinkOperatorX(ObjectPool* pool, int operator_id, const RowDescriptor& row_desc,
const std::vector<TExpr>& t_output_expr)
: Base(operator_id, 0, 0),
OlapTableSinkOperatorX(ObjectPool* pool, int operator_id, int node_id,
const RowDescriptor& row_desc, const std::vector<TExpr>& t_output_expr)
: Base(operator_id, node_id, node_id),
_row_desc(row_desc),
_t_output_expr(t_output_expr),
_pool(pool) {};
Expand Down
6 changes: 3 additions & 3 deletions be/src/exec/operator/olap_table_sink_v2_operator.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,9 +39,9 @@ class OlapTableSinkV2LocalState final
class OlapTableSinkV2OperatorX final : public DataSinkOperatorX<OlapTableSinkV2LocalState> {
public:
using Base = DataSinkOperatorX<OlapTableSinkV2LocalState>;
OlapTableSinkV2OperatorX(ObjectPool* pool, int operator_id, const RowDescriptor& row_desc,
const std::vector<TExpr>& t_output_expr)
: Base(operator_id, 0, 0),
OlapTableSinkV2OperatorX(ObjectPool* pool, int operator_id, int node_id,
const RowDescriptor& row_desc, const std::vector<TExpr>& t_output_expr)
: Base(operator_id, node_id, node_id),
_row_desc(row_desc),
_t_output_expr(t_output_expr),
_pool(pool) {};
Expand Down
6 changes: 3 additions & 3 deletions be/src/exec/operator/result_sink_operator.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -94,11 +94,11 @@ Status ResultSinkLocalState::open(RuntimeState* state) {
return Status::OK();
}

ResultSinkOperatorX::ResultSinkOperatorX(int operator_id, const RowDescriptor& row_desc,
ResultSinkOperatorX::ResultSinkOperatorX(int operator_id, int node_id,
const RowDescriptor& row_desc,
const std::vector<TExpr>& t_output_expr,
const TResultSink& sink)
: DataSinkOperatorX(operator_id, std::numeric_limits<int>::max(),
std::numeric_limits<int>::max()),
: DataSinkOperatorX(operator_id, node_id, node_id),
_sink_type(!sink.__isset.type || sink.type == TResultSinkType::MYSQL_PROTOCOL
? TResultSinkType::MYSQL_PROTOCOL
: sink.type),
Expand Down
2 changes: 1 addition & 1 deletion be/src/exec/operator/result_sink_operator.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -156,7 +156,7 @@ class ResultSinkLocalState final : public PipelineXSinkLocalState<BasicSharedSta

class ResultSinkOperatorX final : public DataSinkOperatorX<ResultSinkLocalState> {
public:
ResultSinkOperatorX(int operator_id, const RowDescriptor& row_desc,
ResultSinkOperatorX(int operator_id, int node_id, const RowDescriptor& row_desc,
const std::vector<TExpr>& select_exprs, const TResultSink& sink);
Status prepare(RuntimeState* state) override;

Expand Down
17 changes: 11 additions & 6 deletions be/src/exec/pipeline/pipeline_fragment_context.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1074,8 +1074,11 @@ Status PipelineFragmentContext::_create_data_sink(ObjectPool* pool, const TDataS
return Status::InternalError("Missing data buffer sink.");
}

_sink = std::make_shared<ResultSinkOperatorX>(next_sink_operator_id(), row_desc,
output_exprs, thrift_sink.result_sink);
auto& pipeline = _pipelines[cur_pipeline_id];
int child_node_id = pipeline->operators().back()->node_id();
_sink = std::make_shared<ResultSinkOperatorX>(next_sink_operator_id(), child_node_id + 1,
row_desc, output_exprs,
thrift_sink.result_sink);
break;
}
case TDataSinkType::DICTIONARY_SINK: {
Expand All @@ -1089,14 +1092,16 @@ Status PipelineFragmentContext::_create_data_sink(ObjectPool* pool, const TDataS
}
case TDataSinkType::GROUP_COMMIT_OLAP_TABLE_SINK:
case TDataSinkType::OLAP_TABLE_SINK: {
auto& pipeline = _pipelines[cur_pipeline_id];
int child_node_id = pipeline->operators().back()->node_id();
if (state->query_options().enable_memtable_on_sink_node &&
!_has_inverted_index_v1_or_partial_update(thrift_sink.olap_table_sink) &&
!config::is_cloud_mode()) {
_sink = std::make_shared<OlapTableSinkV2OperatorX>(pool, next_sink_operator_id(),
row_desc, output_exprs);
_sink = std::make_shared<OlapTableSinkV2OperatorX>(
pool, next_sink_operator_id(), child_node_id + 1, row_desc, output_exprs);
} else {
_sink = std::make_shared<OlapTableSinkOperatorX>(pool, next_sink_operator_id(),
row_desc, output_exprs);
_sink = std::make_shared<OlapTableSinkOperatorX>(
pool, next_sink_operator_id(), child_node_id + 1, row_desc, output_exprs);
}
break;
}
Expand Down
Loading