Refactored PartitionAwareInsertDestination::getPartitionId.
diff --git a/storage/InsertDestination.cpp b/storage/InsertDestination.cpp
index 3caa80f..ff39c55 100644
--- a/storage/InsertDestination.cpp
+++ b/storage/InsertDestination.cpp
@@ -619,19 +619,17 @@
[this,
&always_mark_full,
&num_partitions](auto *accessor) -> void { // NOLINT(build/c++11)
- std::vector<std::unique_ptr<TupleIdSequence>> partition_membership;
+ std::vector<std::unique_ptr<TupleIdSequence>> partition_membership(num_partitions);
// Create a tuple-id sequence for each partition.
for (std::size_t partition = 0; partition < num_partitions; ++partition) {
- partition_membership.emplace_back(std::make_unique<TupleIdSequence>(accessor->getEndPosition()));
+ partition_membership[partition] = std::make_unique<TupleIdSequence>(accessor->getEndPosition());
}
// Iterate over ValueAccessor for each tuple,
// set a bit in the appropriate TupleIdSequence.
accessor->beginIteration();
- while (accessor->next()) {
- partition_membership[this->getPartitionId(accessor)]->set(accessor->getCurrentPosition(), true);
- }
+ this->setPartitionMembership(&partition_membership, accessor);
// For each partition, create an adapter around Value Accessor and
// TupleIdSequence.
@@ -670,19 +668,17 @@
&attribute_map,
&always_mark_full,
&num_partitions](auto *accessor) -> void { // NOLINT(build/c++11)
- std::vector<std::unique_ptr<TupleIdSequence>> partition_membership;
+ std::vector<std::unique_ptr<TupleIdSequence>> partition_membership(num_partitions);
// Create a tuple-id sequence for each partition.
for (std::size_t partition = 0; partition < num_partitions; ++partition) {
- partition_membership.emplace_back(std::make_unique<TupleIdSequence>(accessor->getEndPosition()));
+ partition_membership[partition] = std::make_unique<TupleIdSequence>(accessor->getEndPosition());
}
// Iterate over ValueAccessor for each tuple,
// set a bit in the appropriate TupleIdSequence.
accessor->beginIteration();
- while (accessor->next()) {
- partition_membership[this->getPartitionId(accessor)]->set(accessor->getCurrentPosition(), true);
- }
+ this->setPartitionMembership(&partition_membership, accessor);
// For each partition, create an adapter around Value Accessor and
// TupleIdSequence.
diff --git a/storage/InsertDestination.hpp b/storage/InsertDestination.hpp
index 878ebb4..c3d5641 100644
--- a/storage/InsertDestination.hpp
+++ b/storage/InsertDestination.hpp
@@ -36,6 +36,7 @@
#include "storage/StorageBlock.hpp"
#include "storage/StorageBlockInfo.hpp"
#include "storage/StorageBlockLayout.hpp"
+#include "storage/TupleIdSequence.hpp"
#include "storage/ValueAccessor.hpp"
#include "threading/SpinMutex.hpp"
#include "threading/ThreadIDBasedMap.hpp"
@@ -612,21 +613,35 @@
}
partition_id getPartitionId(const Tuple &tuple) const {
- PartitionSchemeHeader::PartitionValues values;
- for (const attribute_id attr_id : partition_scheme_header_->getPartitionAttributeIds()) {
- values.push_back(tuple.getAttributeValue(attr_id));
+ const auto &partition_attr_ids = partition_scheme_header_->getPartitionAttributeIds();
+
+ PartitionSchemeHeader::PartitionValues values(partition_attr_ids.size());
+ for (std::size_t i = 0; i < partition_attr_ids.size(); ++i) {
+ values[i] = tuple.getAttributeValue(partition_attr_ids[i]);
}
return values.empty() ? input_partition_id_ : partition_scheme_header_->getPartitionId(values);
}
- partition_id getPartitionId(ValueAccessor *accessor) const {
- PartitionSchemeHeader::PartitionValues values;
- for (const attribute_id attr_id : partition_scheme_header_->getPartitionAttributeIds()) {
- values.push_back(accessor->getTypedValueVirtual(attr_id));
- }
+ template<typename ValueAccessorT>
+ void setPartitionMembership(std::vector<std::unique_ptr<TupleIdSequence>> *partition_membership,
+ ValueAccessorT *accessor) const {
+ const auto &partition_attr_ids = partition_scheme_header_->getPartitionAttributeIds();
- return values.empty() ? input_partition_id_ : partition_scheme_header_->getPartitionId(values);
+ if (partition_attr_ids.empty()) {
+ while (accessor->next()) {
+ (*partition_membership)[input_partition_id_]->set(accessor->getCurrentPosition(), true);
+ }
+ } else {
+ PartitionSchemeHeader::PartitionValues values(partition_attr_ids.size());
+ while (accessor->next()) {
+ for (std::size_t i = 0; i < partition_attr_ids.size(); ++i) {
+ values[i] = accessor->getTypedValue(partition_attr_ids[i]);
+ }
+ (*partition_membership)[partition_scheme_header_->getPartitionId(values)]->set(
+ accessor->getCurrentPosition(), true);
+ }
+ }
}
std::unique_ptr<const PartitionSchemeHeader> partition_scheme_header_;
@@ -646,6 +661,7 @@
DISALLOW_COPY_AND_ASSIGN(PartitionAwareInsertDestination);
};
+
/** @} */
} // namespace quickstep
