You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@quickstep.apache.org by ji...@apache.org on 2017/01/28 03:01:46 UTC
[11/12] incubator-quickstep git commit: Added the partitioned hash
join in the distributed version.
Added the partitioned hash join in the distributed version.
Project: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/commit/968ce3f7
Tree: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/tree/968ce3f7
Diff: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/diff/968ce3f7
Branch: refs/heads/reorder-attrs
Commit: 968ce3f75969817b3dcc16c0c1c3218817cfe9b8
Parents: 7bce0b8
Author: Zuyu Zhang <zu...@apache.org>
Authored: Wed Jan 25 15:53:39 2017 -0800
Committer: Zuyu Zhang <zu...@apache.org>
Committed: Wed Jan 25 16:41:35 2017 -0800
----------------------------------------------------------------------
query_execution/CMakeLists.txt | 3 +
query_execution/ForemanDistributed.cpp | 7 +-
query_execution/PolicyEnforcerDistributed.cpp | 2 +
query_execution/PolicyEnforcerDistributed.hpp | 11 +-
query_execution/QueryExecutionTypedefs.hpp | 6 ++
query_execution/QueryManagerDistributed.cpp | 11 ++
query_execution/QueryManagerDistributed.hpp | 40 ++++---
.../tests/execution_generator/CMakeLists.txt | 6 ++
relational_operators/BuildHashOperator.cpp | 9 +-
relational_operators/BuildHashOperator.hpp | 18 ----
relational_operators/CMakeLists.txt | 1 +
relational_operators/DestroyHashOperator.cpp | 4 +-
relational_operators/DestroyHashOperator.hpp | 8 +-
relational_operators/HashJoinOperator.cpp | 31 +++---
relational_operators/HashJoinOperator.hpp | 105 ++++---------------
relational_operators/WorkOrder.proto | 10 +-
relational_operators/WorkOrderFactory.cpp | 10 --
17 files changed, 115 insertions(+), 167 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/968ce3f7/query_execution/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/query_execution/CMakeLists.txt b/query_execution/CMakeLists.txt
index 0f74384..c4c4079 100644
--- a/query_execution/CMakeLists.txt
+++ b/query_execution/CMakeLists.txt
@@ -156,6 +156,7 @@ if (ENABLE_DISTRIBUTED)
target_link_libraries(quickstep_queryexecution_PolicyEnforcerDistributed
glog
quickstep_catalog_CatalogRelation
+ quickstep_catalog_CatalogTypedefs
quickstep_catalog_Catalog_proto
quickstep_queryexecution_PolicyEnforcerBase
quickstep_queryexecution_QueryContext
@@ -252,7 +253,9 @@ target_link_libraries(quickstep_queryexecution_QueryManagerBase
quickstep_utility_Macros)
if (ENABLE_DISTRIBUTED)
target_link_libraries(quickstep_queryexecution_QueryManagerDistributed
+ quickstep_catalog_CatalogTypedefs
quickstep_queryexecution_QueryContext
+ quickstep_queryexecution_QueryContext_proto
quickstep_queryexecution_QueryExecutionMessages_proto
quickstep_queryexecution_QueryExecutionState
quickstep_queryexecution_QueryExecutionTypedefs
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/968ce3f7/query_execution/ForemanDistributed.cpp
----------------------------------------------------------------------
diff --git a/query_execution/ForemanDistributed.cpp b/query_execution/ForemanDistributed.cpp
index 0fa701d..fe4c483 100644
--- a/query_execution/ForemanDistributed.cpp
+++ b/query_execution/ForemanDistributed.cpp
@@ -278,23 +278,28 @@ bool ForemanDistributed::isHashJoinRelatedWorkOrder(const S::WorkOrderMessage &p
size_t *shiftboss_index_for_hash_join) {
const S::WorkOrder &work_order_proto = proto.work_order();
QueryContext::join_hash_table_id join_hash_table_index;
+ partition_id part_id;
switch (work_order_proto.work_order_type()) {
case S::BUILD_HASH:
join_hash_table_index = work_order_proto.GetExtension(S::BuildHashWorkOrder::join_hash_table_index);
+ part_id = work_order_proto.GetExtension(S::BuildHashWorkOrder::partition_id);
break;
case S::HASH_JOIN:
join_hash_table_index = work_order_proto.GetExtension(S::HashJoinWorkOrder::join_hash_table_index);
+ part_id = work_order_proto.GetExtension(S::HashJoinWorkOrder::partition_id);
break;
case S::DESTROY_HASH:
join_hash_table_index = work_order_proto.GetExtension(S::DestroyHashWorkOrder::join_hash_table_index);
+ part_id = work_order_proto.GetExtension(S::DestroyHashWorkOrder::partition_id);
break;
default:
return false;
}
static_cast<PolicyEnforcerDistributed*>(policy_enforcer_.get())->getShiftbossIndexForHashJoin(
- proto.query_id(), join_hash_table_index, next_shiftboss_index_to_schedule, shiftboss_index_for_hash_join);
+ proto.query_id(), join_hash_table_index, part_id, next_shiftboss_index_to_schedule,
+ shiftboss_index_for_hash_join);
return true;
}
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/968ce3f7/query_execution/PolicyEnforcerDistributed.cpp
----------------------------------------------------------------------
diff --git a/query_execution/PolicyEnforcerDistributed.cpp b/query_execution/PolicyEnforcerDistributed.cpp
index c5642bc..e9f57d3 100644
--- a/query_execution/PolicyEnforcerDistributed.cpp
+++ b/query_execution/PolicyEnforcerDistributed.cpp
@@ -173,11 +173,13 @@ void PolicyEnforcerDistributed::getShiftbossIndexForAggregation(
void PolicyEnforcerDistributed::getShiftbossIndexForHashJoin(
const std::size_t query_id,
const QueryContext::join_hash_table_id join_hash_table_index,
+ const partition_id part_id,
const std::size_t next_shiftboss_index_to_schedule,
std::size_t *shiftboss_index) {
DCHECK(admitted_queries_.find(query_id) != admitted_queries_.end());
QueryManagerDistributed *query_manager = static_cast<QueryManagerDistributed*>(admitted_queries_[query_id].get());
query_manager->getShiftbossIndexForHashJoin(join_hash_table_index,
+ part_id,
next_shiftboss_index_to_schedule,
shiftboss_index);
}
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/968ce3f7/query_execution/PolicyEnforcerDistributed.hpp
----------------------------------------------------------------------
diff --git a/query_execution/PolicyEnforcerDistributed.hpp b/query_execution/PolicyEnforcerDistributed.hpp
index 2c00a6b..248948a 100644
--- a/query_execution/PolicyEnforcerDistributed.hpp
+++ b/query_execution/PolicyEnforcerDistributed.hpp
@@ -21,6 +21,7 @@
#include <utility>
#include <vector>
+#include "catalog/CatalogTypedefs.hpp"
#include "query_execution/PolicyEnforcerBase.hpp"
#include "query_execution/QueryContext.hpp"
#include "query_execution/QueryExecutionMessages.pb.h"
@@ -124,19 +125,21 @@ class PolicyEnforcerDistributed final : public PolicyEnforcerBase {
/**
* @brief Get or set the index of Shiftboss for a HashJoin related WorkOrder.
- * If it is the first BuildHash on <join_hash_table_index>, <shiftboss_index>
- * will be set to <next_shiftboss_index_to_schedule>. Otherwise,
- * <shiftboss_index> will be set to the index of the Shiftboss that has
- * executed the first BuildHash.
+ * If it is the first BuildHash on <join_hash_table_index, part_id>,
+ * <shiftboss_index> will be set to <next_shiftboss_index_to_schedule>.
+ * Otherwise, <shiftboss_index> will be set to the index of the Shiftboss that
+ * has executed the first BuildHash.
*
* @param query_id The query id.
* @param join_hash_table_index The Hash Table for the Join.
+ * @param part_id The partition ID.
* @param next_shiftboss_index The index of Shiftboss to schedule a next WorkOrder.
* @param shiftboss_index The index of Shiftboss to schedule the WorkOrder.
**/
void getShiftbossIndexForHashJoin(
const std::size_t query_id,
const QueryContext::join_hash_table_id join_hash_table_index,
+ const partition_id part_id,
const std::size_t next_shiftboss_index_to_schedule,
std::size_t *shiftboss_index);
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/968ce3f7/query_execution/QueryExecutionTypedefs.hpp
----------------------------------------------------------------------
diff --git a/query_execution/QueryExecutionTypedefs.hpp b/query_execution/QueryExecutionTypedefs.hpp
index faf2132..9f78302 100644
--- a/query_execution/QueryExecutionTypedefs.hpp
+++ b/query_execution/QueryExecutionTypedefs.hpp
@@ -62,6 +62,12 @@ using ClientIDMap = ThreadIDBasedMap<client_id,
'a',
'p'>;
+#ifdef QUICKSTEP_DISTRIBUTED
+
+constexpr std::size_t kInvalidShiftbossIndex = static_cast<std::size_t>(-1);
+
+#endif // QUICKSTEP_DISTRIBUTED
+
// We sort the following message types in the order of a life cycle of a query.
enum QueryExecutionMessageType : message_type_id {
kAdmitRequestMessage = 0, // Requesting a query (or queries) to be admitted, from
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/968ce3f7/query_execution/QueryManagerDistributed.cpp
----------------------------------------------------------------------
diff --git a/query_execution/QueryManagerDistributed.cpp b/query_execution/QueryManagerDistributed.cpp
index 5c7e0d8..6ac96ab 100644
--- a/query_execution/QueryManagerDistributed.cpp
+++ b/query_execution/QueryManagerDistributed.cpp
@@ -23,8 +23,10 @@
#include <cstdlib>
#include <memory>
#include <utility>
+#include <vector>
#include "query_execution/QueryContext.hpp"
+#include "query_execution/QueryContext.pb.h"
#include "query_execution/QueryExecutionMessages.pb.h"
#include "query_execution/QueryExecutionTypedefs.hpp"
#include "query_execution/QueryExecutionUtil.hpp"
@@ -45,6 +47,7 @@ using std::malloc;
using std::move;
using std::size_t;
using std::unique_ptr;
+using std::vector;
namespace quickstep {
@@ -65,6 +68,14 @@ QueryManagerDistributed::QueryManagerDistributed(QueryHandle *query_handle,
processOperator(index, false);
}
}
+
+ const serialization::QueryContext &query_context_proto = query_handle->getQueryContextProto();
+ shiftboss_indexes_for_aggrs_.resize(query_context_proto.aggregation_states_size(), kInvalidShiftbossIndex);
+
+ for (int i = 0; i < query_context_proto.join_hash_tables_size(); ++i) {
+ shiftboss_indexes_for_hash_joins_.push_back(
+ vector<size_t>(query_context_proto.join_hash_tables(i).num_partitions(), kInvalidShiftbossIndex));
+ }
}
serialization::WorkOrderMessage* QueryManagerDistributed::getNextWorkOrderMessage(
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/968ce3f7/query_execution/QueryManagerDistributed.hpp
----------------------------------------------------------------------
diff --git a/query_execution/QueryManagerDistributed.hpp b/query_execution/QueryManagerDistributed.hpp
index 7a07fcb..631b15a 100644
--- a/query_execution/QueryManagerDistributed.hpp
+++ b/query_execution/QueryManagerDistributed.hpp
@@ -22,10 +22,12 @@
#include <cstddef>
#include <memory>
-#include <unordered_map>
+#include <vector>
+#include "catalog/CatalogTypedefs.hpp"
#include "query_execution/QueryContext.hpp"
#include "query_execution/QueryExecutionState.hpp"
+#include "query_execution/QueryExecutionTypedefs.hpp"
#include "query_execution/QueryManagerBase.hpp"
#include "query_execution/WorkOrderProtosContainer.hpp"
#include "utility/Macros.hpp"
@@ -105,13 +107,12 @@ class QueryManagerDistributed final : public QueryManagerBase {
void getShiftbossIndexForAggregation(const QueryContext::aggregation_state_id aggr_state_index,
const std::size_t next_shiftboss_index_to_schedule,
std::size_t *shiftboss_index) {
- const auto cit = shiftboss_indexes_for_aggrs_.find(aggr_state_index);
- if (cit != shiftboss_indexes_for_aggrs_.end()) {
- *shiftboss_index = cit->second;
- } else {
- shiftboss_indexes_for_aggrs_.emplace(aggr_state_index, next_shiftboss_index_to_schedule);
- *shiftboss_index = next_shiftboss_index_to_schedule;
+ DCHECK_LT(aggr_state_index, shiftboss_indexes_for_aggrs_.size());
+ if (shiftboss_indexes_for_aggrs_[aggr_state_index] == kInvalidShiftbossIndex) {
+ shiftboss_indexes_for_aggrs_[aggr_state_index] = next_shiftboss_index_to_schedule;
}
+
+ *shiftboss_index = shiftboss_indexes_for_aggrs_[aggr_state_index];
}
/**
@@ -119,19 +120,22 @@ class QueryManagerDistributed final : public QueryManagerBase {
* Shiftboss index is not found, set using <next_shiftboss_index_to_schedule>.
*
* @param join_hash_table_index The Hash Table for the Join.
+ * @param part_id The partition ID.
* @param next_shiftboss_index The index of Shiftboss to schedule a next WorkOrder.
* @param shiftboss_index The index of Shiftboss to schedule the WorkOrder.
**/
void getShiftbossIndexForHashJoin(const QueryContext::join_hash_table_id join_hash_table_index,
+ const partition_id part_id,
const std::size_t next_shiftboss_index_to_schedule,
std::size_t *shiftboss_index) {
- const auto cit = shiftboss_indexes_for_hash_joins_.find(join_hash_table_index);
- if (cit != shiftboss_indexes_for_hash_joins_.end()) {
- *shiftboss_index = cit->second;
- } else {
- shiftboss_indexes_for_hash_joins_.emplace(join_hash_table_index, next_shiftboss_index_to_schedule);
- *shiftboss_index = next_shiftboss_index_to_schedule;
+ DCHECK_LT(join_hash_table_index, shiftboss_indexes_for_hash_joins_.size());
+ DCHECK_LT(part_id, shiftboss_indexes_for_hash_joins_[join_hash_table_index].size());
+
+ if (shiftboss_indexes_for_hash_joins_[join_hash_table_index][part_id] == kInvalidShiftbossIndex) {
+ shiftboss_indexes_for_hash_joins_[join_hash_table_index][part_id] = next_shiftboss_index_to_schedule;
}
+
+ *shiftboss_index = shiftboss_indexes_for_hash_joins_[join_hash_table_index][part_id];
}
private:
@@ -156,11 +160,13 @@ class QueryManagerDistributed final : public QueryManagerBase {
std::unique_ptr<WorkOrderProtosContainer> normal_workorder_protos_container_;
- // A map from an aggregation id to its scheduled Shiftboss index.
- std::unordered_map<QueryContext::aggregation_state_id, std::size_t> shiftboss_indexes_for_aggrs_;
+ // From an aggregation id (QueryContext::aggregation_state_id) to its
+ // scheduled Shiftboss index.
+ std::vector<std::size_t> shiftboss_indexes_for_aggrs_;
- // A map from a join hash table to its scheduled Shiftboss index.
- std::unordered_map<QueryContext::join_hash_table_id, std::size_t> shiftboss_indexes_for_hash_joins_;
+ // Get the scheduled Shiftboss index given
+ // [QueryContext::join_hash_table_id][partition_id].
+ std::vector<std::vector<std::size_t>> shiftboss_indexes_for_hash_joins_;
DISALLOW_COPY_AND_ASSIGN(QueryManagerDistributed);
};
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/968ce3f7/query_optimizer/tests/execution_generator/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/query_optimizer/tests/execution_generator/CMakeLists.txt b/query_optimizer/tests/execution_generator/CMakeLists.txt
index 2705130..d38f4aa 100644
--- a/query_optimizer/tests/execution_generator/CMakeLists.txt
+++ b/query_optimizer/tests/execution_generator/CMakeLists.txt
@@ -51,6 +51,11 @@ if (ENABLE_DISTRIBUTED)
"${CMAKE_CURRENT_SOURCE_DIR}/Join.test"
"${CMAKE_CURRENT_BINARY_DIR}/DistributedJoin.test"
"${CMAKE_CURRENT_BINARY_DIR}/DistributedJoin/")
+ add_test(quickstep_queryoptimizer_tests_distributed_executiongenerator_partition
+ "../quickstep_queryoptimizer_tests_ExecutionGeneratorTest"
+ "${CMAKE_CURRENT_SOURCE_DIR}/Partition.test"
+ "${CMAKE_CURRENT_BINARY_DIR}/DistributedPartition.test"
+ "${CMAKE_CURRENT_BINARY_DIR}/DistributedPartition/")
add_test(quickstep_queryoptimizer_tests_distributed_executiongenerator_select
"../quickstep_queryoptimizer_tests_DistributedExecutionGeneratorTest"
"${CMAKE_CURRENT_SOURCE_DIR}/Select.test"
@@ -146,6 +151,7 @@ if (ENABLE_DISTRIBUTED)
file(MAKE_DIRECTORY ${CMAKE_CURRENT_BINARY_DIR}/DistributedIndex)
file(MAKE_DIRECTORY ${CMAKE_CURRENT_BINARY_DIR}/DistributedInsert)
file(MAKE_DIRECTORY ${CMAKE_CURRENT_BINARY_DIR}/DistributedJoin)
+ file(MAKE_DIRECTORY ${CMAKE_CURRENT_BINARY_DIR}/DistributedPartition)
file(MAKE_DIRECTORY ${CMAKE_CURRENT_BINARY_DIR}/DistributedSelect)
file(MAKE_DIRECTORY ${CMAKE_CURRENT_BINARY_DIR}/DistributedStringPatternMatching)
file(MAKE_DIRECTORY ${CMAKE_CURRENT_BINARY_DIR}/DistributedTableGenerator)
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/968ce3f7/relational_operators/BuildHashOperator.cpp
----------------------------------------------------------------------
diff --git a/relational_operators/BuildHashOperator.cpp b/relational_operators/BuildHashOperator.cpp
index 14ec204..8f40fbb 100644
--- a/relational_operators/BuildHashOperator.cpp
+++ b/relational_operators/BuildHashOperator.cpp
@@ -80,7 +80,7 @@ bool BuildHashOperator::getAllWorkOrders(
for (const block_id block : input_relation_block_ids_[part_id]) {
container->addNormalWorkOrder(
new BuildHashWorkOrder(query_id_, input_relation_, join_key_attributes_, any_join_key_attributes_nullable_,
- num_partitions_, part_id, block, hash_table, storage_manager,
+ part_id, block, hash_table, storage_manager,
CreateLIPFilterBuilderHelper(lip_deployment_index_, query_context)),
op_index_);
}
@@ -94,9 +94,9 @@ bool BuildHashOperator::getAllWorkOrders(
input_relation_block_ids_[part_id].size()) {
container->addNormalWorkOrder(
new BuildHashWorkOrder(query_id_, input_relation_, join_key_attributes_, any_join_key_attributes_nullable_,
- num_partitions_, part_id,
- input_relation_block_ids_[part_id][num_workorders_generated_[part_id]], hash_table,
- storage_manager, CreateLIPFilterBuilderHelper(lip_deployment_index_, query_context)),
+ part_id, input_relation_block_ids_[part_id][num_workorders_generated_[part_id]],
+ hash_table, storage_manager,
+ CreateLIPFilterBuilderHelper(lip_deployment_index_, query_context)),
op_index_);
++num_workorders_generated_[part_id];
}
@@ -142,7 +142,6 @@ serialization::WorkOrder* BuildHashOperator::createWorkOrderProto(const block_id
}
proto->SetExtension(serialization::BuildHashWorkOrder::any_join_key_attributes_nullable,
any_join_key_attributes_nullable_);
- proto->SetExtension(serialization::BuildHashWorkOrder::num_partitions, num_partitions_);
proto->SetExtension(serialization::BuildHashWorkOrder::join_hash_table_index, hash_table_index_);
proto->SetExtension(serialization::BuildHashWorkOrder::partition_id, part_id);
proto->SetExtension(serialization::BuildHashWorkOrder::block_id, block);
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/968ce3f7/relational_operators/BuildHashOperator.hpp
----------------------------------------------------------------------
diff --git a/relational_operators/BuildHashOperator.hpp b/relational_operators/BuildHashOperator.hpp
index c9f29cb..273e3b1 100644
--- a/relational_operators/BuildHashOperator.hpp
+++ b/relational_operators/BuildHashOperator.hpp
@@ -173,8 +173,6 @@ class BuildHashWorkOrder : public WorkOrder {
* @param join_key_attributes The IDs of equijoin attributes in
* input_relation.
* @param any_join_key_attributes_nullable If any attribute is nullable.
- * @param num_partitions The number of partitions in 'input_relation'. If no
- * partitions, it is one.
* @param part_id The partition id of 'input_relation'.
* @param build_block_id The block id.
* @param hash_table The JoinHashTable to use.
@@ -185,7 +183,6 @@ class BuildHashWorkOrder : public WorkOrder {
const CatalogRelationSchema &input_relation,
const std::vector<attribute_id> &join_key_attributes,
const bool any_join_key_attributes_nullable,
- const std::size_t num_partitions,
const partition_id part_id,
const block_id build_block_id,
JoinHashTable *hash_table,
@@ -195,7 +192,6 @@ class BuildHashWorkOrder : public WorkOrder {
input_relation_(input_relation),
join_key_attributes_(join_key_attributes),
any_join_key_attributes_nullable_(any_join_key_attributes_nullable),
- num_partitions_(num_partitions),
part_id_(part_id),
build_block_id_(build_block_id),
hash_table_(DCHECK_NOTNULL(hash_table)),
@@ -210,8 +206,6 @@ class BuildHashWorkOrder : public WorkOrder {
* @param join_key_attributes The IDs of equijoin attributes in
* input_relation.
* @param any_join_key_attributes_nullable If any attribute is nullable.
- * @param num_partitions The number of partitions in 'input_relation'. If no
- * partitions, it is one.
* @param part_id The partition id of 'input_relation'.
* @param build_block_id The block id.
* @param hash_table The JoinHashTable to use.
@@ -222,7 +216,6 @@ class BuildHashWorkOrder : public WorkOrder {
const CatalogRelationSchema &input_relation,
std::vector<attribute_id> &&join_key_attributes,
const bool any_join_key_attributes_nullable,
- const std::size_t num_partitions,
const partition_id part_id,
const block_id build_block_id,
JoinHashTable *hash_table,
@@ -232,7 +225,6 @@ class BuildHashWorkOrder : public WorkOrder {
input_relation_(input_relation),
join_key_attributes_(std::move(join_key_attributes)),
any_join_key_attributes_nullable_(any_join_key_attributes_nullable),
- num_partitions_(num_partitions),
part_id_(part_id),
build_block_id_(build_block_id),
hash_table_(DCHECK_NOTNULL(hash_table)),
@@ -248,15 +240,6 @@ class BuildHashWorkOrder : public WorkOrder {
void execute() override;
/**
- * @brief Get the number of partitions.
- *
- * @return The number of partitions.
- */
- std::size_t num_partitions() const {
- return num_partitions_;
- }
-
- /**
* @brief Get the partition id.
*
* @return The partition id.
@@ -269,7 +252,6 @@ class BuildHashWorkOrder : public WorkOrder {
const CatalogRelationSchema &input_relation_;
const std::vector<attribute_id> join_key_attributes_;
const bool any_join_key_attributes_nullable_;
- const std::size_t num_partitions_;
const partition_id part_id_;
const block_id build_block_id_;
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/968ce3f7/relational_operators/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/relational_operators/CMakeLists.txt b/relational_operators/CMakeLists.txt
index 66ea2d1..78da7b8 100644
--- a/relational_operators/CMakeLists.txt
+++ b/relational_operators/CMakeLists.txt
@@ -207,6 +207,7 @@ target_link_libraries(quickstep_relationaloperators_HashJoinOperator
quickstep_catalog_CatalogRelationSchema
quickstep_catalog_CatalogTypedefs
quickstep_catalog_PartitionScheme
+ quickstep_catalog_PartitionSchemeHeader
quickstep_expressions_predicate_Predicate
quickstep_expressions_scalar_Scalar
quickstep_queryexecution_QueryContext
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/968ce3f7/relational_operators/DestroyHashOperator.cpp
----------------------------------------------------------------------
diff --git a/relational_operators/DestroyHashOperator.cpp b/relational_operators/DestroyHashOperator.cpp
index 4827ef5..5b84bba 100644
--- a/relational_operators/DestroyHashOperator.cpp
+++ b/relational_operators/DestroyHashOperator.cpp
@@ -35,7 +35,7 @@ bool DestroyHashOperator::getAllWorkOrders(
const tmb::client_id scheduler_client_id,
tmb::MessageBus *bus) {
if (blocking_dependencies_met_ && !work_generated_) {
- for (std::size_t part_id = 0; part_id < num_partitions_; ++part_id) {
+ for (std::size_t part_id = 0; part_id < build_num_partitions_; ++part_id) {
container->addNormalWorkOrder(
new DestroyHashWorkOrder(query_id_, hash_table_index_, part_id, query_context),
op_index_);
@@ -47,7 +47,7 @@ bool DestroyHashOperator::getAllWorkOrders(
bool DestroyHashOperator::getAllWorkOrderProtos(WorkOrderProtosContainer *container) {
if (blocking_dependencies_met_ && !work_generated_) {
- for (std::size_t part_id = 0; part_id < num_partitions_; ++part_id) {
+ for (std::size_t part_id = 0; part_id < build_num_partitions_; ++part_id) {
serialization::WorkOrder *proto = new serialization::WorkOrder;
proto->set_work_order_type(serialization::DESTROY_HASH);
proto->set_query_id(query_id_);
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/968ce3f7/relational_operators/DestroyHashOperator.hpp
----------------------------------------------------------------------
diff --git a/relational_operators/DestroyHashOperator.hpp b/relational_operators/DestroyHashOperator.hpp
index 8a1fabd..b707999 100644
--- a/relational_operators/DestroyHashOperator.hpp
+++ b/relational_operators/DestroyHashOperator.hpp
@@ -53,14 +53,14 @@ class DestroyHashOperator : public RelationalOperator {
* @brief Constructor.
*
* @param query_id The ID of the query to which this operator belongs.
- * @param num_partitions The number of partitions.
+ * @param build_num_partitions The number of partitions in 'build_relation'.
* @param hash_table_index The index of the JoinHashTable in QueryContext.
**/
DestroyHashOperator(const std::size_t query_id,
- const std::size_t num_partitions,
+ const std::size_t build_num_partitions,
const QueryContext::join_hash_table_id hash_table_index)
: RelationalOperator(query_id),
- num_partitions_(num_partitions),
+ build_num_partitions_(build_num_partitions),
hash_table_index_(hash_table_index),
work_generated_(false) {}
@@ -79,7 +79,7 @@ class DestroyHashOperator : public RelationalOperator {
bool getAllWorkOrderProtos(WorkOrderProtosContainer *container) override;
private:
- const std::size_t num_partitions_;
+ const std::size_t build_num_partitions_;
const QueryContext::join_hash_table_id hash_table_index_;
bool work_generated_;
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/968ce3f7/relational_operators/HashJoinOperator.cpp
----------------------------------------------------------------------
diff --git a/relational_operators/HashJoinOperator.cpp b/relational_operators/HashJoinOperator.cpp
index 0062b93..fd3841f 100644
--- a/relational_operators/HashJoinOperator.cpp
+++ b/relational_operators/HashJoinOperator.cpp
@@ -213,15 +213,15 @@ bool HashJoinOperator::getAllNonOuterJoinWorkOrders(
return true;
}
- for (std::size_t part_id = 0; part_id < num_partitions_; ++part_id) {
+ for (std::size_t part_id = 0; part_id < build_num_partitions_; ++part_id) {
const JoinHashTable &hash_table =
*(query_context->getJoinHashTable(hash_table_index_, part_id));
for (const block_id probe_block_id : probe_relation_block_ids_[part_id]) {
container->addNormalWorkOrder(
new JoinWorkOrderClass(query_id_, build_relation_, probe_relation_, join_key_attributes_,
- any_join_key_attributes_nullable_, num_partitions_, part_id, probe_block_id,
- residual_predicate, selection, hash_table, output_destination, storage_manager,
+ any_join_key_attributes_nullable_, part_id, probe_block_id, residual_predicate,
+ selection, hash_table, output_destination, storage_manager,
CreateLIPFilterAdaptiveProberHelper(lip_deployment_index_, query_context)),
op_index_);
}
@@ -229,14 +229,14 @@ bool HashJoinOperator::getAllNonOuterJoinWorkOrders(
started_ = true;
return true;
} else {
- for (std::size_t part_id = 0; part_id < num_partitions_; ++part_id) {
+ for (std::size_t part_id = 0; part_id < build_num_partitions_; ++part_id) {
const JoinHashTable &hash_table =
*(query_context->getJoinHashTable(hash_table_index_, part_id));
while (num_workorders_generated_[part_id] < probe_relation_block_ids_[part_id].size()) {
container->addNormalWorkOrder(
new JoinWorkOrderClass(query_id_, build_relation_, probe_relation_, join_key_attributes_,
- any_join_key_attributes_nullable_, num_partitions_, part_id,
+ any_join_key_attributes_nullable_, part_id,
probe_relation_block_ids_[part_id][num_workorders_generated_[part_id]],
residual_predicate, selection, hash_table, output_destination, storage_manager,
CreateLIPFilterAdaptiveProberHelper(lip_deployment_index_, query_context)),
@@ -269,16 +269,15 @@ bool HashJoinOperator::getAllOuterJoinWorkOrders(
return true;
}
- for (std::size_t part_id = 0; part_id < num_partitions_; ++part_id) {
+ for (std::size_t part_id = 0; part_id < build_num_partitions_; ++part_id) {
const JoinHashTable &hash_table =
*(query_context->getJoinHashTable(hash_table_index_, part_id));
for (const block_id probe_block_id : probe_relation_block_ids_[part_id]) {
container->addNormalWorkOrder(
new HashOuterJoinWorkOrder(query_id_, build_relation_, probe_relation_, join_key_attributes_,
- any_join_key_attributes_nullable_, num_partitions_, part_id, probe_block_id,
- selection, is_selection_on_build_, hash_table, output_destination,
- storage_manager,
+ any_join_key_attributes_nullable_, part_id, probe_block_id, selection,
+ is_selection_on_build_, hash_table, output_destination, storage_manager,
CreateLIPFilterAdaptiveProberHelper(lip_deployment_index_, query_context)),
op_index_);
}
@@ -286,14 +285,14 @@ bool HashJoinOperator::getAllOuterJoinWorkOrders(
started_ = true;
return true;
} else {
- for (std::size_t part_id = 0; part_id < num_partitions_; ++part_id) {
+ for (std::size_t part_id = 0; part_id < build_num_partitions_; ++part_id) {
const JoinHashTable &hash_table =
*(query_context->getJoinHashTable(hash_table_index_, part_id));
while (num_workorders_generated_[part_id] < probe_relation_block_ids_[part_id].size()) {
container->addNormalWorkOrder(
new HashOuterJoinWorkOrder(query_id_, build_relation_, probe_relation_, join_key_attributes_,
- any_join_key_attributes_nullable_, num_partitions_, part_id,
+ any_join_key_attributes_nullable_, part_id,
probe_relation_block_ids_[part_id][num_workorders_generated_[part_id]],
selection, is_selection_on_build_, hash_table, output_destination,
storage_manager,
@@ -336,7 +335,7 @@ bool HashJoinOperator::getAllNonOuterJoinWorkOrderProtos(
return true;
}
- for (std::size_t part_id = 0; part_id < num_partitions_; ++part_id) {
+ for (std::size_t part_id = 0; part_id < build_num_partitions_; ++part_id) {
for (const block_id probe_block_id : probe_relation_block_ids_[part_id]) {
container->addWorkOrderProto(
createNonOuterJoinWorkOrderProto(hash_join_type, probe_block_id, part_id),
@@ -346,7 +345,7 @@ bool HashJoinOperator::getAllNonOuterJoinWorkOrderProtos(
started_ = true;
return true;
} else {
- for (std::size_t part_id = 0; part_id < num_partitions_; ++part_id) {
+ for (std::size_t part_id = 0; part_id < build_num_partitions_; ++part_id) {
while (num_workorders_generated_[part_id] < probe_relation_block_ids_[part_id].size()) {
container->addWorkOrderProto(
createNonOuterJoinWorkOrderProto(hash_join_type,
@@ -376,7 +375,6 @@ serialization::WorkOrder* HashJoinOperator::createNonOuterJoinWorkOrderProto(
}
proto->SetExtension(serialization::HashJoinWorkOrder::any_join_key_attributes_nullable,
any_join_key_attributes_nullable_);
- proto->SetExtension(serialization::HashJoinWorkOrder::num_partitions, num_partitions_);
proto->SetExtension(serialization::HashJoinWorkOrder::insert_destination_index, output_destination_index_);
proto->SetExtension(serialization::HashJoinWorkOrder::join_hash_table_index, hash_table_index_);
proto->SetExtension(serialization::HashJoinWorkOrder::partition_id, part_id);
@@ -399,7 +397,7 @@ bool HashJoinOperator::getAllOuterJoinWorkOrderProtos(WorkOrderProtosContainer *
return true;
}
- for (std::size_t part_id = 0; part_id < num_partitions_; ++part_id) {
+ for (std::size_t part_id = 0; part_id < build_num_partitions_; ++part_id) {
for (const block_id probe_block_id : probe_relation_block_ids_[part_id]) {
container->addWorkOrderProto(createOuterJoinWorkOrderProto(probe_block_id, part_id), op_index_);
}
@@ -407,7 +405,7 @@ bool HashJoinOperator::getAllOuterJoinWorkOrderProtos(WorkOrderProtosContainer *
started_ = true;
return true;
} else {
- for (std::size_t part_id = 0; part_id < num_partitions_; ++part_id) {
+ for (std::size_t part_id = 0; part_id < build_num_partitions_; ++part_id) {
while (num_workorders_generated_[part_id] < probe_relation_block_ids_[part_id].size()) {
container->addWorkOrderProto(
createOuterJoinWorkOrderProto(probe_relation_block_ids_[part_id][num_workorders_generated_[part_id]],
@@ -436,7 +434,6 @@ serialization::WorkOrder* HashJoinOperator::createOuterJoinWorkOrderProto(const
}
proto->SetExtension(serialization::HashJoinWorkOrder::any_join_key_attributes_nullable,
any_join_key_attributes_nullable_);
- proto->SetExtension(serialization::HashJoinWorkOrder::num_partitions, num_partitions_);
proto->SetExtension(serialization::HashJoinWorkOrder::insert_destination_index, output_destination_index_);
proto->SetExtension(serialization::HashJoinWorkOrder::join_hash_table_index, hash_table_index_);
proto->SetExtension(serialization::HashJoinWorkOrder::selection_index, selection_index_);
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/968ce3f7/relational_operators/HashJoinOperator.hpp
----------------------------------------------------------------------
diff --git a/relational_operators/HashJoinOperator.hpp b/relational_operators/HashJoinOperator.hpp
index e655f70..acfe3d2 100644
--- a/relational_operators/HashJoinOperator.hpp
+++ b/relational_operators/HashJoinOperator.hpp
@@ -29,6 +29,7 @@
#include "catalog/CatalogRelation.hpp"
#include "catalog/CatalogTypedefs.hpp"
#include "catalog/PartitionScheme.hpp"
+#include "catalog/PartitionSchemeHeader.hpp"
#include "query_execution/QueryContext.hpp"
#include "relational_operators/RelationalOperator.hpp"
#include "relational_operators/WorkOrder.hpp"
@@ -101,8 +102,8 @@ class HashJoinOperator : public RelationalOperator {
* @param join_key_attributes The IDs of equijoin attributes in
* probe_relation.
* @param any_join_key_attributes_nullable If any attribute is nullable.
- * @param num_partitions The number of partitions in 'input_relation'. If no
- * partitions, it is one.
+ * @param build_num_partitions The number of partitions in 'build_relation'.
+ * If no partitions, it is one.
* @param output_relation The output relation.
* @param output_destination_index The index of the InsertDestination in the
* QueryContext to insert the join results.
@@ -128,7 +129,7 @@ class HashJoinOperator : public RelationalOperator {
const bool probe_relation_is_stored,
const std::vector<attribute_id> &join_key_attributes,
const bool any_join_key_attributes_nullable,
- const std::size_t num_partitions,
+ const std::size_t build_num_partitions,
const CatalogRelation &output_relation,
const QueryContext::insert_destination_id output_destination_index,
const QueryContext::join_hash_table_id hash_table_index,
@@ -142,7 +143,7 @@ class HashJoinOperator : public RelationalOperator {
probe_relation_is_stored_(probe_relation_is_stored),
join_key_attributes_(join_key_attributes),
any_join_key_attributes_nullable_(any_join_key_attributes_nullable),
- num_partitions_(num_partitions),
+ build_num_partitions_(build_num_partitions),
output_relation_(output_relation),
output_destination_index_(output_destination_index),
hash_table_index_(hash_table_index),
@@ -152,8 +153,8 @@ class HashJoinOperator : public RelationalOperator {
? std::vector<bool>()
: *is_selection_on_build),
join_type_(join_type),
- probe_relation_block_ids_(num_partitions),
- num_workorders_generated_(num_partitions),
+ probe_relation_block_ids_(build_num_partitions),
+ num_workorders_generated_(build_num_partitions),
started_(false) {
DCHECK(join_type != JoinType::kLeftOuterJoin ||
(is_selection_on_build != nullptr &&
@@ -162,12 +163,15 @@ class HashJoinOperator : public RelationalOperator {
if (probe_relation_is_stored) {
if (probe_relation.hasPartitionScheme()) {
const PartitionScheme &part_scheme = *probe_relation.getPartitionScheme();
- for (std::size_t part_id = 0; part_id < num_partitions_; ++part_id) {
+ DCHECK_EQ(build_num_partitions_, part_scheme.getPartitionSchemeHeader().getNumPartitions());
+ for (std::size_t part_id = 0; part_id < build_num_partitions_; ++part_id) {
probe_relation_block_ids_[part_id] = part_scheme.getBlocksInPartition(part_id);
}
} else {
- // No partition.
- probe_relation_block_ids_[0] = probe_relation.getBlocksSnapshot();
+ // Broadcast hash join if probe has no partitions.
+ for (std::size_t part_id = 0; part_id < build_num_partitions_; ++part_id) {
+ probe_relation_block_ids_[part_id] = probe_relation.getBlocksSnapshot();
+ }
}
}
}
@@ -209,7 +213,14 @@ class HashJoinOperator : public RelationalOperator {
const partition_id part_id) override {
DCHECK_EQ(probe_relation_.getID(), input_relation_id);
- probe_relation_block_ids_[part_id].push_back(input_block_id);
+ if (probe_relation_.hasPartitionScheme()) {
+ probe_relation_block_ids_[part_id].push_back(input_block_id);
+ } else {
+ // Broadcast hash join if probe has no partitions.
+ for (std::size_t build_part_id = 0; build_part_id < build_num_partitions_; ++build_part_id) {
+ probe_relation_block_ids_[build_part_id].push_back(input_block_id);
+ }
+ }
}
QueryContext::insert_destination_id getInsertDestinationID() const override {
@@ -261,7 +272,7 @@ class HashJoinOperator : public RelationalOperator {
const bool probe_relation_is_stored_;
const std::vector<attribute_id> join_key_attributes_;
const bool any_join_key_attributes_nullable_;
- const std::size_t num_partitions_;
+ const std::size_t build_num_partitions_;
const CatalogRelation &output_relation_;
const QueryContext::insert_destination_id output_destination_index_;
const QueryContext::join_hash_table_id hash_table_index_;
@@ -295,8 +306,6 @@ class HashInnerJoinWorkOrder : public WorkOrder {
* @param join_key_attributes The IDs of equijoin attributes in \c
* probe_relation.
* @param any_join_key_attributes_nullable If any attribute is nullable.
- * @param num_partitions The number of partitions in 'probe_relation'. If no
- * partitions, it is one.
* @param part_id The partition id of 'probe_relation'.
* @param lookup_block_id The block id of the probe_relation.
* @param residual_predicate If non-null, apply as an additional filter to
@@ -317,7 +326,6 @@ class HashInnerJoinWorkOrder : public WorkOrder {
const CatalogRelationSchema &probe_relation,
const std::vector<attribute_id> &join_key_attributes,
const bool any_join_key_attributes_nullable,
- const std::size_t num_partitions,
const partition_id part_id,
const block_id lookup_block_id,
const Predicate *residual_predicate,
@@ -331,7 +339,6 @@ class HashInnerJoinWorkOrder : public WorkOrder {
probe_relation_(probe_relation),
join_key_attributes_(join_key_attributes),
any_join_key_attributes_nullable_(any_join_key_attributes_nullable),
- num_partitions_(num_partitions),
part_id_(part_id),
block_id_(lookup_block_id),
residual_predicate_(residual_predicate),
@@ -352,8 +359,6 @@ class HashInnerJoinWorkOrder : public WorkOrder {
* @param join_key_attributes The IDs of equijoin attributes in \c
* probe_relation.
* @param any_join_key_attributes_nullable If any attribute is nullable.
- * @param num_partitions The number of partitions in 'probe_relation'. If no
- * partitions, it is one.
* @param part_id The partition id of 'probe_relation'.
* @param lookup_block_id The block id of the probe_relation.
* @param residual_predicate If non-null, apply as an additional filter to
@@ -374,7 +379,6 @@ class HashInnerJoinWorkOrder : public WorkOrder {
const CatalogRelationSchema &probe_relation,
std::vector<attribute_id> &&join_key_attributes,
const bool any_join_key_attributes_nullable,
- const std::size_t num_partitions,
const partition_id part_id,
const block_id lookup_block_id,
const Predicate *residual_predicate,
@@ -388,7 +392,6 @@ class HashInnerJoinWorkOrder : public WorkOrder {
probe_relation_(probe_relation),
join_key_attributes_(std::move(join_key_attributes)),
any_join_key_attributes_nullable_(any_join_key_attributes_nullable),
- num_partitions_(num_partitions),
part_id_(part_id),
block_id_(lookup_block_id),
residual_predicate_(residual_predicate),
@@ -411,15 +414,6 @@ class HashInnerJoinWorkOrder : public WorkOrder {
void execute() override;
/**
- * @brief Get the number of partitions.
- *
- * @return The number of partitions.
- */
- std::size_t num_partitions() const {
- return num_partitions_;
- }
-
- /**
* @brief Get the partition id.
*
* @return The partition id.
@@ -433,7 +427,6 @@ class HashInnerJoinWorkOrder : public WorkOrder {
const CatalogRelationSchema &probe_relation_;
const std::vector<attribute_id> join_key_attributes_;
const bool any_join_key_attributes_nullable_;
- const std::size_t num_partitions_;
const partition_id part_id_;
const block_id block_id_;
const Predicate *residual_predicate_;
@@ -465,8 +458,6 @@ class HashSemiJoinWorkOrder : public WorkOrder {
* @param join_key_attributes The IDs of equijoin attributes in \c
* probe_relation.
* @param any_join_key_attributes_nullable If any attribute is nullable.
- * @param num_partitions The number of partitions in 'probe_relation'. If no
- * partitions, it is one.
* @param part_id The partition id of 'probe_relation'.
* @param lookup_block_id The block id of the probe_relation.
* @param residual_predicate If non-null, apply as an additional filter to
@@ -487,7 +478,6 @@ class HashSemiJoinWorkOrder : public WorkOrder {
const CatalogRelationSchema &probe_relation,
const std::vector<attribute_id> &join_key_attributes,
const bool any_join_key_attributes_nullable,
- const std::size_t num_partitions,
const partition_id part_id,
const block_id lookup_block_id,
const Predicate *residual_predicate,
@@ -501,7 +491,6 @@ class HashSemiJoinWorkOrder : public WorkOrder {
probe_relation_(probe_relation),
join_key_attributes_(join_key_attributes),
any_join_key_attributes_nullable_(any_join_key_attributes_nullable),
- num_partitions_(num_partitions),
part_id_(part_id),
block_id_(lookup_block_id),
residual_predicate_(residual_predicate),
@@ -522,8 +511,6 @@ class HashSemiJoinWorkOrder : public WorkOrder {
* @param join_key_attributes The IDs of equijoin attributes in \c
* probe_relation.
* @param any_join_key_attributes_nullable If any attribute is nullable.
- * @param num_partitions The number of partitions in 'probe_relation'. If no
- * partitions, it is one.
* @param part_id The partition id of 'probe_relation'.
* @param lookup_block_id The block id of the probe_relation.
* @param residual_predicate If non-null, apply as an additional filter to
@@ -544,7 +531,6 @@ class HashSemiJoinWorkOrder : public WorkOrder {
const CatalogRelationSchema &probe_relation,
std::vector<attribute_id> &&join_key_attributes,
const bool any_join_key_attributes_nullable,
- const std::size_t num_partitions,
const partition_id part_id,
const block_id lookup_block_id,
const Predicate *residual_predicate,
@@ -558,7 +544,6 @@ class HashSemiJoinWorkOrder : public WorkOrder {
probe_relation_(probe_relation),
join_key_attributes_(std::move(join_key_attributes)),
any_join_key_attributes_nullable_(any_join_key_attributes_nullable),
- num_partitions_(num_partitions),
part_id_(part_id),
block_id_(lookup_block_id),
residual_predicate_(residual_predicate),
@@ -573,15 +558,6 @@ class HashSemiJoinWorkOrder : public WorkOrder {
void execute() override;
/**
- * @brief Get the number of partitions.
- *
- * @return The number of partitions.
- */
- std::size_t num_partitions() const {
- return num_partitions_;
- }
-
- /**
* @brief Get the partition id.
*
* @return The partition id.
@@ -599,7 +575,6 @@ class HashSemiJoinWorkOrder : public WorkOrder {
const CatalogRelationSchema &probe_relation_;
const std::vector<attribute_id> join_key_attributes_;
const bool any_join_key_attributes_nullable_;
- const std::size_t num_partitions_;
const partition_id part_id_;
const block_id block_id_;
const Predicate *residual_predicate_;
@@ -631,8 +606,6 @@ class HashAntiJoinWorkOrder : public WorkOrder {
* @param join_key_attributes The IDs of equijoin attributes in \c
* probe_relation.
* @param any_join_key_attributes_nullable If any attribute is nullable.
- * @param num_partitions The number of partitions in 'probe_relation'. If no
- * partitions, it is one.
* @param part_id The partition id of 'probe_relation'.
* @param lookup_block_id The block id of the probe_relation.
* @param residual_predicate If non-null, apply as an additional filter to
@@ -653,7 +626,6 @@ class HashAntiJoinWorkOrder : public WorkOrder {
const CatalogRelationSchema &probe_relation,
const std::vector<attribute_id> &join_key_attributes,
const bool any_join_key_attributes_nullable,
- const std::size_t num_partitions,
const partition_id part_id,
const block_id lookup_block_id,
const Predicate *residual_predicate,
@@ -667,7 +639,6 @@ class HashAntiJoinWorkOrder : public WorkOrder {
probe_relation_(probe_relation),
join_key_attributes_(join_key_attributes),
any_join_key_attributes_nullable_(any_join_key_attributes_nullable),
- num_partitions_(num_partitions),
part_id_(part_id),
block_id_(lookup_block_id),
residual_predicate_(residual_predicate),
@@ -688,8 +659,6 @@ class HashAntiJoinWorkOrder : public WorkOrder {
* @param join_key_attributes The IDs of equijoin attributes in \c
* probe_relation.
* @param any_join_key_attributes_nullable If any attribute is nullable.
- * @param num_partitions The number of partitions in 'probe_relation'. If no
- * partitions, it is one.
* @param part_id The partition id of 'probe_relation'.
* @param lookup_block_id The block id of the probe_relation.
* @param residual_predicate If non-null, apply as an additional filter to
@@ -710,7 +679,6 @@ class HashAntiJoinWorkOrder : public WorkOrder {
const CatalogRelationSchema &probe_relation,
std::vector<attribute_id> &&join_key_attributes,
const bool any_join_key_attributes_nullable,
- const std::size_t num_partitions,
const partition_id part_id,
const block_id lookup_block_id,
const Predicate *residual_predicate,
@@ -724,7 +692,6 @@ class HashAntiJoinWorkOrder : public WorkOrder {
probe_relation_(probe_relation),
join_key_attributes_(std::move(join_key_attributes)),
any_join_key_attributes_nullable_(any_join_key_attributes_nullable),
- num_partitions_(num_partitions),
part_id_(part_id),
block_id_(lookup_block_id),
residual_predicate_(residual_predicate),
@@ -745,15 +712,6 @@ class HashAntiJoinWorkOrder : public WorkOrder {
}
/**
- * @brief Get the number of partitions.
- *
- * @return The number of partitions.
- */
- std::size_t num_partitions() const {
- return num_partitions_;
- }
-
- /**
* @brief Get the partition id.
*
* @return The partition id.
@@ -771,7 +729,6 @@ class HashAntiJoinWorkOrder : public WorkOrder {
const CatalogRelationSchema &probe_relation_;
const std::vector<attribute_id> join_key_attributes_;
const bool any_join_key_attributes_nullable_;
- const std::size_t num_partitions_;
const partition_id part_id_;
const block_id block_id_;
const Predicate *residual_predicate_;
@@ -802,8 +759,6 @@ class HashOuterJoinWorkOrder : public WorkOrder {
* @param join_key_attributes The IDs of equijoin attributes in \c
* probe_relation.
* @param any_join_key_attributes_nullable If any attribute is nullable.
- * @param num_partitions The number of partitions in 'probe_relation'. If no
- * partitions, it is one.
* @param part_id The partition id of 'probe_relation'.
* @param lookup_block_id The block id of the probe_relation.
* @param selection A list of Scalars corresponding to the relation attributes
@@ -823,7 +778,6 @@ class HashOuterJoinWorkOrder : public WorkOrder {
const CatalogRelationSchema &probe_relation,
const std::vector<attribute_id> &join_key_attributes,
const bool any_join_key_attributes_nullable,
- const std::size_t num_partitions,
const partition_id part_id,
const block_id lookup_block_id,
const std::vector<std::unique_ptr<const Scalar>> &selection,
@@ -837,7 +791,6 @@ class HashOuterJoinWorkOrder : public WorkOrder {
probe_relation_(probe_relation),
join_key_attributes_(join_key_attributes),
any_join_key_attributes_nullable_(any_join_key_attributes_nullable),
- num_partitions_(num_partitions),
part_id_(part_id),
block_id_(lookup_block_id),
selection_(selection),
@@ -858,8 +811,6 @@ class HashOuterJoinWorkOrder : public WorkOrder {
* @param join_key_attributes The IDs of equijoin attributes in \c
* probe_relation.
* @param any_join_key_attributes_nullable If any attribute is nullable.
- * @param num_partitions The number of partitions in 'probe_relation'. If no
- * partitions, it is one.
* @param part_id The partition id of 'probe_relation'.
* @param lookup_block_id The block id of the probe_relation.
* @param selection A list of Scalars corresponding to the relation attributes
@@ -878,7 +829,6 @@ class HashOuterJoinWorkOrder : public WorkOrder {
const CatalogRelationSchema &probe_relation,
std::vector<attribute_id> &&join_key_attributes,
const bool any_join_key_attributes_nullable,
- const std::size_t num_partitions,
const partition_id part_id,
const block_id lookup_block_id,
const std::vector<std::unique_ptr<const Scalar>> &selection,
@@ -892,7 +842,6 @@ class HashOuterJoinWorkOrder : public WorkOrder {
probe_relation_(probe_relation),
join_key_attributes_(std::move(join_key_attributes)),
any_join_key_attributes_nullable_(any_join_key_attributes_nullable),
- num_partitions_(num_partitions),
part_id_(part_id),
block_id_(lookup_block_id),
selection_(selection),
@@ -907,15 +856,6 @@ class HashOuterJoinWorkOrder : public WorkOrder {
void execute() override;
/**
- * @brief Get the number of partitions.
- *
- * @return The number of partitions.
- */
- std::size_t num_partitions() const {
- return num_partitions_;
- }
-
- /**
* @brief Get the partition id.
*
* @return The partition id.
@@ -929,7 +869,6 @@ class HashOuterJoinWorkOrder : public WorkOrder {
const CatalogRelationSchema &probe_relation_;
const std::vector<attribute_id> join_key_attributes_;
const bool any_join_key_attributes_nullable_;
- const std::size_t num_partitions_;
const partition_id part_id_;
const block_id block_id_;
const std::vector<std::unique_ptr<const Scalar>> &selection_;
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/968ce3f7/relational_operators/WorkOrder.proto
----------------------------------------------------------------------
diff --git a/relational_operators/WorkOrder.proto b/relational_operators/WorkOrder.proto
index b914fce..f8d9246 100644
--- a/relational_operators/WorkOrder.proto
+++ b/relational_operators/WorkOrder.proto
@@ -63,16 +63,15 @@ message AggregationWorkOrder {
}
}
-// Next tag: 40.
+// Next tag: 39.
message BuildHashWorkOrder {
extend WorkOrder {
// All required.
optional int32 relation_id = 32;
repeated int32 join_key_attributes = 33;
optional bool any_join_key_attributes_nullable = 34;
- optional uint64 num_partitions = 38;
optional uint32 join_hash_table_index = 35;
- optional uint64 partition_id = 39;
+ optional uint64 partition_id = 38;
optional fixed64 block_id = 36;
optional int32 lip_deployment_index = 37;
}
@@ -113,7 +112,7 @@ message FinalizeAggregationWorkOrder {
}
}
-// Next tag: 174.
+// Next tag: 173.
message HashJoinWorkOrder {
enum HashJoinWorkOrderType {
HASH_ANTI_JOIN = 0;
@@ -129,10 +128,9 @@ message HashJoinWorkOrder {
optional int32 probe_relation_id = 162;
repeated int32 join_key_attributes = 163;
optional bool any_join_key_attributes_nullable = 164;
- optional uint64 num_partitions = 172;
optional int32 insert_destination_index = 165;
optional uint32 join_hash_table_index = 166;
- optional uint64 partition_id = 173;
+ optional uint64 partition_id = 172;
optional int32 selection_index = 167;
optional fixed64 block_id = 168;
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/968ce3f7/relational_operators/WorkOrderFactory.cpp
----------------------------------------------------------------------
diff --git a/relational_operators/WorkOrderFactory.cpp b/relational_operators/WorkOrderFactory.cpp
index c09bcbe..a6cba02 100644
--- a/relational_operators/WorkOrderFactory.cpp
+++ b/relational_operators/WorkOrderFactory.cpp
@@ -107,7 +107,6 @@ WorkOrder* WorkOrderFactory::ReconstructFromProto(const serialization::WorkOrder
proto.GetExtension(serialization::BuildHashWorkOrder::relation_id)),
move(join_key_attributes),
proto.GetExtension(serialization::BuildHashWorkOrder::any_join_key_attributes_nullable),
- proto.GetExtension(serialization::BuildHashWorkOrder::num_partitions),
part_id,
proto.GetExtension(serialization::BuildHashWorkOrder::block_id),
query_context->getJoinHashTable(
@@ -199,9 +198,6 @@ WorkOrder* WorkOrderFactory::ReconstructFromProto(const serialization::WorkOrder
const block_id lookup_block_id =
proto.GetExtension(serialization::HashJoinWorkOrder::block_id);
- const std::size_t num_partitions =
- proto.GetExtension(serialization::HashJoinWorkOrder::num_partitions);
-
const Predicate *residual_predicate = nullptr;
if (hash_join_work_order_type != serialization::HashJoinWorkOrder::HASH_OUTER_JOIN) {
residual_predicate =
@@ -233,7 +229,6 @@ WorkOrder* WorkOrderFactory::ReconstructFromProto(const serialization::WorkOrder
probe_relation,
move(join_key_attributes),
any_join_key_attributes_nullable,
- num_partitions,
part_id,
lookup_block_id,
residual_predicate,
@@ -251,7 +246,6 @@ WorkOrder* WorkOrderFactory::ReconstructFromProto(const serialization::WorkOrder
probe_relation,
move(join_key_attributes),
any_join_key_attributes_nullable,
- num_partitions,
part_id,
lookup_block_id,
residual_predicate,
@@ -277,7 +271,6 @@ WorkOrder* WorkOrderFactory::ReconstructFromProto(const serialization::WorkOrder
probe_relation,
move(join_key_attributes),
any_join_key_attributes_nullable,
- num_partitions,
part_id,
lookup_block_id,
selection,
@@ -295,7 +288,6 @@ WorkOrder* WorkOrderFactory::ReconstructFromProto(const serialization::WorkOrder
probe_relation,
move(join_key_attributes),
any_join_key_attributes_nullable,
- num_partitions,
part_id,
lookup_block_id,
residual_predicate,
@@ -544,7 +536,6 @@ bool WorkOrderFactory::ProtoIsValid(const serialization::WorkOrder &proto,
return proto.HasExtension(serialization::BuildHashWorkOrder::any_join_key_attributes_nullable) &&
proto.HasExtension(serialization::BuildHashWorkOrder::block_id) &&
proto.HasExtension(serialization::BuildHashWorkOrder::join_hash_table_index) &&
- proto.HasExtension(serialization::BuildHashWorkOrder::num_partitions) &&
proto.HasExtension(serialization::BuildHashWorkOrder::partition_id) &&
query_context.isValidJoinHashTableId(
proto.GetExtension(serialization::BuildHashWorkOrder::join_hash_table_index),
@@ -648,7 +639,6 @@ bool WorkOrderFactory::ProtoIsValid(const serialization::WorkOrder &proto,
query_context.isValidInsertDestinationId(
proto.GetExtension(serialization::HashJoinWorkOrder::insert_destination_index)) &&
proto.HasExtension(serialization::HashJoinWorkOrder::join_hash_table_index) &&
- proto.HasExtension(serialization::HashJoinWorkOrder::num_partitions) &&
proto.HasExtension(serialization::HashJoinWorkOrder::partition_id) &&
query_context.isValidJoinHashTableId(
proto.GetExtension(serialization::HashJoinWorkOrder::join_hash_table_index),