| /** |
| * 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 <algorithm> |
| #include <cstddef> |
| #include <cstdint> |
| #include <iostream> |
| #include <limits> |
| #include <memory> |
| #include <random> |
| #include <sstream> |
| #include <string> |
| #include <utility> |
| #include <vector> |
| |
| #include "catalog/CatalogAttribute.hpp" |
| #include "catalog/CatalogDatabase.hpp" |
| #include "catalog/CatalogRelation.hpp" |
| #include "catalog/CatalogTypedefs.hpp" |
| #include "expressions/Expressions.pb.h" |
| #include "expressions/scalar/Scalar.hpp" |
| #include "expressions/scalar/ScalarAttribute.hpp" |
| #include "query_execution/QueryContext.hpp" |
| #include "query_execution/QueryContext.pb.h" |
| #include "query_execution/QueryExecutionMessages.pb.h" |
| #include "query_execution/QueryExecutionTypedefs.hpp" |
| #include "query_execution/WorkOrdersContainer.hpp" |
| #include "relational_operators/RelationalOperator.hpp" |
| #include "relational_operators/SortMergeRunOperator.hpp" |
| #include "relational_operators/SortMergeRunOperatorHelpers.hpp" |
| #include "relational_operators/WorkOrder.hpp" |
| #include "storage/BasicColumnStoreValueAccessor.hpp" |
| #include "storage/CompressedColumnStoreValueAccessor.hpp" |
| #include "storage/CompressedPackedRowStoreValueAccessor.hpp" |
| #include "storage/CountedReference.hpp" |
| #include "storage/InsertDestination.hpp" |
| #include "storage/InsertDestination.pb.h" |
| #include "storage/SplitRowStoreValueAccessor.hpp" |
| #include "storage/StorageBlock.hpp" |
| #include "storage/StorageBlockInfo.hpp" |
| #include "storage/StorageBlockLayout.hpp" |
| #include "storage/StorageManager.hpp" |
| #include "storage/TupleStorageSubBlock.hpp" |
| #include "storage/ValueAccessor.hpp" |
| #include "storage/ValueAccessorUtil.hpp" |
| #include "threading/ThreadIDBasedMap.hpp" |
| #include "types/IntType.hpp" |
| #include "types/Type.hpp" |
| #include "types/TypeID.hpp" |
| #include "types/TypedValue.hpp" |
| #include "types/containers/ColumnVectorsValueAccessor.hpp" |
| #include "types/containers/Tuple.hpp" |
| #include "utility/PtrVector.hpp" |
| #include "utility/SortConfiguration.hpp" |
| #include "utility/SortConfiguration.pb.h" |
| |
| #include "gflags/gflags.h" |
| #include "glog/logging.h" |
| #include "gtest/gtest.h" |
| |
| #include "tmb/id_typedefs.h" |
| #include "tmb/tagged_message.h" |
| |
| using std::move; |
| using std::unique_ptr; |
| using std::vector; |
| |
| namespace quickstep { |
| |
| namespace { |
| |
| constexpr std::size_t kQueryId = 0; |
| constexpr const std::size_t kOpIndex = 0; |
| |
| // Helper struct for test tuple that will that will be inserted and sorted. |
| class TestTuple { |
| public: |
| static const int kCol1Bits; |
| static const int kCol2Bits; |
| static const int kCol3Bits; |
| |
| // Intialize tuple based on seed number. |
| explicit TestTuple(int num) |
| : tid_(num), |
| col1_(Bits(num, 3, kCol1Bits)), |
| col2_(Bits(num, 6, kCol2Bits)), |
| col3_(Bits(num, 1, kCol3Bits)) {} |
| |
| int tid_; // Tuple-ID |
| int col1_; // Col-1 |
| int col2_; // Col-2 |
| int col3_; // Col-3 |
| |
| static inline int Bits(int value, int offset, int length) { |
| return (value >> offset) & (0xffff >> (16 - length)); |
| } |
| }; |
| |
| const int TestTuple::kCol1Bits = 5; |
| const int TestTuple::kCol2Bits = 2; |
| const int TestTuple::kCol3Bits = 3; |
| |
| // To simplify testing nullable attributes, we create another meta tuple from |
| // column values from TestTuple struct. Column 1, 2, 3 are same as TestTuple. |
| // Columns 4, 5, 6 are same as 1, 2, 3 respectively when non-zero, and null |
| // otherwise. That is, i^th and (i + 3)^th columns are same when the value is |
| // non-zero. When the value of i^th column is zero, (i + 3)^th column is NULL. |
| // null_c{i} captures if column i was NULL in the tuple. |
| struct TestTupleAttrs { |
| int c1, c2, c3; |
| bool null_c4, null_c5, null_c6; |
| }; |
| |
| // Convert from TestTuple to TestTupleAttrs. |
| struct TestTupleAttrs TupleToTupleAttr(const Tuple &tuple) { |
| auto attr = tuple.getAttributeValueVector(); |
| struct TestTupleAttrs out {}; |
| out.c1 = attr[0].getLiteral<int>(); |
| out.c2 = attr[1].getLiteral<int>(); |
| out.c3 = attr[2].getLiteral<int>(); |
| out.null_c4 = attr[3].isNull(); |
| out.null_c5 = attr[4].isNull(); |
| out.null_c6 = attr[5].isNull(); |
| return out; |
| } |
| |
| } // namespace |
| |
| namespace merge_run_operator { |
| |
| // Test Run class. |
| class RunTest : public ::testing::Test { |
| public: |
| static const tuple_id kNumTuples; |
| static const tuple_id kNumTuplesPerBlock; |
| |
| protected: |
| static const relation_id kTableId = 100; |
| static const relation_id kResultTableId = kTableId + 1; |
| static const char kTableName[]; |
| static const char kStoragePath[]; |
| |
| virtual void SetUp() { |
| // Initialize the TMB, register this thread as sender and receiver for |
| // appropriate types of messages. |
| bus_.Initialize(); |
| |
| foreman_client_id_ = bus_.Connect(); |
| bus_.RegisterClientAsReceiver(foreman_client_id_, kCatalogRelationNewBlockMessage); |
| bus_.RegisterClientAsReceiver(foreman_client_id_, kDataPipelineMessage); |
| |
| const tmb::client_id worker_thread_client_id = bus_.Connect(); |
| bus_.RegisterClientAsSender(worker_thread_client_id, kCatalogRelationNewBlockMessage); |
| bus_.RegisterClientAsSender(worker_thread_client_id, kDataPipelineMessage); |
| |
| thread_id_map_ = ClientIDMap::Instance(); |
| // Usually the worker thread makes the following call. In this test setup, |
| // we don't have a worker thread hence we have to explicitly make the call. |
| thread_id_map_->addValue(worker_thread_client_id); |
| |
| storage_manager_.reset(new StorageManager(kStoragePath)); |
| table_.reset(new CatalogRelation(nullptr, kTableName, kTableId)); |
| const Type &int_type = IntType::InstanceNonNullable(); |
| table_->addAttribute(new CatalogAttribute(table_.get(), "col-1", int_type)); |
| table_->addAttribute(new CatalogAttribute(table_.get(), "col-2", int_type)); |
| table_->addAttribute(new CatalogAttribute(table_.get(), "col-3", int_type)); |
| col1_ = table_->getAttributeByName("col-1")->getID(); |
| col2_ = table_->getAttributeByName("col-2")->getID(); |
| col3_ = table_->getAttributeByName("col-3")->getID(); |
| table_->setDefaultStorageBlockLayout(StorageBlockLayout::GenerateDefaultLayout(*table_, false)); |
| |
| insert_destination_.reset( |
| new BlockPoolInsertDestination(*table_, |
| nullptr, |
| storage_manager_.get(), |
| kOpIndex, |
| 0, // dummy query ID. |
| foreman_client_id_, |
| &bus_)); |
| } |
| |
| virtual void TearDown() { |
| // Usually the worker thread makes the following call. In this test setup, |
| // we don't have a worker thread hence we have to explicitly make the call. |
| thread_id_map_->removeValue(); |
| |
| // Drop blocks from relations and InsertDestination. |
| const vector<block_id> tmp_blocks = insert_destination_->getTouchedBlocks(); |
| for (const block_id block : tmp_blocks) { |
| storage_manager_->deleteBlockOrBlobFile(block); |
| } |
| |
| const vector<block_id> blocks = table_->getBlocksSnapshot(); |
| for (const block_id block : blocks) { |
| storage_manager_->deleteBlockOrBlobFile(block); |
| } |
| } |
| |
| // Helper method to insert test tuples. |
| Tuple *createTuple(tuple_id id) { |
| std::vector<TypedValue> values; |
| values.emplace_back(static_cast<int>(id)); |
| values.emplace_back(static_cast<int>(id)); |
| values.emplace_back(static_cast<int>(id)); |
| return new Tuple(std::move(values)); |
| } |
| |
| std::unique_ptr<StorageManager> storage_manager_; |
| std::unique_ptr<CatalogRelation> table_; |
| std::unique_ptr<InsertDestination> insert_destination_; |
| attribute_id col1_; |
| attribute_id col2_; |
| attribute_id col3_; |
| merge_run_operator::Run run_; |
| |
| MessageBusImpl bus_; |
| tmb::client_id foreman_client_id_; |
| |
| // This map is needed for InsertDestination and some operators that send |
| // messages to Foreman directly. To know the reason behind the design of this |
| // map, see the note in InsertDestination.hpp. |
| ClientIDMap *thread_id_map_; |
| }; |
| |
| const char RunTest::kTableName[] = "table"; |
| const char RunTest::kStoragePath[] = "./sort_merge_run_operator_test_data"; |
| const tuple_id RunTest::kNumTuples = 100; |
| const tuple_id RunTest::kNumTuplesPerBlock = 10; |
| |
| // Test creating an empty run. |
| TEST_F(RunTest, RunCreatorEmptyRun) { |
| { |
| RunCreator run_creator(&run_, insert_destination_.get()); |
| // Need run_creator to destruct to finalize the output run. |
| } |
| EXPECT_EQ(0u, run_.size()); |
| } |
| |
| // Test creating a run with few tuples. |
| TEST_F(RunTest, RunCreatorFewTuples) { |
| tuple_id max_tuples = 10; |
| |
| { |
| RunCreator run_creator(&run_, insert_destination_.get()); |
| std::unique_ptr<Tuple> tuple; |
| for (tuple_id i = 0; i < max_tuples; ++i) { |
| tuple.reset(createTuple(i)); |
| run_creator.appendTuple(*tuple); |
| } |
| EXPECT_EQ(1u, run_.size()); |
| // Need run_creator to destruct to finalize the output run. |
| } |
| |
| |
| BlockReference block(storage_manager_->getBlock(run_.back(), *table_)); |
| std::unique_ptr<ValueAccessor> accessor( |
| block->getTupleStorageSubBlock().createValueAccessor()); |
| |
| tuple_id num_tuples = 0; |
| InvokeOnValueAccessorNotAdapter( |
| accessor.get(), |
| [&](auto *accessor) -> void { // NOLINT(build/c++11) |
| while (accessor->next()) { |
| const tuple_id current_num_tuples = num_tuples; |
| EXPECT_EQ(static_cast<int>(current_num_tuples), |
| accessor->getTypedValue(col1_).template getLiteral<int>()); |
| ++num_tuples; |
| } |
| }); |
| EXPECT_EQ(max_tuples, num_tuples); |
| } |
| |
| // Test creating a run with multiple blocks. |
| TEST_F(RunTest, RunCreatorMultipleBlocks) { |
| std::size_t max_blocks = 3; |
| tuple_id max_tuples = 0; |
| |
| { |
| RunCreator run_creator(&run_, insert_destination_.get()); |
| std::unique_ptr<Tuple> tuple; |
| std::size_t blocks = 0; |
| while (blocks < max_blocks) { |
| tuple.reset(createTuple(max_tuples)); |
| if (run_creator.appendTuple(*tuple)) { |
| ++blocks; |
| } |
| ++max_tuples; |
| } |
| EXPECT_EQ(max_blocks + 1, run_.size()); |
| // Need run_creator to destruct to finalize the output run. |
| } |
| |
| tuple_id num_tuples = 0; |
| for (block_id bid : run_) { |
| BlockReference block(storage_manager_->getBlock(bid, *table_)); |
| std::unique_ptr<ValueAccessor> accessor( |
| block->getTupleStorageSubBlock().createValueAccessor()); |
| |
| InvokeOnValueAccessorNotAdapter( |
| accessor.get(), |
| [&](auto *accessor) -> void { // NOLINT(build/c++11) |
| while (accessor->next()) { |
| const tuple_id current_num_tuples = num_tuples; |
| EXPECT_EQ(static_cast<int>(current_num_tuples), |
| accessor->getTypedValue(col1_).template getLiteral<int>()); |
| ++num_tuples; |
| } |
| }); // NOLINT(whitespace/parens) |
| } |
| EXPECT_EQ(max_tuples, num_tuples); |
| } |
| |
| // Test run iterator to see if tuples are accessed in the same order as stored. |
| TEST_F(RunTest, IterateTuples) { |
| std::unique_ptr<Tuple> tuple; |
| MutableBlockReference storage_block; |
| for (tuple_id i = 0; i < kNumTuples; i += kNumTuplesPerBlock) { |
| // Create block |
| block_id block_id = storage_manager_->createBlock(*table_, table_->getDefaultStorageBlockLayout()); |
| storage_block = storage_manager_->getBlockMutable(block_id, *table_); |
| table_->addBlock(block_id); |
| run_.push_back(block_id); |
| |
| // Insert tuples |
| tuple_id block_bound = i + kNumTuplesPerBlock < kNumTuples |
| ? i + kNumTuplesPerBlock |
| : kNumTuples; |
| for (tuple_id tid = i; tid < block_bound; ++tid) { |
| tuple.reset(createTuple(tid)); |
| EXPECT_TRUE(storage_block->insertTupleInBatch(*tuple)); |
| } |
| storage_block->rebuild(); |
| |
| // Create empty block |
| block_id = storage_manager_->createBlock(*table_, table_->getDefaultStorageBlockLayout()); |
| storage_block = storage_manager_->getBlockMutable(block_id, *table_); |
| table_->addBlock(block_id); |
| run_.push_back(block_id); |
| storage_block->rebuild(); |
| } |
| |
| BlockReference block(storage_manager_->getBlock(run_.front(), *table_)); |
| std::unique_ptr<ValueAccessor> dummy_accessor( |
| block->getTupleStorageSubBlock().createValueAccessor()); |
| tuple_id num_tuples = 0; |
| InvokeOnValueAccessorNotAdapter( |
| dummy_accessor.get(), |
| [&](auto *dummy_accessor) -> void { // NOLINT(build/c++11) |
| RunIterator< |
| typename std::remove_reference<decltype(*dummy_accessor)>::type> |
| run_iterator(run_, storage_manager_.get(), *table_); |
| while (run_iterator.next()) { |
| const tuple_id current_num_tuples = num_tuples; |
| EXPECT_EQ(static_cast<int>(current_num_tuples), |
| run_iterator.getValueAccessor() |
| ->getTypedValue(col1_).template getLiteral<int>()); |
| ++num_tuples; |
| } |
| }); |
| EXPECT_EQ(kNumTuples, num_tuples); |
| } |
| |
| // Test RunMerger class. |
| class RunMergerTest : public ::testing::Test { |
| public: |
| static const std::size_t kNumTuplesPerBlock; |
| static const tuple_id kNumBlocksPerRun; |
| static const std::size_t kNumRuns; |
| static const tuple_id kNumTuples; |
| static const std::size_t kTopK; |
| |
| protected: |
| static const relation_id kTableId = 100; |
| static const relation_id kResultTableId = kTableId + 1; |
| static const char kTableName[]; |
| static const char kStoragePath[]; |
| |
| virtual void SetUp() { |
| // Initialize the TMB, register this thread as sender and receiver for |
| // appropriate types of messages. |
| bus_.Initialize(); |
| foreman_client_id_ = bus_.Connect(); |
| bus_.RegisterClientAsReceiver(foreman_client_id_, kDataPipelineMessage); |
| bus_.RegisterClientAsReceiver(foreman_client_id_, kCatalogRelationNewBlockMessage); |
| |
| const tmb::client_id worker_thread_client_id = bus_.Connect(); |
| bus_.RegisterClientAsSender(worker_thread_client_id, kCatalogRelationNewBlockMessage); |
| bus_.RegisterClientAsSender(worker_thread_client_id, kDataPipelineMessage); |
| |
| thread_id_map_ = ClientIDMap::Instance(); |
| // Usually the worker thread makes the following call. In this test setup, |
| // we don't have a worker thread hence we have to explicitly make the call. |
| thread_id_map_->addValue(worker_thread_client_id); |
| |
| storage_manager_.reset(new StorageManager(kStoragePath)); |
| |
| // Initialize table attributes. |
| table_.reset(new CatalogRelation(nullptr, kTableName, kTableId)); |
| const Type &int_type = IntType::InstanceNonNullable(); |
| const Type &null_int_type = IntType::InstanceNullable(); |
| table_->addAttribute(new CatalogAttribute(table_.get(), "col-1", int_type)); |
| table_->addAttribute(new CatalogAttribute(table_.get(), "col-2", int_type)); |
| table_->addAttribute(new CatalogAttribute(table_.get(), "col-3", int_type)); |
| table_->addAttribute(new CatalogAttribute(table_.get(), "null-col-1", null_int_type)); |
| table_->addAttribute(new CatalogAttribute(table_.get(), "null-col-2", null_int_type)); |
| table_->addAttribute(new CatalogAttribute(table_.get(), "null-col-3", null_int_type)); |
| table_->addAttribute(new CatalogAttribute(table_.get(), "tid", null_int_type)); |
| col1_ = table_->getAttributeByName("col-1")->getID(); |
| col2_ = table_->getAttributeByName("col-2")->getID(); |
| col3_ = table_->getAttributeByName("col-3")->getID(); |
| null_col1_ = table_->getAttributeByName("null-col-1")->getID(); |
| null_col2_ = table_->getAttributeByName("null-col-2")->getID(); |
| null_col3_ = table_->getAttributeByName("null-col-3")->getID(); |
| tid_col_ = table_->getAttributeByName("tid")->getID(); |
| table_->setDefaultStorageBlockLayout(StorageBlockLayout::GenerateDefaultLayout(*table_, false)); |
| |
| insert_destination_.reset( |
| new BlockPoolInsertDestination(*table_, |
| nullptr, |
| storage_manager_.get(), |
| kOpIndex, |
| 0, // dummy query ID. |
| foreman_client_id_, |
| &bus_)); |
| } |
| |
| virtual void TearDown() { |
| // Usually the worker thread makes the following call. In this test setup, |
| // we don't have a worker thread hence we have to explicitly make the call. |
| thread_id_map_->removeValue(); |
| |
| // Drop blocks from relations and InsertDestination. |
| const vector<block_id> tmp_blocks = insert_destination_->getTouchedBlocks(); |
| for (const block_id block : tmp_blocks) { |
| storage_manager_->deleteBlockOrBlobFile(block); |
| } |
| |
| const vector<block_id> blocks = table_->getBlocksSnapshot(); |
| for (const block_id block : blocks) { |
| storage_manager_->deleteBlockOrBlobFile(block); |
| } |
| } |
| |
| // Helper method to create test tuples. |
| Tuple *createTuple(tuple_id id) { |
| std::vector<TypedValue> values; |
| TestTuple tuple(id); |
| values.emplace_back(static_cast<int>(tuple.col1_)); |
| values.emplace_back(static_cast<int>(tuple.col2_)); |
| values.emplace_back(static_cast<int>(tuple.col3_)); |
| if (!tuple.col1_) { |
| values.emplace_back(kInt); |
| } else { |
| values.emplace_back(static_cast<int>(tuple.col1_)); |
| } |
| if (!tuple.col2_) { |
| values.emplace_back(kInt); |
| } else { |
| values.emplace_back(static_cast<int>(tuple.col2_)); |
| } |
| if (!tuple.col3_) { |
| values.emplace_back(kInt); |
| } else { |
| values.emplace_back(static_cast<int>(tuple.col3_)); |
| } |
| values.emplace_back(static_cast<int>(tuple.tid_)); |
| return new Tuple(std::move(values)); |
| } |
| |
| // Method to runs which are sorted based on supplied Tuple comparator. |
| template <typename ComparatorT> |
| void createRuns(const ComparatorT &comparator, bool maintain_top_k = false) { |
| std::random_device rd; |
| std::mt19937 gen(rd()); |
| std::uniform_int_distribution<> dist(0, (1 << 8) - 1); |
| std::unique_ptr<Tuple> tuple; |
| std::vector<TestTuple> top_k_tuples; |
| MutableBlockReference storage_block; |
| |
| auto sort_comparator = |
| [this, comparator](const TestTuple &l, const TestTuple &r) -> bool { |
| std::unique_ptr<Tuple> left(createTuple(l.tid_)), |
| right(createTuple(r.tid_)); |
| return comparator(*left, *right); |
| }; |
| |
| input_runs_.clear(); |
| for (std::size_t run_id = 0; run_id < kNumRuns; ++run_id) { |
| std::vector<TestTuple> tuples; |
| input_runs_.emplace_back(); |
| for (std::size_t i = 0; i < kNumBlocksPerRun * kNumTuplesPerBlock; ++i) { |
| tuples.emplace_back(dist(gen)); |
| } |
| |
| // Sort tuples before creating the run. |
| std::sort(tuples.begin(), tuples.end(), sort_comparator); |
| |
| // Maintain top-K to test top-K output after merge. |
| if (maintain_top_k) { |
| top_k_tuples.insert(top_k_tuples.end(), |
| tuples.begin(), |
| tuples.begin() + std::min(tuples.size(), kTopK)); |
| std::partial_sort( |
| top_k_tuples.begin(), |
| top_k_tuples.begin() + std::min(top_k_tuples.size(), kTopK), |
| top_k_tuples.end(), |
| sort_comparator); |
| if (top_k_tuples.size() > kTopK) { |
| top_k_tuples.erase(top_k_tuples.begin() + kTopK, top_k_tuples.end()); |
| } |
| } |
| |
| // Insert sorted tuples in run. |
| for (std::size_t id = 0; id < kNumBlocksPerRun; ++id) { |
| std::string run_name = "run[" + std::to_string(run_id) + "]"; |
| // Create block. |
| block_id block_id = storage_manager_->createBlock(*table_, table_->getDefaultStorageBlockLayout()); |
| storage_block = storage_manager_->getBlockMutable(block_id, *table_); |
| table_->addBlock(block_id); |
| input_runs_.back().push_back(block_id); |
| for (std::size_t tid = 0; tid < kNumTuplesPerBlock; ++tid) { |
| // Insert tuples. |
| tuple.reset(createTuple(tuples[id * kNumTuplesPerBlock + tid].tid_)); |
| printTuple(run_name.c_str(), *tuple); |
| EXPECT_TRUE(storage_block->insertTupleInBatch(*tuple)); |
| } |
| storage_block->rebuild(); |
| } |
| } |
| |
| if (maintain_top_k) { |
| // Last expected ORDER BY key of the sort. |
| last_expected_tuple_.reset(createTuple(top_k_tuples.back().tid_)); |
| } |
| |
| // Insert empty run. |
| input_runs_.emplace_back(); |
| } |
| |
| // Debug print method. |
| void printTuple(const char *prefix, const Tuple &tuple) { |
| if (VLOG_IS_ON(2)) { |
| std::ostringstream out; |
| out << prefix << ' '; |
| CatalogRelation::const_iterator attr_it = table_->begin(); |
| Tuple::const_iterator value_it = tuple.begin(); |
| for (; attr_it != table_->end(); ++attr_it, ++value_it) { |
| if (value_it->isNull()) { |
| out << "NULL"; |
| } else { |
| out << attr_it->getType().printValueToString(*value_it); |
| } |
| out << '|'; |
| } |
| out << '\n'; |
| VLOG(2) << out.str(); |
| } |
| } |
| |
| // Merge runs. |
| void mergeRuns(const std::vector<attribute_id> &attrs, |
| vector<bool> &&ordering, |
| vector<bool> &&null_ordering, |
| const std::size_t top_k = 0) { |
| PtrVector<Scalar> order_by; |
| for (const attribute_id attr : attrs) { |
| order_by.push_back(new ScalarAttribute(*table_->getAttributeById(attr))); |
| } |
| SortConfiguration sort_config(order_by, move(ordering), move(null_ordering)); |
| RunMerger merge(sort_config, |
| std::move(input_runs_), |
| top_k, |
| *table_, |
| insert_destination_.get(), |
| 0, |
| storage_manager_.get()); |
| merge.doMerge(); |
| output_run_ = std::move(*merge.getOutputRunMutable()); |
| EXPECT_GT(output_run_.size(), 0u); |
| } |
| |
| // Check if the merged run is sorted. |
| template <typename ComparatorT> |
| void checkOutputRun(const ComparatorT comparator, |
| const tuple_id expected_num_tuples = kNumTuples) { |
| BlockReference block( |
| storage_manager_->getBlock(output_run_.front(), *table_)); |
| std::unique_ptr<ValueAccessor> dummy_accessor( |
| block->getTupleStorageSubBlock().createValueAccessor()); |
| tuple_id num_tuples = 0; |
| std::unique_ptr<Tuple> prev, current; |
| InvokeOnValueAccessorNotAdapter( |
| dummy_accessor.get(), |
| [&](auto *dummy_accessor) -> void { // NOLINT(build/c++11) |
| RunIterator< |
| typename std::remove_reference<decltype(*dummy_accessor)>::type> |
| iterator(output_run_, storage_manager_.get(), *table_); |
| while (iterator.next()) { |
| prev = std::move(current); |
| current.reset(iterator.getValueAccessor()->getTuple()); |
| ++num_tuples; |
| if (num_tuples > 1) { |
| EXPECT_FALSE(comparator(*current, *prev)); |
| } |
| std::unique_ptr<Tuple> tuple(iterator.getValueAccessor()->getTuple()); |
| this->printTuple(">", *tuple); |
| } |
| }); |
| EXPECT_EQ(expected_num_tuples, num_tuples); |
| last_actual_tuple_ = std::move(current); |
| } |
| |
| // Check attribute value of last expected tuple and last output tuple. |
| void checkLastTupleAttrValueIsEqual(const attribute_id id) { |
| printTuple("Expected:", *last_expected_tuple_); |
| printTuple("Actual: ", *last_actual_tuple_); |
| EXPECT_EQ(last_expected_tuple_->getAttributeValue(id).isNull(), |
| last_actual_tuple_->getAttributeValue(id).isNull()); |
| if (!last_expected_tuple_->getAttributeValue(id).isNull() && |
| !last_actual_tuple_->getAttributeValue(id).isNull()) { |
| EXPECT_TRUE(last_expected_tuple_->getAttributeValue(id).fastEqualCheck( |
| last_actual_tuple_->getAttributeValue(id))) |
| << "Expected: " |
| << last_expected_tuple_->getAttributeValue(id).getLiteral<int>() |
| << "; Actual: " |
| << last_actual_tuple_->getAttributeValue(id).getLiteral<int>(); |
| } |
| } |
| |
| std::unique_ptr<StorageManager> storage_manager_; |
| std::unique_ptr<CatalogRelation> table_; |
| std::unique_ptr<InsertDestination> insert_destination_; |
| attribute_id col1_; |
| attribute_id col2_; |
| attribute_id col3_; |
| attribute_id null_col1_; |
| attribute_id null_col2_; |
| attribute_id null_col3_; |
| attribute_id tid_col_; |
| merge_run_operator::Run output_run_; |
| std::vector<merge_run_operator::Run> input_runs_; |
| std::unique_ptr<Tuple> last_actual_tuple_; |
| std::unique_ptr<Tuple> last_expected_tuple_; // This is only populated for top-k tests. |
| |
| MessageBusImpl bus_; |
| tmb::client_id foreman_client_id_; |
| |
| // This map is needed for InsertDestination and some operators that send |
| // messages to Foreman directly. To know the reason behind the design of this |
| // map, see the note in InsertDestination.hpp. |
| ClientIDMap *thread_id_map_; |
| }; |
| |
| const char RunMergerTest::kTableName[] = "table"; |
| const char RunMergerTest::kStoragePath[] = "./sort_merge_run_operator_test_data"; |
| const std::size_t RunMergerTest::kNumTuplesPerBlock = 10; |
| const tuple_id RunMergerTest::kNumBlocksPerRun = 10; |
| const std::size_t RunMergerTest::kNumRuns = 10; |
| const tuple_id RunMergerTest::kNumTuples = |
| RunMergerTest::kNumTuplesPerBlock * RunMergerTest::kNumBlocksPerRun * RunMergerTest::kNumRuns; |
| const std::size_t RunMergerTest::kTopK = 5; |
| |
| namespace { |
| |
| // Comparator for col-1 ASC. |
| auto kCol1AscComparator = [](const Tuple &left, const Tuple &right) -> bool { |
| TestTupleAttrs l = TupleToTupleAttr(left); |
| TestTupleAttrs r = TupleToTupleAttr(right); |
| if (l.c1 < r.c1) { |
| return true; |
| } |
| return false; |
| }; |
| |
| // Comparator for col-1 DESC. |
| auto kCol1DescComparator = [](const Tuple &left, const Tuple &right) -> bool { |
| TestTupleAttrs l = TupleToTupleAttr(left); |
| TestTupleAttrs r = TupleToTupleAttr(right); |
| if (l.c1 > r.c1) { |
| return true; |
| } |
| return false; |
| }; |
| |
| // Comparator for null-col-1 ASC NULLS LAST. |
| auto kNullLastCol1AscComparator = |
| [](const Tuple &left, const Tuple &right) -> bool { |
| TestTupleAttrs l = TupleToTupleAttr(left); |
| TestTupleAttrs r = TupleToTupleAttr(right); |
| l.c1 = l.null_c4 ? std::numeric_limits<int>::max() : l.c1; |
| r.c1 = r.null_c4 ? std::numeric_limits<int>::max() : r.c1; |
| if (l.c1 < r.c1) { |
| return true; |
| } |
| return false; |
| }; |
| |
| // Comparator for null-col-1 ASC NULLS FIRST. |
| auto kNullFirstCol1AscComparator = |
| [](const Tuple &left, const Tuple &right) -> bool { |
| TestTupleAttrs l = TupleToTupleAttr(left); |
| TestTupleAttrs r = TupleToTupleAttr(right); |
| l.c1 = l.null_c4 ? std::numeric_limits<int>::min() : l.c1; |
| r.c1 = r.null_c4 ? std::numeric_limits<int>::min() : r.c1; |
| if (l.c1 < r.c1) { |
| return true; |
| } |
| return false; |
| }; |
| |
| // Comparator for null-col-1 DESC NULLS LAST. |
| auto kNullLastCol1DescComparator = |
| [](const Tuple &left, const Tuple &right) -> bool { |
| TestTupleAttrs l = TupleToTupleAttr(left); |
| TestTupleAttrs r = TupleToTupleAttr(right); |
| l.c1 = l.null_c4 ? std::numeric_limits<int>::min() : l.c1; |
| r.c1 = r.null_c4 ? std::numeric_limits<int>::min() : r.c1; |
| if (l.c1 > r.c1) { |
| return true; |
| } |
| return false; |
| }; |
| |
| // Comparator for null-col-1 DESC NULLS FIRST. |
| auto kNullFirstCol1DescComparator = |
| [](const Tuple &left, const Tuple &right) -> bool { |
| TestTupleAttrs l = TupleToTupleAttr(left); |
| TestTupleAttrs r = TupleToTupleAttr(right); |
| l.c1 = l.null_c4 ? std::numeric_limits<int>::max() : l.c1; |
| r.c1 = r.null_c4 ? std::numeric_limits<int>::max() : r.c1; |
| if (l.c1 > r.c1) { |
| return true; |
| } |
| return false; |
| }; |
| |
| // Comparator for col-1 ASC, col-2 ASC, col-3 ASC. |
| auto kCol1Col2Col3AscComparator = |
| [](const Tuple &left, const Tuple &right) -> bool { |
| TestTupleAttrs l = TupleToTupleAttr(left); |
| TestTupleAttrs r = TupleToTupleAttr(right); |
| if (l.c1 < r.c1) { |
| return true; |
| } else if (l.c1 == r.c1) { |
| if (l.c2 < r.c2) { |
| return true; |
| } else if (l.c2 == r.c2) { |
| if (l.c3 < r.c3) { |
| return true; |
| } |
| } |
| } |
| return false; |
| }; |
| |
| // Comparator for col-1 DESC, col-2 DESC, col-3 DESC. |
| auto kCol1Col2Col3DescComparator = |
| [](const Tuple &left, const Tuple &right) -> bool { |
| TestTupleAttrs l = TupleToTupleAttr(left); |
| TestTupleAttrs r = TupleToTupleAttr(right); |
| if (l.c1 > r.c1) { |
| return true; |
| } else if (l.c1 == r.c1) { |
| if (l.c2 > r.c2) { |
| return true; |
| } else if (l.c2 == r.c2) { |
| if (l.c3 > r.c3) { |
| return true; |
| } |
| } |
| } |
| return false; |
| }; |
| |
| // Comparator for col-1 ASC NULLS LAST, col-2 ASC NULLS LAST, col-3 ASC NULLS |
| // LAST. |
| auto kNullLastCol1Col2Col3AscComparator = |
| [](const Tuple &left, const Tuple &right) -> bool { |
| TestTupleAttrs l = TupleToTupleAttr(left); |
| TestTupleAttrs r = TupleToTupleAttr(right); |
| l.c1 = l.null_c4 ? std::numeric_limits<int>::max() : l.c1; |
| l.c2 = l.null_c5 ? std::numeric_limits<int>::max() : l.c2; |
| l.c3 = l.null_c6 ? std::numeric_limits<int>::max() : l.c3; |
| r.c1 = r.null_c4 ? std::numeric_limits<int>::max() : r.c1; |
| r.c2 = r.null_c5 ? std::numeric_limits<int>::max() : r.c2; |
| r.c3 = r.null_c6 ? std::numeric_limits<int>::max() : r.c3; |
| if (l.c1 < r.c1) { |
| return true; |
| } else if (l.c1 == r.c1) { |
| if (l.c2 < r.c2) { |
| return true; |
| } else if (l.c2 == r.c2) { |
| if (l.c3 < r.c3) { |
| return true; |
| } |
| } |
| } |
| return false; |
| }; |
| |
| // Comparator for col-1 DESC NULLS LAST, col-2 DESC NULLS LAST, col-3 DESC NULLS |
| // LAST. |
| auto kNullLastCol1Col2Col3DescComparator = |
| [](const Tuple &left, const Tuple &right) -> bool { |
| TestTupleAttrs l = TupleToTupleAttr(left); |
| TestTupleAttrs r = TupleToTupleAttr(right); |
| l.c1 = l.null_c4 ? std::numeric_limits<int>::min() : l.c1; |
| l.c2 = l.null_c5 ? std::numeric_limits<int>::min() : l.c2; |
| l.c3 = l.null_c6 ? std::numeric_limits<int>::min() : l.c3; |
| r.c1 = r.null_c4 ? std::numeric_limits<int>::min() : r.c1; |
| r.c2 = r.null_c5 ? std::numeric_limits<int>::min() : r.c2; |
| r.c3 = r.null_c6 ? std::numeric_limits<int>::min() : r.c3; |
| if (l.c1 > r.c1) { |
| return true; |
| } else if (l.c1 == r.c1) { |
| if (l.c2 > r.c2) { |
| return true; |
| } else if (l.c2 == r.c2) { |
| if (l.c3 > r.c3) { |
| return true; |
| } |
| } |
| } |
| return false; |
| }; |
| |
| // Comparator for col-1 ASC NULLS FIRST, col-2 ASC NULLS FIRST, col-3 ASC NULLS |
| // FIRST. |
| auto kNullFirstCol1Col2Col3AscComparator = |
| [](const Tuple &left, const Tuple &right) -> bool { |
| TestTupleAttrs l = TupleToTupleAttr(left); |
| TestTupleAttrs r = TupleToTupleAttr(right); |
| l.c1 = l.null_c4 ? std::numeric_limits<int>::min() : l.c1; |
| l.c2 = l.null_c5 ? std::numeric_limits<int>::min() : l.c2; |
| l.c3 = l.null_c6 ? std::numeric_limits<int>::min() : l.c3; |
| r.c1 = r.null_c4 ? std::numeric_limits<int>::min() : r.c1; |
| r.c2 = r.null_c5 ? std::numeric_limits<int>::min() : r.c2; |
| r.c3 = r.null_c6 ? std::numeric_limits<int>::min() : r.c3; |
| if (l.c1 < r.c1) { |
| return true; |
| } else if (l.c1 == r.c1) { |
| if (l.c2 < r.c2) { |
| return true; |
| } else if (l.c2 == r.c2) { |
| if (l.c3 < r.c3) { |
| return true; |
| } |
| } |
| } |
| return false; |
| }; |
| |
| // Comparator for col-1 DESC NULLS FIRST, col-2 DESC NULLS FIRST, col-3 DESC |
| // NULLS FIRST. |
| auto kNullFirstCol1Col2Col3DescComparator = |
| [](const Tuple &left, const Tuple &right) -> bool { |
| TestTupleAttrs l = TupleToTupleAttr(left); |
| TestTupleAttrs r = TupleToTupleAttr(right); |
| l.c1 = l.null_c4 ? std::numeric_limits<int>::max() : l.c1; |
| l.c2 = l.null_c5 ? std::numeric_limits<int>::max() : l.c2; |
| l.c3 = l.null_c6 ? std::numeric_limits<int>::max() : l.c3; |
| r.c1 = r.null_c4 ? std::numeric_limits<int>::max() : r.c1; |
| r.c2 = r.null_c5 ? std::numeric_limits<int>::max() : r.c2; |
| r.c3 = r.null_c6 ? std::numeric_limits<int>::max() : r.c3; |
| if (l.c1 > r.c1) { |
| return true; |
| } else if (l.c1 == r.c1) { |
| if (l.c2 > r.c2) { |
| return true; |
| } else if (l.c2 == r.c2) { |
| if (l.c3 > r.c3) { |
| return true; |
| } |
| } |
| } |
| return false; |
| }; |
| |
| // Comparator for col-1 ASC NULLS FIRST, col-2 DESC NULLS LAST, col-3 ASC NULLS |
| // LAST. |
| auto kMixedNullFLLCol1Col2Col3Comparator = |
| [](const Tuple &left, const Tuple &right) -> bool { |
| TestTupleAttrs l = TupleToTupleAttr(left); |
| TestTupleAttrs r = TupleToTupleAttr(right); |
| l.c1 = l.null_c4 ? std::numeric_limits<int>::min() : l.c1; |
| l.c2 = l.null_c5 ? std::numeric_limits<int>::min() : l.c2; |
| l.c3 = l.null_c6 ? std::numeric_limits<int>::max() : l.c3; |
| r.c1 = r.null_c4 ? std::numeric_limits<int>::min() : r.c1; |
| r.c2 = r.null_c5 ? std::numeric_limits<int>::min() : r.c2; |
| r.c3 = r.null_c6 ? std::numeric_limits<int>::max() : r.c3; |
| if (l.c1 < r.c1) { |
| return true; |
| } else if (l.c1 == r.c1) { |
| if (l.c2 > r.c2) { |
| return true; |
| } else if (l.c2 == r.c2) { |
| if (l.c3 < r.c3) { |
| return true; |
| } |
| } |
| } |
| return false; |
| }; |
| |
| } // namespace |
| |
| TEST_F(RunMergerTest, 1Column_NonNull_Asc) { |
| std::vector<attribute_id> attrs{col1_}; |
| std::vector<bool> ordering{kSortAscending}; |
| std::vector<bool> null_ordering{kSortNullLast}; |
| |
| createRuns(kCol1AscComparator); |
| mergeRuns(attrs, move(ordering), move(null_ordering)); |
| checkOutputRun(kCol1AscComparator); |
| } |
| |
| TEST_F(RunMergerTest, 1Column_NonNull_Asc_TopK) { |
| std::vector<attribute_id> attrs{col1_}; |
| std::vector<bool> ordering{kSortAscending}; |
| std::vector<bool> null_ordering{kSortNullLast}; |
| |
| createRuns(kCol1AscComparator, true); |
| mergeRuns(attrs, move(ordering), move(null_ordering), RunMergerTest::kTopK); |
| checkOutputRun(kCol1AscComparator, RunMergerTest::kTopK); |
| checkLastTupleAttrValueIsEqual(col1_); |
| } |
| |
| TEST_F(RunMergerTest, 1Column_NonNull_Desc) { |
| std::vector<attribute_id> attrs{col1_}; |
| std::vector<bool> ordering{kSortDescending}; |
| std::vector<bool> null_ordering{kSortNullLast}; |
| |
| createRuns(kCol1DescComparator); |
| mergeRuns(attrs, move(ordering), move(null_ordering)); |
| checkOutputRun(kCol1DescComparator); |
| } |
| |
| TEST_F(RunMergerTest, 1Column_NonNull_Desc_TopK) { |
| std::vector<attribute_id> attrs{col1_}; |
| std::vector<bool> ordering{kSortDescending}; |
| std::vector<bool> null_ordering{kSortNullLast}; |
| |
| createRuns(kCol1DescComparator, true); |
| mergeRuns(attrs, move(ordering), move(null_ordering), RunMergerTest::kTopK); |
| checkOutputRun(kCol1DescComparator, RunMergerTest::kTopK); |
| checkLastTupleAttrValueIsEqual(col1_); |
| } |
| |
| TEST_F(RunMergerTest, 1Column_NullLast_Asc) { |
| std::vector<attribute_id> attrs{null_col1_}; |
| std::vector<bool> ordering{kSortAscending}; |
| std::vector<bool> null_ordering{kSortNullLast}; |
| |
| createRuns(kNullLastCol1AscComparator); |
| mergeRuns(attrs, move(ordering), move(null_ordering)); |
| checkOutputRun(kNullLastCol1AscComparator); |
| } |
| |
| TEST_F(RunMergerTest, 1Column_NullLast_Asc_TopK) { |
| std::vector<attribute_id> attrs{null_col1_}; |
| std::vector<bool> ordering{kSortAscending}; |
| std::vector<bool> null_ordering{kSortNullLast}; |
| |
| createRuns(kNullLastCol1AscComparator, true); |
| mergeRuns(attrs, move(ordering), move(null_ordering), RunMergerTest::kTopK); |
| checkOutputRun(kNullLastCol1AscComparator, RunMergerTest::kTopK); |
| checkLastTupleAttrValueIsEqual(null_col1_); |
| } |
| |
| TEST_F(RunMergerTest, 1Column_NullFirst_Asc) { |
| std::vector<attribute_id> attrs{null_col1_}; |
| std::vector<bool> ordering{kSortAscending}; |
| std::vector<bool> null_ordering{kSortNullFirst}; |
| |
| createRuns(kNullFirstCol1AscComparator); |
| mergeRuns(attrs, move(ordering), move(null_ordering)); |
| checkOutputRun(kNullFirstCol1AscComparator); |
| } |
| |
| TEST_F(RunMergerTest, 1Column_NullFirst_Asc_TopK) { |
| std::vector<attribute_id> attrs{null_col1_}; |
| std::vector<bool> ordering{kSortAscending}; |
| std::vector<bool> null_ordering{kSortNullFirst}; |
| |
| createRuns(kNullFirstCol1AscComparator, true); |
| mergeRuns(attrs, move(ordering), move(null_ordering), RunMergerTest::kTopK); |
| checkOutputRun(kNullFirstCol1AscComparator, RunMergerTest::kTopK); |
| checkLastTupleAttrValueIsEqual(null_col1_); |
| } |
| |
| TEST_F(RunMergerTest, 1Column_NullLast_Desc) { |
| std::vector<attribute_id> attrs{null_col1_}; |
| std::vector<bool> ordering{kSortDescending}; |
| std::vector<bool> null_ordering{kSortNullLast}; |
| |
| createRuns(kNullLastCol1DescComparator); |
| mergeRuns(attrs, move(ordering), move(null_ordering)); |
| checkOutputRun(kNullLastCol1DescComparator); |
| } |
| |
| TEST_F(RunMergerTest, 1Column_NullLast_Desc_TopK) { |
| std::vector<attribute_id> attrs{null_col1_}; |
| std::vector<bool> ordering{kSortDescending}; |
| std::vector<bool> null_ordering{kSortNullLast}; |
| |
| createRuns(kNullLastCol1DescComparator, true); |
| mergeRuns(attrs, move(ordering), move(null_ordering), RunMergerTest::kTopK); |
| checkOutputRun(kNullLastCol1DescComparator, RunMergerTest::kTopK); |
| checkLastTupleAttrValueIsEqual(null_col1_); |
| } |
| |
| TEST_F(RunMergerTest, 1Column_NullFirst_Desc) { |
| std::vector<attribute_id> attrs{null_col1_}; |
| std::vector<bool> ordering{kSortDescending}; |
| std::vector<bool> null_ordering{kSortNullFirst}; |
| |
| createRuns(kNullFirstCol1DescComparator); |
| mergeRuns(attrs, move(ordering), move(null_ordering)); |
| checkOutputRun(kNullFirstCol1DescComparator); |
| } |
| |
| TEST_F(RunMergerTest, 1Column_NullFirst_Desc_TopK) { |
| std::vector<attribute_id> attrs{null_col1_}; |
| std::vector<bool> ordering{kSortDescending}; |
| std::vector<bool> null_ordering{kSortNullFirst}; |
| |
| createRuns(kNullFirstCol1DescComparator, true); |
| mergeRuns(attrs, move(ordering), move(null_ordering), RunMergerTest::kTopK); |
| checkOutputRun(kNullFirstCol1DescComparator, RunMergerTest::kTopK); |
| checkLastTupleAttrValueIsEqual(null_col1_); |
| } |
| |
| TEST_F(RunMergerTest, 3Column_NonNull_Asc) { |
| std::vector<attribute_id> attrs{col1_, col2_, col3_}; |
| std::vector<bool> ordering{kSortAscending, kSortAscending, kSortAscending}; |
| std::vector<bool> null_ordering{kSortNullLast, kSortNullLast, kSortNullLast}; |
| |
| createRuns(kCol1Col2Col3AscComparator); |
| mergeRuns(attrs, move(ordering), move(null_ordering)); |
| checkOutputRun(kCol1Col2Col3AscComparator); |
| } |
| |
| TEST_F(RunMergerTest, 3Column_NonNull_Asc_TopK) { |
| std::vector<attribute_id> attrs{col1_, col2_, col3_}; |
| std::vector<bool> ordering{kSortAscending, kSortAscending, kSortAscending}; |
| std::vector<bool> null_ordering{kSortNullLast, kSortNullLast, kSortNullLast}; |
| |
| createRuns(kCol1Col2Col3AscComparator, true); |
| mergeRuns(attrs, move(ordering), move(null_ordering), RunMergerTest::kTopK); |
| checkOutputRun(kCol1Col2Col3AscComparator, RunMergerTest::kTopK); |
| checkLastTupleAttrValueIsEqual(col1_); |
| checkLastTupleAttrValueIsEqual(col2_); |
| checkLastTupleAttrValueIsEqual(col3_); |
| } |
| |
| TEST_F(RunMergerTest, 3Column_NonNull_Desc) { |
| std::vector<attribute_id> attrs{col1_, col2_, col3_}; |
| std::vector<bool> ordering{kSortDescending, kSortDescending, kSortDescending}; |
| std::vector<bool> null_ordering{kSortNullLast, kSortNullLast, kSortNullLast}; |
| |
| createRuns(kCol1Col2Col3DescComparator); |
| mergeRuns(attrs, move(ordering), move(null_ordering)); |
| checkOutputRun(kCol1Col2Col3DescComparator); |
| } |
| |
| TEST_F(RunMergerTest, 3Column_NonNull_Desc_TopK) { |
| std::vector<attribute_id> attrs{col1_, col2_, col3_}; |
| std::vector<bool> ordering{kSortDescending, kSortDescending, kSortDescending}; |
| std::vector<bool> null_ordering{kSortNullLast, kSortNullLast, kSortNullLast}; |
| |
| createRuns(kCol1Col2Col3DescComparator, true); |
| mergeRuns(attrs, move(ordering), move(null_ordering), RunMergerTest::kTopK); |
| checkOutputRun(kCol1Col2Col3DescComparator, RunMergerTest::kTopK); |
| checkLastTupleAttrValueIsEqual(col1_); |
| checkLastTupleAttrValueIsEqual(col2_); |
| checkLastTupleAttrValueIsEqual(col3_); |
| } |
| |
| TEST_F(RunMergerTest, 3Column_NullLast_Asc) { |
| std::vector<attribute_id> attrs{null_col1_, null_col2_, null_col3_}; |
| std::vector<bool> ordering{kSortAscending, kSortAscending, kSortAscending}; |
| std::vector<bool> null_ordering{kSortNullLast, kSortNullLast, kSortNullLast}; |
| |
| createRuns(kNullLastCol1Col2Col3AscComparator); |
| mergeRuns(attrs, move(ordering), move(null_ordering)); |
| checkOutputRun(kNullLastCol1Col2Col3AscComparator); |
| } |
| |
| TEST_F(RunMergerTest, 3Column_NullLast_Asc_TopK) { |
| std::vector<attribute_id> attrs{null_col1_, null_col2_, null_col3_}; |
| std::vector<bool> ordering{kSortAscending, kSortAscending, kSortAscending}; |
| std::vector<bool> null_ordering{kSortNullLast, kSortNullLast, kSortNullLast}; |
| |
| createRuns(kNullLastCol1Col2Col3AscComparator, true); |
| mergeRuns(attrs, move(ordering), move(null_ordering), RunMergerTest::kTopK); |
| checkOutputRun(kNullLastCol1Col2Col3AscComparator, RunMergerTest::kTopK); |
| checkLastTupleAttrValueIsEqual(null_col1_); |
| checkLastTupleAttrValueIsEqual(null_col2_); |
| checkLastTupleAttrValueIsEqual(null_col3_); |
| } |
| |
| TEST_F(RunMergerTest, 3Column_NullLast_Desc) { |
| std::vector<attribute_id> attrs{null_col1_, null_col2_, null_col3_}; |
| std::vector<bool> ordering{kSortDescending, kSortDescending, kSortDescending}; |
| std::vector<bool> null_ordering{kSortNullLast, kSortNullLast, kSortNullLast}; |
| |
| createRuns(kNullLastCol1Col2Col3DescComparator); |
| mergeRuns(attrs, move(ordering), move(null_ordering)); |
| checkOutputRun(kNullLastCol1Col2Col3DescComparator); |
| } |
| |
| TEST_F(RunMergerTest, 3Column_NullLast_Desc_TopK) { |
| std::vector<attribute_id> attrs{null_col1_, null_col2_, null_col3_}; |
| std::vector<bool> ordering{kSortDescending, kSortDescending, kSortDescending}; |
| std::vector<bool> null_ordering{kSortNullLast, kSortNullLast, kSortNullLast}; |
| |
| createRuns(kNullLastCol1Col2Col3DescComparator, true); |
| mergeRuns(attrs, move(ordering), move(null_ordering), RunMergerTest::kTopK); |
| checkOutputRun(kNullLastCol1Col2Col3DescComparator, RunMergerTest::kTopK); |
| checkLastTupleAttrValueIsEqual(null_col1_); |
| checkLastTupleAttrValueIsEqual(null_col2_); |
| checkLastTupleAttrValueIsEqual(null_col3_); |
| } |
| |
| TEST_F(RunMergerTest, 3Column_NullFirst_Asc) { |
| std::vector<attribute_id> attrs{null_col1_, null_col2_, null_col3_}; |
| std::vector<bool> ordering{kSortAscending, kSortAscending, kSortAscending}; |
| std::vector<bool> null_ordering{kSortNullFirst, kSortNullFirst, kSortNullFirst}; |
| |
| createRuns(kNullFirstCol1Col2Col3AscComparator); |
| mergeRuns(attrs, move(ordering), move(null_ordering)); |
| checkOutputRun(kNullFirstCol1Col2Col3AscComparator); |
| } |
| |
| TEST_F(RunMergerTest, 3Column_NullFirst_Asc_TopK) { |
| std::vector<attribute_id> attrs{null_col1_, null_col2_, null_col3_}; |
| std::vector<bool> ordering{kSortAscending, kSortAscending, kSortAscending}; |
| std::vector<bool> null_ordering{kSortNullFirst, kSortNullFirst, kSortNullFirst}; |
| |
| createRuns(kNullFirstCol1Col2Col3AscComparator, true); |
| mergeRuns(attrs, move(ordering), move(null_ordering), RunMergerTest::kTopK); |
| checkOutputRun(kNullFirstCol1Col2Col3AscComparator, RunMergerTest::kTopK); |
| checkLastTupleAttrValueIsEqual(null_col1_); |
| checkLastTupleAttrValueIsEqual(null_col2_); |
| checkLastTupleAttrValueIsEqual(null_col3_); |
| } |
| |
| TEST_F(RunMergerTest, 3Column_NullFirst_Desc) { |
| std::vector<attribute_id> attrs{null_col1_, null_col2_, null_col3_}; |
| std::vector<bool> ordering{kSortDescending, kSortDescending, kSortDescending}; |
| std::vector<bool> null_ordering{kSortNullFirst, kSortNullFirst, kSortNullFirst}; |
| |
| createRuns(kNullFirstCol1Col2Col3DescComparator); |
| mergeRuns(attrs, move(ordering), move(null_ordering)); |
| checkOutputRun(kNullFirstCol1Col2Col3DescComparator); |
| } |
| |
| TEST_F(RunMergerTest, 3Column_NullFirst_Desc_TopK) { |
| std::vector<attribute_id> attrs{null_col1_, null_col2_, null_col3_}; |
| std::vector<bool> ordering{kSortDescending, kSortDescending, kSortDescending}; |
| std::vector<bool> null_ordering{kSortNullFirst, kSortNullFirst, kSortNullFirst}; |
| |
| createRuns(kNullFirstCol1Col2Col3DescComparator, true); |
| mergeRuns(attrs, move(ordering), move(null_ordering), RunMergerTest::kTopK); |
| checkOutputRun(kNullFirstCol1Col2Col3DescComparator, RunMergerTest::kTopK); |
| checkLastTupleAttrValueIsEqual(null_col1_); |
| checkLastTupleAttrValueIsEqual(null_col2_); |
| checkLastTupleAttrValueIsEqual(null_col3_); |
| } |
| |
| TEST_F(RunMergerTest, 3Column_MixedNullOrdering_MixedOrdering) { |
| std::vector<attribute_id> attrs{null_col1_, null_col2_, null_col3_}; |
| std::vector<bool> ordering{kSortAscending, kSortDescending, kSortAscending}; |
| std::vector<bool> null_ordering{kSortNullFirst, kSortNullLast, kSortNullLast}; |
| |
| createRuns(kMixedNullFLLCol1Col2Col3Comparator); |
| mergeRuns(attrs, move(ordering), move(null_ordering)); |
| checkOutputRun(kMixedNullFLLCol1Col2Col3Comparator); |
| } |
| |
| TEST_F(RunMergerTest, 3Column_MixedNullOrdering_MixedOrdering_TopK) { |
| std::vector<attribute_id> attrs{null_col1_, null_col2_, null_col3_}; |
| std::vector<bool> ordering{kSortAscending, kSortDescending, kSortAscending}; |
| std::vector<bool> null_ordering{kSortNullFirst, kSortNullLast, kSortNullLast}; |
| |
| createRuns(kMixedNullFLLCol1Col2Col3Comparator, true); |
| mergeRuns(attrs, move(ordering), move(null_ordering), RunMergerTest::kTopK); |
| checkOutputRun(kMixedNullFLLCol1Col2Col3Comparator, RunMergerTest::kTopK); |
| checkLastTupleAttrValueIsEqual(null_col1_); |
| checkLastTupleAttrValueIsEqual(null_col2_); |
| checkLastTupleAttrValueIsEqual(null_col3_); |
| } |
| |
| } // namespace merge_run_operator |
| |
| // Test SortMergeRunOperator class. Since RunMerger is tested for various |
| // sort configurations, we only test if merge trees are computed and executed |
| // correctly keeping the sort configuration the same. |
| class SortMergeRunOperatorTest : public ::testing::Test { |
| protected: |
| static const relation_id kTableId = 100; |
| static const relation_id kResultTableId = kTableId + 1; |
| static const relation_id kRunTableId = kTableId + 2; |
| static const char kTableName[]; |
| static const char kResultTableName[]; |
| static const char kRunTableName[]; |
| static const char kStoragePath[]; |
| static const char kDatabaseName[]; |
| |
| virtual void SetUp() { |
| // Initialize the TMB, register this thread as sender and receiver for |
| // appropriate types of messages. |
| bus_.Initialize(); |
| |
| const tmb::client_id worker_thread_client_id = bus_.Connect(); |
| bus_.RegisterClientAsSender(worker_thread_client_id, kCatalogRelationNewBlockMessage); |
| bus_.RegisterClientAsSender(worker_thread_client_id, kDataPipelineMessage); |
| bus_.RegisterClientAsSender(worker_thread_client_id, kWorkOrderFeedbackMessage); |
| |
| thread_id_map_ = ClientIDMap::Instance(); |
| // Usually the worker thread makes the following call. In this test setup, |
| // we don't have a worker thread hence we have to explicitly make the call. |
| thread_id_map_->addValue(worker_thread_client_id); |
| |
| foreman_client_id_ = bus_.Connect(); |
| bus_.RegisterClientAsReceiver(foreman_client_id_, kCatalogRelationNewBlockMessage); |
| bus_.RegisterClientAsReceiver(foreman_client_id_, kDataPipelineMessage); |
| bus_.RegisterClientAsReceiver(foreman_client_id_, kWorkOrderFeedbackMessage); |
| |
| storage_manager_.reset(new StorageManager(kStoragePath)); |
| |
| // Create a database. |
| db_.reset(new CatalogDatabase(nullptr, kDatabaseName)); |
| |
| // Create input_table_, owned by db_. |
| input_table_ = createTable(kTableName, kTableId); |
| db_->addRelation(input_table_); |
| |
| col1_ = input_table_->getAttributeByName("col-1")->getID(); |
| col2_ = input_table_->getAttributeByName("col-2")->getID(); |
| col3_ = input_table_->getAttributeByName("col-3")->getID(); |
| null_col1_ = input_table_->getAttributeByName("null-col-1")->getID(); |
| null_col2_ = input_table_->getAttributeByName("null-col-2")->getID(); |
| null_col3_ = input_table_->getAttributeByName("null-col-3")->getID(); |
| tid_col_ = input_table_->getAttributeByName("tid")->getID(); |
| |
| // Create result_table_, owned by db_. |
| result_table_ = createTable(kResultTableName, kResultTableId); |
| const relation_id result_table_id = db_->addRelation(result_table_); |
| |
| ASSERT_EQ(col1_, result_table_->getAttributeByName("col-1")->getID()); |
| ASSERT_EQ(col2_, result_table_->getAttributeByName("col-2")->getID()); |
| ASSERT_EQ(col3_, result_table_->getAttributeByName("col-3")->getID()); |
| ASSERT_EQ(null_col1_, result_table_->getAttributeByName("null-col-1")->getID()); |
| ASSERT_EQ(null_col2_, result_table_->getAttributeByName("null-col-2")->getID()); |
| ASSERT_EQ(null_col3_, result_table_->getAttributeByName("null-col-3")->getID()); |
| ASSERT_EQ(tid_col_, result_table_->getAttributeByName("tid")->getID()); |
| |
| query_context_proto_.set_query_id(0); // dummy query ID. |
| |
| // Setup the InsertDestination proto in the query context proto. |
| insert_destination_index_ = query_context_proto_.insert_destinations_size(); |
| serialization::InsertDestination *insert_destination_proto = query_context_proto_.add_insert_destinations(); |
| |
| insert_destination_proto->set_insert_destination_type(serialization::InsertDestinationType::BLOCK_POOL); |
| insert_destination_proto->set_relation_id(result_table_id); |
| insert_destination_proto->set_relational_op_index(kOpIndex); |
| |
| // Create run_table_, owned by db_. |
| run_table_ = createTable(kRunTableName, kRunTableId); |
| const relation_id run_table_id = db_->addRelation(run_table_); |
| |
| ASSERT_EQ(col1_, run_table_->getAttributeByName("col-1")->getID()); |
| ASSERT_EQ(col2_, run_table_->getAttributeByName("col-2")->getID()); |
| ASSERT_EQ(col3_, run_table_->getAttributeByName("col-3")->getID()); |
| ASSERT_EQ(null_col1_, run_table_->getAttributeByName("null-col-1")->getID()); |
| ASSERT_EQ(null_col2_, run_table_->getAttributeByName("null-col-2")->getID()); |
| ASSERT_EQ(null_col3_, run_table_->getAttributeByName("null-col-3")->getID()); |
| ASSERT_EQ(tid_col_, run_table_->getAttributeByName("tid")->getID()); |
| |
| run_destination_index_ = query_context_proto_.insert_destinations_size(); |
| insert_destination_proto = query_context_proto_.add_insert_destinations(); |
| |
| insert_destination_proto->set_insert_destination_type(serialization::InsertDestinationType::BLOCK_POOL); |
| insert_destination_proto->set_relation_id(run_table_id); |
| insert_destination_proto->set_relational_op_index(kOpIndex); |
| |
| // Set up the QueryContext. |
| query_context_.reset(new QueryContext(query_context_proto_, |
| *db_, |
| storage_manager_.get(), |
| foreman_client_id_, |
| &bus_)); |
| } |
| |
| virtual void TearDown() { |
| // Usually the worker thread makes the following call. In this test setup, |
| // we don't have a worker thread hence we have to explicitly make the call. |
| thread_id_map_->removeValue(); |
| |
| // Drop blocks from relations. |
| const vector<block_id> input_blocks = input_table_->getBlocksSnapshot(); |
| for (const block_id block : input_blocks) { |
| storage_manager_->deleteBlockOrBlobFile(block); |
| } |
| |
| const vector<block_id> result_blocks = result_table_->getBlocksSnapshot(); |
| for (const block_id block : result_blocks) { |
| storage_manager_->deleteBlockOrBlobFile(block); |
| } |
| |
| const vector<block_id> run_blocks = run_table_->getBlocksSnapshot(); |
| for (const block_id block : run_blocks) { |
| storage_manager_->deleteBlockOrBlobFile(block); |
| } |
| } |
| |
| CatalogRelation *createTable(const char *name, const relation_id rel_id) { |
| CatalogRelation *table = new CatalogRelation(nullptr, name, rel_id); |
| const Type &int_type = IntType::InstanceNonNullable(); |
| const Type &null_int_type = IntType::InstanceNullable(); |
| table->addAttribute(new CatalogAttribute(table, "col-1", int_type)); |
| table->addAttribute(new CatalogAttribute(table, "col-2", int_type)); |
| table->addAttribute(new CatalogAttribute(table, "col-3", int_type)); |
| table->addAttribute(new CatalogAttribute(table, "null-col-1", null_int_type)); |
| table->addAttribute(new CatalogAttribute(table, "null-col-2", null_int_type)); |
| table->addAttribute(new CatalogAttribute(table, "null-col-3", null_int_type)); |
| table->addAttribute(new CatalogAttribute(table, "tid", int_type)); |
| table->setDefaultStorageBlockLayout(StorageBlockLayout::GenerateDefaultLayout(*table, false)); |
| return table; |
| } |
| |
| // Helper method to create tuple. |
| Tuple *createTuple(tuple_id id) { |
| std::vector<TypedValue> values; |
| TestTuple tuple(id); |
| values.emplace_back(static_cast<int>(tuple.col1_)); |
| values.emplace_back(static_cast<int>(tuple.col2_)); |
| values.emplace_back(static_cast<int>(tuple.col3_)); |
| if (!tuple.col1_) { |
| values.emplace_back(kInt); |
| } else { |
| values.emplace_back(static_cast<int>(tuple.col1_)); |
| } |
| if (!tuple.col2_) { |
| values.emplace_back(kInt); |
| } else { |
| values.emplace_back(static_cast<int>(tuple.col2_)); |
| } |
| if (!tuple.col3_) { |
| values.emplace_back(kInt); |
| } else { |
| values.emplace_back(static_cast<int>(tuple.col3_)); |
| } |
| values.emplace_back(static_cast<int>(tuple.tid_)); |
| return new Tuple(std::move(values)); |
| } |
| |
| // Create (sorted) blocks for input to SortMergeRunOperator. |
| template <typename ComparatorT> |
| void createBlocks(const std::size_t num_tuples_per_block, |
| const std::size_t num_blocks, |
| const ComparatorT &comparator, |
| const std::size_t top_k = 0) { |
| std::random_device rd; |
| std::mt19937 gen(rd()); |
| std::uniform_int_distribution<> dist(0, (1 << 8) - 1); |
| std::unique_ptr<Tuple> tuple; |
| std::vector<TestTuple> top_k_tuples; |
| MutableBlockReference storage_block; |
| |
| auto sort_comparator = |
| [this, comparator](const TestTuple &l, const TestTuple &r) -> bool { |
| std::unique_ptr<Tuple> left(createTuple(l.tid_)), |
| right(createTuple(r.tid_)); |
| return comparator(*left, *right); |
| }; |
| |
| expect_num_tuples_ = 0; |
| for (std::size_t bid = 0; bid < num_blocks; ++bid) { |
| // Generate tuples. |
| std::vector<TestTuple> tuples; |
| for (std::size_t i = 0; i < num_tuples_per_block; ++i) { |
| tuples.emplace_back(dist(gen)); |
| } |
| |
| // Sort tuples before creating the run. |
| std::sort(tuples.begin(), tuples.end(), sort_comparator); |
| |
| // Maintain top-k. |
| if (top_k > 0) { |
| top_k_tuples.insert(top_k_tuples.end(), |
| tuples.begin(), |
| tuples.begin() + std::min(tuples.size(), top_k)); |
| std::partial_sort( |
| top_k_tuples.begin(), |
| top_k_tuples.begin() + std::min(top_k, top_k_tuples.size()), |
| top_k_tuples.end(), |
| sort_comparator); |
| if (top_k_tuples.size() > top_k) { |
| top_k_tuples.erase(top_k_tuples.begin() + top_k, top_k_tuples.end()); |
| } |
| } |
| |
| // Create block. |
| block_id block_id = storage_manager_->createBlock(*input_table_, input_table_->getDefaultStorageBlockLayout()); |
| storage_block = storage_manager_->getBlockMutable(block_id, *input_table_); |
| input_table_->addBlock(block_id); |
| |
| // Insert sorted tuples. |
| for (std::size_t i = 0; i < num_tuples_per_block; ++i) { |
| tuple.reset(createTuple(tuples[i].tid_)); |
| EXPECT_TRUE(storage_block->insertTupleInBatch(*tuple)); |
| } |
| storage_block->rebuild(); |
| expect_num_tuples_ += num_tuples_per_block; |
| } |
| |
| if (top_k > 0) { |
| expect_num_tuples_ = top_k; |
| last_expected_tuple_.reset(createTuple(top_k_tuples.back().tid_)); |
| } |
| } |
| |
| // Debug print method. |
| void printTuple(const char *prefix, const Tuple &tuple) { |
| if (VLOG_IS_ON(4)) { |
| std::ostringstream out; |
| out << prefix << ' '; |
| CatalogRelation::const_iterator attr_it = input_table_->begin(); |
| Tuple::const_iterator value_it = tuple.begin(); |
| for (; attr_it != input_table_->end(); ++attr_it, ++value_it) { |
| if (value_it->isNull()) { |
| out << "NULL"; |
| } else { |
| out << attr_it->getType().printValueToString(*value_it); |
| } |
| out << '|'; |
| } |
| out << '\n'; |
| VLOG(4) << out.str(); |
| } |
| } |
| |
| void processMessages(const std::size_t num_expected_feedback_messages = 0) { |
| AnnotatedMessage msg; |
| std::size_t num_receieved_feedback_messages = 0; |
| do { |
| if (bus_.ReceiveIfAvailable(foreman_client_id_, &msg)) { |
| const TaggedMessage &tagged_message = msg.tagged_message; |
| switch (tagged_message.message_type()) { |
| case kWorkOrderFeedbackMessage: { |
| // Dispatch feedback messages to SortMergeRunOperator. |
| WorkOrder::FeedbackMessage feedback_msg( |
| const_cast<void *>(tagged_message.message()), |
| tagged_message.message_bytes()); |
| EXPECT_EQ(kOpIndex, feedback_msg.header().rel_op_index); |
| merge_op_->receiveFeedbackMessage(feedback_msg); |
| ++num_receieved_feedback_messages; |
| break; |
| } |
| case kCatalogRelationNewBlockMessage: { |
| serialization::CatalogRelationNewBlockMessage proto; |
| CHECK(proto.ParseFromArray(tagged_message.message(), tagged_message.message_bytes())); |
| |
| CatalogRelation *relation = db_->getRelationByIdMutable(proto.relation_id()); |
| relation->addBlock(proto.block_id()); |
| break; |
| } |
| default: |
| // It is safe to discard other kinds of messages (e.g. pipeline) in |
| // this funtion. |
| break; |
| } |
| } |
| } while (num_receieved_feedback_messages < num_expected_feedback_messages); |
| } |
| |
| void executeOperatorUntilDone() { |
| bool done; |
| WorkOrdersContainer container(kOpIndex + 1, 0); |
| do { |
| done = merge_op_->getAllWorkOrders(&container, |
| query_context_.get(), |
| storage_manager_.get(), |
| foreman_client_id_, |
| &bus_); |
| while (container.hasNormalWorkOrder(kOpIndex)) { |
| std::unique_ptr<WorkOrder> order(container.getNormalWorkOrder(kOpIndex)); |
| order->execute(); |
| processMessages(1); |
| } |
| } while (!done); |
| } |
| |
| bool executeOperator() { |
| bool done = false; |
| WorkOrdersContainer container(kOpIndex + 1, 0); |
| bool executed; |
| do { |
| if (!done) { |
| // Find work orders to execute, if not done already. |
| done = merge_op_->getAllWorkOrders(&container, |
| query_context_.get(), |
| storage_manager_.get(), |
| foreman_client_id_, |
| &bus_); |
| } |
| |
| executed = false; |
| if (container.hasNormalWorkOrder(kOpIndex)) { |
| std::unique_ptr<WorkOrder> order(container.getNormalWorkOrder(kOpIndex)); |
| order->execute(); |
| processMessages(1); |
| executed = true; |
| } |
| } while (container.hasNormalWorkOrder(kOpIndex) || executed); |
| return done; |
| } |
| |
| // Send doneFeedingInputBlocks callback. |
| void feedingDone(RelationalOperator *op) { |
| SortMergeRunOperator *merge_op = static_cast<SortMergeRunOperator *>(op); |
| merge_op->doneFeedingInputBlocks(input_table_->getID()); |
| } |
| |
| QueryContext::sort_config_id createSortConfigProto(const std::vector<attribute_id> &attrs, |
| const std::vector<bool> &ordering, |
| const std::vector<bool> &null_ordering) { |
| // Setup the SortConfiguration proto. |
| DCHECK_EQ(attrs.size(), ordering.size()); |
| DCHECK_EQ(attrs.size(), null_ordering.size()); |
| |
| const QueryContext::sort_config_id sort_config_index = query_context_proto_.sort_configs_size(); |
| serialization::SortConfiguration *sort_config_proto = query_context_proto_.add_sort_configs(); |
| |
| for (std::size_t i = 0; i < attrs.size(); ++i) { |
| serialization::SortConfiguration::OrderBy *order_by_proto = sort_config_proto->add_order_by_list(); |
| |
| unique_ptr<Scalar> scalar( |
| new ScalarAttribute(*input_table_->getAttributeById(attrs[i]))); |
| order_by_proto->mutable_expression()->CopyFrom(scalar->getProto()); |
| order_by_proto->set_is_ascending(ordering[i]); |
| order_by_proto->set_null_first(null_ordering[i]); |
| } |
| |
| return sort_config_index; |
| } |
| |
| // Execute in pipeline breaking mode. |
| void executeNotPipelined(const std::vector<attribute_id> &attrs, |
| const std::vector<bool> &ordering, |
| const std::vector<bool> &null_ordering, |
| const std::uint32_t merge_factor, |
| const std::size_t top_k = 0) { |
| const QueryContext::sort_config_id sort_config_index = createSortConfigProto(attrs, ordering, null_ordering); |
| |
| merge_op_.reset(new SortMergeRunOperator(kQueryId, |
| *input_table_, |
| *result_table_, |
| insert_destination_index_, |
| *run_table_, |
| run_destination_index_, |
| sort_config_index, |
| merge_factor, |
| top_k, |
| true)); |
| |
| merge_op_->setOperatorIndex(kOpIndex); |
| |
| // Set up the QueryContext. |
| query_context_.reset(new QueryContext(query_context_proto_, |
| *db_, |
| storage_manager_.get(), |
| foreman_client_id_, |
| &bus_)); |
| |
| executeOperatorUntilDone(); |
| } |
| |
| void feedBlocks(const std::size_t num_blocks, std::vector<block_id> *blocks) { |
| std::size_t count = std::min(blocks->size(), num_blocks); |
| std::vector<block_id> to_feed(blocks->begin() + blocks->size() - count, |
| blocks->end()); |
| |
| // Feed blocks. |
| DVLOG(1) << "Feeding " << to_feed.size() << " blocks."; |
| for (const block_id block : to_feed) { |
| merge_op_->feedInputBlock(block, input_table_->getID(), 0 /* partition_id */); |
| } |
| |
| // Remove fed blocks. |
| blocks->erase(blocks->begin() + blocks->size() - count, blocks->end()); |
| } |
| |
| // Execute in pipelined mode. |
| void executePipelined(const std::vector<attribute_id> &attrs, |
| const std::vector<bool> &ordering, |
| const std::vector<bool> &null_ordering, |
| const std::uint32_t merge_factor, |
| const std::size_t num_blocks_to_feed_per_iteration, |
| const std::size_t top_k = 0) { |
| const QueryContext::sort_config_id sort_config_index = createSortConfigProto(attrs, ordering, null_ordering); |
| |
| merge_op_.reset(new SortMergeRunOperator(kQueryId, |
| *input_table_, |
| *result_table_, |
| insert_destination_index_, |
| *run_table_, |
| run_destination_index_, |
| sort_config_index, |
| merge_factor, |
| top_k, |
| false)); |
| merge_op_->setOperatorIndex(kOpIndex); |
| |
| // Set up the QueryContext. |
| query_context_.reset(new QueryContext(query_context_proto_, |
| *db_, |
| storage_manager_.get(), |
| foreman_client_id_, |
| &bus_)); |
| |
| std::vector<block_id> blocks = input_table_->getBlocksSnapshot(); |
| while (!blocks.empty()) { |
| feedBlocks(num_blocks_to_feed_per_iteration, &blocks); |
| EXPECT_FALSE(executeOperator()); |
| |
| if (blocks.empty()) { |
| merge_op_->doneFeedingInputBlocks(input_table_->getID()); |
| } |
| |
| // Expect the operator to finish execution with 'done' response when block |
| // feedDone() is signalled. |
| EXPECT_EQ(blocks.empty(), executeOperator()); |
| } |
| } |
| |
| // Check if the merged run is sorted. |
| template <typename ComparatorT> |
| void checkOutputRun(const ComparatorT comparator) { |
| std::vector<block_id> sorted_blocks = result_table_->getBlocksSnapshot(); |
| std::unique_ptr<Tuple> prev, current; |
| std::unique_ptr<ValueAccessor> accessor; |
| |
| ASSERT_GT(sorted_blocks.size(), 0u); |
| tuple_id num_tuples = 0; |
| for (const block_id &id : sorted_blocks) { |
| BlockReference block(storage_manager_->getBlock(id, *result_table_)); |
| accessor.reset(block->getTupleStorageSubBlock().createValueAccessor()); |
| InvokeOnValueAccessorNotAdapter( |
| accessor.get(), |
| [&](auto *accessor) -> void { // NOLINT(build/c++11) |
| while (accessor->next()) { |
| prev = std::move(current); |
| current.reset(accessor->getTuple()); |
| ++num_tuples; |
| if (prev) { |
| EXPECT_FALSE(comparator(*current, *prev)); |
| } |
| this->printTuple(">", *current); |
| } |
| }); |
| } |
| |
| last_actual_tuple_ = std::move(current); |
| EXPECT_LT(0, expect_num_tuples_); |
| EXPECT_EQ(expect_num_tuples_, num_tuples); |
| } |
| |
| void checkLastTupleAttrValueIsEqual(const attribute_id id) { |
| printTuple("Expected:", *last_expected_tuple_); |
| printTuple("Actual: ", *last_actual_tuple_); |
| EXPECT_EQ(last_expected_tuple_->getAttributeValue(id).isNull(), |
| last_actual_tuple_->getAttributeValue(id).isNull()); |
| if (!last_expected_tuple_->getAttributeValue(id).isNull() && |
| !last_actual_tuple_->getAttributeValue(id).isNull()) { |
| EXPECT_TRUE(last_expected_tuple_->getAttributeValue(id).fastEqualCheck( |
| last_actual_tuple_->getAttributeValue(id))); |
| } |
| } |
| |
| std::unique_ptr<StorageManager> storage_manager_; |
| |
| serialization::QueryContext query_context_proto_; |
| std::unique_ptr<QueryContext> query_context_; |
| std::unique_ptr<CatalogDatabase> db_; |
| CatalogRelation *input_table_, *result_table_, *run_table_; |
| |
| QueryContext::insert_destination_id insert_destination_index_, run_destination_index_; |
| |
| std::unique_ptr<RelationalOperator> merge_op_; |
| |
| attribute_id col1_; |
| attribute_id col2_; |
| attribute_id col3_; |
| attribute_id null_col1_; |
| attribute_id null_col2_; |
| attribute_id null_col3_; |
| attribute_id tid_col_; |
| tuple_id expect_num_tuples_; |
| std::unique_ptr<Tuple> last_expected_tuple_; |
| std::unique_ptr<Tuple> last_actual_tuple_; |
| |
| MessageBusImpl bus_; |
| tmb::client_id foreman_client_id_; |
| |
| // This map is needed for InsertDestination and some operators that send |
| // messages to Foreman directly. To know the reason behind the design of this |
| // map, see the note in InsertDestination.hpp. |
| ClientIDMap *thread_id_map_; |
| }; |
| |
| const char SortMergeRunOperatorTest::kTableName[] = "table"; |
| const char SortMergeRunOperatorTest::kResultTableName[] = "result-table"; |
| const char SortMergeRunOperatorTest::kRunTableName[] = "run-table"; |
| const char SortMergeRunOperatorTest::kStoragePath[] = "./sort_merge_run_operator_test_data"; |
| const char SortMergeRunOperatorTest::kDatabaseName[] = "database"; |
| |
| namespace { |
| |
| static const std::size_t kNumTuplesPerBlock = 100; |
| |
| auto kDefaultSortComparator = |
| [](const Tuple &left, const Tuple &right) -> bool { |
| TestTupleAttrs l = TupleToTupleAttr(left); |
| TestTupleAttrs r = TupleToTupleAttr(right); |
| if (l.c1 < r.c1) { |
| return true; |
| } |
| return false; |
| }; |
| |
| std::vector<bool> kDefaultSortOrdering{kSortAscending}; |
| std::vector<bool> kDefaultSortNullOrdering{kSortNullLast}; |
| |
| } // namespace |
| |
| TEST_F(SortMergeRunOperatorTest, NotPipelined_SinglePass_SameRunsAsMergeFactor) { |
| const std::size_t merge_factor = 8; |
| const std::size_t num_blocks = merge_factor; |
| std::vector<attribute_id> kDefaultSortAttributes{col1_}; |
| |
| createBlocks(kNumTuplesPerBlock, num_blocks, kDefaultSortComparator); |
| executeNotPipelined(kDefaultSortAttributes, kDefaultSortOrdering, kDefaultSortNullOrdering, merge_factor); |
| checkOutputRun(kDefaultSortComparator); |
| } |
| |
| TEST_F(SortMergeRunOperatorTest, NotPipelined_SinglePass_SameRunsAsMergeFactor_TopK) { |
| const std::size_t merge_factor = 8; |
| const std::size_t num_blocks = merge_factor; |
| std::vector<attribute_id> kDefaultSortAttributes{col1_}; |
| std::size_t top_k = 20; |
| |
| createBlocks(kNumTuplesPerBlock, num_blocks, kDefaultSortComparator, top_k); |
| executeNotPipelined(kDefaultSortAttributes, kDefaultSortOrdering, kDefaultSortNullOrdering, merge_factor, top_k); |
| checkOutputRun(kDefaultSortComparator); |
| checkLastTupleAttrValueIsEqual(col1_); |
| } |
| |
| TEST_F(SortMergeRunOperatorTest, NotPipelined_SinglePass_LessRunsThanMergeFactor) { |
| const std::size_t merge_factor = 8; |
| const std::size_t num_blocks = merge_factor / 2; |
| std::vector<attribute_id> kDefaultSortAttributes{col1_}; |
| |
| createBlocks(kNumTuplesPerBlock, num_blocks, kDefaultSortComparator); |
| executeNotPipelined(kDefaultSortAttributes, kDefaultSortOrdering, kDefaultSortNullOrdering, merge_factor); |
| checkOutputRun(kDefaultSortComparator); |
| } |
| |
| TEST_F(SortMergeRunOperatorTest, NotPipelined_SinglePass_OneRun) { |
| const std::size_t merge_factor = 8; |
| const std::size_t num_blocks = 1; |
| std::vector<attribute_id> kDefaultSortAttributes{col1_}; |
| |
| createBlocks(kNumTuplesPerBlock, num_blocks, kDefaultSortComparator); |
| executeNotPipelined(kDefaultSortAttributes, kDefaultSortOrdering, kDefaultSortNullOrdering, merge_factor); |
| checkOutputRun(kDefaultSortComparator); |
| } |
| |
| TEST_F(SortMergeRunOperatorTest, NotPipelined_TwoPass_CompleteMerges) { |
| // Should cause: |
| // [3, 3, 3]-way merges in first pass. |
| // [3]-way merges in second pass. |
| const std::size_t merge_factor = 3; |
| const std::size_t num_blocks = merge_factor * merge_factor; |
| std::vector<attribute_id> kDefaultSortAttributes{col1_}; |
| |
| createBlocks(kNumTuplesPerBlock, num_blocks, kDefaultSortComparator); |
| executeNotPipelined(kDefaultSortAttributes, kDefaultSortOrdering, kDefaultSortNullOrdering, merge_factor); |
| checkOutputRun(kDefaultSortComparator); |
| } |
| |
| TEST_F(SortMergeRunOperatorTest, NotPipelined_TwoPass_PartialFirstPass) { |
| // Should cause: |
| // [3, 3, 2]-way merges in first pass. |
| // [3]-way merges in second pass. |
| const std::size_t merge_factor = 3; |
| const std::size_t num_blocks = merge_factor * merge_factor - 1; |
| std::vector<attribute_id> kDefaultSortAttributes{col1_}; |
| |
| createBlocks(kNumTuplesPerBlock, num_blocks, kDefaultSortComparator); |
| executeNotPipelined(kDefaultSortAttributes, kDefaultSortOrdering, kDefaultSortNullOrdering, merge_factor); |
| checkOutputRun(kDefaultSortComparator); |
| } |
| |
| TEST_F(SortMergeRunOperatorTest, NotPipelined_TwoPass_PartialBothPasses) { |
| // Should cause: |
| // [3, 2]-way merge in first pass. |
| // 2-way merge in second pass. |
| const std::size_t merge_factor = 3; |
| const std::size_t num_blocks = merge_factor + 2; |
| std::vector<attribute_id> kDefaultSortAttributes{col1_}; |
| |
| createBlocks(kNumTuplesPerBlock, num_blocks, kDefaultSortComparator); |
| executeNotPipelined(kDefaultSortAttributes, kDefaultSortOrdering, kDefaultSortNullOrdering, merge_factor); |
| checkOutputRun(kDefaultSortComparator); |
| } |
| |
| TEST_F(SortMergeRunOperatorTest, NotPipelined_ThreePass_CompleteMerges) { |
| // Should cause: |
| // [3, 3, 3, 3, 3, 3, 3, 3, 3]-way merges in first pass. |
| // [3, 3, 3]-way merges in second pass. |
| // [3]-way merges in third pass. |
| const std::size_t merge_factor = 3; |
| const std::size_t num_blocks = merge_factor * merge_factor * merge_factor; |
| std::vector<attribute_id> kDefaultSortAttributes{col1_}; |
| |
| createBlocks(kNumTuplesPerBlock, num_blocks, kDefaultSortComparator); |
| executeNotPipelined(kDefaultSortAttributes, kDefaultSortOrdering, kDefaultSortNullOrdering, merge_factor); |
| checkOutputRun(kDefaultSortComparator); |
| } |
| |
| TEST_F(SortMergeRunOperatorTest, NotPipelined_ThreePass_CompleteMerges_TopK) { |
| // Should cause: |
| // [3, 3, 3, 3, 3, 3, 3, 3, 3]-way merges in first pass. |
| // [3, 3, 3]-way merges in second pass. |
| // [3]-way merges in third pass. |
| const std::size_t merge_factor = 3; |
| const std::size_t num_blocks = merge_factor * merge_factor * merge_factor; |
| std::vector<attribute_id> kDefaultSortAttributes{col1_}; |
| const std::size_t top_k = 33; |
| |
| createBlocks(kNumTuplesPerBlock, num_blocks, kDefaultSortComparator, top_k); |
| executeNotPipelined(kDefaultSortAttributes, kDefaultSortOrdering, kDefaultSortNullOrdering, merge_factor, top_k); |
| checkOutputRun(kDefaultSortComparator); |
| checkLastTupleAttrValueIsEqual(col1_); |
| } |
| |
| TEST_F(SortMergeRunOperatorTest, NotPipelined_ThreePass_PartialMergesInFirstPass) { |
| // Should cause: |
| // [3, 3, 3, 3, 3, 3, 3, 3, 1]-way merges in first pass. |
| // [3, 3, 3]-way merges in second pass. |
| // [3]-way merges in third pass. |
| const std::size_t merge_factor = 3; |
| const std::size_t num_blocks = merge_factor * merge_factor * merge_factor - 2; |
| std::vector<attribute_id> kDefaultSortAttributes{col1_}; |
| |
| createBlocks(kNumTuplesPerBlock, num_blocks, kDefaultSortComparator); |
| executeNotPipelined(kDefaultSortAttributes, kDefaultSortOrdering, kDefaultSortNullOrdering, merge_factor); |
| checkOutputRun(kDefaultSortComparator); |
| } |
| |
| TEST_F(SortMergeRunOperatorTest, NotPipelined_ThreePass_PartialMergesInFirstSecondPass) { |
| // Should cause: |
| // [3, 3, 3, 3, 3, 3, 3, 2]-way merges in first pass. |
| // [3, 3, 2]-way merges in second pass. |
| // [3]-way merges in third pass. |
| const std::size_t merge_factor = 3; |
| const std::size_t num_blocks = merge_factor * merge_factor * merge_factor - 4; |
| std::vector<attribute_id> kDefaultSortAttributes{col1_}; |
| |
| createBlocks(kNumTuplesPerBlock, num_blocks, kDefaultSortComparator); |
| executeNotPipelined(kDefaultSortAttributes, kDefaultSortOrdering, kDefaultSortNullOrdering, merge_factor); |
| checkOutputRun(kDefaultSortComparator); |
| } |
| |
| TEST_F(SortMergeRunOperatorTest, NotPipelined_ThreePass_PartialMergesInAllPasses) { |
| // Should cause: |
| // [3, 3, 3, 3, 2]-way merges in first pass. |
| // [3, 2]-way merges in second pass. |
| // [2]-way merges in third pass. |
| const std::size_t merge_factor = 3; |
| const std::size_t num_blocks = 2 * merge_factor * merge_factor - 4; |
| std::vector<attribute_id> kDefaultSortAttributes{col1_}; |
| |
| createBlocks(kNumTuplesPerBlock, num_blocks, kDefaultSortComparator); |
| executeNotPipelined(kDefaultSortAttributes, kDefaultSortOrdering, kDefaultSortNullOrdering, merge_factor); |
| checkOutputRun(kDefaultSortComparator); |
| } |
| |
| TEST_F(SortMergeRunOperatorTest, NotPipelined_ManyPasses_MergeFactor7) { |
| const std::size_t merge_factor = 7; |
| const std::size_t num_blocks = 509; |
| std::vector<attribute_id> kDefaultSortAttributes{col1_}; |
| |
| createBlocks(kNumTuplesPerBlock, num_blocks, kDefaultSortComparator); |
| executeNotPipelined(kDefaultSortAttributes, kDefaultSortOrdering, kDefaultSortNullOrdering, merge_factor); |
| checkOutputRun(kDefaultSortComparator); |
| } |
| |
| TEST_F(SortMergeRunOperatorTest, NotPipelined_ManyPasses_MergeFactor7_TopK) { |
| const std::size_t merge_factor = 7; |
| const std::size_t num_blocks = 509; |
| std::vector<attribute_id> kDefaultSortAttributes{col1_}; |
| const std::size_t top_k = 17; |
| |
| createBlocks(kNumTuplesPerBlock, num_blocks, kDefaultSortComparator, top_k); |
| executeNotPipelined(kDefaultSortAttributes, kDefaultSortOrdering, kDefaultSortNullOrdering, merge_factor, top_k); |
| checkOutputRun(kDefaultSortComparator); |
| checkLastTupleAttrValueIsEqual(col1_); |
| } |
| |
| TEST_F(SortMergeRunOperatorTest, NotPipelined_ManyPasses_MergeFactor17) { |
| const std::size_t merge_factor = 17; |
| const std::size_t num_blocks = 509; |
| std::vector<attribute_id> kDefaultSortAttributes{col1_}; |
| |
| createBlocks(kNumTuplesPerBlock, num_blocks, kDefaultSortComparator); |
| executeNotPipelined(kDefaultSortAttributes, kDefaultSortOrdering, kDefaultSortNullOrdering, merge_factor); |
| checkOutputRun(kDefaultSortComparator); |
| } |
| |
| TEST_F(SortMergeRunOperatorTest, NotPipelined_ManyPasses_MergeFactor17_TopK) { |
| const std::size_t merge_factor = 17; |
| const std::size_t num_blocks = 509; |
| std::vector<attribute_id> kDefaultSortAttributes{col1_}; |
| const std::size_t top_k = 53; |
| |
| createBlocks(kNumTuplesPerBlock, num_blocks, kDefaultSortComparator, top_k); |
| executeNotPipelined(kDefaultSortAttributes, kDefaultSortOrdering, kDefaultSortNullOrdering, merge_factor, top_k); |
| checkOutputRun(kDefaultSortComparator); |
| checkLastTupleAttrValueIsEqual(col1_); |
| } |
| |
| TEST_F(SortMergeRunOperatorTest, Pipleined_SinglePass_LessRunsThanMergeFactor_AllFeed) { |
| const std::size_t merge_factor = 8; |
| const std::size_t num_blocks = merge_factor / 2; |
| std::vector<attribute_id> kDefaultSortAttributes{col1_}; |
| |
| createBlocks(kNumTuplesPerBlock, num_blocks, kDefaultSortComparator); |
| executePipelined(kDefaultSortAttributes, kDefaultSortOrdering, kDefaultSortNullOrdering, merge_factor, num_blocks); |
| checkOutputRun(kDefaultSortComparator); |
| } |
| |
| TEST_F(SortMergeRunOperatorTest, Pipleined_SinglePass_LessRunsThanMergeFactor_SlowFeed) { |
| const std::size_t merge_factor = 16; |
| const std::size_t num_blocks = merge_factor - 1; |
| std::vector<attribute_id> kDefaultSortAttributes{col1_}; |
| |
| createBlocks(kNumTuplesPerBlock, num_blocks, kDefaultSortComparator); |
| executePipelined(kDefaultSortAttributes, kDefaultSortOrdering, kDefaultSortNullOrdering, merge_factor, 1); |
| checkOutputRun(kDefaultSortComparator); |
| } |
| |
| TEST_F(SortMergeRunOperatorTest, Pipleined_SinglePass_SameRunsAsMergeFactor_AllFeed) { |
| const std::size_t merge_factor = 8; |
| const std::size_t num_blocks = merge_factor; |
| std::vector<attribute_id> kDefaultSortAttributes{col1_}; |
| |
| createBlocks(kNumTuplesPerBlock, num_blocks, kDefaultSortComparator); |
| executePipelined(kDefaultSortAttributes, kDefaultSortOrdering, kDefaultSortNullOrdering, merge_factor, num_blocks); |
| checkOutputRun(kDefaultSortComparator); |
| } |
| |
| TEST_F(SortMergeRunOperatorTest, Pipleined_SinglePass_SameRunsAsMergeFactor_AllFeed_TopK) { |
| const std::size_t merge_factor = 8; |
| const std::size_t num_blocks = merge_factor; |
| std::vector<attribute_id> kDefaultSortAttributes{col1_}; |
| const std::size_t top_k = 13; |
| |
| createBlocks(kNumTuplesPerBlock, num_blocks, kDefaultSortComparator, top_k); |
| executePipelined( |
| kDefaultSortAttributes, kDefaultSortOrdering, kDefaultSortNullOrdering, merge_factor, num_blocks, top_k); |
| checkOutputRun(kDefaultSortComparator); |
| checkLastTupleAttrValueIsEqual(col1_); |
| } |
| |
| TEST_F(SortMergeRunOperatorTest, Pipleined_SinglePass_SameRunsAsMergeFactor_SlowFeed) { |
| const std::size_t merge_factor = 8; |
| const std::size_t num_blocks = merge_factor; |
| std::vector<attribute_id> kDefaultSortAttributes{col1_}; |
| |
| createBlocks(kNumTuplesPerBlock, num_blocks, kDefaultSortComparator); |
| executePipelined(kDefaultSortAttributes, kDefaultSortOrdering, kDefaultSortNullOrdering, merge_factor, 1); |
| checkOutputRun(kDefaultSortComparator); |
| } |
| |
| TEST_F(SortMergeRunOperatorTest, Pipleined_SinglePass_OneRun) { |
| const std::size_t merge_factor = 8; |
| const std::size_t num_blocks = 1; |
| std::vector<attribute_id> kDefaultSortAttributes{col1_}; |
| |
| createBlocks(kNumTuplesPerBlock, num_blocks, kDefaultSortComparator); |
| executePipelined(kDefaultSortAttributes, kDefaultSortOrdering, kDefaultSortNullOrdering, merge_factor, num_blocks); |
| checkOutputRun(kDefaultSortComparator); |
| } |
| |
| TEST_F(SortMergeRunOperatorTest, Pipleined_TwoPass_CompleteMerges_AllFeed) { |
| // Should cause: |
| // [3, 3, 3]-way merges in first pass. |
| // [3]-way merges in second pass. |
| const std::size_t merge_factor = 3; |
| const std::size_t num_blocks = merge_factor * merge_factor; |
| std::vector<attribute_id> kDefaultSortAttributes{col1_}; |
| |
| createBlocks(kNumTuplesPerBlock, num_blocks, kDefaultSortComparator); |
| executePipelined(kDefaultSortAttributes, kDefaultSortOrdering, kDefaultSortNullOrdering, merge_factor, num_blocks); |
| checkOutputRun(kDefaultSortComparator); |
| } |
| |
| TEST_F(SortMergeRunOperatorTest, Pipleined_TwoPass_CompleteMerges_SlowFeed) { |
| // Should cause: |
| // [3, 3, 3]-way merges in first pass. |
| // [3]-way merges in second pass. |
| const std::size_t merge_factor = 3; |
| const std::size_t num_blocks = merge_factor * merge_factor; |
| std::vector<attribute_id> kDefaultSortAttributes{col1_}; |
| |
| createBlocks(kNumTuplesPerBlock, num_blocks, kDefaultSortComparator); |
| executePipelined(kDefaultSortAttributes, kDefaultSortOrdering, kDefaultSortNullOrdering, merge_factor, 1); |
| checkOutputRun(kDefaultSortComparator); |
| } |
| |
| TEST_F(SortMergeRunOperatorTest, Pipleined_TwoPass_PartialFirstPass_AllFeed) { |
| // Should cause: |
| // [3, 3, 2]-way merges in first pass. |
| // [3]-way merges in second pass. |
| const std::size_t merge_factor = 3; |
| const std::size_t num_blocks = merge_factor * merge_factor - 1; |
| std::vector<attribute_id> kDefaultSortAttributes{col1_}; |
| |
| createBlocks(kNumTuplesPerBlock, num_blocks, kDefaultSortComparator); |
| executePipelined(kDefaultSortAttributes, kDefaultSortOrdering, kDefaultSortNullOrdering, merge_factor, num_blocks); |
| checkOutputRun(kDefaultSortComparator); |
| } |
| |
| TEST_F(SortMergeRunOperatorTest, Pipleined_TwoPass_PartialFirstPass_SlowFeed) { |
| // Should cause: |
| // [3, 3, 2]-way merges in first pass. |
| // [3]-way merges in second pass. |
| const std::size_t merge_factor = 3; |
| const std::size_t num_blocks = merge_factor * merge_factor - 1; |
| std::vector<attribute_id> kDefaultSortAttributes{col1_}; |
| |
| createBlocks(kNumTuplesPerBlock, num_blocks, kDefaultSortComparator); |
| executePipelined(kDefaultSortAttributes, kDefaultSortOrdering, kDefaultSortNullOrdering, merge_factor, 1); |
| checkOutputRun(kDefaultSortComparator); |
| } |
| |
| TEST_F(SortMergeRunOperatorTest, Pipleined_TwoPass_PartialBothPasses_AllFeed) { |
| // Should cause: |
| // [3, 2]-way merge in first pass. |
| // 2-way merge in second pass. |
| const std::size_t merge_factor = 3; |
| const std::size_t num_blocks = merge_factor + 2; |
| std::vector<attribute_id> kDefaultSortAttributes{col1_}; |
| |
| createBlocks(kNumTuplesPerBlock, num_blocks, kDefaultSortComparator); |
| executePipelined(kDefaultSortAttributes, kDefaultSortOrdering, kDefaultSortNullOrdering, merge_factor, num_blocks); |
| checkOutputRun(kDefaultSortComparator); |
| } |
| |
| TEST_F(SortMergeRunOperatorTest, Pipleined_TwoPass_PartialBothPasses_SlowFeed) { |
| // Should cause: |
| // [3, 2]-way merge in first pass. |
| // 2-way merge in second pass. |
| const std::size_t merge_factor = 3; |
| const std::size_t num_blocks = merge_factor + 2; |
| std::vector<attribute_id> kDefaultSortAttributes{col1_}; |
| |
| createBlocks(kNumTuplesPerBlock, num_blocks, kDefaultSortComparator); |
| executePipelined(kDefaultSortAttributes, kDefaultSortOrdering, kDefaultSortNullOrdering, merge_factor, 1); |
| checkOutputRun(kDefaultSortComparator); |
| } |
| |
| TEST_F(SortMergeRunOperatorTest, Pipleined_ThreePass_CompleteMerges_AllFeed) { |
| // Should cause: |
| // [3, 3, 3, 3, 3, 3, 3, 3, 3]-way merges in first pass. |
| // [3, 3, 3]-way merges in second pass. |
| // [3]-way merges in third pass. |
| const std::size_t merge_factor = 3; |
| const std::size_t num_blocks = merge_factor * merge_factor * merge_factor; |
| std::vector<attribute_id> kDefaultSortAttributes{col1_}; |
| |
| createBlocks(kNumTuplesPerBlock, num_blocks, kDefaultSortComparator); |
| executePipelined(kDefaultSortAttributes, kDefaultSortOrdering, kDefaultSortNullOrdering, merge_factor, num_blocks); |
| checkOutputRun(kDefaultSortComparator); |
| } |
| |
| TEST_F(SortMergeRunOperatorTest, Pipleined_ThreePass_CompleteMerges_SlowFeed) { |
| // Should cause: |
| // [3, 3, 3, 3, 3, 3, 3, 3, 3]-way merges in first pass. |
| // [3, 3, 3]-way merges in second pass. |
| // [3]-way merges in third pass. |
| const std::size_t merge_factor = 3; |
| const std::size_t num_blocks = merge_factor * merge_factor * merge_factor; |
| std::vector<attribute_id> kDefaultSortAttributes{col1_}; |
| |
| createBlocks(kNumTuplesPerBlock, num_blocks, kDefaultSortComparator); |
| executePipelined(kDefaultSortAttributes, kDefaultSortOrdering, kDefaultSortNullOrdering, merge_factor, 1); |
| checkOutputRun(kDefaultSortComparator); |
| } |
| |
| TEST_F(SortMergeRunOperatorTest, Pipleined_ThreePass_CompleteMerges_SlowFeed_TopK) { |
| // Should cause: |
| // [3, 3, 3, 3, 3, 3, 3, 3, 3]-way merges in first pass. |
| // [3, 3, 3]-way merges in second pass. |
| // [3]-way merges in third pass. |
| const std::size_t merge_factor = 3; |
| const std::size_t num_blocks = merge_factor * merge_factor * merge_factor; |
| std::vector<attribute_id> kDefaultSortAttributes{col1_}; |
| const std::size_t top_k = 19; |
| |
| createBlocks(kNumTuplesPerBlock, num_blocks, kDefaultSortComparator, top_k); |
| executePipelined(kDefaultSortAttributes, kDefaultSortOrdering, kDefaultSortNullOrdering, merge_factor, 1, top_k); |
| checkOutputRun(kDefaultSortComparator); |
| checkLastTupleAttrValueIsEqual(col1_); |
| } |
| |
| TEST_F(SortMergeRunOperatorTest, Pipleined_ThreePass_PartialMergesInFirstPass_AllFeed) { |
| // Should cause: |
| // [3, 3, 3, 3, 3, 3, 3, 3, 1]-way merges in first pass. |
| // [3, 3, 3]-way merges in second pass. |
| // [3]-way merges in third pass. |
| const std::size_t merge_factor = 3; |
| const std::size_t num_blocks = merge_factor * merge_factor * merge_factor - 2; |
| std::vector<attribute_id> kDefaultSortAttributes{col1_}; |
| |
| createBlocks(kNumTuplesPerBlock, num_blocks, kDefaultSortComparator); |
| executePipelined(kDefaultSortAttributes, kDefaultSortOrdering, kDefaultSortNullOrdering, merge_factor, num_blocks); |
| checkOutputRun(kDefaultSortComparator); |
| } |
| |
| TEST_F(SortMergeRunOperatorTest, Pipleined_ThreePass_PartialMergesInFirstPass_SlowFeed) { |
| // Should cause: |
| // [3, 3, 3, 3, 3, 3, 3, 3, 1]-way merges in first pass. |
| // [3, 3, 3]-way merges in second pass. |
| // [3]-way merges in third pass. |
| const std::size_t merge_factor = 3; |
| const std::size_t num_blocks = merge_factor * merge_factor * merge_factor - 2; |
| std::vector<attribute_id> kDefaultSortAttributes{col1_}; |
| |
| createBlocks(kNumTuplesPerBlock, num_blocks, kDefaultSortComparator); |
| executePipelined(kDefaultSortAttributes, kDefaultSortOrdering, kDefaultSortNullOrdering, merge_factor, 1); |
| checkOutputRun(kDefaultSortComparator); |
| } |
| |
| TEST_F(SortMergeRunOperatorTest, Pipleined_ThreePass_PartialMergesInFirstSecondPass_AllFeed) { |
| // Should cause: |
| // [3, 3, 3, 3, 3, 3, 3, 2]-way merges in first pass. |
| // [3, 3, 2]-way merges in second pass. |
| // [3]-way merges in third pass. |
| const std::size_t merge_factor = 3; |
| const std::size_t num_blocks = merge_factor * merge_factor * merge_factor - 4; |
| std::vector<attribute_id> kDefaultSortAttributes{col1_}; |
| |
| createBlocks(kNumTuplesPerBlock, num_blocks, kDefaultSortComparator); |
| executePipelined(kDefaultSortAttributes, kDefaultSortOrdering, kDefaultSortNullOrdering, merge_factor, num_blocks); |
| checkOutputRun(kDefaultSortComparator); |
| } |
| |
| TEST_F(SortMergeRunOperatorTest, Pipleined_ThreePass_PartialMergesInFirstSecondPass_SlowFeed) { |
| // Should cause: |
| // [3, 3, 3, 3, 3, 3, 3, 2]-way merges in first pass. |
| // [3, 3, 2]-way merges in second pass. |
| // [3]-way merges in third pass. |
| const std::size_t merge_factor = 3; |
| const std::size_t num_blocks = merge_factor * merge_factor * merge_factor - 4; |
| std::vector<attribute_id> kDefaultSortAttributes{col1_}; |
| |
| createBlocks(kNumTuplesPerBlock, num_blocks, kDefaultSortComparator); |
| executePipelined(kDefaultSortAttributes, kDefaultSortOrdering, kDefaultSortNullOrdering, merge_factor, 1); |
| checkOutputRun(kDefaultSortComparator); |
| } |
| |
| TEST_F(SortMergeRunOperatorTest, Pipleined_ThreePass_PartialMergesInAllPasses_AllFeed) { |
| // Should cause: |
| // [3, 3, 3, 3, 2]-way merges in first pass. |
| // [3, 2]-way merges in second pass. |
| // [2]-way merges in third pass. |
| const std::size_t merge_factor = 3; |
| const std::size_t num_blocks = 2 * merge_factor * merge_factor - 4; |
| std::vector<attribute_id> kDefaultSortAttributes{col1_}; |
| |
| createBlocks(kNumTuplesPerBlock, num_blocks, kDefaultSortComparator); |
| executePipelined(kDefaultSortAttributes, kDefaultSortOrdering, kDefaultSortNullOrdering, merge_factor, num_blocks); |
| checkOutputRun(kDefaultSortComparator); |
| } |
| |
| TEST_F(SortMergeRunOperatorTest, Pipleined_ThreePass_PartialMergesInAllPasses_SlowFeed) { |
| // Should cause: |
| // [3, 3, 3, 3, 2]-way merges in first pass. |
| // [3, 2]-way merges in second pass. |
| // [2]-way merges in third pass. |
| const std::size_t merge_factor = 3; |
| const std::size_t num_blocks = 2 * merge_factor * merge_factor - 4; |
| std::vector<attribute_id> kDefaultSortAttributes{col1_}; |
| |
| createBlocks(kNumTuplesPerBlock, num_blocks, kDefaultSortComparator); |
| executePipelined(kDefaultSortAttributes, kDefaultSortOrdering, kDefaultSortNullOrdering, merge_factor, 1); |
| checkOutputRun(kDefaultSortComparator); |
| } |
| |
| TEST_F(SortMergeRunOperatorTest, Pipleined_ManyPasses_MergeFactor7_AllFeed) { |
| const std::size_t merge_factor = 7; |
| const std::size_t num_blocks = 509; |
| std::vector<attribute_id> kDefaultSortAttributes{col1_}; |
| |
| createBlocks(kNumTuplesPerBlock, num_blocks, kDefaultSortComparator); |
| executePipelined(kDefaultSortAttributes, kDefaultSortOrdering, kDefaultSortNullOrdering, merge_factor, num_blocks); |
| checkOutputRun(kDefaultSortComparator); |
| } |
| |
| TEST_F(SortMergeRunOperatorTest, Pipleined_ManyPasses_MergeFactor7_AllFeed_TopK) { |
| const std::size_t merge_factor = 7; |
| const std::size_t num_blocks = 509; |
| std::vector<attribute_id> kDefaultSortAttributes{col1_}; |
| const std::size_t top_k = 78; |
| |
| createBlocks(kNumTuplesPerBlock, num_blocks, kDefaultSortComparator, top_k); |
| executePipelined(kDefaultSortAttributes, kDefaultSortOrdering, kDefaultSortNullOrdering, merge_factor, num_blocks, |
| top_k); |
| checkOutputRun(kDefaultSortComparator); |
| checkLastTupleAttrValueIsEqual(col1_); |
| } |
| |
| TEST_F(SortMergeRunOperatorTest, Pipleined_ManyPasses_MergeFactor7_SlowFeed) { |
| const std::size_t merge_factor = 7; |
| const std::size_t num_blocks = 509; |
| std::vector<attribute_id> kDefaultSortAttributes{col1_}; |
| |
| createBlocks(kNumTuplesPerBlock, num_blocks, kDefaultSortComparator); |
| executePipelined(kDefaultSortAttributes, kDefaultSortOrdering, kDefaultSortNullOrdering, merge_factor, 1); |
| checkOutputRun(kDefaultSortComparator); |
| } |
| |
| TEST_F(SortMergeRunOperatorTest, Pipleined_ManyPasses_MergeFactor7_SlowFeed_TopK) { |
| const std::size_t merge_factor = 7; |
| const std::size_t num_blocks = 509; |
| std::vector<attribute_id> kDefaultSortAttributes{col1_}; |
| const std::size_t top_k = 37; |
| |
| createBlocks(kNumTuplesPerBlock, num_blocks, kDefaultSortComparator, top_k); |
| executePipelined(kDefaultSortAttributes, kDefaultSortOrdering, kDefaultSortNullOrdering, merge_factor, 1, top_k); |
| checkOutputRun(kDefaultSortComparator); |
| checkLastTupleAttrValueIsEqual(col1_); |
| } |
| |
| TEST_F(SortMergeRunOperatorTest, Pipleined_ManyPasses_MergeFactor17_AllFeed) { |
| const std::size_t merge_factor = 17; |
| const std::size_t num_blocks = 509; |
| std::vector<attribute_id> kDefaultSortAttributes{col1_}; |
| |
| createBlocks(kNumTuplesPerBlock, num_blocks, kDefaultSortComparator); |
| executePipelined(kDefaultSortAttributes, kDefaultSortOrdering, kDefaultSortNullOrdering, merge_factor, num_blocks); |
| checkOutputRun(kDefaultSortComparator); |
| } |
| |
| TEST_F(SortMergeRunOperatorTest, Pipleined_ManyPasses_MergeFactor17_AllFeed_TopK) { |
| const std::size_t merge_factor = 17; |
| const std::size_t num_blocks = 509; |
| std::vector<attribute_id> kDefaultSortAttributes{col1_}; |
| const std::size_t top_k = 32; |
| |
| createBlocks(kNumTuplesPerBlock, num_blocks, kDefaultSortComparator, top_k); |
| executePipelined( |
| kDefaultSortAttributes, kDefaultSortOrdering, kDefaultSortNullOrdering, merge_factor, num_blocks, top_k); |
| checkOutputRun(kDefaultSortComparator); |
| checkLastTupleAttrValueIsEqual(col1_); |
| } |
| |
| TEST_F(SortMergeRunOperatorTest, Pipleined_ManyPasses_MergeFactor17_SlowFeed) { |
| const std::size_t merge_factor = 17; |
| const std::size_t num_blocks = 509; |
| std::vector<attribute_id> kDefaultSortAttributes{col1_}; |
| |
| createBlocks(kNumTuplesPerBlock, num_blocks, kDefaultSortComparator); |
| executePipelined(kDefaultSortAttributes, kDefaultSortOrdering, kDefaultSortNullOrdering, merge_factor, 1); |
| checkOutputRun(kDefaultSortComparator); |
| } |
| |
| TEST_F(SortMergeRunOperatorTest, Pipleined_ManyPasses_MergeFactor17_SlowFeed_TopK) { |
| const std::size_t merge_factor = 17; |
| const std::size_t num_blocks = 509; |
| std::vector<attribute_id> kDefaultSortAttributes{col1_}; |
| const std::size_t top_k = 42; |
| |
| createBlocks(kNumTuplesPerBlock, num_blocks, kDefaultSortComparator, top_k); |
| executePipelined(kDefaultSortAttributes, kDefaultSortOrdering, kDefaultSortNullOrdering, merge_factor, 1, top_k); |
| checkOutputRun(kDefaultSortComparator); |
| checkLastTupleAttrValueIsEqual(col1_); |
| } |
| |
| } // namespace quickstep |
| |
| int main(int argc, char* argv[]) { |
| google::InitGoogleLogging(argv[0]); |
| // Honor FLAGS_buffer_pool_slots in StorageManager. |
| gflags::ParseCommandLineFlags(&argc, &argv, true); |
| testing::InitGoogleTest(&argc, argv); |
| |
| return RUN_ALL_TESTS(); |
| } |
