query_execution/tests/BlockLocator_unittest.cpp - incubator-retired-quickstep - Git at Google

 /**
  *   Copyright 2016 Pivotal Software, Inc.
  *
  *   Licensed under the Apache License, Version 2.0 (the "License");
  *   you may not use this file except in compliance with the License.
  *   You may obtain a copy of the License at
  *
  *       http://www.apache.org/licenses/LICENSE-2.0
  *
  *   Unless required by applicable law or agreed to in writing, software
  *   distributed under the License is distributed on an "AS IS" BASIS,
  *   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  *   See the License for the specific language governing permissions and
  *   limitations under the License.
  **/

 #include <cstdlib>
 #include <memory>
 #include <string>
 #include <utility>
 #include <vector>

 #include "catalog/CatalogAttribute.hpp"
 #include "catalog/CatalogRelation.hpp"
 #include "query_execution/BlockLocator.hpp"
 #include "query_execution/QueryExecutionMessages.pb.h"
 #include "query_execution/QueryExecutionTypedefs.hpp"
 #include "query_execution/QueryExecutionUtil.hpp"
 #include "storage/StorageBlob.hpp"
 #include "storage/StorageBlock.hpp"
 #include "storage/StorageBlockInfo.hpp"
 #include "storage/StorageConstants.hpp"
 #include "storage/StorageManager.hpp"
 #include "types/TypeFactory.hpp"
 #include "types/TypeID.hpp"

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

 #include "tmb/id_typedefs.h"
 #include "tmb/message_bus.h"
 #include "tmb/tagged_message.h"

 using std::free;
 using std::malloc;
 using std::move;
 using std::string;
 using std::unique_ptr;
 using std::vector;

 using tmb::AnnotatedMessage;
 using tmb::MessageBus;
 using tmb::TaggedMessage;

 namespace quickstep {

 class BlockLocatorTest : public ::testing::Test {
  protected:
   static const char kStoragePath[];
   static const char kDomainNetworkAddress[];

   ~BlockLocatorTest() {
     locator_->join();
   }

   virtual void SetUp() {
     bus_.Initialize();

     locator_.reset(new BlockLocator(&bus_));
     locator_client_id_ = locator_->getBusClientID();
     locator_->start();

     worker_client_id_ = bus_.Connect();

     bus_.RegisterClientAsSender(worker_client_id_, kBlockDomainRegistrationMessage);
     bus_.RegisterClientAsReceiver(worker_client_id_, kBlockDomainRegistrationResponseMessage);

     bus_.RegisterClientAsSender(worker_client_id_, kLocateBlockMessage);
     bus_.RegisterClientAsReceiver(worker_client_id_, kLocateBlockResponseMessage);

     bus_.RegisterClientAsSender(worker_client_id_, kPoisonMessage);

     block_domain_ = getBlockDomain(kDomainNetworkAddress);

     storage_manager_.reset(
         new StorageManager(kStoragePath, block_domain_, locator_client_id_, &bus_));
   }

   virtual void TearDown() {
     storage_manager_.reset();

     serialization::EmptyMessage proto;

     const int proto_length = proto.ByteSize();
     char *proto_bytes = static_cast<char*>(malloc(proto_length));
     CHECK(proto.SerializeToArray(proto_bytes, proto_length));

     TaggedMessage message(static_cast<const void*>(proto_bytes),
                           proto_length,
                           kPoisonMessage);
     free(proto_bytes);

     LOG(INFO) << "Worker (id '" << worker_client_id_
               << "') sent PoisonMessage (typed '" << kPoisonMessage
               << "') to BlockLocator (id '" << locator_client_id_ << "')";
     CHECK(MessageBus::SendStatus::kOK ==
         QueryExecutionUtil::SendTMBMessage(&bus_,
                                            worker_client_id_,
                                            locator_client_id_,
                                            move(message)));
   }

   vector<block_id_domain> getPeerDomains(const block_id block) {
     serialization::BlockMessage proto;
     proto.set_block_id(block);

     const int proto_length = proto.ByteSize();
     char *proto_bytes = static_cast<char*>(malloc(proto_length));
     CHECK(proto.SerializeToArray(proto_bytes, proto_length));

     TaggedMessage message(static_cast<const void*>(proto_bytes),
                           proto_length,
                           kLocateBlockMessage);
     free(proto_bytes);

   LOG(INFO) << "Worker (id '" << worker_client_id_
             << "') sent LocateBlockMessage (typed '" << kLocateBlockMessage
             << "') to BlockLocator";
     CHECK(MessageBus::SendStatus::kOK ==
         QueryExecutionUtil::SendTMBMessage(&bus_,
                                            worker_client_id_,
                                            locator_client_id_,
                                            move(message)));

     const AnnotatedMessage annotated_message(bus_.Receive(worker_client_id_, 0, true));
     const TaggedMessage &tagged_message = annotated_message.tagged_message;
     CHECK_EQ(kLocateBlockResponseMessage, tagged_message.message_type());
     LOG(INFO) << "Worker (id '" << worker_client_id_
               << "') received LocateBlockResponseMessage from BlockLocator";

     serialization::LocateBlockResponseMessage response_proto;
     CHECK(response_proto.ParseFromArray(tagged_message.message(), tagged_message.message_bytes()));

     vector<block_id_domain> domains;
     for (int i = 0; i < response_proto.block_domains_size(); ++i) {
       domains.push_back(response_proto.block_domains(i));
     }

     return domains;
   }

   void checkLoaded(const block_id block) {
     const vector<string> peer_domain_network_addresses = storage_manager_->getPeerDomainNetworkAddresses(block);
     EXPECT_EQ(1u, peer_domain_network_addresses.size());
     EXPECT_STREQ(kDomainNetworkAddress, peer_domain_network_addresses[0].data());

     const vector<block_id_domain> domains = getPeerDomains(block);
     EXPECT_EQ(1u, domains.size());
     EXPECT_EQ(block_domain_, domains[0]);
   }

   void checkEvicted(const block_id block) {
     const vector<string> peer_domain_network_addresses = storage_manager_->getPeerDomainNetworkAddresses(block);
     EXPECT_TRUE(peer_domain_network_addresses.empty());

     const vector<block_id_domain> domains = getPeerDomains(block);
     EXPECT_TRUE(domains.empty());
   }

   tmb::client_id worker_client_id_;

   block_id_domain block_domain_;
   unique_ptr<StorageManager> storage_manager_;

  private:
   block_id_domain getBlockDomain(const string &network_address) {
     serialization::BlockDomainRegistrationMessage proto;
     proto.set_domain_network_address(network_address);

     const int proto_length = proto.ByteSize();
     char *proto_bytes = static_cast<char*>(malloc(proto_length));
     CHECK(proto.SerializeToArray(proto_bytes, proto_length));

     TaggedMessage message(static_cast<const void*>(proto_bytes),
                           proto_length,
                           kBlockDomainRegistrationMessage);
     free(proto_bytes);

     LOG(INFO) << "Worker (id '" << worker_client_id_
               << "') sent BlockDomainRegistrationMessage (typed '" << kBlockDomainRegistrationMessage
               << "') to BlockLocator (id '" << locator_client_id_ << "')";

     CHECK(MessageBus::SendStatus::kOK ==
         QueryExecutionUtil::SendTMBMessage(&bus_,
                                            worker_client_id_,
                                            locator_client_id_,
                                            move(message)));

     const AnnotatedMessage annotated_message(bus_.Receive(worker_client_id_, 0, true));
     const TaggedMessage &tagged_message = annotated_message.tagged_message;
     EXPECT_EQ(locator_client_id_, annotated_message.sender);
     EXPECT_EQ(kBlockDomainRegistrationResponseMessage, tagged_message.message_type());
     LOG(INFO) << "Worker (id '" << worker_client_id_
               << "') received BlockDomainRegistrationResponseMessage from BlockLocator";

     serialization::BlockDomainMessage response_proto;
     CHECK(response_proto.ParseFromArray(tagged_message.message(), tagged_message.message_bytes()));

     return static_cast<block_id_domain>(response_proto.block_domain());
   }

   MessageBusImpl bus_;

   unique_ptr<BlockLocator> locator_;
   tmb::client_id locator_client_id_;
 };

 const char BlockLocatorTest::kStoragePath[] = "./block_locator_test_data/";
 const char BlockLocatorTest::kDomainNetworkAddress[] = "ip:port";

 TEST_F(BlockLocatorTest, BlockTest) {
   CatalogRelation relation(nullptr, "rel");
   relation.addAttribute(new CatalogAttribute(nullptr, "attr_int", TypeFactory::GetType(kInt)));

   const block_id block =
       storage_manager_->createBlock(relation, relation.getDefaultStorageBlockLayout());
   checkLoaded(block);

   ASSERT_TRUE(storage_manager_->saveBlockOrBlob(block));
   storage_manager_->evictBlockOrBlob(block);
   checkEvicted(block);

   {
     const BlockReference block_ref = storage_manager_->getBlock(block, relation);
   }
   checkLoaded(block);

   storage_manager_->deleteBlockOrBlobFile(block);
   checkEvicted(block);
 }

 TEST_F(BlockLocatorTest, BlobTest) {
   const block_id blob = storage_manager_->createBlob(kDefaultBlockSizeInSlots);
   checkLoaded(blob);

   ASSERT_TRUE(storage_manager_->saveBlockOrBlob(blob));
   storage_manager_->evictBlockOrBlob(blob);
   checkEvicted(blob);

   {
     const BlobReference blob_ref = storage_manager_->getBlob(blob);
   }
   checkLoaded(blob);

   storage_manager_->deleteBlockOrBlobFile(blob);
   checkEvicted(blob);
 }

 }  // 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();
 }
	/**
	* Copyright 2016 Pivotal Software, Inc.
	*
	* Licensed under the Apache License, Version 2.0 (the "License");
	* you may not use this file except in compliance with the License.
	* You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing, software
	* distributed under the License is distributed on an "AS IS" BASIS,
	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	* See the License for the specific language governing permissions and
	* limitations under the License.
	**/

	#include <cstdlib>
	#include <memory>
	#include <string>
	#include <utility>
	#include <vector>

	#include "catalog/CatalogAttribute.hpp"
	#include "catalog/CatalogRelation.hpp"
	#include "query_execution/BlockLocator.hpp"
	#include "query_execution/QueryExecutionMessages.pb.h"
	#include "query_execution/QueryExecutionTypedefs.hpp"
	#include "query_execution/QueryExecutionUtil.hpp"
	#include "storage/StorageBlob.hpp"
	#include "storage/StorageBlock.hpp"
	#include "storage/StorageBlockInfo.hpp"
	#include "storage/StorageConstants.hpp"
	#include "storage/StorageManager.hpp"
	#include "types/TypeFactory.hpp"
	#include "types/TypeID.hpp"

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

	#include "tmb/id_typedefs.h"
	#include "tmb/message_bus.h"
	#include "tmb/tagged_message.h"

	using std::free;
	using std::malloc;
	using std::move;
	using std::string;
	using std::unique_ptr;
	using std::vector;

	using tmb::AnnotatedMessage;
	using tmb::MessageBus;
	using tmb::TaggedMessage;

	namespace quickstep {

	class BlockLocatorTest : public ::testing::Test {
	protected:
	static const char kStoragePath[];
	static const char kDomainNetworkAddress[];

	~BlockLocatorTest() {
	locator_->join();
	}

	virtual void SetUp() {
	bus_.Initialize();

	locator_.reset(new BlockLocator(&bus_));
	locator_client_id_ = locator_->getBusClientID();
	locator_->start();

	worker_client_id_ = bus_.Connect();

	bus_.RegisterClientAsSender(worker_client_id_, kBlockDomainRegistrationMessage);
	bus_.RegisterClientAsReceiver(worker_client_id_, kBlockDomainRegistrationResponseMessage);

	bus_.RegisterClientAsSender(worker_client_id_, kLocateBlockMessage);
	bus_.RegisterClientAsReceiver(worker_client_id_, kLocateBlockResponseMessage);

	bus_.RegisterClientAsSender(worker_client_id_, kPoisonMessage);

	block_domain_ = getBlockDomain(kDomainNetworkAddress);

	storage_manager_.reset(
	new StorageManager(kStoragePath, block_domain_, locator_client_id_, &bus_));
	}

	virtual void TearDown() {
	storage_manager_.reset();

	serialization::EmptyMessage proto;

	const int proto_length = proto.ByteSize();
	char proto_bytes = static_cast<char>(malloc(proto_length));
	CHECK(proto.SerializeToArray(proto_bytes, proto_length));

	TaggedMessage message(static_cast<const void*>(proto_bytes),
	proto_length,
	kPoisonMessage);
	free(proto_bytes);

	LOG(INFO) << "Worker (id '" << worker_client_id_
	<< "') sent PoisonMessage (typed '" << kPoisonMessage
	<< "') to BlockLocator (id '" << locator_client_id_ << "')";
	CHECK(MessageBus::SendStatus::kOK ==
	QueryExecutionUtil::SendTMBMessage(&bus_,
	worker_client_id_,
	locator_client_id_,
	move(message)));
	}

	vector<block_id_domain> getPeerDomains(const block_id block) {
	serialization::BlockMessage proto;
	proto.set_block_id(block);

	const int proto_length = proto.ByteSize();
	char proto_bytes = static_cast<char>(malloc(proto_length));
	CHECK(proto.SerializeToArray(proto_bytes, proto_length));

	TaggedMessage message(static_cast<const void*>(proto_bytes),
	proto_length,
	kLocateBlockMessage);
	free(proto_bytes);

	LOG(INFO) << "Worker (id '" << worker_client_id_
	<< "') sent LocateBlockMessage (typed '" << kLocateBlockMessage
	<< "') to BlockLocator";
	CHECK(MessageBus::SendStatus::kOK ==
	QueryExecutionUtil::SendTMBMessage(&bus_,
	worker_client_id_,
	locator_client_id_,
	move(message)));

	const AnnotatedMessage annotated_message(bus_.Receive(worker_client_id_, 0, true));
	const TaggedMessage &tagged_message = annotated_message.tagged_message;
	CHECK_EQ(kLocateBlockResponseMessage, tagged_message.message_type());
	LOG(INFO) << "Worker (id '" << worker_client_id_
	<< "') received LocateBlockResponseMessage from BlockLocator";

	serialization::LocateBlockResponseMessage response_proto;
	CHECK(response_proto.ParseFromArray(tagged_message.message(), tagged_message.message_bytes()));

	vector<block_id_domain> domains;
	for (int i = 0; i < response_proto.block_domains_size(); ++i) {
	domains.push_back(response_proto.block_domains(i));
	}

	return domains;
	}

	void checkLoaded(const block_id block) {
	const vector<string> peer_domain_network_addresses = storage_manager_->getPeerDomainNetworkAddresses(block);
	EXPECT_EQ(1u, peer_domain_network_addresses.size());
	EXPECT_STREQ(kDomainNetworkAddress, peer_domain_network_addresses[0].data());

	const vector<block_id_domain> domains = getPeerDomains(block);
	EXPECT_EQ(1u, domains.size());
	EXPECT_EQ(block_domain_, domains[0]);
	}

	void checkEvicted(const block_id block) {
	const vector<string> peer_domain_network_addresses = storage_manager_->getPeerDomainNetworkAddresses(block);
	EXPECT_TRUE(peer_domain_network_addresses.empty());

	const vector<block_id_domain> domains = getPeerDomains(block);
	EXPECT_TRUE(domains.empty());
	}

	tmb::client_id worker_client_id_;

	block_id_domain block_domain_;
	unique_ptr<StorageManager> storage_manager_;

	private:
	block_id_domain getBlockDomain(const string &network_address) {
	serialization::BlockDomainRegistrationMessage proto;
	proto.set_domain_network_address(network_address);

	const int proto_length = proto.ByteSize();
	char proto_bytes = static_cast<char>(malloc(proto_length));
	CHECK(proto.SerializeToArray(proto_bytes, proto_length));

	TaggedMessage message(static_cast<const void*>(proto_bytes),
	proto_length,
	kBlockDomainRegistrationMessage);
	free(proto_bytes);

	LOG(INFO) << "Worker (id '" << worker_client_id_
	<< "') sent BlockDomainRegistrationMessage (typed '" << kBlockDomainRegistrationMessage
	<< "') to BlockLocator (id '" << locator_client_id_ << "')";

	CHECK(MessageBus::SendStatus::kOK ==
	QueryExecutionUtil::SendTMBMessage(&bus_,
	worker_client_id_,
	locator_client_id_,
	move(message)));

	const AnnotatedMessage annotated_message(bus_.Receive(worker_client_id_, 0, true));
	const TaggedMessage &tagged_message = annotated_message.tagged_message;
	EXPECT_EQ(locator_client_id_, annotated_message.sender);
	EXPECT_EQ(kBlockDomainRegistrationResponseMessage, tagged_message.message_type());
	LOG(INFO) << "Worker (id '" << worker_client_id_
	<< "') received BlockDomainRegistrationResponseMessage from BlockLocator";

	serialization::BlockDomainMessage response_proto;
	CHECK(response_proto.ParseFromArray(tagged_message.message(), tagged_message.message_bytes()));

	return static_cast<block_id_domain>(response_proto.block_domain());
	}

	MessageBusImpl bus_;

	unique_ptr<BlockLocator> locator_;
	tmb::client_id locator_client_id_;
	};

	const char BlockLocatorTest::kStoragePath[] = "./block_locator_test_data/";
	const char BlockLocatorTest::kDomainNetworkAddress[] = "ip:port";

	TEST_F(BlockLocatorTest, BlockTest) {
	CatalogRelation relation(nullptr, "rel");
	relation.addAttribute(new CatalogAttribute(nullptr, "attr_int", TypeFactory::GetType(kInt)));

	const block_id block =
	storage_manager_->createBlock(relation, relation.getDefaultStorageBlockLayout());
	checkLoaded(block);

	ASSERT_TRUE(storage_manager_->saveBlockOrBlob(block));
	storage_manager_->evictBlockOrBlob(block);
	checkEvicted(block);

	{
	const BlockReference block_ref = storage_manager_->getBlock(block, relation);
	}
	checkLoaded(block);

	storage_manager_->deleteBlockOrBlobFile(block);
	checkEvicted(block);
	}

	TEST_F(BlockLocatorTest, BlobTest) {
	const block_id blob = storage_manager_->createBlob(kDefaultBlockSizeInSlots);
	checkLoaded(blob);

	ASSERT_TRUE(storage_manager_->saveBlockOrBlob(blob));
	storage_manager_->evictBlockOrBlob(blob);
	checkEvicted(blob);

	{
	const BlobReference blob_ref = storage_manager_->getBlob(blob);
	}
	checkLoaded(blob);

	storage_manager_->deleteBlockOrBlobFile(blob);
	checkEvicted(blob);
	}

	} // 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();
	}