storage/tests/StorageBlockSort_unittest.cpp - incubator-retired-quickstep - Git at Google

 /**
  * 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 <limits>
 #include <memory>
 #include <string>
 #include <vector>

 #include "gtest/gtest.h"
 #include "glog/logging.h"

 #include "catalog/CatalogAttribute.hpp"
 #include "catalog/CatalogRelation.hpp"
 #include "expressions/scalar/Scalar.hpp"
 #include "expressions/scalar/ScalarAttribute.hpp"
 #include "storage/StorageBlockLayout.hpp"
 #include "storage/StorageManager.hpp"
 #include "types/DoubleType.hpp"
 #include "types/FloatType.hpp"
 #include "types/IntType.hpp"
 #include "types/LongType.hpp"
 #include "types/containers/Tuple.hpp"

 namespace quickstep {

 template <typename ScalarType>
 Scalar* SetupColumn(const std::string &attr, const CatalogRelation &table);

 template <>
 Scalar* SetupColumn<ScalarAttribute>(const std::string &attr, const CatalogRelation &table) {
   return new ScalarAttribute(*table.getAttributeByName(attr));
 }

 namespace {

 class TestTuple {
  public:
   explicit TestTuple(int num)
       : tid_(num), col1_(Bits(num, 3, 5)), col2_(Bits(num, 6, 2)), col3_(Bits(num, 2, 4)), col4_(Bits(num, 1, 5)) {}

   int tid_;
   int col1_;
   int col2_;
   int col3_;
   int col4_;

   static inline int Bits(int value, int offset, int length) { return (value >> offset) & (0xffff >> (16 - length)); }
 };

 }  // namespace

 class StorageBlockSortTest : public ::testing::Test {
  public:
   static const int kNumTuples;

  protected:
   static const char kStoragePath[];
   static const char kTableName[];
   static const relation_id kTableId = 100;
   static const block_id kBlockId = 101;

   static constexpr bool kAscending = true;
   static constexpr bool kDescending = false;

   virtual void SetUp() {
     storage_manager_.reset(new StorageManager(kStoragePath));

     table_.reset(new CatalogRelation(nullptr, kTableName, kTableId));

     // Add attributes
     const Type &int_type = IntType::InstanceNullable();
     table_->addAttribute(new CatalogAttribute(table_.get(), "tid", int_type));
     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(), "col-4", int_type));

     std::unique_ptr<StorageBlockLayout> layout(StorageBlockLayout::GenerateDefaultLayout(*table_, false));
     block_id_ = storage_manager_->createBlock(*table_, *layout);
     MutableBlockReference block = storage_manager_->getBlockMutable(block_id_, *table_);
     table_->addBlock(block_id_);

     for (int i = 0; i < kNumTuples; ++i) {
       tuple_mapping_.emplace_back(i);
     }
     std::random_shuffle(tuple_mapping_.begin(), tuple_mapping_.end());

     tuple_mapping_.reserve(kNumTuples);
     for (size_t i = 0; i < tuple_mapping_.size(); ++i) {
       std::unique_ptr<Tuple> tuple(createTuple(tuple_mapping_[i]));
       block->insertTupleInBatch(*tuple);
     }

     block->rebuild();
   }

   inline void addAttributeValue(std::vector<TypedValue> *values, int value) {
     if (value) {
       values->emplace_back(static_cast<int>(value));
     } else {
       // 0 => NULL
       values->emplace_back(TypeID::kInt);
     }
   }

   inline Tuple* createTuple(int num) {
     std::vector<TypedValue> values;
     TestTuple row(num);
     addAttributeValue(&values, row.tid_);
     addAttributeValue(&values, row.col1_);
     addAttributeValue(&values, row.col2_);
     addAttributeValue(&values, row.col3_);
     addAttributeValue(&values, row.col4_);
     Tuple *tuple = new Tuple(std::move(values));
     return tuple;
   }

   template <typename Configuration, typename ScalarType>
   void testSort(void) {
     Configuration test_instance;

     PtrVector<Scalar> sort_columns;
     std::vector<bool> sort_is_ascending;
     std::vector<bool> null_first;
     test_instance.template configure<ScalarType>(*table_, &sort_columns, &sort_is_ascending, &null_first);

     OrderedTupleIdSequence sorted_sequence;
     MutableBlockReference block = storage_manager_->getBlockMutable(block_id_, *table_);
     block->sort(sort_columns, sort_is_ascending, null_first, &sorted_sequence, nullptr);

     ASSERT_EQ(static_cast<std::size_t>(kNumTuples), sorted_sequence.size());
     test_instance.assertTupleIdSequence(sorted_sequence, tuple_mapping_);
   }

   std::unique_ptr<StorageManager> storage_manager_;
   std::unique_ptr<CatalogRelation> table_;
   std::vector<tuple_id> tuple_mapping_;
   block_id block_id_;
 };

 const int StorageBlockSortTest::kNumTuples = 1 << 8;
 const char StorageBlockSortTest::kStoragePath[] = "table";
 const char StorageBlockSortTest::kTableName[] = "./test_data";

 namespace {

 template <bool ascending>
 class SingleColumnTest {
  public:
   template <typename ScalarType>
   void configure(const CatalogRelation &table,
                  PtrVector<Scalar> *sort_columns,
                  std::vector<bool> *sort_is_ascending,
                  std::vector<bool> *sort_null_first) {
     sort_columns->push_back(SetupColumn<ScalarType>("tid", table));
     sort_is_ascending->push_back(ascending);
     sort_null_first->push_back(false);
   }

   void assertTupleIdSequence(const OrderedTupleIdSequence &sequence, const std::vector<tuple_id> &tuple_mapping) {
     for (size_t id = 1; id < sequence.size(); ++id) {
       TestTuple prev(tuple_mapping[sequence[id-1]]);
       TestTuple current(tuple_mapping[sequence[id]]);
       // NOTE: All values are unique in this test. Hence, we use LT and GT.
       ASSERT_GT(prev.tid_, 0);
       if (current.tid_ != 0) {
         if (ascending) {
           ASSERT_LT(prev.tid_, current.tid_);
         } else {
           ASSERT_GT(prev.tid_, current.tid_);
         }
       }
     }
   }
 };

 template <bool col1_ascending, bool tid_ascending>
 class TwoColumnTest {
  public:
   template <typename ScalarType>
   void configure(const CatalogRelation &table,
                  PtrVector<Scalar> *sort_columns,
                  std::vector<bool> *sort_is_ascending,
                  std::vector<bool> *null_first) {
     sort_columns->push_back(SetupColumn<ScalarType>("col-1", table));
     sort_is_ascending->push_back(col1_ascending);
     null_first->push_back(false);

     sort_columns->push_back(SetupColumn<ScalarType>("tid", table));
     sort_is_ascending->push_back(tid_ascending);
     null_first->push_back(false);
   }

   ::testing::AssertionResult compareTupleId(const int prev, const int current) {
     if (tid_ascending) {
       if (prev > current) {
         return ::testing::AssertionFailure() << "TID " << prev << " not less than " << current;
       }
     } else {
       if (prev < current) {
         return ::testing::AssertionFailure() << "TID " << prev << " not greater than " << current;
       }
     }
     return ::testing::AssertionSuccess();
   }

   ::testing::AssertionResult compare(const TestTuple &prev, const TestTuple &current) {
     if (col1_ascending) {
       if (prev.col1_ > current.col1_) {
         return ::testing::AssertionFailure() << "Col-1 " << prev.col1_ << " not less than " << current.col1_;
       } else if (prev.col1_ == current.col1_) {
         return compareTupleId(prev.tid_, current.tid_);
       }
     } else {
       if (prev.col1_ < current.col1_) {
         return ::testing::AssertionFailure() << "Col-1 " << prev.col1_ << " not greater than " << current.col1_;
       } else if (prev.col1_ == current.col1_) {
         return compareTupleId(prev.tid_, current.tid_);
       }
     }
     return ::testing::AssertionSuccess();
   }

   inline void fixupTestTupleForNULLs(TestTuple *tuple) {
     if (tuple->col1_ == 0) {
       tuple->col1_ = col1_ascending ? std::numeric_limits<int>::max() : std::numeric_limits<int>::min();
     }
     if (tuple->tid_ == 0) {
       tuple->tid_ = tid_ascending ? std::numeric_limits<int>::max() : std::numeric_limits<int>::min();
     }
   }

   void assertTupleIdSequence(const OrderedTupleIdSequence &sequence, const std::vector<tuple_id> &tuple_mapping) {
     for (size_t id = 1; id < sequence.size(); ++id) {
       TestTuple prev(tuple_mapping[sequence[id-1]]);
       fixupTestTupleForNULLs(&prev);
       TestTuple current(tuple_mapping[sequence[id]]);
       fixupTestTupleForNULLs(&current);
       ASSERT_TRUE(compare(prev, current));
     }
   }
 };

 class ThreeColumnTest {
  public:
   template <typename ScalarType>
   void configure(const CatalogRelation &table,
                  PtrVector<Scalar> *sort_columns,
                  std::vector<bool> *sort_is_ascending,
                  std::vector<bool> *null_first) {
     sort_columns->push_back(SetupColumn<ScalarType>("col-1", table));
     sort_is_ascending->push_back(true);
     null_first->push_back(false);

     sort_columns->push_back(SetupColumn<ScalarType>("col-3", table));
     sort_is_ascending->push_back(true);
     null_first->push_back(true);

     sort_columns->push_back(SetupColumn<ScalarType>("tid", table));
     sort_is_ascending->push_back(true);
     null_first->push_back(false);
   }

   ::testing::AssertionResult compare(const TestTuple &prev, const TestTuple &current) {
     if (prev.col1_ > current.col1_) {
       return ::testing::AssertionFailure() << "Col-1 " << prev.col1_ << " is greater than " << current.col1_;
     } else if (prev.col1_ == current.col1_) {
       if (prev.col3_ > current.col3_) {
         return ::testing::AssertionFailure() << "Col-3 " << prev.col3_ << " is greater than " << current.col3_;
       } else if (prev.col3_ == current.col3_) {
         if (prev.tid_ > current.tid_) {
           return ::testing::AssertionFailure() << "TID " << prev.tid_ << " is greater than " << current.tid_;
         }
       }
     }
     return ::testing::AssertionSuccess();
   }

   inline void fixupTestTupleForNULLs(TestTuple *tuple) {
     if (tuple->col1_ == 0) {
       tuple->col1_ = std::numeric_limits<int>::max();
     }
     if (tuple->col3_ == 0) {
       tuple->col3_ = std::numeric_limits<int>::min();
     }
     if (tuple->tid_ == 0) {
       tuple->tid_ = std::numeric_limits<int>::max();
     }
   }

   inline void printTuple(const char *prefix, const TestTuple &tuple) {
     if (VLOG_IS_ON(2)) {
       VLOG(2) << prefix << tuple.col1_ << '|' << tuple.col3_ << '|' << tuple.tid_;
     }
   }

   void assertTupleIdSequence(const OrderedTupleIdSequence &sequence, const std::vector<tuple_id> &tuple_mapping) {
     for (size_t id = 1; id < sequence.size(); ++id) {
       TestTuple prev(tuple_mapping[sequence[id-1]]);
       fixupTestTupleForNULLs(&prev);
       TestTuple current(tuple_mapping[sequence[id]]);
       fixupTestTupleForNULLs(&current);
       if (id == 1) printTuple("", prev);
       printTuple("", current);
       ASSERT_TRUE(compare(prev, current));
     }
   }
 };

 class AllColumnTest {
  public:
   template <typename ScalarType>
   void configure(const CatalogRelation &table,
                  PtrVector<Scalar> *sort_columns,
                  std::vector<bool> *sort_is_ascending,
                  std::vector<bool> *null_first) {
     sort_columns->push_back(SetupColumn<ScalarType>("col-1", table));
     sort_is_ascending->push_back(true);
     null_first->push_back(false);

     sort_columns->push_back(SetupColumn<ScalarType>("col-2", table));
     sort_is_ascending->push_back(true);
     null_first->push_back(false);

     sort_columns->push_back(SetupColumn<ScalarType>("col-3", table));
     sort_is_ascending->push_back(true);
     null_first->push_back(true);

     sort_columns->push_back(SetupColumn<ScalarType>("col-4", table));
     sort_is_ascending->push_back(true);
     null_first->push_back(true);

     sort_columns->push_back(SetupColumn<ScalarType>("tid", table));
     sort_is_ascending->push_back(true);
     null_first->push_back(true);
   }

   inline void fixupTestTupleForNULLs(TestTuple *tuple) {
     if (tuple->col1_ == 0) {
       tuple->col1_ = std::numeric_limits<int>::max();
     }
     if (tuple->col2_ == 0) {
       tuple->col2_ = std::numeric_limits<int>::max();
     }
     if (tuple->col3_ == 0) {
       tuple->col3_ = std::numeric_limits<int>::min();
     }
     if (tuple->col4_ == 0) {
       tuple->col4_ = std::numeric_limits<int>::min();
     }
     if (tuple->tid_ == 0) {
       tuple->tid_ = std::numeric_limits<int>::min();
     }
   }

   inline void printTuple(const TestTuple &test) {
     if (VLOG_IS_ON(2)) {
       VLOG(2) << test.col1_ << '|' << test.col2_ << '|' << test.col3_ << '|'
               << test.col4_ << '|' << test.tid_;
     }
   }

   ::testing::AssertionResult compare(const TestTuple &prev, const TestTuple &current) {
     if (prev.col1_ > current.col1_) {
       return ::testing::AssertionFailure() << "Col-1 " << prev.col1_ << " is greater than " << current.col1_;
     } else if (prev.col1_ == current.col1_) {
       if (prev.col2_ > current.col2_) {
         return ::testing::AssertionFailure() << "Col-2 " << prev.col2_ << " is greater than " << current.col2_;
       } else if (prev.col2_ == current.col2_) {
         if (prev.col3_ > current.col3_) {
           return ::testing::AssertionFailure() << "Col-3 " << prev.col3_ << " is greater than " << current.col3_;
         } else if (prev.col3_ == current.col3_) {
           if (prev.col4_ > current.col4_) {
             return ::testing::AssertionFailure() << "Col-4 " << prev.col4_ << " is greater than " << current.col4_;
           } else if (prev.col4_ == current.col4_) {
             if (prev.tid_ > current.tid_) {
               return ::testing::AssertionFailure() << "TID " << prev.tid_ << " is greater than " << current.tid_;
             }
           }
         }
       }
     }
     return ::testing::AssertionSuccess();
   }

   void assertTupleIdSequence(const OrderedTupleIdSequence &sequence, const std::vector<tuple_id> &tuple_mapping) {
     for (size_t id = 1; id < sequence.size(); ++id) {
       TestTuple prev(tuple_mapping[sequence[id-1]]);
       fixupTestTupleForNULLs(&prev);
       TestTuple current(tuple_mapping[sequence[id]]);
       fixupTestTupleForNULLs(&current);
       if (id == 1) printTuple(prev);
       printTuple(current);
       ASSERT_TRUE(compare(prev, current));
     }
   }
 };

 }  // namespace

 TEST_F(StorageBlockSortTest, SingleColumnSort_Ascending_ScalarAttribute) {
   testSort<SingleColumnTest<kAscending>, ScalarAttribute>();
 }

 TEST_F(StorageBlockSortTest, SingleColumnSort_Descending_ScalarAttribute) {
   testSort<SingleColumnTest<kDescending>, ScalarAttribute>();
 }

 TEST_F(StorageBlockSortTest, TwoColumnSort_AscendingAscending_ScalarAttribute) {
   testSort<TwoColumnTest<kAscending, kAscending>, ScalarAttribute>();
 }

 TEST_F(StorageBlockSortTest, TwoColumnSort_DescendingAscending_ScalarAttribute) {
   testSort<TwoColumnTest<kDescending, kAscending>, ScalarAttribute>();
 }

 TEST_F(StorageBlockSortTest, TwoColumnSort_AscendingDescending_ScalarAttribute) {
   testSort<TwoColumnTest<kAscending, kDescending>, ScalarAttribute>();
 }

 TEST_F(StorageBlockSortTest, TwoColumnSort_DescendingDescending_ScalarAttribute) {
   testSort<TwoColumnTest<kDescending, kDescending>, ScalarAttribute>();
 }

 TEST_F(StorageBlockSortTest, ThreeColumnSort_ScalarAttribute) {
   testSort<ThreeColumnTest, ScalarAttribute>();
 }

 TEST_F(StorageBlockSortTest, AllColumnSort_ScalarAttribute) {
   testSort<AllColumnTest, ScalarAttribute>();
 }

 }  // namespace quickstep
	/**
	* 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 <limits>
	#include <memory>
	#include <string>
	#include <vector>

	#include "gtest/gtest.h"
	#include "glog/logging.h"

	#include "catalog/CatalogAttribute.hpp"
	#include "catalog/CatalogRelation.hpp"
	#include "expressions/scalar/Scalar.hpp"
	#include "expressions/scalar/ScalarAttribute.hpp"
	#include "storage/StorageBlockLayout.hpp"
	#include "storage/StorageManager.hpp"
	#include "types/DoubleType.hpp"
	#include "types/FloatType.hpp"
	#include "types/IntType.hpp"
	#include "types/LongType.hpp"
	#include "types/containers/Tuple.hpp"

	namespace quickstep {

	template <typename ScalarType>
	Scalar* SetupColumn(const std::string &attr, const CatalogRelation &table);

	template <>
	Scalar* SetupColumn<ScalarAttribute>(const std::string &attr, const CatalogRelation &table) {
	return new ScalarAttribute(*table.getAttributeByName(attr));
	}

	namespace {

	class TestTuple {
	public:
	explicit TestTuple(int num)
	: tid_(num), col1_(Bits(num, 3, 5)), col2_(Bits(num, 6, 2)), col3_(Bits(num, 2, 4)), col4_(Bits(num, 1, 5)) {}

	int tid_;
	int col1_;
	int col2_;
	int col3_;
	int col4_;

	static inline int Bits(int value, int offset, int length) { return (value >> offset) & (0xffff >> (16 - length)); }
	};

	} // namespace

	class StorageBlockSortTest : public ::testing::Test {
	public:
	static const int kNumTuples;

	protected:
	static const char kStoragePath[];
	static const char kTableName[];
	static const relation_id kTableId = 100;
	static const block_id kBlockId = 101;

	static constexpr bool kAscending = true;
	static constexpr bool kDescending = false;

	virtual void SetUp() {
	storage_manager_.reset(new StorageManager(kStoragePath));

	table_.reset(new CatalogRelation(nullptr, kTableName, kTableId));

	// Add attributes
	const Type &int_type = IntType::InstanceNullable();
	table_->addAttribute(new CatalogAttribute(table_.get(), "tid", int_type));
	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(), "col-4", int_type));

	std::unique_ptr<StorageBlockLayout> layout(StorageBlockLayout::GenerateDefaultLayout(*table_, false));
	block_id_ = storage_manager_->createBlock(table_, layout);
	MutableBlockReference block = storage_manager_->getBlockMutable(block_id_, *table_);
	table_->addBlock(block_id_);

	for (int i = 0; i < kNumTuples; ++i) {
	tuple_mapping_.emplace_back(i);
	}
	std::random_shuffle(tuple_mapping_.begin(), tuple_mapping_.end());

	tuple_mapping_.reserve(kNumTuples);
	for (size_t i = 0; i < tuple_mapping_.size(); ++i) {
	std::unique_ptr<Tuple> tuple(createTuple(tuple_mapping_[i]));
	block->insertTupleInBatch(*tuple);
	}

	block->rebuild();
	}

	inline void addAttributeValue(std::vector<TypedValue> *values, int value) {
	if (value) {
	values->emplace_back(static_cast<int>(value));
	} else {
	// 0 => NULL
	values->emplace_back(TypeID::kInt);
	}
	}

	inline Tuple* createTuple(int num) {
	std::vector<TypedValue> values;
	TestTuple row(num);
	addAttributeValue(&values, row.tid_);
	addAttributeValue(&values, row.col1_);
	addAttributeValue(&values, row.col2_);
	addAttributeValue(&values, row.col3_);
	addAttributeValue(&values, row.col4_);
	Tuple *tuple = new Tuple(std::move(values));
	return tuple;
	}

	template <typename Configuration, typename ScalarType>
	void testSort(void) {
	Configuration test_instance;

	PtrVector<Scalar> sort_columns;
	std::vector<bool> sort_is_ascending;
	std::vector<bool> null_first;
	test_instance.template configure<ScalarType>(*table_, &sort_columns, &sort_is_ascending, &null_first);

	OrderedTupleIdSequence sorted_sequence;
	MutableBlockReference block = storage_manager_->getBlockMutable(block_id_, *table_);
	block->sort(sort_columns, sort_is_ascending, null_first, &sorted_sequence, nullptr);

	ASSERT_EQ(static_cast<std::size_t>(kNumTuples), sorted_sequence.size());
	test_instance.assertTupleIdSequence(sorted_sequence, tuple_mapping_);
	}

	std::unique_ptr<StorageManager> storage_manager_;
	std::unique_ptr<CatalogRelation> table_;
	std::vector<tuple_id> tuple_mapping_;
	block_id block_id_;
	};

	const int StorageBlockSortTest::kNumTuples = 1 << 8;
	const char StorageBlockSortTest::kStoragePath[] = "table";
	const char StorageBlockSortTest::kTableName[] = "./test_data";

	namespace {

	template <bool ascending>
	class SingleColumnTest {
	public:
	template <typename ScalarType>
	void configure(const CatalogRelation &table,
	PtrVector<Scalar> *sort_columns,
	std::vector<bool> *sort_is_ascending,
	std::vector<bool> *sort_null_first) {
	sort_columns->push_back(SetupColumn<ScalarType>("tid", table));
	sort_is_ascending->push_back(ascending);
	sort_null_first->push_back(false);
	}

	void assertTupleIdSequence(const OrderedTupleIdSequence &sequence, const std::vector<tuple_id> &tuple_mapping) {
	for (size_t id = 1; id < sequence.size(); ++id) {
	TestTuple prev(tuple_mapping[sequence[id-1]]);
	TestTuple current(tuple_mapping[sequence[id]]);
	// NOTE: All values are unique in this test. Hence, we use LT and GT.
	ASSERT_GT(prev.tid_, 0);
	if (current.tid_ != 0) {
	if (ascending) {
	ASSERT_LT(prev.tid_, current.tid_);
	} else {
	ASSERT_GT(prev.tid_, current.tid_);
	}
	}
	}
	}
	};

	template <bool col1_ascending, bool tid_ascending>
	class TwoColumnTest {
	public:
	template <typename ScalarType>
	void configure(const CatalogRelation &table,
	PtrVector<Scalar> *sort_columns,
	std::vector<bool> *sort_is_ascending,
	std::vector<bool> *null_first) {
	sort_columns->push_back(SetupColumn<ScalarType>("col-1", table));
	sort_is_ascending->push_back(col1_ascending);
	null_first->push_back(false);

	sort_columns->push_back(SetupColumn<ScalarType>("tid", table));
	sort_is_ascending->push_back(tid_ascending);
	null_first->push_back(false);
	}

	::testing::AssertionResult compareTupleId(const int prev, const int current) {
	if (tid_ascending) {
	if (prev > current) {
	return ::testing::AssertionFailure() << "TID " << prev << " not less than " << current;
	}
	} else {
	if (prev < current) {
	return ::testing::AssertionFailure() << "TID " << prev << " not greater than " << current;
	}
	}
	return ::testing::AssertionSuccess();
	}

	::testing::AssertionResult compare(const TestTuple &prev, const TestTuple &current) {
	if (col1_ascending) {
	if (prev.col1_ > current.col1_) {
	return ::testing::AssertionFailure() << "Col-1 " << prev.col1_ << " not less than " << current.col1_;
	} else if (prev.col1_ == current.col1_) {
	return compareTupleId(prev.tid_, current.tid_);
	}
	} else {
	if (prev.col1_ < current.col1_) {
	return ::testing::AssertionFailure() << "Col-1 " << prev.col1_ << " not greater than " << current.col1_;
	} else if (prev.col1_ == current.col1_) {
	return compareTupleId(prev.tid_, current.tid_);
	}
	}
	return ::testing::AssertionSuccess();
	}

	inline void fixupTestTupleForNULLs(TestTuple *tuple) {
	if (tuple->col1_ == 0) {
	tuple->col1_ = col1_ascending ? std::numeric_limits<int>::max() : std::numeric_limits<int>::min();
	}
	if (tuple->tid_ == 0) {
	tuple->tid_ = tid_ascending ? std::numeric_limits<int>::max() : std::numeric_limits<int>::min();
	}
	}

	void assertTupleIdSequence(const OrderedTupleIdSequence &sequence, const std::vector<tuple_id> &tuple_mapping) {
	for (size_t id = 1; id < sequence.size(); ++id) {
	TestTuple prev(tuple_mapping[sequence[id-1]]);
	fixupTestTupleForNULLs(&prev);
	TestTuple current(tuple_mapping[sequence[id]]);
	fixupTestTupleForNULLs(&current);
	ASSERT_TRUE(compare(prev, current));
	}
	}
	};

	class ThreeColumnTest {
	public:
	template <typename ScalarType>
	void configure(const CatalogRelation &table,
	PtrVector<Scalar> *sort_columns,
	std::vector<bool> *sort_is_ascending,
	std::vector<bool> *null_first) {
	sort_columns->push_back(SetupColumn<ScalarType>("col-1", table));
	sort_is_ascending->push_back(true);
	null_first->push_back(false);

	sort_columns->push_back(SetupColumn<ScalarType>("col-3", table));
	sort_is_ascending->push_back(true);
	null_first->push_back(true);

	sort_columns->push_back(SetupColumn<ScalarType>("tid", table));
	sort_is_ascending->push_back(true);
	null_first->push_back(false);
	}

	::testing::AssertionResult compare(const TestTuple &prev, const TestTuple &current) {
	if (prev.col1_ > current.col1_) {
	return ::testing::AssertionFailure() << "Col-1 " << prev.col1_ << " is greater than " << current.col1_;
	} else if (prev.col1_ == current.col1_) {
	if (prev.col3_ > current.col3_) {
	return ::testing::AssertionFailure() << "Col-3 " << prev.col3_ << " is greater than " << current.col3_;
	} else if (prev.col3_ == current.col3_) {
	if (prev.tid_ > current.tid_) {
	return ::testing::AssertionFailure() << "TID " << prev.tid_ << " is greater than " << current.tid_;
	}
	}
	}
	return ::testing::AssertionSuccess();
	}

	inline void fixupTestTupleForNULLs(TestTuple *tuple) {
	if (tuple->col1_ == 0) {
	tuple->col1_ = std::numeric_limits<int>::max();
	}
	if (tuple->col3_ == 0) {
	tuple->col3_ = std::numeric_limits<int>::min();
	}
	if (tuple->tid_ == 0) {
	tuple->tid_ = std::numeric_limits<int>::max();
	}
	}

	inline void printTuple(const char *prefix, const TestTuple &tuple) {
	if (VLOG_IS_ON(2)) {
	VLOG(2) << prefix << tuple.col1_ << '\|' << tuple.col3_ << '\|' << tuple.tid_;
	}
	}

	void assertTupleIdSequence(const OrderedTupleIdSequence &sequence, const std::vector<tuple_id> &tuple_mapping) {
	for (size_t id = 1; id < sequence.size(); ++id) {
	TestTuple prev(tuple_mapping[sequence[id-1]]);
	fixupTestTupleForNULLs(&prev);
	TestTuple current(tuple_mapping[sequence[id]]);
	fixupTestTupleForNULLs(&current);
	if (id == 1) printTuple("", prev);
	printTuple("", current);
	ASSERT_TRUE(compare(prev, current));
	}
	}
	};

	class AllColumnTest {
	public:
	template <typename ScalarType>
	void configure(const CatalogRelation &table,
	PtrVector<Scalar> *sort_columns,
	std::vector<bool> *sort_is_ascending,
	std::vector<bool> *null_first) {
	sort_columns->push_back(SetupColumn<ScalarType>("col-1", table));
	sort_is_ascending->push_back(true);
	null_first->push_back(false);

	sort_columns->push_back(SetupColumn<ScalarType>("col-2", table));
	sort_is_ascending->push_back(true);
	null_first->push_back(false);

	sort_columns->push_back(SetupColumn<ScalarType>("col-3", table));
	sort_is_ascending->push_back(true);
	null_first->push_back(true);

	sort_columns->push_back(SetupColumn<ScalarType>("col-4", table));
	sort_is_ascending->push_back(true);
	null_first->push_back(true);

	sort_columns->push_back(SetupColumn<ScalarType>("tid", table));
	sort_is_ascending->push_back(true);
	null_first->push_back(true);
	}

	inline void fixupTestTupleForNULLs(TestTuple *tuple) {
	if (tuple->col1_ == 0) {
	tuple->col1_ = std::numeric_limits<int>::max();
	}
	if (tuple->col2_ == 0) {
	tuple->col2_ = std::numeric_limits<int>::max();
	}
	if (tuple->col3_ == 0) {
	tuple->col3_ = std::numeric_limits<int>::min();
	}
	if (tuple->col4_ == 0) {
	tuple->col4_ = std::numeric_limits<int>::min();
	}
	if (tuple->tid_ == 0) {
	tuple->tid_ = std::numeric_limits<int>::min();
	}
	}

	inline void printTuple(const TestTuple &test) {
	if (VLOG_IS_ON(2)) {
	VLOG(2) << test.col1_ << '\|' << test.col2_ << '\|' << test.col3_ << '\|'
	<< test.col4_ << '\|' << test.tid_;
	}
	}

	::testing::AssertionResult compare(const TestTuple &prev, const TestTuple &current) {
	if (prev.col1_ > current.col1_) {
	return ::testing::AssertionFailure() << "Col-1 " << prev.col1_ << " is greater than " << current.col1_;
	} else if (prev.col1_ == current.col1_) {
	if (prev.col2_ > current.col2_) {
	return ::testing::AssertionFailure() << "Col-2 " << prev.col2_ << " is greater than " << current.col2_;
	} else if (prev.col2_ == current.col2_) {
	if (prev.col3_ > current.col3_) {
	return ::testing::AssertionFailure() << "Col-3 " << prev.col3_ << " is greater than " << current.col3_;
	} else if (prev.col3_ == current.col3_) {
	if (prev.col4_ > current.col4_) {
	return ::testing::AssertionFailure() << "Col-4 " << prev.col4_ << " is greater than " << current.col4_;
	} else if (prev.col4_ == current.col4_) {
	if (prev.tid_ > current.tid_) {
	return ::testing::AssertionFailure() << "TID " << prev.tid_ << " is greater than " << current.tid_;
	}
	}
	}
	}
	}
	return ::testing::AssertionSuccess();
	}

	void assertTupleIdSequence(const OrderedTupleIdSequence &sequence, const std::vector<tuple_id> &tuple_mapping) {
	for (size_t id = 1; id < sequence.size(); ++id) {
	TestTuple prev(tuple_mapping[sequence[id-1]]);
	fixupTestTupleForNULLs(&prev);
	TestTuple current(tuple_mapping[sequence[id]]);
	fixupTestTupleForNULLs(&current);
	if (id == 1) printTuple(prev);
	printTuple(current);
	ASSERT_TRUE(compare(prev, current));
	}
	}
	};

	} // namespace

	TEST_F(StorageBlockSortTest, SingleColumnSort_Ascending_ScalarAttribute) {
	testSort<SingleColumnTest<kAscending>, ScalarAttribute>();
	}

	TEST_F(StorageBlockSortTest, SingleColumnSort_Descending_ScalarAttribute) {
	testSort<SingleColumnTest<kDescending>, ScalarAttribute>();
	}

	TEST_F(StorageBlockSortTest, TwoColumnSort_AscendingAscending_ScalarAttribute) {
	testSort<TwoColumnTest<kAscending, kAscending>, ScalarAttribute>();
	}

	TEST_F(StorageBlockSortTest, TwoColumnSort_DescendingAscending_ScalarAttribute) {
	testSort<TwoColumnTest<kDescending, kAscending>, ScalarAttribute>();
	}

	TEST_F(StorageBlockSortTest, TwoColumnSort_AscendingDescending_ScalarAttribute) {
	testSort<TwoColumnTest<kAscending, kDescending>, ScalarAttribute>();
	}

	TEST_F(StorageBlockSortTest, TwoColumnSort_DescendingDescending_ScalarAttribute) {
	testSort<TwoColumnTest<kDescending, kDescending>, ScalarAttribute>();
	}

	TEST_F(StorageBlockSortTest, ThreeColumnSort_ScalarAttribute) {
	testSort<ThreeColumnTest, ScalarAttribute>();
	}

	TEST_F(StorageBlockSortTest, AllColumnSort_ScalarAttribute) {
	testSort<AllColumnTest, ScalarAttribute>();
	}

	} // namespace quickstep