You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@quickstep.apache.org by ji...@apache.org on 2016/10/28 19:34:14 UTC
[01/11] incubator-quickstep git commit: Fix a bug in destroy payload
[Forced Update!]
Repository: incubator-quickstep
Updated Branches:
refs/heads/exact-filter b9db77d96 -> f10f5677b (forced update)
Fix a bug in destroy payload
- Change in pointer arithmetic now ensures that destroyPayload will be
called for correct payload entry in each bucket.
Project: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/commit/a11cdfd5
Tree: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/tree/a11cdfd5
Diff: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/diff/a11cdfd5
Branch: refs/heads/exact-filter
Commit: a11cdfd5d9b364cacd755ade30d1e99b51893e3b
Parents: 5b24791
Author: Harshad Deshmukh <hb...@apache.org>
Authored: Thu Oct 20 17:06:46 2016 -0500
Committer: Harshad Deshmukh <hb...@apache.org>
Committed: Thu Oct 20 17:06:46 2016 -0500
----------------------------------------------------------------------
storage/FastSeparateChainingHashTable.hpp | 11 ++++-------
1 file changed, 4 insertions(+), 7 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/a11cdfd5/storage/FastSeparateChainingHashTable.hpp
----------------------------------------------------------------------
diff --git a/storage/FastSeparateChainingHashTable.hpp b/storage/FastSeparateChainingHashTable.hpp
index 17ee1d2..231da7e 100644
--- a/storage/FastSeparateChainingHashTable.hpp
+++ b/storage/FastSeparateChainingHashTable.hpp
@@ -141,19 +141,16 @@ class FastSeparateChainingHashTable
HashTablePreallocationState *prealloc_state) override;
void destroyPayload() override {
- void *hash_buckets = buckets_;
const std::size_t num_buckets =
header_->buckets_allocated.load(std::memory_order_relaxed);
- const std::size_t bucket_size = bucket_size_;
- void *bucket_ptr = static_cast<char *>(hash_buckets) + kValueOffset;
+ void *bucket_ptr = static_cast<char *>(buckets_) + kValueOffset;
for (std::size_t bucket_num = 0; bucket_num < num_buckets; ++bucket_num) {
- void *value_internal_ptr = bucket_ptr;
for (std::size_t handle_num = 0; handle_num < handles_.size(); ++handle_num) {
- value_internal_ptr =
- static_cast<char *>(value_internal_ptr) + this->payload_offsets_[handle_num];
+ void *value_internal_ptr =
+ static_cast<char *>(bucket_ptr) + this->payload_offsets_[handle_num];
handles_[handle_num]->destroyPayload(static_cast<std::uint8_t *>(value_internal_ptr));
}
- bucket_ptr = static_cast<char *>(bucket_ptr) + bucket_size;
+ bucket_ptr = static_cast<char *>(bucket_ptr) + bucket_size_;
}
}
[05/11] incubator-quickstep git commit: QUICKSTEP-59 Convert
setBitRegularVersion() to a branchless version
Posted by ji...@apache.org.
QUICKSTEP-59 Convert setBitRegularVersion() to a branchless version
Project: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/commit/539a95a1
Tree: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/tree/539a95a1
Diff: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/diff/539a95a1
Branch: refs/heads/exact-filter
Commit: 539a95a1531eda83aa6bd47f5496d909754d0124
Parents: 9c32ea4
Author: Saket Saurabh <ss...@cs.wisc.edu>
Authored: Mon Oct 17 17:50:40 2016 -0500
Committer: Harshad Deshmukh <hb...@apache.org>
Committed: Sun Oct 23 08:30:41 2016 -0500
----------------------------------------------------------------------
utility/BitVector.hpp | 33 ++++++++++++++++++++++++++-------
1 file changed, 26 insertions(+), 7 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/539a95a1/utility/BitVector.hpp
----------------------------------------------------------------------
diff --git a/utility/BitVector.hpp b/utility/BitVector.hpp
index 2b56b8c..c404b7e 100644
--- a/utility/BitVector.hpp
+++ b/utility/BitVector.hpp
@@ -890,13 +890,32 @@ class BitVector {
}
inline void setBitRegularVersion(const std::size_t bit_num, bool value) {
- if (value) {
- data_array_[bit_num >> kHigherOrderShift]
- |= (TopBit<std::size_t>() >> (bit_num & kLowerOrderMask));
- } else {
- data_array_[bit_num >> kHigherOrderShift]
- &= ~(TopBit<std::size_t>() >> (bit_num & kLowerOrderMask));
- }
+ // The mask has all the bits set to 1, except the bit at the required bit position
+ // in the word that has be to set/unset based on the boolean value.
+ std::size_t mask = ~(TopBit<std::size_t>() >> (bit_num & kLowerOrderMask));
+
+ // The bit_value type casts the value argument (true or false) to either (1 or 0).
+ std::size_t bit_value = value;
+
+ // Subtracting 1 from the bit_value, will generate a value
+ // that will either be all zeros (if the required bit_value is true),
+ // or all ones (if the required bit_value was false). The trick here is that
+ // subtracting one from an unsigned zero value will set all the bits to 1.
+ // Finally, bitwise OR with the mask followed by negation will generate an op_value
+ // that will have all bits as 0's if the required bit_value was false, or
+ // will have only 1 at the required bit position if the bit_value was true.
+ std::size_t op_value = ~((bit_value - 1) | mask);
+
+ // Get the index position in the data_array_ that is being operated upon.
+ std::size_t index_pos_in_data_array = bit_num >> kHigherOrderShift;
+
+ // First, by doing a bitwise AND with the mask, we clear the required bit position.
+ data_array_[index_pos_in_data_array] &= mask;
+
+ // Then, we set the required bit position with the op_value the we generated above.
+ // Note that bitwise OR with op_value will set the required bit position to 1 if the
+ // value was indeed true, but it will be a no-op if the value was false.
+ data_array_[index_pos_in_data_array] |= op_value;
}
template <typename VectorType>
[03/11] incubator-quickstep git commit: ExecutionGenerator and
QueryContext support for LIPFilters.
Posted by ji...@apache.org.
ExecutionGenerator and QueryContext support for LIPFilters.
Project: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/commit/8746eed4
Tree: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/tree/8746eed4
Diff: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/diff/8746eed4
Branch: refs/heads/exact-filter
Commit: 8746eed4d06ee07d64cbef8ab7851c53c50e95be
Parents: a58a747
Author: Jianqiao Zhu <ji...@cs.wisc.edu>
Authored: Wed Sep 7 13:20:43 2016 -0500
Committer: Jianqiao Zhu <ji...@cs.wisc.edu>
Committed: Fri Oct 21 12:34:59 2016 -0500
----------------------------------------------------------------------
query_execution/CMakeLists.txt | 7 +-
query_execution/QueryContext.cpp | 33 ++-
query_execution/QueryContext.hpp | 96 ++++++++
query_execution/QueryContext.proto | 25 +-
query_optimizer/CMakeLists.txt | 21 ++
query_optimizer/ExecutionGenerator.cpp | 36 ++-
query_optimizer/ExecutionGenerator.hpp | 4 +
query_optimizer/LIPFilterGenerator.cpp | 194 +++++++++++++++
query_optimizer/LIPFilterGenerator.hpp | 194 +++++++++++++++
query_optimizer/QueryPlan.hpp | 32 +++
relational_operators/RelationalOperator.hpp | 12 +-
utility/lip_filter/CMakeLists.txt | 55 ++++-
utility/lip_filter/LIPFilter.hpp | 64 +++++
utility/lip_filter/LIPFilter.proto | 58 +++++
utility/lip_filter/LIPFilterAdaptiveProber.hpp | 243 +++++++++++++++++++
utility/lip_filter/LIPFilterBuilder.hpp | 109 +++++++++
utility/lip_filter/LIPFilterDeployment.cpp | 87 +++++++
utility/lip_filter/LIPFilterDeployment.hpp | 111 +++++++++
utility/lip_filter/LIPFilterFactory.cpp | 74 ++++++
utility/lip_filter/LIPFilterFactory.hpp | 69 ++++++
utility/lip_filter/SingleIdentityHashFilter.hpp | 185 ++++++++++++++
21 files changed, 1684 insertions(+), 25 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/8746eed4/query_execution/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/query_execution/CMakeLists.txt b/query_execution/CMakeLists.txt
index dafdea4..b5e07df 100644
--- a/query_execution/CMakeLists.txt
+++ b/query_execution/CMakeLists.txt
@@ -190,9 +190,11 @@ target_link_libraries(quickstep_queryexecution_QueryContext
quickstep_types_TypedValue
quickstep_types_containers_Tuple
quickstep_utility_Macros
- quickstep_utility_SortConfiguration)
+ quickstep_utility_SortConfiguration
+ quickstep_utility_lipfilter_LIPFilter
+ quickstep_utility_lipfilter_LIPFilterDeployment
+ quickstep_utility_lipfilter_LIPFilterFactory)
target_link_libraries(quickstep_queryexecution_QueryContext_proto
- quickstep_utility_BloomFilter_proto
quickstep_expressions_Expressions_proto
quickstep_expressions_tablegenerator_GeneratorFunction_proto
quickstep_storage_AggregationOperationState_proto
@@ -201,6 +203,7 @@ target_link_libraries(quickstep_queryexecution_QueryContext_proto
quickstep_storage_WindowAggregationOperationState_proto
quickstep_types_containers_Tuple_proto
quickstep_utility_SortConfiguration_proto
+ quickstep_utility_lipfilter_LIPFilter_proto
${PROTOBUF_LIBRARY})
target_link_libraries(quickstep_queryexecution_QueryExecutionMessages_proto
quickstep_catalog_Catalog_proto
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/8746eed4/query_execution/QueryContext.cpp
----------------------------------------------------------------------
diff --git a/query_execution/QueryContext.cpp b/query_execution/QueryContext.cpp
index 6612611..0e6636d 100644
--- a/query_execution/QueryContext.cpp
+++ b/query_execution/QueryContext.cpp
@@ -40,6 +40,9 @@
#include "types/TypedValue.hpp"
#include "types/containers/Tuple.hpp"
#include "utility/SortConfiguration.hpp"
+#include "utility/lip_filter/LIPFilter.hpp"
+#include "utility/lip_filter/LIPFilterDeployment.hpp"
+#include "utility/lip_filter/LIPFilterFactory.hpp"
#include "glog/logging.h"
@@ -92,6 +95,18 @@ QueryContext::QueryContext(const serialization::QueryContext &proto,
bus));
}
+ for (int i = 0; i < proto.lip_filters_size(); ++i) {
+ lip_filters_.emplace_back(
+ std::unique_ptr<LIPFilter>(
+ LIPFilterFactory::ReconstructFromProto(proto.lip_filters(i))));
+ }
+
+ for (int i = 0; i < proto.lip_filter_deployments_size(); ++i) {
+ lip_deployments_.emplace_back(
+ std::make_unique<LIPFilterDeployment>(
+ proto.lip_filter_deployments(i), lip_filters_));
+ }
+
for (int i = 0; i < proto.predicates_size(); ++i) {
predicates_.emplace_back(
PredicateFactory::ReconstructFromProto(proto.predicates(i), database));
@@ -151,12 +166,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) {
@@ -185,6 +194,18 @@ bool QueryContext::ProtoIsValid(const serialization::QueryContext &proto,
}
}
+ for (int i = 0; i < proto.lip_filters_size(); ++i) {
+ if (!LIPFilterFactory::ProtoIsValid(proto.lip_filters(i))) {
+ return false;
+ }
+ }
+
+ for (int i = 0; i < proto.lip_filter_deployments_size(); ++i) {
+ if (!LIPFilterDeployment::ProtoIsValid(proto.lip_filter_deployments(i))) {
+ return false;
+ }
+ }
+
for (int i = 0; i < proto.predicates_size(); ++i) {
if (!PredicateFactory::ProtoIsValid(proto.predicates(i), database)) {
return false;
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/8746eed4/query_execution/QueryContext.hpp
----------------------------------------------------------------------
diff --git a/query_execution/QueryContext.hpp b/query_execution/QueryContext.hpp
index 7cc44ad..4ebb042 100644
--- a/query_execution/QueryContext.hpp
+++ b/query_execution/QueryContext.hpp
@@ -37,6 +37,8 @@
#include "types/containers/Tuple.hpp"
#include "utility/Macros.hpp"
#include "utility/SortConfiguration.hpp"
+#include "utility/lip_filter/LIPFilter.hpp"
+#include "utility/lip_filter/LIPFilterDeployment.hpp"
#include "glog/logging.h"
@@ -84,6 +86,17 @@ class QueryContext {
typedef std::uint32_t join_hash_table_id;
/**
+ * @brief A unique identifier for a LIPFilterDeployment per query.
+ **/
+ typedef std::uint32_t lip_deployment_id;
+ static constexpr lip_deployment_id kInvalidLIPDeploymentId = static_cast<lip_deployment_id>(-1);
+
+ /**
+ * @brief A unique identifier for a LIPFilter per query.
+ **/
+ typedef std::uint32_t lip_filter_id;
+
+ /**
* @brief A unique identifier for a Predicate per query.
*
* @note A negative value indicates a null Predicate.
@@ -295,6 +308,87 @@ class QueryContext {
}
/**
+ * @brief Whether the given LIPFilterDeployment id is valid.
+ *
+ * @param id The LIPFilterDeployment id.
+ *
+ * @return True if valid, otherwise false.
+ **/
+ bool isValidLIPDeploymentId(const lip_deployment_id id) const {
+ return id < lip_deployments_.size();
+ }
+
+ /**
+ * @brief Get a constant pointer to the LIPFilterDeployment.
+ *
+ * @param id The LIPFilterDeployment id.
+ *
+ * @return The constant pointer to LIPFilterDeployment that is
+ * already created in the constructor.
+ **/
+ inline const LIPFilterDeployment* getLIPDeployment(
+ const lip_deployment_id id) const {
+ DCHECK_LT(id, lip_deployments_.size());
+ return lip_deployments_[id].get();
+ }
+
+ /**
+ * @brief Destory the given LIPFilterDeployment.
+ *
+ * @param id The id of the LIPFilterDeployment to destroy.
+ **/
+ inline void destroyLIPDeployment(const lip_deployment_id id) {
+ DCHECK_LT(id, lip_deployments_.size());
+ lip_deployments_[id].reset();
+ }
+
+ /**
+ * @brief Whether the given LIPFilter id is valid.
+ *
+ * @param id The LIPFilter id.
+ *
+ * @return True if valid, otherwise false.
+ **/
+ bool isValidLIPFilterId(const lip_filter_id id) const {
+ return id < lip_filters_.size();
+ }
+
+ /**
+ * @brief Get a mutable reference to the LIPFilter.
+ *
+ * @param id The LIPFilter id.
+ *
+ * @return The LIPFilter, already created in the constructor.
+ **/
+ inline LIPFilter* getLIPFilterMutable(const lip_filter_id id) {
+ DCHECK_LT(id, lip_filters_.size());
+ return lip_filters_[id].get();
+ }
+
+ /**
+ * @brief Get a constant pointer to the LIPFilter.
+ *
+ * @param id The LIPFilter id.
+ *
+ * @return The constant pointer to LIPFilter that is
+ * already created in the constructor.
+ **/
+ inline const LIPFilter* getLIPFilter(const lip_filter_id id) const {
+ DCHECK_LT(id, lip_filters_.size());
+ return lip_filters_[id].get();
+ }
+
+ /**
+ * @brief Destory the given LIPFilter.
+ *
+ * @param id The id of the LIPFilter to destroy.
+ **/
+ inline void destroyLIPFilter(const lip_filter_id id) {
+ DCHECK_LT(id, lip_filters_.size());
+ lip_filters_[id].reset();
+ }
+
+ /**
* @brief Whether the given Predicate id is valid or no predicate.
*
* @param id The Predicate id.
@@ -472,6 +566,8 @@ class QueryContext {
std::vector<std::unique_ptr<const GeneratorFunctionHandle>> generator_functions_;
std::vector<std::unique_ptr<InsertDestination>> insert_destinations_;
std::vector<std::unique_ptr<JoinHashTable>> join_hash_tables_;
+ std::vector<std::unique_ptr<LIPFilterDeployment>> lip_deployments_;
+ std::vector<std::unique_ptr<LIPFilter>> lip_filters_;
std::vector<std::unique_ptr<const Predicate>> predicates_;
std::vector<std::vector<std::unique_ptr<const Scalar>>> scalar_groups_;
std::vector<std::unique_ptr<const SortConfiguration>> sort_configs_;
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/8746eed4/query_execution/QueryContext.proto
----------------------------------------------------------------------
diff --git a/query_execution/QueryContext.proto b/query_execution/QueryContext.proto
index 1a586a4..ab0f520 100644
--- a/query_execution/QueryContext.proto
+++ b/query_execution/QueryContext.proto
@@ -26,8 +26,8 @@ import "storage/HashTable.proto";
import "storage/InsertDestination.proto";
import "storage/WindowAggregationOperationState.proto";
import "types/containers/Tuple.proto";
-import "utility/BloomFilter.proto";
import "utility/SortConfiguration.proto";
+import "utility/lip_filter/LIPFilter.proto";
message QueryContext {
message ScalarGroup {
@@ -46,19 +46,20 @@ message QueryContext {
}
repeated AggregationOperationState aggregation_states = 1;
- repeated BloomFilter bloom_filters = 2;
- repeated GeneratorFunctionHandle generator_functions = 3;
- repeated HashTable join_hash_tables = 4;
- repeated InsertDestination insert_destinations = 5;
- repeated Predicate predicates = 6;
- repeated ScalarGroup scalar_groups = 7;
- repeated SortConfiguration sort_configs = 8;
- repeated Tuple tuples = 9;
+ repeated GeneratorFunctionHandle generator_functions = 2;
+ repeated HashTable join_hash_tables = 3;
+ repeated InsertDestination insert_destinations = 4;
+ repeated LIPFilter lip_filters = 5;
+ repeated LIPFilterDeployment lip_filter_deployments = 6;
+ repeated Predicate predicates = 7;
+ repeated ScalarGroup scalar_groups = 8;
+ repeated SortConfiguration sort_configs = 9;
+ repeated Tuple tuples = 10;
// NOTE(zuyu): For UpdateWorkOrder only.
- repeated UpdateGroup update_groups = 10;
+ repeated UpdateGroup update_groups = 11;
- repeated WindowAggregationOperationState window_aggregation_states = 11;
+ repeated WindowAggregationOperationState window_aggregation_states = 12;
- required uint64 query_id = 12;
+ required uint64 query_id = 13;
}
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/8746eed4/query_optimizer/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/query_optimizer/CMakeLists.txt b/query_optimizer/CMakeLists.txt
index 8333d4b..00d5163 100644
--- a/query_optimizer/CMakeLists.txt
+++ b/query_optimizer/CMakeLists.txt
@@ -41,6 +41,7 @@ add_subdirectory(tests)
# Declare micro-libs:
add_library(quickstep_queryoptimizer_ExecutionGenerator ExecutionGenerator.cpp ExecutionGenerator.hpp)
+add_library(quickstep_queryoptimizer_LIPFilterGenerator LIPFilterGenerator.cpp LIPFilterGenerator.hpp)
add_library(quickstep_queryoptimizer_LogicalGenerator LogicalGenerator.cpp LogicalGenerator.hpp)
add_library(quickstep_queryoptimizer_LogicalToPhysicalMapper
../empty_src.cpp
@@ -72,6 +73,7 @@ target_link_libraries(quickstep_queryoptimizer_ExecutionGenerator
quickstep_expressions_windowaggregation_WindowAggregateFunction_proto
quickstep_queryexecution_QueryContext
quickstep_queryexecution_QueryContext_proto
+ quickstep_queryoptimizer_LIPFilterGenerator
quickstep_queryoptimizer_OptimizerContext
quickstep_queryoptimizer_QueryHandle
quickstep_queryoptimizer_QueryPlan
@@ -99,6 +101,7 @@ target_link_libraries(quickstep_queryoptimizer_ExecutionGenerator
quickstep_queryoptimizer_physical_HashJoin
quickstep_queryoptimizer_physical_InsertSelection
quickstep_queryoptimizer_physical_InsertTuple
+ quickstep_queryoptimizer_physical_LIPFilterConfiguration
quickstep_queryoptimizer_physical_NestedLoopsJoin
quickstep_queryoptimizer_physical_PatternMatcher
quickstep_queryoptimizer_physical_Physical
@@ -152,6 +155,23 @@ if (ENABLE_DISTRIBUTED)
target_link_libraries(quickstep_queryoptimizer_ExecutionGenerator
quickstep_catalog_Catalog_proto)
endif()
+target_link_libraries(quickstep_queryoptimizer_LIPFilterGenerator
+ glog
+ quickstep_catalog_CatalogAttribute
+ quickstep_queryexecution_QueryContext
+ quickstep_queryexecution_QueryContext_proto
+ quickstep_queryoptimizer_QueryPlan
+ quickstep_queryoptimizer_expressions_ExprId
+ quickstep_queryoptimizer_physical_Aggregate
+ quickstep_queryoptimizer_physical_HashJoin
+ quickstep_queryoptimizer_physical_LIPFilterConfiguration
+ quickstep_queryoptimizer_physical_Physical
+ quickstep_queryoptimizer_physical_Selection
+ quickstep_relationaloperators_RelationalOperator
+ quickstep_types_Type
+ quickstep_utility_Macros
+ quickstep_utility_lipfilter_LIPFilter
+ quickstep_utility_lipfilter_LIPFilter_proto)
target_link_libraries(quickstep_queryoptimizer_LogicalGenerator
glog
quickstep_parser_ParseStatement
@@ -225,6 +245,7 @@ target_link_libraries(quickstep_queryoptimizer_Validator
add_library(quickstep_queryoptimizer ../empty_src.cpp QueryOptimizerModule.hpp)
target_link_libraries(quickstep_queryoptimizer
quickstep_queryoptimizer_ExecutionGenerator
+ quickstep_queryoptimizer_LIPFilterGenerator
quickstep_queryoptimizer_LogicalGenerator
quickstep_queryoptimizer_LogicalToPhysicalMapper
quickstep_queryoptimizer_Optimizer
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/8746eed4/query_optimizer/ExecutionGenerator.cpp
----------------------------------------------------------------------
diff --git a/query_optimizer/ExecutionGenerator.cpp b/query_optimizer/ExecutionGenerator.cpp
index 5a701b7..2e0d8f3 100644
--- a/query_optimizer/ExecutionGenerator.cpp
+++ b/query_optimizer/ExecutionGenerator.cpp
@@ -54,6 +54,7 @@
#include "expressions/window_aggregation/WindowAggregateFunction.pb.h"
#include "query_execution/QueryContext.hpp"
#include "query_execution/QueryContext.pb.h"
+#include "query_optimizer/LIPFilterGenerator.hpp"
#include "query_optimizer/OptimizerContext.hpp"
#include "query_optimizer/QueryHandle.hpp"
#include "query_optimizer/QueryPlan.hpp"
@@ -76,6 +77,7 @@
#include "query_optimizer/physical/HashJoin.hpp"
#include "query_optimizer/physical/InsertSelection.hpp"
#include "query_optimizer/physical/InsertTuple.hpp"
+#include "query_optimizer/physical/LIPFilterConfiguration.hpp"
#include "query_optimizer/physical/NestedLoopsJoin.hpp"
#include "query_optimizer/physical/PatternMatcher.hpp"
#include "query_optimizer/physical/Physical.hpp"
@@ -153,9 +155,6 @@ static const volatile bool aggregate_hashtable_type_dummy
DEFINE_bool(parallelize_load, true, "Parallelize loading data files.");
-DEFINE_bool(optimize_joins, false,
- "Enable post execution plan generation optimizations for joins.");
-
namespace E = ::quickstep::optimizer::expressions;
namespace P = ::quickstep::optimizer::physical;
namespace S = ::quickstep::serialization;
@@ -171,6 +170,12 @@ void ExecutionGenerator::generatePlan(const P::PhysicalPtr &physical_plan) {
cost_model_for_hash_join_.reset(
new cost::SimpleCostModel(top_level_physical_plan_->shared_subplans()));
+ const auto &lip_filter_configuration =
+ top_level_physical_plan_->lip_filter_configuration();
+ if (lip_filter_configuration != nullptr) {
+ lip_filter_generator_.reset(new LIPFilterGenerator(lip_filter_configuration));
+ }
+
const CatalogRelation *result_relation = nullptr;
try {
@@ -179,6 +184,11 @@ void ExecutionGenerator::generatePlan(const P::PhysicalPtr &physical_plan) {
}
generatePlanInternal(top_level_physical_plan_->plan());
+ // Deploy LIPFilters if enabled.
+ if (lip_filter_generator_ != nullptr) {
+ 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 =
@@ -235,6 +245,11 @@ void ExecutionGenerator::generatePlanInternal(
generatePlanInternal(child);
}
+ // If enabled, collect attribute substitution map for LIPFilterGenerator.
+ if (lip_filter_generator_ != nullptr) {
+ lip_filter_generator_->registerAttributeMap(physical_plan, attribute_substitution_map_);
+ }
+
switch (physical_plan->getPhysicalType()) {
case P::PhysicalType::kAggregate:
return convertAggregate(
@@ -566,6 +581,10 @@ void ExecutionGenerator::convertSelection(
std::forward_as_tuple(select_index,
output_relation));
temporary_relation_info_vec_.emplace_back(select_index, output_relation);
+
+ if (lip_filter_generator_ != nullptr) {
+ lip_filter_generator_->addSelectionInfo(physical_selection, select_index);
+ }
}
void ExecutionGenerator::convertSharedSubplanReference(const physical::SharedSubplanReferencePtr &physical_plan) {
@@ -794,6 +813,12 @@ void ExecutionGenerator::convertHashJoin(const P::HashJoinPtr &physical_plan) {
std::forward_as_tuple(join_operator_index,
output_relation));
temporary_relation_info_vec_.emplace_back(join_operator_index, output_relation);
+
+ if (lip_filter_generator_ != nullptr) {
+ lip_filter_generator_->addHashJoinInfo(physical_plan,
+ build_operator_index,
+ join_operator_index);
+ }
}
void ExecutionGenerator::convertNestedLoopsJoin(
@@ -1421,6 +1446,11 @@ void ExecutionGenerator::convertAggregate(
execution_plan_->addDirectDependency(destroy_aggregation_state_operator_index,
finalize_aggregation_operator_index,
true);
+
+ if (lip_filter_generator_ != nullptr) {
+ lip_filter_generator_->addAggregateInfo(physical_plan,
+ aggregation_operator_index);
+ }
}
void ExecutionGenerator::convertSort(const P::SortPtr &physical_sort) {
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/8746eed4/query_optimizer/ExecutionGenerator.hpp
----------------------------------------------------------------------
diff --git a/query_optimizer/ExecutionGenerator.hpp b/query_optimizer/ExecutionGenerator.hpp
index b7d8ef9..55197c9 100644
--- a/query_optimizer/ExecutionGenerator.hpp
+++ b/query_optimizer/ExecutionGenerator.hpp
@@ -33,6 +33,7 @@
#include "catalog/CatalogTypedefs.hpp"
#include "query_execution/QueryContext.hpp"
#include "query_execution/QueryContext.pb.h"
+#include "query_optimizer/LIPFilterGenerator.hpp"
#include "query_optimizer/QueryHandle.hpp"
#include "query_optimizer/QueryPlan.hpp"
#include "query_optimizer/cost_model/CostModel.hpp"
@@ -427,6 +428,9 @@ class ExecutionGenerator {
physical::TopLevelPlanPtr top_level_physical_plan_;
+ // Sub-generator for deploying LIP (lookahead information passing) filters.
+ std::unique_ptr<LIPFilterGenerator> lip_filter_generator_;
+
DISALLOW_COPY_AND_ASSIGN(ExecutionGenerator);
};
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/8746eed4/query_optimizer/LIPFilterGenerator.cpp
----------------------------------------------------------------------
diff --git a/query_optimizer/LIPFilterGenerator.cpp b/query_optimizer/LIPFilterGenerator.cpp
new file mode 100644
index 0000000..404037e
--- /dev/null
+++ b/query_optimizer/LIPFilterGenerator.cpp
@@ -0,0 +1,194 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ **/
+
+#include "query_optimizer/LIPFilterGenerator.hpp"
+
+#include <map>
+#include <utility>
+#include <vector>
+
+#include "catalog/CatalogAttribute.hpp"
+#include "query_execution/QueryContext.pb.h"
+#include "relational_operators/RelationalOperator.hpp"
+#include "types/Type.hpp"
+#include "utility/lip_filter/LIPFilter.hpp"
+#include "utility/lip_filter/LIPFilter.pb.h"
+
+#include "glog/logging.h"
+
+namespace quickstep {
+namespace optimizer {
+
+namespace E = ::quickstep::optimizer::expressions;
+namespace P = ::quickstep::optimizer::physical;
+
+void LIPFilterGenerator::registerAttributeMap(
+ const P::PhysicalPtr &node,
+ const std::unordered_map<E::ExprId, const CatalogAttribute *> &attribute_substitution_map) {
+ // Check if a builder is attached to node.
+ 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));
+ }
+ }
+ // Check if a prober is attached to node.
+ 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();
+ map_entry.emplace(attr_id, attribute_substitution_map.at(attr_id));
+ }
+ }
+}
+
+void LIPFilterGenerator::deployLIPFilters(QueryPlan *execution_plan,
+ serialization::QueryContext *query_context_proto) const {
+ LIPFilterBuilderMap lip_filter_builder_map;
+
+ // Deploy builders
+ const auto &build_info_map = lip_filter_configuration_->getBuildInfoMap();
+ for (const auto &info : builder_infos_) {
+ const auto build_it = build_info_map.find(info.builder_node);
+ if (build_it != build_info_map.end()) {
+ deployBuilderInternal(execution_plan,
+ query_context_proto,
+ info.builder_node,
+ info.builder_operator_index,
+ build_it->second,
+ &lip_filter_builder_map);
+ }
+ }
+
+ // Deploy probers
+ const auto &probe_info_map = lip_filter_configuration_->getProbeInfoMap();
+ for (const auto &info : prober_infos_) {
+ const auto probe_it = probe_info_map.find(info.prober_node);
+ if (probe_it != probe_info_map.end()) {
+ deployProberInteral(execution_plan,
+ query_context_proto,
+ info.prober_node,
+ info.prober_operator_index,
+ probe_it->second,
+ lip_filter_builder_map);
+ }
+ }
+}
+
+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 lip_deployment_index = query_context_proto->lip_filter_deployments_size();
+ auto *lip_filter_deployment_info_proto =
+ query_context_proto->add_lip_filter_deployments();
+ lip_filter_deployment_info_proto->set_action_type(serialization::LIPFilterActionType::BUILD);
+
+ const auto &builder_attribute_map = attribute_map_.at(builder_node);
+ for (const auto &info : build_info_vec) {
+ // Add the LIPFilter information into query context.
+ 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();
+ const CatalogAttribute *target_attr = builder_attribute_map.at(info.build_attribute->id());
+ const Type &attr_type = target_attr->getType();
+
+ switch (info.filter_type) {
+ case LIPFilterType::kSingleIdentityHashFilter: {
+ DCHECK(!attr_type.isVariableLength());
+ lip_filter_proto->set_lip_filter_type(
+ serialization::LIPFilterType::SINGLE_IDENTITY_HASH_FILTER);
+ lip_filter_proto->SetExtension(
+ serialization::SingleIdentityHashFilter::filter_cardinality, info.filter_cardinality);
+ lip_filter_proto->SetExtension(
+ serialization::SingleIdentityHashFilter::attribute_size, attr_type.minimumByteLength());
+ break;
+ }
+ default:
+ LOG(FATAL) << "Unsupported LIPFilter type";
+ break;
+ }
+
+ // Register the builder information which is needed later by the probers.
+ lip_filter_builder_map->emplace(
+ std::make_pair(info.build_attribute->id(), builder_node),
+ std::make_pair(lip_filter_id, builder_operator_index));
+
+ // Add the builder deployment information into query context.
+ auto *lip_filter_entry_proto = lip_filter_deployment_info_proto->add_entries();
+ lip_filter_entry_proto->set_lip_filter_id(lip_filter_id);
+ lip_filter_entry_proto->set_attribute_id(target_attr->getID());
+ lip_filter_entry_proto->mutable_attribute_type()->CopyFrom(attr_type.getProto());
+ }
+
+ // Attach the LIPFilterDeployment information to the RelationalOperator.
+ RelationalOperator *relop =
+ execution_plan->getQueryPlanDAGMutable()->getNodePayloadMutable(builder_operator_index);
+ relop->deployLIPFilters(lip_deployment_index);
+}
+
+void LIPFilterGenerator::deployProberInteral(
+ QueryPlan *execution_plan,
+ serialization::QueryContext *query_context_proto,
+ const physical::PhysicalPtr &prober_node,
+ const QueryPlan::DAGNodeIndex prober_operator_index,
+ const std::vector<physical::LIPFilterProbeInfo> &probe_info_vec,
+ const LIPFilterBuilderMap &lip_filter_builder_map) const {
+ const auto lip_deployment_index = query_context_proto->lip_filter_deployments_size();
+ auto *lip_filter_deployment_info_proto =
+ query_context_proto->add_lip_filter_deployments();
+ lip_filter_deployment_info_proto->set_action_type(serialization::LIPFilterActionType::PROBE);
+
+ const auto &prober_attribute_map = attribute_map_.at(prober_node);
+ for (const auto &info : probe_info_vec) {
+ // Find the corresponding builder for the to-be-probed LIPFilter.
+ const auto &builder_info =
+ lip_filter_builder_map.at(
+ std::make_pair(info.build_attribute->id(), info.builder));
+ const CatalogAttribute *target_attr = prober_attribute_map.at(info.probe_attribute->id());
+
+ // Add the prober deployment information into query context.
+ auto *lip_filter_entry_proto = lip_filter_deployment_info_proto->add_entries();
+ lip_filter_entry_proto->set_lip_filter_id(builder_info.first);
+ lip_filter_entry_proto->set_attribute_id(target_attr->getID());
+ lip_filter_entry_proto->mutable_attribute_type()->CopyFrom(
+ target_attr->getType().getProto());
+
+ // A prober must wait until the corresponding builder has completed building
+ // the LIPFilter.
+ execution_plan->addOrUpgradeDirectDependency(prober_operator_index,
+ builder_info.second,
+ true /* is_pipeline_breaker */);
+ }
+
+ // Attach the LIPFilterDeployment information to the RelationalOperator.
+ RelationalOperator *relop =
+ execution_plan->getQueryPlanDAGMutable()->getNodePayloadMutable(prober_operator_index);
+ relop->deployLIPFilters(lip_deployment_index);
+}
+
+} // namespace optimizer
+} // namespace quickstep
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/8746eed4/query_optimizer/LIPFilterGenerator.hpp
----------------------------------------------------------------------
diff --git a/query_optimizer/LIPFilterGenerator.hpp b/query_optimizer/LIPFilterGenerator.hpp
new file mode 100644
index 0000000..9d191a1
--- /dev/null
+++ b/query_optimizer/LIPFilterGenerator.hpp
@@ -0,0 +1,194 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ **/
+
+#ifndef QUICKSTEP_QUERY_OPTIMIZER_LIP_FILTER_GENERATOR_HPP_
+#define QUICKSTEP_QUERY_OPTIMIZER_LIP_FILTER_GENERATOR_HPP_
+
+#include <map>
+#include <unordered_map>
+#include <utility>
+#include <vector>
+
+#include "query_execution/QueryContext.hpp"
+#include "query_optimizer/QueryPlan.hpp"
+#include "query_optimizer/expressions/ExprId.hpp"
+#include "query_optimizer/physical/LIPFilterConfiguration.hpp"
+#include "query_optimizer/physical/Aggregate.hpp"
+#include "query_optimizer/physical/HashJoin.hpp"
+#include "query_optimizer/physical/Physical.hpp"
+#include "query_optimizer/physical/Selection.hpp"
+#include "utility/Macros.hpp"
+
+#include "glog/logging.h"
+
+namespace quickstep {
+
+namespace serialization { class QueryContext; }
+
+class CatalogAttribute;
+
+namespace optimizer {
+
+/** \addtogroup QueryOptimizer
+ * @{
+ */
+
+/**
+ * @brief Generates backend LIPFilter deployments from physical plan's information.
+ */
+class LIPFilterGenerator {
+ public:
+ /**
+ * @brief Constructor.
+ *
+ * @param lip_filter_configuration The LIPFilter configuration information
+ * generated by physical optimizer.
+ */
+ explicit LIPFilterGenerator(
+ const physical::LIPFilterConfigurationPtr &lip_filter_configuration)
+ : lip_filter_configuration_(lip_filter_configuration) {
+ DCHECK(lip_filter_configuration_ != nullptr);
+ }
+
+ /**
+ * @brief Collect the ExprId to CatalogAttribute mapping information for the
+ * given physical node.
+ *
+ * @param node A physical plan node.
+ * @param attribute_substitution_map A map that maps each ExprId to the
+ * backend relation's CatalogAttribute's.
+ */
+ void registerAttributeMap(
+ const physical::PhysicalPtr &node,
+ const std::unordered_map<expressions::ExprId, const CatalogAttribute *> &attribute_substitution_map);
+
+ /**
+ * @brief Add physical-to-execution mapping information for deploying LIPFilters
+ * to an aggregation.
+ *
+ * @param aggregate A physical Aggregate node.
+ * @param aggregate_operator_index The index of the AggregationOperator that
+ * corresponds to \p aggregate in the execution plan.
+ */
+ void addAggregateInfo(const physical::AggregatePtr &aggregate,
+ const QueryPlan::DAGNodeIndex aggregate_operator_index) {
+ prober_infos_.emplace_back(aggregate, aggregate_operator_index);
+ }
+
+ /**
+ * @brief Add physical-to-execution mapping information for deploying LIPFilters
+ * to a hash-join.
+ *
+ * @param hash_join A physical HashJoin node.
+ * @param build_operator_index The index of the BuildHashOperator that corresponds
+ * to \p hash_join in the execution plan.
+ * @param join_operator_index The index of the HashJoinOperator that corresponds
+ * to \p hash_join in the execution plan.
+ */
+ void addHashJoinInfo(const physical::HashJoinPtr &hash_join,
+ const QueryPlan::DAGNodeIndex build_operator_index,
+ const QueryPlan::DAGNodeIndex join_operator_index) {
+ builder_infos_.emplace_back(hash_join, build_operator_index);
+ prober_infos_.emplace_back(hash_join, join_operator_index);
+ }
+
+ /**
+ * @brief Add physical-to-execution mapping information for deploying LIPFilters
+ * to a selection.
+ *
+ * @param selection A physical Selection node.
+ * @param select_operator_index The index of the SelectOperator that corresponds
+ * to \p selection in the execution plan.
+ */
+ void addSelectionInfo(const physical::SelectionPtr &selection,
+ const QueryPlan::DAGNodeIndex select_operator_index) {
+ prober_infos_.emplace_back(selection, select_operator_index);
+ }
+
+ /**
+ * @brief Deploy the LIPFilters to the execution plan.
+ *
+ * @param execution_plan The execution plan.
+ * @param query_context_proto QueryContext protobuf for the execution plan.
+ */
+ void deployLIPFilters(QueryPlan *execution_plan,
+ serialization::QueryContext *query_context_proto) const;
+
+ private:
+ /**
+ * @brief Internal data structure for representing a LIPFilter builder.
+ */
+ struct BuilderInfo {
+ BuilderInfo(const physical::PhysicalPtr &builder_node_in,
+ const QueryPlan::DAGNodeIndex builder_operator_index_in)
+ : builder_node(builder_node_in),
+ builder_operator_index(builder_operator_index_in) {
+ }
+ const physical::PhysicalPtr builder_node;
+ const QueryPlan::DAGNodeIndex builder_operator_index;
+ };
+
+ /**
+ * @brief Internal data structure for representing a LIPFilter prober.
+ */
+ struct ProberInfo {
+ ProberInfo(const physical::PhysicalPtr &prober_node_in,
+ const QueryPlan::DAGNodeIndex prober_operator_index_in)
+ : prober_node(prober_node_in),
+ prober_operator_index(prober_operator_index_in) {
+ }
+ const physical::PhysicalPtr prober_node;
+ const QueryPlan::DAGNodeIndex prober_operator_index;
+ };
+
+ // Maps each LIPFilter's building attribute to the LIPFilter's id in QueryContext
+ // as well as the LIPFilter's building relational operator's 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(QueryPlan *execution_plan,
+ serialization::QueryContext *query_context_proto,
+ const physical::PhysicalPtr &prober_node,
+ const QueryPlan::DAGNodeIndex prober_operator_index,
+ const std::vector<physical::LIPFilterProbeInfo> &probe_info_vec,
+ const LIPFilterBuilderMap &lip_filter_builder_map) const;
+
+ const physical::LIPFilterConfigurationPtr lip_filter_configuration_;
+
+ std::vector<BuilderInfo> builder_infos_;
+ std::vector<ProberInfo> prober_infos_;
+
+ std::map<physical::PhysicalPtr, std::map<expressions::ExprId, const CatalogAttribute *>> attribute_map_;
+
+ DISALLOW_COPY_AND_ASSIGN(LIPFilterGenerator);
+};
+
+/** @} */
+
+} // namespace optimizer
+} // namespace quickstep
+
+#endif /* QUICKSTEP_QUERY_OPTIMIZER_LIP_FILTER_GENERATOR_HPP_ */
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/8746eed4/query_optimizer/QueryPlan.hpp
----------------------------------------------------------------------
diff --git a/query_optimizer/QueryPlan.hpp b/query_optimizer/QueryPlan.hpp
index 5cd174c..ef6dff4 100644
--- a/query_optimizer/QueryPlan.hpp
+++ b/query_optimizer/QueryPlan.hpp
@@ -74,6 +74,38 @@ class QueryPlan {
}
/**
+ * @brief Creates a link or upgrades the existing link from \p producer_operator_index
+ * to \p consumer_operator_index in the DAG.
+ *
+ * Depending on whether there is an existing link from \p producer_operator_index
+ * to \p consumer_operator_index:
+ * - Case 1, no existing link:
+ * Creates a link with metadata set to is_pipeline_breaker.
+ * - Case 2, existing link with metadata \p m:
+ * Set m = (m | is_pipeline_break).
+ *
+ * @param consumer_operator_index The index of the consumer operator.
+ * @param producer_operator_index The index of the producer operator.
+ * @param is_pipeline_breaker True if the result from the producer cannot be
+ * pipelined to the consumer, otherwise false.
+ */
+ inline void addOrUpgradeDirectDependency(DAGNodeIndex consumer_operator_index,
+ DAGNodeIndex producer_operator_index,
+ bool is_pipeline_breaker) {
+ const auto &dependents = dag_operators_.getDependents(producer_operator_index);
+ const auto consumer_it = dependents.find(consumer_operator_index);
+ if (consumer_it == dependents.end()) {
+ dag_operators_.createLink(producer_operator_index,
+ consumer_operator_index,
+ is_pipeline_breaker);
+ } else {
+ dag_operators_.setLinkMetadata(producer_operator_index,
+ consumer_operator_index,
+ consumer_it->second | is_pipeline_breaker);
+ }
+ }
+
+ /**
* @brief Creates dependencies for a DropTable operator with index
* \p drop_operator_index. If \p producer_operator_index
* has any dependent, creates a link from \p drop_operator_index
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/8746eed4/relational_operators/RelationalOperator.hpp
----------------------------------------------------------------------
diff --git a/relational_operators/RelationalOperator.hpp b/relational_operators/RelationalOperator.hpp
index f0303e5..3eea189 100644
--- a/relational_operators/RelationalOperator.hpp
+++ b/relational_operators/RelationalOperator.hpp
@@ -245,6 +245,13 @@ class RelationalOperator {
return op_index_;
}
+ /**
+ * @brief Deploy a group of LIPFilters to this operator.
+ */
+ void deployLIPFilters(const QueryContext::lip_deployment_id lip_deployment_index) {
+ lip_deployment_index_ = lip_deployment_index;
+ }
+
protected:
/**
* @brief Constructor
@@ -257,7 +264,8 @@ class RelationalOperator {
const bool blocking_dependencies_met = false)
: query_id_(query_id),
blocking_dependencies_met_(blocking_dependencies_met),
- done_feeding_input_relation_(false) {}
+ done_feeding_input_relation_(false),
+ lip_deployment_index_(QueryContext::kInvalidLIPDeploymentId) {}
const std::size_t query_id_;
@@ -265,6 +273,8 @@ class RelationalOperator {
bool done_feeding_input_relation_;
std::size_t op_index_;
+ QueryContext::lip_deployment_id lip_deployment_index_;
+
private:
DISALLOW_COPY_AND_ASSIGN(RelationalOperator);
};
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/8746eed4/utility/lip_filter/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/utility/lip_filter/CMakeLists.txt b/utility/lip_filter/CMakeLists.txt
index 2232abe..b7224d2 100644
--- a/utility/lip_filter/CMakeLists.txt
+++ b/utility/lip_filter/CMakeLists.txt
@@ -15,5 +15,58 @@
# specific language governing permissions and limitations
# under the License.
+QS_PROTOBUF_GENERATE_CPP(utility_lipfilter_LIPFilter_proto_srcs
+ utility_lipfilter_LIPFilter_proto_hdrs
+ LIPFilter.proto)
+
# Declare micro-libs:
-add_library(quickstep_utility_lipfilter_LIPFilter ../../empty_src.cpp LIPFilter.hpp)
\ No newline at end of file
+add_library(quickstep_utility_lipfilter_LIPFilter ../../empty_src.cpp LIPFilter.hpp)
+add_library(quickstep_utility_lipfilter_LIPFilterAdaptiveProber ../../empty_src.cpp LIPFilterAdaptiveProber.hpp)
+add_library(quickstep_utility_lipfilter_LIPFilterBuilder ../../empty_src.cpp LIPFilterBuilder.hpp)
+add_library(quickstep_utility_lipfilter_LIPFilterDeployment LIPFilterDeployment.cpp LIPFilterDeployment.hpp)
+add_library(quickstep_utility_lipfilter_LIPFilterFactory LIPFilterFactory.cpp LIPFilterFactory.hpp)
+add_library(quickstep_utility_lipfilter_LIPFilter_proto
+ ${utility_lipfilter_LIPFilter_proto_srcs})
+add_library(quickstep_utility_lipfilter_SingleIdentityHashFilter ../../empty_src.cpp SingleIdentityHashFilter.hpp)
+
+# Link dependencies:
+target_link_libraries(quickstep_utility_lipfilter_LIPFilter
+ quickstep_catalog_CatalogTypedefs
+ quickstep_storage_StorageBlockInfo
+ quickstep_utility_Macros)
+target_link_libraries(quickstep_utility_lipfilter_LIPFilterAdaptiveProber
+ quickstep_catalog_CatalogTypedefs
+ quickstep_storage_StorageBlockInfo
+ quickstep_storage_TupleIdSequence
+ quickstep_storage_ValueAccessor
+ quickstep_types_Type
+ quickstep_utility_Macros
+ quickstep_utility_lipfilter_LIPFilter)
+target_link_libraries(quickstep_utility_lipfilter_LIPFilterBuilder
+ quickstep_catalog_CatalogTypedefs
+ quickstep_utility_Macros
+ quickstep_utility_lipfilter_LIPFilter)
+target_link_libraries(quickstep_utility_lipfilter_LIPFilterDeployment
+ quickstep_catalog_CatalogTypedefs
+ quickstep_types_TypeFactory
+ quickstep_utility_Macros
+ quickstep_utility_lipfilter_LIPFilter
+ quickstep_utility_lipfilter_LIPFilterAdaptiveProber
+ quickstep_utility_lipfilter_LIPFilterBuilder
+ quickstep_utility_lipfilter_LIPFilter_proto)
+target_link_libraries(quickstep_utility_lipfilter_LIPFilterFactory
+ quickstep_utility_lipfilter_LIPFilter_proto
+ quickstep_utility_lipfilter_SingleIdentityHashFilter
+ quickstep_utility_Macros)
+target_link_libraries(quickstep_utility_lipfilter_LIPFilter_proto
+ ${PROTOBUF_LIBRARY}
+ quickstep_types_Type_proto)
+target_link_libraries(quickstep_utility_lipfilter_SingleIdentityHashFilter
+ quickstep_catalog_CatalogTypedefs
+ quickstep_storage_StorageBlockInfo
+ quickstep_storage_StorageConstants
+ quickstep_storage_ValueAccessor
+ quickstep_storage_ValueAccessorUtil
+ quickstep_types_Type
+ quickstep_utility_lipfilter_LIPFilter
+ quickstep_utility_Macros)
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/8746eed4/utility/lip_filter/LIPFilter.hpp
----------------------------------------------------------------------
diff --git a/utility/lip_filter/LIPFilter.hpp b/utility/lip_filter/LIPFilter.hpp
index 33165ed..682d69f 100644
--- a/utility/lip_filter/LIPFilter.hpp
+++ b/utility/lip_filter/LIPFilter.hpp
@@ -20,8 +20,18 @@
#ifndef QUICKSTEP_UTILITY_LIP_FILTER_LIP_FILTER_HPP_
#define QUICKSTEP_UTILITY_LIP_FILTER_LIP_FILTER_HPP_
+#include <cstddef>
+#include <vector>
+
+#include "catalog/CatalogTypedefs.hpp"
+#include "storage/StorageBlockInfo.hpp"
+#include "utility/Macros.hpp"
+
namespace quickstep {
+class Type;
+class ValueAccessor;
+
/** \addtogroup Utility
* @{
*/
@@ -32,6 +42,60 @@ enum class LIPFilterType {
kSingleIdentityHashFilter
};
+/**
+ * @brief Base class for LIP (Lookahead Information Passing) filters.
+ */
+class LIPFilter {
+ public:
+ /**
+ * @breif Get the type of this LIPFilter.
+ *
+ * @return The type of this LIPFilter.
+ */
+ LIPFilterType getType() const {
+ return type_;
+ }
+
+ /**
+ * @brief Insert the values drawn from a ValueAccessor into this LIPFilter.
+ *
+ * @param accessor A ValueAccessor which will be used to access the values.
+ * @param attr_id The attribute id of the values to be read from accessor.
+ * @param attr_type The type of the values.
+ */
+ virtual void insertValueAccessor(ValueAccessor *accessor,
+ const attribute_id attr_id,
+ const Type *attr_type) = 0;
+
+ /**
+ * @brief Filter the given batch of tuples from a ValueAccessor. Remove any
+ * tuple in the batch that does not have a hit in this filter.
+ *
+ * @param accessor A ValueAccessor which will be used to access the tuples.
+ * @param attr_id The attribute id of the values to be filtered.
+ * @param is_attr_nullable Whether the values can be NULL.
+ * @param batch The batch of tuple ids to be filtered. This vector will also
+ * be updated in place in this method to hold the output tuple ids.
+ * @param batch_size The input batch size.
+ *
+ * @return The output batch size.
+ */
+ virtual std::size_t filterBatch(ValueAccessor *accessor,
+ const attribute_id attr_id,
+ const bool is_attr_nullable,
+ std::vector<tuple_id> *batch,
+ const std::size_t batch_size) const = 0;
+
+ protected:
+ explicit LIPFilter(const LIPFilterType &type)
+ : type_(type) {}
+
+ private:
+ LIPFilterType type_;
+
+ DISALLOW_COPY_AND_ASSIGN(LIPFilter);
+};
+
/** @} */
} // namespace quickstep
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/8746eed4/utility/lip_filter/LIPFilter.proto
----------------------------------------------------------------------
diff --git a/utility/lip_filter/LIPFilter.proto b/utility/lip_filter/LIPFilter.proto
new file mode 100644
index 0000000..def13dd
--- /dev/null
+++ b/utility/lip_filter/LIPFilter.proto
@@ -0,0 +1,58 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+syntax = "proto2";
+
+package quickstep.serialization;
+
+import "types/Type.proto";
+
+enum LIPFilterType {
+ BLOOM_FILTER = 1;
+ EXACT_FILTER = 2;
+ SINGLE_IDENTITY_HASH_FILTER = 3;
+}
+
+message LIPFilter {
+ required LIPFilterType lip_filter_type = 1;
+
+ extensions 16 to max;
+}
+
+message SingleIdentityHashFilter {
+ extend LIPFilter {
+ // All required
+ optional uint64 filter_cardinality = 16;
+ optional uint64 attribute_size = 17;
+ }
+}
+
+enum LIPFilterActionType {
+ BUILD = 1;
+ PROBE = 2;
+}
+
+message LIPFilterDeployment {
+ message Entry {
+ required uint32 lip_filter_id = 1;
+ required int32 attribute_id = 2;
+ required Type attribute_type = 3;
+ }
+
+ required LIPFilterActionType action_type = 1;
+ repeated Entry entries = 2;
+}
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/8746eed4/utility/lip_filter/LIPFilterAdaptiveProber.hpp
----------------------------------------------------------------------
diff --git a/utility/lip_filter/LIPFilterAdaptiveProber.hpp b/utility/lip_filter/LIPFilterAdaptiveProber.hpp
new file mode 100644
index 0000000..e1a75d6
--- /dev/null
+++ b/utility/lip_filter/LIPFilterAdaptiveProber.hpp
@@ -0,0 +1,243 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ **/
+
+#ifndef QUICKSTEP_UTILITY_LIP_FILTER_LIP_FILTER_ADAPTIVE_PROBER_HPP_
+#define QUICKSTEP_UTILITY_LIP_FILTER_LIP_FILTER_ADAPTIVE_PROBER_HPP_
+
+#include <algorithm>
+#include <cstddef>
+#include <cstdint>
+#include <memory>
+#include <vector>
+
+#include "catalog/CatalogTypedefs.hpp"
+#include "storage/StorageBlockInfo.hpp"
+#include "storage/TupleIdSequence.hpp"
+#include "storage/ValueAccessor.hpp"
+#include "types/Type.hpp"
+#include "utility/Macros.hpp"
+#include "utility/lip_filter/LIPFilter.hpp"
+
+#include "glog/logging.h"
+
+namespace quickstep {
+
+/** \addtogroup Utility
+ * @{
+ */
+
+/**
+ * @brief Helper class for adaptively applying a group of LIPFilters to a
+ * ValueAccessor. Here "adaptive" means that the application ordering
+ * of the filters will be adjusted on the fly based on the filters' miss
+ * rates.
+ */
+class LIPFilterAdaptiveProber {
+ public:
+ /**
+ * @brief Constructor.
+ *
+ * @param lip_filters The LIPFilters that will be probed.
+ * @param The target attribute ids for the LIPFilters.
+ * @param The target attribute types for the LIPFilters.
+ */
+ LIPFilterAdaptiveProber(const std::vector<LIPFilter *> &lip_filters,
+ const std::vector<attribute_id> &attr_ids,
+ const std::vector<const Type *> &attr_types) {
+ DCHECK_EQ(lip_filters.size(), attr_ids.size());
+ DCHECK_EQ(lip_filters.size(), attr_types.size());
+
+ probe_entries_.reserve(lip_filters.size());
+ for (std::size_t i = 0; i < lip_filters.size(); ++i) {
+ DCHECK(lip_filters[i] != nullptr);
+ probe_entries_.emplace_back(
+ new ProbeEntry(lip_filters[i], attr_ids[i], attr_types[i]));
+ }
+ }
+
+ /**
+ * @brief Destructor.
+ */
+ ~LIPFilterAdaptiveProber() {
+ for (ProbeEntry *entry : probe_entries_) {
+ delete entry;
+ }
+ }
+
+ /**
+ * @brief Apply this group of LIPFilters to the given ValueAccessor.
+ *
+ * @param accessor A ValueAccessor to be filtered.
+ * @return A TupleIdSequence for the hit tuples in the ValueAccessor.
+ */
+ TupleIdSequence* filterValueAccessor(ValueAccessor *accessor) {
+ const TupleIdSequence *existence_map = accessor->getTupleIdSequenceVirtual();
+ if (existence_map == nullptr) {
+ return filterValueAccessorNoExistenceMap(accessor);
+ } else {
+ return filterValueAccessorWithExistenceMap(accessor, existence_map);
+ }
+ }
+
+ private:
+ /**
+ * @brief Internal data structure for representing each LIPFilter probing entry.
+ */
+ struct ProbeEntry {
+ ProbeEntry(const LIPFilter *lip_filter_in,
+ const attribute_id attr_id_in,
+ const Type *attr_type_in)
+ : lip_filter(lip_filter_in),
+ attr_id(attr_id_in),
+ attr_type(attr_type_in),
+ miss(0),
+ cnt(0) {
+ }
+
+ /**
+ * @brief Whether a LIPFilter is more selective than the other.
+ */
+ static bool isBetterThan(const ProbeEntry *a,
+ const ProbeEntry *b) {
+ return a->miss_rate > b->miss_rate;
+ }
+
+ const LIPFilter *lip_filter;
+ const attribute_id attr_id;
+ const Type *attr_type;
+ std::uint32_t miss;
+ std::uint32_t cnt;
+ float miss_rate;
+ };
+
+ /**
+ * @brief Sepecialized filterValueAccessor implementation where the given
+ * ValueAccessor has no existence map.
+ */
+ inline TupleIdSequence* filterValueAccessorNoExistenceMap(ValueAccessor *accessor) {
+ const std::uint32_t num_tuples = accessor->getNumTuplesVirtual();
+ std::unique_ptr<TupleIdSequence> matches(new TupleIdSequence(num_tuples));
+ std::uint32_t next_batch_size = 64u;
+ std::vector<tuple_id> batch(num_tuples);
+
+ // Apply the filters in a batched manner.
+ std::uint32_t batch_start = 0;
+ do {
+ const std::uint32_t batch_size =
+ std::min(next_batch_size, num_tuples - batch_start);
+ for (std::uint32_t i = 0; i < batch_size; ++i) {
+ batch[i] = batch_start + i;
+ }
+
+ const std::uint32_t num_hits = filterBatch(accessor, &batch, batch_size);
+ for (std::uint32_t i = 0; i < num_hits; ++i) {
+ matches->set(batch[i], true);
+ }
+
+ batch_start += batch_size;
+ next_batch_size *= 2;
+ } while (batch_start < num_tuples);
+
+ return matches.release();
+ }
+
+ /**
+ * @brief Sepecialized filterValueAccessor implementation where the given
+ * ValueAccessor has an existence map.
+ */
+ inline TupleIdSequence* filterValueAccessorWithExistenceMap(ValueAccessor *accessor,
+ const TupleIdSequence *existence_map) {
+ std::unique_ptr<TupleIdSequence> matches(
+ new TupleIdSequence(existence_map->length()));
+ std::uint32_t next_batch_size = 64u;
+ std::uint32_t num_tuples_left = existence_map->numTuples();
+ std::vector<tuple_id> batch(num_tuples_left);
+
+ // Apply the filters in a batched manner.
+ TupleIdSequence::const_iterator tuple_it = existence_map->before_begin();
+ do {
+ const std::uint32_t batch_size =
+ next_batch_size < num_tuples_left ? next_batch_size : num_tuples_left;
+ for (std::uint32_t i = 0; i < batch_size; ++i) {
+ ++tuple_it;
+ batch[i] = *tuple_it;
+ }
+
+ const std::uint32_t num_hits = filterBatch(accessor, &batch, batch_size);
+ for (std::uint32_t i = 0; i < num_hits; ++i) {
+ matches->set(batch[i], true);
+ }
+
+ num_tuples_left -= batch_size;
+ next_batch_size *= 2;
+ } while (num_tuples_left > 0);
+
+ return matches.release();
+ }
+
+ /**
+ * @brief Filter the given batch of tuples from the ValueAccessor. Remove any
+ * tuple in the batch that misses any filter.
+ */
+ inline std::size_t filterBatch(ValueAccessor *accessor,
+ std::vector<tuple_id> *batch,
+ std::uint32_t batch_size) {
+ // Apply the LIPFilters one by one to the batch and update corresponding
+ // cnt/miss statistics.
+ for (auto *entry : probe_entries_) {
+ const std::uint32_t out_size =
+ entry->lip_filter->filterBatch(accessor,
+ entry->attr_id,
+ entry->attr_type->isNullable(),
+ batch,
+ batch_size);
+ entry->cnt += batch_size;
+ entry->miss += batch_size - out_size;
+ batch_size = out_size;
+ }
+
+ // Adaptively adjust the application ordering after each batch.
+ adaptEntryOrder();
+
+ return batch_size;
+ }
+
+ /**
+ * @brief Adjust LIPFilter application ordering with regard to their miss
+ * rates (i.e. selectivites).
+ */
+ inline void adaptEntryOrder() {
+ for (auto &entry : probe_entries_) {
+ entry->miss_rate = static_cast<float>(entry->miss) / entry->cnt;
+ }
+ std::sort(probe_entries_.begin(),
+ probe_entries_.end(),
+ ProbeEntry::isBetterThan);
+ }
+
+ std::vector<ProbeEntry *> probe_entries_;
+
+ DISALLOW_COPY_AND_ASSIGN(LIPFilterAdaptiveProber);
+};
+
+/** @} */
+
+} // namespace quickstep
+
+#endif // QUICKSTEP_UTILITY_LIP_FILTER_LIP_FILTER_ADAPTIVE_PROBER_HPP_
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/8746eed4/utility/lip_filter/LIPFilterBuilder.hpp
----------------------------------------------------------------------
diff --git a/utility/lip_filter/LIPFilterBuilder.hpp b/utility/lip_filter/LIPFilterBuilder.hpp
new file mode 100644
index 0000000..deb8f66
--- /dev/null
+++ b/utility/lip_filter/LIPFilterBuilder.hpp
@@ -0,0 +1,109 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ **/
+
+#ifndef QUICKSTEP_UTILITY_LIP_FILTER_LIP_FILTER_BUILDER_HPP_
+#define QUICKSTEP_UTILITY_LIP_FILTER_LIP_FILTER_BUILDER_HPP_
+
+#include <cstddef>
+#include <memory>
+#include <vector>
+
+#include "catalog/CatalogTypedefs.hpp"
+#include "utility/Macros.hpp"
+#include "utility/lip_filter/LIPFilter.hpp"
+
+#include "glog/logging.h"
+
+namespace quickstep {
+
+class Type;
+class ValueAccessor;
+
+/** \addtogroup Utility
+ * @{
+ */
+
+class LIPFilterBuilder;
+typedef std::shared_ptr<LIPFilterBuilder> LIPFilterBuilderPtr;
+
+/**
+ * @brief Helper class for building LIPFilters from a relation (i.e. ValueAccessor).
+ */
+class LIPFilterBuilder {
+ public:
+ /**
+ * @brief Constructor.
+ *
+ * @param lip_filters The LIPFilters that will be built.
+ * @param attr_ids The target attribute ids for the LIPFilters.
+ * @param attr_types The target attribute types for the LIPFilters.
+ */
+ LIPFilterBuilder(const std::vector<LIPFilter *> &lip_filters,
+ const std::vector<attribute_id> &attr_ids,
+ const std::vector<const Type *> &attr_types) {
+ DCHECK_EQ(lip_filters.size(), attr_ids.size());
+ DCHECK_EQ(lip_filters.size(), attr_types.size());
+
+ build_entries_.reserve(lip_filters.size());
+ for (std::size_t i = 0; i < lip_filters.size(); ++i) {
+ build_entries_.emplace_back(lip_filters[i], attr_ids[i], attr_types[i]);
+ }
+ }
+
+ /**
+ * @brief Insert all the values from the given ValueAccessor into the attached
+ * LIPFilters with regard to each target attribute id in \p attr_ids_.
+ *
+ * @param accessor The ValueAccessor which will be used to access the values.
+ */
+ void insertValueAccessor(ValueAccessor *accessor) {
+ for (auto &entry : build_entries_) {
+ entry.lip_filter->insertValueAccessor(accessor,
+ entry.attr_id,
+ entry.attr_type);
+ }
+ }
+
+ private:
+ /**
+ * @brief Internal data structure for representing each LIPFilter building entry.
+ */
+ struct BuildEntry {
+ BuildEntry(LIPFilter *lip_filter_in,
+ const attribute_id attr_id_in,
+ const Type *attr_type_in)
+ : lip_filter(lip_filter_in),
+ attr_id(attr_id_in),
+ attr_type(attr_type_in) {
+ }
+ LIPFilter *lip_filter;
+ const attribute_id attr_id;
+ const Type *attr_type;
+ };
+
+ std::vector<BuildEntry> build_entries_;
+
+ DISALLOW_COPY_AND_ASSIGN(LIPFilterBuilder);
+};
+
+/** @} */
+
+} // namespace quickstep
+
+#endif // QUICKSTEP_UTILITY_LIP_FILTER_LIP_FILTER_BUILDER_HPP_
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/8746eed4/utility/lip_filter/LIPFilterDeployment.cpp
----------------------------------------------------------------------
diff --git a/utility/lip_filter/LIPFilterDeployment.cpp b/utility/lip_filter/LIPFilterDeployment.cpp
new file mode 100644
index 0000000..cd4d90f
--- /dev/null
+++ b/utility/lip_filter/LIPFilterDeployment.cpp
@@ -0,0 +1,87 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ **/
+
+#include "utility/lip_filter/LIPFilterDeployment.hpp"
+
+#include <memory>
+#include <vector>
+
+#include "types/TypeFactory.hpp"
+#include "utility/lip_filter/LIPFilter.hpp"
+#include "utility/lip_filter/LIPFilter.pb.h"
+#include "utility/lip_filter/LIPFilterBuilder.hpp"
+#include "utility/lip_filter/LIPFilterAdaptiveProber.hpp"
+
+#include "glog/logging.h"
+
+namespace quickstep {
+
+LIPFilterDeployment::LIPFilterDeployment(
+ const serialization::LIPFilterDeployment &proto,
+ const std::vector<std::unique_ptr<LIPFilter>> &lip_filters) {
+ switch (proto.action_type()) {
+ case serialization::LIPFilterActionType::BUILD:
+ action_type_ = LIPFilterActionType::kBuild;
+ break;
+ case serialization::LIPFilterActionType::PROBE:
+ action_type_ = LIPFilterActionType::kProbe;
+ break;
+ default:
+ LOG(FATAL) << "Unsupported LIPFilterActionType: "
+ << serialization::LIPFilterActionType_Name(proto.action_type());
+ }
+
+ for (int i = 0; i < proto.entries_size(); ++i) {
+ const auto &entry_proto = proto.entries(i);
+ lip_filters_.emplace_back(lip_filters.at(entry_proto.lip_filter_id()).get());
+ attr_ids_.emplace_back(entry_proto.attribute_id());
+ attr_types_.emplace_back(&TypeFactory::ReconstructFromProto(entry_proto.attribute_type()));
+ }
+}
+
+bool LIPFilterDeployment::ProtoIsValid(
+ const serialization::LIPFilterDeployment &proto) {
+ if (proto.action_type() != serialization::LIPFilterActionType::BUILD &&
+ proto.action_type() != serialization::LIPFilterActionType::PROBE) {
+ LOG(FATAL) << "Unsupported LIPFilterActionType: "
+ << serialization::LIPFilterActionType_Name(proto.action_type());
+ }
+ if (proto.entries_size() == 0) {
+ return false;
+ }
+ for (int i = 0; i < proto.entries_size(); ++i) {
+ const auto &entry_proto = proto.entries(i);
+ if (!TypeFactory::ProtoIsValid(entry_proto.attribute_type())) {
+ return false;
+ }
+ }
+ return true;
+}
+
+LIPFilterBuilder* LIPFilterDeployment::createLIPFilterBuilder() const {
+ DCHECK(action_type_ == LIPFilterActionType::kBuild);
+ return new LIPFilterBuilder(lip_filters_, attr_ids_, attr_types_);
+}
+
+LIPFilterAdaptiveProber* LIPFilterDeployment::createLIPFilterAdaptiveProber() const {
+ DCHECK(action_type_ == LIPFilterActionType::kProbe);
+ return new LIPFilterAdaptiveProber(lip_filters_, attr_ids_, attr_types_);
+}
+
+} // namespace quickstep
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/8746eed4/utility/lip_filter/LIPFilterDeployment.hpp
----------------------------------------------------------------------
diff --git a/utility/lip_filter/LIPFilterDeployment.hpp b/utility/lip_filter/LIPFilterDeployment.hpp
new file mode 100644
index 0000000..9b37f88
--- /dev/null
+++ b/utility/lip_filter/LIPFilterDeployment.hpp
@@ -0,0 +1,111 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ **/
+
+#ifndef QUICKSTEP_UTILITY_LIP_FILTER_LIP_FILTER_DEPLOYMENT_HPP_
+#define QUICKSTEP_UTILITY_LIP_FILTER_LIP_FILTER_DEPLOYMENT_HPP_
+
+#include <memory>
+#include <vector>
+
+#include "catalog/CatalogTypedefs.hpp"
+#include "utility/Macros.hpp"
+
+namespace quickstep {
+
+namespace serialization { class LIPFilterDeployment; }
+
+class LIPFilter;
+class LIPFilterBuilder;
+class LIPFilterAdaptiveProber;
+class Type;
+
+/** \addtogroup Utility
+ * @{
+ */
+
+enum class LIPFilterActionType {
+ kBuild = 0,
+ kProbe
+};
+
+/**
+ * @brief Helper class for organizing a group of LIPFilters in the backend.
+ * Each LIPFilterDeployment object is attached to a RelationalOperator.
+ */
+class LIPFilterDeployment {
+ public:
+ /**
+ * @brief Constructor.
+ *
+ * @param proto The Protocol Buffer representation of a LIPFilterDeployment.
+ * @param lip_filters The LIPFilter objects to be deployed.
+ */
+ LIPFilterDeployment(const serialization::LIPFilterDeployment &proto,
+ const std::vector<std::unique_ptr<LIPFilter>> &lip_filters);
+ /**
+ * @brief Determine if a serialized protobuf representation of a
+ * LIPFilterDeployment is fully-formed and valid.
+ *
+ * @param proto A serialized protobuf representation of a LIPFilterDeployment
+ * to check for validity.
+ * @return Whether proto is fully-formed and valid.
+ **/
+ static bool ProtoIsValid(const serialization::LIPFilterDeployment &proto);
+
+ /**
+ * @brief Get the action type for this group of LIPFilters (i.e. whether
+ * to build or probe the filters).
+ *
+ * @return The action type.
+ */
+ LIPFilterActionType getActionType() const {
+ return action_type_;
+ }
+
+ /**
+ * @brief Create a LIPFilterBuilder for this group of LIPFilters.
+ *
+ * @return A new LIPFilterBuilder object for this group of LIPFilters.
+ * Caller should take ownership of the returned object.
+ */
+ LIPFilterBuilder* createLIPFilterBuilder() const;
+
+ /**
+ * @brief Create a LIPFilterAdaptiveProber for this group of LIPFilters.
+ *
+ * @return A new LIPFilterAdaptiveProber object for this group of LIPFilters.
+ * Caller should take ownership of the returned object.
+ */
+ LIPFilterAdaptiveProber* createLIPFilterAdaptiveProber() const;
+
+ private:
+ LIPFilterActionType action_type_;
+
+ std::vector<LIPFilter *> lip_filters_;
+ std::vector<attribute_id> attr_ids_;
+ std::vector<const Type *> attr_types_;
+
+ DISALLOW_COPY_AND_ASSIGN(LIPFilterDeployment);
+};
+
+/** @} */
+
+} // namespace quickstep
+
+#endif // QUICKSTEP_UTILITY_LIP_FILTER_LIP_FILTER_DEPLOYMENT_HPP_
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/8746eed4/utility/lip_filter/LIPFilterFactory.cpp
----------------------------------------------------------------------
diff --git a/utility/lip_filter/LIPFilterFactory.cpp b/utility/lip_filter/LIPFilterFactory.cpp
new file mode 100644
index 0000000..ebc4a0e
--- /dev/null
+++ b/utility/lip_filter/LIPFilterFactory.cpp
@@ -0,0 +1,74 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ **/
+
+#include "utility/lip_filter/LIPFilterFactory.hpp"
+
+#include <cstddef>
+#include <cstdint>
+
+#include "utility/lip_filter/LIPFilter.pb.h"
+#include "utility/lip_filter/SingleIdentityHashFilter.hpp"
+
+#include "glog/logging.h"
+
+namespace quickstep {
+
+LIPFilter* LIPFilterFactory::ReconstructFromProto(const serialization::LIPFilter &proto) {
+ switch (proto.lip_filter_type()) {
+ case serialization::LIPFilterType::SINGLE_IDENTITY_HASH_FILTER: {
+ const std::size_t attr_size =
+ proto.GetExtension(serialization::SingleIdentityHashFilter::attribute_size);
+ const std::size_t filter_cardinality =
+ proto.GetExtension(serialization::SingleIdentityHashFilter::filter_cardinality);
+
+ if (attr_size >= 8) {
+ return new SingleIdentityHashFilter<std::uint64_t>(filter_cardinality);
+ } else if (attr_size >= 4) {
+ return new SingleIdentityHashFilter<std::uint32_t>(filter_cardinality);
+ } else if (attr_size >= 2) {
+ return new SingleIdentityHashFilter<std::uint16_t>(filter_cardinality);
+ } else {
+ return new SingleIdentityHashFilter<std::uint8_t>(filter_cardinality);
+ }
+ }
+ // TODO(jianqiao): handle the BLOOM_FILTER and EXACT_FILTER implementations.
+ default:
+ LOG(FATAL) << "Unsupported LIP filter type: "
+ << serialization::LIPFilterType_Name(proto.lip_filter_type());
+ }
+ return nullptr;
+}
+
+bool LIPFilterFactory::ProtoIsValid(const serialization::LIPFilter &proto) {
+ switch (proto.lip_filter_type()) {
+ case serialization::LIPFilterType::SINGLE_IDENTITY_HASH_FILTER: {
+ const std::size_t attr_size =
+ proto.GetExtension(serialization::SingleIdentityHashFilter::attribute_size);
+ const std::size_t filter_cardinality =
+ proto.GetExtension(serialization::SingleIdentityHashFilter::filter_cardinality);
+ return (attr_size != 0 && filter_cardinality != 0);
+ }
+ default:
+ LOG(FATAL) << "Unsupported LIP filter type: "
+ << serialization::LIPFilterType_Name(proto.lip_filter_type());
+ }
+ return false;
+}
+
+} // namespace quickstep
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/8746eed4/utility/lip_filter/LIPFilterFactory.hpp
----------------------------------------------------------------------
diff --git a/utility/lip_filter/LIPFilterFactory.hpp b/utility/lip_filter/LIPFilterFactory.hpp
new file mode 100644
index 0000000..b8301b8
--- /dev/null
+++ b/utility/lip_filter/LIPFilterFactory.hpp
@@ -0,0 +1,69 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ **/
+
+#ifndef QUICKSTEP_UTILITY_LIP_FILTER_LIP_FILTER_FACTORY_HPP_
+#define QUICKSTEP_UTILITY_LIP_FILTER_LIP_FILTER_FACTORY_HPP_
+
+#include "utility/Macros.hpp"
+
+namespace quickstep {
+
+namespace serialization { class LIPFilter; }
+
+class LIPFilter;
+
+/** \addtogroup Utility
+ * @{
+ */
+
+/**
+ * @brief All-static factory object that provides access to various implementations
+ * of LIPFilters.
+ */
+class LIPFilterFactory {
+ public:
+ /**
+ * @brief Reconstruct a LIPFilter from its serialized Protocol Buffer form.
+ *
+ * @param proto The Protocol Buffer representation of a LIPFilter object,
+ * @return A new LIPFilter reconstructed from the supplied Protocol Buffer.
+ * Caller should take ownership of the returned object.
+ */
+ static LIPFilter* ReconstructFromProto(const serialization::LIPFilter &proto);
+
+ /**
+ * @brief Check whether a serialization::LIPFilter is fully-formed and
+ * all parts are valid.
+ *
+ * @param proto A serialized Protocol Buffer representation of a LIPFilter.
+ * @return Whether proto is fully-formed and valid.
+ **/
+ static bool ProtoIsValid(const serialization::LIPFilter &proto);
+
+ private:
+ LIPFilterFactory() {}
+
+ DISALLOW_COPY_AND_ASSIGN(LIPFilterFactory);
+};
+
+/** @} */
+
+} // namespace quickstep
+
+#endif // QUICKSTEP_UTILITY_LIP_FILTER_LIP_FILTER_FACTORY_HPP_
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/8746eed4/utility/lip_filter/SingleIdentityHashFilter.hpp
----------------------------------------------------------------------
diff --git a/utility/lip_filter/SingleIdentityHashFilter.hpp b/utility/lip_filter/SingleIdentityHashFilter.hpp
new file mode 100644
index 0000000..2823818
--- /dev/null
+++ b/utility/lip_filter/SingleIdentityHashFilter.hpp
@@ -0,0 +1,185 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ **/
+
+#ifndef QUICKSTEP_UTILITY_LIP_FILTER_SINGLE_IDENTITY_HASH_FILTER_HPP_
+#define QUICKSTEP_UTILITY_LIP_FILTER_SINGLE_IDENTITY_HASH_FILTER_HPP_
+
+#include <atomic>
+#include <cstddef>
+#include <cstdint>
+#include <cstring>
+#include <vector>
+
+#include "catalog/CatalogTypedefs.hpp"
+#include "storage/StorageBlockInfo.hpp"
+#include "storage/StorageConstants.hpp"
+#include "storage/ValueAccessor.hpp"
+#include "storage/ValueAccessorUtil.hpp"
+#include "types/Type.hpp"
+#include "utility/Macros.hpp"
+#include "utility/lip_filter/LIPFilter.hpp"
+
+#include "glog/logging.h"
+
+namespace quickstep {
+
+/** \addtogroup Utility
+ * @{
+ */
+
+/**
+ * @brief Specialized bloom filter that uses only one hash function. The hash
+ * function is an "identity" function that it simply reinterpret_casts
+ * a value's byte stream into the specified CppType as the value's hash
+ * code.
+ *
+ * @note SingleIdentityHashFilter is most effective when applied to fixed-length
+ * integer values. It cannot be applied to variable-length values unless
+ * the corresponding value Type has its minimumByteLength() greater than
+ * sizeof(CppType).
+ */
+template <typename CppType>
+class SingleIdentityHashFilter : public LIPFilter {
+ public:
+ /**
+ * @brief Constructor.
+ *
+ * @param filter_cardinality The cardinality of this hash filter.
+ */
+ explicit SingleIdentityHashFilter(const std::size_t filter_cardinality)
+ : LIPFilter(LIPFilterType::kSingleIdentityHashFilter),
+ filter_cardinality_(filter_cardinality),
+ bit_array_(GetByteSize(filter_cardinality)) {
+ DCHECK_GE(filter_cardinality, 0u);
+ std::memset(bit_array_.data(),
+ 0x0,
+ sizeof(std::atomic<std::uint8_t>) * GetByteSize(filter_cardinality));
+ }
+
+ void insertValueAccessor(ValueAccessor *accessor,
+ const attribute_id attr_id,
+ const Type *attr_type) override {
+ InvokeOnAnyValueAccessor(
+ accessor,
+ [&](auto *accessor) -> void { // NOLINT(build/c++11)
+ if (attr_type->isNullable()) {
+ this->insertValueAccessorInternal<true>(accessor, attr_id);
+ } else {
+ this->insertValueAccessorInternal<false>(accessor, attr_id);
+ }
+ });
+ }
+
+ std::size_t filterBatch(ValueAccessor *accessor,
+ const attribute_id attr_id,
+ const bool is_attr_nullable,
+ std::vector<tuple_id> *batch,
+ const std::size_t batch_size) const override {
+ DCHECK(batch != nullptr);
+ DCHECK_LE(batch_size, batch->size());
+
+ return InvokeOnAnyValueAccessor(
+ accessor,
+ [&](auto *accessor) -> std::size_t { // NOLINT(build/c++11)
+ if (is_attr_nullable) {
+ return this->filterBatchInternal<true>(accessor, attr_id, batch, batch_size);
+ } else {
+ return this->filterBatchInternal<false>(accessor, attr_id, batch, batch_size);
+ }
+ });
+ }
+
+ private:
+ /**
+ * @brief Round up bit_size to multiples of 8.
+ */
+ inline static std::size_t GetByteSize(const std::size_t bit_size) {
+ return (bit_size + 7) / 8;
+ }
+
+ /**
+ * @brief Iterate through the accessor and hash values into the internal bit
+ * array.
+ */
+ template <bool is_attr_nullable, typename ValueAccessorT>
+ inline void insertValueAccessorInternal(ValueAccessorT *accessor,
+ const attribute_id attr_id) {
+ accessor->beginIteration();
+ while (accessor->next()) {
+ const void *value = accessor->template getUntypedValue<is_attr_nullable>(attr_id);
+ if (!is_attr_nullable || value != nullptr) {
+ insert(value);
+ }
+ }
+ }
+
+ /**
+ * @brief Filter the given batch of tuples from a ValueAccessor. Write the
+ * tuple ids which survive in the filtering back to \p batch.
+ */
+ template <bool is_attr_nullable, typename ValueAccessorT>
+ inline std::size_t filterBatchInternal(const ValueAccessorT *accessor,
+ const attribute_id attr_id,
+ std::vector<tuple_id> *batch,
+ const std::size_t batch_size) const {
+ std::size_t out_size = 0;
+ for (std::size_t i = 0; i < batch_size; ++i) {
+ const tuple_id tid = batch->at(i);
+ const void *value =
+ accessor->template getUntypedValueAtAbsolutePosition(attr_id, tid);
+ if (is_attr_nullable && value == nullptr) {
+ continue;
+ }
+ if (contains(value)) {
+ batch->at(out_size) = tid;
+ ++out_size;
+ }
+ }
+ return out_size;
+ }
+
+ /**
+ * @brief Inserts a given value into the hash filter.
+ */
+ inline void insert(const void *key_begin) {
+ const CppType hash = *reinterpret_cast<const CppType *>(key_begin) % filter_cardinality_;
+ bit_array_[hash >> 3u].fetch_or(1u << (hash & 7u), std::memory_order_relaxed);
+ }
+
+ /**
+ * @brief Test membership of a given value in the hash filter.
+ * If true is returned, then a value may or may not be present in the hash filter.
+ * If false is returned, a value is certainly not present in the hash filter.
+ */
+ inline bool contains(const void *key_begin) const {
+ const CppType hash = *reinterpret_cast<const CppType *>(key_begin) % filter_cardinality_;
+ return (bit_array_[hash >> 3u].load(std::memory_order_relaxed) & (1u << (hash & 7u)));
+ }
+
+ std::size_t filter_cardinality_;
+ alignas(kCacheLineBytes) std::vector<std::atomic<std::uint8_t>> bit_array_;
+
+ DISALLOW_COPY_AND_ASSIGN(SingleIdentityHashFilter);
+};
+
+/** @} */
+
+} // namespace quickstep
+
+#endif // QUICKSTEP_UTILITY_LIP_FILTER_SINGLE_IDENTITY_HASH_FILTER_HPP_
[11/11] incubator-quickstep git commit: Analyze min/max stat for each
column.
Posted by ji...@apache.org.
Analyze min/max stat for each column.
Project: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/commit/f10f5677
Tree: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/tree/f10f5677
Diff: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/diff/f10f5677
Branch: refs/heads/exact-filter
Commit: f10f5677b7b7c03ad684967a3dc0ea0cfc41502a
Parents: 7f0067b
Author: Jianqiao Zhu <ji...@cs.wisc.edu>
Authored: Thu Oct 27 14:16:32 2016 -0500
Committer: Jianqiao Zhu <ji...@cs.wisc.edu>
Committed: Fri Oct 28 14:31:52 2016 -0500
----------------------------------------------------------------------
catalog/Catalog.proto | 10 +-
catalog/CatalogRelationStatistics.cpp | 49 ++++--
catalog/CatalogRelationStatistics.hpp | 151 ++++++++++++++++---
cli/CommandExecutor.cpp | 34 ++++-
cli/QuickstepCli.cpp | 6 +
.../cost_model/StarSchemaSimpleCostModel.cpp | 16 +-
6 files changed, 220 insertions(+), 46 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/f10f5677/catalog/Catalog.proto
----------------------------------------------------------------------
diff --git a/catalog/Catalog.proto b/catalog/Catalog.proto
index 90ce37e..58de363 100644
--- a/catalog/Catalog.proto
+++ b/catalog/Catalog.proto
@@ -81,13 +81,15 @@ message IndexScheme {
}
message CatalogRelationStatistics {
- optional fixed64 num_tuples = 1;
+ optional TypedValue num_tuples = 1;
- message NumDistinctValuesEntry {
+ message ColumnStats {
required int32 attr_id = 1;
- required fixed64 num_distinct_values = 2;
+ optional TypedValue num_distinct_values = 2;
+ optional TypedValue min_value = 3;
+ optional TypedValue max_value = 4;
}
- repeated NumDistinctValuesEntry num_distinct_values_map = 2;
+ repeated ColumnStats column_stats = 2;
}
message CatalogRelationSchema {
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/f10f5677/catalog/CatalogRelationStatistics.cpp
----------------------------------------------------------------------
diff --git a/catalog/CatalogRelationStatistics.cpp b/catalog/CatalogRelationStatistics.cpp
index 6a51570..0f67234 100644
--- a/catalog/CatalogRelationStatistics.cpp
+++ b/catalog/CatalogRelationStatistics.cpp
@@ -23,27 +23,56 @@
namespace quickstep {
+const TypedValue CatalogRelationStatistics::kNullValue =
+ NullType::InstanceNullable().makeNullValue();
+
CatalogRelationStatistics::CatalogRelationStatistics(
const serialization::CatalogRelationStatistics &proto) {
if (proto.has_num_tuples()) {
- num_tuples_ = proto.num_tuples();
+ num_tuples_ = TypedValue::ReconstructFromProto(proto.num_tuples());
+ } else {
+ num_tuples_ = kNullValue;
}
- for (int i = 0; i < proto.num_distinct_values_map_size(); ++i) {
- const auto &entry = proto.num_distinct_values_map(i);
- num_distinct_values_map_.emplace(entry.attr_id(),
- entry.num_distinct_values());
+
+ for (int i = 0; i < proto.column_stats_size(); ++i) {
+ const auto &stat_proto = proto.column_stats(i);
+ auto &stat = column_stats_[stat_proto.attr_id()];
+ if (stat_proto.has_num_distinct_values()) {
+ stat.num_distinct_values =
+ TypedValue::ReconstructFromProto(stat_proto.num_distinct_values());
+ }
+ if (stat_proto.has_min_value()) {
+ stat.min_value =
+ TypedValue::ReconstructFromProto(stat_proto.min_value());
+ }
+ if (stat_proto.has_max_value()) {
+ stat.max_value =
+ TypedValue::ReconstructFromProto(stat_proto.max_value());
+ }
}
}
serialization::CatalogRelationStatistics CatalogRelationStatistics::getProto() const {
serialization::CatalogRelationStatistics proto;
- if (num_tuples_ != 0) {
- proto.set_num_tuples(num_tuples_);
+ if (!num_tuples_.isNull()) {
+ proto.mutable_num_tuples()->CopyFrom(num_tuples_.getProto());
}
- for (const auto &pair : num_distinct_values_map_) {
- auto entry = proto.add_num_distinct_values_map();
+ for (const auto &pair : column_stats_) {
+ auto entry = proto.add_column_stats();
entry->set_attr_id(pair.first);
- entry->set_num_distinct_values(pair.second);
+ const auto &stat = pair.second;
+ if (!stat.num_distinct_values.isNull()) {
+ entry->mutable_num_distinct_values()->CopyFrom(
+ stat.num_distinct_values.getProto());
+ }
+ if (!stat.min_value.isNull()) {
+ entry->mutable_min_value()->CopyFrom(
+ stat.min_value.getProto());
+ }
+ if (!stat.max_value.isNull()) {
+ entry->mutable_max_value()->CopyFrom(
+ stat.max_value.getProto());
+ }
}
return proto;
}
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/f10f5677/catalog/CatalogRelationStatistics.hpp
----------------------------------------------------------------------
diff --git a/catalog/CatalogRelationStatistics.hpp b/catalog/CatalogRelationStatistics.hpp
index f2056f3..ecb3110 100644
--- a/catalog/CatalogRelationStatistics.hpp
+++ b/catalog/CatalogRelationStatistics.hpp
@@ -26,6 +26,9 @@
#include "catalog/Catalog.pb.h"
#include "catalog/CatalogTypedefs.hpp"
+#include "types/LongType.hpp"
+#include "types/NullType.hpp"
+#include "types/TypedValue.hpp"
#include "utility/Macros.hpp"
namespace quickstep {
@@ -44,7 +47,7 @@ class CatalogRelationStatistics {
* @brief Constructor.
**/
CatalogRelationStatistics()
- : num_tuples_(0) {}
+ : num_tuples_(kNullValue) {}
/**
* @brief Reconstruct a CatalogRelationStatistics object from its serialized
@@ -64,12 +67,12 @@ class CatalogRelationStatistics {
serialization::CatalogRelationStatistics getProto() const;
/**
- * @brief Set the number of tuples statistic.
+ * @brief Check whether the number of tuples statistic is available.
*
- * @param num_tuples The number of tuples statistic.
+ * @return True if the number of tuples statistic is available, false otherwise.
*/
- void setNumTuples(std::size_t num_tuples) {
- num_tuples_ = num_tuples;
+ bool hasNumTuples() const {
+ return !num_tuples_.isNull();
}
/**
@@ -78,41 +81,145 @@ class CatalogRelationStatistics {
* @return The number of tuples. Returns 0 if the statistic is not set.
*/
std::size_t getNumTuples() const {
- return num_tuples_;
+ DCHECK(hasNumTuples());
+ return num_tuples_.getLiteral<std::int64_t>();
+ }
+
+ /**
+ * @brief Set the number of tuples statistic.
+ *
+ * @param num_tuples The number of tuples statistic.
+ */
+ void setNumTuples(const std::size_t num_tuples) {
+ num_tuples_ = LongType::InstanceNonNullable().makeValue(&num_tuples);
+ }
+
+ /**
+ * @brief Check whether the number of distinct values statistic is available
+ * for a column.
+ *
+ * @param The id of the column.
+ * @return True if the number of distinct values statistic is available,
+ * false otherwise.
+ */
+ bool hasNumDistinctValues(const attribute_id attr_id) const {
+ const ColumnStats *stats = getColumnStats(attr_id);
+ return (stats != nullptr && !stats->num_distinct_values.isNull());
+ }
+
+ /**
+ * @brief Get the number of distinct values statistic for a column.
+ *
+ * @param The id of the column.
+ * @return The number of distinct values statistic for the column.
+ */
+ std::size_t getNumDistinctValues(const attribute_id attr_id) const {
+ DCHECK(hasNumDistinctValues(attr_id));
+ return column_stats_.at(attr_id).num_distinct_values.getLiteral<std::int64_t>();
}
/**
- * @brief Set the number of distinct values statistic for a column (catalog attribute).
+ * @brief Set the number of distinct values statistic for a column.
*
* @param attr_id The id of the column.
* @param num_distinct_values The number of distinct values statistic.
*/
- void setNumDistinctValues(attribute_id attr_id, std::size_t num_distinct_values) {
- num_distinct_values_map_[attr_id] = num_distinct_values;
+ void setNumDistinctValues(const attribute_id attr_id,
+ const std::size_t num_distinct_values) {
+ column_stats_[attr_id].num_distinct_values =
+ LongType::InstanceNonNullable().makeValue(&num_distinct_values);
}
/**
- * @brief Get the number of distinct values statistic for a column (catalog attribute).
+ * @brief Check whether the minimum value statistic is available for a column.
*
* @param The id of the column.
- * @return The number of distinct values statistic for the column. Returns 0
- * if the statistic is not set.
+ * @return True if the minimum value statistic is available, false otherwise.
*/
- std::size_t getNumDistinctValues(attribute_id attr_id) const {
- const auto it = num_distinct_values_map_.find(attr_id);
- if (it == num_distinct_values_map_.end()) {
- return static_cast<std::size_t>(0);
- } else {
- return it->second;
- }
+ bool hasMinValue(const attribute_id attr_id) const {
+ const ColumnStats *stats = getColumnStats(attr_id);
+ return (stats != nullptr && !stats->min_value.isNull());
+ }
+
+ /**
+ * @brief Get the minimum value statistic for a column.
+ *
+ * @param The id of the column.
+ * @return The minimum value statistic for the column.
+ */
+ const TypedValue& getMinValue(const attribute_id attr_id) const {
+ DCHECK(hasMinValue(attr_id));
+ return column_stats_.at(attr_id).min_value;
+ }
+
+ /**
+ * @brief Set the minimum value statistic for a column.
+ *
+ * @param The id of the column.
+ * @return The minimum value statistic for the column.
+ */
+ void setMinValue(const attribute_id attr_id,
+ const TypedValue &min_value) {
+ column_stats_[attr_id].min_value = min_value;
+ }
+
+ /**
+ * @brief Check whether the maximum value statistic is available for a column.
+ *
+ * @param The id of the column.
+ * @return True if the maximum value statistic is available, false otherwise.
+ */
+ bool hasMaxValue(const attribute_id attr_id) const {
+ const ColumnStats *stats = getColumnStats(attr_id);
+ return (stats != nullptr && !stats->max_value.isNull());
+ }
+
+ /**
+ * @brief Get the maximum value statistic for a column.
+ *
+ * @param The id of the column.
+ * @return The maximum value statistic for the column.
+ */
+ const TypedValue& getMaxValue(const attribute_id attr_id) const {
+ DCHECK(hasMaxValue(attr_id));
+ return column_stats_.at(attr_id).max_value;
+ }
+
+ /**
+ * @brief Set the maximum value statistic for a column.
+ *
+ * @param The id of the column.
+ * @return The maximum value statistic for the column.
+ */
+ void setMaxValue(const attribute_id attr_id,
+ const TypedValue &max_value) {
+ column_stats_[attr_id].max_value = max_value;
}
private:
+ struct ColumnStats {
+ ColumnStats()
+ : num_distinct_values(kNullValue),
+ min_value(kNullValue),
+ max_value(kNullValue) {
+ }
+ TypedValue num_distinct_values;
+ TypedValue min_value;
+ TypedValue max_value;
+ };
+
+ inline const ColumnStats* getColumnStats(const attribute_id attr_id) const {
+ const auto it = column_stats_.find(attr_id);
+ return (it == column_stats_.end() ? nullptr : &it->second);
+ }
+
+ static const TypedValue kNullValue;
+
// Total number of tuples in the relation.
- std::size_t num_tuples_;
+ TypedValue num_tuples_;
- // Number of distinct values for each column.
- std::unordered_map<attribute_id, std::size_t> num_distinct_values_map_;
+ // Statistics for each column.
+ std::unordered_map<attribute_id, ColumnStats> column_stats_;
DISALLOW_COPY_AND_ASSIGN(CatalogRelationStatistics);
};
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/f10f5677/cli/CommandExecutor.cpp
----------------------------------------------------------------------
diff --git a/cli/CommandExecutor.cpp b/cli/CommandExecutor.cpp
index fea4b74..aa4a039 100644
--- a/cli/CommandExecutor.cpp
+++ b/cli/CommandExecutor.cpp
@@ -33,6 +33,8 @@
#include "catalog/CatalogRelationSchema.hpp"
#include "cli/DropRelation.hpp"
#include "cli/PrintToScreen.hpp"
+#include "expressions/aggregation/AggregateFunctionMax.hpp"
+#include "expressions/aggregation/AggregateFunctionMin.hpp"
#include "parser/ParseStatement.hpp"
#include "parser/ParseString.hpp"
#include "parser/SqlParserWrapper.hpp"
@@ -322,9 +324,26 @@ void executeAnalyze(const PtrVector<ParseString> *arguments,
// Get the number of distinct values for each column.
for (const CatalogAttribute &attribute : relation) {
+ const Type &attr_type = attribute.getType();
+ bool is_min_applicable =
+ AggregateFunctionMin::Instance().canApplyToTypes({&attr_type});
+ bool is_max_applicable =
+ AggregateFunctionMax::Instance().canApplyToTypes({&attr_type});
+
std::string query_string = "SELECT COUNT(DISTINCT ";
query_string.append(attribute.getName());
- query_string.append(") FROM ");
+ query_string.append(")");
+ if (is_min_applicable) {
+ query_string.append(", MIN(");
+ query_string.append(attribute.getName());
+ query_string.append(")");
+ }
+ if (is_max_applicable) {
+ query_string.append(", MAX(");
+ query_string.append(attribute.getName());
+ query_string.append(")");
+ }
+ query_string.append(" FROM ");
query_string.append(relation.getName());
query_string.append(";");
@@ -340,9 +359,18 @@ void executeAnalyze(const PtrVector<ParseString> *arguments,
auto *stat = mutable_relation->getStatisticsMutable();
const attribute_id attr_id = attribute.getID();
- DCHECK(results[0].getTypeID() == TypeID::kLong);
+ auto results_it = results.begin();
+ DCHECK(results_it->getTypeID() == TypeID::kLong);
stat->setNumDistinctValues(attr_id,
- results[0].getLiteral<std::int64_t>());
+ results_it->getLiteral<std::int64_t>());
+ if (is_min_applicable) {
+ ++results_it;
+ stat->setMinValue(attr_id, *results_it);
+ }
+ if (is_max_applicable) {
+ ++results_it;
+ stat->setMaxValue(attr_id, *results_it);
+ }
}
// Get the number of tuples for the relation.
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/f10f5677/cli/QuickstepCli.cpp
----------------------------------------------------------------------
diff --git a/cli/QuickstepCli.cpp b/cli/QuickstepCli.cpp
index 8269197..b89a0f3 100644
--- a/cli/QuickstepCli.cpp
+++ b/cli/QuickstepCli.cpp
@@ -84,6 +84,7 @@ typedef quickstep::LineReaderDumb LineReaderImpl;
#include "gflags/gflags.h"
#include "glog/logging.h"
+#include "google/protobuf/text_format.h"
#include "tmb/address.h"
#include "tmb/id_typedefs.h"
@@ -300,6 +301,11 @@ int main(int argc, char* argv[]) {
LOG(FATAL) << "NON-STANDARD EXCEPTION DURING STARTUP";
}
+ std::string str;
+ google::protobuf::TextFormat::PrintToString(
+ query_processor->getDefaultDatabase()->getProto(), &str);
+ std::cerr << str << "\n";
+
// Parse the CPU affinities for workers and the preloader thread, if enabled
// to warm up the buffer pool.
const vector<int> worker_cpu_affinities =
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/f10f5677/query_optimizer/cost_model/StarSchemaSimpleCostModel.cpp
----------------------------------------------------------------------
diff --git a/query_optimizer/cost_model/StarSchemaSimpleCostModel.cpp b/query_optimizer/cost_model/StarSchemaSimpleCostModel.cpp
index 1075739..d8682f8 100644
--- a/query_optimizer/cost_model/StarSchemaSimpleCostModel.cpp
+++ b/query_optimizer/cost_model/StarSchemaSimpleCostModel.cpp
@@ -25,6 +25,7 @@
#include <vector>
#include "catalog/CatalogRelation.hpp"
+#include "catalog/CatalogRelationStatistics.hpp"
#include "query_optimizer/expressions/AttributeReference.hpp"
#include "query_optimizer/expressions/ComparisonExpression.hpp"
#include "query_optimizer/expressions/ExprId.hpp"
@@ -104,10 +105,11 @@ std::size_t StarSchemaSimpleCostModel::estimateCardinalityForTopLevelPlan(
std::size_t StarSchemaSimpleCostModel::estimateCardinalityForTableReference(
const P::TableReferencePtr &physical_plan) {
- std::size_t num_tuples = physical_plan->relation()->getStatistics().getNumTuples();
- if (num_tuples == 0) {
- num_tuples = physical_plan->relation()->estimateTupleCardinality();
- }
+ const CatalogRelation *relation = physical_plan->relation();
+ const CatalogRelationStatistics &stat = relation->getStatistics();
+ const std::size_t num_tuples =
+ stat.hasNumTuples() ? stat.getNumTuples()
+ : relation->estimateTupleCardinality();
return num_tuples;
}
@@ -384,9 +386,9 @@ std::size_t StarSchemaSimpleCostModel::getNumDistinctValues(
const std::vector<E::AttributeReferencePtr> &attributes = table_reference->attribute_list();
for (std::size_t i = 0; i < attributes.size(); ++i) {
if (attributes[i]->id() == attribute_id) {
- std::size_t num_distinct_values = relation.getStatistics().getNumDistinctValues(i);
- if (num_distinct_values > 0) {
- return num_distinct_values;
+ const CatalogRelationStatistics &stat = relation.getStatistics();
+ if (stat.hasNumDistinctValues(i)) {
+ return stat.getNumDistinctValues(i);
}
break;
}
[04/11] incubator-quickstep git commit: Remember handle count in
FastSeparateChainingHashTable
Posted by ji...@apache.org.
Remember handle count in FastSeparateChainingHashTable
- Using handles_ vector directly in FastSeparateChainingHashTable can
cause problems. One example is a distinctify hash table which doesn't
accept any payload. In this case, the reference to handles_ vector
becomes garbage after the constructor gets executed. Therefore to
avoid faults about accessing an invalid reference, we use the number
of handles as a guarding mechanism.
Project: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/commit/9c32ea45
Tree: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/tree/9c32ea45
Diff: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/diff/9c32ea45
Branch: refs/heads/exact-filter
Commit: 9c32ea45c5971568fae3d611a0fc9d963d7501bf
Parents: 8746eed
Author: Harshad Deshmukh <hb...@apache.org>
Authored: Fri Oct 21 16:15:53 2016 -0500
Committer: Harshad Deshmukh <hb...@apache.org>
Committed: Fri Oct 21 16:15:53 2016 -0500
----------------------------------------------------------------------
storage/FastSeparateChainingHashTable.hpp | 8 +++++---
1 file changed, 5 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/9c32ea45/storage/FastSeparateChainingHashTable.hpp
----------------------------------------------------------------------
diff --git a/storage/FastSeparateChainingHashTable.hpp b/storage/FastSeparateChainingHashTable.hpp
index 231da7e..2435d45 100644
--- a/storage/FastSeparateChainingHashTable.hpp
+++ b/storage/FastSeparateChainingHashTable.hpp
@@ -145,10 +145,10 @@ class FastSeparateChainingHashTable
header_->buckets_allocated.load(std::memory_order_relaxed);
void *bucket_ptr = static_cast<char *>(buckets_) + kValueOffset;
for (std::size_t bucket_num = 0; bucket_num < num_buckets; ++bucket_num) {
- for (std::size_t handle_num = 0; handle_num < handles_.size(); ++handle_num) {
+ for (std::size_t handle_id = 0; handle_id < num_handles_; ++handle_id) {
void *value_internal_ptr =
- static_cast<char *>(bucket_ptr) + this->payload_offsets_[handle_num];
- handles_[handle_num]->destroyPayload(static_cast<std::uint8_t *>(value_internal_ptr));
+ static_cast<char *>(bucket_ptr) + this->payload_offsets_[handle_id];
+ handles_[handle_id]->destroyPayload(static_cast<std::uint8_t *>(value_internal_ptr));
}
bucket_ptr = static_cast<char *>(bucket_ptr) + bucket_size_;
}
@@ -223,6 +223,7 @@ class FastSeparateChainingHashTable
bool isFull(const std::size_t extra_variable_storage) const;
const std::vector<AggregationHandle *> &handles_;
+ const std::size_t num_handles_;
// Helper object to manage key storage.
HashTableKeyManager<serializable, force_key_copy> key_manager_;
@@ -292,6 +293,7 @@ FastSeparateChainingHashTable<resizable,
kBucketAlignment(alignof(std::atomic<std::size_t>)),
kValueOffset(sizeof(std::atomic<std::size_t>) + sizeof(std::size_t)),
handles_(handles),
+ num_handles_(handles.size()),
key_manager_(this->key_types_, kValueOffset + this->total_payload_size_),
bucket_size_(ComputeBucketSize(key_manager_.getFixedKeySize())) {
init_payload_ =
[10/11] incubator-quickstep git commit: QUICKSTEP-46 fixed
Posted by ji...@apache.org.
QUICKSTEP-46 fixed
Project: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/commit/7f0067b7
Tree: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/tree/7f0067b7
Diff: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/diff/7f0067b7
Branch: refs/heads/exact-filter
Commit: 7f0067b7913d8f2b68e8ac771fb9a87090d773e9
Parents: 1effc79
Author: tarun <ta...@gmail.com>
Authored: Mon Oct 10 20:29:42 2016 -0500
Committer: tarunbansal <ta...@gmail.com>
Committed: Fri Oct 28 10:02:34 2016 -0500
----------------------------------------------------------------------
relational_operators/CMakeLists.txt | 14 ++++
relational_operators/TextScanOperator.cpp | 73 ++++++++++++++++----
relational_operators/TextScanOperator.hpp | 19 ++++-
.../tests/text_scan_faulty_golden_output.txt | 5 ++
.../tests/text_scan_faulty_input.txt | 4 ++
5 files changed, 99 insertions(+), 16 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/7f0067b7/relational_operators/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/relational_operators/CMakeLists.txt b/relational_operators/CMakeLists.txt
index 8dd65d0..0735bce 100644
--- a/relational_operators/CMakeLists.txt
+++ b/relational_operators/CMakeLists.txt
@@ -784,3 +784,17 @@ add_test(TextScanOperator_unittest
${TEXT_SCAN_INPUT_FILE}
${TEXT_SCAN_GOLDEN_OUTPUT_FILE}
${TEXT_SCAN_FAILURE_OUTPUT_FILE})
+file(TO_NATIVE_PATH
+ "${CMAKE_CURRENT_SOURCE_DIR}/tests/text_scan_faulty_input.txt"
+ TEXT_SCAN_FAULTY_INPUT_FILE)
+file(TO_NATIVE_PATH
+ "${CMAKE_CURRENT_SOURCE_DIR}/tests/text_scan_faulty_golden_output.txt"
+ TEXT_SCAN_FAULTY_GOLDEN_OUTPUT_FILE)
+file(TO_NATIVE_PATH
+ "${CMAKE_CURRENT_BINARY_DIR}/text_scan_faulty_failure_output.txt"
+ TEXT_SCAN_FAULTY_FAILURE_OUTPUT_FILE)
+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
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/7f0067b7/relational_operators/TextScanOperator.cpp
----------------------------------------------------------------------
diff --git a/relational_operators/TextScanOperator.cpp b/relational_operators/TextScanOperator.cpp
index 4151bac..aa734d3 100644
--- a/relational_operators/TextScanOperator.cpp
+++ b/relational_operators/TextScanOperator.cpp
@@ -196,7 +196,9 @@ serialization::WorkOrder* TextScanOperator::createWorkOrderProto(const string &f
void TextScanWorkOrder::execute() {
const CatalogRelationSchema &relation = output_destination_->getRelation();
std::vector<Tuple> tuples;
+ bool is_faulty;
+ std::vector<TypedValue> vector_tuple_returned;
constexpr std::size_t kSmallBufferSize = 0x4000;
char *buffer = reinterpret_cast<char *>(malloc(std::max(text_segment_size_, kSmallBufferSize)));
@@ -218,7 +220,6 @@ void TextScanWorkOrder::execute() {
} else {
--row_ptr;
}
-
if (row_ptr >= buffer_end) {
// This block does not even contain a newline character.
return;
@@ -238,16 +239,23 @@ void TextScanWorkOrder::execute() {
// RIGHT AFTER the LAST newline character in this text segment.
// Process the tuples which are between the first newline character and the
- // last newline character.
+ // last newline character. SKIP any row which is corrupt instead of ABORTING the
+ // whole COPY operation.
while (row_ptr < end_ptr) {
if (*row_ptr == '\r' || *row_ptr == '\n') {
// Skip empty lines.
++row_ptr;
} else {
- tuples.emplace_back(parseRow(&row_ptr, relation));
+ vector_tuple_returned = parseRow(&row_ptr, relation, &is_faulty);
+ if (is_faulty) {
+ // Skip faulty rows
+ LOG(INFO) << "Faulty row found. Hence switching to next row.";
+ } else {
+ // Convert vector returned to tuple only when a valid row is encountered.
+ tuples.emplace_back(Tuple(std::move(vector_tuple_returned)));
+ }
}
}
-
// Process the tuple that is right after the last newline character.
// NOTE(jianqiao): dynamic_read_size is trying to balance between the cases
// that the last tuple is very small / very large.
@@ -279,7 +287,15 @@ void TextScanWorkOrder::execute() {
row_string.push_back('\n');
}
row_ptr = row_string.c_str();
- tuples.emplace_back(parseRow(&row_ptr, relation));
+
+ vector_tuple_returned = parseRow(&row_ptr, relation, &is_faulty);
+ if (is_faulty) {
+ // Skip the faulty row.
+ LOG(INFO) << "Faulty row found. Hence switching to next row.";
+ } else {
+ // Convert vector returned to tuple only when a valid row is encountered.
+ tuples.emplace_back(Tuple(std::move(vector_tuple_returned)));
+ }
}
std::fclose(file);
@@ -312,19 +328,26 @@ void TextScanWorkOrder::execute() {
output_destination_->bulkInsertTuples(&column_vectors);
}
-Tuple TextScanWorkOrder::parseRow(const char **row_ptr,
- const CatalogRelationSchema &relation) const {
+std::vector<TypedValue> TextScanWorkOrder::parseRow(const char **row_ptr,
+ const CatalogRelationSchema &relation, bool *is_faulty) const {
std::vector<TypedValue> attribute_values;
+ // Always assume current row is not faulty initially.
+ *is_faulty = false;
bool is_null_literal;
bool has_reached_end_of_line = false;
std::string value_str;
for (const auto &attr : relation) {
if (has_reached_end_of_line) {
- throw TextScanFormatError("Row has too few fields");
+ // Do not abort if one of the row is faulty.
+ // Set is_faulty to true and SKIP the current row.
+ *is_faulty = true;
+ LOG(INFO) << "Row has too few fields.";
+ return attribute_values;
}
value_str.clear();
+
extractFieldString(row_ptr,
&is_null_literal,
&has_reached_end_of_line,
@@ -333,24 +356,46 @@ Tuple TextScanWorkOrder::parseRow(const char **row_ptr,
if (is_null_literal) {
// NULL literal.
if (!attr.getType().isNullable()) {
- throw TextScanFormatError(
- "NULL literal '\\N' was specified for a column with a "
- "non-nullable Type");
+ *is_faulty = true;
+ LOG(INFO) << "NULL literal '\\N' was specified for a column with a "
+ "non-nullable Type.";
+ skipFaultyRow(row_ptr);
+ return attribute_values;
}
attribute_values.emplace_back(attr.getType().makeNullValue());
} else {
attribute_values.emplace_back();
if (!attr.getType().parseValueFromString(value_str, &(attribute_values.back()))) {
- throw TextScanFormatError("Failed to parse value");
+ // Do not abort if one of the row is faulty.
+ *is_faulty = true;
+ LOG(INFO) << "Failed to parse value.";
+ skipFaultyRow(row_ptr);
+ return attribute_values;
}
}
}
if (!has_reached_end_of_line) {
- throw TextScanFormatError("Row has too many fields");
+ // Do not abort if one of the row is faulty.
+ // Set is_faulty to true and SKIP the current row.
+ *is_faulty = true;
+ LOG(INFO) << "Row has too many fields.";
+ skipFaultyRow(row_ptr);
}
- return Tuple(std::move(attribute_values));
+ return attribute_values;
+}
+
+void TextScanWorkOrder::skipFaultyRow(const char **field_ptr) const {
+ const char *cur_ptr = *field_ptr;
+ // Move row pointer to the end of faulty row.
+ for (;; ++cur_ptr) {
+ const char c = *cur_ptr;
+ if (c == '\n') {
+ break;
+ }
+ }
+ *field_ptr = cur_ptr + 1;
}
void TextScanWorkOrder::extractFieldString(const char **field_ptr,
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/7f0067b7/relational_operators/TextScanOperator.hpp
----------------------------------------------------------------------
diff --git a/relational_operators/TextScanOperator.hpp b/relational_operators/TextScanOperator.hpp
index 24af844..65863b3 100644
--- a/relational_operators/TextScanOperator.hpp
+++ b/relational_operators/TextScanOperator.hpp
@@ -24,6 +24,7 @@
#include <cstddef>
#include <exception>
#include <string>
+#include <vector>
#include "catalog/CatalogRelation.hpp"
#include "catalog/CatalogTypedefs.hpp"
@@ -241,6 +242,18 @@ class TextScanWorkOrder : public WorkOrder {
std::string *field_string) const;
/**
+ * @brief This method helps incorporate fault tolerance while ingesting data.
+ * It is called whenever a faulty row is encountered and it is
+ * required to move \p *field_ptr to the next row.
+ *
+ * @param field_ptr \p *field_ptr points to the current position of the input
+ * char stream while parsing a faulty row. After the call, \p *field_ptr
+ * will be modified to the start position of the NEXT record in the
+ * stream.
+ */
+ void skipFaultyRow(const char **field_ptr) const;
+
+ /**
* @brief Make a tuple by parsing all of the individual fields from a char stream.
*
* @param \p *row_ptr points to the current position of the input char stream
@@ -248,10 +261,12 @@ class TextScanWorkOrder : public WorkOrder {
* After the call, \p *row_ptr will be modified to the start position of
* the NEXT text row.
* @param relation The relation schema for the tuple.
+ * @param is_faulty OUTPUT parameter. Set to true if the row is faulty,
* @return The tuple parsed from the char stream.
*/
- Tuple parseRow(const char **row_ptr,
- const CatalogRelationSchema &relation) const;
+std::vector<TypedValue> parseRow(const char **row_ptr,
+ const CatalogRelationSchema &relation,
+ bool *is_faulty) const;
/**
* @brief Parse up to three octal digits (0-7) starting at \p *literal_ptr as
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/7f0067b7/relational_operators/tests/text_scan_faulty_golden_output.txt
----------------------------------------------------------------------
diff --git a/relational_operators/tests/text_scan_faulty_golden_output.txt b/relational_operators/tests/text_scan_faulty_golden_output.txt
new file mode 100644
index 0000000..e07bedf
--- /dev/null
+++ b/relational_operators/tests/text_scan_faulty_golden_output.txt
@@ -0,0 +1,5 @@
++--------------------+------------------------+--------------------+-----------------------------------------+----------------------------------------+--------------------+
+|long_attr |double_attr |char_attr |datetime_attr |interval_attr |varchar_attr |
++--------------------+------------------------+--------------------+-----------------------------------------+----------------------------------------+--------------------+
+| 1234| 12.34| foo| 1994-04-27T08:20:50| 12 days 00:00:00| right_row|
++--------------------+------------------------+--------------------+-----------------------------------------+----------------------------------------+--------------------+
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/7f0067b7/relational_operators/tests/text_scan_faulty_input.txt
----------------------------------------------------------------------
diff --git a/relational_operators/tests/text_scan_faulty_input.txt b/relational_operators/tests/text_scan_faulty_input.txt
new file mode 100644
index 0000000..aa00d39
--- /dev/null
+++ b/relational_operators/tests/text_scan_faulty_input.txt
@@ -0,0 +1,4 @@
+1234 12.34 foo 1994-04-27T08:20:50 12 days right_row
+1234 abcd foo 1994-04-27T08:20:50 12 days row_with_wrong_datatype_value
+1234 foo 1994-04-27T08:20:50 12 days row_with_less_values
+1234 abcd foo 1994-04-27T08:20:50 12 days bar row_with_more_values
[09/11] incubator-quickstep git commit: Fix copyright notice in the
header files.
Posted by ji...@apache.org.
Fix copyright notice in the header files.
Project: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/commit/1effc794
Tree: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/tree/1effc794
Diff: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/diff/1effc794
Branch: refs/heads/exact-filter
Commit: 1effc794e2a577de5e086e0965b8e7804578da4b
Parents: 1e1434a
Author: Harshad Deshmukh <hb...@apache.org>
Authored: Thu Oct 27 13:42:27 2016 -0500
Committer: Harshad Deshmukh <hb...@apache.org>
Committed: Thu Oct 27 13:42:27 2016 -0500
----------------------------------------------------------------------
storage/FastHashTable.hpp | 28 ++++++++++++++--------------
storage/PartitionedHashTablePool.hpp | 26 ++++++++++++++------------
2 files changed, 28 insertions(+), 26 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/1effc794/storage/FastHashTable.hpp
----------------------------------------------------------------------
diff --git a/storage/FastHashTable.hpp b/storage/FastHashTable.hpp
index 74d9ee3..4a82a62 100644
--- a/storage/FastHashTable.hpp
+++ b/storage/FastHashTable.hpp
@@ -1,20 +1,20 @@
/**
- * Copyright 2011-2015 Quickstep Technologies LLC.
- * Copyright 2015-2016 Pivotal Software, Inc.
- * Copyright 2016, Quickstep Research Group, Computer Sciences Department,
- * University of Wisconsin\u2014Madison.
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
*
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
+ * http://www.apache.org/licenses/LICENSE-2.0
*
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
**/
#ifndef QUICKSTEP_STORAGE_FAST_HASH_TABLE_HPP_
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/1effc794/storage/PartitionedHashTablePool.hpp
----------------------------------------------------------------------
diff --git a/storage/PartitionedHashTablePool.hpp b/storage/PartitionedHashTablePool.hpp
index bcc4d23..95d1810 100644
--- a/storage/PartitionedHashTablePool.hpp
+++ b/storage/PartitionedHashTablePool.hpp
@@ -1,18 +1,20 @@
/**
- * Copyright 2016, Quickstep Research Group, Computer Sciences Department,
- * University of Wisconsin\u2014Madison.
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
*
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
+ * http://www.apache.org/licenses/LICENSE-2.0
*
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
**/
#ifndef QUICKSTEP_STORAGE_PARTITIONED_HASH_TABLE_POOL_HPP_
[07/11] incubator-quickstep git commit: Add backend support for
LIPFilters.
Posted by ji...@apache.org.
Add backend support for LIPFilters.
Project: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/commit/393eba55
Tree: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/tree/393eba55
Diff: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/diff/393eba55
Branch: refs/heads/exact-filter
Commit: 393eba550efdabbdccb2bf166fe8deae8bd3ed77
Parents: 539a95a
Author: Jianqiao Zhu <ji...@cs.wisc.edu>
Authored: Wed Sep 7 13:20:43 2016 -0500
Committer: Jianqiao Zhu <ji...@cs.wisc.edu>
Committed: Mon Oct 24 17:00:55 2016 -0500
----------------------------------------------------------------------
expressions/scalar/ScalarAttribute.cpp | 2 +-
query_execution/QueryContext.hpp | 8 +-
query_optimizer/PhysicalGenerator.cpp | 2 +-
query_optimizer/rules/AttachLIPFilters.cpp | 17 ++-
query_optimizer/rules/CMakeLists.txt | 1 +
query_optimizer/tests/OptimizerTextTest.cpp | 6 +-
relational_operators/AggregationOperator.cpp | 12 +-
relational_operators/AggregationOperator.hpp | 10 +-
relational_operators/BuildHashOperator.cpp | 17 ++-
relational_operators/BuildHashOperator.hpp | 18 ++-
relational_operators/CMakeLists.txt | 11 ++
relational_operators/HashJoinOperator.cpp | 161 ++++++++++++++++------
relational_operators/HashJoinOperator.hpp | 64 ++++++---
relational_operators/SelectOperator.cpp | 87 +++++++++---
relational_operators/SelectOperator.hpp | 16 ++-
relational_operators/WorkOrder.proto | 7 +-
relational_operators/WorkOrderFactory.cpp | 73 +++++++++-
storage/AggregationOperationState.cpp | 49 ++++---
storage/AggregationOperationState.hpp | 9 +-
storage/CMakeLists.txt | 5 +-
storage/StorageBlock.cpp | 112 +++++----------
storage/StorageBlock.hpp | 82 ++++-------
utility/PlanVisualizer.cpp | 4 +-
utility/lip_filter/CMakeLists.txt | 4 +
utility/lip_filter/LIPFilterBuilder.hpp | 3 -
utility/lip_filter/LIPFilterUtil.hpp | 79 +++++++++++
26 files changed, 588 insertions(+), 271 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/393eba55/expressions/scalar/ScalarAttribute.cpp
----------------------------------------------------------------------
diff --git a/expressions/scalar/ScalarAttribute.cpp b/expressions/scalar/ScalarAttribute.cpp
index b29286b..cc42084 100644
--- a/expressions/scalar/ScalarAttribute.cpp
+++ b/expressions/scalar/ScalarAttribute.cpp
@@ -168,7 +168,7 @@ ColumnVector* ScalarAttribute::getAllValuesForJoin(
ValueAccessor *accessor = using_left_relation ? left_accessor
: right_accessor;
- return InvokeOnValueAccessorNotAdapter(
+ return InvokeOnAnyValueAccessor(
accessor,
[&joined_tuple_ids,
&attr_id,
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/393eba55/query_execution/QueryContext.hpp
----------------------------------------------------------------------
diff --git a/query_execution/QueryContext.hpp b/query_execution/QueryContext.hpp
index 4ebb042..7ad8fa1 100644
--- a/query_execution/QueryContext.hpp
+++ b/query_execution/QueryContext.hpp
@@ -88,7 +88,7 @@ class QueryContext {
/**
* @brief A unique identifier for a LIPFilterDeployment per query.
**/
- typedef std::uint32_t lip_deployment_id;
+ typedef std::int32_t lip_deployment_id;
static constexpr lip_deployment_id kInvalidLIPDeploymentId = static_cast<lip_deployment_id>(-1);
/**
@@ -315,7 +315,7 @@ class QueryContext {
* @return True if valid, otherwise false.
**/
bool isValidLIPDeploymentId(const lip_deployment_id id) const {
- return id < lip_deployments_.size();
+ return static_cast<std::size_t>(id) < lip_deployments_.size();
}
/**
@@ -328,7 +328,7 @@ class QueryContext {
**/
inline const LIPFilterDeployment* getLIPDeployment(
const lip_deployment_id id) const {
- DCHECK_LT(id, lip_deployments_.size());
+ DCHECK_LT(static_cast<std::size_t>(id), lip_deployments_.size());
return lip_deployments_[id].get();
}
@@ -338,7 +338,7 @@ class QueryContext {
* @param id The id of the LIPFilterDeployment to destroy.
**/
inline void destroyLIPDeployment(const lip_deployment_id id) {
- DCHECK_LT(id, lip_deployments_.size());
+ DCHECK_LT(static_cast<std::size_t>(id), lip_deployments_.size());
lip_deployments_[id].reset();
}
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/393eba55/query_optimizer/PhysicalGenerator.cpp
----------------------------------------------------------------------
diff --git a/query_optimizer/PhysicalGenerator.cpp b/query_optimizer/PhysicalGenerator.cpp
index 9db4037..7cb97dc 100644
--- a/query_optimizer/PhysicalGenerator.cpp
+++ b/query_optimizer/PhysicalGenerator.cpp
@@ -50,7 +50,7 @@ DEFINE_bool(reorder_hash_joins, true,
"cardinality and selective tables to be joined first, which is suitable "
"for queries on star-schema tables.");
-DEFINE_bool(use_lip_filters, false,
+DEFINE_bool(use_lip_filters, true,
"If true, use LIP (Lookahead Information Passing) filters to accelerate "
"query processing. LIP filters are effective for queries on star schema "
"tables (e.g. the SSB benchmark) and snowflake schema tables (e.g. the "
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/393eba55/query_optimizer/rules/AttachLIPFilters.cpp
----------------------------------------------------------------------
diff --git a/query_optimizer/rules/AttachLIPFilters.cpp b/query_optimizer/rules/AttachLIPFilters.cpp
index 090fb8c..b3c57ab 100644
--- a/query_optimizer/rules/AttachLIPFilters.cpp
+++ b/query_optimizer/rules/AttachLIPFilters.cpp
@@ -36,6 +36,7 @@
#include "query_optimizer/physical/PhysicalType.hpp"
#include "query_optimizer/physical/Selection.hpp"
#include "query_optimizer/physical/TopLevelPlan.hpp"
+#include "types/TypedValue.hpp"
#include "utility/lip_filter/LIPFilter.hpp"
#include "glog/logging.h"
@@ -174,12 +175,16 @@ const std::vector<AttachLIPFilters::LIPFilterInfoPtr>& AttachLIPFilters
if (selectivity < 1.0) {
std::size_t cardinality = cost_model_->estimateCardinality(build_node);
for (const auto &attr : hash_join->right_join_attributes()) {
- lip_filters.emplace_back(
- std::make_shared<LIPFilterInfo>(attr,
- path.cdr()->node,
- path.depth,
- selectivity,
- cardinality));
+ // NOTE(jianqiao): currently we only consider attributes of primitive
+ // fixed-length types.
+ if (TypedValue::HashIsReversible(attr->getValueType().getTypeID())) {
+ lip_filters.emplace_back(
+ std::make_shared<LIPFilterInfo>(attr,
+ path.cdr()->node,
+ path.depth,
+ selectivity,
+ cardinality));
+ }
}
}
}
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/393eba55/query_optimizer/rules/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/query_optimizer/rules/CMakeLists.txt b/query_optimizer/rules/CMakeLists.txt
index 29875f6..7fffadc 100644
--- a/query_optimizer/rules/CMakeLists.txt
+++ b/query_optimizer/rules/CMakeLists.txt
@@ -50,6 +50,7 @@ target_link_libraries(quickstep_queryoptimizer_rules_AttachLIPFilters
quickstep_queryoptimizer_physical_Selection
quickstep_queryoptimizer_physical_TopLevelPlan
quickstep_queryoptimizer_rules_Rule
+ quickstep_types_TypedValue
quickstep_utility_Macros
quickstep_utility_lipfilter_LIPFilter)
target_link_libraries(quickstep_queryoptimizer_rules_BottomUpRule
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/393eba55/query_optimizer/tests/OptimizerTextTest.cpp
----------------------------------------------------------------------
diff --git a/query_optimizer/tests/OptimizerTextTest.cpp b/query_optimizer/tests/OptimizerTextTest.cpp
index b6be739..759c173 100644
--- a/query_optimizer/tests/OptimizerTextTest.cpp
+++ b/query_optimizer/tests/OptimizerTextTest.cpp
@@ -32,6 +32,7 @@ namespace quickstep {
namespace optimizer {
DECLARE_bool(reorder_hash_joins);
+DECLARE_bool(use_lip_filters);
}
}
@@ -57,9 +58,10 @@ int main(int argc, char** argv) {
test_driver->registerOptions(
quickstep::optimizer::OptimizerTextTestRunner::kTestOptions);
- // Turn off join order optimization for optimizer test since it is up to change
- // and affects a large number of test cases.
+ // Turn off join order optimization and LIPFilter for optimizer test since
+ // it is up to change and affects a large number of test cases.
quickstep::optimizer::FLAGS_reorder_hash_joins = false;
+ quickstep::optimizer::FLAGS_use_lip_filters = false;
::testing::InitGoogleTest(&argc, argv);
int success = RUN_ALL_TESTS();
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/393eba55/relational_operators/AggregationOperator.cpp
----------------------------------------------------------------------
diff --git a/relational_operators/AggregationOperator.cpp b/relational_operators/AggregationOperator.cpp
index 056e76d..e111f5b 100644
--- a/relational_operators/AggregationOperator.cpp
+++ b/relational_operators/AggregationOperator.cpp
@@ -27,6 +27,8 @@
#include "relational_operators/WorkOrder.pb.h"
#include "storage/AggregationOperationState.hpp"
#include "storage/StorageBlockInfo.hpp"
+#include "utility/lip_filter/LIPFilterAdaptiveProber.hpp"
+#include "utility/lip_filter/LIPFilterUtil.hpp"
#include "tmb/id_typedefs.h"
@@ -45,7 +47,8 @@ bool AggregationOperator::getAllWorkOrders(
new AggregationWorkOrder(
query_id_,
input_block_id,
- query_context->getAggregationState(aggr_state_index_)),
+ query_context->getAggregationState(aggr_state_index_),
+ CreateLIPFilterAdaptiveProberHelper(lip_deployment_index_, query_context)),
op_index_);
}
started_ = true;
@@ -57,7 +60,8 @@ bool AggregationOperator::getAllWorkOrders(
new AggregationWorkOrder(
query_id_,
input_relation_block_ids_[num_workorders_generated_],
- query_context->getAggregationState(aggr_state_index_)),
+ query_context->getAggregationState(aggr_state_index_),
+ CreateLIPFilterAdaptiveProberHelper(lip_deployment_index_, query_context)),
op_index_);
++num_workorders_generated_;
}
@@ -92,13 +96,13 @@ serialization::WorkOrder* AggregationOperator::createWorkOrderProto(const block_
proto->SetExtension(serialization::AggregationWorkOrder::block_id, block);
proto->SetExtension(serialization::AggregationWorkOrder::aggr_state_index, aggr_state_index_);
+ proto->SetExtension(serialization::AggregationWorkOrder::lip_deployment_index, lip_deployment_index_);
return proto;
}
-
void AggregationWorkOrder::execute() {
- state_->aggregateBlock(input_block_id_);
+ state_->aggregateBlock(input_block_id_, lip_filter_adaptive_prober_.get());
}
} // namespace quickstep
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/393eba55/relational_operators/AggregationOperator.hpp
----------------------------------------------------------------------
diff --git a/relational_operators/AggregationOperator.hpp b/relational_operators/AggregationOperator.hpp
index 31c1da4..b5ed977 100644
--- a/relational_operators/AggregationOperator.hpp
+++ b/relational_operators/AggregationOperator.hpp
@@ -30,6 +30,7 @@
#include "relational_operators/WorkOrder.hpp"
#include "storage/StorageBlockInfo.hpp"
#include "utility/Macros.hpp"
+#include "utility/lip_filter/LIPFilterAdaptiveProber.hpp"
#include "glog/logging.h"
@@ -137,13 +138,16 @@ class AggregationWorkOrder : public WorkOrder {
* @param query_id The ID of this query.
* @param input_block_id The block id.
* @param state The AggregationState to use.
+ * @param lip_filter_adaptive_prober The attached LIP filter prober.
**/
AggregationWorkOrder(const std::size_t query_id,
const block_id input_block_id,
- AggregationOperationState *state)
+ AggregationOperationState *state,
+ LIPFilterAdaptiveProber *lip_filter_adaptive_prober = nullptr)
: WorkOrder(query_id),
input_block_id_(input_block_id),
- state_(DCHECK_NOTNULL(state)) {}
+ state_(DCHECK_NOTNULL(state)),
+ lip_filter_adaptive_prober_(lip_filter_adaptive_prober) {}
~AggregationWorkOrder() override {}
@@ -153,6 +157,8 @@ class AggregationWorkOrder : public WorkOrder {
const block_id input_block_id_;
AggregationOperationState *state_;
+ std::unique_ptr<LIPFilterAdaptiveProber> lip_filter_adaptive_prober_;
+
DISALLOW_COPY_AND_ASSIGN(AggregationWorkOrder);
};
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/393eba55/relational_operators/BuildHashOperator.cpp
----------------------------------------------------------------------
diff --git a/relational_operators/BuildHashOperator.cpp b/relational_operators/BuildHashOperator.cpp
index 465621c..60e091f 100644
--- a/relational_operators/BuildHashOperator.cpp
+++ b/relational_operators/BuildHashOperator.cpp
@@ -34,6 +34,8 @@
#include "storage/TupleReference.hpp"
#include "storage/TupleStorageSubBlock.hpp"
#include "storage/ValueAccessor.hpp"
+#include "utility/lip_filter/LIPFilterBuilder.hpp"
+#include "utility/lip_filter/LIPFilterUtil.hpp"
#include "glog/logging.h"
@@ -79,7 +81,8 @@ bool BuildHashOperator::getAllWorkOrders(
any_join_key_attributes_nullable_,
input_block_id,
hash_table,
- storage_manager),
+ storage_manager,
+ CreateLIPFilterBuilderHelper(lip_deployment_index_, query_context)),
op_index_);
}
started_ = true;
@@ -95,7 +98,8 @@ bool BuildHashOperator::getAllWorkOrders(
any_join_key_attributes_nullable_,
input_relation_block_ids_[num_workorders_generated_],
hash_table,
- storage_manager),
+ storage_manager,
+ CreateLIPFilterBuilderHelper(lip_deployment_index_, query_context)),
op_index_);
++num_workorders_generated_;
}
@@ -136,17 +140,24 @@ serialization::WorkOrder* BuildHashOperator::createWorkOrderProto(const block_id
any_join_key_attributes_nullable_);
proto->SetExtension(serialization::BuildHashWorkOrder::join_hash_table_index, hash_table_index_);
proto->SetExtension(serialization::BuildHashWorkOrder::block_id, block);
+ proto->SetExtension(serialization::BuildHashWorkOrder::lip_deployment_index, lip_deployment_index_);
return proto;
}
-
void BuildHashWorkOrder::execute() {
BlockReference block(
storage_manager_->getBlock(build_block_id_, input_relation_));
TupleReferenceGenerator generator(build_block_id_);
std::unique_ptr<ValueAccessor> accessor(block->getTupleStorageSubBlock().createValueAccessor());
+
+ // Build LIPFilters if enabled.
+ if (lip_filter_builder_ != nullptr) {
+ lip_filter_builder_->insertValueAccessor(accessor.get());
+ accessor->beginIterationVirtual();
+ }
+
HashTablePutResult result;
if (join_key_attributes_.size() == 1) {
result = hash_table_->putValueAccessor(accessor.get(),
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/393eba55/relational_operators/BuildHashOperator.hpp
----------------------------------------------------------------------
diff --git a/relational_operators/BuildHashOperator.hpp b/relational_operators/BuildHashOperator.hpp
index 4a80a8a..0f96ef2 100644
--- a/relational_operators/BuildHashOperator.hpp
+++ b/relational_operators/BuildHashOperator.hpp
@@ -20,6 +20,7 @@
#ifndef QUICKSTEP_RELATIONAL_OPERATORS_BUILD_HASH_OPERATOR_HPP_
#define QUICKSTEP_RELATIONAL_OPERATORS_BUILD_HASH_OPERATOR_HPP_
+#include <memory>
#include <string>
#include <utility>
#include <vector>
@@ -31,6 +32,7 @@
#include "relational_operators/WorkOrder.hpp"
#include "storage/StorageBlockInfo.hpp"
#include "utility/Macros.hpp"
+#include "utility/lip_filter/LIPFilterBuilder.hpp"
#include "glog/logging.h"
@@ -162,6 +164,7 @@ class BuildHashWorkOrder : public WorkOrder {
* @param build_block_id The 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.
**/
BuildHashWorkOrder(const std::size_t query_id,
const CatalogRelationSchema &input_relation,
@@ -169,14 +172,16 @@ class BuildHashWorkOrder : public WorkOrder {
const bool any_join_key_attributes_nullable,
const block_id build_block_id,
JoinHashTable *hash_table,
- StorageManager *storage_manager)
+ StorageManager *storage_manager,
+ LIPFilterBuilder *lip_filter_builder = nullptr)
: WorkOrder(query_id),
input_relation_(input_relation),
join_key_attributes_(join_key_attributes),
any_join_key_attributes_nullable_(any_join_key_attributes_nullable),
build_block_id_(build_block_id),
hash_table_(DCHECK_NOTNULL(hash_table)),
- storage_manager_(DCHECK_NOTNULL(storage_manager)) {}
+ storage_manager_(DCHECK_NOTNULL(storage_manager)),
+ lip_filter_builder_(lip_filter_builder) {}
/**
* @brief Constructor for the distributed version.
@@ -189,6 +194,7 @@ class BuildHashWorkOrder : public WorkOrder {
* @param build_block_id The 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.
**/
BuildHashWorkOrder(const std::size_t query_id,
const CatalogRelationSchema &input_relation,
@@ -196,14 +202,16 @@ class BuildHashWorkOrder : public WorkOrder {
const bool any_join_key_attributes_nullable,
const block_id build_block_id,
JoinHashTable *hash_table,
- StorageManager *storage_manager)
+ StorageManager *storage_manager,
+ LIPFilterBuilder *lip_filter_builder = nullptr)
: WorkOrder(query_id),
input_relation_(input_relation),
join_key_attributes_(std::move(join_key_attributes)),
any_join_key_attributes_nullable_(any_join_key_attributes_nullable),
build_block_id_(build_block_id),
hash_table_(DCHECK_NOTNULL(hash_table)),
- storage_manager_(DCHECK_NOTNULL(storage_manager)) {}
+ storage_manager_(DCHECK_NOTNULL(storage_manager)),
+ lip_filter_builder_(lip_filter_builder) {}
~BuildHashWorkOrder() override {}
@@ -222,6 +230,8 @@ class BuildHashWorkOrder : public WorkOrder {
JoinHashTable *hash_table_;
StorageManager *storage_manager_;
+ std::unique_ptr<LIPFilterBuilder> lip_filter_builder_;
+
DISALLOW_COPY_AND_ASSIGN(BuildHashWorkOrder);
};
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/393eba55/relational_operators/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/relational_operators/CMakeLists.txt b/relational_operators/CMakeLists.txt
index a9645b4..8dd65d0 100644
--- a/relational_operators/CMakeLists.txt
+++ b/relational_operators/CMakeLists.txt
@@ -92,6 +92,8 @@ target_link_libraries(quickstep_relationaloperators_AggregationOperator
quickstep_storage_AggregationOperationState
quickstep_storage_StorageBlockInfo
quickstep_utility_Macros
+ quickstep_utility_lipfilter_LIPFilterAdaptiveProber
+ quickstep_utility_lipfilter_LIPFilterUtil
tmb)
target_link_libraries(quickstep_relationaloperators_BuildHashOperator
glog
@@ -111,6 +113,8 @@ target_link_libraries(quickstep_relationaloperators_BuildHashOperator
quickstep_storage_TupleStorageSubBlock
quickstep_storage_ValueAccessor
quickstep_utility_Macros
+ quickstep_utility_lipfilter_LIPFilterBuilder
+ quickstep_utility_lipfilter_LIPFilterUtil
tmb)
target_link_libraries(quickstep_relationaloperators_CreateIndexOperator
glog
@@ -223,6 +227,8 @@ target_link_libraries(quickstep_relationaloperators_HashJoinOperator
quickstep_types_containers_ColumnVector
quickstep_types_containers_ColumnVectorsValueAccessor
quickstep_utility_Macros
+ quickstep_utility_lipfilter_LIPFilterAdaptiveProber
+ quickstep_utility_lipfilter_LIPFilterUtil
tmb)
target_link_libraries(quickstep_relationaloperators_InsertOperator
glog
@@ -322,7 +328,11 @@ target_link_libraries(quickstep_relationaloperators_SelectOperator
quickstep_storage_StorageBlock
quickstep_storage_StorageBlockInfo
quickstep_storage_StorageManager
+ quickstep_storage_TupleIdSequence
+ quickstep_storage_ValueAccessor
quickstep_utility_Macros
+ quickstep_utility_lipfilter_LIPFilterAdaptiveProber
+ quickstep_utility_lipfilter_LIPFilterUtil
tmb)
if(QUICKSTEP_HAVE_LIBNUMA)
target_link_libraries(quickstep_relationaloperators_SelectOperator
@@ -492,6 +502,7 @@ target_link_libraries(quickstep_relationaloperators_WorkOrderFactory
quickstep_relationaloperators_WorkOrder_proto
quickstep_storage_StorageBlockInfo
quickstep_utility_Macros
+ quickstep_utility_lipfilter_LIPFilterUtil
tmb)
target_link_libraries(quickstep_relationaloperators_WorkOrder_proto
quickstep_relationaloperators_SortMergeRunOperator_proto
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/393eba55/relational_operators/HashJoinOperator.cpp
----------------------------------------------------------------------
diff --git a/relational_operators/HashJoinOperator.cpp b/relational_operators/HashJoinOperator.cpp
index 779c0fe..4a91f86 100644
--- a/relational_operators/HashJoinOperator.cpp
+++ b/relational_operators/HashJoinOperator.cpp
@@ -48,6 +48,8 @@
#include "types/TypedValue.hpp"
#include "types/containers/ColumnVector.hpp"
#include "types/containers/ColumnVectorsValueAccessor.hpp"
+#include "utility/lip_filter/LIPFilterAdaptiveProber.hpp"
+#include "utility/lip_filter/LIPFilterUtil.hpp"
#include "gflags/gflags.h"
#include "glog/logging.h"
@@ -95,28 +97,22 @@ class MapBasedJoinedTupleCollector {
class SemiAntiJoinTupleCollector {
public:
- explicit SemiAntiJoinTupleCollector(const TupleStorageSubBlock &tuple_store) {
- filter_.reset(tuple_store.getExistenceMap());
- }
+ explicit SemiAntiJoinTupleCollector(TupleIdSequence *filter)
+ : filter_(filter) {}
template <typename ValueAccessorT>
inline void operator()(const ValueAccessorT &accessor) {
filter_->set(accessor.getCurrentPosition(), false);
}
- const TupleIdSequence* filter() const {
- return filter_.get();
- }
-
private:
- std::unique_ptr<TupleIdSequence> filter_;
+ TupleIdSequence *filter_;
};
class OuterJoinTupleCollector {
public:
- explicit OuterJoinTupleCollector(const TupleStorageSubBlock &tuple_store) {
- filter_.reset(tuple_store.getExistenceMap());
- }
+ explicit OuterJoinTupleCollector(TupleIdSequence *filter)
+ : filter_(filter) {}
template <typename ValueAccessorT>
inline void operator()(const ValueAccessorT &accessor,
@@ -134,14 +130,10 @@ class OuterJoinTupleCollector {
return &joined_tuples_;
}
- const TupleIdSequence* filter() const {
- return filter_.get();
- }
-
private:
std::unordered_map<block_id, std::vector<std::pair<tuple_id, tuple_id>>> joined_tuples_;
// BitVector on the probe relation. 1 if the corresponding tuple has no match.
- std::unique_ptr<TupleIdSequence> filter_;
+ TupleIdSequence *filter_;
};
} // namespace
@@ -203,7 +195,8 @@ bool HashJoinOperator::getAllNonOuterJoinWorkOrders(
selection,
hash_table,
output_destination,
- storage_manager),
+ storage_manager,
+ CreateLIPFilterAdaptiveProberHelper(lip_deployment_index_, query_context)),
op_index_);
}
started_ = true;
@@ -223,7 +216,8 @@ bool HashJoinOperator::getAllNonOuterJoinWorkOrders(
selection,
hash_table,
output_destination,
- storage_manager),
+ storage_manager,
+ CreateLIPFilterAdaptiveProberHelper(lip_deployment_index_, query_context)),
op_index_);
++num_workorders_generated_;
} // end while
@@ -264,7 +258,8 @@ bool HashJoinOperator::getAllOuterJoinWorkOrders(
is_selection_on_build_,
hash_table,
output_destination,
- storage_manager),
+ storage_manager,
+ CreateLIPFilterAdaptiveProberHelper(lip_deployment_index_, query_context)),
op_index_);
}
started_ = true;
@@ -284,7 +279,8 @@ bool HashJoinOperator::getAllOuterJoinWorkOrders(
is_selection_on_build_,
hash_table,
output_destination,
- storage_manager),
+ storage_manager,
+ CreateLIPFilterAdaptiveProberHelper(lip_deployment_index_, query_context)),
op_index_);
++num_workorders_generated_;
}
@@ -360,6 +356,7 @@ serialization::WorkOrder* HashJoinOperator::createNonOuterJoinWorkOrderProto(
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_);
+ proto->SetExtension(serialization::HashJoinWorkOrder::lip_deployment_index, lip_deployment_index_);
return proto;
}
@@ -408,6 +405,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::lip_deployment_index, lip_deployment_index_);
for (const bool is_attribute_on_build : is_selection_on_build_) {
proto->AddExtension(serialization::HashJoinWorkOrder::is_selection_on_build, is_attribute_on_build);
@@ -421,8 +419,19 @@ void HashInnerJoinWorkOrder::execute() {
BlockReference probe_block(
storage_manager_->getBlock(block_id_, probe_relation_));
const TupleStorageSubBlock &probe_store = probe_block->getTupleStorageSubBlock();
-
std::unique_ptr<ValueAccessor> probe_accessor(probe_store.createValueAccessor());
+
+ // Probe the LIPFilters to generate an existence bitmap for probe_accessor, if enabled.
+ std::unique_ptr<TupleIdSequence> existence_map;
+ std::unique_ptr<ValueAccessor> base_accessor;
+ if (lip_filter_adaptive_prober_ != nullptr) {
+ base_accessor.reset(probe_accessor.release());
+ existence_map.reset(
+ lip_filter_adaptive_prober_->filterValueAccessor(base_accessor.get()));
+ probe_accessor.reset(
+ base_accessor->createSharedTupleIdSequenceAdapterVirtual(*existence_map));
+ }
+
MapBasedJoinedTupleCollector collector;
if (join_key_attributes_.size() == 1) {
hash_table_.getAllFromValueAccessor(
@@ -526,9 +535,19 @@ void HashSemiJoinWorkOrder::executeWithResidualPredicate() {
BlockReference probe_block = storage_manager_->getBlock(block_id_,
probe_relation_);
const TupleStorageSubBlock &probe_store = probe_block->getTupleStorageSubBlock();
-
std::unique_ptr<ValueAccessor> probe_accessor(probe_store.createValueAccessor());
+ // Probe the LIPFilters to generate an existence bitmap for probe_accessor, if enabled.
+ std::unique_ptr<TupleIdSequence> existence_map;
+ std::unique_ptr<ValueAccessor> base_accessor;
+ if (lip_filter_adaptive_prober_ != nullptr) {
+ base_accessor.reset(probe_accessor.release());
+ existence_map.reset(
+ lip_filter_adaptive_prober_->filterValueAccessor(base_accessor.get()));
+ probe_accessor.reset(
+ base_accessor->createSharedTupleIdSequenceAdapterVirtual(*existence_map));
+ }
+
// We collect all the matching probe relation tuples, as there's a residual
// preidcate that needs to be applied after collecting these matches.
MapBasedJoinedTupleCollector collector;
@@ -548,7 +567,6 @@ void HashSemiJoinWorkOrder::executeWithResidualPredicate() {
// Get a filter for tuples in the given probe block.
TupleIdSequence filter(probe_store.getMaxTupleID() + 1);
- filter.setRange(0, filter.length(), false);
for (const std::pair<const block_id,
std::vector<std::pair<tuple_id, tuple_id>>>
&build_block_entry : *collector.getJoinedTuples()) {
@@ -607,9 +625,24 @@ void HashSemiJoinWorkOrder::executeWithoutResidualPredicate() {
BlockReference probe_block = storage_manager_->getBlock(block_id_,
probe_relation_);
const TupleStorageSubBlock &probe_store = probe_block->getTupleStorageSubBlock();
-
std::unique_ptr<ValueAccessor> probe_accessor(probe_store.createValueAccessor());
- SemiAntiJoinTupleCollector collector(probe_store);
+
+ // Probe the LIPFilters to generate an existence bitmap for probe_accessor, if enabled.
+ std::unique_ptr<TupleIdSequence> existence_map;
+ std::unique_ptr<ValueAccessor> base_accessor;
+ if (lip_filter_adaptive_prober_ != nullptr) {
+ base_accessor.reset(probe_accessor.release());
+ existence_map.reset(
+ lip_filter_adaptive_prober_->filterValueAccessor(base_accessor.get()));
+ probe_accessor.reset(
+ base_accessor->createSharedTupleIdSequenceAdapterVirtual(*existence_map));
+ }
+
+ if (existence_map == nullptr) {
+ existence_map.reset(probe_store.getExistenceMap());
+ }
+
+ SemiAntiJoinTupleCollector collector(existence_map.get());
// We collect all the probe relation tuples which have at least one matching
// tuple in the build relation. As a performance optimization, the hash table
// just looks for the existence of the probing key in the hash table and sets
@@ -637,7 +670,7 @@ void HashSemiJoinWorkOrder::executeWithoutResidualPredicate() {
probe_block->getIndicesConsistent());
std::unique_ptr<ValueAccessor> probe_accessor_with_filter(
- probe_store.createValueAccessor(collector.filter()));
+ probe_accessor->createSharedTupleIdSequenceAdapterVirtual(*existence_map));
ColumnVectorsValueAccessor temp_result;
for (vector<unique_ptr<const Scalar>>::const_iterator selection_it = selection_.begin();
selection_it != selection_.end(); ++selection_it) {
@@ -654,9 +687,24 @@ void HashAntiJoinWorkOrder::executeWithoutResidualPredicate() {
BlockReference probe_block = storage_manager_->getBlock(block_id_,
probe_relation_);
const TupleStorageSubBlock &probe_store = probe_block->getTupleStorageSubBlock();
-
std::unique_ptr<ValueAccessor> probe_accessor(probe_store.createValueAccessor());
- SemiAntiJoinTupleCollector collector(probe_store);
+
+ // Probe the LIPFilters to generate an existence bitmap for probe_accessor, if enabled.
+ std::unique_ptr<TupleIdSequence> existence_map;
+ std::unique_ptr<ValueAccessor> base_accessor;
+ if (lip_filter_adaptive_prober_ != nullptr) {
+ base_accessor.reset(probe_accessor.release());
+ existence_map.reset(
+ lip_filter_adaptive_prober_->filterValueAccessor(base_accessor.get()));
+ probe_accessor.reset(
+ base_accessor->createSharedTupleIdSequenceAdapterVirtual(*existence_map));
+ }
+
+ if (existence_map == nullptr) {
+ existence_map.reset(probe_store.getExistenceMap());
+ }
+
+ SemiAntiJoinTupleCollector collector(existence_map.get());
// We probe the hash table to find the keys which have an entry in the
// hash table.
if (join_key_attributes_.size() == 1) {
@@ -680,7 +728,7 @@ void HashAntiJoinWorkOrder::executeWithoutResidualPredicate() {
probe_block->getIndicesConsistent());
std::unique_ptr<ValueAccessor> probe_accessor_with_filter(
- probe_store.createValueAccessor(collector.filter()));
+ probe_accessor->createSharedTupleIdSequenceAdapterVirtual(*existence_map));
ColumnVectorsValueAccessor temp_result;
for (vector<unique_ptr<const Scalar>>::const_iterator selection_it = selection_.begin();
selection_it != selection_.end(); ++selection_it) {
@@ -698,8 +746,19 @@ void HashAntiJoinWorkOrder::executeWithResidualPredicate() {
BlockReference probe_block = storage_manager_->getBlock(block_id_,
probe_relation_);
const TupleStorageSubBlock &probe_store = probe_block->getTupleStorageSubBlock();
-
std::unique_ptr<ValueAccessor> probe_accessor(probe_store.createValueAccessor());
+
+ // Probe the LIPFilters to generate an existence bitmap for probe_accessor, if enabled.
+ std::unique_ptr<TupleIdSequence> existence_map;
+ std::unique_ptr<ValueAccessor> base_accessor;
+ if (lip_filter_adaptive_prober_ != nullptr) {
+ base_accessor.reset(probe_accessor.release());
+ existence_map.reset(
+ lip_filter_adaptive_prober_->filterValueAccessor(base_accessor.get()));
+ probe_accessor.reset(
+ base_accessor->createSharedTupleIdSequenceAdapterVirtual(*existence_map));
+ }
+
MapBasedJoinedTupleCollector collector;
// We probe the hash table and get all the matches. Unlike
// executeWithoutResidualPredicate(), we have to collect all the matching
@@ -719,8 +778,12 @@ void HashAntiJoinWorkOrder::executeWithResidualPredicate() {
&collector);
}
- // Create a filter for all the tuples from the given probe block.
- std::unique_ptr<TupleIdSequence> filter(probe_store.getExistenceMap());
+ // If the existence map has not been initialized by the pre-filtering LIPFilters.
+ // Then create it for all the tuples from the given probe block.
+ if (existence_map == nullptr) {
+ existence_map.reset(probe_store.getExistenceMap());
+ }
+
for (const std::pair<const block_id, std::vector<std::pair<tuple_id, tuple_id>>>
&build_block_entry : *collector.getJoinedTuples()) {
// First element of the pair build_block_entry is the build block ID
@@ -733,7 +796,7 @@ void HashAntiJoinWorkOrder::executeWithResidualPredicate() {
std::unique_ptr<ValueAccessor> build_accessor(build_store.createValueAccessor());
for (const std::pair<tuple_id, tuple_id> &hash_match
: build_block_entry.second) {
- if (!filter->get(hash_match.second)) {
+ if (!existence_map->get(hash_match.second)) {
// We have already seen this tuple, skip it.
continue;
}
@@ -743,9 +806,9 @@ void HashAntiJoinWorkOrder::executeWithResidualPredicate() {
*probe_accessor,
probe_relation_id,
hash_match.second)) {
- // Note that the filter marks a match as false, as needed by the anti
- // join definition.
- filter->set(hash_match.second, false);
+ // Note that the existence map marks a match as false, as needed by the
+ // anti join definition.
+ existence_map->set(hash_match.second, false);
}
}
}
@@ -755,7 +818,7 @@ void HashAntiJoinWorkOrder::executeWithResidualPredicate() {
probe_block->getIndicesConsistent());
std::unique_ptr<ValueAccessor> probe_accessor_with_filter(
- probe_store.createValueAccessor(filter.get()));
+ probe_accessor->createSharedTupleIdSequenceAdapterVirtual(*existence_map));
ColumnVectorsValueAccessor temp_result;
for (vector<unique_ptr<const Scalar>>::const_iterator selection_it = selection_.begin();
selection_it != selection_.end();
@@ -775,9 +838,24 @@ void HashOuterJoinWorkOrder::execute() {
const BlockReference probe_block = storage_manager_->getBlock(block_id_,
probe_relation_);
const TupleStorageSubBlock &probe_store = probe_block->getTupleStorageSubBlock();
-
std::unique_ptr<ValueAccessor> probe_accessor(probe_store.createValueAccessor());
- OuterJoinTupleCollector collector(probe_store);
+
+ // Probe the LIPFilters to generate an existence bitmap for probe_accessor, if enabled.
+ std::unique_ptr<TupleIdSequence> existence_map;
+ std::unique_ptr<ValueAccessor> base_accessor;
+ if (lip_filter_adaptive_prober_ != nullptr) {
+ base_accessor.reset(probe_accessor.release());
+ existence_map.reset(
+ lip_filter_adaptive_prober_->filterValueAccessor(base_accessor.get()));
+ probe_accessor.reset(
+ base_accessor->createSharedTupleIdSequenceAdapterVirtual(*existence_map));
+ }
+
+ if (existence_map == nullptr) {
+ existence_map.reset(probe_store.getExistenceMap());
+ }
+
+ OuterJoinTupleCollector collector(existence_map.get());
if (join_key_attributes_.size() == 1) {
hash_table_.getAllFromValueAccessorWithExtraWorkForFirstMatch(
probe_accessor.get(),
@@ -822,11 +900,10 @@ void HashOuterJoinWorkOrder::execute() {
probe_block->getIndicesConsistent());
// Populate the output tuples for non-matches.
- const TupleIdSequence *filter = collector.filter();
- const TupleIdSequence::size_type num_tuples_without_matches = filter->size();
+ const TupleIdSequence::size_type num_tuples_without_matches = existence_map->size();
if (num_tuples_without_matches > 0) {
std::unique_ptr<ValueAccessor> probe_accessor_with_filter(
- probe_store.createValueAccessor(filter));
+ probe_accessor->createSharedTupleIdSequenceAdapterVirtual(*existence_map));
ColumnVectorsValueAccessor temp_result;
for (std::size_t i = 0; i < selection_.size(); ++i) {
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/393eba55/relational_operators/HashJoinOperator.hpp
----------------------------------------------------------------------
diff --git a/relational_operators/HashJoinOperator.hpp b/relational_operators/HashJoinOperator.hpp
index fa393b6..0ed1eeb 100644
--- a/relational_operators/HashJoinOperator.hpp
+++ b/relational_operators/HashJoinOperator.hpp
@@ -35,6 +35,7 @@
#include "storage/HashTable.hpp"
#include "storage/StorageBlockInfo.hpp"
#include "utility/Macros.hpp"
+#include "utility/lip_filter/LIPFilterAdaptiveProber.hpp"
#include "glog/logging.h"
@@ -295,6 +296,7 @@ class HashInnerJoinWorkOrder : public WorkOrder {
* @param hash_table The JoinHashTable to use.
* @param output_destination The InsertDestination to insert the join results.
* @param storage_manager The StorageManager to use.
+ * @param lip_filter_adaptive_prober The attached LIP filter prober.
**/
HashInnerJoinWorkOrder(
const std::size_t query_id,
@@ -307,7 +309,8 @@ class HashInnerJoinWorkOrder : public WorkOrder {
const std::vector<std::unique_ptr<const Scalar>> &selection,
const JoinHashTable &hash_table,
InsertDestination *output_destination,
- StorageManager *storage_manager)
+ StorageManager *storage_manager,
+ LIPFilterAdaptiveProber *lip_filter_adaptive_prober = nullptr)
: WorkOrder(query_id),
build_relation_(build_relation),
probe_relation_(probe_relation),
@@ -318,7 +321,8 @@ class HashInnerJoinWorkOrder : public WorkOrder {
selection_(selection),
hash_table_(hash_table),
output_destination_(DCHECK_NOTNULL(output_destination)),
- storage_manager_(DCHECK_NOTNULL(storage_manager)) {}
+ storage_manager_(DCHECK_NOTNULL(storage_manager)),
+ lip_filter_adaptive_prober_(lip_filter_adaptive_prober) {}
/**
* @brief Constructor for the distributed version.
@@ -342,6 +346,7 @@ class HashInnerJoinWorkOrder : public WorkOrder {
* @param hash_table The JoinHashTable to use.
* @param output_destination The InsertDestination to insert the join results.
* @param storage_manager The StorageManager to use.
+ * @param lip_filter_adaptive_prober The attached LIP filter prober.
**/
HashInnerJoinWorkOrder(
const std::size_t query_id,
@@ -354,7 +359,8 @@ class HashInnerJoinWorkOrder : public WorkOrder {
const std::vector<std::unique_ptr<const Scalar>> &selection,
const JoinHashTable &hash_table,
InsertDestination *output_destination,
- StorageManager *storage_manager)
+ StorageManager *storage_manager,
+ LIPFilterAdaptiveProber *lip_filter_adaptive_prober = nullptr)
: WorkOrder(query_id),
build_relation_(build_relation),
probe_relation_(probe_relation),
@@ -365,7 +371,8 @@ class HashInnerJoinWorkOrder : public WorkOrder {
selection_(selection),
hash_table_(hash_table),
output_destination_(DCHECK_NOTNULL(output_destination)),
- storage_manager_(DCHECK_NOTNULL(storage_manager)) {}
+ storage_manager_(DCHECK_NOTNULL(storage_manager)),
+ lip_filter_adaptive_prober_(lip_filter_adaptive_prober) {}
~HashInnerJoinWorkOrder() override {}
@@ -392,6 +399,8 @@ class HashInnerJoinWorkOrder : public WorkOrder {
InsertDestination *output_destination_;
StorageManager *storage_manager_;
+ std::unique_ptr<LIPFilterAdaptiveProber> lip_filter_adaptive_prober_;
+
DISALLOW_COPY_AND_ASSIGN(HashInnerJoinWorkOrder);
};
@@ -423,6 +432,7 @@ class HashSemiJoinWorkOrder : public WorkOrder {
* @param hash_table The JoinHashTable to use.
* @param output_destination The InsertDestination to insert the join results.
* @param storage_manager The StorageManager to use.
+ * @param lip_filter_adaptive_prober The attached LIP filter prober.
**/
HashSemiJoinWorkOrder(
const std::size_t query_id,
@@ -435,7 +445,8 @@ class HashSemiJoinWorkOrder : public WorkOrder {
const std::vector<std::unique_ptr<const Scalar>> &selection,
const JoinHashTable &hash_table,
InsertDestination *output_destination,
- StorageManager *storage_manager)
+ StorageManager *storage_manager,
+ LIPFilterAdaptiveProber *lip_filter_adaptive_prober = nullptr)
: WorkOrder(query_id),
build_relation_(build_relation),
probe_relation_(probe_relation),
@@ -446,7 +457,8 @@ class HashSemiJoinWorkOrder : public WorkOrder {
selection_(selection),
hash_table_(hash_table),
output_destination_(DCHECK_NOTNULL(output_destination)),
- storage_manager_(DCHECK_NOTNULL(storage_manager)) {}
+ storage_manager_(DCHECK_NOTNULL(storage_manager)),
+ lip_filter_adaptive_prober_(lip_filter_adaptive_prober) {}
/**
* @brief Constructor for the distributed version.
@@ -470,6 +482,7 @@ class HashSemiJoinWorkOrder : public WorkOrder {
* @param hash_table The JoinHashTable to use.
* @param output_destination The InsertDestination to insert the join results.
* @param storage_manager The StorageManager to use.
+ * @param lip_filter_adaptive_prober The attached LIP filter prober.
**/
HashSemiJoinWorkOrder(
const std::size_t query_id,
@@ -482,7 +495,8 @@ class HashSemiJoinWorkOrder : public WorkOrder {
const std::vector<std::unique_ptr<const Scalar>> &selection,
const JoinHashTable &hash_table,
InsertDestination *output_destination,
- StorageManager *storage_manager)
+ StorageManager *storage_manager,
+ LIPFilterAdaptiveProber *lip_filter_adaptive_prober = nullptr)
: WorkOrder(query_id),
build_relation_(build_relation),
probe_relation_(probe_relation),
@@ -493,7 +507,8 @@ class HashSemiJoinWorkOrder : public WorkOrder {
selection_(selection),
hash_table_(hash_table),
output_destination_(DCHECK_NOTNULL(output_destination)),
- storage_manager_(DCHECK_NOTNULL(storage_manager)) {}
+ storage_manager_(DCHECK_NOTNULL(storage_manager)),
+ lip_filter_adaptive_prober_(lip_filter_adaptive_prober) {}
~HashSemiJoinWorkOrder() override {}
@@ -516,6 +531,8 @@ class HashSemiJoinWorkOrder : public WorkOrder {
InsertDestination *output_destination_;
StorageManager *storage_manager_;
+ std::unique_ptr<LIPFilterAdaptiveProber> lip_filter_adaptive_prober_;
+
DISALLOW_COPY_AND_ASSIGN(HashSemiJoinWorkOrder);
};
@@ -547,6 +564,7 @@ class HashAntiJoinWorkOrder : public WorkOrder {
* @param hash_table The JoinHashTable to use.
* @param output_destination The InsertDestination to insert the join results.
* @param storage_manager The StorageManager to use.
+ * @param lip_filter_adaptive_prober The attached LIP filter prober.
**/
HashAntiJoinWorkOrder(
const std::size_t query_id,
@@ -559,7 +577,8 @@ class HashAntiJoinWorkOrder : public WorkOrder {
const std::vector<std::unique_ptr<const Scalar>> &selection,
const JoinHashTable &hash_table,
InsertDestination *output_destination,
- StorageManager *storage_manager)
+ StorageManager *storage_manager,
+ LIPFilterAdaptiveProber *lip_filter_adaptive_prober = nullptr)
: WorkOrder(query_id),
build_relation_(build_relation),
probe_relation_(probe_relation),
@@ -570,7 +589,8 @@ class HashAntiJoinWorkOrder : public WorkOrder {
selection_(selection),
hash_table_(hash_table),
output_destination_(DCHECK_NOTNULL(output_destination)),
- storage_manager_(DCHECK_NOTNULL(storage_manager)) {}
+ storage_manager_(DCHECK_NOTNULL(storage_manager)),
+ lip_filter_adaptive_prober_(lip_filter_adaptive_prober) {}
/**
* @brief Constructor for the distributed version.
@@ -594,6 +614,7 @@ class HashAntiJoinWorkOrder : public WorkOrder {
* @param hash_table The JoinHashTable to use.
* @param output_destination The InsertDestination to insert the join results.
* @param storage_manager The StorageManager to use.
+ * @param lip_filter_adaptive_prober The attached LIP filter prober.
**/
HashAntiJoinWorkOrder(
const std::size_t query_id,
@@ -606,7 +627,8 @@ class HashAntiJoinWorkOrder : public WorkOrder {
const std::vector<std::unique_ptr<const Scalar>> &selection,
const JoinHashTable &hash_table,
InsertDestination *output_destination,
- StorageManager *storage_manager)
+ StorageManager *storage_manager,
+ LIPFilterAdaptiveProber *lip_filter_adaptive_prober = nullptr)
: WorkOrder(query_id),
build_relation_(build_relation),
probe_relation_(probe_relation),
@@ -617,7 +639,8 @@ class HashAntiJoinWorkOrder : public WorkOrder {
selection_(selection),
hash_table_(hash_table),
output_destination_(DCHECK_NOTNULL(output_destination)),
- storage_manager_(DCHECK_NOTNULL(storage_manager)) {}
+ storage_manager_(DCHECK_NOTNULL(storage_manager)),
+ lip_filter_adaptive_prober_(lip_filter_adaptive_prober) {}
~HashAntiJoinWorkOrder() override {}
@@ -646,6 +669,8 @@ class HashAntiJoinWorkOrder : public WorkOrder {
InsertDestination *output_destination_;
StorageManager *storage_manager_;
+ std::unique_ptr<LIPFilterAdaptiveProber> lip_filter_adaptive_prober_;
+
DISALLOW_COPY_AND_ASSIGN(HashAntiJoinWorkOrder);
};
@@ -675,6 +700,7 @@ class HashOuterJoinWorkOrder : public WorkOrder {
* @param hash_table The JoinHashTable to use.
* @param output_destination The InsertDestination to insert the join results.
* @param storage_manager The StorageManager to use.
+ * @param lip_filter_adaptive_prober The attached LIP filter prober.
**/
HashOuterJoinWorkOrder(
const std::size_t query_id,
@@ -687,7 +713,8 @@ class HashOuterJoinWorkOrder : public WorkOrder {
const std::vector<bool> &is_selection_on_build,
const JoinHashTable &hash_table,
InsertDestination *output_destination,
- StorageManager *storage_manager)
+ StorageManager *storage_manager,
+ LIPFilterAdaptiveProber *lip_filter_adaptive_prober)
: WorkOrder(query_id),
build_relation_(build_relation),
probe_relation_(probe_relation),
@@ -698,7 +725,8 @@ class HashOuterJoinWorkOrder : public WorkOrder {
is_selection_on_build_(is_selection_on_build),
hash_table_(hash_table),
output_destination_(output_destination),
- storage_manager_(storage_manager) {}
+ storage_manager_(storage_manager),
+ lip_filter_adaptive_prober_(lip_filter_adaptive_prober) {}
/**
* @brief Constructor for the distributed version.
@@ -733,7 +761,8 @@ class HashOuterJoinWorkOrder : public WorkOrder {
std::vector<bool> &&is_selection_on_build,
const JoinHashTable &hash_table,
InsertDestination *output_destination,
- StorageManager *storage_manager)
+ StorageManager *storage_manager,
+ LIPFilterAdaptiveProber *lip_filter_adaptive_prober)
: WorkOrder(query_id),
build_relation_(build_relation),
probe_relation_(probe_relation),
@@ -744,7 +773,8 @@ class HashOuterJoinWorkOrder : public WorkOrder {
is_selection_on_build_(std::move(is_selection_on_build)),
hash_table_(hash_table),
output_destination_(output_destination),
- storage_manager_(storage_manager) {}
+ storage_manager_(storage_manager),
+ lip_filter_adaptive_prober_(lip_filter_adaptive_prober) {}
~HashOuterJoinWorkOrder() override {}
@@ -763,6 +793,8 @@ class HashOuterJoinWorkOrder : public WorkOrder {
InsertDestination *output_destination_;
StorageManager *storage_manager_;
+ std::unique_ptr<LIPFilterAdaptiveProber> lip_filter_adaptive_prober_;
+
DISALLOW_COPY_AND_ASSIGN(HashOuterJoinWorkOrder);
};
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/393eba55/relational_operators/SelectOperator.cpp
----------------------------------------------------------------------
diff --git a/relational_operators/SelectOperator.cpp b/relational_operators/SelectOperator.cpp
index d56326e..236ee7c 100644
--- a/relational_operators/SelectOperator.cpp
+++ b/relational_operators/SelectOperator.cpp
@@ -30,6 +30,10 @@
#include "storage/StorageBlock.hpp"
#include "storage/StorageBlockInfo.hpp"
#include "storage/StorageManager.hpp"
+#include "storage/TupleIdSequence.hpp"
+#include "storage/ValueAccessor.hpp"
+#include "utility/lip_filter/LIPFilterAdaptiveProber.hpp"
+#include "utility/lip_filter/LIPFilterUtil.hpp"
#include "glog/logging.h"
@@ -40,22 +44,26 @@ namespace quickstep {
class Predicate;
void SelectOperator::addWorkOrders(WorkOrdersContainer *container,
+ QueryContext *query_context,
StorageManager *storage_manager,
const Predicate *predicate,
const std::vector<std::unique_ptr<const Scalar>> *selection,
InsertDestination *output_destination) {
if (input_relation_is_stored_) {
for (const block_id input_block_id : input_relation_block_ids_) {
- container->addNormalWorkOrder(new SelectWorkOrder(query_id_,
- input_relation_,
- input_block_id,
- predicate,
- simple_projection_,
- simple_selection_,
- selection,
- output_destination,
- storage_manager),
- op_index_);
+ container->addNormalWorkOrder(
+ new SelectWorkOrder(
+ query_id_,
+ input_relation_,
+ input_block_id,
+ predicate,
+ simple_projection_,
+ simple_selection_,
+ selection,
+ output_destination,
+ storage_manager,
+ CreateLIPFilterAdaptiveProberHelper(lip_deployment_index_, query_context)),
+ op_index_);
}
} else {
while (num_workorders_generated_ < input_relation_block_ids_.size()) {
@@ -69,7 +77,8 @@ void SelectOperator::addWorkOrders(WorkOrdersContainer *container,
simple_selection_,
selection,
output_destination,
- storage_manager),
+ storage_manager,
+ CreateLIPFilterAdaptiveProberHelper(lip_deployment_index_, query_context)),
op_index_);
++num_workorders_generated_;
}
@@ -78,6 +87,7 @@ void SelectOperator::addWorkOrders(WorkOrdersContainer *container,
#ifdef QUICKSTEP_HAVE_LIBNUMA
void SelectOperator::addPartitionAwareWorkOrders(WorkOrdersContainer *container,
+ QueryContext *query_context,
StorageManager *storage_manager,
const Predicate *predicate,
const std::vector<std::unique_ptr<const Scalar>> *selection,
@@ -99,6 +109,7 @@ void SelectOperator::addPartitionAwareWorkOrders(WorkOrdersContainer *container,
selection,
output_destination,
storage_manager,
+ CreateLIPFilterAdaptiveProberHelper(lip_deployment_index_, query_context),
placement_scheme_->getNUMANodeForBlock(input_block_id)),
op_index_);
}
@@ -120,6 +131,7 @@ void SelectOperator::addPartitionAwareWorkOrders(WorkOrdersContainer *container,
selection,
output_destination,
storage_manager,
+ CreateLIPFilterAdaptiveProberHelper(lip_deployment_index_, query_context),
placement_scheme_->getNUMANodeForBlock(block_in_partition)),
op_index_);
++num_workorders_generated_in_partition_[part_id];
@@ -151,11 +163,21 @@ bool SelectOperator::getAllWorkOrders(
if (input_relation_.hasPartitionScheme()) {
#ifdef QUICKSTEP_HAVE_LIBNUMA
if (input_relation_.hasNUMAPlacementScheme()) {
- addPartitionAwareWorkOrders(container, storage_manager, predicate, selection, output_destination);
+ addPartitionAwareWorkOrders(container,
+ query_context,
+ storage_manager,
+ predicate,
+ selection,
+ output_destination);
}
#endif
} else {
- addWorkOrders(container, storage_manager, predicate, selection, output_destination);
+ addWorkOrders(container,
+ query_context,
+ storage_manager,
+ predicate,
+ selection,
+ output_destination);
}
started_ = true;
}
@@ -164,11 +186,21 @@ bool SelectOperator::getAllWorkOrders(
if (input_relation_.hasPartitionScheme()) {
#ifdef QUICKSTEP_HAVE_LIBNUMA
if (input_relation_.hasNUMAPlacementScheme()) {
- addPartitionAwareWorkOrders(container, storage_manager, predicate, selection, output_destination);
+ addPartitionAwareWorkOrders(container,
+ query_context,
+ storage_manager,
+ predicate,
+ selection,
+ output_destination);
}
#endif
} else {
- addWorkOrders(container, storage_manager, predicate, selection, output_destination);
+ addWorkOrders(container,
+ query_context,
+ storage_manager,
+ predicate,
+ selection,
+ output_destination);
}
return done_feeding_input_relation_;
}
@@ -210,22 +242,41 @@ serialization::WorkOrder* SelectOperator::createWorkOrderProto(const block_id bl
}
}
proto->SetExtension(serialization::SelectWorkOrder::selection_index, selection_index_);
+ proto->SetExtension(serialization::SelectWorkOrder::lip_deployment_index, lip_deployment_index_);
return proto;
}
-
void SelectWorkOrder::execute() {
BlockReference block(
storage_manager_->getBlock(input_block_id_, input_relation_, getPreferredNUMANodes()[0]));
+ // NOTE(jianqiao): For SSB and TPCH queries, experiments show that it is almost
+ // always better to apply the predicate BEFORE the LIPFilters. But as a future
+ // work this ordering may even be adaptive.
+ std::unique_ptr<TupleIdSequence> predicate_matches;
+ if (predicate_ != nullptr) {
+ predicate_matches.reset(
+ block->getMatchesForPredicate(predicate_));
+ }
+
+ std::unique_ptr<TupleIdSequence> matches;
+ if (lip_filter_adaptive_prober_ != nullptr) {
+ std::unique_ptr<ValueAccessor> accessor(
+ block->getTupleStorageSubBlock().createValueAccessor(predicate_matches.get()));
+ matches.reset(
+ lip_filter_adaptive_prober_->filterValueAccessor(accessor.get()));
+ } else {
+ matches.reset(predicate_matches.release());
+ }
+
if (simple_projection_) {
block->selectSimple(simple_selection_,
- predicate_,
+ matches.get(),
output_destination_);
} else {
block->select(*DCHECK_NOTNULL(selection_),
- predicate_,
+ matches.get(),
output_destination_);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/393eba55/relational_operators/SelectOperator.hpp
----------------------------------------------------------------------
diff --git a/relational_operators/SelectOperator.hpp b/relational_operators/SelectOperator.hpp
index 0f5c712..2ace458 100644
--- a/relational_operators/SelectOperator.hpp
+++ b/relational_operators/SelectOperator.hpp
@@ -38,6 +38,7 @@
#include "relational_operators/WorkOrder.hpp"
#include "storage/StorageBlockInfo.hpp"
#include "utility/Macros.hpp"
+#include "utility/lip_filter/LIPFilterAdaptiveProber.hpp"
#include "glog/logging.h"
@@ -49,6 +50,7 @@ namespace quickstep {
class CatalogRelationSchema;
class InsertDestination;
+class LIPFilterDeployment;
class Predicate;
class Scalar;
class StorageManager;
@@ -247,12 +249,14 @@ class SelectOperator : public RelationalOperator {
}
void addWorkOrders(WorkOrdersContainer *container,
+ QueryContext *query_context,
StorageManager *storage_manager,
const Predicate *predicate,
const std::vector<std::unique_ptr<const Scalar>> *selection,
InsertDestination *output_destination);
void addPartitionAwareWorkOrders(WorkOrdersContainer *container,
+ QueryContext *query_context,
StorageManager *storage_manager,
const Predicate *predicate,
const std::vector<std::unique_ptr<const Scalar>> *selection,
@@ -318,6 +322,7 @@ class SelectWorkOrder : public WorkOrder {
* @param output_destination The InsertDestination to insert the selection
* results.
* @param storage_manager The StorageManager to use.
+ * @param lip_filter_adaptive_prober The attached LIP filter prober.
**/
SelectWorkOrder(const std::size_t query_id,
const CatalogRelationSchema &input_relation,
@@ -328,6 +333,7 @@ class SelectWorkOrder : public WorkOrder {
const std::vector<std::unique_ptr<const Scalar>> *selection,
InsertDestination *output_destination,
StorageManager *storage_manager,
+ LIPFilterAdaptiveProber *lip_filter_adaptive_prober = nullptr,
const numa_node_id numa_node = 0)
: WorkOrder(query_id),
input_relation_(input_relation),
@@ -337,7 +343,8 @@ class SelectWorkOrder : public WorkOrder {
simple_selection_(simple_selection),
selection_(selection),
output_destination_(DCHECK_NOTNULL(output_destination)),
- storage_manager_(DCHECK_NOTNULL(storage_manager)) {
+ storage_manager_(DCHECK_NOTNULL(storage_manager)),
+ lip_filter_adaptive_prober_(lip_filter_adaptive_prober) {
preferred_numa_nodes_.push_back(numa_node);
}
@@ -360,6 +367,7 @@ class SelectWorkOrder : public WorkOrder {
* @param output_destination The InsertDestination to insert the selection
* results.
* @param storage_manager The StorageManager to use.
+ * @param lip_filter_adaptive_prober The attached LIP filter prober.
**/
SelectWorkOrder(const std::size_t query_id,
const CatalogRelationSchema &input_relation,
@@ -370,6 +378,7 @@ class SelectWorkOrder : public WorkOrder {
const std::vector<std::unique_ptr<const Scalar>> *selection,
InsertDestination *output_destination,
StorageManager *storage_manager,
+ LIPFilterAdaptiveProber *lip_filter_adaptive_prober = nullptr,
const numa_node_id numa_node = 0)
: WorkOrder(query_id),
input_relation_(input_relation),
@@ -379,7 +388,8 @@ class SelectWorkOrder : public WorkOrder {
simple_selection_(std::move(simple_selection)),
selection_(selection),
output_destination_(DCHECK_NOTNULL(output_destination)),
- storage_manager_(DCHECK_NOTNULL(storage_manager)) {
+ storage_manager_(DCHECK_NOTNULL(storage_manager)),
+ lip_filter_adaptive_prober_(lip_filter_adaptive_prober) {
preferred_numa_nodes_.push_back(numa_node);
}
@@ -407,6 +417,8 @@ class SelectWorkOrder : public WorkOrder {
InsertDestination *output_destination_;
StorageManager *storage_manager_;
+ std::unique_ptr<LIPFilterAdaptiveProber> lip_filter_adaptive_prober_;
+
DISALLOW_COPY_AND_ASSIGN(SelectWorkOrder);
};
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/393eba55/relational_operators/WorkOrder.proto
----------------------------------------------------------------------
diff --git a/relational_operators/WorkOrder.proto b/relational_operators/WorkOrder.proto
index 3eed379..86f34b8 100644
--- a/relational_operators/WorkOrder.proto
+++ b/relational_operators/WorkOrder.proto
@@ -59,6 +59,7 @@ message AggregationWorkOrder {
// All required.
optional uint32 aggr_state_index = 16;
optional fixed64 block_id = 17;
+ optional int32 lip_deployment_index = 18;
}
}
@@ -70,6 +71,7 @@ message BuildHashWorkOrder {
optional bool any_join_key_attributes_nullable = 34;
optional uint32 join_hash_table_index = 35;
optional fixed64 block_id = 36;
+ optional int32 lip_deployment_index = 37;
}
}
@@ -131,6 +133,8 @@ message HashJoinWorkOrder {
optional int32 residual_predicate_index = 169;
// Used by HashOuterJoinWorkOrder only.
repeated bool is_selection_on_build = 170;
+
+ optional int32 lip_deployment_index = 171;
}
}
@@ -185,9 +189,10 @@ message SelectWorkOrder {
// When 'simple_projection' is true.
repeated int32 simple_selection = 245;
-
// Otherwise.
optional int32 selection_index = 246;
+
+ optional int32 lip_deployment_index = 247;
}
}
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/393eba55/relational_operators/WorkOrderFactory.cpp
----------------------------------------------------------------------
diff --git a/relational_operators/WorkOrderFactory.cpp b/relational_operators/WorkOrderFactory.cpp
index 2356bab..91a717f 100644
--- a/relational_operators/WorkOrderFactory.cpp
+++ b/relational_operators/WorkOrderFactory.cpp
@@ -50,6 +50,7 @@
#include "relational_operators/WindowAggregationOperator.hpp"
#include "relational_operators/WorkOrder.pb.h"
#include "storage/StorageBlockInfo.hpp"
+#include "utility/lip_filter/LIPFilterUtil.hpp"
#include "glog/logging.h"
@@ -61,6 +62,7 @@ using std::vector;
namespace quickstep {
class InsertDestination;
+class LIPFilterAdaptiveProber;
class Predicate;
class Scalar;
@@ -82,7 +84,9 @@ WorkOrder* WorkOrderFactory::ReconstructFromProto(const serialization::WorkOrder
proto.query_id(),
proto.GetExtension(serialization::AggregationWorkOrder::block_id),
query_context->getAggregationState(
- proto.GetExtension(serialization::AggregationWorkOrder::aggr_state_index)));
+ proto.GetExtension(serialization::AggregationWorkOrder::aggr_state_index)),
+ CreateLIPFilterAdaptiveProberHelper(
+ proto.GetExtension(serialization::AggregationWorkOrder::lip_deployment_index), query_context));
}
case serialization::BUILD_HASH: {
LOG(INFO) << "Creating BuildHashWorkOrder";
@@ -101,7 +105,9 @@ WorkOrder* WorkOrderFactory::ReconstructFromProto(const serialization::WorkOrder
proto.GetExtension(serialization::BuildHashWorkOrder::block_id),
query_context->getJoinHashTable(
proto.GetExtension(serialization::BuildHashWorkOrder::join_hash_table_index)),
- storage_manager);
+ storage_manager,
+ CreateLIPFilterBuilderHelper(
+ proto.GetExtension(serialization::BuildHashWorkOrder::lip_deployment_index), query_context));
}
case serialization::DELETE: {
LOG(INFO) << "Creating DeleteWorkOrder";
@@ -200,6 +206,9 @@ WorkOrder* WorkOrderFactory::ReconstructFromProto(const serialization::WorkOrder
InsertDestination *output_destination =
query_context->getInsertDestination(
proto.GetExtension(serialization::HashJoinWorkOrder::insert_destination_index));
+ LIPFilterAdaptiveProber *lip_filter_adaptive_prober =
+ CreateLIPFilterAdaptiveProberHelper(
+ proto.GetExtension(serialization::HashJoinWorkOrder::lip_deployment_index), query_context);
switch (hash_join_work_order_type) {
case serialization::HashJoinWorkOrder::HASH_ANTI_JOIN: {
@@ -215,7 +224,8 @@ WorkOrder* WorkOrderFactory::ReconstructFromProto(const serialization::WorkOrder
selection,
hash_table,
output_destination,
- storage_manager);
+ storage_manager,
+ lip_filter_adaptive_prober);
}
case serialization::HashJoinWorkOrder::HASH_INNER_JOIN: {
LOG(INFO) << "Creating HashInnerJoinWorkOrder";
@@ -230,7 +240,8 @@ WorkOrder* WorkOrderFactory::ReconstructFromProto(const serialization::WorkOrder
selection,
hash_table,
output_destination,
- storage_manager);
+ storage_manager,
+ lip_filter_adaptive_prober);
}
case serialization::HashJoinWorkOrder::HASH_OUTER_JOIN: {
vector<bool> is_selection_on_build;
@@ -253,7 +264,8 @@ WorkOrder* WorkOrderFactory::ReconstructFromProto(const serialization::WorkOrder
move(is_selection_on_build),
hash_table,
output_destination,
- storage_manager);
+ storage_manager,
+ lip_filter_adaptive_prober);
}
case serialization::HashJoinWorkOrder::HASH_SEMI_JOIN: {
LOG(INFO) << "Creating HashSemiJoinWorkOrder";
@@ -268,7 +280,8 @@ WorkOrder* WorkOrderFactory::ReconstructFromProto(const serialization::WorkOrder
selection,
hash_table,
output_destination,
- storage_manager);
+ storage_manager,
+ lip_filter_adaptive_prober);
}
default:
LOG(FATAL) << "Unknown HashJoinWorkOrder Type in WorkOrderFactory::ReconstructFromProto";
@@ -346,7 +359,9 @@ WorkOrder* WorkOrderFactory::ReconstructFromProto(const serialization::WorkOrder
proto.GetExtension(serialization::SelectWorkOrder::selection_index)),
query_context->getInsertDestination(
proto.GetExtension(serialization::SelectWorkOrder::insert_destination_index)),
- storage_manager);
+ storage_manager,
+ CreateLIPFilterAdaptiveProberHelper(
+ proto.GetExtension(serialization::HashJoinWorkOrder::lip_deployment_index), query_context));
}
case serialization::SORT_MERGE_RUN: {
LOG(INFO) << "Creating SortMergeRunWorkOrder";
@@ -459,6 +474,17 @@ bool WorkOrderFactory::ProtoIsValid(const serialization::WorkOrder &proto,
switch (proto.work_order_type()) {
case serialization::AGGREGATION: {
+ if (!proto.HasExtension(serialization::AggregationWorkOrder::lip_deployment_index)) {
+ return false;
+ } else {
+ const QueryContext::lip_deployment_id lip_deployment_index =
+ proto.GetExtension(serialization::AggregationWorkOrder::lip_deployment_index);
+ if (lip_deployment_index != QueryContext::kInvalidLIPDeploymentId &&
+ !query_context.isValidLIPDeploymentId(lip_deployment_index)) {
+ return false;
+ }
+ }
+
return proto.HasExtension(serialization::AggregationWorkOrder::block_id) &&
proto.HasExtension(serialization::AggregationWorkOrder::aggr_state_index) &&
query_context.isValidAggregationStateId(
@@ -482,6 +508,17 @@ bool WorkOrderFactory::ProtoIsValid(const serialization::WorkOrder &proto,
}
}
+ if (!proto.HasExtension(serialization::BuildHashWorkOrder::lip_deployment_index)) {
+ return false;
+ } else {
+ const QueryContext::lip_deployment_id lip_deployment_index =
+ proto.GetExtension(serialization::BuildHashWorkOrder::lip_deployment_index);
+ if (lip_deployment_index != QueryContext::kInvalidLIPDeploymentId &&
+ !query_context.isValidLIPDeploymentId(lip_deployment_index)) {
+ return false;
+ }
+ }
+
return proto.HasExtension(serialization::BuildHashWorkOrder::any_join_key_attributes_nullable) &&
proto.HasExtension(serialization::BuildHashWorkOrder::block_id) &&
proto.HasExtension(serialization::BuildHashWorkOrder::join_hash_table_index) &&
@@ -556,6 +593,17 @@ bool WorkOrderFactory::ProtoIsValid(const serialization::WorkOrder &proto,
}
}
+ if (!proto.HasExtension(serialization::HashJoinWorkOrder::lip_deployment_index)) {
+ return false;
+ } else {
+ const QueryContext::lip_deployment_id lip_deployment_index =
+ proto.GetExtension(serialization::HashJoinWorkOrder::lip_deployment_index);
+ if (lip_deployment_index != QueryContext::kInvalidLIPDeploymentId &&
+ !query_context.isValidLIPDeploymentId(lip_deployment_index)) {
+ return false;
+ }
+ }
+
if (hash_join_work_order_type == serialization::HashJoinWorkOrder::HASH_OUTER_JOIN) {
if (!proto.HasExtension(serialization::HashJoinWorkOrder::is_selection_on_build)) {
return false;
@@ -655,6 +703,17 @@ bool WorkOrderFactory::ProtoIsValid(const serialization::WorkOrder &proto,
return false;
}
+ if (!proto.HasExtension(serialization::SelectWorkOrder::lip_deployment_index)) {
+ return false;
+ } else {
+ const QueryContext::lip_deployment_id lip_deployment_index =
+ proto.GetExtension(serialization::SelectWorkOrder::lip_deployment_index);
+ if (lip_deployment_index != QueryContext::kInvalidLIPDeploymentId &&
+ !query_context.isValidLIPDeploymentId(lip_deployment_index)) {
+ return false;
+ }
+ }
+
return proto.HasExtension(serialization::SelectWorkOrder::insert_destination_index) &&
query_context.isValidInsertDestinationId(
proto.GetExtension(serialization::SelectWorkOrder::insert_destination_index)) &&
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/393eba55/storage/AggregationOperationState.cpp
----------------------------------------------------------------------
diff --git a/storage/AggregationOperationState.cpp b/storage/AggregationOperationState.cpp
index 249026d..eb7ca79 100644
--- a/storage/AggregationOperationState.cpp
+++ b/storage/AggregationOperationState.cpp
@@ -46,10 +46,13 @@
#include "storage/StorageBlock.hpp"
#include "storage/StorageBlockInfo.hpp"
#include "storage/StorageManager.hpp"
+#include "storage/TupleIdSequence.hpp"
+#include "storage/ValueAccessor.hpp"
#include "types/TypedValue.hpp"
#include "types/containers/ColumnVector.hpp"
#include "types/containers/ColumnVectorsValueAccessor.hpp"
#include "types/containers/Tuple.hpp"
+#include "utility/lip_filter/LIPFilterAdaptiveProber.hpp"
#include "glog/logging.h"
@@ -331,11 +334,12 @@ bool AggregationOperationState::ProtoIsValid(
return true;
}
-void AggregationOperationState::aggregateBlock(const block_id input_block) {
+void AggregationOperationState::aggregateBlock(const block_id input_block,
+ LIPFilterAdaptiveProber *lip_filter_adaptive_prober) {
if (group_by_list_.empty()) {
aggregateBlockSingleState(input_block);
} else {
- aggregateBlockHashTable(input_block);
+ aggregateBlockHashTable(input_block, lip_filter_adaptive_prober);
}
}
@@ -367,10 +371,13 @@ void AggregationOperationState::aggregateBlockSingleState(
BlockReference block(
storage_manager_->getBlock(input_block, input_relation_));
- // If there is a filter predicate, 'reuse_matches' holds the set of matching
- // tuples so that it can be reused across multiple aggregates (i.e. we only
- // pay the cost of evaluating the predicate once).
- std::unique_ptr<TupleIdSequence> reuse_matches;
+ std::unique_ptr<TupleIdSequence> matches;
+ if (predicate_ != nullptr) {
+ std::unique_ptr<ValueAccessor> accessor(
+ block->getTupleStorageSubBlock().createValueAccessor());
+ matches.reset(block->getMatchesForPredicate(predicate_.get(), matches.get()));
+ }
+
for (std::size_t agg_idx = 0; agg_idx < handles_.size(); ++agg_idx) {
const std::vector<attribute_id> *local_arguments_as_attributes = nullptr;
#ifdef QUICKSTEP_ENABLE_VECTOR_COPY_ELISION_SELECTION
@@ -387,9 +394,8 @@ void AggregationOperationState::aggregateBlockSingleState(
arguments_[agg_idx],
local_arguments_as_attributes,
{}, /* group_by */
- predicate_.get(),
+ matches.get(),
distinctify_hashtables_[agg_idx].get(),
- &reuse_matches,
nullptr /* reuse_group_by_vectors */);
local_state.emplace_back(nullptr);
} else {
@@ -397,8 +403,7 @@ void AggregationOperationState::aggregateBlockSingleState(
local_state.emplace_back(block->aggregate(*handles_[agg_idx],
arguments_[agg_idx],
local_arguments_as_attributes,
- predicate_.get(),
- &reuse_matches));
+ matches.get()));
}
}
@@ -407,14 +412,22 @@ void AggregationOperationState::aggregateBlockSingleState(
}
void AggregationOperationState::aggregateBlockHashTable(
- const block_id input_block) {
+ const block_id input_block,
+ LIPFilterAdaptiveProber *lip_filter_adaptive_prober) {
BlockReference block(
storage_manager_->getBlock(input_block, input_relation_));
- // If there is a filter predicate, 'reuse_matches' holds the set of matching
- // tuples so that it can be reused across multiple aggregates (i.e. we only
- // pay the cost of evaluating the predicate once).
- std::unique_ptr<TupleIdSequence> reuse_matches;
+ // Apply the predicate first, then the LIPFilters, to generate a TupleIdSequence
+ // as the existence map for the tuples.
+ std::unique_ptr<TupleIdSequence> matches;
+ if (predicate_ != nullptr) {
+ matches.reset(block->getMatchesForPredicate(predicate_.get()));
+ }
+ if (lip_filter_adaptive_prober != nullptr) {
+ std::unique_ptr<ValueAccessor> accessor(
+ block->getTupleStorageSubBlock().createValueAccessor(matches.get()));
+ matches.reset(lip_filter_adaptive_prober->filterValueAccessor(accessor.get()));
+ }
// This holds values of all the GROUP BY attributes so that the can be reused
// across multiple aggregates (i.e. we only pay the cost of evaluatin the
@@ -431,9 +444,8 @@ void AggregationOperationState::aggregateBlockHashTable(
arguments_[agg_idx],
nullptr, /* arguments_as_attributes */
group_by_list_,
- predicate_.get(),
+ matches.get(),
distinctify_hashtables_[agg_idx].get(),
- &reuse_matches,
&reuse_group_by_vectors);
}
}
@@ -447,9 +459,8 @@ void AggregationOperationState::aggregateBlockHashTable(
DCHECK(agg_hash_table != nullptr);
block->aggregateGroupBy(arguments_,
group_by_list_,
- predicate_.get(),
+ matches.get(),
agg_hash_table,
- &reuse_matches,
&reuse_group_by_vectors);
group_by_hashtable_pool_->returnHashTable(agg_hash_table);
}
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/393eba55/storage/AggregationOperationState.hpp
----------------------------------------------------------------------
diff --git a/storage/AggregationOperationState.hpp b/storage/AggregationOperationState.hpp
index 3b0f286..c251983 100644
--- a/storage/AggregationOperationState.hpp
+++ b/storage/AggregationOperationState.hpp
@@ -41,6 +41,7 @@ class AggregateFunction;
class CatalogDatabaseLite;
class CatalogRelationSchema;
class InsertDestination;
+class LIPFilterAdaptiveProber;
class StorageManager;
/** \addtogroup Storage
@@ -155,8 +156,11 @@ class AggregationOperationState {
*
* @param input_block The block ID of the storage block where the aggreates
* are going to be computed.
+ * @param lip_filter_adaptive_prober The LIPFilter prober for pre-filtering
+ * the block.
**/
- void aggregateBlock(const block_id input_block);
+ void aggregateBlock(const block_id input_block,
+ LIPFilterAdaptiveProber *lip_filter_adaptive_prober = nullptr);
/**
* @brief Generate the final results for the aggregates managed by this
@@ -185,7 +189,8 @@ class AggregationOperationState {
// Aggregate on input block.
void aggregateBlockSingleState(const block_id input_block);
- void aggregateBlockHashTable(const block_id input_block);
+ void aggregateBlockHashTable(const block_id input_block,
+ LIPFilterAdaptiveProber *lip_filter_adaptive_prober);
void finalizeSingleState(InsertDestination *output_destination);
void finalizeHashTable(InsertDestination *output_destination);
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/393eba55/storage/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/storage/CMakeLists.txt b/storage/CMakeLists.txt
index 325a7cb..0e32cc1 100644
--- a/storage/CMakeLists.txt
+++ b/storage/CMakeLists.txt
@@ -292,11 +292,14 @@ target_link_libraries(quickstep_storage_AggregationOperationState
quickstep_storage_StorageBlock
quickstep_storage_StorageBlockInfo
quickstep_storage_StorageManager
+ quickstep_storage_TupleIdSequence
+ quickstep_storage_ValueAccessor
quickstep_types_TypedValue
quickstep_types_containers_ColumnVector
quickstep_types_containers_ColumnVectorsValueAccessor
quickstep_types_containers_Tuple
- quickstep_utility_Macros)
+ quickstep_utility_Macros
+ quickstep_utility_lipfilter_LIPFilterAdaptiveProber)
target_link_libraries(quickstep_storage_AggregationOperationState_proto
quickstep_expressions_Expressions_proto
quickstep_expressions_aggregation_AggregateFunction_proto
[08/11] incubator-quickstep git commit: Support for performing
partitioned aggregation.
Posted by ji...@apache.org.
Support for performing partitioned aggregation.
- Used for creating a pool of hash tables such that each hash table
belongs to a unique partition.
- The partitioning is done on the group-by keys.
- Wrote a utility function to compute composite hash of a group of
TypedValues.
- Added a check for whether the aggregation is partitioned or not.
- The conditions for whether the aggregation can be partitioned
are as follows:
1. The query has a GROUP BY clause.
2. There are no aggrgeations with a DISTINCT clause.
3. The estimated number of groups are greater than a pre-defined
threshold.
4. The query has at least one aggregation function.
- Method for partitioned aggregation with GROUP BY
- StorageBlock now provides a method for performing GROUP BY aggregation
in a partitioned way.
- The Tuple class now supports a method to compute the hash of the entire
tuple (i.e. hash key is the composite key made up of all the
attributes in the tuple).
- AggregationOperationState calls appropriate method (i.e.
aggregateGroupBy or aggregateGroupByPartitioned) based on the way in
which aggregation is being performed.
Project: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/commit/1e1434ad
Tree: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/tree/1e1434ad
Diff: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/diff/1e1434ad
Branch: refs/heads/exact-filter
Commit: 1e1434ad41db59009b4f30578f3832ff62799b35
Parents: 393eba5
Author: Harshad Deshmukh <hb...@apache.org>
Authored: Wed Oct 26 09:09:52 2016 -0500
Committer: Harshad Deshmukh <hb...@apache.org>
Committed: Wed Oct 26 09:10:09 2016 -0500
----------------------------------------------------------------------
.../FinalizeAggregationOperator.cpp | 30 ++-
.../FinalizeAggregationOperator.hpp | 9 +-
storage/AggregationOperationState.cpp | 195 +++++++++++++++---
storage/AggregationOperationState.hpp | 74 ++++++-
storage/CMakeLists.txt | 12 ++
storage/HashTablePool.hpp | 2 +-
storage/PartitionedHashTablePool.hpp | 204 +++++++++++++++++++
storage/StorageBlock.cpp | 55 +++++
storage/StorageBlock.hpp | 50 +++++
types/containers/CMakeLists.txt | 1 +
types/containers/Tuple.hpp | 8 +
utility/CMakeLists.txt | 6 +
utility/CompositeHash.hpp | 52 +++++
13 files changed, 657 insertions(+), 41 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/1e1434ad/relational_operators/FinalizeAggregationOperator.cpp
----------------------------------------------------------------------
diff --git a/relational_operators/FinalizeAggregationOperator.cpp b/relational_operators/FinalizeAggregationOperator.cpp
index 7e337de..0cbf635 100644
--- a/relational_operators/FinalizeAggregationOperator.cpp
+++ b/relational_operators/FinalizeAggregationOperator.cpp
@@ -41,16 +41,27 @@ bool FinalizeAggregationOperator::getAllWorkOrders(
if (blocking_dependencies_met_ && !started_) {
started_ = true;
- container->addNormalWorkOrder(
- new FinalizeAggregationWorkOrder(
- query_id_,
- query_context->getAggregationState(aggr_state_index_),
- query_context->getInsertDestination(output_destination_index_)),
- op_index_);
+ AggregationOperationState *agg_state =
+ query_context->getAggregationState(aggr_state_index_);
+ DCHECK(agg_state != nullptr);
+ for (int part_id = 0;
+ part_id < static_cast<int>(agg_state->getNumPartitions());
+ ++part_id) {
+ container->addNormalWorkOrder(
+ new FinalizeAggregationWorkOrder(
+ query_id_,
+ agg_state,
+ query_context->getInsertDestination(output_destination_index_),
+ part_id),
+ op_index_);
+ }
}
return started_;
}
+// TODO(quickstep-team) : Think about how the number of partitions could be
+// accessed in this function. Until then, we can't use partitioned aggregation
+// with the distributed version.
bool FinalizeAggregationOperator::getAllWorkOrderProtos(WorkOrderProtosContainer *container) {
if (blocking_dependencies_met_ && !started_) {
started_ = true;
@@ -68,9 +79,12 @@ bool FinalizeAggregationOperator::getAllWorkOrderProtos(WorkOrderProtosContainer
return started_;
}
-
void FinalizeAggregationWorkOrder::execute() {
- state_->finalizeAggregate(output_destination_);
+ if (state_->isAggregatePartitioned()) {
+ state_->finalizeAggregatePartitioned(part_id_, output_destination_);
+ } else {
+ state_->finalizeAggregate(output_destination_);
+ }
}
} // namespace quickstep
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/1e1434ad/relational_operators/FinalizeAggregationOperator.hpp
----------------------------------------------------------------------
diff --git a/relational_operators/FinalizeAggregationOperator.hpp b/relational_operators/FinalizeAggregationOperator.hpp
index 0aeac2a..ae7127a 100644
--- a/relational_operators/FinalizeAggregationOperator.hpp
+++ b/relational_operators/FinalizeAggregationOperator.hpp
@@ -119,13 +119,17 @@ class FinalizeAggregationWorkOrder : public WorkOrder {
* @param state The AggregationState to use.
* @param output_destination The InsertDestination to insert aggregation
* results.
+ * @param part_id The partition ID for which the Finalize aggregation work
+ * order is issued. Ignore if aggregation is not partitioned.
*/
FinalizeAggregationWorkOrder(const std::size_t query_id,
AggregationOperationState *state,
- InsertDestination *output_destination)
+ InsertDestination *output_destination,
+ const int part_id = -1)
: WorkOrder(query_id),
state_(DCHECK_NOTNULL(state)),
- output_destination_(DCHECK_NOTNULL(output_destination)) {}
+ output_destination_(DCHECK_NOTNULL(output_destination)),
+ part_id_(part_id) {}
~FinalizeAggregationWorkOrder() override {}
@@ -134,6 +138,7 @@ class FinalizeAggregationWorkOrder : public WorkOrder {
private:
AggregationOperationState *state_;
InsertDestination *output_destination_;
+ const int part_id_;
DISALLOW_COPY_AND_ASSIGN(FinalizeAggregationWorkOrder);
};
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/1e1434ad/storage/AggregationOperationState.cpp
----------------------------------------------------------------------
diff --git a/storage/AggregationOperationState.cpp b/storage/AggregationOperationState.cpp
index eb7ca79..b942c1b 100644
--- a/storage/AggregationOperationState.cpp
+++ b/storage/AggregationOperationState.cpp
@@ -60,6 +60,14 @@ using std::unique_ptr;
namespace quickstep {
+DEFINE_int32(num_aggregation_partitions,
+ 41,
+ "The number of partitions used for performing the aggregation");
+DEFINE_int32(partition_aggregation_num_groups_threshold,
+ 500000,
+ "The threshold used for deciding whether the aggregation is done "
+ "in a partitioned way or not");
+
AggregationOperationState::AggregationOperationState(
const CatalogRelationSchema &input_relation,
const std::vector<const AggregateFunction *> &aggregate_functions,
@@ -72,6 +80,8 @@ AggregationOperationState::AggregationOperationState(
const std::vector<HashTableImplType> &distinctify_hash_table_impl_types,
StorageManager *storage_manager)
: input_relation_(input_relation),
+ is_aggregate_partitioned_(checkAggregatePartitioned(
+ estimated_num_entries, is_distinct, group_by, aggregate_functions)),
predicate_(predicate),
group_by_list_(std::move(group_by)),
arguments_(std::move(arguments)),
@@ -175,11 +185,10 @@ AggregationOperationState::AggregationOperationState(
key_types.insert(
key_types.end(), argument_types.begin(), argument_types.end());
// TODO(jianqiao): estimated_num_entries is quite inaccurate for
- // estimating
- // the number of entries in the distinctify hash table. We may estimate
- // for each distinct aggregation an estimated_num_distinct_keys value
- // during
- // query optimization, if it worths.
+ // estimating the number of entries in the distinctify hash table.
+ // We may estimate for each distinct aggregation an
+ // estimated_num_distinct_keys value during query optimization, if it's
+ // worth.
distinctify_hashtables_.emplace_back(
AggregationStateFastHashTableFactory::CreateResizable(
*distinctify_hash_table_impl_types_it,
@@ -195,14 +204,24 @@ AggregationOperationState::AggregationOperationState(
}
if (!group_by_handles.empty()) {
- // Aggregation with GROUP BY: create a HashTable pool for per-group
- // states.
- group_by_hashtable_pool_.reset(new HashTablePool(estimated_num_entries,
- hash_table_impl_type,
- group_by_types,
- payload_sizes,
- group_by_handles,
- storage_manager));
+ // Aggregation with GROUP BY: create a HashTable pool.
+ if (!is_aggregate_partitioned_) {
+ group_by_hashtable_pool_.reset(new HashTablePool(estimated_num_entries,
+ hash_table_impl_type,
+ group_by_types,
+ payload_sizes,
+ group_by_handles,
+ storage_manager));
+ } else {
+ partitioned_group_by_hashtable_pool_.reset(
+ new PartitionedHashTablePool(estimated_num_entries,
+ FLAGS_num_aggregation_partitions,
+ hash_table_impl_type,
+ group_by_types,
+ payload_sizes,
+ group_by_handles,
+ storage_manager));
+ }
}
}
}
@@ -450,19 +469,71 @@ void AggregationOperationState::aggregateBlockHashTable(
}
}
- // Call StorageBlock::aggregateGroupBy() to aggregate this block's values
- // directly into the (threadsafe) shared global HashTable for this
- // aggregate.
- DCHECK(group_by_hashtable_pool_ != nullptr);
- AggregationStateHashTableBase *agg_hash_table =
+ if (!is_aggregate_partitioned_) {
+ // Call StorageBlock::aggregateGroupBy() to aggregate this block's values
+ // directly into the (threadsafe) shared global HashTable for this
+ // aggregate.
+ DCHECK(group_by_hashtable_pool_ != nullptr);
+ AggregationStateHashTableBase *agg_hash_table =
group_by_hashtable_pool_->getHashTableFast();
- DCHECK(agg_hash_table != nullptr);
- block->aggregateGroupBy(arguments_,
- group_by_list_,
- matches.get(),
- agg_hash_table,
- &reuse_group_by_vectors);
- group_by_hashtable_pool_->returnHashTable(agg_hash_table);
+ DCHECK(agg_hash_table != nullptr);
+ block->aggregateGroupBy(arguments_,
+ group_by_list_,
+ matches.get(),
+ agg_hash_table,
+ &reuse_group_by_vectors);
+ group_by_hashtable_pool_->returnHashTable(agg_hash_table);
+ } else {
+ ColumnVectorsValueAccessor temp_result;
+ // IDs of 'arguments' as attributes in the ValueAccessor we create below.
+ std::vector<attribute_id> argument_ids;
+
+ // IDs of GROUP BY key element(s) in the ValueAccessor we create below.
+ std::vector<attribute_id> key_ids;
+ const std::size_t num_partitions = partitioned_group_by_hashtable_pool_->getNumPartitions();
+ block->aggregateGroupByPartitioned(
+ arguments_,
+ group_by_list_,
+ matches.get(),
+ num_partitions,
+ &temp_result,
+ &argument_ids,
+ &key_ids,
+ &reuse_group_by_vectors);
+ // Compute the partitions for the tuple formed by group by values.
+ std::vector<std::unique_ptr<TupleIdSequence>> partition_membership;
+ partition_membership.resize(num_partitions);
+
+ // Create a tuple-id sequence for each partition.
+ for (std::size_t partition = 0; partition < num_partitions; ++partition) {
+ partition_membership[partition].reset(
+ new TupleIdSequence(temp_result.getEndPosition()));
+ }
+
+ // Iterate over ValueAccessor for each tuple,
+ // set a bit in the appropriate TupleIdSequence.
+ temp_result.beginIteration();
+ while (temp_result.next()) {
+ // We need a unique_ptr because getTupleWithAttributes() uses "new".
+ std::unique_ptr<Tuple> curr_tuple(temp_result.getTupleWithAttributes(key_ids));
+ const std::size_t curr_tuple_partition_id =
+ curr_tuple->getTupleHash() % num_partitions;
+ partition_membership[curr_tuple_partition_id]->set(
+ temp_result.getCurrentPosition(), true);
+ }
+ // For each partition, create an adapter around Value Accessor and
+ // TupleIdSequence.
+ std::vector<std::unique_ptr<
+ TupleIdSequenceAdapterValueAccessor<ColumnVectorsValueAccessor>>> adapter;
+ adapter.resize(num_partitions);
+ for (std::size_t partition = 0; partition < num_partitions; ++partition) {
+ adapter[partition].reset(temp_result.createSharedTupleIdSequenceAdapter(
+ *(partition_membership)[partition]));
+ partitioned_group_by_hashtable_pool_->getHashTable(partition)
+ ->upsertValueAccessorCompositeKeyFast(
+ argument_ids, adapter[partition].get(), key_ids, true);
+ }
+ }
}
void AggregationOperationState::finalizeSingleState(
@@ -606,13 +677,81 @@ void AggregationOperationState::finalizeHashTable(
}
void AggregationOperationState::destroyAggregationHashTablePayload() {
- if (group_by_hashtable_pool_ != nullptr) {
- auto all_hash_tables = group_by_hashtable_pool_->getAllHashTables();
- DCHECK(all_hash_tables != nullptr);
+ std::vector<std::unique_ptr<AggregationStateHashTableBase>> *all_hash_tables =
+ nullptr;
+ if (!is_aggregate_partitioned_) {
+ if (group_by_hashtable_pool_ != nullptr) {
+ all_hash_tables = group_by_hashtable_pool_->getAllHashTables();
+ }
+ } else {
+ if (partitioned_group_by_hashtable_pool_ != nullptr) {
+ all_hash_tables = partitioned_group_by_hashtable_pool_->getAllHashTables();
+ }
+ }
+ if (all_hash_tables != nullptr) {
for (std::size_t ht_index = 0; ht_index < all_hash_tables->size(); ++ht_index) {
(*all_hash_tables)[ht_index]->destroyPayload();
}
}
}
+void AggregationOperationState::finalizeAggregatePartitioned(
+ const std::size_t partition_id, InsertDestination *output_destination) {
+ // Each element of 'group_by_keys' is a vector of values for a particular
+ // group (which is also the prefix of the finalized Tuple for that group).
+ std::vector<std::vector<TypedValue>> group_by_keys;
+
+ // Collect per-aggregate finalized values.
+ std::vector<std::unique_ptr<ColumnVector>> final_values;
+ for (std::size_t agg_idx = 0; agg_idx < handles_.size(); ++agg_idx) {
+ AggregationStateHashTableBase *hash_table =
+ partitioned_group_by_hashtable_pool_->getHashTable(partition_id);
+ ColumnVector *agg_result_col = handles_[agg_idx]->finalizeHashTable(
+ *hash_table, &group_by_keys, agg_idx);
+ if (agg_result_col != nullptr) {
+ final_values.emplace_back(agg_result_col);
+ }
+ }
+
+ // Reorganize 'group_by_keys' in column-major order so that we can make a
+ // ColumnVectorsValueAccessor to bulk-insert results.
+ //
+ // TODO(chasseur): Shuffling around the GROUP BY keys like this is suboptimal
+ // if there is only one aggregate. The need to do this should hopefully go
+ // away when we work out storing composite structures for multiple aggregates
+ // in a single HashTable.
+ std::vector<std::unique_ptr<ColumnVector>> group_by_cvs;
+ std::size_t group_by_element_idx = 0;
+ for (const std::unique_ptr<const Scalar> &group_by_element : group_by_list_) {
+ const Type &group_by_type = group_by_element->getType();
+ if (NativeColumnVector::UsableForType(group_by_type)) {
+ NativeColumnVector *element_cv = new NativeColumnVector(group_by_type, group_by_keys.size());
+ group_by_cvs.emplace_back(element_cv);
+ for (std::vector<TypedValue> &group_key : group_by_keys) {
+ element_cv->appendTypedValue(std::move(group_key[group_by_element_idx]));
+ }
+ } else {
+ IndirectColumnVector *element_cv = new IndirectColumnVector(group_by_type, group_by_keys.size());
+ group_by_cvs.emplace_back(element_cv);
+ for (std::vector<TypedValue> &group_key : group_by_keys) {
+ element_cv->appendTypedValue(std::move(group_key[group_by_element_idx]));
+ }
+ }
+ ++group_by_element_idx;
+ }
+
+ // Stitch together a ColumnVectorsValueAccessor combining the GROUP BY keys
+ // and the finalized aggregates.
+ ColumnVectorsValueAccessor complete_result;
+ for (std::unique_ptr<ColumnVector> &group_by_cv : group_by_cvs) {
+ complete_result.addColumn(group_by_cv.release());
+ }
+ for (std::unique_ptr<ColumnVector> &final_value_cv : final_values) {
+ complete_result.addColumn(final_value_cv.release());
+ }
+
+ // Bulk-insert the complete result.
+ output_destination->bulkInsertTuples(&complete_result);
+}
+
} // namespace quickstep
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/1e1434ad/storage/AggregationOperationState.hpp
----------------------------------------------------------------------
diff --git a/storage/AggregationOperationState.hpp b/storage/AggregationOperationState.hpp
index c251983..e0826b0 100644
--- a/storage/AggregationOperationState.hpp
+++ b/storage/AggregationOperationState.hpp
@@ -32,9 +32,12 @@
#include "storage/AggregationOperationState.pb.h"
#include "storage/HashTableBase.hpp"
#include "storage/HashTablePool.hpp"
+#include "storage/PartitionedHashTablePool.hpp"
#include "storage/StorageBlockInfo.hpp"
#include "utility/Macros.hpp"
+#include "gflags/gflags.h"
+
namespace quickstep {
class AggregateFunction;
@@ -44,6 +47,9 @@ class InsertDestination;
class LIPFilterAdaptiveProber;
class StorageManager;
+DECLARE_int32(num_aggregation_partitions);
+DECLARE_int32(partition_aggregation_num_groups_threshold);
+
/** \addtogroup Storage
* @{
*/
@@ -176,9 +182,39 @@ class AggregationOperationState {
**/
void destroyAggregationHashTablePayload();
+ /**
+ * @brief Generate the final results for the aggregates managed by this
+ * AggregationOperationState and write them out to StorageBlock(s).
+ * In this implementation, each thread picks a hash table belonging to
+ * a partition and writes its values to StorageBlock(s). There is no
+ * need to merge multiple hash tables in one, because there is no
+ * overlap in the keys across two hash tables.
+ *
+ * @param partition_id The ID of the partition for which finalize is being
+ * performed.
+ * @param output_destination An InsertDestination where the finalized output
+ * tuple(s) from this aggregate are to be written.
+ **/
+ void finalizeAggregatePartitioned(
+ const std::size_t partition_id, InsertDestination *output_destination);
+
static void mergeGroupByHashTables(AggregationStateHashTableBase *src,
AggregationStateHashTableBase *dst);
+ bool isAggregatePartitioned() const {
+ return is_aggregate_partitioned_;
+ }
+
+ /**
+ * @brief Get the number of partitions to be used for the aggregation.
+ * For non-partitioned aggregations, we return 1.
+ **/
+ std::size_t getNumPartitions() const {
+ return is_aggregate_partitioned_
+ ? partitioned_group_by_hashtable_pool_->getNumPartitions()
+ : 1;
+ }
+
int dflag;
private:
@@ -195,12 +231,41 @@ class AggregationOperationState {
void finalizeSingleState(InsertDestination *output_destination);
void finalizeHashTable(InsertDestination *output_destination);
- // A vector of group by hash table pools.
- std::unique_ptr<HashTablePool> group_by_hashtable_pool_;
+ bool checkAggregatePartitioned(
+ const std::size_t estimated_num_groups,
+ const std::vector<bool> &is_distinct,
+ const std::vector<std::unique_ptr<const Scalar>> &group_by,
+ const std::vector<const AggregateFunction *> &aggregate_functions) const {
+ // If there's no aggregation, return false.
+ if (aggregate_functions.empty()) {
+ return false;
+ }
+ // Check if there's a distinct operation involved in any aggregate, if so
+ // the aggregate can't be partitioned.
+ for (auto distinct : is_distinct) {
+ if (distinct) {
+ return false;
+ }
+ }
+ // There's no distinct aggregation involved, Check if there's at least one
+ // GROUP BY operation.
+ if (group_by.empty()) {
+ return false;
+ }
+ // There are GROUP BYs without DISTINCT. Check if the estimated number of
+ // groups is large enough to warrant a partitioned aggregation.
+ return estimated_num_groups >
+ static_cast<std::size_t>(
+ FLAGS_partition_aggregation_num_groups_threshold);
+ }
// Common state for all aggregates in this operation: the input relation, the
// filter predicate (if any), and the list of GROUP BY expressions (if any).
const CatalogRelationSchema &input_relation_;
+
+ // Whether the aggregation is partitioned or not.
+ const bool is_aggregate_partitioned_;
+
std::unique_ptr<const Predicate> predicate_;
std::vector<std::unique_ptr<const Scalar>> group_by_list_;
@@ -233,6 +298,11 @@ class AggregationOperationState {
std::vector<std::unique_ptr<AggregationStateHashTableBase>>
group_by_hashtables_;
+ // A vector of group by hash table pools.
+ std::unique_ptr<HashTablePool> group_by_hashtable_pool_;
+
+ std::unique_ptr<PartitionedHashTablePool> partitioned_group_by_hashtable_pool_;
+
StorageManager *storage_manager_;
DISALLOW_COPY_AND_ASSIGN(AggregationOperationState);
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/1e1434ad/storage/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/storage/CMakeLists.txt b/storage/CMakeLists.txt
index 0e32cc1..be60662 100644
--- a/storage/CMakeLists.txt
+++ b/storage/CMakeLists.txt
@@ -235,6 +235,7 @@ add_library(quickstep_storage_PackedRowStoreTupleStorageSubBlock
add_library(quickstep_storage_PackedRowStoreValueAccessor
../empty_src.cpp
PackedRowStoreValueAccessor.hpp)
+add_library(quickstep_storage_PartitionedHashTablePool ../empty_src.cpp PartitionedHashTablePool.hpp)
add_library(quickstep_storage_PreloaderThread PreloaderThread.cpp PreloaderThread.hpp)
add_library(quickstep_storage_SMAIndexSubBlock SMAIndexSubBlock.cpp SMAIndexSubBlock.hpp)
add_library(quickstep_storage_SeparateChainingHashTable ../empty_src.cpp SeparateChainingHashTable.hpp)
@@ -270,6 +271,7 @@ add_library(quickstep_storage_WindowAggregationOperationState_proto ${storage_Wi
# Link dependencies:
target_link_libraries(quickstep_storage_AggregationOperationState
+ ${GFLAGS_LIB_NAME}
glog
quickstep_catalog_CatalogDatabaseLite
quickstep_catalog_CatalogRelationSchema
@@ -289,6 +291,7 @@ target_link_libraries(quickstep_storage_AggregationOperationState
quickstep_storage_HashTableFactory
quickstep_storage_HashTablePool
quickstep_storage_InsertDestination
+ quickstep_storage_PartitionedHashTablePool
quickstep_storage_StorageBlock
quickstep_storage_StorageBlockInfo
quickstep_storage_StorageManager
@@ -852,6 +855,14 @@ target_link_libraries(quickstep_storage_PackedRowStoreValueAccessor
quickstep_types_TypedValue
quickstep_utility_BitVector
quickstep_utility_Macros)
+target_link_libraries(quickstep_storage_PartitionedHashTablePool
+ glog
+ quickstep_expressions_aggregation_AggregationHandle
+ quickstep_storage_FastHashTable
+ quickstep_storage_FastHashTableFactory
+ quickstep_storage_HashTableBase
+ quickstep_utility_Macros
+ quickstep_utility_StringUtil)
target_link_libraries(quickstep_storage_PreloaderThread
glog
quickstep_catalog_CatalogDatabase
@@ -1169,6 +1180,7 @@ target_link_libraries(quickstep_storage
quickstep_storage_LinearOpenAddressingHashTable
quickstep_storage_PackedRowStoreTupleStorageSubBlock
quickstep_storage_PackedRowStoreValueAccessor
+ quickstep_storage_PartitionedHashTablePool
quickstep_storage_PreloaderThread
quickstep_storage_SMAIndexSubBlock
quickstep_storage_SeparateChainingHashTable
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/1e1434ad/storage/HashTablePool.hpp
----------------------------------------------------------------------
diff --git a/storage/HashTablePool.hpp b/storage/HashTablePool.hpp
index 3cdfcb3..96cf849 100644
--- a/storage/HashTablePool.hpp
+++ b/storage/HashTablePool.hpp
@@ -173,7 +173,7 @@ class HashTablePool {
* @param All the hash tables in the pool.
*
**/
- const std::vector<std::unique_ptr<AggregationStateHashTableBase>>*
+ std::vector<std::unique_ptr<AggregationStateHashTableBase>>*
getAllHashTables() {
return &hash_tables_;
}
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/1e1434ad/storage/PartitionedHashTablePool.hpp
----------------------------------------------------------------------
diff --git a/storage/PartitionedHashTablePool.hpp b/storage/PartitionedHashTablePool.hpp
new file mode 100644
index 0000000..bcc4d23
--- /dev/null
+++ b/storage/PartitionedHashTablePool.hpp
@@ -0,0 +1,204 @@
+/**
+ * Copyright 2016, Quickstep Research Group, Computer Sciences Department,
+ * University of Wisconsin\u2014Madison.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ **/
+
+#ifndef QUICKSTEP_STORAGE_PARTITIONED_HASH_TABLE_POOL_HPP_
+#define QUICKSTEP_STORAGE_PARTITIONED_HASH_TABLE_POOL_HPP_
+
+#include <algorithm>
+#include <chrono>
+#include <memory>
+#include <utility>
+#include <vector>
+
+#include "expressions/aggregation/AggregationHandle.hpp"
+#include "storage/HashTableBase.hpp"
+#include "storage/FastHashTable.hpp"
+#include "storage/FastHashTableFactory.hpp"
+#include "utility/Macros.hpp"
+#include "utility/StringUtil.hpp"
+
+#include "glog/logging.h"
+
+namespace quickstep {
+
+class StorageManager;
+class Type;
+
+/** \addtogroup Storage
+ * @{
+ */
+
+/**
+ * @brief A pool of HashTables used for a single aggregation handle. Each
+ * HashTable represents values from a given partition, which is
+ * determined by the keys in the group by clause.
+ **/
+class PartitionedHashTablePool {
+ public:
+ /**
+ * @brief Constructor.
+ *
+ * @param estimated_num_entries The maximum number of entries in a hash table.
+ * @param num_partitions The number of partitions (i.e. number of HashTables)
+ * @param hash_table_impl_type The type of hash table implementation.
+ * @param group_by_types A vector of pointer of types which form the group by
+ * key.
+ * @param agg_handle The aggregation handle.
+ * @param storage_manager A pointer to the storage manager.
+ **/
+ PartitionedHashTablePool(const std::size_t estimated_num_entries,
+ const std::size_t num_partitions,
+ const HashTableImplType hash_table_impl_type,
+ const std::vector<const Type *> &group_by_types,
+ AggregationHandle *agg_handle,
+ StorageManager *storage_manager)
+ : estimated_num_entries_(
+ setHashTableSize(estimated_num_entries, num_partitions)),
+ num_partitions_(num_partitions),
+ hash_table_impl_type_(hash_table_impl_type),
+ group_by_types_(group_by_types),
+ agg_handle_(DCHECK_NOTNULL(agg_handle)),
+ storage_manager_(DCHECK_NOTNULL(storage_manager)) {
+ initializeAllHashTables();
+ }
+
+ /**
+ * @brief Constructor.
+ *
+ * @note This constructor is relevant for the HashTable specialized for
+ * aggregation.
+ *
+ * @param estimated_num_entries The maximum number of entries in a hash table.
+ * @param num_partitions The number of partitions (i.e. number of HashTables)
+ * @param hash_table_impl_type The type of hash table implementation.
+ * @param group_by_types A vector of pointer of types which form the group by
+ * key.
+ * @param payload_sizes The sizes of the payload elements (i.e.
+ * AggregationStates).
+ * @param handles The aggregation handles.
+ * @param storage_manager A pointer to the storage manager.
+ **/
+ PartitionedHashTablePool(const std::size_t estimated_num_entries,
+ const std::size_t num_partitions,
+ const HashTableImplType hash_table_impl_type,
+ const std::vector<const Type *> &group_by_types,
+ const std::vector<std::size_t> &payload_sizes,
+ const std::vector<AggregationHandle *> &handles,
+ StorageManager *storage_manager)
+ : estimated_num_entries_(
+ setHashTableSize(estimated_num_entries, num_partitions)),
+ num_partitions_(num_partitions),
+ hash_table_impl_type_(hash_table_impl_type),
+ group_by_types_(group_by_types),
+ payload_sizes_(payload_sizes),
+ handles_(handles),
+ storage_manager_(DCHECK_NOTNULL(storage_manager)) {
+ initializeAllHashTables();
+ }
+
+ /**
+ * @brief Check out a hash table for insertion.
+ *
+ * @param partition_id The ID of the partitioned HashTable.
+ *
+ * @return A hash table pointer for the given HashTable.
+ **/
+ AggregationStateHashTableBase* getHashTable(const std::size_t partition_id) {
+ DCHECK_LT(partition_id, num_partitions_);
+ DCHECK_LT(partition_id, hash_tables_.size());
+ return hash_tables_[partition_id].get();
+ }
+
+ /**
+ * @brief Get all the hash tables from the pool.
+ *
+ * @warning The caller should ensure that this call is made when no hash table
+ * is being checked in or checked out from the pool. In other words
+ * the hash table pool is in read-only state.
+ *
+ * @param All the hash tables in the pool.
+ *
+ **/
+ std::vector<std::unique_ptr<AggregationStateHashTableBase>>*
+ getAllHashTables() {
+ return &hash_tables_;
+ }
+
+ /**
+ * @brief Get the number of partitions used for the aggregation.
+ **/
+ inline std::size_t getNumPartitions() const {
+ return num_partitions_;
+ }
+
+ private:
+ void initializeAllHashTables() {
+ for (std::size_t part_num = 0; part_num < num_partitions_; ++part_num) {
+ AggregationStateHashTableBase *part_hash_table = createNewHashTableFast();
+ hash_tables_.push_back(
+ std::unique_ptr<AggregationStateHashTableBase>(part_hash_table));
+ }
+ }
+
+ AggregationStateHashTableBase* createNewHashTable() {
+ return agg_handle_->createGroupByHashTable(hash_table_impl_type_,
+ group_by_types_,
+ estimated_num_entries_,
+ storage_manager_);
+ }
+
+ AggregationStateHashTableBase* createNewHashTableFast() {
+ return AggregationStateFastHashTableFactory::CreateResizable(
+ hash_table_impl_type_,
+ group_by_types_,
+ estimated_num_entries_,
+ payload_sizes_,
+ handles_,
+ storage_manager_);
+ }
+
+ inline std::size_t setHashTableSize(const std::size_t overall_estimate,
+ const std::size_t num_partitions) const {
+ CHECK_NE(num_partitions, 0Lu);
+ // The minimum size of the hash table is set to 100.
+ return std::max(static_cast<std::size_t>(overall_estimate / num_partitions),
+ 100Lu);
+ }
+
+ std::vector<std::unique_ptr<AggregationStateHashTableBase>> hash_tables_;
+
+ const std::size_t estimated_num_entries_;
+ const std::size_t num_partitions_;
+
+ const HashTableImplType hash_table_impl_type_;
+
+ const std::vector<const Type *> group_by_types_;
+
+ std::vector<std::size_t> payload_sizes_;
+
+ AggregationHandle *agg_handle_;
+ const std::vector<AggregationHandle *> handles_;
+ StorageManager *storage_manager_;
+
+ DISALLOW_COPY_AND_ASSIGN(PartitionedHashTablePool);
+};
+
+/** @} */
+
+} // namespace quickstep
+
+#endif // QUICKSTEP_STORAGE_HASH_TABLE_POOL_HPP_
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/1e1434ad/storage/StorageBlock.cpp
----------------------------------------------------------------------
diff --git a/storage/StorageBlock.cpp b/storage/StorageBlock.cpp
index dd3e19d..ea74ee6 100644
--- a/storage/StorageBlock.cpp
+++ b/storage/StorageBlock.cpp
@@ -1329,4 +1329,59 @@ const std::size_t StorageBlock::getNumTuples() const {
return tuple_store_->numTuples();
}
+void StorageBlock::aggregateGroupByPartitioned(
+ const std::vector<std::vector<std::unique_ptr<const Scalar>>> &arguments,
+ const std::vector<std::unique_ptr<const Scalar>> &group_by,
+ const TupleIdSequence *filter,
+ const std::size_t num_partitions,
+ ColumnVectorsValueAccessor *temp_result,
+ std::vector<attribute_id> *argument_ids,
+ std::vector<attribute_id> *key_ids,
+ std::vector<std::unique_ptr<ColumnVector>> *reuse_group_by_vectors) const {
+ DCHECK(!group_by.empty())
+ << "Called aggregateGroupByPartitioned() with zero GROUP BY expressions";
+
+ SubBlocksReference sub_blocks_ref(*tuple_store_,
+ indices_,
+ indices_consistent_);
+
+ std::unique_ptr<ValueAccessor> accessor(
+ tuple_store_->createValueAccessor(filter));
+
+ attribute_id attr_id = 0;
+
+ // First, put GROUP BY keys into 'temp_result'.
+ if (reuse_group_by_vectors->empty()) {
+ // Compute GROUP BY values from group_by Scalars, and store them in
+ // reuse_group_by_vectors for reuse by other aggregates on this same
+ // block.
+ reuse_group_by_vectors->reserve(group_by.size());
+ for (const std::unique_ptr<const Scalar> &group_by_element : group_by) {
+ reuse_group_by_vectors->emplace_back(
+ group_by_element->getAllValues(accessor.get(), &sub_blocks_ref));
+ temp_result->addColumn(reuse_group_by_vectors->back().get(), false);
+ key_ids->push_back(attr_id++);
+ }
+ } else {
+ // Reuse precomputed GROUP BY values from reuse_group_by_vectors.
+ DCHECK_EQ(group_by.size(), reuse_group_by_vectors->size())
+ << "Wrong number of reuse_group_by_vectors";
+ for (const std::unique_ptr<ColumnVector> &reuse_cv : *reuse_group_by_vectors) {
+ temp_result->addColumn(reuse_cv.get(), false);
+ key_ids->push_back(attr_id++);
+ }
+ }
+
+ // Compute argument vectors and add them to 'temp_result'.
+ for (const std::vector<std::unique_ptr<const Scalar>> &argument : arguments) {
+ for (const std::unique_ptr<const Scalar> &args : argument) {
+ temp_result->addColumn(args->getAllValues(accessor.get(), &sub_blocks_ref));
+ argument_ids->push_back(attr_id++);
+ }
+ if (argument.empty()) {
+ argument_ids->push_back(kInvalidAttributeID);
+ }
+ }
+}
+
} // namespace quickstep
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/1e1434ad/storage/StorageBlock.hpp
----------------------------------------------------------------------
diff --git a/storage/StorageBlock.hpp b/storage/StorageBlock.hpp
index 488efcc..56b3bdc 100644
--- a/storage/StorageBlock.hpp
+++ b/storage/StorageBlock.hpp
@@ -43,6 +43,7 @@ class AggregationHandle;
class AggregationState;
class CatalogRelationSchema;
class ColumnVector;
+class ColumnVectorsValueAccessor;
class InsertDestinationInterface;
class Predicate;
class Scalar;
@@ -451,6 +452,55 @@ class StorageBlock : public StorageBlockBase {
AggregationStateHashTableBase *hash_table,
std::vector<std::unique_ptr<ColumnVector>> *reuse_group_by_vectors) const;
+
+ /**
+ * @brief Perform the GROUP BY aggregation for the case when aggregation is
+ * partitioned.
+ *
+ * TODO(harshad) - Refactor this class to use only one function
+ * aggregateGroupBy.
+ * @note The difference between this method and the aggregateGroupBy method
+ * is that in this method, the tuples are routed to different HashTables
+ * based on the partition to which they belong to. The partition is
+ * determined by the GROUP BY attributes. Right now hash based
+ * partitioning is performed.
+ *
+ * @note This function only creates the ColumnVectorsValueAccessor needed for
+ * the insertion in the hash table. The actual insertion in respective
+ * hash tables should be handled by the caller. See
+ * AggregationOperationState::aggregateHashTable() for one such
+ * implementation.
+ *
+ * @param arguments The arguments to the aggregation function as Scalars.
+ * @param group_by The list of GROUP BY attributes/expressions. The tuples in
+ * this storage block are grouped by these attributes before
+ * aggregation.
+ * @param filter If non-NULL, then only tuple IDs which are set in the
+ * filter will be checked (all others will be assumed to be false).
+ * @param num_partitions The number of partitions used for the aggregation.
+ * @param temp_result The ColumnVectorsValueAccessor used for collecting
+ * the attribute values from this StorageBlock.
+ * @param arguments_ids The attribute IDs used for the aggregation, which
+ * come from the arguments vector. If arguments is empty, this vector
+ * is filled with invalid attribute IDs.
+ * @param key_ids The attribute IDs of the group by attributes.
+ * @param reuse_group_by_vectors This parameter is used to store and reuse
+ * GROUP BY attribute vectors pre-computed in an earlier invocation of
+ * aggregateGroupBy(). \c reuse_group_by_vectors is never \c nullptr
+ * for ease of use. Current invocation of aggregateGroupBy() will reuse
+ * ColumnVectors if non-empty, otherwise computes ColumnVectors based
+ * on \c group_by and stores them in \c reuse_group_by_vectors.
+ **/
+ void aggregateGroupByPartitioned(
+ const std::vector<std::vector<std::unique_ptr<const Scalar>>> &arguments,
+ const std::vector<std::unique_ptr<const Scalar>> &group_by,
+ const TupleIdSequence *filter,
+ const std::size_t num_partitions,
+ ColumnVectorsValueAccessor *temp_result,
+ std::vector<attribute_id> *argument_ids,
+ std::vector<attribute_id> *key_ids,
+ std::vector<std::unique_ptr<ColumnVector>> *reuse_group_by_vectors) const;
+
/**
* @brief Inserts the GROUP BY expressions and aggregation arguments together
* as keys into the distinctify hash table.
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/1e1434ad/types/containers/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/types/containers/CMakeLists.txt b/types/containers/CMakeLists.txt
index aacb63a..c2a6623 100644
--- a/types/containers/CMakeLists.txt
+++ b/types/containers/CMakeLists.txt
@@ -49,6 +49,7 @@ target_link_libraries(quickstep_types_containers_Tuple
quickstep_catalog_CatalogTypedefs
quickstep_types_TypedValue
quickstep_types_containers_Tuple_proto
+ quickstep_utility_CompositeHash
quickstep_utility_Macros)
target_link_libraries(quickstep_types_containers_Tuple_proto
quickstep_types_TypedValue_proto
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/1e1434ad/types/containers/Tuple.hpp
----------------------------------------------------------------------
diff --git a/types/containers/Tuple.hpp b/types/containers/Tuple.hpp
index 60f832c..6237d54 100644
--- a/types/containers/Tuple.hpp
+++ b/types/containers/Tuple.hpp
@@ -28,6 +28,7 @@
#include "catalog/CatalogTypedefs.hpp"
#include "types/TypedValue.hpp"
#include "types/containers/Tuple.pb.h"
+#include "utility/CompositeHash.hpp"
#include "utility/Macros.hpp"
#include "glog/logging.h"
@@ -218,6 +219,13 @@ class Tuple {
return attribute_values_.size();
}
+ /**
+ * @brief Get the hash value of the tuple.
+ **/
+ std::size_t getTupleHash() const {
+ return HashCompositeKey(attribute_values_);
+ }
+
private:
/**
* @brief Constructor which does not create any attributes, nor pre-reserve
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/1e1434ad/utility/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/utility/CMakeLists.txt b/utility/CMakeLists.txt
index 395e264..e9be2ec 100644
--- a/utility/CMakeLists.txt
+++ b/utility/CMakeLists.txt
@@ -169,6 +169,7 @@ add_library(quickstep_utility_BloomFilter_proto
add_library(quickstep_utility_CalculateInstalledMemory CalculateInstalledMemory.cpp CalculateInstalledMemory.hpp)
add_library(quickstep_utility_Cast ../empty_src.cpp Cast.hpp)
add_library(quickstep_utility_CheckSnprintf ../empty_src.cpp CheckSnprintf.hpp)
+add_library(quickstep_utility_CompositeHash ../empty_src.cpp CompositeHash.hpp)
add_library(quickstep_utility_DAG ../empty_src.cpp DAG.hpp)
add_library(quickstep_utility_DisjointTreeForest ../empty_src.cpp DisjointTreeForest.hpp)
add_library(quickstep_utility_EqualsAnyConstant ../empty_src.cpp EqualsAnyConstant.hpp)
@@ -230,6 +231,10 @@ target_link_libraries(quickstep_utility_CalculateInstalledMemory
glog)
target_link_libraries(quickstep_utility_CheckSnprintf
glog)
+target_link_libraries(quickstep_utility_CompositeHash
+ quickstep_types_TypedValue
+ quickstep_utility_HashPair
+ glog)
target_link_libraries(quickstep_utility_DAG
glog
quickstep_utility_Macros)
@@ -325,6 +330,7 @@ target_link_libraries(quickstep_utility
quickstep_utility_CalculateInstalledMemory
quickstep_utility_Cast
quickstep_utility_CheckSnprintf
+ quickstep_utility_CompositeHash
quickstep_utility_DAG
quickstep_utility_DisjointTreeForest
quickstep_utility_EqualsAnyConstant
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/1e1434ad/utility/CompositeHash.hpp
----------------------------------------------------------------------
diff --git a/utility/CompositeHash.hpp b/utility/CompositeHash.hpp
new file mode 100644
index 0000000..517bc96
--- /dev/null
+++ b/utility/CompositeHash.hpp
@@ -0,0 +1,52 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ **/
+
+#ifndef QUICKSTEP_UTILITY_COMPOSITE_HASH_HPP_
+#define QUICKSTEP_UTILITY_COMPOSITE_HASH_HPP_
+
+#include <cstddef>
+#include <vector>
+
+#include "types/TypedValue.hpp"
+#include "utility/HashPair.hpp"
+
+#include "glog/logging.h"
+
+namespace quickstep {
+
+/**
+ * @brief Compute the hash value of a composite key.
+ *
+ * @param key A vector of TypedValues which together form the composite key.
+ * @return The hash value.
+ **/
+static std::size_t HashCompositeKey(const std::vector<TypedValue> &key) {
+ DCHECK(!key.empty());
+ std::size_t hash = key.front().getHash();
+ for (std::vector<TypedValue>::const_iterator key_it = key.begin() + 1;
+ key_it != key.end();
+ ++key_it) {
+ hash = CombineHashes(hash, key_it->getHash());
+ }
+ return hash;
+}
+
+} // namespace quickstep
+
+#endif // QUICKSTEP_UTILITY_COMPOSITE_HASH_HPP_
[02/11] incubator-quickstep git commit: Initiated query execution
data structure in all Shiftbosses.
Posted by ji...@apache.org.
Initiated query execution data structure in all Shiftbosses.
Project: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/commit/a58a7475
Tree: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/tree/a58a7475
Diff: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/diff/a58a7475
Branch: refs/heads/exact-filter
Commit: a58a74751ac3b0caf55aae09c13161dd375f4147
Parents: a11cdfd
Author: Zuyu Zhang <zu...@apache.org>
Authored: Sun Oct 9 18:44:00 2016 -0700
Committer: Zuyu Zhang <zu...@apache.org>
Committed: Fri Oct 21 08:06:59 2016 -0700
----------------------------------------------------------------------
query_execution/ForemanDistributed.cpp | 15 ++++-
query_execution/ForemanDistributed.hpp | 5 ++
query_execution/PolicyEnforcerBase.cpp | 2 +
query_execution/PolicyEnforcerBase.hpp | 15 +++++
query_execution/PolicyEnforcerDistributed.cpp | 67 +++++++++-----------
query_execution/QueryExecutionMessages.proto | 6 +-
query_execution/QueryExecutionUtil.hpp | 13 ++++
query_execution/Shiftboss.cpp | 2 +
.../DistributedExecutionGeneratorTestRunner.hpp | 2 +-
9 files changed, 86 insertions(+), 41 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/a58a7475/query_execution/ForemanDistributed.cpp
----------------------------------------------------------------------
diff --git a/query_execution/ForemanDistributed.cpp b/query_execution/ForemanDistributed.cpp
index 9c20465..56b319b 100644
--- a/query_execution/ForemanDistributed.cpp
+++ b/query_execution/ForemanDistributed.cpp
@@ -18,6 +18,8 @@
#include <cstdio>
#include <cstdlib>
#include <memory>
+#include <unordered_map>
+#include <unordered_set>
#include <utility>
#include <vector>
@@ -163,7 +165,9 @@ void ForemanDistributed::run() {
break;
}
case kQueryInitiateResponseMessage: {
- // TODO(zuyu): check the query id.
+ S::QueryInitiateResponseMessage proto;
+ CHECK(proto.ParseFromArray(tagged_message.message(), tagged_message.message_bytes()));
+ CHECK(policy_enforcer_->existQuery(proto.query_id()));
break;
}
case kCatalogRelationNewBlockMessage: // Fall through
@@ -183,7 +187,14 @@ void ForemanDistributed::run() {
S::SaveQueryResultResponseMessage proto;
CHECK(proto.ParseFromArray(tagged_message.message(), tagged_message.message_bytes()));
- processSaveQueryResultResponseMessage(proto.cli_id(), proto.relation_id());
+ const std::size_t query_id = proto.query_id();
+ query_result_saved_shiftbosses_[query_id].insert(proto.shiftboss_index());
+
+ // TODO(quickstep-team): Dynamically scale-up/down Shiftbosses.
+ if (query_result_saved_shiftbosses_[query_id].size() == shiftboss_directory_.size()) {
+ processSaveQueryResultResponseMessage(proto.cli_id(), proto.relation_id());
+ query_result_saved_shiftbosses_.erase(query_id);
+ }
break;
}
case kPoisonMessage: {
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/a58a7475/query_execution/ForemanDistributed.hpp
----------------------------------------------------------------------
diff --git a/query_execution/ForemanDistributed.hpp b/query_execution/ForemanDistributed.hpp
index fc1ede5..b42795c 100644
--- a/query_execution/ForemanDistributed.hpp
+++ b/query_execution/ForemanDistributed.hpp
@@ -18,6 +18,8 @@
#include <cstddef>
#include <cstdio>
#include <memory>
+#include <unordered_map>
+#include <unordered_set>
#include <vector>
#include "catalog/CatalogTypedefs.hpp"
@@ -120,6 +122,9 @@ class ForemanDistributed final : public ForemanBase {
std::unique_ptr<PolicyEnforcerDistributed> policy_enforcer_;
+ // From a query id to a set of Shiftbosses that save query result.
+ std::unordered_map<std::size_t, std::unordered_set<std::size_t>> query_result_saved_shiftbosses_;
+
DISALLOW_COPY_AND_ASSIGN(ForemanDistributed);
};
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/a58a7475/query_execution/PolicyEnforcerBase.cpp
----------------------------------------------------------------------
diff --git a/query_execution/PolicyEnforcerBase.cpp b/query_execution/PolicyEnforcerBase.cpp
index 4174bd6..745ded6 100644
--- a/query_execution/PolicyEnforcerBase.cpp
+++ b/query_execution/PolicyEnforcerBase.cpp
@@ -142,6 +142,8 @@ void PolicyEnforcerBase::removeQuery(const std::size_t query_id) {
<< " that hasn't finished its execution";
}
admitted_queries_.erase(query_id);
+
+ removed_query_ids_.insert(query_id);
}
bool PolicyEnforcerBase::admitQueries(
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/a58a7475/query_execution/PolicyEnforcerBase.hpp
----------------------------------------------------------------------
diff --git a/query_execution/PolicyEnforcerBase.hpp b/query_execution/PolicyEnforcerBase.hpp
index 62906e9..ea2c06f 100644
--- a/query_execution/PolicyEnforcerBase.hpp
+++ b/query_execution/PolicyEnforcerBase.hpp
@@ -24,6 +24,7 @@
#include <memory>
#include <queue>
#include <unordered_map>
+#include <unordered_set>
#include <vector>
#include "query_execution/QueryExecutionTypedefs.hpp"
@@ -106,6 +107,16 @@ class PolicyEnforcerBase {
void processMessage(const TaggedMessage &tagged_message);
/**
+ * @brief Check if the given query id ever exists.
+ *
+ * @return True if the query ever exists, otherwise false.
+ **/
+ inline bool existQuery(const std::size_t query_id) const {
+ return admitted_queries_.find(query_id) != admitted_queries_.end() ||
+ removed_query_ids_.find(query_id) != removed_query_ids_.end();
+ }
+
+ /**
* @brief Check if there are any queries to be executed.
*
* @return True if there is at least one active or waiting query, false if
@@ -163,6 +174,10 @@ class PolicyEnforcerBase {
// Key = query ID, value = QueryManagerBase* for the key query.
std::unordered_map<std::size_t, std::unique_ptr<QueryManagerBase>> admitted_queries_;
+ // TODO(quickstep-team): Delete a 'query_id' after receiving all
+ // 'QueryInitiateResponseMessage's for the 'query_id'.
+ std::unordered_set<std::size_t> removed_query_ids_;
+
// The queries which haven't been admitted yet.
std::queue<QueryHandle*> waiting_queries_;
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/a58a7475/query_execution/PolicyEnforcerDistributed.cpp
----------------------------------------------------------------------
diff --git a/query_execution/PolicyEnforcerDistributed.cpp b/query_execution/PolicyEnforcerDistributed.cpp
index 47491ed..c06fd86 100644
--- a/query_execution/PolicyEnforcerDistributed.cpp
+++ b/query_execution/PolicyEnforcerDistributed.cpp
@@ -37,6 +37,7 @@
#include "gflags/gflags.h"
#include "glog/logging.h"
+#include "tmb/address.h"
#include "tmb/id_typedefs.h"
#include "tmb/message_bus.h"
#include "tmb/tagged_message.h"
@@ -170,25 +171,18 @@ void PolicyEnforcerDistributed::initiateQueryInShiftboss(QueryHandle *query_hand
kQueryInitiateMessage);
free(proto_bytes);
- // TODO(zuyu): Multiple Shiftbosses support.
+ // TODO(quickstep-team): Dynamically scale-up/down Shiftbosses.
+ tmb::Address shiftboss_addresses;
+ for (std::size_t i = 0; i < shiftboss_directory_->size(); ++i) {
+ shiftboss_addresses.AddRecipient(shiftboss_directory_->getClientId(i));
+ }
+
DLOG(INFO) << "PolicyEnforcerDistributed sent QueryInitiateMessage (typed '" << kQueryInitiateMessage
- << "') to Shiftboss with TMB client ID " << shiftboss_directory_->getClientId(0);
- const tmb::MessageBus::SendStatus send_status =
- QueryExecutionUtil::SendTMBMessage(bus_,
- foreman_client_id_,
- shiftboss_directory_->getClientId(0),
- move(message));
- CHECK(send_status == tmb::MessageBus::SendStatus::kOK);
-
- // Wait Shiftboss for QueryInitiateResponseMessage.
- const tmb::AnnotatedMessage annotated_message = bus_->Receive(foreman_client_id_, 0, true);
- const TaggedMessage &tagged_message = annotated_message.tagged_message;
- DCHECK_EQ(kQueryInitiateResponseMessage, tagged_message.message_type());
- DLOG(INFO) << "PolicyEnforcerDistributed received typed '" << tagged_message.message_type()
- << "' message from client " << annotated_message.sender;
-
- S::QueryInitiateResponseMessage proto_response;
- CHECK(proto_response.ParseFromArray(tagged_message.message(), tagged_message.message_bytes()));
+ << "') to all Shiftbosses";
+ QueryExecutionUtil::BroadcastMessage(foreman_client_id_,
+ shiftboss_addresses,
+ move(message),
+ bus_);
}
void PolicyEnforcerDistributed::onQueryCompletion(QueryManagerBase *query_manager) {
@@ -198,8 +192,14 @@ void PolicyEnforcerDistributed::onQueryCompletion(QueryManagerBase *query_manage
const tmb::client_id cli_id = query_handle->getClientId();
const std::size_t query_id = query_handle->query_id();
+ // TODO(quickstep-team): Dynamically scale-up/down Shiftbosses.
+ tmb::Address shiftboss_addresses;
+ for (std::size_t i = 0; i < shiftboss_directory_->size(); ++i) {
+ shiftboss_addresses.AddRecipient(shiftboss_directory_->getClientId(i));
+ }
+
if (query_result == nullptr) {
- // Clean up query execution states, i.e., QueryContext, in Shiftboss.
+ // Clean up query execution states, i.e., QueryContext, in Shiftbosses.
serialization::QueryTeardownMessage proto;
proto.set_query_id(query_id);
@@ -211,15 +211,12 @@ void PolicyEnforcerDistributed::onQueryCompletion(QueryManagerBase *query_manage
proto_length,
kQueryTeardownMessage);
- // TODO(zuyu): Support multiple shiftbosses.
DLOG(INFO) << "PolicyEnforcerDistributed sent QueryTeardownMessage (typed '" << kQueryTeardownMessage
- << "') to Shiftboss with TMB client ID " << shiftboss_directory_->getClientId(0);
- tmb::MessageBus::SendStatus send_status =
- QueryExecutionUtil::SendTMBMessage(bus_,
- foreman_client_id_,
- shiftboss_directory_->getClientId(0),
- move(message));
- CHECK(send_status == tmb::MessageBus::SendStatus::kOK);
+ << "') to all Shiftbosses";
+ QueryExecutionUtil::BroadcastMessage(foreman_client_id_,
+ shiftboss_addresses,
+ move(message),
+ bus_);
TaggedMessage cli_message(kQueryExecutionSuccessMessage);
@@ -227,7 +224,7 @@ void PolicyEnforcerDistributed::onQueryCompletion(QueryManagerBase *query_manage
DLOG(INFO) << "PolicyEnforcerDistributed sent QueryExecutionSuccessMessage (typed '"
<< kQueryExecutionSuccessMessage
<< "') to CLI with TMB client id " << cli_id;
- send_status =
+ const tmb::MessageBus::SendStatus send_status =
QueryExecutionUtil::SendTMBMessage(bus_,
foreman_client_id_,
cli_id,
@@ -257,15 +254,13 @@ void PolicyEnforcerDistributed::onQueryCompletion(QueryManagerBase *query_manage
kSaveQueryResultMessage);
free(proto_bytes);
- // TODO(zuyu): Support multiple shiftbosses.
+ // TODO(quickstep-team): Dynamically scale-up/down Shiftbosses.
DLOG(INFO) << "PolicyEnforcerDistributed sent SaveQueryResultMessage (typed '" << kSaveQueryResultMessage
- << "') to Shiftboss with TMB client ID " << shiftboss_directory_->getClientId(0);
- const tmb::MessageBus::SendStatus send_status =
- QueryExecutionUtil::SendTMBMessage(bus_,
- foreman_client_id_,
- shiftboss_directory_->getClientId(0),
- move(message));
- CHECK(send_status == tmb::MessageBus::SendStatus::kOK);
+ << "') to all Shiftbosses";
+ QueryExecutionUtil::BroadcastMessage(foreman_client_id_,
+ shiftboss_addresses,
+ move(message),
+ bus_);
}
} // namespace quickstep
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/a58a7475/query_execution/QueryExecutionMessages.proto
----------------------------------------------------------------------
diff --git a/query_execution/QueryExecutionMessages.proto b/query_execution/QueryExecutionMessages.proto
index 060efa1..1a2cb78 100644
--- a/query_execution/QueryExecutionMessages.proto
+++ b/query_execution/QueryExecutionMessages.proto
@@ -128,8 +128,10 @@ message SaveQueryResultMessage {
}
message SaveQueryResultResponseMessage {
- required int32 relation_id = 1;
- required uint32 cli_id = 2; // tmb::client_id.
+ required uint64 query_id = 1;
+ required int32 relation_id = 2;
+ required uint32 cli_id = 3; // tmb::client_id.
+ required uint64 shiftboss_index = 4;
}
message QueryExecutionSuccessMessage {
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/a58a7475/query_execution/QueryExecutionUtil.hpp
----------------------------------------------------------------------
diff --git a/query_execution/QueryExecutionUtil.hpp b/query_execution/QueryExecutionUtil.hpp
index 7a3a3b3..b41965c 100644
--- a/query_execution/QueryExecutionUtil.hpp
+++ b/query_execution/QueryExecutionUtil.hpp
@@ -121,6 +121,19 @@ class QueryExecutionUtil {
DCHECK_EQ(kWorkloadCompletionMessage, tagged_message.message_type());
}
+ static void BroadcastMessage(const tmb::client_id sender_id,
+ const tmb::Address &addresses,
+ tmb::TaggedMessage &&tagged_message, // NOLINT(whitespace/operators)
+ tmb::MessageBus *bus) {
+ // The sender broadcasts the given message to all 'addresses'.
+ tmb::MessageStyle style;
+ style.Broadcast(true);
+
+ const tmb::MessageBus::SendStatus send_status =
+ bus->Send(sender_id, addresses, style, std::move(tagged_message));
+ CHECK(send_status == tmb::MessageBus::SendStatus::kOK);
+ }
+
static void BroadcastPoisonMessage(const tmb::client_id sender_id, tmb::MessageBus *bus) {
// Terminate all threads.
// The sender thread broadcasts poison message to the workers and foreman.
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/a58a7475/query_execution/Shiftboss.cpp
----------------------------------------------------------------------
diff --git a/query_execution/Shiftboss.cpp b/query_execution/Shiftboss.cpp
index 5c2c5e0..a434527 100644
--- a/query_execution/Shiftboss.cpp
+++ b/query_execution/Shiftboss.cpp
@@ -189,8 +189,10 @@ void Shiftboss::run() {
query_contexts_.erase(proto.query_id());
serialization::SaveQueryResultResponseMessage proto_response;
+ proto_response.set_query_id(proto.query_id());
proto_response.set_relation_id(proto.relation_id());
proto_response.set_cli_id(proto.cli_id());
+ proto_response.set_shiftboss_index(shiftboss_index_);
const size_t proto_response_length = proto_response.ByteSize();
char *proto_response_bytes = static_cast<char*>(malloc(proto_response_length));
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/a58a7475/query_optimizer/tests/DistributedExecutionGeneratorTestRunner.hpp
----------------------------------------------------------------------
diff --git a/query_optimizer/tests/DistributedExecutionGeneratorTestRunner.hpp b/query_optimizer/tests/DistributedExecutionGeneratorTestRunner.hpp
index e4d0765..ab10841 100644
--- a/query_optimizer/tests/DistributedExecutionGeneratorTestRunner.hpp
+++ b/query_optimizer/tests/DistributedExecutionGeneratorTestRunner.hpp
@@ -49,7 +49,7 @@ namespace quickstep {
namespace optimizer {
namespace {
-constexpr int kNumInstances = 1;
+constexpr int kNumInstances = 3;
} // namespace
/**
[06/11] incubator-quickstep git commit: Add backend support for
LIPFilters.
Posted by ji...@apache.org.
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/393eba55/storage/StorageBlock.cpp
----------------------------------------------------------------------
diff --git a/storage/StorageBlock.cpp b/storage/StorageBlock.cpp
index ec5990f..dd3e19d 100644
--- a/storage/StorageBlock.cpp
+++ b/storage/StorageBlock.cpp
@@ -340,7 +340,7 @@ void StorageBlock::sample(const bool is_block_sample,
}
void StorageBlock::select(const vector<unique_ptr<const Scalar>> &selection,
- const Predicate *predicate,
+ const TupleIdSequence *filter,
InsertDestinationInterface *destination) const {
ColumnVectorsValueAccessor temp_result;
{
@@ -348,14 +348,8 @@ void StorageBlock::select(const vector<unique_ptr<const Scalar>> &selection,
indices_,
indices_consistent_);
- std::unique_ptr<TupleIdSequence> matches;
- std::unique_ptr<ValueAccessor> accessor;
- if (predicate == nullptr) {
- accessor.reset(tuple_store_->createValueAccessor());
- } else {
- matches.reset(getMatchesForPredicate(predicate));
- accessor.reset(tuple_store_->createValueAccessor(matches.get()));
- }
+ std::unique_ptr<ValueAccessor> accessor(
+ tuple_store_->createValueAccessor(filter));
for (vector<unique_ptr<const Scalar>>::const_iterator selection_cit = selection.begin();
selection_cit != selection.end();
@@ -370,16 +364,10 @@ void StorageBlock::select(const vector<unique_ptr<const Scalar>> &selection,
}
void StorageBlock::selectSimple(const std::vector<attribute_id> &selection,
- const Predicate *predicate,
+ const TupleIdSequence *filter,
InsertDestinationInterface *destination) const {
- std::unique_ptr<ValueAccessor> accessor;
- std::unique_ptr<TupleIdSequence> matches;
- if (predicate == nullptr) {
- accessor.reset(tuple_store_->createValueAccessor());
- } else {
- matches.reset(getMatchesForPredicate(predicate));
- accessor.reset(tuple_store_->createValueAccessor(matches.get()));
- }
+ std::unique_ptr<ValueAccessor> accessor(
+ tuple_store_->createValueAccessor(filter));
destination->bulkInsertTuplesWithRemappedAttributes(selection,
accessor.get());
@@ -389,37 +377,28 @@ AggregationState* StorageBlock::aggregate(
const AggregationHandle &handle,
const std::vector<std::unique_ptr<const Scalar>> &arguments,
const std::vector<attribute_id> *arguments_as_attributes,
- const Predicate *predicate,
- std::unique_ptr<TupleIdSequence> *reuse_matches) const {
- // If there is a filter predicate that hasn't already been evaluated,
- // evaluate it now and save the results for other aggregates on this same
- // block.
- if (predicate && !*reuse_matches) {
- reuse_matches->reset(getMatchesForPredicate(predicate));
- }
-
+ const TupleIdSequence *filter) const {
#ifdef QUICKSTEP_ENABLE_VECTOR_COPY_ELISION_SELECTION
// If all the arguments to this aggregate are plain relation attributes,
// aggregate directly on a ValueAccessor from this block to avoid a copy.
if ((arguments_as_attributes != nullptr) && (!arguments_as_attributes->empty())) {
DCHECK_EQ(arguments.size(), arguments_as_attributes->size())
<< "Mismatch between number of arguments and number of attribute_ids";
- return aggregateHelperValueAccessor(handle, *arguments_as_attributes, reuse_matches->get());
+ return aggregateHelperValueAccessor(handle, *arguments_as_attributes, filter);
}
// TODO(shoban): We may want to optimize for ScalarLiteral here.
#endif
// Call aggregateHelperColumnVector() to materialize each argument as a
// ColumnVector, then aggregate over those.
- return aggregateHelperColumnVector(handle, arguments, reuse_matches->get());
+ return aggregateHelperColumnVector(handle, arguments, filter);
}
void StorageBlock::aggregateGroupBy(
const std::vector<std::vector<std::unique_ptr<const Scalar>>> &arguments,
const std::vector<std::unique_ptr<const Scalar>> &group_by,
- const Predicate *predicate,
+ const TupleIdSequence *filter,
AggregationStateHashTableBase *hash_table,
- std::unique_ptr<TupleIdSequence> *reuse_matches,
std::vector<std::unique_ptr<ColumnVector>> *reuse_group_by_vectors) const {
DCHECK_GT(group_by.size(), 0u)
<< "Called aggregateGroupBy() with zero GROUP BY expressions";
@@ -438,23 +417,7 @@ void StorageBlock::aggregateGroupBy(
// this aggregate, as well as the GROUP BY expression values.
ColumnVectorsValueAccessor temp_result;
{
- std::unique_ptr<ValueAccessor> accessor;
- if (predicate) {
- if (!*reuse_matches) {
- // If there is a filter predicate that hasn't already been evaluated,
- // evaluate it now and save the results for other aggregates on this
- // same block.
- reuse_matches->reset(getMatchesForPredicate(predicate));
- }
-
- // Create a filtered ValueAccessor that only iterates over predicate
- // matches.
- accessor.reset(tuple_store_->createValueAccessor(reuse_matches->get()));
- } else {
- // Create a ValueAccessor that iterates over all tuples in this block
- accessor.reset(tuple_store_->createValueAccessor());
- }
-
+ std::unique_ptr<ValueAccessor> accessor(tuple_store_->createValueAccessor(filter));
attribute_id attr_id = 0;
// First, put GROUP BY keys into 'temp_result'.
@@ -503,9 +466,8 @@ void StorageBlock::aggregateDistinct(
const std::vector<std::unique_ptr<const Scalar>> &arguments,
const std::vector<attribute_id> *arguments_as_attributes,
const std::vector<std::unique_ptr<const Scalar>> &group_by,
- const Predicate *predicate,
+ const TupleIdSequence *filter,
AggregationStateHashTableBase *distinctify_hash_table,
- std::unique_ptr<TupleIdSequence> *reuse_matches,
std::vector<std::unique_ptr<ColumnVector>> *reuse_group_by_vectors) const {
DCHECK_GT(arguments.size(), 0u)
<< "Called aggregateDistinct() with zero argument expressions";
@@ -517,22 +479,7 @@ void StorageBlock::aggregateDistinct(
// this aggregate, as well as the GROUP BY expression values.
ColumnVectorsValueAccessor temp_result;
{
- std::unique_ptr<ValueAccessor> accessor;
- if (predicate) {
- if (!*reuse_matches) {
- // If there is a filter predicate that hasn't already been evaluated,
- // evaluate it now and save the results for other aggregates on this
- // same block.
- reuse_matches->reset(getMatchesForPredicate(predicate));
- }
-
- // Create a filtered ValueAccessor that only iterates over predicate
- // matches.
- accessor.reset(tuple_store_->createValueAccessor(reuse_matches->get()));
- } else {
- // Create a ValueAccessor that iterates over all tuples in this block
- accessor.reset(tuple_store_->createValueAccessor());
- }
+ std::unique_ptr<ValueAccessor> accessor(tuple_store_->createValueAccessor(filter));
#ifdef QUICKSTEP_ENABLE_VECTOR_COPY_ELISION_SELECTION
// If all the arguments to this aggregate are plain relation attributes,
@@ -1246,23 +1193,36 @@ bool StorageBlock::rebuildIndexes(bool short_circuit) {
return all_indices_consistent_;
}
-TupleIdSequence* StorageBlock::getMatchesForPredicate(const Predicate *predicate) const {
+TupleIdSequence* StorageBlock::getMatchesForPredicate(const Predicate *predicate,
+ const TupleIdSequence *filter) const {
if (predicate == nullptr) {
- return tuple_store_->getExistenceMap();
+ TupleIdSequence *matches = tuple_store_->getExistenceMap();
+ if (filter != nullptr) {
+ matches->intersectWith(*filter);
+ }
+ return matches;
}
std::unique_ptr<ValueAccessor> value_accessor(tuple_store_->createValueAccessor());
- std::unique_ptr<TupleIdSequence> existence_map;
- if (!tuple_store_->isPacked()) {
- existence_map.reset(tuple_store_->getExistenceMap());
- }
SubBlocksReference sub_blocks_ref(*tuple_store_,
indices_,
indices_consistent_);
- return predicate->getAllMatches(value_accessor.get(),
- &sub_blocks_ref,
- nullptr,
- existence_map.get());
+
+ if (!tuple_store_->isPacked()) {
+ std::unique_ptr<TupleIdSequence> existence_map(tuple_store_->getExistenceMap());
+ if (filter != nullptr) {
+ existence_map->intersectWith(*filter);
+ }
+ return predicate->getAllMatches(value_accessor.get(),
+ &sub_blocks_ref,
+ nullptr,
+ existence_map.get());
+ } else {
+ return predicate->getAllMatches(value_accessor.get(),
+ &sub_blocks_ref,
+ nullptr,
+ filter);
+ }
}
std::unordered_map<attribute_id, TypedValue>* StorageBlock::generateUpdatedValues(
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/393eba55/storage/StorageBlock.hpp
----------------------------------------------------------------------
diff --git a/storage/StorageBlock.hpp b/storage/StorageBlock.hpp
index bab5bab..488efcc 100644
--- a/storage/StorageBlock.hpp
+++ b/storage/StorageBlock.hpp
@@ -313,6 +313,17 @@ class StorageBlock : public StorageBlockBase {
ValueAccessor *accessor);
/**
+ * @brief Get the IDs of tuples in this StorageBlock which match a given Predicate.
+ *
+ * @param predicate The predicate to match.
+ * @param filter If non-NULL, then only tuple IDs which are set in the
+ * filter will be checked (all others will be assumed to be false).
+ * @return A TupleIdSequence which contains matching tuple IDs for predicate.
+ **/
+ TupleIdSequence* getMatchesForPredicate(const Predicate *predicate,
+ const TupleIdSequence *filter = nullptr) const;
+
+ /**
* @brief Perform a random sampling of data on the StorageBlock. The number
* of records sampled is determined by the sample percentage in case of
* tuple sample. For block sample all the records in a block are taken.
@@ -336,8 +347,8 @@ class StorageBlock : public StorageBlockBase {
*
* @param selection A list of scalars, which will be evaluated to obtain
* attribute values for each result tuple.
- * @param predicate A predicate for selection. NULL indicates that all tuples
- * should be matched.
+ * @param filter If non-NULL, then only tuple IDs which are set in the
+ * filter will be checked (all others will be assumed to be false).
* @param destination Where to insert the tuples resulting from the SELECT
* query.
* @exception TupleTooLargeForBlock A tuple produced by this selection was
@@ -348,18 +359,18 @@ class StorageBlock : public StorageBlockBase {
*
**/
void select(const std::vector<std::unique_ptr<const Scalar>> &selection,
- const Predicate *predicate,
+ const TupleIdSequence *filter,
InsertDestinationInterface *destination) const;
/**
* @brief Perform a simple SELECT query on this StorageBlock which only
* projects attributes and does not evaluate expressions.
*
+ * @param selection The attributes to project.
+ * @param filter If non-NULL, then only tuple IDs which are set in the
+ * filter will be checked (all others will be assumed to be false).
* @param destination Where to insert the tuples resulting from the SELECT
* query.
- * @param selection The attributes to project.
- * @param predicate A predicate for selection. NULL indicates that all tuples
- * should be matched.
* @exception TupleTooLargeForBlock A tuple produced by this selection was
* too large to insert into an empty block provided by
* destination. Selection may be partially complete (with some
@@ -371,7 +382,7 @@ class StorageBlock : public StorageBlockBase {
* an inconsistent IndexSubBlock (see indicesAreConsistent()).
**/
void selectSimple(const std::vector<attribute_id> &selection,
- const Predicate *predicate,
+ const TupleIdSequence *filter,
InsertDestinationInterface *destination) const;
/**
@@ -384,23 +395,8 @@ class StorageBlock : public StorageBlockBase {
* @param arguments_as_attributes If non-NULL, indicates a valid attribute_id
* for each of the elements in arguments, and is used to elide a copy.
* Has no effect if NULL, or if VECTOR_COPY_ELISION_LEVEL is NONE.
- * @param predicate A predicate for selection. nullptr indicates that all
- * tuples should be aggregated on.
- * @param reuse_matches This parameter is used to store and reuse tuple-id
- * sequence of matches pre-computed in an earlier invocations to
- * aggregate(). \c reuse_matches is never \c nullptr for ease of use.
- * Current invocation of aggregate() will reuse TupleIdSequence if
- * passed, otherwise compute a TupleIdSequence based on \c predicate
- * and store in \c reuse_matches. We use std::unique_ptr for each of
- * use, since the caller will not have to selective free.
- *
- * For example, see this relevant pseudo-C++ code:
- * \code
- * std::unique_ptr<TupleIdSequence> matches;
- * for each aggregate {
- * block.aggregate(..., &matches);
- * }
- * \endcode
+ * @param filter If non-NULL, then only tuple IDs which are set in the
+ * filter will be checked (all others will be assumed to be false).
*
* @return Aggregated state for this block in the form of an
* AggregationState. AggregationHandle::mergeStates() can be called
@@ -412,8 +408,7 @@ class StorageBlock : public StorageBlockBase {
const AggregationHandle &handle,
const std::vector<std::unique_ptr<const Scalar>> &arguments,
const std::vector<attribute_id> *arguments_as_attributes,
- const Predicate *predicate,
- std::unique_ptr<TupleIdSequence> *reuse_matches) const;
+ const TupleIdSequence *filter) const;
/**
* @brief Perform GROUP BY aggregation on the tuples in the this storage
@@ -423,18 +418,10 @@ class StorageBlock : public StorageBlockBase {
* @param group_by The list of GROUP BY attributes/expressions. The tuples in
* this storage block are grouped by these attributes before
* aggregation.
- * @param predicate A predicate for selection. nullptr indicates that all
- * tuples should be aggregated on.
+ * @param filter If non-NULL, then only tuple IDs which are set in the
+ * filter will be checked (all others will be assumed to be false).
* @param hash_table Hash table to store aggregation state mapped based on
* GROUP BY value list (defined by \c group_by).
- * @param reuse_matches This parameter is used to store and reuse tuple-id
- * sequence of matches pre-computed in an earlier invocations of
- * aggregateGroupBy(). \c reuse_matches is never \c nullptr for ease of
- * use. Current invocation of aggregateGroupBy() will reuse
- * TupleIdSequence if passed, otherwise computes a TupleIdSequence based
- * on \c predicate and stores in \c reuse_matches. We use
- * std::unique_ptr for each of use, since the caller will not have to
- * selective free.
* @param reuse_group_by_vectors This parameter is used to store and reuse
* GROUP BY attribute vectors pre-computed in an earlier invocation of
* aggregateGroupBy(). \c reuse_group_by_vectors is never \c nullptr
@@ -444,10 +431,9 @@ class StorageBlock : public StorageBlockBase {
*
* For sample usage of aggregateGroupBy, see this relevant pseudo-C++ code:
* \code
- * std::unique_ptr<TupleIdSequence> matches;
* std::vector<std::unique_ptr<ColumnVector>> group_by_vectors;
* for each aggregate {
- * block.aggregateGroupBy(..., &matches, &group_by_vectors);
+ * block.aggregateGroupBy(..., &group_by_vectors);
* }
* \endcode
**/
@@ -461,9 +447,8 @@ class StorageBlock : public StorageBlockBase {
void aggregateGroupBy(
const std::vector<std::vector<std::unique_ptr<const Scalar>>> &arguments,
const std::vector<std::unique_ptr<const Scalar>> &group_by,
- const Predicate *predicate,
+ const TupleIdSequence *filter,
AggregationStateHashTableBase *hash_table,
- std::unique_ptr<TupleIdSequence> *reuse_matches,
std::vector<std::unique_ptr<ColumnVector>> *reuse_group_by_vectors) const;
/**
@@ -481,19 +466,11 @@ class StorageBlock : public StorageBlockBase {
* for each of the elements in arguments, and is used to elide a copy.
* Has no effect if NULL, or if VECTOR_COPY_ELISION_LEVEL is NONE.
* @param group_by The list of GROUP BY attributes/expressions.
- * @param predicate A predicate for selection. \c nullptr indicates that all
- * tuples should be aggregated on.
+ * @param filter If non-NULL, then only tuple IDs which are set in the
+ * filter will be checked (all others will be assumed to be false).
* @param distinctify_hash_table Hash table to store the arguments and GROUP
* BY expressions together as hash table key and a bool constant \c true
* as hash table value. (So the hash table actually serves as a hash set.)
- * @param reuse_matches This parameter is used to store and reuse tuple-id
- * sequence of matches pre-computed in an earlier invocations of
- * aggregateGroupBy(). \c reuse_matches is never \c nullptr for ease of
- * use. Current invocation of aggregateGroupBy() will reuse
- * TupleIdSequence if passed, otherwise computes a TupleIdSequence based
- * on \c predicate and stores in \c reuse_matches. We use
- * std::unique_ptr for each of use, since the caller will not have to
- * selective free.
* @param reuse_group_by_vectors This parameter is used to store and reuse
* GROUP BY attribute vectors pre-computed in an earlier invocation of
* aggregateGroupBy(). \c reuse_group_by_vectors is never \c nullptr
@@ -505,9 +482,8 @@ class StorageBlock : public StorageBlockBase {
const std::vector<std::unique_ptr<const Scalar>> &arguments,
const std::vector<attribute_id> *arguments_as_attributes,
const std::vector<std::unique_ptr<const Scalar>> &group_by,
- const Predicate *predicate,
+ const TupleIdSequence *filter,
AggregationStateHashTableBase *distinctify_hash_table,
- std::unique_ptr<TupleIdSequence> *reuse_matches,
std::vector<std::unique_ptr<ColumnVector>> *reuse_group_by_vectors) const;
/**
@@ -627,8 +603,6 @@ class StorageBlock : public StorageBlockBase {
// StorageBlock's header.
bool rebuildIndexes(bool short_circuit);
- TupleIdSequence* getMatchesForPredicate(const Predicate *predicate) const;
-
std::unordered_map<attribute_id, TypedValue>* generateUpdatedValues(
const ValueAccessor &accessor,
const tuple_id tuple,
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/393eba55/utility/PlanVisualizer.cpp
----------------------------------------------------------------------
diff --git a/utility/PlanVisualizer.cpp b/utility/PlanVisualizer.cpp
index 2adf674..5d70c86 100644
--- a/utility/PlanVisualizer.cpp
+++ b/utility/PlanVisualizer.cpp
@@ -132,9 +132,7 @@ void PlanVisualizer::visit(const P::PhysicalPtr &input) {
for (const auto &attr : child->getOutputAttributes()) {
if (referenced_ids.find(attr->id()) != referenced_ids.end()) {
- edge_info.labels.emplace_back(
- attr->attribute_alias() + ", est # distinct = " +
- std::to_string(cost_model_->estimateNumDistinctValues(attr->id(), child)));
+ edge_info.labels.emplace_back(attr->attribute_alias());
}
}
}
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/393eba55/utility/lip_filter/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/utility/lip_filter/CMakeLists.txt b/utility/lip_filter/CMakeLists.txt
index b7224d2..23b3763 100644
--- a/utility/lip_filter/CMakeLists.txt
+++ b/utility/lip_filter/CMakeLists.txt
@@ -25,6 +25,7 @@ add_library(quickstep_utility_lipfilter_LIPFilterAdaptiveProber ../../empty_src.
add_library(quickstep_utility_lipfilter_LIPFilterBuilder ../../empty_src.cpp LIPFilterBuilder.hpp)
add_library(quickstep_utility_lipfilter_LIPFilterDeployment LIPFilterDeployment.cpp LIPFilterDeployment.hpp)
add_library(quickstep_utility_lipfilter_LIPFilterFactory LIPFilterFactory.cpp LIPFilterFactory.hpp)
+add_library(quickstep_utility_lipfilter_LIPFilterUtil ../../empty_src.cpp LIPFilterUtil.hpp)
add_library(quickstep_utility_lipfilter_LIPFilter_proto
${utility_lipfilter_LIPFilter_proto_srcs})
add_library(quickstep_utility_lipfilter_SingleIdentityHashFilter ../../empty_src.cpp SingleIdentityHashFilter.hpp)
@@ -58,6 +59,9 @@ target_link_libraries(quickstep_utility_lipfilter_LIPFilterFactory
quickstep_utility_lipfilter_LIPFilter_proto
quickstep_utility_lipfilter_SingleIdentityHashFilter
quickstep_utility_Macros)
+target_link_libraries(quickstep_utility_lipfilter_LIPFilterUtil
+ quickstep_queryexecution_QueryContext
+ quickstep_utility_lipfilter_LIPFilterDeployment)
target_link_libraries(quickstep_utility_lipfilter_LIPFilter_proto
${PROTOBUF_LIBRARY}
quickstep_types_Type_proto)
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/393eba55/utility/lip_filter/LIPFilterBuilder.hpp
----------------------------------------------------------------------
diff --git a/utility/lip_filter/LIPFilterBuilder.hpp b/utility/lip_filter/LIPFilterBuilder.hpp
index deb8f66..aa84a06 100644
--- a/utility/lip_filter/LIPFilterBuilder.hpp
+++ b/utility/lip_filter/LIPFilterBuilder.hpp
@@ -39,9 +39,6 @@ class ValueAccessor;
* @{
*/
-class LIPFilterBuilder;
-typedef std::shared_ptr<LIPFilterBuilder> LIPFilterBuilderPtr;
-
/**
* @brief Helper class for building LIPFilters from a relation (i.e. ValueAccessor).
*/
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/393eba55/utility/lip_filter/LIPFilterUtil.hpp
----------------------------------------------------------------------
diff --git a/utility/lip_filter/LIPFilterUtil.hpp b/utility/lip_filter/LIPFilterUtil.hpp
new file mode 100644
index 0000000..5d43c49
--- /dev/null
+++ b/utility/lip_filter/LIPFilterUtil.hpp
@@ -0,0 +1,79 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ **/
+
+#ifndef QUICKSTEP_UTILITY_LIP_FILTER_LIP_FILTER_UTIL_HPP_
+#define QUICKSTEP_UTILITY_LIP_FILTER_LIP_FILTER_UTIL_HPP_
+
+#include "query_execution/QueryContext.hpp"
+#include "utility/lip_filter/LIPFilterDeployment.hpp"
+
+namespace quickstep {
+
+class LIPFilterBuilder;
+class LIPFilterAdaptiveProber;
+
+/** \addtogroup Utility
+ * @{
+ */
+
+/**
+ * @brief Create a LIPFilterBuilder for the given LIPFilterDeployment in QueryContext.
+ *
+ * @param lip_deployment_index The id of the LIPFilterDeployment in QueryContext.
+ * @param query_context The QueryContext.
+ * @return A LIPFilterBuilder object, or nullptr if \p lip_deployment_index is invalid.
+ */
+inline LIPFilterBuilder* CreateLIPFilterBuilderHelper(
+ const QueryContext::lip_deployment_id lip_deployment_index,
+ const QueryContext *query_context) {
+ if (lip_deployment_index == QueryContext::kInvalidLIPDeploymentId) {
+ return nullptr;
+ } else {
+ const LIPFilterDeployment *lip_filter_deployment =
+ query_context->getLIPDeployment(lip_deployment_index);
+ return lip_filter_deployment->createLIPFilterBuilder();
+ }
+}
+
+/**
+ * @brief Create a LIPFilterAdaptiveProber for the given LIPFilterDeployment
+ * in QueryContext.
+ *
+ * @param lip_deployment_index The id of the LIPFilterDeployment in QueryContext.
+ * @param query_context The QueryContext.
+ * @return A LIPFilterAdaptiveProber object, or nullptr if \p lip_deployment_index
+ * is invalid.
+ */
+inline LIPFilterAdaptiveProber* CreateLIPFilterAdaptiveProberHelper(
+ const QueryContext::lip_deployment_id lip_deployment_index,
+ const QueryContext *query_context) {
+ if (lip_deployment_index == QueryContext::kInvalidLIPDeploymentId) {
+ return nullptr;
+ } else {
+ const LIPFilterDeployment *lip_filter_deployment =
+ query_context->getLIPDeployment(lip_deployment_index);
+ return lip_filter_deployment->createLIPFilterAdaptiveProber();
+ }
+}
+
+/** @} */
+
+} // namespace quickstep
+
+#endif // QUICKSTEP_UTILITY_LIP_FILTER_LIP_FILTER_UTIL_HPP_