You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@impala.apache.org by "Rahul Shivu Mahadev (Code Review)" <ge...@cloudera.org> on 2018/07/26 07:38:38 UTC
[Impala-ASF-CR] IMPALA-3825: Distribute Runtime Filtering Aggregation This patch moves runtime filter aggregation from the coordinator to other exec nodes. * QueryState will now house the Aggregated filters. * FilterState has been decoupled from coordinator code * A glob
Rahul Shivu Mahadev has uploaded this change for review. ( http://gerrit.cloudera.org:8080/11055
Change subject: IMPALA-3825: Distribute Runtime Filtering Aggregation This patch moves runtime filter aggregation from the coordinator to other exec nodes. * QueryState will now house the Aggregated filters. * FilterState has been decoupled from coordinator code * A glob
......................................................................
IMPALA-3825: Distribute Runtime Filtering Aggregation
This patch moves runtime filter aggregation from the coordinator
to other exec nodes.
* QueryState will now house the Aggregated filters.
* FilterState has been decoupled from coordinator code
* A global flag has been added to switch back to legacy aggregation
Change-Id: I94e183a0353fc46f8d3eccae029d2d52c5cdc40c
---
M be/src/common/global-flags.cc
M be/src/runtime/CMakeLists.txt
M be/src/runtime/coordinator-backend-state.cc
M be/src/runtime/coordinator-backend-state.h
D be/src/runtime/coordinator-filter-state.h
M be/src/runtime/coordinator.cc
M be/src/runtime/coordinator.h
A be/src/runtime/filter-state.cc
A be/src/runtime/filter-state.h
M be/src/runtime/query-exec-mgr.cc
M be/src/runtime/query-state.cc
M be/src/runtime/query-state.h
M be/src/runtime/runtime-filter-bank.cc
M be/src/service/impala-server.cc
M common/thrift/ImpalaInternalService.thrift
15 files changed, 689 insertions(+), 265 deletions(-)
git pull ssh://gerrit.cloudera.org:29418/Impala-ASF refs/changes/55/11055/1
--
To view, visit http://gerrit.cloudera.org:8080/11055
To unsubscribe, visit http://gerrit.cloudera.org:8080/settings
Gerrit-Project: Impala-ASF
Gerrit-Branch: master
Gerrit-MessageType: newchange
Gerrit-Change-Id: I94e183a0353fc46f8d3eccae029d2d52c5cdc40c
Gerrit-Change-Number: 11055
Gerrit-PatchSet: 1
Gerrit-Owner: Rahul Shivu Mahadev <ra...@cloudera.com>
Gerrit-Reviewer: Michael Ho <kw...@cloudera.com>
Gerrit-Reviewer: Sailesh Mukil <sa...@cloudera.com>
[Impala-ASF-CR] IMPALA-3825: Distribute Runtime Filtering Aggregation
Posted by "Rahul Shivu Mahadev (Code Review)" <ge...@cloudera.org>.
Hello Michael Ho, Sailesh Mukil, Impala Public Jenkins,
I'd like you to reexamine a change. Please visit
http://gerrit.cloudera.org:8080/11055
to look at the new patch set (#4).
Change subject: IMPALA-3825: Distribute Runtime Filtering Aggregation
......................................................................
IMPALA-3825: Distribute Runtime Filtering Aggregation
This patch moves runtime filter aggregation from the coordinator
to other exec nodes.
* QueryState will now house the Aggregated filters.
* FilterState has been decoupled from coordinator code
* A global flag has been added to switch back to legacy aggregation
* The coordinator will now pick an aggregator for every filter and
send the information via ExecQueryFInstances RPC. The corresponding
QueryState will set up the filter routing table locally and send
filter updates to the filter consumers.
Failure Modes:
1. Aggregator node runs out of memory :
There is a way to detect this and is currently employed by the
coordinator, same behavior here i.e logs error.
2. Aggregator node crashes :
Coordinator will detect this and cancel the query via cancel rpc,
nothing to worry here.
3. Filter Producer/Producers crash before sending the filter to the
Aggregator : Coordinator will detect the crash and inform other nodes.
Change-Id: I94e183a0353fc46f8d3eccae029d2d52c5cdc40c
---
M be/src/common/global-flags.cc
M be/src/runtime/CMakeLists.txt
M be/src/runtime/coordinator-backend-state.cc
M be/src/runtime/coordinator-backend-state.h
M be/src/runtime/coordinator.cc
M be/src/runtime/coordinator.h
A be/src/runtime/filter-state.cc
R be/src/runtime/filter-state.h
M be/src/runtime/query-exec-mgr.cc
M be/src/runtime/query-state.cc
M be/src/runtime/query-state.h
M be/src/runtime/runtime-filter-bank.cc
M be/src/service/impala-server.cc
M common/thrift/ImpalaInternalService.thrift
14 files changed, 591 insertions(+), 151 deletions(-)
git pull ssh://gerrit.cloudera.org:29418/Impala-ASF refs/changes/55/11055/4
--
To view, visit http://gerrit.cloudera.org:8080/11055
To unsubscribe, visit http://gerrit.cloudera.org:8080/settings
Gerrit-Project: Impala-ASF
Gerrit-Branch: master
Gerrit-MessageType: newpatchset
Gerrit-Change-Id: I94e183a0353fc46f8d3eccae029d2d52c5cdc40c
Gerrit-Change-Number: 11055
Gerrit-PatchSet: 4
Gerrit-Owner: Rahul Shivu Mahadev <ra...@cloudera.com>
Gerrit-Reviewer: Impala Public Jenkins <im...@cloudera.com>
Gerrit-Reviewer: Michael Ho <kw...@cloudera.com>
Gerrit-Reviewer: Sailesh Mukil <sa...@cloudera.com>
[Impala-ASF-CR] IMPALA-3825: Distribute Runtime Filtering Aggregation
Posted by "Impala Public Jenkins (Code Review)" <ge...@cloudera.org>.
Impala Public Jenkins has posted comments on this change. ( http://gerrit.cloudera.org:8080/11055 )
Change subject: IMPALA-3825: Distribute Runtime Filtering Aggregation
......................................................................
Patch Set 1:
Build Failed
https://jenkins.impala.io/job/gerrit-code-review-checks/72/ : Initial code review checks failed. See linked job for details on the failure.
--
To view, visit http://gerrit.cloudera.org:8080/11055
To unsubscribe, visit http://gerrit.cloudera.org:8080/settings
Gerrit-Project: Impala-ASF
Gerrit-Branch: master
Gerrit-MessageType: comment
Gerrit-Change-Id: I94e183a0353fc46f8d3eccae029d2d52c5cdc40c
Gerrit-Change-Number: 11055
Gerrit-PatchSet: 1
Gerrit-Owner: Rahul Shivu Mahadev <ra...@cloudera.com>
Gerrit-Reviewer: Impala Public Jenkins <im...@cloudera.com>
Gerrit-Reviewer: Michael Ho <kw...@cloudera.com>
Gerrit-Reviewer: Sailesh Mukil <sa...@cloudera.com>
Gerrit-Comment-Date: Thu, 26 Jul 2018 08:14:53 +0000
Gerrit-HasComments: No
[Impala-ASF-CR] IMPALA-3825: Distribute Runtime Filtering Aggregation
Posted by "Rahul Shivu Mahadev (Code Review)" <ge...@cloudera.org>.
Rahul Shivu Mahadev has posted comments on this change. ( http://gerrit.cloudera.org:8080/11055 )
Change subject: IMPALA-3825: Distribute Runtime Filtering Aggregation
......................................................................
Patch Set 3:
(37 comments)
http://gerrit.cloudera.org:8080/#/c/11055/3/be/src/runtime/coordinator-backend-state.cc
File be/src/runtime/coordinator-backend-state.cc:
http://gerrit.cloudera.org:8080/#/c/11055/3/be/src/runtime/coordinator-backend-state.cc@95
PS3, Line 95: x
> naming
Done
http://gerrit.cloudera.org:8080/#/c/11055/3/be/src/runtime/coordinator-backend-state.cc@98
PS3, Line 98: LOG(INFO) << "DebuggingPublishFilter AggregatorAddress " << tfs.aggregator_address;
> remove
Done
http://gerrit.cloudera.org:8080/#/c/11055/3/be/src/runtime/coordinator-backend-state.cc@95
PS3, Line 95: for (auto const& x : filter_routing_table) {
: TFilterState tfs;
: x.second.ToThrift(&tfs);
: LOG(INFO) << "DebuggingPublishFilter AggregatorAddress " << tfs.aggregator_address;
: rpc_params->filter_routing_table.insert(
: std::pair<int32_t, TFilterState>(x.first, tfs));
: }
> Add a comment about what you're doing here.
Done
http://gerrit.cloudera.org:8080/#/c/11055/3/be/src/runtime/coordinator.cc
File be/src/runtime/coordinator.cc:
http://gerrit.cloudera.org:8080/#/c/11055/3/be/src/runtime/coordinator.cc@260
PS3, Line 260: x
> naming. Call it 'filter' or something similar
Done
http://gerrit.cloudera.org:8080/#/c/11055/3/be/src/runtime/coordinator.cc@262
PS3, Line 262: num_filters_
> Is this needed as a member variable? Doesn't seem to be used anywhere.
do you think I should make it static ? and have better distribution across queries ?
http://gerrit.cloudera.org:8080/#/c/11055/3/be/src/runtime/coordinator.cc@690
PS3, Line 690: if (params.__isset.filter_updates_received) {
> Add a comment above this line:
Done
http://gerrit.cloudera.org:8080/#/c/11055/3/be/src/runtime/coordinator.cc@691
PS3, Line 691: if (backend_state->GetNumReceivedFilters() < params.filter_updates_received) {
> Add another comment above this line:
Done
http://gerrit.cloudera.org:8080/#/c/11055/3/be/src/runtime/coordinator.cc@884
PS3, Line 884: LOG(INFO)
: << "DebuggingPublishFilter Coordinator sending filter to fragment with id "
: << fragment_idx;
> remove
Done
http://gerrit.cloudera.org:8080/#/c/11055/3/be/src/runtime/filter-state.h
File be/src/runtime/filter-state.h:
http://gerrit.cloudera.org:8080/#/c/11055/3/be/src/runtime/filter-state.h@141
PS3, Line 141: FilterTarget
> const ref
Done
http://gerrit.cloudera.org:8080/#/c/11055/3/be/src/runtime/filter-state.h@150
PS3, Line 150: /// Need to cast the int type of this class to int32_t of thrift
> Still needed?
Done
http://gerrit.cloudera.org:8080/#/c/11055/3/be/src/runtime/filter-state.h@151
PS3, Line 151: set
> Why not unordered?
Done
http://gerrit.cloudera.org:8080/#/c/11055/3/be/src/runtime/filter-state.h@152
PS3, Line 152: int i
> int32_t here and remove the cast in the next line. Also, rename to 'idx'.
Done
http://gerrit.cloudera.org:8080/#/c/11055/3/be/src/runtime/filter-state.h@167
PS3, Line 167: TFilterTarget
> const ref
Done
http://gerrit.cloudera.org:8080/#/c/11055/3/be/src/runtime/filter-state.h@171
PS3, Line 171: boost
> std
Done
http://gerrit.cloudera.org:8080/#/c/11055/3/be/src/runtime/filter-state.h@130
PS3, Line 130: bool disabled() const {
: if (is_bloom_filter()) {
: return bloom_filter_.always_true;
: } else {
: DCHECK(is_min_max_filter());
: return min_max_filter_.always_true;
: }
: }
:
: void ToThrift(TFilterState* f) const {
: std::vector<TFilterTarget> t_targets;
: for (FilterTarget filter_target : targets_) {
: TFilterTarget thrift_filter_target;
: filter_target.ToThrift(&thrift_filter_target);
: t_targets.push_back(thrift_filter_target);
: }
: f->__set_targets(t_targets);
: f->__set_desc(desc_);
: f->__set_src(src_);
: f->__set_pending_count(pending_count_);
: /// Need to cast the int type of this class to int32_t of thrift
: std::set<int32_t> src_fragment_instance_idxs;
: for (int i : src_fragment_instance_idxs_) {
: src_fragment_instance_idxs.insert((int32_t)i);
: }
: f->__set_src_fragment_instance_idxs(src_fragment_instance_idxs);
: f->__set_bloom_filter(bloom_filter_);
: f->__set_min_max_filter(min_max_filter_);
: f->__set_first_arrival_time(first_arrival_time_);
: f->__set_completion_time(completion_time_);
: f->__set_aggregator_address(aggregator_address_);
: }
:
: static FilterState FromThrift(const TFilterState& f) {
: FilterState fs(f.desc, f.src, f.aggregator_address, f.pending_count,
: f.first_arrival_time, f.completion_time, f.bloom_filter, f.min_max_filter);
: std::vector<FilterTarget> targets;
: for (TFilterTarget filter_target : f.targets) {
: targets.push_back(FilterTarget::FromThrift(filter_target));
: }
: fs.set_targets(targets);
: boost::unordered_set<int> src_fragment_instance_idxs;
: for (int idx : f.src_fragment_instance_idxs) {
: src_fragment_instance_idxs.insert(idx);
: }
: fs.set_src_fragment_instance_idxs(src_fragment_instance_idxs);
:
: return fs;
: }
> These can be moved to the .cc file.
Done
http://gerrit.cloudera.org:8080/#/c/11055/3/be/src/runtime/filter-state.h@191
PS3, Line 191: TPlanNodeId src_;
: std::vector<FilterTarget> targets_;
> Add a comment for these both. It wasn't added to the CoordinatorFilterState
These two fields were infact present in coordinator-filter-state.
https://github.com/apache/impala/blob/master/be/src/runtime/coordinator-filter-state.h#L106
http://gerrit.cloudera.org:8080/#/c/11055/3/be/src/runtime/filter-state.h@198
PS3, Line 198: boost::unordered_set
> We can use the std version for this
Done
http://gerrit.cloudera.org:8080/#/c/11055/3/be/src/runtime/query-exec-mgr.cc
File be/src/runtime/query-exec-mgr.cc:
http://gerrit.cloudera.org:8080/#/c/11055/3/be/src/runtime/query-exec-mgr.cc@58
PS3, Line 58: if (params.filter_routing_table.size() != 0) {
> Add a comment above this:
Done
http://gerrit.cloudera.org:8080/#/c/11055/3/be/src/runtime/query-state.h
File be/src/runtime/query-state.h:
http://gerrit.cloudera.org:8080/#/c/11055/3/be/src/runtime/query-state.h@181
PS3, Line 181: const std::map<int32_t, TFilterState>
> Add the parameter name as well.
Done
http://gerrit.cloudera.org:8080/#/c/11055/3/be/src/runtime/query-state.h@206
PS3, Line 206: std::map<int32_t, FilterState> filter_routing_table_;
:
: SpinLock filter_lock_;
:
: std::map<TNetworkAddress, std::set<int32_t>> backend_list_;
> Add comments for these.
Done
http://gerrit.cloudera.org:8080/#/c/11055/3/be/src/runtime/query-state.cc
File be/src/runtime/query-state.cc:
http://gerrit.cloudera.org:8080/#/c/11055/3/be/src/runtime/query-state.cc@457
PS3, Line 457: const std::map<int32_t, TFilterState>
> Take this as a const ref, so that we avoid copying the map on calling this
Done
http://gerrit.cloudera.org:8080/#/c/11055/3/be/src/runtime/query-state.cc@458
PS3, Line 458: std::pair<int, TFilterState> it
> Using 'auto' for iterators is reasonable.
Done
http://gerrit.cloudera.org:8080/#/c/11055/3/be/src/runtime/query-state.cc@459
PS3, Line 459: s
> Avoid using names that aren't self explanatory.
Done
http://gerrit.cloudera.org:8080/#/c/11055/3/be/src/runtime/query-state.cc@466
PS3, Line 466: std::map<int32_t, FilterState>::iterator
> auto
Done
http://gerrit.cloudera.org:8080/#/c/11055/3/be/src/runtime/query-state.cc@551
PS3, Line 551: t
> 'backend'
Done
http://gerrit.cloudera.org:8080/#/c/11055/3/be/src/runtime/runtime-filter-bank.cc
File be/src/runtime/runtime-filter-bank.cc:
http://gerrit.cloudera.org:8080/#/c/11055/3/be/src/runtime/runtime-filter-bank.cc@134
PS3, Line 134: Couldn't send filter to aggregator
> Couldn't open a connection to the aggregator node
Done
http://gerrit.cloudera.org:8080/#/c/11055/3/be/src/runtime/runtime-filter-bank.cc@137
PS3, Line 137: for (int i = 0; i < MAX_FSEND_RETRY; ++i) {
> (Just FYI for now) If we decide to make this run asynchronously as part of
Ok
http://gerrit.cloudera.org:8080/#/c/11055/3/be/src/runtime/runtime-filter-bank.cc@138
PS3, Line 138: LOG(INFO) << "DebuggingPublishFilter trying to contact aggregator\n";
> This isn't needed. It might make the logs too noisy.
Done
http://gerrit.cloudera.org:8080/#/c/11055/3/be/src/runtime/runtime-filter-bank.cc@141
PS3, Line 141: LOG(INFO) << res.status.status_code;
> Not needed.
Done
http://gerrit.cloudera.org:8080/#/c/11055/3/be/src/runtime/runtime-filter-bank.cc@142
PS3, Line 142: if (res.status.status_code == TErrorCode::OK) {
: break;
: }
> You need to check 'status' too. That is the status of the RPC from the send
Done
http://gerrit.cloudera.org:8080/#/c/11055/3/be/src/runtime/runtime-filter-bank.cc@147
PS3, Line 147: }
> Add a newline before this
Done
http://gerrit.cloudera.org:8080/#/c/11055/3/be/src/runtime/runtime-filter-bank.cc@196
PS3, Line 196: LOG(INFO) << "DebuggingPublishFilter Produced filter with filter id" << filter_id;
> Not needed. The logs will get too noisy.
Done
http://gerrit.cloudera.org:8080/#/c/11055/3/be/src/runtime/runtime-filter-bank.cc@203
PS3, Line 203: TODO
> super-nit: "TODO:"
Done
http://gerrit.cloudera.org:8080/#/c/11055/3/be/src/service/impala-server.cc
File be/src/service/impala-server.cc:
http://gerrit.cloudera.org:8080/#/c/11055/3/be/src/service/impala-server.cc@2221
PS3, Line 2221: s
> naming
Done
http://gerrit.cloudera.org:8080/#/c/11055/3/be/src/service/impala-server.cc@2221
PS3, Line 2221: QueryState* s =
: ExecEnv::GetInstance()->query_exec_mgr()->GetQueryState(params.query_id);
> Move inside the 'if' condition.
Done
http://gerrit.cloudera.org:8080/#/c/11055/3/be/src/service/impala-server.cc@2230
PS3, Line 2230: result.status.__set_status_code(TErrorCode::OK);
: result.__isset.status = true;
> Better to set this after the call to UpdateFilter
Done
http://gerrit.cloudera.org:8080/#/c/11055/3/common/thrift/ImpalaInternalService.thrift
File common/thrift/ImpalaInternalService.thrift:
http://gerrit.cloudera.org:8080/#/c/11055/3/common/thrift/ImpalaInternalService.thrift@822
PS3, Line 822:
> nit:whitespace
Done
--
To view, visit http://gerrit.cloudera.org:8080/11055
To unsubscribe, visit http://gerrit.cloudera.org:8080/settings
Gerrit-Project: Impala-ASF
Gerrit-Branch: master
Gerrit-MessageType: comment
Gerrit-Change-Id: I94e183a0353fc46f8d3eccae029d2d52c5cdc40c
Gerrit-Change-Number: 11055
Gerrit-PatchSet: 3
Gerrit-Owner: Rahul Shivu Mahadev <ra...@cloudera.com>
Gerrit-Reviewer: Impala Public Jenkins <im...@cloudera.com>
Gerrit-Reviewer: Michael Ho <kw...@cloudera.com>
Gerrit-Reviewer: Rahul Shivu Mahadev <ra...@cloudera.com>
Gerrit-Reviewer: Sailesh Mukil <sa...@cloudera.com>
Gerrit-Comment-Date: Tue, 31 Jul 2018 20:42:59 +0000
Gerrit-HasComments: Yes
[Impala-ASF-CR] IMPALA-3825: Distribute Runtime Filtering Aggregation
Posted by "Tim Armstrong (Code Review)" <ge...@cloudera.org>.
Tim Armstrong has abandoned this change. ( http://gerrit.cloudera.org:8080/11055 )
Change subject: IMPALA-3825: Distribute Runtime Filtering Aggregation
......................................................................
Abandoned
Cleaning up some old CRs
--
To view, visit http://gerrit.cloudera.org:8080/11055
To unsubscribe, visit http://gerrit.cloudera.org:8080/settings
Gerrit-Project: Impala-ASF
Gerrit-Branch: master
Gerrit-MessageType: abandon
Gerrit-Change-Id: I94e183a0353fc46f8d3eccae029d2d52c5cdc40c
Gerrit-Change-Number: 11055
Gerrit-PatchSet: 4
Gerrit-Owner: Rahul Shivu Mahadev <ra...@cloudera.com>
Gerrit-Reviewer: Impala Public Jenkins <im...@cloudera.com>
Gerrit-Reviewer: Michael Ho <mi...@gmail.com>
Gerrit-Reviewer: Rahul Shivu Mahadev <ra...@cloudera.com>
Gerrit-Reviewer: Sailesh Mukil <sa...@apache.org>
[Impala-ASF-CR] IMPALA-3825: Distribute Runtime Filtering Aggregation
Posted by "Sailesh Mukil (Code Review)" <ge...@cloudera.org>.
Sailesh Mukil has posted comments on this change. ( http://gerrit.cloudera.org:8080/11055 )
Change subject: IMPALA-3825: Distribute Runtime Filtering Aggregation
......................................................................
Patch Set 3:
(40 comments)
http://gerrit.cloudera.org:8080/#/c/11055/3//COMMIT_MSG
Commit Message:
http://gerrit.cloudera.org:8080/#/c/11055/3//COMMIT_MSG@14
PS3, Line 14:
Add comments about the following, so that it's easier for reviewers:
- Explain in a para or two how your patch achieves the distribution; i.e. explain your approach in plain english.
- What kind of new failure modes can happen because of this change, as opposed to before? (Talk about the Exec() RPC race with the UpdateFilter() RPC)
- How long do we expect the aggregators to be accessible? i.e. what is the lifetime of an aggregator tied to?
- How are you updating the runtime profile?
http://gerrit.cloudera.org:8080/#/c/11055/3/be/src/runtime/coordinator-backend-state.cc
File be/src/runtime/coordinator-backend-state.cc:
http://gerrit.cloudera.org:8080/#/c/11055/3/be/src/runtime/coordinator-backend-state.cc@95
PS3, Line 95: x
naming
http://gerrit.cloudera.org:8080/#/c/11055/3/be/src/runtime/coordinator-backend-state.cc@98
PS3, Line 98: LOG(INFO) << "DebuggingPublishFilter AggregatorAddress " << tfs.aggregator_address;
remove
http://gerrit.cloudera.org:8080/#/c/11055/3/be/src/runtime/coordinator-backend-state.cc@95
PS3, Line 95: for (auto const& x : filter_routing_table) {
: TFilterState tfs;
: x.second.ToThrift(&tfs);
: LOG(INFO) << "DebuggingPublishFilter AggregatorAddress " << tfs.aggregator_address;
: rpc_params->filter_routing_table.insert(
: std::pair<int32_t, TFilterState>(x.first, tfs));
: }
Add a comment about what you're doing here.
http://gerrit.cloudera.org:8080/#/c/11055/3/be/src/runtime/coordinator.cc
File be/src/runtime/coordinator.cc:
http://gerrit.cloudera.org:8080/#/c/11055/3/be/src/runtime/coordinator.cc@260
PS3, Line 260: x
naming. Call it 'filter' or something similar
http://gerrit.cloudera.org:8080/#/c/11055/3/be/src/runtime/coordinator.cc@262
PS3, Line 262: num_filters_
Is this needed as a member variable? Doesn't seem to be used anywhere.
http://gerrit.cloudera.org:8080/#/c/11055/3/be/src/runtime/coordinator.cc@690
PS3, Line 690: if (params.__isset.filter_updates_received) {
Add a comment above this line:
"Update aggregator's filter metrics in the runtime profile"
http://gerrit.cloudera.org:8080/#/c/11055/3/be/src/runtime/coordinator.cc@691
PS3, Line 691: if (backend_state->GetNumReceivedFilters() < params.filter_updates_received) {
Add another comment above this line:
"Make sure not to double count filter updates from the same aggregator."
http://gerrit.cloudera.org:8080/#/c/11055/3/be/src/runtime/coordinator.cc@690
PS3, Line 690: if (params.__isset.filter_updates_received) {
: if (backend_state->GetNumReceivedFilters() < params.filter_updates_received) {
: filter_updates_received_->Add(
: params.filter_updates_received - backend_state->GetNumReceivedFilters());
: backend_state->SetNumReceivedFilters(params.filter_updates_received);
: }
: }
There's a race here. If 2 updates for the same Backend execute in parallel, you'll end up having an incorrect number of updated filters.
http://gerrit.cloudera.org:8080/#/c/11055/3/be/src/runtime/coordinator.cc@884
PS3, Line 884: LOG(INFO)
: << "DebuggingPublishFilter Coordinator sending filter to fragment with id "
: << fragment_idx;
remove
http://gerrit.cloudera.org:8080/#/c/11055/3/be/src/runtime/filter-state.h
File be/src/runtime/filter-state.h:
http://gerrit.cloudera.org:8080/#/c/11055/3/be/src/runtime/filter-state.h@141
PS3, Line 141: FilterTarget
const ref
http://gerrit.cloudera.org:8080/#/c/11055/3/be/src/runtime/filter-state.h@150
PS3, Line 150: /// Need to cast the int type of this class to int32_t of thrift
Still needed?
http://gerrit.cloudera.org:8080/#/c/11055/3/be/src/runtime/filter-state.h@151
PS3, Line 151: set
Why not unordered?
http://gerrit.cloudera.org:8080/#/c/11055/3/be/src/runtime/filter-state.h@152
PS3, Line 152: int i
int32_t here and remove the cast in the next line. Also, rename to 'idx'.
http://gerrit.cloudera.org:8080/#/c/11055/3/be/src/runtime/filter-state.h@167
PS3, Line 167: TFilterTarget
const ref
http://gerrit.cloudera.org:8080/#/c/11055/3/be/src/runtime/filter-state.h@171
PS3, Line 171: boost
std
http://gerrit.cloudera.org:8080/#/c/11055/3/be/src/runtime/filter-state.h@130
PS3, Line 130: bool disabled() const {
: if (is_bloom_filter()) {
: return bloom_filter_.always_true;
: } else {
: DCHECK(is_min_max_filter());
: return min_max_filter_.always_true;
: }
: }
:
: void ToThrift(TFilterState* f) const {
: std::vector<TFilterTarget> t_targets;
: for (FilterTarget filter_target : targets_) {
: TFilterTarget thrift_filter_target;
: filter_target.ToThrift(&thrift_filter_target);
: t_targets.push_back(thrift_filter_target);
: }
: f->__set_targets(t_targets);
: f->__set_desc(desc_);
: f->__set_src(src_);
: f->__set_pending_count(pending_count_);
: /// Need to cast the int type of this class to int32_t of thrift
: std::set<int32_t> src_fragment_instance_idxs;
: for (int i : src_fragment_instance_idxs_) {
: src_fragment_instance_idxs.insert((int32_t)i);
: }
: f->__set_src_fragment_instance_idxs(src_fragment_instance_idxs);
: f->__set_bloom_filter(bloom_filter_);
: f->__set_min_max_filter(min_max_filter_);
: f->__set_first_arrival_time(first_arrival_time_);
: f->__set_completion_time(completion_time_);
: f->__set_aggregator_address(aggregator_address_);
: }
:
: static FilterState FromThrift(const TFilterState& f) {
: FilterState fs(f.desc, f.src, f.aggregator_address, f.pending_count,
: f.first_arrival_time, f.completion_time, f.bloom_filter, f.min_max_filter);
: std::vector<FilterTarget> targets;
: for (TFilterTarget filter_target : f.targets) {
: targets.push_back(FilterTarget::FromThrift(filter_target));
: }
: fs.set_targets(targets);
: boost::unordered_set<int> src_fragment_instance_idxs;
: for (int idx : f.src_fragment_instance_idxs) {
: src_fragment_instance_idxs.insert(idx);
: }
: fs.set_src_fragment_instance_idxs(src_fragment_instance_idxs);
:
: return fs;
: }
These can be moved to the .cc file.
http://gerrit.cloudera.org:8080/#/c/11055/3/be/src/runtime/filter-state.h@191
PS3, Line 191: TPlanNodeId src_;
: std::vector<FilterTarget> targets_;
Add a comment for these both. It wasn't added to the CoordinatorFilterState, but it should have.
http://gerrit.cloudera.org:8080/#/c/11055/3/be/src/runtime/filter-state.h@198
PS3, Line 198: boost::unordered_set
We can use the std version for this
http://gerrit.cloudera.org:8080/#/c/11055/3/be/src/runtime/query-exec-mgr.cc
File be/src/runtime/query-exec-mgr.cc:
http://gerrit.cloudera.org:8080/#/c/11055/3/be/src/runtime/query-exec-mgr.cc@58
PS3, Line 58: if (params.filter_routing_table.size() != 0) {
Add a comment above this:
"Initialize the filter routing table given by the coordinator."
http://gerrit.cloudera.org:8080/#/c/11055/3/be/src/runtime/query-state.h
File be/src/runtime/query-state.h:
http://gerrit.cloudera.org:8080/#/c/11055/3/be/src/runtime/query-state.h@181
PS3, Line 181: const std::map<int32_t, TFilterState>
Add the parameter name as well.
http://gerrit.cloudera.org:8080/#/c/11055/3/be/src/runtime/query-state.h@206
PS3, Line 206: std::map<int32_t, FilterState> filter_routing_table_;
:
: SpinLock filter_lock_;
:
: std::map<TNetworkAddress, std::set<int32_t>> backend_list_;
Add comments for these.
http://gerrit.cloudera.org:8080/#/c/11055/3/be/src/runtime/query-state.cc
File be/src/runtime/query-state.cc:
http://gerrit.cloudera.org:8080/#/c/11055/3/be/src/runtime/query-state.cc@457
PS3, Line 457: const std::map<int32_t, TFilterState>
Take this as a const ref, so that we avoid copying the map on calling this function.
http://gerrit.cloudera.org:8080/#/c/11055/3/be/src/runtime/query-state.cc@458
PS3, Line 458: std::pair<int, TFilterState> it
Using 'auto' for iterators is reasonable.
https://google.github.io/styleguide/cppguide.html#auto
http://gerrit.cloudera.org:8080/#/c/11055/3/be/src/runtime/query-state.cc@459
PS3, Line 459: s
Avoid using names that aren't self explanatory.
Eg: This could be called 'filter'
http://gerrit.cloudera.org:8080/#/c/11055/3/be/src/runtime/query-state.cc@466
PS3, Line 466: std::map<int32_t, FilterState>::iterator
auto
http://gerrit.cloudera.org:8080/#/c/11055/3/be/src/runtime/query-state.cc@551
PS3, Line 551: t
'backend'
http://gerrit.cloudera.org:8080/#/c/11055/3/be/src/runtime/runtime-filter-bank.cc
File be/src/runtime/runtime-filter-bank.cc:
http://gerrit.cloudera.org:8080/#/c/11055/3/be/src/runtime/runtime-filter-bank.cc@134
PS3, Line 134: Couldn't send filter to aggregator
Couldn't open a connection to the aggregator node
http://gerrit.cloudera.org:8080/#/c/11055/3/be/src/runtime/runtime-filter-bank.cc@137
PS3, Line 137: for (int i = 0; i < MAX_FSEND_RETRY; ++i) {
(Just FYI for now) If we decide to make this run asynchronously as part of the rpc_pool() ThreadPool (see comment on L205), then we can add some sleeps between the retries.
http://gerrit.cloudera.org:8080/#/c/11055/3/be/src/runtime/runtime-filter-bank.cc@138
PS3, Line 138: LOG(INFO) << "DebuggingPublishFilter trying to contact aggregator\n";
This isn't needed. It might make the logs too noisy.
http://gerrit.cloudera.org:8080/#/c/11055/3/be/src/runtime/runtime-filter-bank.cc@141
PS3, Line 141: LOG(INFO) << res.status.status_code;
Not needed.
http://gerrit.cloudera.org:8080/#/c/11055/3/be/src/runtime/runtime-filter-bank.cc@142
PS3, Line 142: if (res.status.status_code == TErrorCode::OK) {
: break;
: }
You need to check 'status' too. That is the status of the RPC from the senders point of view. If it's not OK, that means the RPC failed to send for some reason (host not reachable, connection timed out, etc.)
'res.status' will contain the status set by the remote node as a response to this RPC. (Eg: "Aggregator not found, etc.)
So you need to check both here.
We don't do this in SendFilterToCoordinator() since we don't care if the RPC fails in that case.
http://gerrit.cloudera.org:8080/#/c/11055/3/be/src/runtime/runtime-filter-bank.cc@147
PS3, Line 147: }
Add a newline before this
http://gerrit.cloudera.org:8080/#/c/11055/3/be/src/runtime/runtime-filter-bank.cc@196
PS3, Line 196: LOG(INFO) << "DebuggingPublishFilter Produced filter with filter id" << filter_id;
Not needed. The logs will get too noisy.
http://gerrit.cloudera.org:8080/#/c/11055/3/be/src/runtime/runtime-filter-bank.cc@203
PS3, Line 203: TODO
super-nit: "TODO:"
http://gerrit.cloudera.org:8080/#/c/11055/3/be/src/runtime/runtime-filter-bank.cc@204
PS3, Line 204: SendFilterToAggregator(
: aggregator_address, params, ExecEnv::GetInstance()->impalad_client_cache());
My take is that we can also run this as part of the rpc_pool() threadpool, so that it doesn't slow this fragment instance down. Let's make a decision once Michael and/or Tim have had a chance to chime in.
http://gerrit.cloudera.org:8080/#/c/11055/3/be/src/service/impala-server.cc
File be/src/service/impala-server.cc:
http://gerrit.cloudera.org:8080/#/c/11055/3/be/src/service/impala-server.cc@2221
PS3, Line 2221: s
naming
http://gerrit.cloudera.org:8080/#/c/11055/3/be/src/service/impala-server.cc@2221
PS3, Line 2221: QueryState* s =
: ExecEnv::GetInstance()->query_exec_mgr()->GetQueryState(params.query_id);
Move inside the 'if' condition.
http://gerrit.cloudera.org:8080/#/c/11055/3/be/src/service/impala-server.cc@2230
PS3, Line 2230: result.status.__set_status_code(TErrorCode::OK);
: result.__isset.status = true;
Better to set this after the call to UpdateFilter
http://gerrit.cloudera.org:8080/#/c/11055/3/common/thrift/ImpalaInternalService.thrift
File common/thrift/ImpalaInternalService.thrift:
http://gerrit.cloudera.org:8080/#/c/11055/3/common/thrift/ImpalaInternalService.thrift@822
PS3, Line 822:
nit:whitespace
--
To view, visit http://gerrit.cloudera.org:8080/11055
To unsubscribe, visit http://gerrit.cloudera.org:8080/settings
Gerrit-Project: Impala-ASF
Gerrit-Branch: master
Gerrit-MessageType: comment
Gerrit-Change-Id: I94e183a0353fc46f8d3eccae029d2d52c5cdc40c
Gerrit-Change-Number: 11055
Gerrit-PatchSet: 3
Gerrit-Owner: Rahul Shivu Mahadev <ra...@cloudera.com>
Gerrit-Reviewer: Impala Public Jenkins <im...@cloudera.com>
Gerrit-Reviewer: Michael Ho <kw...@cloudera.com>
Gerrit-Reviewer: Sailesh Mukil <sa...@cloudera.com>
Gerrit-Comment-Date: Sun, 29 Jul 2018 22:31:17 +0000
Gerrit-HasComments: Yes
[Impala-ASF-CR] IMPALA-3825: Distribute Runtime Filtering Aggregation
Posted by "Impala Public Jenkins (Code Review)" <ge...@cloudera.org>.
Impala Public Jenkins has posted comments on this change. ( http://gerrit.cloudera.org:8080/11055 )
Change subject: IMPALA-3825: Distribute Runtime Filtering Aggregation
......................................................................
Patch Set 3:
Build Started https://jenkins.impala.io/job/gerrit-code-review-checks/81/
Running initial code review checks. This is experimental - please report any issues to tarmstrong@cloudera.com or on this JIRA: IMPALA-7317
--
To view, visit http://gerrit.cloudera.org:8080/11055
To unsubscribe, visit http://gerrit.cloudera.org:8080/settings
Gerrit-Project: Impala-ASF
Gerrit-Branch: master
Gerrit-MessageType: comment
Gerrit-Change-Id: I94e183a0353fc46f8d3eccae029d2d52c5cdc40c
Gerrit-Change-Number: 11055
Gerrit-PatchSet: 3
Gerrit-Owner: Rahul Shivu Mahadev <ra...@cloudera.com>
Gerrit-Reviewer: Impala Public Jenkins <im...@cloudera.com>
Gerrit-Reviewer: Michael Ho <kw...@cloudera.com>
Gerrit-Reviewer: Sailesh Mukil <sa...@cloudera.com>
Gerrit-Comment-Date: Fri, 27 Jul 2018 00:00:20 +0000
Gerrit-HasComments: No
[Impala-ASF-CR] IMPALA-3825: Distribute Runtime Filtering Aggregation
Posted by "Michael Ho (Code Review)" <ge...@cloudera.org>.
Michael Ho has posted comments on this change. ( http://gerrit.cloudera.org:8080/11055 )
Change subject: IMPALA-3825: Distribute Runtime Filtering Aggregation
......................................................................
Patch Set 4:
(1 comment)
May help to add a unit test to exercise retrying logic when sending a filter to the aggregator.
http://gerrit.cloudera.org:8080/#/c/11055/4/common/thrift/ImpalaInternalService.thrift
File common/thrift/ImpalaInternalService.thrift:
http://gerrit.cloudera.org:8080/#/c/11055/4/common/thrift/ImpalaInternalService.thrift@878
PS4, Line 878: optional map<Types.TNetworkAddress, set<i32>> backend_list
May help to add a comment on what this is for.
--
To view, visit http://gerrit.cloudera.org:8080/11055
To unsubscribe, visit http://gerrit.cloudera.org:8080/settings
Gerrit-Project: Impala-ASF
Gerrit-Branch: master
Gerrit-MessageType: comment
Gerrit-Change-Id: I94e183a0353fc46f8d3eccae029d2d52c5cdc40c
Gerrit-Change-Number: 11055
Gerrit-PatchSet: 4
Gerrit-Owner: Rahul Shivu Mahadev <ra...@cloudera.com>
Gerrit-Reviewer: Impala Public Jenkins <im...@cloudera.com>
Gerrit-Reviewer: Michael Ho <kw...@cloudera.com>
Gerrit-Reviewer: Rahul Shivu Mahadev <ra...@cloudera.com>
Gerrit-Reviewer: Sailesh Mukil <sa...@cloudera.com>
Gerrit-Comment-Date: Fri, 10 Aug 2018 21:50:22 +0000
Gerrit-HasComments: Yes
[Impala-ASF-CR] IMPALA-3825: Distribute Runtime Filtering Aggregation This patch moves runtime filter aggregation from the coordinator to other exec nodes. * QueryState will now house the Aggregated filters. * FilterState has been decoupled from coordinator code * A glob
Posted by "Impala Public Jenkins (Code Review)" <ge...@cloudera.org>.
Impala Public Jenkins has posted comments on this change. ( http://gerrit.cloudera.org:8080/11055 )
Change subject: IMPALA-3825: Distribute Runtime Filtering Aggregation This patch moves runtime filter aggregation from the coordinator to other exec nodes. * QueryState will now house the Aggregated filters. * FilterState has been decoupled from coordinator code * A glob
......................................................................
Patch Set 1:
Build Started https://jenkins.impala.io/job/gerrit-code-review-checks/72/
Running initial code review checks. This is experimental - please report any issues to tarmstrong@cloudera.com or on this JIRA: IMPALA-7317
--
To view, visit http://gerrit.cloudera.org:8080/11055
To unsubscribe, visit http://gerrit.cloudera.org:8080/settings
Gerrit-Project: Impala-ASF
Gerrit-Branch: master
Gerrit-MessageType: comment
Gerrit-Change-Id: I94e183a0353fc46f8d3eccae029d2d52c5cdc40c
Gerrit-Change-Number: 11055
Gerrit-PatchSet: 1
Gerrit-Owner: Rahul Shivu Mahadev <ra...@cloudera.com>
Gerrit-Reviewer: Impala Public Jenkins <im...@cloudera.com>
Gerrit-Reviewer: Michael Ho <kw...@cloudera.com>
Gerrit-Reviewer: Sailesh Mukil <sa...@cloudera.com>
Gerrit-Comment-Date: Thu, 26 Jul 2018 07:38:47 +0000
Gerrit-HasComments: No
[Impala-ASF-CR] IMPALA-3825: Distribute Runtime Filtering Aggregation
Posted by "Impala Public Jenkins (Code Review)" <ge...@cloudera.org>.
Impala Public Jenkins has posted comments on this change. ( http://gerrit.cloudera.org:8080/11055 )
Change subject: IMPALA-3825: Distribute Runtime Filtering Aggregation
......................................................................
Patch Set 3:
Build Successful
https://jenkins.impala.io/job/gerrit-code-review-checks/81/ : Initial code review checks passed. Use gerrit-verify-dryrun-external or gerrit-verify-dryrun to run full precommit tests.
--
To view, visit http://gerrit.cloudera.org:8080/11055
To unsubscribe, visit http://gerrit.cloudera.org:8080/settings
Gerrit-Project: Impala-ASF
Gerrit-Branch: master
Gerrit-MessageType: comment
Gerrit-Change-Id: I94e183a0353fc46f8d3eccae029d2d52c5cdc40c
Gerrit-Change-Number: 11055
Gerrit-PatchSet: 3
Gerrit-Owner: Rahul Shivu Mahadev <ra...@cloudera.com>
Gerrit-Reviewer: Impala Public Jenkins <im...@cloudera.com>
Gerrit-Reviewer: Michael Ho <kw...@cloudera.com>
Gerrit-Reviewer: Sailesh Mukil <sa...@cloudera.com>
Gerrit-Comment-Date: Fri, 27 Jul 2018 00:34:21 +0000
Gerrit-HasComments: No
[Impala-ASF-CR] IMPALA-3825: Distribute Runtime Filtering Aggregation
Posted by "Rahul Shivu Mahadev (Code Review)" <ge...@cloudera.org>.
Hello Michael Ho, Sailesh Mukil, Impala Public Jenkins,
I'd like you to reexamine a change. Please visit
http://gerrit.cloudera.org:8080/11055
to look at the new patch set (#3).
Change subject: IMPALA-3825: Distribute Runtime Filtering Aggregation
......................................................................
IMPALA-3825: Distribute Runtime Filtering Aggregation
This patch moves runtime filter aggregation from the coordinator
to other exec nodes.
* QueryState will now house the Aggregated filters.
* FilterState has been decoupled from coordinator code
* A global flag has been added to switch back to legacy aggregation
Change-Id: I94e183a0353fc46f8d3eccae029d2d52c5cdc40c
---
M be/src/common/global-flags.cc
M be/src/runtime/CMakeLists.txt
M be/src/runtime/coordinator-backend-state.cc
M be/src/runtime/coordinator-backend-state.h
D be/src/runtime/coordinator-filter-state.h
M be/src/runtime/coordinator.cc
M be/src/runtime/coordinator.h
A be/src/runtime/filter-state.cc
A be/src/runtime/filter-state.h
M be/src/runtime/query-exec-mgr.cc
M be/src/runtime/query-state.cc
M be/src/runtime/query-state.h
M be/src/runtime/runtime-filter-bank.cc
M be/src/service/impala-server.cc
M common/thrift/ImpalaInternalService.thrift
15 files changed, 690 insertions(+), 265 deletions(-)
git pull ssh://gerrit.cloudera.org:29418/Impala-ASF refs/changes/55/11055/3
--
To view, visit http://gerrit.cloudera.org:8080/11055
To unsubscribe, visit http://gerrit.cloudera.org:8080/settings
Gerrit-Project: Impala-ASF
Gerrit-Branch: master
Gerrit-MessageType: newpatchset
Gerrit-Change-Id: I94e183a0353fc46f8d3eccae029d2d52c5cdc40c
Gerrit-Change-Number: 11055
Gerrit-PatchSet: 3
Gerrit-Owner: Rahul Shivu Mahadev <ra...@cloudera.com>
Gerrit-Reviewer: Impala Public Jenkins <im...@cloudera.com>
Gerrit-Reviewer: Michael Ho <kw...@cloudera.com>
Gerrit-Reviewer: Sailesh Mukil <sa...@cloudera.com>
[Impala-ASF-CR] IMPALA-3825: Distribute Runtime Filtering Aggregation
Posted by "Impala Public Jenkins (Code Review)" <ge...@cloudera.org>.
Impala Public Jenkins has posted comments on this change. ( http://gerrit.cloudera.org:8080/11055 )
Change subject: IMPALA-3825: Distribute Runtime Filtering Aggregation
......................................................................
Patch Set 4:
Build Successful
https://jenkins.impala.io/job/gerrit-code-review-checks/119/ : Initial code review checks passed. Use gerrit-verify-dryrun-external or gerrit-verify-dryrun to run full precommit tests.
--
To view, visit http://gerrit.cloudera.org:8080/11055
To unsubscribe, visit http://gerrit.cloudera.org:8080/settings
Gerrit-Project: Impala-ASF
Gerrit-Branch: master
Gerrit-MessageType: comment
Gerrit-Change-Id: I94e183a0353fc46f8d3eccae029d2d52c5cdc40c
Gerrit-Change-Number: 11055
Gerrit-PatchSet: 4
Gerrit-Owner: Rahul Shivu Mahadev <ra...@cloudera.com>
Gerrit-Reviewer: Impala Public Jenkins <im...@cloudera.com>
Gerrit-Reviewer: Michael Ho <kw...@cloudera.com>
Gerrit-Reviewer: Rahul Shivu Mahadev <ra...@cloudera.com>
Gerrit-Reviewer: Sailesh Mukil <sa...@cloudera.com>
Gerrit-Comment-Date: Tue, 31 Jul 2018 21:18:11 +0000
Gerrit-HasComments: No
[Impala-ASF-CR] IMPALA-3825: Distribute Runtime Filtering Aggregation
Posted by "Rahul Shivu Mahadev (Code Review)" <ge...@cloudera.org>.
Hello Michael Ho, Sailesh Mukil, Impala Public Jenkins,
I'd like you to reexamine a change. Please visit
http://gerrit.cloudera.org:8080/11055
to look at the new patch set (#2).
Change subject: IMPALA-3825: Distribute Runtime Filtering Aggregation
......................................................................
IMPALA-3825: Distribute Runtime Filtering Aggregation
This patch moves runtime filter aggregation from the coordinator
to other exec nodes.
* QueryState will now house the Aggregated filters.
* FilterState has been decoupled from coordinator code
* A global flag has been added to switch back to legacy aggregation
Change-Id: I94e183a0353fc46f8d3eccae029d2d52c5cdc40c
---
M be/src/common/global-flags.cc
M be/src/runtime/CMakeLists.txt
M be/src/runtime/coordinator-backend-state.cc
M be/src/runtime/coordinator-backend-state.h
D be/src/runtime/coordinator-filter-state.h
M be/src/runtime/coordinator.cc
M be/src/runtime/coordinator.h
A be/src/runtime/filter-state.cc
A be/src/runtime/filter-state.h
M be/src/runtime/query-exec-mgr.cc
M be/src/runtime/query-state.cc
M be/src/runtime/query-state.h
M be/src/runtime/runtime-filter-bank.cc
M be/src/service/impala-server.cc
M common/thrift/ImpalaInternalService.thrift
15 files changed, 689 insertions(+), 265 deletions(-)
git pull ssh://gerrit.cloudera.org:29418/Impala-ASF refs/changes/55/11055/2
--
To view, visit http://gerrit.cloudera.org:8080/11055
To unsubscribe, visit http://gerrit.cloudera.org:8080/settings
Gerrit-Project: Impala-ASF
Gerrit-Branch: master
Gerrit-MessageType: newpatchset
Gerrit-Change-Id: I94e183a0353fc46f8d3eccae029d2d52c5cdc40c
Gerrit-Change-Number: 11055
Gerrit-PatchSet: 2
Gerrit-Owner: Rahul Shivu Mahadev <ra...@cloudera.com>
Gerrit-Reviewer: Impala Public Jenkins <im...@cloudera.com>
Gerrit-Reviewer: Michael Ho <kw...@cloudera.com>
Gerrit-Reviewer: Sailesh Mukil <sa...@cloudera.com>