You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@quickstep.apache.org by hb...@apache.org on 2016/08/19 15:35:47 UTC
[5/5] incubator-quickstep git commit: Parallel work order generation
support.
Parallel work order generation support.
Project: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/commit/31557e7e
Tree: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/tree/31557e7e
Diff: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/diff/31557e7e
Branch: refs/heads/partitioned-aggregation
Commit: 31557e7e29f54918891e7df1288b43e39ca95d38
Parents: 5b4b0ca
Author: Harshad Deshmukh <hb...@apache.org>
Authored: Fri Aug 19 10:35:09 2016 -0500
Committer: Harshad Deshmukh <hb...@apache.org>
Committed: Fri Aug 19 10:35:09 2016 -0500
----------------------------------------------------------------------
.../FinalizeAggregationOperator.cpp | 35 ++++++++++++++++----
.../FinalizeAggregationOperator.hpp | 12 +++++--
2 files changed, 37 insertions(+), 10 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/31557e7e/relational_operators/FinalizeAggregationOperator.cpp
----------------------------------------------------------------------
diff --git a/relational_operators/FinalizeAggregationOperator.cpp b/relational_operators/FinalizeAggregationOperator.cpp
index 20d0ee5..c08f319 100644
--- a/relational_operators/FinalizeAggregationOperator.cpp
+++ b/relational_operators/FinalizeAggregationOperator.cpp
@@ -39,12 +39,29 @@ bool FinalizeAggregationOperator::getAllWorkOrders(
if (blocking_dependencies_met_ && !started_) {
started_ = true;
- container->addNormalWorkOrder(
- new FinalizeAggregationWorkOrder(
- query_id_,
- query_context->releaseAggregationState(aggr_state_index_),
- query_context->getInsertDestination(output_destination_index_)),
- op_index_);
+ DCHECK(query_context->getAggregationState(aggr_state_index_) != nullptr);
+ if (query_context->getAggregationState(aggr_state_index_)->isAggregatePartitioned()) {
+ // The same AggregationState is shared across all the WorkOrders.
+ for (std::size_t part_id = 0;
+ part_id < query_context->getAggregationState(aggr_state_index_)
+ ->getNumPartitions();
+ ++part_id) {
+ container->addNormalWorkOrder(
+ new FinalizeAggregationWorkOrder(
+ query_id_,
+ query_context->getAggregationState(aggr_state_index_),
+ query_context->getInsertDestination(output_destination_index_),
+ static_cast<int>(part_id)),
+ op_index_);
+ }
+ } else {
+ container->addNormalWorkOrder(
+ new FinalizeAggregationWorkOrder(
+ query_id_,
+ query_context->getAggregationState(aggr_state_index_),
+ query_context->getInsertDestination(output_destination_index_)),
+ op_index_);
+ }
}
return started_;
}
@@ -68,7 +85,11 @@ bool FinalizeAggregationOperator::getAllWorkOrderProtos(WorkOrderProtosContainer
void FinalizeAggregationWorkOrder::execute() {
- state_->finalizeAggregate(output_destination_);
+ if (state_->isAggregatePartitioned()) {
+ state_->finalizeAggregatePartitioned(part_id_, output_destination_);
+ } else {
+ state_->finalizeAggregate(output_destination_);
+ }
}
} // namespace quickstep
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/31557e7e/relational_operators/FinalizeAggregationOperator.hpp
----------------------------------------------------------------------
diff --git a/relational_operators/FinalizeAggregationOperator.hpp b/relational_operators/FinalizeAggregationOperator.hpp
index 158a637..b7609fc 100644
--- a/relational_operators/FinalizeAggregationOperator.hpp
+++ b/relational_operators/FinalizeAggregationOperator.hpp
@@ -113,21 +113,27 @@ class FinalizeAggregationWorkOrder : public WorkOrder {
* @param state The AggregationState to use.
* @param output_destination The InsertDestination to insert aggregation
* results.
+ * @param part_id The partition ID for which the Finalize aggregation work
+ * order is issued. Ignore this field if the aggregation is not
+ * partitioned.
*/
FinalizeAggregationWorkOrder(const std::size_t query_id,
AggregationOperationState *state,
- InsertDestination *output_destination)
+ InsertDestination *output_destination,
+ int part_id = -1)
: WorkOrder(query_id),
state_(DCHECK_NOTNULL(state)),
- output_destination_(DCHECK_NOTNULL(output_destination)) {}
+ output_destination_(DCHECK_NOTNULL(output_destination)),
+ part_id_(part_id) {}
~FinalizeAggregationWorkOrder() override {}
void execute() override;
private:
- std::unique_ptr<AggregationOperationState> state_;
+ AggregationOperationState *state_;
InsertDestination *output_destination_;
+ const int part_id_;
DISALLOW_COPY_AND_ASSIGN(FinalizeAggregationWorkOrder);
};