Merge branch 'master' into in-operator-support
diff --git a/relational_operators/CMakeLists.txt b/relational_operators/CMakeLists.txt
index eea94cd..ac41912 100644
--- a/relational_operators/CMakeLists.txt
+++ b/relational_operators/CMakeLists.txt
@@ -168,11 +168,13 @@
quickstep_storage_StorageBlock
quickstep_storage_StorageBlockInfo
quickstep_storage_StorageManager
+ quickstep_storage_SubBlocksReference
quickstep_storage_TupleReference
quickstep_storage_TupleStorageSubBlock
quickstep_storage_ValueAccessor
quickstep_types_containers_ColumnVectorsValueAccessor
quickstep_utility_Macros
+ quickstep_utility_PtrList
tmb)
target_link_libraries(quickstep_relationaloperators_InsertOperator
glog
diff --git a/relational_operators/HashJoinOperator.cpp b/relational_operators/HashJoinOperator.cpp
index d947340..404006f 100644
--- a/relational_operators/HashJoinOperator.cpp
+++ b/relational_operators/HashJoinOperator.cpp
@@ -35,6 +35,7 @@
#include "storage/StorageBlock.hpp"
#include "storage/StorageBlockInfo.hpp"
#include "storage/StorageManager.hpp"
+#include "storage/SubBlocksReference.hpp"
#include "storage/TupleReference.hpp"
#include "storage/TupleStorageSubBlock.hpp"
#include "storage/ValueAccessor.hpp"
@@ -85,7 +86,7 @@
// Consolidation is a no-op for this version, but we provide this trivial
// call so that MapBasedJoinedTupleCollector and
// VectorBasedJoinedTupleCollector have the same interface and can both be
- // used in the templated HashJoinWorkOrder::executeWithCollectorType() method.
+ // used in the templated HashInnerJoinWorkOrder::executeWithCollectorType() method.
inline void consolidate() const {
}
@@ -183,6 +184,57 @@
consolidated_joined_tuples_;
};
+class SemiAntiJoinTupleCollector {
+ public:
+ SemiAntiJoinTupleCollector(const TupleStorageSubBlock &tuple_store) {
+ filter_.reset(tuple_store.getExistenceMap());
+ }
+
+ 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_;
+};
+
+class OuterJoinTupleCollector {
+ public:
+ OuterJoinTupleCollector(const TupleStorageSubBlock &tuple_store) {
+ filter_.reset(tuple_store.getExistenceMap());
+ }
+
+ template <typename ValueAccessorT>
+ inline void operator()(const ValueAccessorT &accessor,
+ const TupleReference &tref) {
+ joined_tuples_[tref.block].emplace_back(tref.tuple, accessor.getCurrentPosition());
+ }
+
+ template <typename ValueAccessorT>
+ inline void recordMatch(const ValueAccessorT &accessor) {
+ filter_->set(accessor.getCurrentPosition(), false);
+ }
+
+ inline std::unordered_map<block_id, std::vector<std::pair<tuple_id, tuple_id>>>*
+ getJoinedTupleMap() {
+ 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_;
+};
+
} // namespace
bool HashJoinOperator::getAllWorkOrders(
@@ -191,31 +243,57 @@
StorageManager *storage_manager,
const tmb::client_id foreman_client_id,
tmb::MessageBus *bus) {
+ switch (join_type_) {
+ case JoinType::kInnerJoin:
+ return getAllNonOuterJoinWorkOrders<HashInnerJoinWorkOrder>(
+ container, query_context, storage_manager);
+ case JoinType::kLeftOuterJoin:
+ return getAllOuterJoinWorkOrders(container, query_context,
+ storage_manager);
+ case JoinType::kLeftSemiJoin:
+ return getAllNonOuterJoinWorkOrders<HashSemiJoinWorkOrder>(
+ container, query_context, storage_manager);
+ case JoinType::kLeftAntiJoin:
+ return getAllNonOuterJoinWorkOrders<HashAntiJoinWorkOrder>(
+ container, query_context, storage_manager);
+ default:
+ LOG(FATAL) << "Unknown join type in HashJoinOperator::getAllWorkOrders()";
+ }
+}
+
+template <class JoinWorkOrderClass>
+bool HashJoinOperator::getAllNonOuterJoinWorkOrders(
+ WorkOrdersContainer *container,
+ QueryContext *query_context,
+ StorageManager *storage_manager) {
// We wait until the building of global hash table is complete.
if (blocking_dependencies_met_) {
DCHECK(query_context != nullptr);
- const Predicate *residual_predicate = query_context->getPredicate(residual_predicate_index_);
+ const Predicate *residual_predicate =
+ query_context->getPredicate(residual_predicate_index_);
const vector<unique_ptr<const Scalar>> &selection =
- query_context->getScalarGroup(selection_index_);
+ query_context->getScalarGroup(selection_on_probe_index_);
InsertDestination *output_destination =
query_context->getInsertDestination(output_destination_index_);
- JoinHashTable *hash_table = query_context->getJoinHashTable(hash_table_index_);
+ const JoinHashTable &hash_table =
+ *(query_context->getJoinHashTable(hash_table_index_));
if (probe_relation_is_stored_) {
if (!started_) {
for (const block_id probe_block_id : probe_relation_block_ids_) {
container->addNormalWorkOrder(
- new HashJoinWorkOrder(build_relation_,
- probe_relation_,
- join_key_attributes_,
- any_join_key_attributes_nullable_,
- probe_block_id,
- residual_predicate,
- selection,
- output_destination,
- hash_table,
- storage_manager),
+ new JoinWorkOrderClass(
+ build_relation_,
+ probe_relation_,
+ join_key_attributes_,
+ any_join_key_attributes_nullable_,
+ probe_block_id,
+ selection,
+ hash_table,
+ residual_predicate,
+ output_destination,
+ storage_manager),
op_index_);
}
started_ = true;
@@ -224,27 +302,99 @@
} else {
while (num_workorders_generated_ < probe_relation_block_ids_.size()) {
container->addNormalWorkOrder(
- new HashJoinWorkOrder(
+ new JoinWorkOrderClass(
build_relation_,
probe_relation_,
join_key_attributes_,
any_join_key_attributes_nullable_,
probe_relation_block_ids_[num_workorders_generated_],
- residual_predicate,
selection,
- output_destination,
hash_table,
+ residual_predicate,
+ output_destination,
storage_manager),
op_index_);
++num_workorders_generated_;
- } // end while
+ }
return done_feeding_input_relation_;
- } // end else (input_relation_is_stored is false)
- } // end if (blocking_dependencies_met)
+ } // end else (probe_relation_is_stored_)
+ } // end if (blocking_dependencies_met_)
return false;
}
-void HashJoinWorkOrder::execute() {
+bool HashJoinOperator::getAllOuterJoinWorkOrders(
+ WorkOrdersContainer *container,
+ QueryContext *query_context,
+ StorageManager *storage_manager) {
+ // We wait until the building of global hash table is complete.
+ if (blocking_dependencies_met_) {
+ DCHECK(query_context != nullptr);
+
+ const vector<unique_ptr<const Scalar>> &selection_on_probe =
+ query_context->getScalarGroup(selection_on_probe_index_);
+ const vector<unique_ptr<const Scalar>> &selection_on_build =
+ query_context->getScalarGroup(selection_on_build_index_);
+ InsertDestination *output_destination =
+ query_context->getInsertDestination(output_destination_index_);
+ const JoinHashTable &hash_table =
+ *(query_context->getJoinHashTable(hash_table_index_));
+
+ // TODO(harshad, jianqiao) Construct the vector below in ExecutionGenerator
+ // and pass it as an argument to the HashJoinOperator.
+ std::vector<const Type*> selection_on_build_types;
+ for (auto selection_on_build_it = selection_on_build.begin();
+ selection_on_build_it != selection_on_build.end();
+ ++selection_on_build_it) {
+ selection_on_build_types.emplace_back(
+ (&(*selection_on_build_it)->getType().getNullableVersion()));
+ }
+
+ if (probe_relation_is_stored_) {
+ if (!started_) {
+ for (const block_id probe_block_id : probe_relation_block_ids_) {
+ container->addNormalWorkOrder(
+ new HashOuterJoinWorkOrder(
+ build_relation_,
+ probe_relation_,
+ join_key_attributes_,
+ any_join_key_attributes_nullable_,
+ hash_table,
+ selection_on_probe,
+ selection_on_build,
+ selection_on_build_types,
+ probe_block_id,
+ output_destination,
+ storage_manager),
+ op_index_);
+ }
+ started_ = true;
+ }
+ return started_;
+ } else {
+ while (num_workorders_generated_ < probe_relation_block_ids_.size()) {
+ container->addNormalWorkOrder(
+ new HashOuterJoinWorkOrder(
+ build_relation_,
+ probe_relation_,
+ join_key_attributes_,
+ any_join_key_attributes_nullable_,
+ hash_table,
+ selection_on_probe,
+ selection_on_build,
+ selection_on_build_types,
+ probe_relation_block_ids_[num_workorders_generated_],
+ output_destination,
+ storage_manager),
+ op_index_);
+ ++num_workorders_generated_;
+ }
+ return done_feeding_input_relation_;
+ } // end else (probe_relation_is_stored_)
+ } // end if (blocking_dependencies_met_)
+ return false;
+}
+
+void HashInnerJoinWorkOrder::execute() {
if (FLAGS_vector_based_joined_tuple_collector) {
executeWithCollectorType<VectorBasedJoinedTupleCollector>();
} else {
@@ -253,7 +403,7 @@
}
template <typename CollectorT>
-void HashJoinWorkOrder::executeWithCollectorType() {
+void HashInnerJoinWorkOrder::executeWithCollectorType() {
BlockReference probe_block(
storage_manager_->getBlock(block_id_, probe_relation_));
const TupleStorageSubBlock &probe_store = probe_block->getTupleStorageSubBlock();
@@ -261,13 +411,13 @@
std::unique_ptr<ValueAccessor> probe_accessor(probe_store.createValueAccessor());
CollectorT collector;
if (join_key_attributes_.size() == 1) {
- hash_table_->getAllFromValueAccessor(
+ hash_table_.getAllFromValueAccessor(
probe_accessor.get(),
join_key_attributes_.front(),
any_join_key_attributes_nullable_,
&collector);
} else {
- hash_table_->getAllFromValueAccessorCompositeKey(
+ hash_table_.getAllFromValueAccessorCompositeKey(
probe_accessor.get(),
join_key_attributes_,
any_join_key_attributes_nullable_,
@@ -348,4 +498,359 @@
}
}
+void HashSemiJoinWorkOrder::execute() {
+ if (residual_predicate_ == nullptr) {
+ executeWithoutResidualPredicate();
+ } else {
+ executeWithResidualPredicate();
+ }
+}
+
+void HashSemiJoinWorkOrder::executeWithResidualPredicate() {
+ const relation_id build_relation_id = build_relation_.getID();
+ const relation_id probe_relation_id = probe_relation_.getID();
+
+ 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());
+
+ // TODO(harshad) - Make this function work with both types of collectors.
+
+ // 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;
+ if (join_key_attributes_.size() == 1) {
+ hash_table_.getAllFromValueAccessor(
+ probe_accessor.get(),
+ join_key_attributes_.front(),
+ any_join_key_attributes_nullable_,
+ &collector);
+ } else {
+ hash_table_.getAllFromValueAccessorCompositeKey(
+ probe_accessor.get(),
+ join_key_attributes_,
+ any_join_key_attributes_nullable_,
+ &collector);
+ }
+
+ // 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()) {
+ // First element of the pair build_block_entry is the build block ID
+ // 2nd element of the pair is a vector of pairs, in each of which -
+ // 1st element is a matching tuple ID from the inner (build) relation.
+ // 2nd element is a matching tuple ID from the outer (probe) relation.
+
+ // Get the block from the build relation for this pair of matched tuples.
+ BlockReference build_block =
+ storage_manager_->getBlock(build_block_entry.first, build_relation_);
+ const TupleStorageSubBlock &build_store =
+ build_block->getTupleStorageSubBlock();
+ std::unique_ptr<ValueAccessor> build_accessor(
+ build_store.createValueAccessor());
+ for (const std::pair<tuple_id, tuple_id> &hash_match
+ : build_block_entry.second) {
+ // For each pair, 1st element is a tuple ID from the build relation in the
+ // given build block, 2nd element is a tuple ID from the probe relation.
+ if (filter.get(hash_match.second)) {
+ // We have already found matches for this tuple that belongs to the
+ // probe side, skip it.
+ continue;
+ }
+ if (residual_predicate_->matchesForJoinedTuples(*build_accessor,
+ build_relation_id,
+ hash_match.first,
+ *probe_accessor,
+ probe_relation_id,
+ hash_match.second)) {
+ filter.set(hash_match.second, true);
+ }
+ }
+ }
+
+ SubBlocksReference sub_blocks_ref(probe_store,
+ probe_block->getIndices(),
+ probe_block->getIndicesConsistent());
+
+ std::unique_ptr<ValueAccessor> probe_accessor_with_filter(
+ probe_store.createValueAccessor(&filter));
+ ColumnVectorsValueAccessor temp_result;
+ for (vector<unique_ptr<const Scalar>>::const_iterator selection_it = selection_.begin();
+ selection_it != selection_.end();
+ ++selection_it) {
+ temp_result.addColumn((*selection_it)->getAllValues(
+ probe_accessor_with_filter.get(), &sub_blocks_ref));
+ }
+
+ output_destination_->bulkInsertTuples(&temp_result);
+}
+
+void HashSemiJoinWorkOrder::executeWithoutResidualPredicate() {
+ DCHECK(residual_predicate_ == nullptr);
+
+ 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);
+ // 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
+ // the bit for the probing key in the collector. The optimization works
+ // because there is no residual predicate in this case, unlike
+ // executeWithResidualPredicate().
+ if (join_key_attributes_.size() == 1u) {
+ // Call the collector to set the bit to 0 for every key without a match.
+ hash_table_.runOverKeysFromValueAccessorIfMatchNotFound(
+ probe_accessor.get(),
+ join_key_attributes_.front(),
+ any_join_key_attributes_nullable_,
+ &collector);
+ } else {
+ // Call the collector to set the bit to 0 for every key without a match.
+ hash_table_.runOverKeysFromValueAccessorIfMatchNotFoundCompositeKey(
+ probe_accessor.get(),
+ join_key_attributes_,
+ any_join_key_attributes_nullable_,
+ &collector);
+ }
+
+ SubBlocksReference sub_blocks_ref(probe_store,
+ probe_block->getIndices(),
+ probe_block->getIndicesConsistent());
+
+ std::unique_ptr<ValueAccessor> probe_accessor_with_filter(
+ probe_store.createValueAccessor(collector.filter()));
+ ColumnVectorsValueAccessor temp_result;
+ for (vector<unique_ptr<const Scalar>>::const_iterator selection_it = selection_.begin();
+ selection_it != selection_.end(); ++selection_it) {
+ temp_result.addColumn((*selection_it)->getAllValues(
+ probe_accessor_with_filter.get(), &sub_blocks_ref));
+ }
+
+ output_destination_->bulkInsertTuples(&temp_result);
+}
+
+void HashAntiJoinWorkOrder::executeWithoutResidualPredicate() {
+ DEBUG_ASSERT(residual_predicate_ == nullptr);
+
+ 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);
+ // We probe the hash table to find the keys which have an entry in the
+ // hash table.
+ if (join_key_attributes_.size() == 1) {
+ // Call the collector to set the bit to 0 for every key with a match.
+ hash_table_.runOverKeysFromValueAccessorIfMatchFound(
+ probe_accessor.get(),
+ join_key_attributes_.front(),
+ any_join_key_attributes_nullable_,
+ &collector);
+ } else {
+ // Call the collector to set the bit to 0 for every key with a match.
+ hash_table_.runOverKeysFromValueAccessorIfMatchFoundCompositeKey(
+ probe_accessor.get(),
+ join_key_attributes_,
+ any_join_key_attributes_nullable_,
+ &collector);
+ }
+
+ SubBlocksReference sub_blocks_ref(probe_store,
+ probe_block->getIndices(),
+ probe_block->getIndicesConsistent());
+
+ std::unique_ptr<ValueAccessor> probe_accessor_with_filter(
+ probe_store.createValueAccessor(collector.filter()));
+ ColumnVectorsValueAccessor temp_result;
+ for (vector<unique_ptr<const Scalar>>::const_iterator selection_it = selection_.begin();
+ selection_it != selection_.end(); ++selection_it) {
+ temp_result.addColumn((*selection_it)->getAllValues(
+ probe_accessor_with_filter.get(), &sub_blocks_ref));
+ }
+
+ output_destination_->bulkInsertTuples(&temp_result);
+}
+
+void HashAntiJoinWorkOrder::executeWithResidualPredicate() {
+ const relation_id build_relation_id = build_relation_.getID();
+ const relation_id probe_relation_id = probe_relation_.getID();
+
+ 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());
+ // TODO(harshad) - Make the following code work with both types of collectors.
+ MapBasedJoinedTupleCollector collector;
+ // We probe the hash table and get all the matches. Unlike
+ // executeWithoutResidualPredicate(), we have to collect all the matching
+ // tuples, because after this step we still have to evalute the residual
+ // predicate.
+ if (join_key_attributes_.size() == 1) {
+ hash_table_.getAllFromValueAccessor(
+ probe_accessor.get(),
+ join_key_attributes_.front(),
+ any_join_key_attributes_nullable_,
+ &collector);
+ } else {
+ hash_table_.getAllFromValueAccessorCompositeKey(
+ probe_accessor.get(),
+ join_key_attributes_,
+ any_join_key_attributes_nullable_,
+ &collector);
+ }
+
+ // Create a filter for all the tuples from the given probe block.
+ std::unique_ptr<TupleIdSequence> filter(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
+ // 2nd element of the pair is a vector of pairs, in each of which -
+ // 1st element is a matching tuple ID from the inner (build) relation.
+ // 2nd element is a matching tuple ID from the outer (probe) relation.
+ BlockReference build_block = storage_manager_->getBlock(build_block_entry.first,
+ build_relation_);
+ const TupleStorageSubBlock &build_store = build_block->getTupleStorageSubBlock();
+ 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)) {
+ // We have already seen this tuple, skip it.
+ continue;
+ }
+ if (residual_predicate_->matchesForJoinedTuples(*build_accessor,
+ build_relation_id,
+ hash_match.first,
+ *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);
+ }
+ }
+ }
+
+ SubBlocksReference sub_blocks_ref(probe_store,
+ probe_block->getIndices(),
+ probe_block->getIndicesConsistent());
+
+ std::unique_ptr<ValueAccessor> probe_accessor_with_filter(
+ probe_store.createValueAccessor(filter.get()));
+ ColumnVectorsValueAccessor temp_result;
+ for (vector<unique_ptr<const Scalar>>::const_iterator selection_it = selection_.begin();
+ selection_it != selection_.end();
+ ++selection_it) {
+ temp_result.addColumn((*selection_it)->getAllValues(probe_accessor_with_filter.get(),
+ &sub_blocks_ref));
+ }
+
+ output_destination_->bulkInsertTuples(&temp_result);
+}
+
+void HashOuterJoinWorkOrder::execute() {
+ const relation_id build_relation_id = build_relation_.getID();
+ const relation_id probe_relation_id = probe_relation_.getID();
+
+ 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);
+ if (join_key_attributes_.size() == 1) {
+ hash_table_.getAllFromValueAccessorWithExtraWorkForFirstMatch(
+ probe_accessor.get(),
+ join_key_attributes_.front(),
+ any_join_key_attributes_nullable_,
+ &collector);
+ } else {
+ hash_table_.getAllFromValueAccessorCompositeKeyWithExtraWorkForFirstMatch(
+ probe_accessor.get(),
+ join_key_attributes_,
+ any_join_key_attributes_nullable_,
+ &collector);
+ }
+
+ for (const std::pair<const block_id, std::vector<std::pair<tuple_id, tuple_id>>>
+ &build_block_entry : *collector.getJoinedTupleMap()) {
+ BlockReference build_block =
+ storage_manager_->getBlock(build_block_entry.first, build_relation_);
+ const TupleStorageSubBlock &build_store =
+ build_block->getTupleStorageSubBlock();
+
+ std::unique_ptr<ValueAccessor> build_accessor(
+ build_store.createValueAccessor());
+ ColumnVectorsValueAccessor temp_result;
+ for (auto selection_it = selection_on_probe_.begin();
+ selection_it != selection_on_probe_.end();
+ ++selection_it) {
+ temp_result.addColumn(
+ (*selection_it)->getAllValuesForJoin(
+ build_relation_id,
+ build_accessor.get(),
+ probe_relation_id,
+ probe_accessor.get(),
+ build_block_entry.second));
+ }
+ for (auto selection_it = selection_on_build_.begin();
+ selection_it != selection_on_build_.end();
+ ++selection_it) {
+ temp_result.addColumn(
+ (*selection_it)->getAllValuesForJoin(
+ build_relation_id,
+ build_accessor.get(),
+ probe_relation_id,
+ probe_accessor.get(),
+ build_block_entry.second));
+ }
+
+ output_destination_->bulkInsertTuples(&temp_result);
+ }
+
+ SubBlocksReference sub_blocks_ref(probe_store,
+ probe_block->getIndices(),
+ 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();
+ if (num_tuples_without_matches > 0) {
+ std::unique_ptr<ValueAccessor> probe_accessor_with_filter(
+ probe_store.createValueAccessor(filter));
+ ColumnVectorsValueAccessor temp_result;
+ for (auto selection_it = selection_on_probe_.begin();
+ selection_it != selection_on_probe_.end();
+ ++selection_it) {
+ temp_result.addColumn(
+ (*selection_it)->getAllValues(probe_accessor_with_filter.get(),
+ &sub_blocks_ref));
+ }
+
+ for (const Type *selection_on_build_type : selection_on_build_types_) {
+ if (NativeColumnVector::UsableForType(*selection_on_build_type)) {
+ NativeColumnVector *result = new NativeColumnVector(
+ *selection_on_build_type, num_tuples_without_matches);
+ result->fillWithNulls();
+ temp_result.addColumn(result);
+ } else {
+ IndirectColumnVector *result = new IndirectColumnVector(
+ *selection_on_build_type, num_tuples_without_matches);
+ result->fillWithValue(TypedValue(selection_on_build_type->getTypeID()));
+ temp_result.addColumn(result);
+ }
+ }
+ output_destination_->bulkInsertTuples(&temp_result);
+ }
+}
+
} // namespace quickstep
diff --git a/relational_operators/HashJoinOperator.hpp b/relational_operators/HashJoinOperator.hpp
index 9b8c87b..4141e14 100644
--- a/relational_operators/HashJoinOperator.hpp
+++ b/relational_operators/HashJoinOperator.hpp
@@ -30,6 +30,7 @@
#include "storage/HashTable.hpp"
#include "storage/StorageBlockInfo.hpp"
#include "utility/Macros.hpp"
+#include "utility/PtrList.hpp"
#include "glog/logging.h"
@@ -51,11 +52,18 @@
*/
/**
- * @brief An operator which performs a hash-join on two
- * relations.
+ * @brief An operator which performs a hash-join, including inner-join,
+ * semi-join, anti-join and outer-join on two relations.
**/
class HashJoinOperator : public RelationalOperator {
public:
+ enum class JoinType {
+ kInnerJoin = 0,
+ kLeftOuterJoin,
+ kLeftSemiJoin,
+ kLeftAntiJoin
+ };
+
/**
* @brief Constructor.
*
@@ -93,35 +101,48 @@
* additional filter to pairs of tuples that match the hash-join (i.e.
* key equality) predicate. Effectively, this makes the join predicate
* the conjunction of the key-equality predicate and residual predicate.
- * @param selection_index The group index of Scalars in QueryContext,
- * corresponding to the attributes of the relation referred by
+ * Note that this field is not relevant for anti-join.
+ * @param selection_on_probe_index The group index of Scalars in QueryContext,
+ * corresponding to the attributes of the probe relation referred by
* output_relation_id. Each Scalar is evaluated for the joined tuples,
* and the resulting value is inserted into the join result.
+ * @param selection_on_build_index The group index of Scalars in QueryContext,
+ * corresponding to the attributes of the build relation referred by
+ * output_relation_id. Each Scalar is evaluated for the joined tuples,
+ * and the resulting value is inserted into the join result.
+ * @param join_type The type of join corresponding to this operator.
**/
- HashJoinOperator(const CatalogRelation &build_relation,
- const CatalogRelation &probe_relation,
- const bool probe_relation_is_stored,
- const std::vector<attribute_id> &join_key_attributes,
- const bool any_join_key_attributes_nullable,
- const CatalogRelation &output_relation,
- const QueryContext::insert_destination_id output_destination_index,
- const QueryContext::join_hash_table_id hash_table_index,
- const QueryContext::predicate_id residual_predicate_index,
- const QueryContext::scalar_group_id selection_index)
- : build_relation_(build_relation),
- probe_relation_(probe_relation),
- probe_relation_is_stored_(probe_relation_is_stored),
- join_key_attributes_(join_key_attributes),
- any_join_key_attributes_nullable_(any_join_key_attributes_nullable),
- output_relation_(output_relation),
- output_destination_index_(output_destination_index),
- hash_table_index_(hash_table_index),
- residual_predicate_index_(residual_predicate_index),
- selection_index_(selection_index),
- probe_relation_block_ids_(probe_relation_is_stored ? probe_relation.getBlocksSnapshot()
- : std::vector<block_id>()),
- num_workorders_generated_(0),
- started_(false) {}
+ HashJoinOperator(
+ const CatalogRelation &build_relation,
+ const CatalogRelation &probe_relation,
+ const bool probe_relation_is_stored,
+ const std::vector<attribute_id> &join_key_attributes,
+ const bool any_join_key_attributes_nullable,
+ const CatalogRelation &output_relation,
+ const QueryContext::insert_destination_id output_destination_index,
+ const QueryContext::join_hash_table_id hash_table_index,
+ const QueryContext::predicate_id residual_predicate_index,
+ const QueryContext::scalar_group_id selection_on_probe_index,
+ const QueryContext::scalar_group_id selection_on_build_index =
+ QueryContext::kInvalidScalarGroupId,
+ const JoinType join_type = JoinType::kInnerJoin)
+ : build_relation_(build_relation),
+ probe_relation_(probe_relation),
+ probe_relation_is_stored_(probe_relation_is_stored),
+ join_key_attributes_(join_key_attributes),
+ any_join_key_attributes_nullable_(any_join_key_attributes_nullable),
+ output_relation_(output_relation),
+ output_destination_index_(output_destination_index),
+ hash_table_index_(hash_table_index),
+ residual_predicate_index_(residual_predicate_index),
+ selection_on_probe_index_(selection_on_probe_index),
+ selection_on_build_index_(selection_on_build_index),
+ join_type_(join_type),
+ probe_relation_block_ids_(probe_relation_is_stored
+ ? probe_relation.getBlocksSnapshot()
+ : std::vector<block_id>()),
+ num_workorders_generated_(0),
+ started_(false) {}
~HashJoinOperator() override {}
@@ -163,6 +184,15 @@
}
private:
+ template <class JoinWorkOrderClass>
+ bool getAllNonOuterJoinWorkOrders(WorkOrdersContainer *container,
+ QueryContext *query_context,
+ StorageManager *storage_manager);
+
+ bool getAllOuterJoinWorkOrders(WorkOrdersContainer *container,
+ QueryContext *query_context,
+ StorageManager *storage_manager);
+
const CatalogRelation &build_relation_;
const CatalogRelation &probe_relation_;
const bool probe_relation_is_stored_;
@@ -172,7 +202,9 @@
const QueryContext::insert_destination_id output_destination_index_;
const QueryContext::join_hash_table_id hash_table_index_;
const QueryContext::predicate_id residual_predicate_index_;
- const QueryContext::scalar_group_id selection_index_;
+ const QueryContext::scalar_group_id selection_on_probe_index_;
+ const QueryContext::scalar_group_id selection_on_build_index_;
+ const JoinType join_type_;
std::vector<block_id> probe_relation_block_ids_;
std::size_t num_workorders_generated_;
@@ -183,9 +215,9 @@
};
/**
- * @brief A WorkOrder produced by HashJoinOperator.
+ * @brief An inner join WorkOrder produced by HashJoinOperator.
**/
-class HashJoinWorkOrder : public WorkOrder {
+class HashInnerJoinWorkOrder : public WorkOrder {
public:
/**
* @brief Constructor.
@@ -198,26 +230,26 @@
* probe_relation.
* @param any_join_key_attributes_nullable If any attribute is nullable.
* @param lookup_block_id The block id of the probe_relation.
+ * @param selection A list of Scalars corresponding to the relation attributes
+ * in \c output_destination. Each Scalar is evaluated for the joined
+ * tuples, and the resulting value is inserted into the join result.
+ * @param hash_table The JoinHashTable to use.
* @param residual_predicate If non-null, apply as an additional filter to
* pairs of tuples that match the hash-join (i.e. key equality)
* predicate. Effectively, this makes the join predicate the
* conjunction of the key-equality predicate and residual_predicate.
- * @param selection A list of Scalars corresponding to the relation attributes
- * in \c output_destination. Each Scalar is evaluated for the joined
- * tuples, and the resulting value is inserted into the join result.
* @param output_destination The InsertDestination to insert the join results.
- * @param hash_table The JoinHashTable to use.
* @param storage_manager The StorageManager to use.
**/
- HashJoinWorkOrder(const CatalogRelationSchema &build_relation,
+ HashInnerJoinWorkOrder(const CatalogRelationSchema &build_relation,
const CatalogRelationSchema &probe_relation,
const std::vector<attribute_id> &join_key_attributes,
const bool any_join_key_attributes_nullable,
const block_id lookup_block_id,
- const Predicate *residual_predicate,
const std::vector<std::unique_ptr<const Scalar>> &selection,
+ const JoinHashTable &hash_table,
+ const Predicate *residual_predicate,
InsertDestination *output_destination,
- JoinHashTable *hash_table,
StorageManager *storage_manager)
: build_relation_(build_relation),
probe_relation_(probe_relation),
@@ -226,11 +258,11 @@
block_id_(lookup_block_id),
residual_predicate_(residual_predicate),
selection_(selection),
+ hash_table_(hash_table),
output_destination_(DCHECK_NOTNULL(output_destination)),
- hash_table_(DCHECK_NOTNULL(hash_table)),
storage_manager_(DCHECK_NOTNULL(storage_manager)) {}
- ~HashJoinWorkOrder() override {}
+ ~HashInnerJoinWorkOrder() override {}
/**
* @exception TupleTooLargeForBlock A tuple produced by this join was too
@@ -253,11 +285,233 @@
const Predicate *residual_predicate_;
const std::vector<std::unique_ptr<const Scalar>> &selection_;
+ const JoinHashTable &hash_table_;
InsertDestination *output_destination_;
- JoinHashTable *hash_table_;
StorageManager *storage_manager_;
- DISALLOW_COPY_AND_ASSIGN(HashJoinWorkOrder);
+ DISALLOW_COPY_AND_ASSIGN(HashInnerJoinWorkOrder);
+};
+
+/**
+ * @brief A left semi-join WorkOrder produced by the HashJoinOperator to execute
+ * EXISTS() clause.
+ **/
+class HashSemiJoinWorkOrder : public WorkOrder {
+ public:
+ /**
+ * @brief Constructor.
+ *
+ * @param build_relation The relation that the hash table was originally built
+ * on (i.e. the inner relation in the join).
+ * @param probe_relation The relation to probe the hash table with (i.e. the
+ * outer relation in the join).
+ * @param join_key_attributes The IDs of equijoin attributes in \c
+ * probe_relation.
+ * @param any_join_key_attributes_nullable If any attribute is nullable.
+ * @param lookup_block_id The block id of the probe_relation.
+ * @param selection A list of Scalars corresponding to the relation attributes
+ * in \c output_destination. Each Scalar is evaluated for the joined
+ * tuples, and the resulting value is inserted into the join result.
+ * @param hash_table The JoinHashTable to use.
+ * @param residual_predicate If non-null, apply as an additional filter to
+ * pairs of tuples that match the hash-join (i.e. key equality)
+ * predicate. Effectively, this makes the join predicate the
+ * conjunction of the key-equality predicate and residual_predicate.
+ * @param output_destination The InsertDestination to insert the join results.
+ * @param storage_manager The StorageManager to use.
+ **/
+ HashSemiJoinWorkOrder(const CatalogRelationSchema &build_relation,
+ const CatalogRelationSchema &probe_relation,
+ const std::vector<attribute_id> &join_key_attributes,
+ const bool any_join_key_attributes_nullable,
+ const block_id lookup_block_id,
+ const std::vector<std::unique_ptr<const Scalar>> &selection,
+ const JoinHashTable &hash_table,
+ const Predicate *residual_predicate,
+ InsertDestination *output_destination,
+ StorageManager *storage_manager)
+ : build_relation_(build_relation),
+ probe_relation_(probe_relation),
+ join_key_attributes_(join_key_attributes),
+ any_join_key_attributes_nullable_(any_join_key_attributes_nullable),
+ hash_table_(hash_table),
+ selection_(selection),
+ residual_predicate_(residual_predicate),
+ block_id_(lookup_block_id),
+ output_destination_(output_destination),
+ storage_manager_(storage_manager) {}
+
+ ~HashSemiJoinWorkOrder() override {}
+
+ void execute() override;
+
+ private:
+ void executeWithoutResidualPredicate();
+
+ void executeWithResidualPredicate();
+
+ const CatalogRelationSchema &build_relation_;
+ const CatalogRelationSchema &probe_relation_;
+ const std::vector<attribute_id> join_key_attributes_;
+ const bool any_join_key_attributes_nullable_;
+ const JoinHashTable &hash_table_;
+ const std::vector<std::unique_ptr<const Scalar>> &selection_;
+ const Predicate *residual_predicate_;
+ const block_id block_id_;
+
+ InsertDestination *output_destination_;
+ StorageManager *storage_manager_;
+
+ DISALLOW_COPY_AND_ASSIGN(HashSemiJoinWorkOrder);
+};
+
+/**
+ * @brief A left anti-join WorkOrder produced by the HashJoinOperator to execute
+ * NOT EXISTS() clause.
+ **/
+class HashAntiJoinWorkOrder : public WorkOrder {
+ public:
+ /**
+ * @brief Constructor.
+ * TODO(harshad) - Sync the doxygen.
+ *
+ * @param build_relation The relation that the hash table was originally built
+ * on (i.e. the inner relation in the join).
+ * @param probe_relation The relation to probe the hash table with (i.e. the
+ * outer relation in the join).
+ * @param join_key_attributes The IDs of equijoin attributes in \c
+ * probe_relation.
+ * @param any_join_key_attributes_nullable If any attribute is nullable.
+ * @param lookup_block_id The block id of the probe_relation.
+ * @param selection A list of Scalars corresponding to the relation attributes
+ * in \c output_destination. Each Scalar is evaluated for the joined
+ * tuples, and the resulting value is inserted into the join result.
+ * @param hash_table The JoinHashTable to use.
+ * @param residual_predicate If non-null, apply as an additional filter to
+ * pairs of tuples that match the hash-join (i.e. key equality)
+ * predicate. Effectively, this makes the join predicate the
+ * conjunction of the key-equality predicate and residual_predicate.
+ * @param output_destination The InsertDestination to insert the join results.
+ * @param storage_manager The StorageManager to use.
+ **/
+ HashAntiJoinWorkOrder(const CatalogRelationSchema &build_relation,
+ const CatalogRelationSchema &probe_relation,
+ const std::vector<attribute_id> &join_key_attributes,
+ const bool any_join_key_attributes_nullable,
+ const block_id lookup_block_id,
+ const std::vector<std::unique_ptr<const Scalar>> &selection,
+ const JoinHashTable &hash_table,
+ const Predicate *residual_predicate,
+ InsertDestination *output_destination,
+ StorageManager *storage_manager)
+ : build_relation_(build_relation),
+ probe_relation_(probe_relation),
+ join_key_attributes_(join_key_attributes),
+ any_join_key_attributes_nullable_(any_join_key_attributes_nullable),
+ selection_(selection),
+ residual_predicate_(residual_predicate),
+ block_id_(lookup_block_id),
+ hash_table_(hash_table),
+ output_destination_(output_destination),
+ storage_manager_(storage_manager) {}
+
+ ~HashAntiJoinWorkOrder() override {}
+
+ void execute() override {
+ if (residual_predicate_ == nullptr) {
+ executeWithoutResidualPredicate();
+ } else {
+ executeWithResidualPredicate();
+ }
+ }
+
+ private:
+ void executeWithoutResidualPredicate();
+
+ void executeWithResidualPredicate();
+
+ const CatalogRelationSchema &build_relation_;
+ const CatalogRelationSchema &probe_relation_;
+ const std::vector<attribute_id> join_key_attributes_;
+ const bool any_join_key_attributes_nullable_;
+ const std::vector<std::unique_ptr<const Scalar>> &selection_;
+ const Predicate *residual_predicate_;
+ const block_id block_id_;
+ const JoinHashTable &hash_table_;
+
+ InsertDestination *output_destination_;
+ StorageManager *storage_manager_;
+
+ DISALLOW_COPY_AND_ASSIGN(HashAntiJoinWorkOrder);
+};
+
+/**
+ * @brief A left outer join WorkOrder produced by the HashJoinOperator.
+ **/
+ class HashOuterJoinWorkOrder : public WorkOrder {
+ public:
+ /**
+ * @brief Constructor.
+ *
+ * @param build_relation The relation that the hash table was originally built
+ * on (i.e. the inner relation in the join).
+ * @param probe_relation The relation to probe the hash table with (i.e. the
+ * outer relation in the join).
+ * @param join_key_attributes The IDs of equijoin attributes in \c
+ * probe_relation.
+ * @param any_join_key_attributes_nullable If any attribute is nullable.
+ * @param hash_table The JoinHashTable to use.
+ * @param selection_on_probe A list of Scalars from probe relation,
+ * corresponding to the relation attributes in \c output_destination.
+ * @param selection_on_build A list of Scalars from build relation,
+ * corresponding to the relation attributes in \c output_destination.
+ * @param lookup_block_id The block id of the probe_relation.
+ * @param output_destination The InsertDestination to insert the join results.
+ * @param storage_manager The StorageManager to use.
+ **/
+ HashOuterJoinWorkOrder(const CatalogRelationSchema &build_relation,
+ const CatalogRelationSchema &probe_relation,
+ const std::vector<attribute_id> &join_key_attributes,
+ const bool any_join_key_attributes_nullable,
+ const JoinHashTable &hash_table,
+ const std::vector<std::unique_ptr<const Scalar>> &selection_on_probe,
+ const std::vector<std::unique_ptr<const Scalar>> &selection_on_build,
+ const std::vector<const Type*> &selection_on_build_types,
+ const block_id lookup_block_id,
+ InsertDestination *output_destination,
+ StorageManager *storage_manager)
+ : build_relation_(build_relation),
+ probe_relation_(probe_relation),
+ join_key_attributes_(join_key_attributes),
+ any_join_key_attributes_nullable_(any_join_key_attributes_nullable),
+ hash_table_(hash_table),
+ selection_on_probe_(selection_on_probe),
+ selection_on_build_(selection_on_build),
+ selection_on_build_types_(selection_on_build_types),
+ block_id_(lookup_block_id),
+ output_destination_(output_destination),
+ storage_manager_(storage_manager) {}
+
+ ~HashOuterJoinWorkOrder() override {}
+
+ void execute() override;
+
+ private:
+ const CatalogRelationSchema &build_relation_;
+ const CatalogRelationSchema &probe_relation_;
+ const std::vector<attribute_id> join_key_attributes_;
+ const bool any_join_key_attributes_nullable_;
+ const JoinHashTable &hash_table_;
+ const std::vector<std::unique_ptr<const Scalar>> &selection_on_probe_;
+ const std::vector<std::unique_ptr<const Scalar>> &selection_on_build_;
+ const std::vector<const Type*> &selection_on_build_types_;
+
+ const block_id block_id_;
+
+ InsertDestination *output_destination_;
+ StorageManager *storage_manager_;
+
+ DISALLOW_COPY_AND_ASSIGN(HashOuterJoinWorkOrder);
};
/** @} */
diff --git a/storage/HashTable.hpp b/storage/HashTable.hpp
index ab74e2c..0178970 100644
--- a/storage/HashTable.hpp
+++ b/storage/HashTable.hpp
@@ -707,6 +707,89 @@
FunctorT *functor) const;
/**
+ * @brief Lookup (multiple) keys from a ValueAccessor, apply a functor to
+ * the matching values and additionally call a hasMatch() function of
+ * the functor when the first match for a key is found.
+ * @warning This method assumes that no concurrent calls to put(),
+ * putCompositeKey(), putValueAccessor(),
+ * putValueAccessorCompositeKey(), upsert(), upsertCompositeKey(),
+ * upsertValueAccessor(), or upsertValueAccessorCompositeKey() are
+ * taking place (i.e. that this HashTable is immutable for the
+ * duration of the call and as long as the returned pointer may be
+ * dereferenced). Concurrent calls to getSingle(),
+ * getSingleCompositeKey(), getAll(), getAllCompositeKey(),
+ * getAllFromValueAccessor(), getAllFromValueAccessorCompositeKey(),
+ * forEach(), and forEachCompositeKey() are safe.
+ * @note This version is for single scalar keys. See also
+ * getAllFromValueAccessorCompositeKeyWithExtraWorkForFirstMatch().
+ *
+ * @param accessor A ValueAccessor which will be used to access keys.
+ * beginIteration() should be called on accessor before calling this
+ * method.
+ * @param key_attr_id The attribute ID of the keys to be read from accessor.
+ * @param check_for_null_keys If true, each key will be checked to see if it
+ * is null before looking it up (null keys are skipped). This must be
+ * set to true if some of the keys that will be read from accessor may
+ * be null.
+ * @param functor A pointer to a functor, which should provide two functions:
+ * 1) An operator that takes 2 arguments: const ValueAccessor& (or better
+ * yet, a templated call operator which takes a const reference to
+ * some subclass of ValueAccessor as its first argument) and
+ * const ValueT&. The operator will be invoked once for each pair of a
+ * key taken from accessor and matching value.
+ * 2) A function hasMatch that takes 1 argument: const ValueAccessor&.
+ * The function will be called only once for a key from accessor when
+ * the first match is found.
+ */
+ template <typename FunctorT>
+ void getAllFromValueAccessorWithExtraWorkForFirstMatch(
+ ValueAccessor *accessor,
+ const attribute_id key_attr_id,
+ const bool check_for_null_keys,
+ FunctorT *functor) const;
+
+ /**
+ * @brief Lookup (multiple) keys from a ValueAccessor, apply a functor to
+ * the matching values and additionally call a hasMatch() function of
+ * the functor when the first match for a key is found. Composite key
+ * version.
+ * @warning This method assumes that no concurrent calls to put(),
+ * putCompositeKey(), putValueAccessor(),
+ * putValueAccessorCompositeKey(), upsert(), upsertCompositeKey(),
+ * upsertValueAccessor(), or upsertValueAccessorCompositeKey() are
+ * taking place (i.e. that this HashTable is immutable for the
+ * duration of the call and as long as the returned pointer may be
+ * dereferenced). Concurrent calls to getSingle(),
+ * getSingleCompositeKey(), getAll(), getAllCompositeKey(),
+ * getAllFromValueAccessor(), getAllFromValueAccessorCompositeKey(),
+ * forEach(), and forEachCompositeKey() are safe.
+ *
+ * @param accessor A ValueAccessor which will be used to access keys.
+ * beginIteration() should be called on accessor before calling this
+ * method.
+ * @param key_attr_id The attribute ID of the keys to be read from accessor.
+ * @param check_for_null_keys If true, each key will be checked to see if it
+ * is null before looking it up (null keys are skipped). This must be
+ * set to true if some of the keys that will be read from accessor may
+ * be null.
+ * @param functor A pointer to a functor, which should provide two functions:
+ * 1) An operator that takes 2 arguments: const ValueAccessor& (or better
+ * yet, a templated call operator which takes a const reference to
+ * some subclass of ValueAccessor as its first argument) and
+ * const ValueT&. The operator will be invoked once for each pair of a
+ * key taken from accessor and matching value.
+ * 2) A function hasMatch that takes 1 argument: const ValueAccessor&.
+ * The function will be called only once for a key from accessor when
+ * the first match is found.
+ */
+ template <typename FunctorT>
+ void getAllFromValueAccessorCompositeKeyWithExtraWorkForFirstMatch(
+ ValueAccessor *accessor,
+ const std::vector<attribute_id> &key_attr_ids,
+ const bool check_for_null_keys,
+ FunctorT *functor) const;
+
+ /**
* @brief Lookup (multiple) keys from a ValueAccessor and apply a functor to
* the matching values. Composite key version.
*
@@ -746,6 +829,113 @@
FunctorT *functor) const;
/**
+ * @brief Apply the functor to each key with a match in the hash table.
+ *
+ * @param accessor A ValueAccessor which will be used to access keys.
+ * beginIteration() should be called on accessor before calling this
+ * method.
+ * @param key_attr_id The attribute ID of the keys to be read from accessor.
+ * @param check_for_null_keys If true, each key will be checked to see if it
+ * is null before looking it up (null keys are skipped). This must be
+ * set to true if some of the keys that will be read from accessor may
+ * be null.
+ * @param functor A pointer to a functor which should provide an operator that
+ * takes 1 argument: const ValueAccessor&. The operator will be called
+ * only once for a key from accessor if there is a match.
+ */
+ template <typename FunctorT>
+ void runOverKeysFromValueAccessorIfMatchFound(ValueAccessor *accessor,
+ const attribute_id key_attr_id,
+ const bool check_for_null_keys,
+ FunctorT *functor) const {
+ return runOverKeysFromValueAccessor<true>(accessor,
+ key_attr_id,
+ check_for_null_keys,
+ functor);
+ }
+
+ /**
+ * @brief Apply the functor to each key with a match in the hash table.
+ *
+ * @param accessor A ValueAccessor which will be used to access keys.
+ * beginIteration() should be called on accessor before calling this
+ * method.
+ * @param key_attr_id The attribute ID of the keys to be read from accessor.
+ * @param check_for_null_keys If true, each key will be checked to see if it
+ * is null before looking it up (null keys are skipped). This must be
+ * set to true if some of the keys that will be read from accessor may
+ * be null.
+ * @param functor A pointer to a functor which should provide an operator that
+ * takes 1 argument: const ValueAccessor&. The operator will be called
+ * only once for a key from accessor if there is a match.
+ */
+ template <typename FunctorT>
+ void runOverKeysFromValueAccessorIfMatchFoundCompositeKey(
+ ValueAccessor *accessor,
+ const std::vector<attribute_id> &key_attr_ids,
+ const bool check_for_null_keys,
+ FunctorT *functor) const {
+ return runOverKeysFromValueAccessorCompositeKey<true>(accessor,
+ key_attr_ids,
+ check_for_null_keys,
+ functor);
+ }
+
+ /**
+ * @brief Apply the functor to each key without a match in the hash table.
+ *
+ * @param accessor A ValueAccessor which will be used to access keys.
+ * beginIteration() should be called on accessor before calling this
+ * method.
+ * @param key_attr_id The attribute ID of the keys to be read from accessor.
+ * @param check_for_null_keys If true, each key will be checked to see if it
+ * is null before looking it up (null keys are skipped). This must be
+ * set to true if some of the keys that will be read from accessor may
+ * be null.
+ * @param functor A pointer to a functor which should provide an operator that
+ * takes 1 argument: const ValueAccessor&. The operator will be called
+ * only once for a key from accessor if there is no match.
+ */
+ template <typename FunctorT>
+ void runOverKeysFromValueAccessorIfMatchNotFound(
+ ValueAccessor *accessor,
+ const attribute_id key_attr_id,
+ const bool check_for_null_keys,
+ FunctorT *functor) const {
+ return runOverKeysFromValueAccessor<false>(accessor,
+ key_attr_id,
+ check_for_null_keys,
+ functor);
+ }
+
+ /**
+ * @brief Apply the functor to each key without a match in the hash table.
+ *
+ * @param accessor A ValueAccessor which will be used to access keys.
+ * beginIteration() should be called on accessor before calling this
+ * method.
+ * @param key_attr_id The attribute ID of the keys to be read from accessor.
+ * @param check_for_null_keys If true, each key will be checked to see if it
+ * is null before looking it up (null keys are skipped). This must be
+ * set to true if some of the keys that will be read from accessor may
+ * be null.
+ * @param functor A pointer to a functor which should provide an operator that
+ * takes 1 argument: const ValueAccessor&. The operator will be called
+ * only once for a key from accessor if there is no match.
+ */
+ template <typename FunctorT>
+ void runOverKeysFromValueAccessorIfMatchNotFoundCompositeKey(
+ ValueAccessor *accessor,
+ const std::vector<attribute_id> &key_attr_ids,
+ const bool check_for_null_keys,
+ FunctorT *functor) const {
+ return runOverKeysFromValueAccessorCompositeKey<false>(accessor,
+ key_attr_ids,
+ check_for_null_keys,
+ functor);
+ }
+
+ /**
* @brief Apply a functor to each key, value pair in this hash table.
*
* @warning This method assumes that no concurrent calls to put(),
@@ -965,6 +1155,10 @@
const ValueT **value,
std::size_t *entry_num) const = 0;
+ // Return true if key exists in the hash table.
+ virtual bool hasKey(const TypedValue &key) const = 0;
+ virtual bool hasCompositeKey(const std::vector<TypedValue> &key) const = 0;
+
// For a resizable HashTable, grow to accomodate more entries. If
// 'extra_buckets' is not zero, it may serve as a "hint" to implementations
// that at least the requested number of extra buckets are required when
@@ -1048,6 +1242,21 @@
return false;
}
+ // If run_if_match_found is true, apply the functor to each key if a match is
+ // found; otherwise, apply the functor if no match is found.
+ template <bool run_if_match_found, typename FunctorT>
+ void runOverKeysFromValueAccessor(ValueAccessor *accessor,
+ const attribute_id key_attr_id,
+ const bool check_for_null_keys,
+ FunctorT *functor) const;
+
+ template <bool run_if_match_found, typename FunctorT>
+ void runOverKeysFromValueAccessorCompositeKey(
+ ValueAccessor *accessor,
+ const std::vector<attribute_id> &key_attr_ids,
+ const bool check_for_null_keys,
+ FunctorT *functor) const;
+
// Method containing the actual logic implementing getAllFromValueAccessor().
// Has extra template parameters that control behavior to avoid some
// inner-loop branching.
@@ -1678,6 +1887,184 @@
bool force_key_copy,
bool allow_duplicate_keys>
template <typename FunctorT>
+void HashTable<ValueT,
+ resizable,
+ serializable,
+ force_key_copy,
+ allow_duplicate_keys>::
+ getAllFromValueAccessorWithExtraWorkForFirstMatch(
+ ValueAccessor *accessor,
+ const attribute_id key_attr_id,
+ const bool check_for_null_keys,
+ FunctorT *functor) const {
+ InvokeOnAnyValueAccessor(
+ accessor,
+ [&](auto *accessor) -> void { // NOLINT(build/c++11)
+ while (accessor->next()) {
+ TypedValue key = accessor->getTypedValue(key_attr_id);
+ if (check_for_null_keys && key.isNull()) {
+ continue;
+ }
+ const std::size_t hash_code = adjust_hashes_ ? AdjustHash(key.getHash())
+ : key.getHash();
+ std::size_t entry_num = 0;
+ const ValueT *value;
+ if (getNextEntryForKey(key, hash_code, &value, &entry_num)) {
+ functor->recordMatch(*accessor);
+ (*functor)(*accessor, *value);
+ if (!allow_duplicate_keys) {
+ continue;
+ }
+ while (getNextEntryForKey(key, hash_code, &value, &entry_num)) {
+ (*functor)(*accessor, *value);
+ }
+ }
+ }
+ }); // NOLINT(whitespace/parens)
+}
+
+template <typename ValueT,
+ bool resizable,
+ bool serializable,
+ bool force_key_copy,
+ bool allow_duplicate_keys>
+template <typename FunctorT>
+void HashTable<ValueT, resizable, serializable, force_key_copy, allow_duplicate_keys>
+ ::getAllFromValueAccessorCompositeKeyWithExtraWorkForFirstMatch(
+ ValueAccessor *accessor,
+ const std::vector<attribute_id> &key_attr_ids,
+ const bool check_for_null_keys,
+ FunctorT *functor) const {
+ DEBUG_ASSERT(key_types_.size() == key_attr_ids.size());
+ std::vector<TypedValue> key_vector;
+ key_vector.resize(key_attr_ids.size());
+ InvokeOnAnyValueAccessor(
+ accessor,
+ [&](auto *accessor) -> void { // NOLINT(build/c++11)
+ while (accessor->next()) {
+ bool null_key = false;
+ for (std::vector<attribute_id>::size_type key_idx = 0;
+ key_idx < key_types_.size();
+ ++key_idx) {
+ key_vector[key_idx] = accessor->getTypedValue(key_attr_ids[key_idx]);
+ if (check_for_null_keys && key_vector[key_idx].isNull()) {
+ null_key = true;
+ break;
+ }
+ }
+ if (null_key) {
+ continue;
+ }
+
+ const std::size_t hash_code = adjust_hashes_ ? AdjustHash(hashCompositeKey(key_vector))
+ : hashCompositeKey(key_vector);
+ std::size_t entry_num = 0;
+ const ValueT *value;
+ if (getNextEntryForCompositeKey(key_vector, hash_code, &value, &entry_num)) {
+ functor->recordMatch(*accessor);
+ (*functor)(*accessor, *value);
+ if (!allow_duplicate_keys) {
+ continue;
+ }
+ while (getNextEntryForCompositeKey(key_vector, hash_code, &value, &entry_num)) {
+ (*functor)(*accessor, *value);
+ }
+ }
+ }
+ }); // NOLINT(whitespace/parens)
+}
+
+template <typename ValueT,
+ bool resizable,
+ bool serializable,
+ bool force_key_copy,
+ bool allow_duplicate_keys>
+template <bool run_if_match_found, typename FunctorT>
+void HashTable<ValueT,
+ resizable,
+ serializable,
+ force_key_copy,
+ allow_duplicate_keys>::
+ runOverKeysFromValueAccessor(ValueAccessor *accessor,
+ const attribute_id key_attr_id,
+ const bool check_for_null_keys,
+ FunctorT *functor) const {
+ InvokeOnAnyValueAccessor(
+ accessor,
+ [&](auto *accessor) -> void { // NOLINT(build/c++11)
+ while (accessor->next()) {
+ TypedValue key = accessor->getTypedValue(key_attr_id);
+ if (check_for_null_keys && key.isNull()) {
+ if (!run_if_match_found) {
+ (*functor)(*accessor);
+ continue;
+ }
+ }
+ if (run_if_match_found) {
+ if (hasKey(key)) {
+ (*functor)(*accessor);
+ }
+ } else {
+ if (!hasKey(key)) {
+ (*functor)(*accessor);
+ }
+ }
+ }
+ }); // NOLINT(whitespace/parens)
+}
+
+template <typename ValueT,
+ bool resizable,
+ bool serializable,
+ bool force_key_copy,
+ bool allow_duplicate_keys>
+template <bool run_if_match_found, typename FunctorT>
+void HashTable<ValueT, resizable, serializable, force_key_copy, allow_duplicate_keys>
+ ::runOverKeysFromValueAccessorCompositeKey(ValueAccessor *accessor,
+ const std::vector<attribute_id> &key_attr_ids,
+ const bool check_for_null_keys,
+ FunctorT *functor) const {
+ DEBUG_ASSERT(key_types_.size() == key_attr_ids.size());
+ std::vector<TypedValue> key_vector;
+ key_vector.resize(key_attr_ids.size());
+ InvokeOnAnyValueAccessor(
+ accessor,
+ [&](auto *accessor) -> void { // NOLINT(build/c++11)
+ while (accessor->next()) {
+ bool null_key = false;
+ for (std::vector<attribute_id>::size_type key_idx = 0;
+ key_idx < key_types_.size();
+ ++key_idx) {
+ key_vector[key_idx] = accessor->getTypedValue(key_attr_ids[key_idx]);
+ if (check_for_null_keys && key_vector[key_idx].isNull()) {
+ null_key = true;
+ break;
+ }
+ }
+ if (null_key) {
+ if (!run_if_match_found) {
+ (*functor)(*accessor);
+ continue;
+ }
+ }
+
+ if (run_if_match_found) {
+ if (hasCompositeKey(key_vector)) {
+ (*functor)(*accessor);
+ }
+ } else if (!hasCompositeKey(key_vector)) {
+ (*functor)(*accessor);
+ }
+ }
+ }); // NOLINT(whitespace/parens)
+}
+
+template <typename ValueT,
+ bool resizable,
+ bool serializable,
+ bool force_key_copy,
+ bool allow_duplicate_keys>
+template <typename FunctorT>
std::size_t HashTable<ValueT, resizable, serializable, force_key_copy, allow_duplicate_keys>
::forEach(FunctorT *functor) const {
std::size_t entries_visited = 0;
diff --git a/storage/LinearOpenAddressingHashTable.hpp b/storage/LinearOpenAddressingHashTable.hpp
index e5ca0b0..a2ebe18 100644
--- a/storage/LinearOpenAddressingHashTable.hpp
+++ b/storage/LinearOpenAddressingHashTable.hpp
@@ -151,6 +151,9 @@
const ValueT **value,
std::size_t *entry_num) const override;
+ bool hasKey(const TypedValue &key) const override;
+ bool hasCompositeKey(const std::vector<TypedValue> &key) const override;
+
void resize(const std::size_t extra_buckets,
const std::size_t extra_variable_storage,
const std::size_t retry_num = 0) override;
@@ -1099,6 +1102,75 @@
return false;
}
+template <typename ValueT,
+ bool resizable,
+ bool serializable,
+ bool force_key_copy,
+ bool allow_duplicate_keys>
+bool LinearOpenAddressingHashTable<ValueT, resizable, serializable, force_key_copy, allow_duplicate_keys>
+ ::hasKey(const TypedValue &key) const {
+ DEBUG_ASSERT(this->key_types_.size() == 1);
+ DEBUG_ASSERT(key.isPlausibleInstanceOf(this->key_types_.front()->getSignature()));
+
+ const std::size_t hash_code = this->AdjustHash(key.getHash());
+ for (std::size_t bucket_num = hash_code % header_->num_buckets;
+ bucket_num < header_->num_buckets + header_->num_overflow_buckets;
+ ++bucket_num) {
+ const char *bucket = static_cast<const char*>(hash_buckets_) + bucket_num * bucket_size_;
+ const std::size_t bucket_hash
+ = reinterpret_cast<const std::atomic<std::size_t>*>(bucket)->load(std::memory_order_relaxed);
+ if (bucket_hash == kEmptyHash) {
+ // Hit an empty bucket, so the search is finished
+ // without finding any match.
+ return false;
+ }
+
+ // None of the get methods should be called while inserts are still taking
+ // place.
+ DEBUG_ASSERT(bucket_hash != kPendingHash);
+
+ if ((bucket_hash == hash_code) && key_manager_.scalarKeyCollisionCheck(key, bucket)) {
+ // Match located.
+ return true;
+ }
+ }
+ return false;
+}
+
+template <typename ValueT,
+ bool resizable,
+ bool serializable,
+ bool force_key_copy,
+ bool allow_duplicate_keys>
+bool LinearOpenAddressingHashTable<ValueT, resizable, serializable, force_key_copy, allow_duplicate_keys>
+ ::hasCompositeKey(const std::vector<TypedValue> &key) const {
+ DEBUG_ASSERT(this->key_types_.size() == key.size());
+
+ const std::size_t hash_code = this->AdjustHash(this->hashCompositeKey(key));
+ for (std::size_t bucket_num = hash_code % header_->num_buckets;
+ bucket_num < header_->num_buckets + header_->num_overflow_buckets;
+ ++bucket_num) {
+ const char *bucket = static_cast<const char*>(hash_buckets_) + bucket_num * bucket_size_;
+ const std::size_t bucket_hash
+ = reinterpret_cast<const std::atomic<std::size_t>*>(bucket)->load(std::memory_order_relaxed);
+ if (bucket_hash == kEmptyHash) {
+ // Hit an empty bucket, so the search is finished
+ // without finding any match.
+ return false;
+ }
+
+ // None of the get methods should be called while inserts are still taking
+ // place.
+ DEBUG_ASSERT(bucket_hash != kPendingHash);
+
+ if ((bucket_hash == hash_code) && key_manager_.compositeKeyCollisionCheck(key, bucket)) {
+ // Match located.
+ return true;
+ }
+ }
+ return false;
+}
+
// TODO(chasseur): Smarter heuristics that are more selective about whether
// to grow hash buckets, variable-length storage, or both, and to what degree.
template <typename ValueT,
diff --git a/storage/SeparateChainingHashTable.hpp b/storage/SeparateChainingHashTable.hpp
index c93e783..c096b1b 100644
--- a/storage/SeparateChainingHashTable.hpp
+++ b/storage/SeparateChainingHashTable.hpp
@@ -145,6 +145,9 @@
const ValueT **value,
std::size_t *entry_num) const override;
+ bool hasKey(const TypedValue &key) const override;
+ bool hasCompositeKey(const std::vector<TypedValue> &key) const override;
+
void resize(const std::size_t extra_buckets,
const std::size_t extra_variable_storage,
const std::size_t retry_num = 0) override;
@@ -1054,6 +1057,57 @@
bool serializable,
bool force_key_copy,
bool allow_duplicate_keys>
+bool SeparateChainingHashTable<ValueT, resizable, serializable, force_key_copy, allow_duplicate_keys>
+ ::hasKey(const TypedValue &key) const {
+ DEBUG_ASSERT(this->key_types_.size() == 1);
+ DEBUG_ASSERT(key.isPlausibleInstanceOf(this->key_types_.front()->getSignature()));
+
+ const std::size_t hash_code = key.getHash();
+ std::size_t bucket_ref = slots_[hash_code % header_->num_slots].load(std::memory_order_relaxed);
+ while (bucket_ref != 0) {
+ DEBUG_ASSERT(bucket_ref != std::numeric_limits<std::size_t>::max());
+ const char *bucket = static_cast<const char*>(buckets_) + (bucket_ref - 1) * bucket_size_;
+ const std::size_t bucket_hash = *reinterpret_cast<const std::size_t*>(
+ bucket + sizeof(std::atomic<std::size_t>));
+ if ((bucket_hash == hash_code) && key_manager_.scalarKeyCollisionCheck(key, bucket)) {
+ // Find a match.
+ return true;
+ }
+ bucket_ref = reinterpret_cast<const std::atomic<std::size_t>*>(bucket)->load(std::memory_order_relaxed);
+ }
+ return false;
+}
+
+template <typename ValueT,
+ bool resizable,
+ bool serializable,
+ bool force_key_copy,
+ bool allow_duplicate_keys>
+bool SeparateChainingHashTable<ValueT, resizable, serializable, force_key_copy, allow_duplicate_keys>
+ ::hasCompositeKey(const std::vector<TypedValue> &key) const {
+ DEBUG_ASSERT(this->key_types_.size() == key.size());
+
+ const std::size_t hash_code = this->hashCompositeKey(key);
+ std::size_t bucket_ref = slots_[hash_code % header_->num_slots].load(std::memory_order_relaxed);
+ while (bucket_ref != 0) {
+ DEBUG_ASSERT(bucket_ref != std::numeric_limits<std::size_t>::max());
+ const char *bucket = static_cast<const char*>(buckets_) + (bucket_ref - 1) * bucket_size_;
+ const std::size_t bucket_hash = *reinterpret_cast<const std::size_t*>(
+ bucket + sizeof(std::atomic<std::size_t>));
+ if ((bucket_hash == hash_code) && key_manager_.compositeKeyCollisionCheck(key, bucket)) {
+ // Find a match.
+ return true;
+ }
+ bucket_ref = reinterpret_cast<const std::atomic<std::size_t>*>(bucket)->load(std::memory_order_relaxed);
+ }
+ return false;
+}
+
+template <typename ValueT,
+ bool resizable,
+ bool serializable,
+ bool force_key_copy,
+ bool allow_duplicate_keys>
void SeparateChainingHashTable<ValueT, resizable, serializable, force_key_copy, allow_duplicate_keys>
::resize(const std::size_t extra_buckets,
const std::size_t extra_variable_storage,
diff --git a/storage/SimpleScalarSeparateChainingHashTable.hpp b/storage/SimpleScalarSeparateChainingHashTable.hpp
index b2d894d..1619e34 100644
--- a/storage/SimpleScalarSeparateChainingHashTable.hpp
+++ b/storage/SimpleScalarSeparateChainingHashTable.hpp
@@ -182,6 +182,12 @@
return getNextEntryForKey(key.front(), hash_code, value, entry_num);
}
+ bool hasKey(const TypedValue &key) const override;
+
+ bool hasCompositeKey(const std::vector<TypedValue> &key) const override {
+ return false;
+ }
+
void resize(const std::size_t extra_buckets,
const std::size_t extra_variable_storage,
const std::size_t retry_num = 0) override;
@@ -772,6 +778,37 @@
bool serializable,
bool force_key_copy,
bool allow_duplicate_keys>
+bool SimpleScalarSeparateChainingHashTable<ValueT,
+ resizable,
+ serializable,
+ force_key_copy,
+ allow_duplicate_keys>
+ ::hasKey(const TypedValue &key) const {
+ DCHECK(!allow_duplicate_keys);
+ DCHECK(key.isPlausibleInstanceOf(this->key_types_.front()->getSignature()));
+
+ const std::size_t hash_code = key.getHashScalarLiteral();
+ std::size_t bucket_ref = slots_[hash_code % header_->num_slots].load(std::memory_order_relaxed);
+ while (bucket_ref != 0) {
+ DCHECK_NE(bucket_ref, std::numeric_limits<std::size_t>::max());
+
+ const Bucket &bucket = buckets_[bucket_ref - 1];
+ if (bucket.hash == hash_code) {
+ // Match located.
+ return true;
+ }
+ bucket_ref = bucket.next.load(std::memory_order_relaxed);
+ }
+
+ // Reached the end of the chain and didn't find a match.
+ return false;
+}
+
+template <typename ValueT,
+ bool resizable,
+ bool serializable,
+ bool force_key_copy,
+ bool allow_duplicate_keys>
void SimpleScalarSeparateChainingHashTable<ValueT,
resizable,
serializable,
diff --git a/storage/StorageBlock.hpp b/storage/StorageBlock.hpp
index da3bc70..97813e2 100644
--- a/storage/StorageBlock.hpp
+++ b/storage/StorageBlock.hpp
@@ -187,6 +187,17 @@
}
/**
+ * @brief Get the flag vector indicating for each IndexSubBlock
+ * whether it is consistent.
+ *
+ * @return The flag vector indicating for each IndexSubBlock
+ * whether it is consistent.
+ */
+ const std::vector<bool>& getIndicesConsistent() const {
+ return indices_consistent_;
+ }
+
+ /**
* @brief Get one of this block's IndexSubBlocks.
*
* @param index_id The ID of the IndexSubBlock. This is simply a serial
@@ -201,6 +212,15 @@
}
/**
+ * @brief Get the IndexSubBlock vector.
+ *
+ * @return The IndexSubBlock vector.
+ */
+ const PtrVector<IndexSubBlock>& getIndices() const {
+ return indices_;
+ }
+
+ /**
* @brief Insert a single tuple into this block.
*
* @param tuple The tuple to insert.
