cli/QuickstepCliDistributed.cpp - incubator-retired-quickstep - Git at Google

 /**
  *   Copyright 2011-2015 Quickstep Technologies LLC.
  *   Copyright 2015-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.
  **/

 /* A standalone command-line interface to distributed QuickStep */

 #include <chrono>
 #include <cstddef>
 #include <cstdint>
 #include <cstdio>
 #include <cstdlib>
 #include <string>
 #include <utility>
 #include <vector>

 #include "catalog/Catalog.pb.h"
 #include "catalog/CatalogRelation.hpp"
 #include "cli/CliConfig.h"  // For QUICKSTEP_USE_LINENOISE.
 #include "cli/LineReader.hpp"

 #ifdef QUICKSTEP_USE_LINENOISE
 #include "cli/LineReaderLineNoise.hpp"
 typedef quickstep::LineReaderLineNoise LineReaderImpl;
 #else
 #include "cli/LineReaderDumb.hpp"
 typedef quickstep::LineReaderDumb LineReaderImpl;
 #endif

 #include "cli/PrintToScreen.hpp"
 #include "parser/ParseStatement.hpp"
 #include "parser/SqlParserWrapper.hpp"
 #include "query_execution/QueryExecutionMessages.pb.h"
 #include "query_execution/QueryExecutionTypedefs.hpp"
 #include "query_execution/QueryExecutionUtil.hpp"
 #include "storage/StorageConfig.h"  // For QUICKSTEP_HAVE_FILE_MANAGER_HDFS.

 #ifdef QUICKSTEP_HAVE_FILE_MANAGER_HDFS
 #include "storage/FileManagerHdfs.hpp"
 #endif

 #include "storage/StorageBlockInfo.hpp"
 #include "storage/StorageManager.hpp"
 #include "utility/Macros.hpp"

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

 #include "tmb/address.h"
 #include "tmb/id_typedefs.h"
 #include "tmb/message_bus.h"
 #include "tmb/message_style.h"
 #include "tmb/native_net_client_message_bus.h"
 #include "tmb/tagged_message.h"

 using std::fprintf;
 using std::free;
 using std::malloc;
 using std::move;
 using std::printf;
 using std::size_t;
 using std::string;

 using quickstep::ParseResult;
 using quickstep::QueryExecutionUtil;
 using quickstep::block_id;
 using quickstep::kDistributedCliRegistrationResponseMessage;
 using quickstep::kDistributedCliRegistrationMessage;
 using quickstep::kQueryExecutionErrorMessage;
 using quickstep::kQueryExecutionSuccessMessage;
 using quickstep::kSqlQueryMessage;

 using tmb::Address;
 using tmb::AnnotatedMessage;
 using tmb::MessageBus;
 using tmb::TaggedMessage;
 using tmb::client_id;

 namespace quickstep {

 #ifdef QUICKSTEP_OS_WINDOWS
 static constexpr char kPathSeparator = '\\';
 static constexpr char kDefaultStoragePath[] = "qsstor\\";
 #else
 static constexpr char kPathSeparator = '/';
 static constexpr char kDefaultStoragePath[] = "qsstor/";
 #endif

 DEFINE_string(storage_path, kDefaultStoragePath,
               "Filesystem path for quickstep database storage.");

 DEFINE_string(tmb_server_ip, "127.0.0.1", "IP Address of the TMB Server.");

 static bool ValidateTmbServerPort(const char *flagname,
                                   std::int32_t value) {
   if (value > 0 && value < 65536) {
     return true;
   } else {
     std::fprintf(stderr, "--%s must be between 1 and 65535 (inclusive)\n", flagname);
     return false;
   }
 }
 DEFINE_int32(tmb_server_port, 4575, "Port of the TMB Server.");
 static const bool tmb_server_port_dummy
     = gflags::RegisterFlagValidator(&FLAGS_tmb_server_port, &ValidateTmbServerPort);

 }  // namespace quickstep

 int main(int argc, char* argv[]) {
   gflags::ParseCommandLineFlags(&argc, &argv, true);
   google::InitGoogleLogging(argv[0]);
   grpc_init();

 #ifdef QUICKSTEP_HAVE_FILE_MANAGER_HDFS
   if (quickstep::FLAGS_use_hdfs) {
     LOG(INFO) << "Using HDFS as the default persistent storage, with namenode at "
               << quickstep::FLAGS_hdfs_namenode_host << ":"
               << quickstep::FLAGS_hdfs_namenode_port << " and block replication factor "
               << quickstep::FLAGS_hdfs_num_replications << "\n";
   }
 #endif

   // Setup the paths used by StorageManager.
   string fixed_storage_path(quickstep::FLAGS_storage_path);
   if (!fixed_storage_path.empty()
       && (fixed_storage_path.back() != quickstep::kPathSeparator)) {
     fixed_storage_path.push_back(quickstep::kPathSeparator);
   }

   quickstep::StorageManager storage_manager(fixed_storage_path);

   tmb::NativeNetClientMessageBus bus;
   bus.AddServer(quickstep::FLAGS_tmb_server_ip.c_str(), quickstep::FLAGS_tmb_server_port);
   bus.Initialize();

   const client_id cli_id = bus.Connect();
   LOG(INFO) << "CLI TMB Client ID: " << cli_id;

   bus.RegisterClientAsSender(cli_id, kDistributedCliRegistrationMessage);
   bus.RegisterClientAsReceiver(cli_id, kDistributedCliRegistrationResponseMessage);

   LOG(INFO) << "Distributed CLI sent DistributedCliRegistrationMessage (typed '"
             << kDistributedCliRegistrationMessage
             << "') to all";

   Address all_addresses;
   all_addresses.All(true);

   // NOTE(zuyu): The singleton Conductor would need one copy of the message.
   tmb::MessageStyle style;

   quickstep::serialization::EmptyMessage proto;

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

   TaggedMessage cli_reg_message(static_cast<const void*>(proto_bytes),
                                 proto_length,
                                 kDistributedCliRegistrationMessage);
   free(proto_bytes);

   MessageBus::SendStatus send_status =
       bus.Send(cli_id, all_addresses, style, move(cli_reg_message));
   DCHECK(send_status == MessageBus::SendStatus::kOK);

   // Wait for Conductor to response.
   AnnotatedMessage cli_reg_response_message(bus.Receive(cli_id, 0, true));
   DCHECK_EQ(kDistributedCliRegistrationResponseMessage,
             cli_reg_response_message.tagged_message.message_type());
   const client_id conductor_id = cli_reg_response_message.sender;

   LOG(INFO) << "Distributed CLI received typed '" << kDistributedCliRegistrationResponseMessage
             << "' message from client " << conductor_id;

   // Prepare for executing a query.
   bus.RegisterClientAsSender(cli_id, kSqlQueryMessage);
   bus.RegisterClientAsReceiver(cli_id, kQueryExecutionSuccessMessage);
   bus.RegisterClientAsReceiver(cli_id, kQueryExecutionErrorMessage);

   LineReaderImpl line_reader("distributed_quickstep> ",
                              "                  ...> ");
   quickstep::SqlParserWrapper parser_wrapper;
   std::chrono::time_point<std::chrono::steady_clock> start, end;

   for (;;) {
     string *command_string = new string();
     *command_string = line_reader.getNextCommand();
     if (command_string->size() == 0) {
       delete command_string;
       break;
     }

     parser_wrapper.feedNextBuffer(command_string);

     bool quitting = false;
     for (;;) {
       ParseResult result = parser_wrapper.getNextStatement();
       if (result.condition == ParseResult::kSuccess) {
         if (result.parsed_statement->getStatementType() == quickstep::ParseStatement::kQuit) {
           quitting = true;
           break;
         }

         LOG(INFO) << "Distributed CLI sent SqlQueryMessage (typed '" << kSqlQueryMessage
                   << "') to Foreman";
         quickstep::serialization::SqlQueryMessage proto;
         proto.set_sql_query(*command_string);

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

         TaggedMessage sql_query_message(static_cast<const void*>(proto_bytes),
                                         proto_length,
                                         kSqlQueryMessage);
         free(proto_bytes);

         QueryExecutionUtil::SendTMBMessage(&bus,
                                            cli_id,
                                            conductor_id,
                                            move(sql_query_message));

         start = std::chrono::steady_clock::now();

         AnnotatedMessage annotated_message(bus.Receive(cli_id, 0, true));
         const TaggedMessage &tagged_message = annotated_message.tagged_message;
         LOG(INFO) << "Distributed CLI received typed '" << tagged_message.message_type()
                   << "' message from client " << annotated_message.sender;
         switch (tagged_message.message_type()) {
           case kQueryExecutionSuccessMessage: {
             end = std::chrono::steady_clock::now();

             quickstep::serialization::QueryExecutionSuccessMessage proto;
             CHECK(proto.ParseFromArray(tagged_message.message(), tagged_message.message_bytes()));

             if (proto.has_query_result_relation()) {
               quickstep::CatalogRelation query_result_relation(proto.query_result_relation());

               quickstep::PrintToScreen::PrintRelation(query_result_relation,
                                                       &storage_manager,
                                                       stdout);

               const std::vector<block_id> blocks(query_result_relation.getBlocksSnapshot());
               for (const block_id block : blocks) {
                 storage_manager.deleteBlockOrBlobFile(block);
               }
             }

             printf("Execution time: %g seconds\nQuery Complete\n", std::chrono::duration<double>(end - start).count());
             break;
           }
           case kQueryExecutionErrorMessage: {
             quickstep::serialization::QueryExecutionErrorMessage proto;
             CHECK(proto.ParseFromArray(tagged_message.message(), tagged_message.message_bytes()));

             fprintf(stderr, "%s", proto.error_message().c_str());
             break;
           }
           default: {
             LOG(ERROR) << "Unknown TMB message type";
           }
         }
       } else {
         if (result.condition == ParseResult::kError) {
           fprintf(stderr, "%s", result.error_message.c_str());
         }
         break;
       }
     }

     if (quitting) {
       break;
     }
   }

   bus.Disconnect(cli_id);

   return 0;
 }
	/**
	* Copyright 2011-2015 Quickstep Technologies LLC.
	* Copyright 2015-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.
	**/

	/* A standalone command-line interface to distributed QuickStep */

	#include <chrono>
	#include <cstddef>
	#include <cstdint>
	#include <cstdio>
	#include <cstdlib>
	#include <string>
	#include <utility>
	#include <vector>

	#include "catalog/Catalog.pb.h"
	#include "catalog/CatalogRelation.hpp"
	#include "cli/CliConfig.h" // For QUICKSTEP_USE_LINENOISE.
	#include "cli/LineReader.hpp"

	#ifdef QUICKSTEP_USE_LINENOISE
	#include "cli/LineReaderLineNoise.hpp"
	typedef quickstep::LineReaderLineNoise LineReaderImpl;
	#else
	#include "cli/LineReaderDumb.hpp"
	typedef quickstep::LineReaderDumb LineReaderImpl;
	#endif

	#include "cli/PrintToScreen.hpp"
	#include "parser/ParseStatement.hpp"
	#include "parser/SqlParserWrapper.hpp"
	#include "query_execution/QueryExecutionMessages.pb.h"
	#include "query_execution/QueryExecutionTypedefs.hpp"
	#include "query_execution/QueryExecutionUtil.hpp"
	#include "storage/StorageConfig.h" // For QUICKSTEP_HAVE_FILE_MANAGER_HDFS.

	#ifdef QUICKSTEP_HAVE_FILE_MANAGER_HDFS
	#include "storage/FileManagerHdfs.hpp"
	#endif

	#include "storage/StorageBlockInfo.hpp"
	#include "storage/StorageManager.hpp"
	#include "utility/Macros.hpp"

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

	#include "tmb/address.h"
	#include "tmb/id_typedefs.h"
	#include "tmb/message_bus.h"
	#include "tmb/message_style.h"
	#include "tmb/native_net_client_message_bus.h"
	#include "tmb/tagged_message.h"

	using std::fprintf;
	using std::free;
	using std::malloc;
	using std::move;
	using std::printf;
	using std::size_t;
	using std::string;

	using quickstep::ParseResult;
	using quickstep::QueryExecutionUtil;
	using quickstep::block_id;
	using quickstep::kDistributedCliRegistrationResponseMessage;
	using quickstep::kDistributedCliRegistrationMessage;
	using quickstep::kQueryExecutionErrorMessage;
	using quickstep::kQueryExecutionSuccessMessage;
	using quickstep::kSqlQueryMessage;

	using tmb::Address;
	using tmb::AnnotatedMessage;
	using tmb::MessageBus;
	using tmb::TaggedMessage;
	using tmb::client_id;

	namespace quickstep {

	#ifdef QUICKSTEP_OS_WINDOWS
	static constexpr char kPathSeparator = '\\';
	static constexpr char kDefaultStoragePath[] = "qsstor\\";
	#else
	static constexpr char kPathSeparator = '/';
	static constexpr char kDefaultStoragePath[] = "qsstor/";
	#endif

	DEFINE_string(storage_path, kDefaultStoragePath,
	"Filesystem path for quickstep database storage.");

	DEFINE_string(tmb_server_ip, "127.0.0.1", "IP Address of the TMB Server.");

	static bool ValidateTmbServerPort(const char *flagname,
	std::int32_t value) {
	if (value > 0 && value < 65536) {
	return true;
	} else {
	std::fprintf(stderr, "--%s must be between 1 and 65535 (inclusive)\n", flagname);
	return false;
	}
	}
	DEFINE_int32(tmb_server_port, 4575, "Port of the TMB Server.");
	static const bool tmb_server_port_dummy
	= gflags::RegisterFlagValidator(&FLAGS_tmb_server_port, &ValidateTmbServerPort);

	} // namespace quickstep

	int main(int argc, char* argv[]) {
	gflags::ParseCommandLineFlags(&argc, &argv, true);
	google::InitGoogleLogging(argv[0]);
	grpc_init();

	#ifdef QUICKSTEP_HAVE_FILE_MANAGER_HDFS
	if (quickstep::FLAGS_use_hdfs) {
	LOG(INFO) << "Using HDFS as the default persistent storage, with namenode at "
	<< quickstep::FLAGS_hdfs_namenode_host << ":"
	<< quickstep::FLAGS_hdfs_namenode_port << " and block replication factor "
	<< quickstep::FLAGS_hdfs_num_replications << "\n";
	}
	#endif

	// Setup the paths used by StorageManager.
	string fixed_storage_path(quickstep::FLAGS_storage_path);
	if (!fixed_storage_path.empty()
	&& (fixed_storage_path.back() != quickstep::kPathSeparator)) {
	fixed_storage_path.push_back(quickstep::kPathSeparator);
	}

	quickstep::StorageManager storage_manager(fixed_storage_path);

	tmb::NativeNetClientMessageBus bus;
	bus.AddServer(quickstep::FLAGS_tmb_server_ip.c_str(), quickstep::FLAGS_tmb_server_port);
	bus.Initialize();

	const client_id cli_id = bus.Connect();
	LOG(INFO) << "CLI TMB Client ID: " << cli_id;

	bus.RegisterClientAsSender(cli_id, kDistributedCliRegistrationMessage);
	bus.RegisterClientAsReceiver(cli_id, kDistributedCliRegistrationResponseMessage);

	LOG(INFO) << "Distributed CLI sent DistributedCliRegistrationMessage (typed '"
	<< kDistributedCliRegistrationMessage
	<< "') to all";

	Address all_addresses;
	all_addresses.All(true);

	// NOTE(zuyu): The singleton Conductor would need one copy of the message.
	tmb::MessageStyle style;

	quickstep::serialization::EmptyMessage proto;

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

	TaggedMessage cli_reg_message(static_cast<const void*>(proto_bytes),
	proto_length,
	kDistributedCliRegistrationMessage);
	free(proto_bytes);

	MessageBus::SendStatus send_status =
	bus.Send(cli_id, all_addresses, style, move(cli_reg_message));
	DCHECK(send_status == MessageBus::SendStatus::kOK);

	// Wait for Conductor to response.
	AnnotatedMessage cli_reg_response_message(bus.Receive(cli_id, 0, true));
	DCHECK_EQ(kDistributedCliRegistrationResponseMessage,
	cli_reg_response_message.tagged_message.message_type());
	const client_id conductor_id = cli_reg_response_message.sender;

	LOG(INFO) << "Distributed CLI received typed '" << kDistributedCliRegistrationResponseMessage
	<< "' message from client " << conductor_id;

	// Prepare for executing a query.
	bus.RegisterClientAsSender(cli_id, kSqlQueryMessage);
	bus.RegisterClientAsReceiver(cli_id, kQueryExecutionSuccessMessage);
	bus.RegisterClientAsReceiver(cli_id, kQueryExecutionErrorMessage);

	LineReaderImpl line_reader("distributed_quickstep> ",
	" ...> ");
	quickstep::SqlParserWrapper parser_wrapper;
	std::chrono::time_point<std::chrono::steady_clock> start, end;

	for (;;) {
	string *command_string = new string();
	*command_string = line_reader.getNextCommand();
	if (command_string->size() == 0) {
	delete command_string;
	break;
	}

	parser_wrapper.feedNextBuffer(command_string);

	bool quitting = false;
	for (;;) {
	ParseResult result = parser_wrapper.getNextStatement();
	if (result.condition == ParseResult::kSuccess) {
	if (result.parsed_statement->getStatementType() == quickstep::ParseStatement::kQuit) {
	quitting = true;
	break;
	}

	LOG(INFO) << "Distributed CLI sent SqlQueryMessage (typed '" << kSqlQueryMessage
	<< "') to Foreman";
	quickstep::serialization::SqlQueryMessage proto;
	proto.set_sql_query(*command_string);

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

	TaggedMessage sql_query_message(static_cast<const void*>(proto_bytes),
	proto_length,
	kSqlQueryMessage);
	free(proto_bytes);

	QueryExecutionUtil::SendTMBMessage(&bus,
	cli_id,
	conductor_id,
	move(sql_query_message));

	start = std::chrono::steady_clock::now();

	AnnotatedMessage annotated_message(bus.Receive(cli_id, 0, true));
	const TaggedMessage &tagged_message = annotated_message.tagged_message;
	LOG(INFO) << "Distributed CLI received typed '" << tagged_message.message_type()
	<< "' message from client " << annotated_message.sender;
	switch (tagged_message.message_type()) {
	case kQueryExecutionSuccessMessage: {
	end = std::chrono::steady_clock::now();

	quickstep::serialization::QueryExecutionSuccessMessage proto;
	CHECK(proto.ParseFromArray(tagged_message.message(), tagged_message.message_bytes()));

	if (proto.has_query_result_relation()) {
	quickstep::CatalogRelation query_result_relation(proto.query_result_relation());

	quickstep::PrintToScreen::PrintRelation(query_result_relation,
	&storage_manager,
	stdout);

	const std::vector<block_id> blocks(query_result_relation.getBlocksSnapshot());
	for (const block_id block : blocks) {
	storage_manager.deleteBlockOrBlobFile(block);
	}
	}

	printf("Execution time: %g seconds\nQuery Complete\n", std::chrono::duration<double>(end - start).count());
	break;
	}
	case kQueryExecutionErrorMessage: {
	quickstep::serialization::QueryExecutionErrorMessage proto;
	CHECK(proto.ParseFromArray(tagged_message.message(), tagged_message.message_bytes()));

	fprintf(stderr, "%s", proto.error_message().c_str());
	break;
	}
	default: {
	LOG(ERROR) << "Unknown TMB message type";
	}
	}
	} else {
	if (result.condition == ParseResult::kError) {
	fprintf(stderr, "%s", result.error_message.c_str());
	}
	break;
	}
	}

	if (quitting) {
	break;
	}
	}

	bus.Disconnect(cli_id);

	return 0;
	}