You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by kx...@apache.org on 2024/01/09 00:44:56 UTC
(doris) branch branch-2.0 updated: [feature](audit-log) add audit-log in insert into (#27641) (#28946)
This is an automated email from the ASF dual-hosted git repository.
kxiao pushed a commit to branch branch-2.0
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-2.0 by this push:
new 26a0d40a957 [feature](audit-log) add audit-log in insert into (#27641) (#28946)
26a0d40a957 is described below
commit 26a0d40a9577ca799f7076e4492b62f0e6ad0bbf
Author: Mryange <59...@users.noreply.github.com>
AuthorDate: Tue Jan 9 08:44:49 2024 +0800
[feature](audit-log) add audit-log in insert into (#27641) (#28946)
---
be/src/pipeline/pipeline_fragment_context.cpp | 3 ++-
be/src/pipeline/pipeline_fragment_context.h | 16 ++++++++++++++--
be/src/pipeline/pipeline_task.cpp | 3 ++-
be/src/runtime/fragment_mgr.cpp | 20 ++++++++++++++++----
be/src/runtime/fragment_mgr.h | 2 ++
be/src/runtime/plan_fragment_executor.cpp | 25 ++++++++++++++++++++++---
be/src/runtime/plan_fragment_executor.h | 11 ++++++++++-
be/src/runtime/query_context.h | 1 +
be/src/runtime/query_statistics.cpp | 9 +++++++++
be/src/runtime/query_statistics.h | 19 +++++++++++++++----
be/src/vec/exec/vexchange_node.cpp | 7 ++++---
gensrc/thrift/FrontendService.thrift | 20 ++++++++++++++++++++
12 files changed, 117 insertions(+), 19 deletions(-)
diff --git a/be/src/pipeline/pipeline_fragment_context.cpp b/be/src/pipeline/pipeline_fragment_context.cpp
index 015944ffeed..c8b2d985aaa 100644
--- a/be/src/pipeline/pipeline_fragment_context.cpp
+++ b/be/src/pipeline/pipeline_fragment_context.cpp
@@ -875,7 +875,8 @@ void PipelineFragmentContext::send_report(bool done) {
_fragment_instance_id, _backend_num, _runtime_state.get(),
std::bind(&PipelineFragmentContext::update_status, this, std::placeholders::_1),
std::bind(&PipelineFragmentContext::cancel, this, std::placeholders::_1,
- std::placeholders::_2)});
+ std::placeholders::_2),
+ _dml_query_statistics()});
}
} // namespace doris::pipeline
diff --git a/be/src/pipeline/pipeline_fragment_context.h b/be/src/pipeline/pipeline_fragment_context.h
index 1b44894b842..a80c77959d0 100644
--- a/be/src/pipeline/pipeline_fragment_context.h
+++ b/be/src/pipeline/pipeline_fragment_context.h
@@ -121,7 +121,11 @@ public:
return _task_group_entity;
}
-private:
+ void set_query_statistics(std::shared_ptr<QueryStatistics> query_statistics) {
+ _query_statistics = query_statistics;
+ }
+
+protected:
Status _create_sink(int sender_id, const TDataSink& t_data_sink, RuntimeState* state);
Status _build_pipelines(ExecNode*, PipelinePtr);
Status _build_pipeline_tasks(const doris::TPipelineFragmentParams& request);
@@ -201,7 +205,15 @@ private:
// This executor will not report status to FE on being cancelled.
bool _is_report_on_cancel;
- DescriptorTbl* _desc_tbl;
+ DescriptorTbl* _desc_tbl = nullptr;
+ static bool _has_inverted_index_or_partial_update(TOlapTableSink sink);
+ std::shared_ptr<QueryStatistics> _dml_query_statistics() {
+ if (_query_statistics && _query_statistics->collect_dml_statistics()) {
+ return _query_statistics;
+ }
+ return nullptr;
+ }
+ std::shared_ptr<QueryStatistics> _query_statistics = nullptr;
};
} // namespace pipeline
} // namespace doris
\ No newline at end of file
diff --git a/be/src/pipeline/pipeline_task.cpp b/be/src/pipeline/pipeline_task.cpp
index 0d3cb413bf7..fa443c01b23 100644
--- a/be/src/pipeline/pipeline_task.cpp
+++ b/be/src/pipeline/pipeline_task.cpp
@@ -58,10 +58,11 @@ PipelineTask::PipelineTask(PipelinePtr& pipeline, uint32_t index, RuntimeState*
_fragment_context(fragment_context),
_parent_profile(parent_profile) {
_pipeline_task_watcher.start();
- _query_statistics.reset(new QueryStatistics());
+ _query_statistics.reset(new QueryStatistics(state->query_options().query_type));
_sink->set_query_statistics(_query_statistics);
_collect_query_statistics_with_every_batch =
_pipeline->collect_query_statistics_with_every_batch();
+ fragment_context->set_query_statistics(_query_statistics);
}
void PipelineTask::_fresh_profile_counter() {
diff --git a/be/src/runtime/fragment_mgr.cpp b/be/src/runtime/fragment_mgr.cpp
index 526a04c2fae..ab915870c24 100644
--- a/be/src/runtime/fragment_mgr.cpp
+++ b/be/src/runtime/fragment_mgr.cpp
@@ -173,7 +173,8 @@ public:
private:
void coordinator_callback(const Status& status, RuntimeProfile* profile,
- RuntimeProfile* load_channel_profile, bool done);
+ RuntimeProfile* load_channel_profile, bool done,
+ std::shared_ptr<QueryStatistics> query_statistics);
// Id of this query
TUniqueId _query_id;
@@ -218,7 +219,8 @@ FragmentExecState::FragmentExecState(const TUniqueId& query_id,
_query_ctx(std::move(query_ctx)),
_executor(exec_env, std::bind<void>(std::mem_fn(&FragmentExecState::coordinator_callback),
this, std::placeholders::_1, std::placeholders::_2,
- std::placeholders::_3, std::placeholders::_4)),
+ std::placeholders::_3, std::placeholders::_4,
+ std::placeholders::_5)),
_set_rsc_info(false),
_timeout_second(-1),
_report_status_cb_impl(report_status_cb_impl) {
@@ -304,13 +306,15 @@ Status FragmentExecState::cancel(const PPlanFragmentCancelReason& reason, const
// Also, the reported status will always reflect the most recent execution status,
// including the final status when execution finishes.
void FragmentExecState::coordinator_callback(const Status& status, RuntimeProfile* profile,
- RuntimeProfile* load_channel_profile, bool done) {
+ RuntimeProfile* load_channel_profile, bool done,
+ std::shared_ptr<QueryStatistics> query_statistics) {
_report_status_cb_impl(
{status, profile, load_channel_profile, done, _coord_addr, _query_id, -1,
_fragment_instance_id, _backend_num, _executor.runtime_state(),
std::bind(&FragmentExecState::update_status, this, std::placeholders::_1),
std::bind(&PlanFragmentExecutor::cancel, &_executor, std::placeholders::_1,
- std::placeholders::_2)});
+ std::placeholders::_2),
+ query_statistics});
DCHECK(status.ok() || done); // if !status.ok() => done
}
@@ -401,6 +405,14 @@ void FragmentMgr::coordinator_callback(const ReportStatusRequest& req) {
params.__set_finished_scan_ranges(req.runtime_state->num_finished_range());
DCHECK(req.runtime_state != nullptr);
+
+ if (req.query_statistics) {
+ TQueryStatistics queryStatistics;
+ DCHECK(req.query_statistics->collect_dml_statistics());
+ req.query_statistics->to_thrift(&queryStatistics);
+ params.__set_query_statistics(queryStatistics);
+ }
+
if (req.runtime_state->query_type() == TQueryType::LOAD && !req.done && req.status.ok()) {
// this is a load plan, and load is not finished, just make a brief report
params.__set_loaded_rows(req.runtime_state->num_rows_load_total());
diff --git a/be/src/runtime/fragment_mgr.h b/be/src/runtime/fragment_mgr.h
index 5f167aa2dc0..1394e54f531 100644
--- a/be/src/runtime/fragment_mgr.h
+++ b/be/src/runtime/fragment_mgr.h
@@ -33,6 +33,7 @@
#include "common/status.h"
#include "gutil/ref_counted.h"
#include "http/rest_monitor_iface.h"
+#include "runtime/query_statistics.h"
#include "runtime_filter_mgr.h"
#include "util/countdown_latch.h"
#include "util/hash_util.hpp" // IWYU pragma: keep
@@ -78,6 +79,7 @@ struct ReportStatusRequest {
RuntimeState* runtime_state;
std::function<Status(Status)> update_fn;
std::function<void(const PPlanFragmentCancelReason&, const std::string&)> cancel_fn;
+ std::shared_ptr<QueryStatistics> query_statistics;
};
// This class used to manage all the fragment execute in this instance
diff --git a/be/src/runtime/plan_fragment_executor.cpp b/be/src/runtime/plan_fragment_executor.cpp
index 2d7f8c79520..2792189e8e2 100644
--- a/be/src/runtime/plan_fragment_executor.cpp
+++ b/be/src/runtime/plan_fragment_executor.cpp
@@ -239,7 +239,7 @@ Status PlanFragmentExecutor::prepare(const TExecPlanFragmentParams& request,
VLOG_NOTICE << "plan_root=\n" << _plan->debug_string();
_prepared = true;
- _query_statistics.reset(new QueryStatistics());
+ _query_statistics.reset(new QueryStatistics(request.query_options.query_type));
if (_sink != nullptr) {
_sink->set_query_statistics(_query_statistics);
}
@@ -372,7 +372,25 @@ Status PlanFragmentExecutor::get_vectorized_internal(::doris::vectorized::Block*
void PlanFragmentExecutor::_collect_query_statistics() {
_query_statistics->clear();
- Status status = _plan->collect_query_statistics(_query_statistics.get());
+ Status status;
+ /// TODO(yxc):
+ // The judgment of enable_local_exchange here is a bug, it should not need to be checked. I will fix this later.
+ bool _is_local = false;
+ if (_runtime_state->query_options().__isset.enable_local_exchange) {
+ _is_local = _runtime_state->query_options().enable_local_exchange;
+ }
+
+ if (_is_local) {
+ if (_runtime_state->num_per_fragment_instances() == 1) {
+ status = _plan->collect_query_statistics(_query_statistics.get());
+ } else {
+ status = _plan->collect_query_statistics(_query_statistics.get(),
+ _runtime_state->per_fragment_instance_idx());
+ }
+ } else {
+ status = _plan->collect_query_statistics(_query_statistics.get());
+ }
+
if (!status.ok()) {
LOG(INFO) << "collect query statistics failed, st=" << status;
return;
@@ -469,7 +487,8 @@ void PlanFragmentExecutor::send_report(bool done) {
// but fragments still need to be cancelled (e.g. limit reached), the coordinator will
// be waiting for a final report and profile.
_report_status_cb(status, _is_report_success ? profile() : nullptr,
- _is_report_success ? load_channel_profile() : nullptr, done || !status.ok());
+ _is_report_success ? load_channel_profile() : nullptr, done || !status.ok(),
+ _dml_query_statistics());
}
void PlanFragmentExecutor::stop_report_thread() {
diff --git a/be/src/runtime/plan_fragment_executor.h b/be/src/runtime/plan_fragment_executor.h
index f2a9f10d9a3..565c9047c4b 100644
--- a/be/src/runtime/plan_fragment_executor.h
+++ b/be/src/runtime/plan_fragment_executor.h
@@ -34,6 +34,7 @@
#include <vector>
#include "common/status.h"
+#include "runtime/query_statistics.h"
#include "runtime/runtime_state.h"
#include "util/runtime_profile.h"
@@ -80,7 +81,8 @@ public:
// functions like PrettyPrint() or to_thrift(), neither of which is const
// because they take locks.
using report_status_callback =
- std::function<void(const Status&, RuntimeProfile*, RuntimeProfile*, bool)>;
+ std::function<void(const Status&, RuntimeProfile*, RuntimeProfile*, bool,
+ std::shared_ptr<QueryStatistics>)>;
// report_status_cb, if !empty(), is used to report the accumulated profile
// information periodically during execution (open() or get_next()).
@@ -244,6 +246,13 @@ private:
void _collect_query_statistics();
+ std::shared_ptr<QueryStatistics> _dml_query_statistics() {
+ if (_query_statistics && _query_statistics->collect_dml_statistics()) {
+ return _query_statistics;
+ }
+ return nullptr;
+ }
+
void _collect_node_statistics();
};
diff --git a/be/src/runtime/query_context.h b/be/src/runtime/query_context.h
index c7651a687dd..c6b996bcae9 100644
--- a/be/src/runtime/query_context.h
+++ b/be/src/runtime/query_context.h
@@ -30,6 +30,7 @@
#include "runtime/datetime_value.h"
#include "runtime/exec_env.h"
#include "runtime/memory/mem_tracker_limiter.h"
+#include "runtime/query_statistics.h"
#include "runtime/runtime_filter_mgr.h"
#include "runtime/runtime_predicate.h"
#include "task_group/task_group.h"
diff --git a/be/src/runtime/query_statistics.cpp b/be/src/runtime/query_statistics.cpp
index dfe63f57626..7630362f5eb 100644
--- a/be/src/runtime/query_statistics.cpp
+++ b/be/src/runtime/query_statistics.cpp
@@ -62,6 +62,15 @@ void QueryStatistics::to_pb(PQueryStatistics* statistics) {
}
}
+void QueryStatistics::to_thrift(TQueryStatistics* statistics) const {
+ DCHECK(statistics != nullptr);
+ statistics->scan_bytes = scan_bytes;
+ statistics->scan_rows = scan_rows;
+ statistics->cpu_ms = cpu_ms;
+ statistics->returned_rows = returned_rows;
+ statistics->max_peak_memory_bytes = max_peak_memory_bytes;
+}
+
void QueryStatistics::from_pb(const PQueryStatistics& statistics) {
scan_rows = statistics.scan_rows();
scan_bytes = statistics.scan_bytes();
diff --git a/be/src/runtime/query_statistics.h b/be/src/runtime/query_statistics.h
index d32b7a60e2a..4923e727961 100644
--- a/be/src/runtime/query_statistics.h
+++ b/be/src/runtime/query_statistics.h
@@ -17,6 +17,8 @@
#pragma once
+#include <gen_cpp/FrontendService_types.h>
+#include <gen_cpp/PaloInternalService_types.h>
#include <stdint.h>
#include <map>
@@ -57,9 +59,14 @@ private:
// or plan's statistics and QueryStatisticsRecvr is responsible for collecting it.
class QueryStatistics {
public:
- QueryStatistics()
- : scan_rows(0), scan_bytes(0), cpu_ms(0), returned_rows(0), max_peak_memory_bytes(0) {}
- ~QueryStatistics();
+ QueryStatistics(TQueryType::type query_type = TQueryType::type::SELECT)
+ : scan_rows(0),
+ scan_bytes(0),
+ cpu_ms(0),
+ returned_rows(0),
+ max_peak_memory_bytes(0),
+ _query_type(query_type) {}
+ virtual ~QueryStatistics();
void merge(const QueryStatistics& other);
@@ -107,11 +114,14 @@ public:
}
void to_pb(PQueryStatistics* statistics);
-
+ void to_thrift(TQueryStatistics* statistics) const;
void from_pb(const PQueryStatistics& statistics);
bool collected() const { return _collected; }
void set_collected() { _collected = true; }
+ // LOAD does not need to collect information on the exchange node.
+ bool collect_dml_statistics() { return _query_type == TQueryType::LOAD; }
+
private:
friend class QueryStatisticsRecvr;
int64_t scan_rows;
@@ -127,6 +137,7 @@ private:
using NodeStatisticsMap = std::unordered_map<int64_t, NodeStatistics*>;
NodeStatisticsMap _nodes_statistics_map;
bool _collected = false;
+ const TQueryType::type _query_type;
};
using QueryStatisticsPtr = std::shared_ptr<QueryStatistics>;
// It is used for collecting sub plan query statistics in DataStreamRecvr.
diff --git a/be/src/vec/exec/vexchange_node.cpp b/be/src/vec/exec/vexchange_node.cpp
index 17a85767ec7..dbd78a6033a 100644
--- a/be/src/vec/exec/vexchange_node.cpp
+++ b/be/src/vec/exec/vexchange_node.cpp
@@ -144,13 +144,14 @@ void VExchangeNode::release_resource(RuntimeState* state) {
Status VExchangeNode::collect_query_statistics(QueryStatistics* statistics) {
RETURN_IF_ERROR(ExecNode::collect_query_statistics(statistics));
- statistics->merge(_sub_plan_query_statistics_recvr.get());
+ if (!statistics->collect_dml_statistics()) {
+ statistics->merge(_sub_plan_query_statistics_recvr.get());
+ }
return Status::OK();
}
Status VExchangeNode::collect_query_statistics(QueryStatistics* statistics, int sender_id) {
RETURN_IF_ERROR(ExecNode::collect_query_statistics(statistics));
- DCHECK(statistics != nullptr);
- if (_sub_plan_query_statistics_recvr) {
+ if (!statistics->collect_dml_statistics()) {
statistics->merge(_sub_plan_query_statistics_recvr.get(), sender_id);
}
return Status::OK();
diff --git a/gensrc/thrift/FrontendService.thrift b/gensrc/thrift/FrontendService.thrift
index 248163ad110..f6121874932 100644
--- a/gensrc/thrift/FrontendService.thrift
+++ b/gensrc/thrift/FrontendService.thrift
@@ -390,6 +390,22 @@ enum FrontendServiceVersion {
V1
}
+struct TDetailedReportParams {
+ 1: optional Types.TUniqueId fragment_instance_id
+ 2: optional RuntimeProfile.TRuntimeProfileTree profile
+ 3: optional RuntimeProfile.TRuntimeProfileTree loadChannelProfile
+}
+
+
+struct TQueryStatistics {
+ // A thrift structure identical to the PQueryStatistics structure.
+ 1: optional i64 scan_rows
+ 2: optional i64 scan_bytes
+ 3: optional i64 returned_rows
+ 4: optional i64 cpu_ms
+ 5: optional i64 max_peak_memory_bytes
+}
+
// The results of an INSERT query, sent to the coordinator as part of
// TReportExecStatusParams
struct TReportExecStatusParams {
@@ -447,6 +463,10 @@ struct TReportExecStatusParams {
21: optional RuntimeProfile.TRuntimeProfileTree loadChannelProfile
22: optional i32 finished_scan_ranges
+
+ 23: optional list<TDetailedReportParams> detailed_report
+
+ 24: optional TQueryStatistics query_statistics
}
struct TFeResult {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org