Return-Path: X-Original-To: archive-asf-public-internal@cust-asf2.ponee.io Delivered-To: archive-asf-public-internal@cust-asf2.ponee.io Received: from cust-asf.ponee.io (cust-asf.ponee.io [163.172.22.183]) by cust-asf2.ponee.io (Postfix) with ESMTP id 2A5E2200C10 for ; Fri, 20 Jan 2017 02:08:31 +0100 (CET) Received: by cust-asf.ponee.io (Postfix) id 29309160B57; Fri, 20 Jan 2017 01:08:31 +0000 (UTC) Delivered-To: archive-asf-public@cust-asf.ponee.io Received: from mail.apache.org (hermes.apache.org [140.211.11.3]) by cust-asf.ponee.io (Postfix) with SMTP id CE89F160B54 for ; Fri, 20 Jan 2017 02:08:28 +0100 (CET) Received: (qmail 1248 invoked by uid 500); 20 Jan 2017 01:08:27 -0000 Mailing-List: contact commits-help@quickstep.incubator.apache.org; run by ezmlm Precedence: bulk List-Help: List-Unsubscribe: List-Post: List-Id: Reply-To: dev@quickstep.incubator.apache.org Delivered-To: mailing list commits@quickstep.incubator.apache.org Received: (qmail 1239 invoked by uid 99); 20 Jan 2017 01:08:27 -0000 Received: from pnap-us-west-generic-nat.apache.org (HELO spamd1-us-west.apache.org) (209.188.14.142) by apache.org (qpsmtpd/0.29) with ESMTP; Fri, 20 Jan 2017 01:08:27 +0000 Received: from localhost (localhost [127.0.0.1]) by spamd1-us-west.apache.org (ASF Mail Server at spamd1-us-west.apache.org) with ESMTP id 448EBC1368 for ; Fri, 20 Jan 2017 01:08:27 +0000 (UTC) X-Virus-Scanned: Debian amavisd-new at spamd1-us-west.apache.org X-Spam-Flag: NO X-Spam-Score: -6.218 X-Spam-Level: X-Spam-Status: No, score=-6.218 tagged_above=-999 required=6.31 tests=[FILL_THIS_FORM=0.001, KAM_ASCII_DIVIDERS=0.8, KAM_LAZY_DOMAIN_SECURITY=1, RCVD_IN_DNSWL_HI=-5, RCVD_IN_MSPIKE_H3=-0.01, RCVD_IN_MSPIKE_WL=-0.01, RP_MATCHES_RCVD=-2.999] autolearn=disabled Received: from mx1-lw-eu.apache.org ([10.40.0.8]) by localhost (spamd1-us-west.apache.org [10.40.0.7]) (amavisd-new, port 10024) with ESMTP id rdOvkyQHy1E2 for ; Fri, 20 Jan 2017 01:08:22 +0000 (UTC) Received: from mail.apache.org (hermes.apache.org [140.211.11.3]) by mx1-lw-eu.apache.org (ASF Mail Server at mx1-lw-eu.apache.org) with SMTP id 54A995F47E for ; Fri, 20 Jan 2017 01:08:20 +0000 (UTC) Received: (qmail 1206 invoked by uid 99); 20 Jan 2017 01:08:19 -0000 Received: from git1-us-west.apache.org (HELO git1-us-west.apache.org) (140.211.11.23) by apache.org (qpsmtpd/0.29) with ESMTP; Fri, 20 Jan 2017 01:08:19 +0000 Received: by git1-us-west.apache.org (ASF Mail Server at git1-us-west.apache.org, from userid 33) id 6EC53DFC16; Fri, 20 Jan 2017 01:08:19 +0000 (UTC) Content-Type: text/plain; charset="us-ascii" MIME-Version: 1.0 Content-Transfer-Encoding: 7bit From: zuyuz@apache.org To: commits@quickstep.incubator.apache.org Message-Id: <7d855a3faff84371ad5cd92d637d35a9@git.apache.org> X-Mailer: ASF-Git Admin Mailer Subject: incubator-quickstep git commit: Added Operator support for Partitioned HashJoin. Date: Fri, 20 Jan 2017 01:08:19 +0000 (UTC) archived-at: Fri, 20 Jan 2017 01:08:31 -0000 Repository: incubator-quickstep Updated Branches: refs/heads/partitioned_join_hash_table [created] 8e452194c Added Operator support for Partitioned HashJoin. Project: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/commit/8e452194 Tree: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/tree/8e452194 Diff: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/diff/8e452194 Branch: refs/heads/partitioned_join_hash_table Commit: 8e452194cf5c4e8e33865ada905b7b123ca281a2 Parents: ee3b7f0 Author: Zuyu Zhang Authored: Tue Jan 17 13:18:08 2017 -0800 Committer: Zuyu Zhang Committed: Thu Jan 19 17:07:58 2017 -0800 ---------------------------------------------------------------------- query_execution/QueryContext.cpp | 14 +- query_execution/QueryContext.hpp | 25 ++- query_execution/QueryContext.proto | 7 +- query_optimizer/ExecutionGenerator.cpp | 30 ++- relational_operators/BuildHashOperator.cpp | 75 ++++---- relational_operators/BuildHashOperator.hpp | 51 ++++- relational_operators/CMakeLists.txt | 9 +- relational_operators/DestroyHashOperator.cpp | 25 ++- relational_operators/DestroyHashOperator.hpp | 9 + relational_operators/HashJoinOperator.cpp | 189 ++++++++++--------- relational_operators/HashJoinOperator.hpp | 102 +++++++++- relational_operators/WorkOrder.proto | 5 + relational_operators/WorkOrderFactory.cpp | 28 ++- .../tests/HashJoinOperator_unittest.cpp | 42 +++-- storage/StorageBlockInfo.hpp | 3 + 15 files changed, 419 insertions(+), 195 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/8e452194/query_execution/QueryContext.cpp ---------------------------------------------------------------------- diff --git a/query_execution/QueryContext.cpp b/query_execution/QueryContext.cpp index 0e6636d..71839a7 100644 --- a/query_execution/QueryContext.cpp +++ b/query_execution/QueryContext.cpp @@ -79,9 +79,15 @@ QueryContext::QueryContext(const serialization::QueryContext &proto, } for (int i = 0; i < proto.join_hash_tables_size(); ++i) { - join_hash_tables_.emplace_back( - JoinHashTableFactory::CreateResizableFromProto(proto.join_hash_tables(i), - storage_manager)); + PartitionedJoinHashTables partitioned_join_hash_tables; + + const serialization::QueryContext::HashTableContext &hash_table_context_proto = proto.join_hash_tables(i); + for (std::uint64_t j = 0; j < hash_table_context_proto.num_partitions(); ++j) { + partitioned_join_hash_tables.emplace_back( + JoinHashTableFactory::CreateResizableFromProto(hash_table_context_proto.join_hash_table(), storage_manager)); + } + + join_hash_tables_.push_back(move(partitioned_join_hash_tables)); } for (int i = 0; i < proto.insert_destinations_size(); ++i) { @@ -178,7 +184,7 @@ bool QueryContext::ProtoIsValid(const serialization::QueryContext &proto, } for (int i = 0; i < proto.join_hash_tables_size(); ++i) { - if (!JoinHashTableFactory::ProtoIsValid(proto.join_hash_tables(i))) { + if (!JoinHashTableFactory::ProtoIsValid(proto.join_hash_tables(i).join_hash_table())) { return false; } } http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/8e452194/query_execution/QueryContext.hpp ---------------------------------------------------------------------- diff --git a/query_execution/QueryContext.hpp b/query_execution/QueryContext.hpp index 7ad8fa1..895c2ea 100644 --- a/query_execution/QueryContext.hpp +++ b/query_execution/QueryContext.hpp @@ -278,33 +278,37 @@ class QueryContext { * @brief Whether the given JoinHashTable id is valid. * * @param id The JoinHashTable id. + * @param part_id The partition id. * * @return True if valid, otherwise false. **/ - bool isValidJoinHashTableId(const join_hash_table_id id) const { - return id < join_hash_tables_.size(); + bool isValidJoinHashTableId(const join_hash_table_id id, const partition_id part_id) const { + return id < join_hash_tables_.size() && + part_id < join_hash_tables_[id].size(); } /** * @brief Get the JoinHashTable. * * @param id The JoinHashTable id in the query. + * @param part_id The partition id. * * @return The JoinHashTable, already created in the constructor. **/ - inline JoinHashTable* getJoinHashTable(const join_hash_table_id id) { - DCHECK_LT(id, join_hash_tables_.size()); - return join_hash_tables_[id].get(); + inline JoinHashTable* getJoinHashTable(const join_hash_table_id id, const partition_id part_id) { + DCHECK(isValidJoinHashTableId(id, part_id)); + return join_hash_tables_[id][part_id].get(); } /** * @brief Destory the given JoinHashTable. * * @param id The id of the JoinHashTable to destroy. + * @param part_id The partition id. **/ - inline void destroyJoinHashTable(const join_hash_table_id id) { - DCHECK_LT(id, join_hash_tables_.size()); - join_hash_tables_[id].reset(); + inline void destroyJoinHashTable(const join_hash_table_id id, const partition_id part_id) { + DCHECK(isValidJoinHashTableId(id, part_id)); + join_hash_tables_[id][part_id].reset(); } /** @@ -562,10 +566,13 @@ class QueryContext { } private: + // Per hash join, the index is the partition id. + typedef std::vector> PartitionedJoinHashTables; + std::vector> aggregation_states_; std::vector> generator_functions_; std::vector> insert_destinations_; - std::vector> join_hash_tables_; + std::vector join_hash_tables_; std::vector> lip_deployments_; std::vector> lip_filters_; std::vector> predicates_; http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/8e452194/query_execution/QueryContext.proto ---------------------------------------------------------------------- diff --git a/query_execution/QueryContext.proto b/query_execution/QueryContext.proto index ab0f520..6dce6b8 100644 --- a/query_execution/QueryContext.proto +++ b/query_execution/QueryContext.proto @@ -30,6 +30,11 @@ import "utility/SortConfiguration.proto"; import "utility/lip_filter/LIPFilter.proto"; message QueryContext { + message HashTableContext { + required HashTable join_hash_table = 1; + optional uint64 num_partitions = 2 [default = 1]; + } + message ScalarGroup { repeated Scalar scalars = 1; } @@ -47,7 +52,7 @@ message QueryContext { repeated AggregationOperationState aggregation_states = 1; repeated GeneratorFunctionHandle generator_functions = 2; - repeated HashTable join_hash_tables = 3; + repeated HashTableContext join_hash_tables = 3; repeated InsertDestination insert_destinations = 4; repeated LIPFilter lip_filters = 5; repeated LIPFilterDeployment lip_filter_deployments = 6; http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/8e452194/query_optimizer/ExecutionGenerator.cpp ---------------------------------------------------------------------- diff --git a/query_optimizer/ExecutionGenerator.cpp b/query_optimizer/ExecutionGenerator.cpp index e0bfb3b..e25b8ad 100644 --- a/query_optimizer/ExecutionGenerator.cpp +++ b/query_optimizer/ExecutionGenerator.cpp @@ -694,15 +694,32 @@ void ExecutionGenerator::convertHashJoin(const P::HashJoinPtr &physical_plan) { build_physical->getOutputAttributes()))); } + const CatalogRelation *build_relation = build_relation_info->relation; + // FIXME(quickstep-team): Add support for self-join. - if (build_relation_info->relation == probe_operator_info->relation) { + if (build_relation == probe_operator_info->relation) { THROW_SQL_ERROR() << "Self-join is not supported"; } // Create join hash table proto. const QueryContext::join_hash_table_id join_hash_table_index = query_context_proto_->join_hash_tables_size(); - S::HashTable *hash_table_proto = query_context_proto_->add_join_hash_tables(); + S::QueryContext::HashTableContext *hash_table_context_proto = + query_context_proto_->add_join_hash_tables(); + + // No partition. + std::size_t num_partitions = 1; + if (build_relation->hasPartitionScheme() && + build_attribute_ids.size() == 1) { + const PartitionSchemeHeader &partition_scheme_header = + build_relation->getPartitionScheme()->getPartitionSchemeHeader(); + if (build_attribute_ids[0] == partition_scheme_header.getPartitionAttributeId()) { + // TODO(zuyu): add optimizer support for partitioned hash joins. + hash_table_context_proto->set_num_partitions(num_partitions); + } + } + + S::HashTable *hash_table_proto = hash_table_context_proto->mutable_join_hash_table(); // SimplifyHashTableImplTypeProto() switches the hash table implementation // from SeparateChaining to SimpleScalarSeparateChaining when there is a @@ -712,7 +729,6 @@ void ExecutionGenerator::convertHashJoin(const P::HashJoinPtr &physical_plan) { HashTableImplTypeProtoFromString(FLAGS_join_hashtable_type), key_types)); - const CatalogRelationSchema *build_relation = build_relation_info->relation; for (const attribute_id build_attribute : build_attribute_ids) { hash_table_proto->add_key_types()->CopyFrom( build_relation->getAttributeById(build_attribute)->getType().getProto()); @@ -725,10 +741,11 @@ void ExecutionGenerator::convertHashJoin(const P::HashJoinPtr &physical_plan) { execution_plan_->addRelationalOperator( new BuildHashOperator( query_handle_->query_id(), - *build_relation_info->relation, + *build_relation, build_relation_info->isStoredRelation(), build_attribute_ids, any_build_attributes_nullable, + num_partitions, join_hash_table_index)); // Create InsertDestination proto. @@ -766,11 +783,12 @@ void ExecutionGenerator::convertHashJoin(const P::HashJoinPtr &physical_plan) { execution_plan_->addRelationalOperator( new HashJoinOperator( query_handle_->query_id(), - *build_relation_info->relation, + *build_relation, *probe_operator_info->relation, probe_operator_info->isStoredRelation(), probe_attribute_ids, any_probe_attributes_nullable, + num_partitions, *output_relation, insert_destination_index, join_hash_table_index, @@ -782,7 +800,7 @@ void ExecutionGenerator::convertHashJoin(const P::HashJoinPtr &physical_plan) { const QueryPlan::DAGNodeIndex destroy_operator_index = execution_plan_->addRelationalOperator(new DestroyHashOperator( - query_handle_->query_id(), join_hash_table_index)); + query_handle_->query_id(), num_partitions, join_hash_table_index)); if (!build_relation_info->isStoredRelation()) { execution_plan_->addDirectDependency(build_operator_index, http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/8e452194/relational_operators/BuildHashOperator.cpp ---------------------------------------------------------------------- diff --git a/relational_operators/BuildHashOperator.cpp b/relational_operators/BuildHashOperator.cpp index 60e091f..bfcaef4 100644 --- a/relational_operators/BuildHashOperator.cpp +++ b/relational_operators/BuildHashOperator.cpp @@ -70,38 +70,36 @@ bool BuildHashOperator::getAllWorkOrders( tmb::MessageBus *bus) { DCHECK(query_context != nullptr); - JoinHashTable *hash_table = query_context->getJoinHashTable(hash_table_index_); if (input_relation_is_stored_) { - if (!started_) { - for (const block_id input_block_id : input_relation_block_ids_) { + if (started_) { + return true; + } + + for (std::size_t part_id = 0; part_id < num_partitions_; ++part_id) { + JoinHashTable *hash_table = query_context->getJoinHashTable(hash_table_index_, part_id); + 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_, - input_block_id, - hash_table, - storage_manager, + new BuildHashWorkOrder(query_id_, input_relation_, join_key_attributes_, any_join_key_attributes_nullable_, + block, part_id, hash_table, storage_manager, CreateLIPFilterBuilderHelper(lip_deployment_index_, query_context)), op_index_); } - started_ = true; } - return started_; + started_ = true; + return true; } else { - while (num_workorders_generated_ < input_relation_block_ids_.size()) { - container->addNormalWorkOrder( - new BuildHashWorkOrder( - query_id_, - input_relation_, - join_key_attributes_, - any_join_key_attributes_nullable_, - input_relation_block_ids_[num_workorders_generated_], - hash_table, - storage_manager, - CreateLIPFilterBuilderHelper(lip_deployment_index_, query_context)), - op_index_); - ++num_workorders_generated_; + for (std::size_t part_id = 0; part_id < num_partitions_; ++part_id) { + JoinHashTable *hash_table = query_context->getJoinHashTable(hash_table_index_, part_id); + while (num_workorders_generated_[part_id] < + input_relation_block_ids_[part_id].size()) { + container->addNormalWorkOrder( + new BuildHashWorkOrder(query_id_, input_relation_, join_key_attributes_, any_join_key_attributes_nullable_, + input_relation_block_ids_[part_id][num_workorders_generated_[part_id]], part_id, + hash_table, storage_manager, + CreateLIPFilterBuilderHelper(lip_deployment_index_, query_context)), + op_index_); + ++num_workorders_generated_[part_id]; + } } return done_feeding_input_relation_; } @@ -109,25 +107,31 @@ bool BuildHashOperator::getAllWorkOrders( bool BuildHashOperator::getAllWorkOrderProtos(WorkOrderProtosContainer *container) { if (input_relation_is_stored_) { - if (!started_) { - for (const block_id input_block_id : input_relation_block_ids_) { - container->addWorkOrderProto(createWorkOrderProto(input_block_id), op_index_); + if (started_) { + return true; + } + + for (std::size_t part_id = 0; part_id < num_partitions_; ++part_id) { + for (const block_id block : input_relation_block_ids_[part_id]) { + container->addWorkOrderProto(createWorkOrderProto(block, part_id), op_index_); } - started_ = true; } + started_ = true; return true; } else { - while (num_workorders_generated_ < input_relation_block_ids_.size()) { - container->addWorkOrderProto( - createWorkOrderProto(input_relation_block_ids_[num_workorders_generated_]), - op_index_); - ++num_workorders_generated_; + for (std::size_t part_id = 0; part_id < num_partitions_; ++part_id) { + while (num_workorders_generated_[part_id] < input_relation_block_ids_[part_id].size()) { + container->addWorkOrderProto( + createWorkOrderProto(input_relation_block_ids_[part_id][num_workorders_generated_[part_id]], part_id), + op_index_); + ++num_workorders_generated_[part_id]; + } } return done_feeding_input_relation_; } } -serialization::WorkOrder* BuildHashOperator::createWorkOrderProto(const block_id block) { +serialization::WorkOrder* BuildHashOperator::createWorkOrderProto(const block_id block, const partition_id part_id) { serialization::WorkOrder *proto = new serialization::WorkOrder; proto->set_work_order_type(serialization::BUILD_HASH); proto->set_query_id(query_id_); @@ -139,6 +143,7 @@ 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::join_hash_table_index, hash_table_index_); + proto->SetExtension(serialization::BuildHashWorkOrder::partition_id, part_id); proto->SetExtension(serialization::BuildHashWorkOrder::block_id, block); proto->SetExtension(serialization::BuildHashWorkOrder::lip_deployment_index, lip_deployment_index_); http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/8e452194/relational_operators/BuildHashOperator.hpp ---------------------------------------------------------------------- diff --git a/relational_operators/BuildHashOperator.hpp b/relational_operators/BuildHashOperator.hpp index dec121c..e3ea74c 100644 --- a/relational_operators/BuildHashOperator.hpp +++ b/relational_operators/BuildHashOperator.hpp @@ -27,6 +27,7 @@ #include "catalog/CatalogRelation.hpp" #include "catalog/CatalogTypedefs.hpp" +#include "catalog/PartitionScheme.hpp" #include "query_execution/QueryContext.hpp" #include "relational_operators/RelationalOperator.hpp" #include "relational_operators/WorkOrder.hpp" @@ -75,6 +76,8 @@ class BuildHashOperator : public RelationalOperator { * @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 hash_table_index The index of the JoinHashTable in QueryContext. * The HashTable's key Type(s) should be the Type(s) of the * join_key_attributes in input_relation. @@ -84,6 +87,7 @@ class BuildHashOperator : public RelationalOperator { const bool input_relation_is_stored, const std::vector &join_key_attributes, const bool any_join_key_attributes_nullable, + const std::size_t num_partitions, const QueryContext::join_hash_table_id hash_table_index) : RelationalOperator(query_id), input_relation_(input_relation), @@ -91,10 +95,22 @@ class BuildHashOperator : public RelationalOperator { join_key_attributes_(join_key_attributes), any_join_key_attributes_nullable_(any_join_key_attributes_nullable), hash_table_index_(hash_table_index), - input_relation_block_ids_(input_relation_is_stored ? input_relation.getBlocksSnapshot() - : std::vector()), - num_workorders_generated_(0), - started_(false) {} + num_partitions_(num_partitions), + input_relation_block_ids_(num_partitions), + num_workorders_generated_(num_partitions), + started_(false) { + if (input_relation_is_stored) { + if (input_relation.hasPartitionScheme()) { + const PartitionScheme &part_scheme = *input_relation.getPartitionScheme(); + for (std::size_t part_id = 0; part_id < num_partitions_; ++part_id) { + input_relation_block_ids_[part_id] = part_scheme.getBlocksInPartition(part_id); + } + } else { + // No partition. + input_relation_block_ids_[0] = input_relation.getBlocksSnapshot(); + } + } + } ~BuildHashOperator() override {} @@ -116,7 +132,7 @@ class BuildHashOperator : public RelationalOperator { void feedInputBlock(const block_id input_block_id, const relation_id input_relation_id, const partition_id part_id) override { - input_relation_block_ids_.push_back(input_block_id); + input_relation_block_ids_[part_id].push_back(input_block_id); } private: @@ -124,8 +140,9 @@ class BuildHashOperator : public RelationalOperator { * @brief Create Work Order proto. * * @param block The block id used in the Work Order. + * @param part_id The partition id of 'block'. **/ - serialization::WorkOrder* createWorkOrderProto(const block_id block); + serialization::WorkOrder* createWorkOrderProto(const block_id block, const partition_id part_id); const CatalogRelation &input_relation_; const bool input_relation_is_stored_; @@ -133,8 +150,10 @@ class BuildHashOperator : public RelationalOperator { const bool any_join_key_attributes_nullable_; const QueryContext::join_hash_table_id hash_table_index_; - std::vector input_relation_block_ids_; - std::vector::size_type num_workorders_generated_; + const std::size_t num_partitions_; + // The index is the partition id. + std::vector input_relation_block_ids_; + std::vector num_workorders_generated_; bool started_; @@ -155,6 +174,7 @@ class BuildHashWorkOrder : public WorkOrder { * input_relation. * @param any_join_key_attributes_nullable If any attribute is nullable. * @param build_block_id The block id. + * @param part_id The partition id of 'build_block_id'. * @param hash_table The JoinHashTable to use. * @param storage_manager The StorageManager to use. * @param lip_filter_builder The attached LIP filter builer. @@ -164,6 +184,7 @@ class BuildHashWorkOrder : public WorkOrder { const std::vector &join_key_attributes, const bool any_join_key_attributes_nullable, const block_id build_block_id, + const partition_id part_id, JoinHashTable *hash_table, StorageManager *storage_manager, LIPFilterBuilder *lip_filter_builder) @@ -172,6 +193,7 @@ class BuildHashWorkOrder : public WorkOrder { join_key_attributes_(join_key_attributes), any_join_key_attributes_nullable_(any_join_key_attributes_nullable), build_block_id_(build_block_id), + part_id_(part_id), hash_table_(DCHECK_NOTNULL(hash_table)), storage_manager_(DCHECK_NOTNULL(storage_manager)), lip_filter_builder_(lip_filter_builder) {} @@ -185,6 +207,7 @@ class BuildHashWorkOrder : public WorkOrder { * input_relation. * @param any_join_key_attributes_nullable If any attribute is nullable. * @param build_block_id The block id. + * @param part_id The partition id of 'build_block_id'. * @param hash_table The JoinHashTable to use. * @param storage_manager The StorageManager to use. * @param lip_filter_builder The attached LIP filter builer. @@ -194,6 +217,7 @@ class BuildHashWorkOrder : public WorkOrder { std::vector &&join_key_attributes, const bool any_join_key_attributes_nullable, const block_id build_block_id, + const partition_id part_id, JoinHashTable *hash_table, StorageManager *storage_manager, LIPFilterBuilder *lip_filter_builder) @@ -202,6 +226,7 @@ class BuildHashWorkOrder : public WorkOrder { join_key_attributes_(std::move(join_key_attributes)), any_join_key_attributes_nullable_(any_join_key_attributes_nullable), build_block_id_(build_block_id), + part_id_(part_id), hash_table_(DCHECK_NOTNULL(hash_table)), storage_manager_(DCHECK_NOTNULL(storage_manager)), lip_filter_builder_(lip_filter_builder) {} @@ -214,11 +239,21 @@ class BuildHashWorkOrder : public WorkOrder { void execute() override; + /** + * @brief Get the partition id. + * + * @return The partition id. + */ + partition_id getPartitionId() const { + return part_id_; + } + private: const CatalogRelationSchema &input_relation_; const std::vector join_key_attributes_; const bool any_join_key_attributes_nullable_; const block_id build_block_id_; + const partition_id part_id_; JoinHashTable *hash_table_; StorageManager *storage_manager_; http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/8e452194/relational_operators/CMakeLists.txt ---------------------------------------------------------------------- diff --git a/relational_operators/CMakeLists.txt b/relational_operators/CMakeLists.txt index 526820e..66ea2d1 100644 --- a/relational_operators/CMakeLists.txt +++ b/relational_operators/CMakeLists.txt @@ -40,8 +40,8 @@ add_library(quickstep_relationaloperators_AggregationOperator AggregationOperato add_library(quickstep_relationaloperators_BuildHashOperator BuildHashOperator.cpp BuildHashOperator.hpp) add_library(quickstep_relationaloperators_CreateIndexOperator CreateIndexOperator.cpp CreateIndexOperator.hpp) add_library(quickstep_relationaloperators_CreateTableOperator CreateTableOperator.cpp CreateTableOperator.hpp) -add_library(quickstep_relationaloperators_DestroyAggregationStateOperator - DestroyAggregationStateOperator.cpp +add_library(quickstep_relationaloperators_DestroyAggregationStateOperator + DestroyAggregationStateOperator.cpp DestroyAggregationStateOperator.hpp) add_library(quickstep_relationaloperators_DeleteOperator DeleteOperator.cpp DeleteOperator.hpp) add_library(quickstep_relationaloperators_DestroyHashOperator DestroyHashOperator.cpp DestroyHashOperator.hpp) @@ -99,6 +99,7 @@ target_link_libraries(quickstep_relationaloperators_BuildHashOperator glog quickstep_catalog_CatalogRelation quickstep_catalog_CatalogTypedefs + quickstep_catalog_PartitionScheme quickstep_queryexecution_QueryContext quickstep_queryexecution_WorkOrderProtosContainer quickstep_queryexecution_WorkOrdersContainer @@ -162,6 +163,7 @@ target_link_libraries(quickstep_relationaloperators_DestroyAggregationStateOpera tmb) target_link_libraries(quickstep_relationaloperators_DestroyHashOperator glog + quickstep_catalog_CatalogTypedefs quickstep_queryexecution_QueryContext quickstep_queryexecution_WorkOrderProtosContainer quickstep_queryexecution_WorkOrdersContainer @@ -204,6 +206,7 @@ target_link_libraries(quickstep_relationaloperators_HashJoinOperator quickstep_catalog_CatalogRelation quickstep_catalog_CatalogRelationSchema quickstep_catalog_CatalogTypedefs + quickstep_catalog_PartitionScheme quickstep_expressions_predicate_Predicate quickstep_expressions_scalar_Scalar quickstep_queryexecution_QueryContext @@ -796,4 +799,4 @@ add_test(TextScanOperator_faulty_unittest TextScanOperator_unittest ${TEXT_SCAN_FAULTY_INPUT_FILE} ${TEXT_SCAN_FAULTY_GOLDEN_OUTPUT_FILE} - ${TEXT_SCAN_FAULTY_FAILURE_OUTPUT_FILE}) \ No newline at end of file + ${TEXT_SCAN_FAULTY_FAILURE_OUTPUT_FILE}) http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/8e452194/relational_operators/DestroyHashOperator.cpp ---------------------------------------------------------------------- diff --git a/relational_operators/DestroyHashOperator.cpp b/relational_operators/DestroyHashOperator.cpp index 96d2561..4827ef5 100644 --- a/relational_operators/DestroyHashOperator.cpp +++ b/relational_operators/DestroyHashOperator.cpp @@ -35,31 +35,36 @@ 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) { + container->addNormalWorkOrder( + new DestroyHashWorkOrder(query_id_, hash_table_index_, part_id, query_context), + op_index_); + } work_generated_ = true; - container->addNormalWorkOrder( - new DestroyHashWorkOrder(query_id_, hash_table_index_, query_context), - op_index_); } return work_generated_; } bool DestroyHashOperator::getAllWorkOrderProtos(WorkOrderProtosContainer *container) { if (blocking_dependencies_met_ && !work_generated_) { - work_generated_ = true; + for (std::size_t part_id = 0; part_id < num_partitions_; ++part_id) { + serialization::WorkOrder *proto = new serialization::WorkOrder; + proto->set_work_order_type(serialization::DESTROY_HASH); + proto->set_query_id(query_id_); + proto->SetExtension(serialization::DestroyHashWorkOrder::join_hash_table_index, hash_table_index_); + proto->SetExtension(serialization::DestroyHashWorkOrder::partition_id, part_id); - serialization::WorkOrder *proto = new serialization::WorkOrder; - proto->set_work_order_type(serialization::DESTROY_HASH); - proto->set_query_id(query_id_); - proto->SetExtension(serialization::DestroyHashWorkOrder::join_hash_table_index, hash_table_index_); + container->addWorkOrderProto(proto, op_index_); + } - container->addWorkOrderProto(proto, op_index_); + work_generated_ = true; } return work_generated_; } void DestroyHashWorkOrder::execute() { - query_context_->destroyJoinHashTable(hash_table_index_); + query_context_->destroyJoinHashTable(hash_table_index_, part_id_); } } // namespace quickstep http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/8e452194/relational_operators/DestroyHashOperator.hpp ---------------------------------------------------------------------- diff --git a/relational_operators/DestroyHashOperator.hpp b/relational_operators/DestroyHashOperator.hpp index d402766..8a1fabd 100644 --- a/relational_operators/DestroyHashOperator.hpp +++ b/relational_operators/DestroyHashOperator.hpp @@ -22,6 +22,7 @@ #include +#include "catalog/CatalogTypedefs.hpp" #include "query_execution/QueryContext.hpp" #include "relational_operators/RelationalOperator.hpp" #include "relational_operators/WorkOrder.hpp" @@ -52,11 +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 hash_table_index The index of the JoinHashTable in QueryContext. **/ DestroyHashOperator(const std::size_t query_id, + const std::size_t num_partitions, const QueryContext::join_hash_table_id hash_table_index) : RelationalOperator(query_id), + num_partitions_(num_partitions), hash_table_index_(hash_table_index), work_generated_(false) {} @@ -75,6 +79,7 @@ class DestroyHashOperator : public RelationalOperator { bool getAllWorkOrderProtos(WorkOrderProtosContainer *container) override; private: + const std::size_t num_partitions_; const QueryContext::join_hash_table_id hash_table_index_; bool work_generated_; @@ -91,13 +96,16 @@ class DestroyHashWorkOrder : public WorkOrder { * * @param query_id The ID of the query to which this WorkOrder belongs. * @param hash_table_index The index of the JoinHashTable in QueryContext. + * @param part_id The partition id. * @param query_context The QueryContext to use. **/ DestroyHashWorkOrder(const std::size_t query_id, const QueryContext::join_hash_table_id hash_table_index, + const partition_id part_id, QueryContext *query_context) : WorkOrder(query_id), hash_table_index_(hash_table_index), + part_id_(part_id), query_context_(DCHECK_NOTNULL(query_context)) {} ~DestroyHashWorkOrder() override {} @@ -106,6 +114,7 @@ class DestroyHashWorkOrder : public WorkOrder { private: const QueryContext::join_hash_table_id hash_table_index_; + const partition_id part_id_; QueryContext *query_context_; DISALLOW_COPY_AND_ASSIGN(DestroyHashWorkOrder); http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/8e452194/relational_operators/HashJoinOperator.cpp ---------------------------------------------------------------------- diff --git a/relational_operators/HashJoinOperator.cpp b/relational_operators/HashJoinOperator.cpp index 2028046..d2eb560 100644 --- a/relational_operators/HashJoinOperator.cpp +++ b/relational_operators/HashJoinOperator.cpp @@ -207,49 +207,43 @@ bool HashJoinOperator::getAllNonOuterJoinWorkOrders( query_context->getScalarGroup(selection_index_); InsertDestination *output_destination = query_context->getInsertDestination(output_destination_index_); - const JoinHashTable &hash_table = - *(query_context->getJoinHashTable(hash_table_index_)); if (probe_relation_is_stored_) { - if (!started_) { - for (const block_id probe_block_id : probe_relation_block_ids_) { + if (started_) { + return true; + } + + for (std::size_t part_id = 0; part_id < 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_, - probe_block_id, - residual_predicate, - selection, - hash_table, - output_destination, - storage_manager, + new JoinWorkOrderClass(query_id_, build_relation_, probe_relation_, join_key_attributes_, + any_join_key_attributes_nullable_, probe_block_id, part_id, residual_predicate, + selection, hash_table, output_destination, storage_manager, CreateLIPFilterAdaptiveProberHelper(lip_deployment_index_, query_context)), op_index_); } - started_ = true; } - return started_; + started_ = true; + return true; } else { - while (num_workorders_generated_ < probe_relation_block_ids_.size()) { - container->addNormalWorkOrder( - new JoinWorkOrderClass( - query_id_, - build_relation_, - probe_relation_, - join_key_attributes_, - any_join_key_attributes_nullable_, - probe_relation_block_ids_[num_workorders_generated_], - residual_predicate, - selection, - hash_table, - output_destination, - storage_manager, - CreateLIPFilterAdaptiveProberHelper(lip_deployment_index_, query_context)), - op_index_); - ++num_workorders_generated_; - } // end while + for (std::size_t part_id = 0; part_id < 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_, + probe_relation_block_ids_[part_id][num_workorders_generated_[part_id]], part_id, + residual_predicate, selection, hash_table, output_destination, storage_manager, + CreateLIPFilterAdaptiveProberHelper(lip_deployment_index_, query_context)), + op_index_); + ++num_workorders_generated_[part_id]; + } // end while + } // end for return done_feeding_input_relation_; } // end else (probe_relation_is_stored_) } // end if (blocking_dependencies_met_) @@ -269,49 +263,43 @@ bool HashJoinOperator::getAllOuterJoinWorkOrders( InsertDestination *output_destination = query_context->getInsertDestination(output_destination_index_); - const JoinHashTable &hash_table = - *(query_context->getJoinHashTable(hash_table_index_)); if (probe_relation_is_stored_) { - if (!started_) { - for (const block_id probe_block_id : probe_relation_block_ids_) { + if (started_) { + return true; + } + + for (std::size_t part_id = 0; part_id < 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_, - probe_block_id, - selection, - is_selection_on_build_, - hash_table, - output_destination, - storage_manager, - CreateLIPFilterAdaptiveProberHelper(lip_deployment_index_, query_context)), + new HashOuterJoinWorkOrder(query_id_, build_relation_, probe_relation_, join_key_attributes_, + any_join_key_attributes_nullable_, probe_block_id, part_id, selection, + is_selection_on_build_, hash_table, output_destination, storage_manager, + CreateLIPFilterAdaptiveProberHelper(lip_deployment_index_, query_context)), op_index_); } - started_ = true; } - return started_; + started_ = true; + return true; } else { - while (num_workorders_generated_ < probe_relation_block_ids_.size()) { - container->addNormalWorkOrder( - new HashOuterJoinWorkOrder( - query_id_, - build_relation_, - probe_relation_, - join_key_attributes_, - any_join_key_attributes_nullable_, - probe_relation_block_ids_[num_workorders_generated_], - selection, - is_selection_on_build_, - hash_table, - output_destination, - storage_manager, - CreateLIPFilterAdaptiveProberHelper(lip_deployment_index_, query_context)), - op_index_); - ++num_workorders_generated_; + for (std::size_t part_id = 0; part_id < 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_, + probe_relation_block_ids_[part_id][num_workorders_generated_[part_id]], + part_id, selection, is_selection_on_build_, hash_table, output_destination, + storage_manager, + CreateLIPFilterAdaptiveProberHelper(lip_deployment_index_, query_context)), + op_index_); + ++num_workorders_generated_[part_id]; + } } return done_feeding_input_relation_; } // end else (probe_relation_is_stored_) @@ -343,22 +331,29 @@ bool HashJoinOperator::getAllNonOuterJoinWorkOrderProtos( } if (probe_relation_is_stored_) { - if (!started_) { - for (const block_id probe_block_id : probe_relation_block_ids_) { + if (started_) { + return true; + } + + for (std::size_t part_id = 0; part_id < 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), + createNonOuterJoinWorkOrderProto(hash_join_type, probe_block_id, part_id), op_index_); } - started_ = true; } + started_ = true; return true; } else { - while (num_workorders_generated_ < probe_relation_block_ids_.size()) { - container->addWorkOrderProto( - createNonOuterJoinWorkOrderProto(hash_join_type, - probe_relation_block_ids_[num_workorders_generated_]), - op_index_); - ++num_workorders_generated_; + for (std::size_t part_id = 0; part_id < num_partitions_; ++part_id) { + while (num_workorders_generated_[part_id] < probe_relation_block_ids_[part_id].size()) { + container->addWorkOrderProto( + createNonOuterJoinWorkOrderProto(hash_join_type, + probe_relation_block_ids_[part_id][num_workorders_generated_[part_id]], + part_id), + op_index_); + ++num_workorders_generated_[part_id]; + } } return done_feeding_input_relation_; @@ -367,7 +362,7 @@ bool HashJoinOperator::getAllNonOuterJoinWorkOrderProtos( serialization::WorkOrder* HashJoinOperator::createNonOuterJoinWorkOrderProto( const serialization::HashJoinWorkOrder::HashJoinWorkOrderType hash_join_type, - const block_id block) { + const block_id block, const partition_id part_id) { serialization::WorkOrder *proto = new serialization::WorkOrder; proto->set_work_order_type(serialization::HASH_JOIN); proto->set_query_id(query_id_); @@ -382,6 +377,7 @@ serialization::WorkOrder* HashJoinOperator::createNonOuterJoinWorkOrderProto( any_join_key_attributes_nullable_); 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); proto->SetExtension(serialization::HashJoinWorkOrder::selection_index, selection_index_); proto->SetExtension(serialization::HashJoinWorkOrder::block_id, block); proto->SetExtension(serialization::HashJoinWorkOrder::residual_predicate_index, residual_predicate_index_); @@ -397,26 +393,34 @@ bool HashJoinOperator::getAllOuterJoinWorkOrderProtos(WorkOrderProtosContainer * } if (probe_relation_is_stored_) { - if (!started_) { - for (const block_id probe_block_id : probe_relation_block_ids_) { - container->addWorkOrderProto(createOuterJoinWorkOrderProto(probe_block_id), op_index_); + if (started_) { + return true; + } + + for (std::size_t part_id = 0; part_id < 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_); } - started_ = true; } + started_ = true; return true; } else { - while (num_workorders_generated_ < probe_relation_block_ids_.size()) { - container->addWorkOrderProto( - createOuterJoinWorkOrderProto(probe_relation_block_ids_[num_workorders_generated_]), - op_index_); - ++num_workorders_generated_; + for (std::size_t part_id = 0; part_id < 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]], + part_id), + op_index_); + ++num_workorders_generated_[part_id]; + } } return done_feeding_input_relation_; } } -serialization::WorkOrder* HashJoinOperator::createOuterJoinWorkOrderProto(const block_id block) { +serialization::WorkOrder* HashJoinOperator::createOuterJoinWorkOrderProto(const block_id block, + const partition_id part_id) { serialization::WorkOrder *proto = new serialization::WorkOrder; proto->set_work_order_type(serialization::HASH_JOIN); proto->set_query_id(query_id_); @@ -434,6 +438,7 @@ serialization::WorkOrder* HashJoinOperator::createOuterJoinWorkOrderProto(const proto->SetExtension(serialization::HashJoinWorkOrder::join_hash_table_index, hash_table_index_); proto->SetExtension(serialization::HashJoinWorkOrder::selection_index, selection_index_); proto->SetExtension(serialization::HashJoinWorkOrder::block_id, block); + proto->SetExtension(serialization::HashJoinWorkOrder::partition_id, part_id); proto->SetExtension(serialization::HashJoinWorkOrder::lip_deployment_index, lip_deployment_index_); for (const bool is_attribute_on_build : is_selection_on_build_) { http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/8e452194/relational_operators/HashJoinOperator.hpp ---------------------------------------------------------------------- diff --git a/relational_operators/HashJoinOperator.hpp b/relational_operators/HashJoinOperator.hpp index 98c87bf..a9cf83e 100644 --- a/relational_operators/HashJoinOperator.hpp +++ b/relational_operators/HashJoinOperator.hpp @@ -28,6 +28,7 @@ #include "catalog/CatalogRelation.hpp" #include "catalog/CatalogTypedefs.hpp" +#include "catalog/PartitionScheme.hpp" #include "query_execution/QueryContext.hpp" #include "relational_operators/RelationalOperator.hpp" #include "relational_operators/WorkOrder.hpp" @@ -100,6 +101,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 output_relation The output relation. * @param output_destination_index The index of the InsertDestination in the * QueryContext to insert the join results. @@ -125,6 +128,7 @@ class HashJoinOperator : public RelationalOperator { const bool probe_relation_is_stored, const std::vector &join_key_attributes, const bool any_join_key_attributes_nullable, + const std::size_t num_partitions, const CatalogRelation &output_relation, const QueryContext::insert_destination_id output_destination_index, const QueryContext::join_hash_table_id hash_table_index, @@ -147,14 +151,25 @@ class HashJoinOperator : public RelationalOperator { ? std::vector() : *is_selection_on_build), join_type_(join_type), - probe_relation_block_ids_(probe_relation_is_stored - ? probe_relation.getBlocksSnapshot() - : std::vector()), - num_workorders_generated_(0), + num_partitions_(num_partitions), + probe_relation_block_ids_(num_partitions), + num_workorders_generated_(num_partitions), started_(false) { DCHECK(join_type != JoinType::kLeftOuterJoin || (is_selection_on_build != nullptr && residual_predicate_index == QueryContext::kInvalidPredicateId)); + + 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) { + probe_relation_block_ids_[part_id] = part_scheme.getBlocksInPartition(part_id); + } + } else { + // No partition. + probe_relation_block_ids_[0] = probe_relation.getBlocksSnapshot(); + } + } } ~HashJoinOperator() override {} @@ -192,8 +207,9 @@ class HashJoinOperator : public RelationalOperator { void feedInputBlock(const block_id input_block_id, const relation_id input_relation_id, const partition_id part_id) override { - DCHECK(input_relation_id == probe_relation_.getID()); - probe_relation_block_ids_.push_back(input_block_id); + DCHECK_EQ(probe_relation_.getID(), input_relation_id); + + probe_relation_block_ids_[part_id].push_back(input_block_id); } QueryContext::insert_destination_id getInsertDestinationID() const override { @@ -229,7 +245,7 @@ class HashJoinOperator : public RelationalOperator { serialization::WorkOrder* createNonOuterJoinWorkOrderProto( const serialization::HashJoinWorkOrder::HashJoinWorkOrderType hash_join_type, - const block_id block); + const block_id block, const partition_id part_id); bool getAllOuterJoinWorkOrderProtos(WorkOrderProtosContainer *container); @@ -238,7 +254,7 @@ class HashJoinOperator : public RelationalOperator { * * @param block The block id used in the Work Order. **/ - serialization::WorkOrder* createOuterJoinWorkOrderProto(const block_id block); + serialization::WorkOrder* createOuterJoinWorkOrderProto(const block_id block, const partition_id part_id); const CatalogRelation &build_relation_; const CatalogRelation &probe_relation_; @@ -253,8 +269,10 @@ class HashJoinOperator : public RelationalOperator { const std::vector is_selection_on_build_; const JoinType join_type_; - std::vector probe_relation_block_ids_; - std::size_t num_workorders_generated_; + const std::size_t num_partitions_; + // The index is the partition id. + std::vector probe_relation_block_ids_; + std::vector num_workorders_generated_; bool started_; @@ -278,6 +296,7 @@ class HashInnerJoinWorkOrder : public WorkOrder { * probe_relation. * @param any_join_key_attributes_nullable If any attribute is nullable. * @param lookup_block_id The block id of the probe_relation. + * @param part_id The partition id of the 'lookup_block_id'. * @param residual_predicate If non-null, apply as an additional filter to * pairs of tuples that match the hash-join (i.e. key equality) * predicate. Effectively, this makes the join predicate the @@ -297,6 +316,7 @@ class HashInnerJoinWorkOrder : public WorkOrder { const std::vector &join_key_attributes, const bool any_join_key_attributes_nullable, const block_id lookup_block_id, + const partition_id part_id, const Predicate *residual_predicate, const std::vector> &selection, const JoinHashTable &hash_table, @@ -309,6 +329,7 @@ class HashInnerJoinWorkOrder : public WorkOrder { join_key_attributes_(join_key_attributes), any_join_key_attributes_nullable_(any_join_key_attributes_nullable), block_id_(lookup_block_id), + part_id_(part_id), residual_predicate_(residual_predicate), selection_(selection), hash_table_(hash_table), @@ -328,6 +349,7 @@ class HashInnerJoinWorkOrder : public WorkOrder { * probe_relation. * @param any_join_key_attributes_nullable If any attribute is nullable. * @param lookup_block_id The block id of the probe_relation. + * @param part_id The partition id of the 'lookup_block_id'. * @param residual_predicate If non-null, apply as an additional filter to * pairs of tuples that match the hash-join (i.e. key equality) * predicate. Effectively, this makes the join predicate the @@ -347,6 +369,7 @@ class HashInnerJoinWorkOrder : public WorkOrder { std::vector &&join_key_attributes, const bool any_join_key_attributes_nullable, const block_id lookup_block_id, + const partition_id part_id, const Predicate *residual_predicate, const std::vector> &selection, const JoinHashTable &hash_table, @@ -359,6 +382,7 @@ class HashInnerJoinWorkOrder : public WorkOrder { join_key_attributes_(std::move(join_key_attributes)), any_join_key_attributes_nullable_(any_join_key_attributes_nullable), block_id_(lookup_block_id), + part_id_(part_id), residual_predicate_(residual_predicate), selection_(selection), hash_table_(hash_table), @@ -378,12 +402,22 @@ class HashInnerJoinWorkOrder : public WorkOrder { **/ void execute() override; + /** + * @brief Get the partition id. + * + * @return The partition id. + */ + partition_id getPartitionId() const { + return part_id_; + } + private: const CatalogRelationSchema &build_relation_; const CatalogRelationSchema &probe_relation_; const std::vector join_key_attributes_; const bool any_join_key_attributes_nullable_; const block_id block_id_; + const partition_id part_id_; const Predicate *residual_predicate_; const std::vector> &selection_; const JoinHashTable &hash_table_; @@ -414,6 +448,7 @@ class HashSemiJoinWorkOrder : public WorkOrder { * probe_relation. * @param any_join_key_attributes_nullable If any attribute is nullable. * @param lookup_block_id The block id of the probe_relation. + * @param part_id The partition id of the 'lookup_block_id'. * @param residual_predicate If non-null, apply as an additional filter to * pairs of tuples that match the hash-join (i.e. key equality) * predicate. Effectively, this makes the join predicate the @@ -433,6 +468,7 @@ class HashSemiJoinWorkOrder : public WorkOrder { const std::vector &join_key_attributes, const bool any_join_key_attributes_nullable, const block_id lookup_block_id, + const partition_id part_id, const Predicate *residual_predicate, const std::vector> &selection, const JoinHashTable &hash_table, @@ -445,6 +481,7 @@ class HashSemiJoinWorkOrder : public WorkOrder { join_key_attributes_(join_key_attributes), any_join_key_attributes_nullable_(any_join_key_attributes_nullable), block_id_(lookup_block_id), + part_id_(part_id), residual_predicate_(residual_predicate), selection_(selection), hash_table_(hash_table), @@ -464,6 +501,7 @@ class HashSemiJoinWorkOrder : public WorkOrder { * probe_relation. * @param any_join_key_attributes_nullable If any attribute is nullable. * @param lookup_block_id The block id of the probe_relation. + * @param part_id The partition id of the 'lookup_block_id'. * @param residual_predicate If non-null, apply as an additional filter to * pairs of tuples that match the hash-join (i.e. key equality) * predicate. Effectively, this makes the join predicate the @@ -483,6 +521,7 @@ class HashSemiJoinWorkOrder : public WorkOrder { std::vector &&join_key_attributes, const bool any_join_key_attributes_nullable, const block_id lookup_block_id, + const partition_id part_id, const Predicate *residual_predicate, const std::vector> &selection, const JoinHashTable &hash_table, @@ -495,6 +534,7 @@ class HashSemiJoinWorkOrder : public WorkOrder { join_key_attributes_(std::move(join_key_attributes)), any_join_key_attributes_nullable_(any_join_key_attributes_nullable), block_id_(lookup_block_id), + part_id_(part_id), residual_predicate_(residual_predicate), selection_(selection), hash_table_(hash_table), @@ -506,6 +546,15 @@ class HashSemiJoinWorkOrder : public WorkOrder { void execute() override; + /** + * @brief Get the partition id. + * + * @return The partition id. + */ + partition_id getPartitionId() const { + return part_id_; + } + private: void executeWithoutResidualPredicate(); @@ -516,6 +565,7 @@ class HashSemiJoinWorkOrder : public WorkOrder { const std::vector join_key_attributes_; const bool any_join_key_attributes_nullable_; const block_id block_id_; + const partition_id part_id_; const Predicate *residual_predicate_; const std::vector> &selection_; const JoinHashTable &hash_table_; @@ -546,6 +596,7 @@ class HashAntiJoinWorkOrder : public WorkOrder { * probe_relation. * @param any_join_key_attributes_nullable If any attribute is nullable. * @param lookup_block_id The block id of the probe_relation. + * @param part_id The partition id of the 'lookup_block_id'. * @param residual_predicate If non-null, apply as an additional filter to * pairs of tuples that match the hash-join (i.e. key equality) * predicate. Effectively, this makes the join predicate the @@ -565,6 +616,7 @@ class HashAntiJoinWorkOrder : public WorkOrder { const std::vector &join_key_attributes, const bool any_join_key_attributes_nullable, const block_id lookup_block_id, + const partition_id part_id, const Predicate *residual_predicate, const std::vector> &selection, const JoinHashTable &hash_table, @@ -577,6 +629,7 @@ class HashAntiJoinWorkOrder : public WorkOrder { join_key_attributes_(join_key_attributes), any_join_key_attributes_nullable_(any_join_key_attributes_nullable), block_id_(lookup_block_id), + part_id_(part_id), residual_predicate_(residual_predicate), selection_(selection), hash_table_(hash_table), @@ -596,6 +649,7 @@ class HashAntiJoinWorkOrder : public WorkOrder { * probe_relation. * @param any_join_key_attributes_nullable If any attribute is nullable. * @param lookup_block_id The block id of the probe_relation. + * @param part_id The partition id of the 'lookup_block_id'. * @param residual_predicate If non-null, apply as an additional filter to * pairs of tuples that match the hash-join (i.e. key equality) * predicate. Effectively, this makes the join predicate the @@ -615,6 +669,7 @@ class HashAntiJoinWorkOrder : public WorkOrder { std::vector &&join_key_attributes, const bool any_join_key_attributes_nullable, const block_id lookup_block_id, + const partition_id part_id, const Predicate *residual_predicate, const std::vector> &selection, const JoinHashTable &hash_table, @@ -627,6 +682,7 @@ class HashAntiJoinWorkOrder : public WorkOrder { join_key_attributes_(std::move(join_key_attributes)), any_join_key_attributes_nullable_(any_join_key_attributes_nullable), block_id_(lookup_block_id), + part_id_(part_id), residual_predicate_(residual_predicate), selection_(selection), hash_table_(hash_table), @@ -644,6 +700,15 @@ class HashAntiJoinWorkOrder : public WorkOrder { } } + /** + * @brief Get the partition id. + * + * @return The partition id. + */ + partition_id getPartitionId() const { + return part_id_; + } + private: void executeWithoutResidualPredicate(); @@ -654,6 +719,7 @@ class HashAntiJoinWorkOrder : public WorkOrder { const std::vector join_key_attributes_; const bool any_join_key_attributes_nullable_; const block_id block_id_; + const partition_id part_id_; const Predicate *residual_predicate_; const std::vector> &selection_; const JoinHashTable &hash_table_; @@ -683,6 +749,7 @@ class HashOuterJoinWorkOrder : public WorkOrder { * probe_relation. * @param any_join_key_attributes_nullable If any attribute is nullable. * @param lookup_block_id The block id of the probe_relation. + * @param part_id The partition id of the 'lookup_block_id'. * @param selection A list of Scalars corresponding to the relation attributes * in \c output_destination. Each Scalar is evaluated for the joined * tuples, and the resulting value is inserted into the join result. @@ -701,6 +768,7 @@ class HashOuterJoinWorkOrder : public WorkOrder { const std::vector &join_key_attributes, const bool any_join_key_attributes_nullable, const block_id lookup_block_id, + const partition_id part_id, const std::vector> &selection, const std::vector &is_selection_on_build, const JoinHashTable &hash_table, @@ -713,6 +781,7 @@ class HashOuterJoinWorkOrder : public WorkOrder { join_key_attributes_(join_key_attributes), any_join_key_attributes_nullable_(any_join_key_attributes_nullable), block_id_(lookup_block_id), + part_id_(part_id), selection_(selection), is_selection_on_build_(is_selection_on_build), hash_table_(hash_table), @@ -732,6 +801,7 @@ class HashOuterJoinWorkOrder : public WorkOrder { * probe_relation. * @param any_join_key_attributes_nullable If any attribute is nullable. * @param lookup_block_id The block id of the probe_relation. + * @param part_id The partition id of the 'lookup_block_id'. * @param selection A list of Scalars corresponding to the relation attributes * in \c output_destination. Each Scalar is evaluated for the joined * tuples, and the resulting value is inserted into the join result. @@ -749,6 +819,7 @@ class HashOuterJoinWorkOrder : public WorkOrder { std::vector &&join_key_attributes, const bool any_join_key_attributes_nullable, const block_id lookup_block_id, + const partition_id part_id, const std::vector> &selection, std::vector &&is_selection_on_build, const JoinHashTable &hash_table, @@ -761,6 +832,7 @@ class HashOuterJoinWorkOrder : public WorkOrder { join_key_attributes_(std::move(join_key_attributes)), any_join_key_attributes_nullable_(any_join_key_attributes_nullable), block_id_(lookup_block_id), + part_id_(part_id), selection_(selection), is_selection_on_build_(std::move(is_selection_on_build)), hash_table_(hash_table), @@ -772,12 +844,22 @@ class HashOuterJoinWorkOrder : public WorkOrder { void execute() override; + /** + * @brief Get the partition id. + * + * @return The partition id. + */ + partition_id getPartitionId() const { + return part_id_; + } + private: const CatalogRelationSchema &build_relation_; const CatalogRelationSchema &probe_relation_; const std::vector join_key_attributes_; const bool any_join_key_attributes_nullable_; const block_id block_id_; + const partition_id part_id_; const std::vector> &selection_; const std::vector is_selection_on_build_; const JoinHashTable &hash_table_; http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/8e452194/relational_operators/WorkOrder.proto ---------------------------------------------------------------------- diff --git a/relational_operators/WorkOrder.proto b/relational_operators/WorkOrder.proto index 86f34b8..f8d9246 100644 --- a/relational_operators/WorkOrder.proto +++ b/relational_operators/WorkOrder.proto @@ -63,6 +63,7 @@ message AggregationWorkOrder { } } +// Next tag: 39. message BuildHashWorkOrder { extend WorkOrder { // All required. @@ -70,6 +71,7 @@ message BuildHashWorkOrder { repeated int32 join_key_attributes = 33; optional bool any_join_key_attributes_nullable = 34; optional uint32 join_hash_table_index = 35; + optional uint64 partition_id = 38; optional fixed64 block_id = 36; optional int32 lip_deployment_index = 37; } @@ -89,6 +91,7 @@ message DestroyHashWorkOrder { extend WorkOrder { // All required. optional uint32 join_hash_table_index = 112; + optional uint64 partition_id = 113; } } @@ -109,6 +112,7 @@ message FinalizeAggregationWorkOrder { } } +// Next tag: 173. message HashJoinWorkOrder { enum HashJoinWorkOrderType { HASH_ANTI_JOIN = 0; @@ -126,6 +130,7 @@ message HashJoinWorkOrder { optional bool any_join_key_attributes_nullable = 164; optional int32 insert_destination_index = 165; optional uint32 join_hash_table_index = 166; + 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/8e452194/relational_operators/WorkOrderFactory.cpp ---------------------------------------------------------------------- diff --git a/relational_operators/WorkOrderFactory.cpp b/relational_operators/WorkOrderFactory.cpp index 871f74d..59f86d2 100644 --- a/relational_operators/WorkOrderFactory.cpp +++ b/relational_operators/WorkOrderFactory.cpp @@ -98,6 +98,9 @@ WorkOrder* WorkOrderFactory::ReconstructFromProto(const serialization::WorkOrder proto.GetExtension(serialization::BuildHashWorkOrder::join_key_attributes, i)); } + const partition_id part_id = + proto.GetExtension(serialization::BuildHashWorkOrder::partition_id); + return new BuildHashWorkOrder( proto.query_id(), catalog_database->getRelationSchemaById( @@ -105,8 +108,9 @@ WorkOrder* WorkOrderFactory::ReconstructFromProto(const serialization::WorkOrder move(join_key_attributes), proto.GetExtension(serialization::BuildHashWorkOrder::any_join_key_attributes_nullable), proto.GetExtension(serialization::BuildHashWorkOrder::block_id), + part_id, query_context->getJoinHashTable( - proto.GetExtension(serialization::BuildHashWorkOrder::join_hash_table_index)), + proto.GetExtension(serialization::BuildHashWorkOrder::join_hash_table_index), part_id), storage_manager, CreateLIPFilterBuilderHelper( proto.GetExtension(serialization::BuildHashWorkOrder::lip_deployment_index), query_context)); @@ -139,6 +143,8 @@ WorkOrder* WorkOrderFactory::ReconstructFromProto(const serialization::WorkOrder proto.query_id(), proto.GetExtension( serialization::DestroyHashWorkOrder::join_hash_table_index), + proto.GetExtension( + serialization::DestroyHashWorkOrder::partition_id), query_context); } case serialization::DROP_TABLE: { @@ -202,9 +208,11 @@ WorkOrder* WorkOrderFactory::ReconstructFromProto(const serialization::WorkOrder const std::vector> &selection = query_context->getScalarGroup( proto.GetExtension(serialization::HashJoinWorkOrder::selection_index)); + const partition_id part_id = + proto.GetExtension(serialization::HashJoinWorkOrder::partition_id); const JoinHashTable &hash_table = *query_context->getJoinHashTable( - proto.GetExtension(serialization::HashJoinWorkOrder::join_hash_table_index)); + proto.GetExtension(serialization::HashJoinWorkOrder::join_hash_table_index), part_id); InsertDestination *output_destination = query_context->getInsertDestination( proto.GetExtension(serialization::HashJoinWorkOrder::insert_destination_index)); @@ -222,6 +230,7 @@ WorkOrder* WorkOrderFactory::ReconstructFromProto(const serialization::WorkOrder move(join_key_attributes), any_join_key_attributes_nullable, lookup_block_id, + part_id, residual_predicate, selection, hash_table, @@ -238,6 +247,7 @@ WorkOrder* WorkOrderFactory::ReconstructFromProto(const serialization::WorkOrder move(join_key_attributes), any_join_key_attributes_nullable, lookup_block_id, + part_id, residual_predicate, selection, hash_table, @@ -262,6 +272,7 @@ WorkOrder* WorkOrderFactory::ReconstructFromProto(const serialization::WorkOrder move(join_key_attributes), any_join_key_attributes_nullable, lookup_block_id, + part_id, selection, move(is_selection_on_build), hash_table, @@ -278,6 +289,7 @@ WorkOrder* WorkOrderFactory::ReconstructFromProto(const serialization::WorkOrder move(join_key_attributes), any_join_key_attributes_nullable, lookup_block_id, + part_id, residual_predicate, selection, hash_table, @@ -524,8 +536,10 @@ 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::partition_id) && query_context.isValidJoinHashTableId( - proto.GetExtension(serialization::BuildHashWorkOrder::join_hash_table_index)); + proto.GetExtension(serialization::BuildHashWorkOrder::join_hash_table_index), + proto.GetExtension(serialization::BuildHashWorkOrder::partition_id)); } case serialization::DELETE: { return proto.HasExtension(serialization::DeleteWorkOrder::relation_id) && @@ -544,8 +558,10 @@ bool WorkOrderFactory::ProtoIsValid(const serialization::WorkOrder &proto, } case serialization::DESTROY_HASH: { return proto.HasExtension(serialization::DestroyHashWorkOrder::join_hash_table_index) && + proto.HasExtension(serialization::DestroyHashWorkOrder::partition_id) && query_context.isValidJoinHashTableId( - proto.GetExtension(serialization::DestroyHashWorkOrder::join_hash_table_index)); + proto.GetExtension(serialization::DestroyHashWorkOrder::join_hash_table_index), + proto.GetExtension(serialization::DestroyHashWorkOrder::partition_id)); } case serialization::DROP_TABLE: { return true; @@ -623,8 +639,10 @@ 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::partition_id) && query_context.isValidJoinHashTableId( - proto.GetExtension(serialization::HashJoinWorkOrder::join_hash_table_index)) && + proto.GetExtension(serialization::HashJoinWorkOrder::join_hash_table_index), + proto.GetExtension(serialization::HashJoinWorkOrder::partition_id)) && proto.HasExtension(serialization::HashJoinWorkOrder::selection_index) && query_context.isValidScalarGroupId( proto.GetExtension(serialization::HashJoinWorkOrder::selection_index)) && http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/8e452194/relational_operators/tests/HashJoinOperator_unittest.cpp ---------------------------------------------------------------------- diff --git a/relational_operators/tests/HashJoinOperator_unittest.cpp b/relational_operators/tests/HashJoinOperator_unittest.cpp index 239547f..2370c92 100644 --- a/relational_operators/tests/HashJoinOperator_unittest.cpp +++ b/relational_operators/tests/HashJoinOperator_unittest.cpp @@ -97,6 +97,8 @@ constexpr tuple_id kBlockSize = 10; constexpr std::size_t kQueryId = 0; constexpr int kOpIndex = 0; +constexpr std::size_t kNumPartitions = 1; + } // namespace class HashJoinOperatorTest : public ::testing::TestWithParam { @@ -192,6 +194,8 @@ class HashJoinOperatorTest : public ::testing::TestWithParam } storage_block->rebuild(); } + + num_partitions_ = kNumPartitions; } virtual void TearDown() { @@ -291,6 +295,8 @@ class HashJoinOperatorTest : public ::testing::TestWithParam unique_ptr db_; // The following CatalogRelations are owned by db_. CatalogRelation *dim_table_, *fact_table_; + + std::size_t num_partitions_; }; TEST_P(HashJoinOperatorTest, LongKeyHashJoinTest) { @@ -302,7 +308,7 @@ TEST_P(HashJoinOperatorTest, LongKeyHashJoinTest) { query_context_proto.join_hash_tables_size(); serialization::HashTable *hash_table_proto = - query_context_proto.add_join_hash_tables(); + query_context_proto.add_join_hash_tables()->mutable_join_hash_table(); switch (GetParam()) { case HashTableImplType::kLinearOpenAddressing: hash_table_proto->set_hash_table_impl_type( @@ -341,6 +347,7 @@ TEST_P(HashJoinOperatorTest, LongKeyHashJoinTest) { true /* is_stored */, std::vector(1, dim_col_long.getID()), dim_col_long.getType().isNullable(), + num_partitions_, join_hash_table_index)); // Create the prober operator with one selection attribute. @@ -370,6 +377,7 @@ TEST_P(HashJoinOperatorTest, LongKeyHashJoinTest) { true /* is_stored */, std::vector(1, fact_col_long.getID()), fact_col_long.getType().isNullable(), + num_partitions_, *result_table, output_destination_index, join_hash_table_index, @@ -427,7 +435,7 @@ TEST_P(HashJoinOperatorTest, LongKeyHashJoinTest) { } // Create cleaner operator. - unique_ptr cleaner(new DestroyHashOperator(kQueryId, join_hash_table_index)); + unique_ptr cleaner(new DestroyHashOperator(kQueryId, num_partitions_, join_hash_table_index)); cleaner->informAllBlockingDependenciesMet(); fetchAndExecuteWorkOrders(cleaner.get()); @@ -443,7 +451,7 @@ TEST_P(HashJoinOperatorTest, IntDuplicateKeyHashJoinTest) { query_context_proto.join_hash_tables_size(); serialization::HashTable *hash_table_proto = - query_context_proto.add_join_hash_tables(); + query_context_proto.add_join_hash_tables()->mutable_join_hash_table(); switch (GetParam()) { case HashTableImplType::kLinearOpenAddressing: hash_table_proto->set_hash_table_impl_type( @@ -485,6 +493,7 @@ TEST_P(HashJoinOperatorTest, IntDuplicateKeyHashJoinTest) { true /* is_stored */, std::vector(1, dim_col_int.getID()), dim_col_int.getType().isNullable(), + num_partitions_, join_hash_table_index)); // Create the prober operator with two selection attributes. @@ -519,6 +528,7 @@ TEST_P(HashJoinOperatorTest, IntDuplicateKeyHashJoinTest) { true /* is_stored */, std::vector(1, fact_col_int.getID()), fact_col_int.getType().isNullable(), + num_partitions_, *result_table, output_destination_index, join_hash_table_index, @@ -598,7 +608,7 @@ TEST_P(HashJoinOperatorTest, IntDuplicateKeyHashJoinTest) { } // Create cleaner operator. - unique_ptr cleaner(new DestroyHashOperator(kQueryId, join_hash_table_index)); + unique_ptr cleaner(new DestroyHashOperator(kQueryId, num_partitions_, join_hash_table_index)); cleaner->informAllBlockingDependenciesMet(); fetchAndExecuteWorkOrders(cleaner.get()); @@ -614,7 +624,7 @@ TEST_P(HashJoinOperatorTest, CharKeyCartesianProductHashJoinTest) { query_context_proto.join_hash_tables_size(); serialization::HashTable *hash_table_proto = - query_context_proto.add_join_hash_tables(); + query_context_proto.add_join_hash_tables()->mutable_join_hash_table(); switch (GetParam()) { case HashTableImplType::kLinearOpenAddressing: hash_table_proto->set_hash_table_impl_type( @@ -648,6 +658,7 @@ TEST_P(HashJoinOperatorTest, CharKeyCartesianProductHashJoinTest) { true /* is_stored */, std::vector(1, dim_col_char.getID()), dim_col_char.getType().isNullable(), + num_partitions_, join_hash_table_index)); // Create prober operator with one selection attribute. @@ -677,6 +688,7 @@ TEST_P(HashJoinOperatorTest, CharKeyCartesianProductHashJoinTest) { true /* is_stored */, std::vector(1, fact_col_char.getID()), fact_col_char.getType().isNullable(), + num_partitions_, *result_table, output_destination_index, join_hash_table_index, @@ -734,7 +746,7 @@ TEST_P(HashJoinOperatorTest, CharKeyCartesianProductHashJoinTest) { } // Create cleaner operator. - unique_ptr cleaner(new DestroyHashOperator(kQueryId, join_hash_table_index)); + unique_ptr cleaner(new DestroyHashOperator(kQueryId, num_partitions_, join_hash_table_index)); cleaner->informAllBlockingDependenciesMet(); fetchAndExecuteWorkOrders(cleaner.get()); @@ -750,7 +762,7 @@ TEST_P(HashJoinOperatorTest, VarCharDuplicateKeyHashJoinTest) { query_context_proto.join_hash_tables_size(); serialization::HashTable *hash_table_proto = - query_context_proto.add_join_hash_tables(); + query_context_proto.add_join_hash_tables()->mutable_join_hash_table(); switch (GetParam()) { case HashTableImplType::kLinearOpenAddressing: hash_table_proto->set_hash_table_impl_type( @@ -785,6 +797,7 @@ TEST_P(HashJoinOperatorTest, VarCharDuplicateKeyHashJoinTest) { true /* is_stored */, std::vector(1, dim_col_varchar.getID()), dim_col_varchar.getType().isNullable(), + num_partitions_, join_hash_table_index)); // Create prober operator with two selection attributes. @@ -819,6 +832,7 @@ TEST_P(HashJoinOperatorTest, VarCharDuplicateKeyHashJoinTest) { true /* is_stored */, std::vector(1, fact_col_varchar.getID()), fact_col_varchar.getType().isNullable(), + num_partitions_, *result_table, output_destination_index, join_hash_table_index, @@ -902,7 +916,7 @@ TEST_P(HashJoinOperatorTest, VarCharDuplicateKeyHashJoinTest) { } // Create the cleaner operator. - unique_ptr cleaner(new DestroyHashOperator(kQueryId, join_hash_table_index)); + unique_ptr cleaner(new DestroyHashOperator(kQueryId, num_partitions_, join_hash_table_index)); cleaner->informAllBlockingDependenciesMet(); fetchAndExecuteWorkOrders(cleaner.get()); @@ -918,7 +932,7 @@ TEST_P(HashJoinOperatorTest, CompositeKeyHashJoinTest) { query_context_proto.join_hash_tables_size(); serialization::HashTable *hash_table_proto = - query_context_proto.add_join_hash_tables(); + query_context_proto.add_join_hash_tables()->mutable_join_hash_table(); switch (GetParam()) { case HashTableImplType::kLinearOpenAddressing: hash_table_proto->set_hash_table_impl_type( @@ -958,6 +972,7 @@ TEST_P(HashJoinOperatorTest, CompositeKeyHashJoinTest) { true /* is_stored */, dim_key_attrs, dim_col_long.getType().isNullable() || dim_col_varchar.getType().isNullable(), + num_partitions_, join_hash_table_index)); // Create the prober operator with two selection attributes. @@ -997,6 +1012,7 @@ TEST_P(HashJoinOperatorTest, CompositeKeyHashJoinTest) { fact_key_attrs, fact_col_long.getType().isNullable() || fact_col_varchar.getType().isNullable(), + num_partitions_, *result_table, output_destination_index, join_hash_table_index, @@ -1079,7 +1095,7 @@ TEST_P(HashJoinOperatorTest, CompositeKeyHashJoinTest) { } // Create cleaner operator. - unique_ptr cleaner(new DestroyHashOperator(kQueryId, join_hash_table_index)); + unique_ptr cleaner(new DestroyHashOperator(kQueryId, num_partitions_, join_hash_table_index)); cleaner->informAllBlockingDependenciesMet(); fetchAndExecuteWorkOrders(cleaner.get()); @@ -1096,7 +1112,7 @@ TEST_P(HashJoinOperatorTest, CompositeKeyHashJoinWithResidualPredicateTest) { query_context_proto.join_hash_tables_size(); serialization::HashTable *hash_table_proto = - query_context_proto.add_join_hash_tables(); + query_context_proto.add_join_hash_tables()->mutable_join_hash_table(); switch (GetParam()) { case HashTableImplType::kLinearOpenAddressing: hash_table_proto->set_hash_table_impl_type( @@ -1136,6 +1152,7 @@ TEST_P(HashJoinOperatorTest, CompositeKeyHashJoinWithResidualPredicateTest) { true /* is_stored */, dim_key_attrs, dim_col_long.getType().isNullable() || dim_col_varchar.getType().isNullable(), + num_partitions_, join_hash_table_index)); // Create prober operator with two selection attributes. @@ -1185,6 +1202,7 @@ TEST_P(HashJoinOperatorTest, CompositeKeyHashJoinWithResidualPredicateTest) { fact_key_attrs, fact_col_long.getType().isNullable() || fact_col_varchar.getType().isNullable(), + num_partitions_, *result_table, output_destination_index, join_hash_table_index, @@ -1267,7 +1285,7 @@ TEST_P(HashJoinOperatorTest, CompositeKeyHashJoinWithResidualPredicateTest) { } // Create cleaner operator. - unique_ptr cleaner(new DestroyHashOperator(kQueryId, join_hash_table_index)); + unique_ptr cleaner(new DestroyHashOperator(kQueryId, num_partitions_, join_hash_table_index)); cleaner->informAllBlockingDependenciesMet(); fetchAndExecuteWorkOrders(cleaner.get()); http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/8e452194/storage/StorageBlockInfo.hpp ---------------------------------------------------------------------- diff --git a/storage/StorageBlockInfo.hpp b/storage/StorageBlockInfo.hpp index 361648f..e3ede3d 100644 --- a/storage/StorageBlockInfo.hpp +++ b/storage/StorageBlockInfo.hpp @@ -25,6 +25,7 @@ #include #include #include +#include #include "utility/Macros.hpp" @@ -49,6 +50,8 @@ static constexpr int kBlockIdDomainLengthInDigits = std::numeric_limits::digits10; static constexpr block_id_domain kMaxDomain = static_cast(0xFFFF); +typedef std::vector blocks_in_partition; + /** * @brief All-static object that provides access to block_id. **/