Repository: incubator-quickstep Updated Branches: refs/heads/lip-refactor 51f020ecc -> 0cca12890
Updates Project: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/commit/0cca1289 Tree: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/tree/0cca1289 Diff: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/diff/0cca1289 Branch: refs/heads/lip-refactor Commit: 0cca1289019c2e5aab623327a2268605af2dbf17 Parents: 51f020e Author: Jianqiao Zhu <jianq...@cs.wisc.edu> Authored: Tue Oct 4 23:22:51 2016 -0500 Committer: Jianqiao Zhu <jianq...@cs.wisc.edu> Committed: Tue Oct 4 23:22:51 2016 -0500 ---------------------------------------------------------------------- query_execution/QueryContext.cpp | 21 +--- query_optimizer/ExecutionGenerator.cpp | 5 +- query_optimizer/LIPFilterGenerator.cpp | 109 ++++++++++++------- query_optimizer/LIPFilterGenerator.hpp | 14 ++- query_optimizer/PhysicalGenerator.cpp | 1 - .../physical/LIPFilterConfiguration.hpp | 18 +-- query_optimizer/rules/AttachLIPFilters.cpp | 4 +- storage/HashTableFactory.hpp | 5 +- utility/PlanVisualizer.cpp | 4 +- 9 files changed, 102 insertions(+), 79 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/0cca1289/query_execution/QueryContext.cpp ---------------------------------------------------------------------- diff --git a/query_execution/QueryContext.cpp b/query_execution/QueryContext.cpp index 1a1b296..47408d4 100644 --- a/query_execution/QueryContext.cpp +++ b/query_execution/QueryContext.cpp @@ -68,10 +68,6 @@ QueryContext::QueryContext(const serialization::QueryContext &proto, storage_manager)); } -// for (int i = 0; i < proto.bloom_filters_size(); ++i) { -// bloom_filters_.emplace_back(new BloomFilter(proto.bloom_filters(i))); -// } - for (int i = 0; i < proto.generator_functions_size(); ++i) { const GeneratorFunctionHandle *func_handle = GeneratorFunctionFactory::Instance().reconstructFromProto(proto.generator_functions(i)); @@ -80,12 +76,11 @@ QueryContext::QueryContext(const serialization::QueryContext &proto, std::unique_ptr<const GeneratorFunctionHandle>(func_handle)); } -// 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, -// bloom_filters_)); -// } + 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)); + } for (int i = 0; i < proto.insert_destinations_size(); ++i) { const serialization::InsertDestination &insert_destination_proto = proto.insert_destinations(i); @@ -157,12 +152,6 @@ bool QueryContext::ProtoIsValid(const serialization::QueryContext &proto, } } -// for (int i = 0; i < proto.bloom_filters_size(); ++i) { -// if (!BloomFilter::ProtoIsValid(proto.bloom_filters(i))) { -// return false; -// } -// } - // Each GeneratorFunctionHandle object is serialized as a function name with // a list of arguments. Here checks that the arguments are valid TypedValue's. for (int i = 0; i < proto.generator_functions_size(); ++i) { http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/0cca1289/query_optimizer/ExecutionGenerator.cpp ---------------------------------------------------------------------- diff --git a/query_optimizer/ExecutionGenerator.cpp b/query_optimizer/ExecutionGenerator.cpp index 7bfa949..0047818 100644 --- a/query_optimizer/ExecutionGenerator.cpp +++ b/query_optimizer/ExecutionGenerator.cpp @@ -176,6 +176,8 @@ void ExecutionGenerator::generatePlan(const P::PhysicalPtr &physical_plan) { } generatePlanInternal(top_level_physical_plan_->plan()); + lip_filter_generator_->deployLIPFilters(execution_plan_, query_context_proto_); + // Set the query result relation if the input plan exists in physical_to_execution_map_, // which indicates the plan is the result of a SELECT query. const std::unordered_map<P::PhysicalPtr, CatalogRelationInfo>::const_iterator it = @@ -1412,7 +1414,8 @@ void ExecutionGenerator::convertAggregate( aggr_state_proto->mutable_predicate()->CopyFrom(predicate->getProto()); } - aggr_state_proto->set_estimated_num_entries(cost_model_->estimateCardinality(physical_plan)); +// aggr_state_proto->set_estimated_num_entries(cost_model_->estimateCardinality(physical_plan)); + aggr_state_proto->set_estimated_num_entries(64uL); const QueryPlan::DAGNodeIndex aggregation_operator_index = execution_plan_->addRelationalOperator( http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/0cca1289/query_optimizer/LIPFilterGenerator.cpp ---------------------------------------------------------------------- diff --git a/query_optimizer/LIPFilterGenerator.cpp b/query_optimizer/LIPFilterGenerator.cpp index abde004..4e24740 100644 --- a/query_optimizer/LIPFilterGenerator.cpp +++ b/query_optimizer/LIPFilterGenerator.cpp @@ -39,18 +39,18 @@ namespace P = ::quickstep::optimizer::physical; void LIPFilterGenerator::registerAttributeMap( const P::PhysicalPtr &node, const std::unordered_map<E::ExprId, const CatalogAttribute *> &attribute_substitution_map) { - const auto &build_infos = lip_filter_configuration_->getBuildInfo(); - const auto build_it = build_infos.find(node); - if (build_it != build_infos.end()) { + const auto &build_info_map = lip_filter_configuration_->getBuildInfoMap(); + const auto build_it = build_info_map.find(node); + if (build_it != build_info_map.end()) { auto &map_entry = attribute_map_[node]; for (const auto &info : build_it->second) { E::ExprId attr_id = info.build_attribute->id(); map_entry.emplace(attr_id, attribute_substitution_map.at(attr_id)); } } - const auto &probe_infos = lip_filter_configuration_->getProbeInfo(); - const auto probe_it = probe_infos.find(node); - if (probe_it != probe_infos.end()) { + const auto &probe_info_map = lip_filter_configuration_->getProbeInfoMap(); + const auto probe_it = probe_info_map.find(node); + if (probe_it != probe_info_map.end()) { auto &map_entry = attribute_map_[node]; for (const auto &info : probe_it->second) { E::ExprId attr_id = info.probe_attribute->id(); @@ -61,45 +61,70 @@ void LIPFilterGenerator::registerAttributeMap( void LIPFilterGenerator::deployLIPFilters(QueryPlan *execution_plan, serialization::QueryContext *query_context_proto) const { - std::map<std::pair<E::ExprId, P::PhysicalPtr>, - std::pair<QueryContext::lip_filter_id, QueryPlan::DAGNodeIndex>> lip_filter_builder_map; - for (const auto &builder : lip_filter_configuration_->getBuildInfo()) { - const P::PhysicalPtr &builder_node = builder.first; - const auto &builder_attribute_map = attribute_map_.at(builder_node); - for (const auto &info : builder.second) { - const QueryContext::lip_filter_id lip_filter_id = query_context_proto->lip_filters_size(); - serialization::LIPFilter *lip_filter_proto = query_context_proto->add_lip_filters(); - - switch (info.filter_type) { - case LIPFilterType::kSingleIdentityHashFilter: - lip_filter_proto->set_lip_filter_type( - serialization::LIPFilterType::SINGLE_IDENTITY_HASH_FILTER); - lip_filter_proto->SetExtension( - serialization::SingleIdentityHashFilter::num_bits, info.filter_size); - break; - default: - LOG(FATAL) << "Unsupported LIPFilter type"; - break; - } - - lip_filter_builder_map.emplace( - std::make_pair(info.build_attribute->id(), builder_node), - std::make_pair(lip_filter_id, builder_op_index_map_.at(builder_node))); - - auto *lip_filter_deployment_info_proto = - query_context_proto->add_lip_filter_deployment_infos(); - lip_filter_deployment_info_proto->set_action_type(serialization::LIPFilterActionType::BUILD); - lip_filter_deployment_info_proto->set_lip_filter_id(lip_filter_id); - - const CatalogAttribute *target_attr = builder_attribute_map.at(info.build_attribute->id()); - lip_filter_deployment_info_proto->set_attribute_id(target_attr->getID()); - lip_filter_deployment_info_proto->mutable_attribute_type()->CopyFrom(target_attr->getProto()); - - std::cerr << "Build " << info.build_attribute->toString() - << " @" << builder_node << "\n"; + LIPFilterBuilderMap lip_filter_builder_map; + + // Deploy builders + const auto &build_info_map = lip_filter_configuration_->getBuildInfoMap(); + for (const auto &hash_join_info : hash_join_infos_) { + const P::PhysicalPtr &builder_node = hash_join_info.hash_join; + const auto build_it = build_info_map.find(builder_node); + if (build_it != build_info_map.end()) { + deployBuilderInternal(execution_plan, + query_context_proto, + builder_node, + hash_join_info.build_operator_index, + build_it->second, + &lip_filter_builder_map); + } + } + + // Deploy probers + // const auto &probe_info_map = lip_filter_configuration_->getProbeInfoMap(); +} + +void LIPFilterGenerator::deployBuilderInternal( + QueryPlan *execution_plan, + serialization::QueryContext *query_context_proto, + const physical::PhysicalPtr &builder_node, + const QueryPlan::DAGNodeIndex builder_operator_index, + const std::vector<physical::LIPFilterBuildInfo> &build_info_vec, + LIPFilterBuilderMap *lip_filter_builder_map) const { + const auto &builder_attribute_map = attribute_map_.at(builder_node); + for (const auto &info : build_info_vec) { + const QueryContext::lip_filter_id lip_filter_id = query_context_proto->lip_filters_size(); + serialization::LIPFilter *lip_filter_proto = query_context_proto->add_lip_filters(); + + switch (info.filter_type) { + case LIPFilterType::kSingleIdentityHashFilter: + lip_filter_proto->set_lip_filter_type( + serialization::LIPFilterType::SINGLE_IDENTITY_HASH_FILTER); + lip_filter_proto->SetExtension( + serialization::SingleIdentityHashFilter::num_bits, info.filter_size); + break; + default: + LOG(FATAL) << "Unsupported LIPFilter type"; + break; } + + lip_filter_builder_map->emplace( + std::make_pair(info.build_attribute->id(), builder_node), + std::make_pair(lip_filter_id, builder_operator_index)); + + auto *lip_filter_deployment_info_proto = + query_context_proto->add_lip_filter_deployment_infos(); + lip_filter_deployment_info_proto->set_action_type(serialization::LIPFilterActionType::BUILD); + lip_filter_deployment_info_proto->set_lip_filter_id(lip_filter_id); + + const CatalogAttribute *target_attr = builder_attribute_map.at(info.build_attribute->id()); + lip_filter_deployment_info_proto->set_attribute_id(target_attr->getID()); + lip_filter_deployment_info_proto->mutable_attribute_type()->CopyFrom( + target_attr->getType().getProto()); + + std::cerr << "Build " << info.build_attribute->toString() + << " @" << builder_node << "\n"; } } + } // namespace optimizer } // namespace quickstep http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/0cca1289/query_optimizer/LIPFilterGenerator.hpp ---------------------------------------------------------------------- diff --git a/query_optimizer/LIPFilterGenerator.hpp b/query_optimizer/LIPFilterGenerator.hpp index 5b74757..05270c9 100644 --- a/query_optimizer/LIPFilterGenerator.hpp +++ b/query_optimizer/LIPFilterGenerator.hpp @@ -63,7 +63,6 @@ class LIPFilterGenerator { void addHashJoinInfo(const physical::HashJoinPtr &hash_join, const QueryPlan::DAGNodeIndex build_operator_index, const QueryPlan::DAGNodeIndex join_operator_index) { - builder_op_index_map_.emplace(hash_join, build_operator_index); hash_join_infos_.emplace_back(hash_join, build_operator_index, join_operator_index); } @@ -112,9 +111,20 @@ class LIPFilterGenerator { const QueryPlan::DAGNodeIndex select_operator_index; }; + typedef std::map<std::pair<expressions::ExprId, physical::PhysicalPtr>, + std::pair<QueryContext::lip_filter_id, QueryPlan::DAGNodeIndex>> LIPFilterBuilderMap; + + void deployBuilderInternal(QueryPlan *execution_plan, + serialization::QueryContext *query_context_proto, + const physical::PhysicalPtr &builder_node, + const QueryPlan::DAGNodeIndex builder_operator_index, + const std::vector<physical::LIPFilterBuildInfo> &build_info_vec, + LIPFilterBuilderMap *lip_filter_builder_map) const; + + void deployProberInteral(); + const physical::LIPFilterConfigurationPtr lip_filter_configuration_; std::map<physical::PhysicalPtr, std::map<expressions::ExprId, const CatalogAttribute *>> attribute_map_; - std::map<physical::PhysicalPtr, QueryPlan::DAGNodeIndex> builder_op_index_map_; std::vector<AggregateInfo> aggregate_infos_; std::vector<HashJoinInfo> hash_join_infos_; std::vector<SelectionInfo> selection_infos_; http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/0cca1289/query_optimizer/PhysicalGenerator.cpp ---------------------------------------------------------------------- diff --git a/query_optimizer/PhysicalGenerator.cpp b/query_optimizer/PhysicalGenerator.cpp index a3fedc9..eb1ec08 100644 --- a/query_optimizer/PhysicalGenerator.cpp +++ b/query_optimizer/PhysicalGenerator.cpp @@ -115,7 +115,6 @@ P::PhysicalPtr PhysicalGenerator::optimizePlan() { quickstep::PlanVisualizer plan_visualizer; std::cerr << "\n" << plan_visualizer.visualize(physical_plan_) << "\n"; } - exit(0); #ifdef QUICKSTEP_DEBUG Validate(physical_plan_); http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/0cca1289/query_optimizer/physical/LIPFilterConfiguration.hpp ---------------------------------------------------------------------- diff --git a/query_optimizer/physical/LIPFilterConfiguration.hpp b/query_optimizer/physical/LIPFilterConfiguration.hpp index 153d677..9b028ad 100644 --- a/query_optimizer/physical/LIPFilterConfiguration.hpp +++ b/query_optimizer/physical/LIPFilterConfiguration.hpp @@ -42,7 +42,7 @@ namespace physical { class Physical; typedef std::shared_ptr<const Physical> PhysicalPtr; -struct LIPFilterBuildInfo{ +struct LIPFilterBuildInfo { LIPFilterBuildInfo(const expressions::AttributeReferencePtr &build_attribute_in, const std::size_t filter_size_in, const LIPFilterType &filter_type_in) @@ -81,7 +81,7 @@ class LIPFilterConfiguration { const PhysicalPtr &builder, const std::size_t filter_size, const LIPFilterType &filter_type) { - build_info_[builder].emplace_back( + build_info_map_[builder].emplace_back( build_attribute, filter_size, filter_type); } @@ -89,21 +89,21 @@ class LIPFilterConfiguration { const PhysicalPtr &prober, const expressions::AttributeReferencePtr &build_attribute, const PhysicalPtr &builder) { - probe_info_[prober].emplace_back( + probe_info_map_[prober].emplace_back( probe_attribute, build_attribute, builder); } - const std::map<PhysicalPtr, std::vector<LIPFilterBuildInfo>>& getBuildInfo() const { - return build_info_; + const std::map<PhysicalPtr, std::vector<LIPFilterBuildInfo>>& getBuildInfoMap() const { + return build_info_map_; } - const std::map<PhysicalPtr, std::vector<LIPFilterProbeInfo>>& getProbeInfo() const { - return probe_info_; + const std::map<PhysicalPtr, std::vector<LIPFilterProbeInfo>>& getProbeInfoMap() const { + return probe_info_map_; } private: - std::map<PhysicalPtr, std::vector<LIPFilterBuildInfo>> build_info_; - std::map<PhysicalPtr, std::vector<LIPFilterProbeInfo>> probe_info_; + std::map<PhysicalPtr, std::vector<LIPFilterBuildInfo>> build_info_map_; + std::map<PhysicalPtr, std::vector<LIPFilterProbeInfo>> probe_info_map_; DISALLOW_COPY_AND_ASSIGN(LIPFilterConfiguration); }; http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/0cca1289/query_optimizer/rules/AttachLIPFilters.cpp ---------------------------------------------------------------------- diff --git a/query_optimizer/rules/AttachLIPFilters.cpp b/query_optimizer/rules/AttachLIPFilters.cpp index e79c883..d67eacd 100644 --- a/query_optimizer/rules/AttachLIPFilters.cpp +++ b/query_optimizer/rules/AttachLIPFilters.cpp @@ -71,8 +71,8 @@ P::PhysicalPtr AttachLIPFilters::apply(const P::PhysicalPtr &input) { // } P::PhysicalPtr output; - if (!lip_filter_configuration_->getBuildInfo().empty() || - !lip_filter_configuration_->getProbeInfo().empty()) { + if (!lip_filter_configuration_->getBuildInfoMap().empty() || + !lip_filter_configuration_->getProbeInfoMap().empty()) { output = top_level_plan->copyWithLIPFilterConfiguration( P::LIPFilterConfigurationPtr(lip_filter_configuration_.release())); } else { http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/0cca1289/storage/HashTableFactory.hpp ---------------------------------------------------------------------- diff --git a/storage/HashTableFactory.hpp b/storage/HashTableFactory.hpp index cd79723..d690557 100644 --- a/storage/HashTableFactory.hpp +++ b/storage/HashTableFactory.hpp @@ -295,14 +295,11 @@ class HashTableFactory { * @param proto A protobuf description of a resizable HashTable. * @param storage_manager The StorageManager to use (a StorageBlob will be * allocated to hold the HashTable's contents). - * @param bloom_filters A vector of pointers to bloom filters that may be used - * during hash table construction in build/probe phase. * @return A new resizable HashTable with parameters specified by proto. **/ static HashTable<ValueT, resizable, serializable, force_key_copy, allow_duplicate_keys>* CreateResizableFromProto(const serialization::HashTable &proto, - StorageManager *storage_manager, - const std::vector<std::unique_ptr<BloomFilter>> &bloom_filters) { + StorageManager *storage_manager) { DCHECK(ProtoIsValid(proto)) << "Attempted to create HashTable from invalid proto description:\n" << proto.DebugString(); http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/0cca1289/utility/PlanVisualizer.cpp ---------------------------------------------------------------------- diff --git a/utility/PlanVisualizer.cpp b/utility/PlanVisualizer.cpp index fdb5812..5156c1e 100644 --- a/utility/PlanVisualizer.cpp +++ b/utility/PlanVisualizer.cpp @@ -168,7 +168,7 @@ void PlanVisualizer::visit(const P::PhysicalPtr &input) { } if (lip_filter_conf_ != nullptr) { - const auto &build_filters = lip_filter_conf_->getBuildInfo(); + const auto &build_filters = lip_filter_conf_->getBuildInfoMap(); const auto build_it = build_filters.find(input); if (build_it != build_filters.end()) { for (const auto &build_info : build_it->second) { @@ -176,7 +176,7 @@ void PlanVisualizer::visit(const P::PhysicalPtr &input) { std::string("[LIP build] ") + build_info.build_attribute->attribute_alias()); } } - const auto &probe_filters = lip_filter_conf_->getProbeInfo(); + const auto &probe_filters = lip_filter_conf_->getProbeInfoMap(); const auto probe_it = probe_filters.find(input); if (probe_it != probe_filters.end()) { for (const auto &probe_info : probe_it->second) {
