Fixed the command execution bug in the distributed version.
diff --git a/cli/tests/CMakeLists.txt b/cli/tests/CMakeLists.txt
index c450e4d..18d321d 100644
--- a/cli/tests/CMakeLists.txt
+++ b/cli/tests/CMakeLists.txt
@@ -76,8 +76,9 @@
glog
gtest
quickstep_catalog_CatalogTypedefs
- quickstep_cli_CommandExecutorUtil
+ quickstep_cli_CommandExecutor
quickstep_cli_Constants
+ quickstep_cli_DefaultsConfigurator
quickstep_cli_DropRelation
quickstep_cli_PrintToScreen
quickstep_parser_ParseStatement
@@ -90,11 +91,10 @@
quickstep_queryexecution_Shiftboss
quickstep_queryexecution_Worker
quickstep_queryexecution_WorkerDirectory
- quickstep_queryoptimizer_Optimizer
- quickstep_queryoptimizer_OptimizerContext
quickstep_queryoptimizer_QueryHandle
- quickstep_queryoptimizer_tests_TestDatabaseLoader
+ quickstep_queryoptimizer_QueryProcessor
quickstep_storage_DataExchangerAsync
+ quickstep_storage_StorageConstants
quickstep_storage_StorageManager
quickstep_utility_Macros
quickstep_utility_MemStream
diff --git a/cli/tests/CommandExecutorTestRunner.cpp b/cli/tests/CommandExecutorTestRunner.cpp
index 4ab94d8..a8307db 100644
--- a/cli/tests/CommandExecutorTestRunner.cpp
+++ b/cli/tests/CommandExecutorTestRunner.cpp
@@ -69,7 +69,7 @@
std::printf("%s\n", parse_statement.toString().c_str());
try {
if (parse_statement.getStatementType() == ParseStatement::kCommand) {
- quickstep::cli::executeCommand(
+ cli::executeCommand(
*result.parsed_statement,
*query_processor_->getDefaultDatabase(),
main_thread_client_id_,
diff --git a/cli/tests/DistributedCommandExecutorTest.cpp b/cli/tests/DistributedCommandExecutorTest.cpp
index b41a70f..c8574b9 100644
--- a/cli/tests/DistributedCommandExecutorTest.cpp
+++ b/cli/tests/DistributedCommandExecutorTest.cpp
@@ -49,8 +49,6 @@
auto test_runner = make_unique<quickstep::DistributedCommandExecutorTestRunner>(argv[3]);
test_driver = make_unique<quickstep::TextBasedTestDriver>(&input_file, test_runner.get());
- test_driver->registerOption(
- quickstep::DistributedCommandExecutorTestRunner::kResetOption);
::testing::InitGoogleTest(&argc, argv);
const int success = RUN_ALL_TESTS();
diff --git a/cli/tests/DistributedCommandExecutorTestRunner.cpp b/cli/tests/DistributedCommandExecutorTestRunner.cpp
index a029975..d0baf9d 100644
--- a/cli/tests/DistributedCommandExecutorTestRunner.cpp
+++ b/cli/tests/DistributedCommandExecutorTestRunner.cpp
@@ -20,6 +20,7 @@
#include "cli/tests/DistributedCommandExecutorTestRunner.hpp"
#include <cstdio>
+#include <cstdlib>
#include <memory>
#include <set>
#include <string>
@@ -27,8 +28,9 @@
#include <vector>
#include "catalog/CatalogTypedefs.hpp"
-#include "cli/CommandExecutorUtil.hpp"
+#include "cli/CommandExecutor.hpp"
#include "cli/Constants.hpp"
+#include "cli/DefaultsConfigurator.hpp"
#include "cli/DropRelation.hpp"
#include "cli/PrintToScreen.hpp"
#include "parser/ParseStatement.hpp"
@@ -37,11 +39,10 @@
#include "query_execution/ForemanDistributed.hpp"
#include "query_execution/QueryExecutionTypedefs.hpp"
#include "query_execution/QueryExecutionUtil.hpp"
-#include "query_optimizer/Optimizer.hpp"
-#include "query_optimizer/OptimizerContext.hpp"
#include "query_optimizer/QueryHandle.hpp"
-#include "query_optimizer/tests/TestDatabaseLoader.hpp"
+#include "query_optimizer/QueryProcessor.hpp"
#include "storage/DataExchangerAsync.hpp"
+#include "storage/StorageConstants.hpp"
#include "storage/StorageManager.hpp"
#include "utility/MemStream.hpp"
#include "utility/SqlError.hpp"
@@ -62,13 +63,9 @@
class CatalogRelation;
-namespace C = cli;
-
-const char *DistributedCommandExecutorTestRunner::kResetOption =
- "reset_before_execution";
-
DistributedCommandExecutorTestRunner::DistributedCommandExecutorTestRunner(const string &storage_path)
- : query_id_(0) {
+ : catalog_path_(storage_path + kCatalogFilename),
+ query_id_(0) {
bus_.Initialize();
cli_id_ = bus_.Connect();
@@ -82,23 +79,13 @@
block_locator_ = make_unique<BlockLocator>(&bus_);
block_locator_->start();
- test_database_loader_ = make_unique<optimizer::TestDatabaseLoader>(
- storage_path,
- block_locator::getBlockDomain(
- test_database_loader_data_exchanger_.network_address(), cli_id_, &locator_client_id_, &bus_),
- locator_client_id_,
- &bus_);
- DCHECK_EQ(block_locator_->getBusClientID(), locator_client_id_);
- test_database_loader_data_exchanger_.set_storage_manager(test_database_loader_->storage_manager());
- test_database_loader_data_exchanger_.start();
-
- test_database_loader_->createTestRelation(false /* allow_vchar */);
- test_database_loader_->loadTestRelation();
+ DefaultsConfigurator::InitializeDefaultDatabase(storage_path, catalog_path_);
+ query_processor_ = std::make_unique<QueryProcessor>(std::string(catalog_path_));
// NOTE(zuyu): Foreman should initialize before Shiftboss so that the former
// could receive a registration message from the latter.
- foreman_ = make_unique<ForemanDistributed>(*block_locator_, &bus_, test_database_loader_->catalog_database(),
- nullptr /* query_processor */);
+ foreman_ = make_unique<ForemanDistributed>(*block_locator_, &bus_, query_processor_->getDefaultDatabase(),
+ query_processor_.get());
foreman_->start();
// We don't use the NUMA aware version of worker code.
@@ -139,17 +126,18 @@
foreman_->join();
- test_database_loader_data_exchanger_.shutdown();
- test_database_loader_.reset();
data_exchanger_.shutdown();
storage_manager_.reset();
CHECK(MessageBus::SendStatus::kOK ==
QueryExecutionUtil::SendTMBMessage(&bus_, cli_id_, locator_client_id_, TaggedMessage(kPoisonMessage)));
- test_database_loader_data_exchanger_.join();
data_exchanger_.join();
block_locator_->join();
+
+ const std::string command = "rm -f " + catalog_path_;
+ CHECK(!std::system(command.c_str()))
+ << "Failed when attempting to remove catalog proto file: " << catalog_path_;
}
void DistributedCommandExecutorTestRunner::runTestCase(
@@ -158,12 +146,6 @@
VLOG(4) << "Test SQL(s): " << input;
- if (options.find(kResetOption) != options.end()) {
- test_database_loader_->clear();
- test_database_loader_->createTestRelation(false /* allow_vchar */);
- test_database_loader_->loadTestRelation();
- }
-
MemStream output_stream;
sql_parser_.feedNextBuffer(new string(input));
@@ -181,32 +163,19 @@
try {
if (parse_statement.getStatementType() == ParseStatement::kCommand) {
- const ParseCommand &command = static_cast<const ParseCommand &>(parse_statement);
- const PtrVector<ParseString> &arguments = *(command.arguments());
- const string &command_str = command.command()->value();
-
- string command_response;
- if (command_str == C::kDescribeDatabaseCommand) {
- command_response = C::ExecuteDescribeDatabase(arguments, *test_database_loader_->catalog_database());
- } else if (command_str == C::kDescribeTableCommand) {
- if (arguments.empty()) {
- command_response = C::ExecuteDescribeDatabase(arguments, *test_database_loader_->catalog_database());
- } else {
- command_response = C::ExecuteDescribeTable(arguments, *test_database_loader_->catalog_database());
- }
- } else {
- THROW_SQL_ERROR_AT(command.command()) << "Unsupported command";
- }
-
- std::fprintf(output_stream.file(), "%s", command_response.c_str());
+ cli::executeCommand(
+ *result.parsed_statement,
+ *query_processor_->getDefaultDatabase(),
+ cli_id_,
+ foreman_->getBusClientID(),
+ &bus_,
+ storage_manager_.get(),
+ query_processor_.get(),
+ output_stream.file());
} else {
- optimizer::OptimizerContext optimizer_context;
auto query_handle = std::make_unique<QueryHandle>(query_id_++, cli_id_);
+ query_processor_->generateQueryHandle(parse_statement, query_handle.get());
- optimizer_.generateQueryHandle(parse_statement,
- test_database_loader_->catalog_database(),
- &optimizer_context,
- query_handle.get());
const CatalogRelation *query_result_relation = query_handle->getQueryResultRelation();
QueryExecutionUtil::ConstructAndSendAdmitRequestMessage(
@@ -217,11 +186,11 @@
if (query_result_relation) {
PrintToScreen::PrintRelation(*query_result_relation,
- test_database_loader_->storage_manager(),
+ storage_manager_.get(),
output_stream.file());
DropRelation::Drop(*query_result_relation,
- test_database_loader_->catalog_database(),
- test_database_loader_->storage_manager());
+ query_processor_->getDefaultDatabase(),
+ storage_manager_.get());
}
}
} catch (const SqlError &error) {
diff --git a/cli/tests/DistributedCommandExecutorTestRunner.hpp b/cli/tests/DistributedCommandExecutorTestRunner.hpp
index 0427a85..5b03b2c 100644
--- a/cli/tests/DistributedCommandExecutorTestRunner.hpp
+++ b/cli/tests/DistributedCommandExecutorTestRunner.hpp
@@ -33,8 +33,7 @@
#include "query_execution/Shiftboss.hpp"
#include "query_execution/Worker.hpp"
#include "query_execution/WorkerDirectory.hpp"
-#include "query_optimizer/Optimizer.hpp"
-#include "query_optimizer/tests/TestDatabaseLoader.hpp"
+#include "query_optimizer/QueryProcessor.hpp"
#include "storage/DataExchangerAsync.hpp"
#include "storage/StorageManager.hpp"
#include "utility/Macros.hpp"
@@ -53,12 +52,6 @@
class DistributedCommandExecutorTestRunner : public TextBasedTestRunner {
public:
/**
- * @brief If this option is enabled, recreate the entire database and
- * repopulate the data before every test.
- */
- static const char *kResetOption;
-
- /**
* @brief Constructor.
*/
explicit DistributedCommandExecutorTestRunner(const std::string &storage_path);
@@ -70,18 +63,18 @@
std::string *output) override;
private:
+ const std::string catalog_path_;
+
std::size_t query_id_;
SqlParserWrapper sql_parser_;
- std::unique_ptr<optimizer::TestDatabaseLoader> test_database_loader_;
- DataExchangerAsync test_database_loader_data_exchanger_;
- optimizer::Optimizer optimizer_;
MessageBusImpl bus_;
tmb::client_id cli_id_, locator_client_id_;
std::unique_ptr<BlockLocator> block_locator_;
+ std::unique_ptr<QueryProcessor> query_processor_;
std::unique_ptr<ForemanDistributed> foreman_;
MessageBusImpl bus_local_;
diff --git a/cli/tests/command_executor/CMakeLists.txt b/cli/tests/command_executor/CMakeLists.txt
index 3850de7..86ebe98 100644
--- a/cli/tests/command_executor/CMakeLists.txt
+++ b/cli/tests/command_executor/CMakeLists.txt
@@ -32,6 +32,11 @@
"${CMAKE_CURRENT_BINARY_DIR}/Dt/")
if (ENABLE_DISTRIBUTED)
+ add_test(quickstep_cli_tests_commandexecutor_analyze_distributed
+ "../quickstep_cli_tests_CommandExecutorTest"
+ "${CMAKE_CURRENT_SOURCE_DIR}/Analyze.test"
+ "${CMAKE_CURRENT_BINARY_DIR}/AnalyzeDistributed.test"
+ "${CMAKE_CURRENT_BINARY_DIR}/AnalyzeDistributed/")
add_test(quickstep_cli_tests_commandexecutor_d_distributed
"../quickstep_cli_tests_DistributedCommandExecutorTest"
"${CMAKE_CURRENT_SOURCE_DIR}/D.test"
@@ -51,6 +56,7 @@
file(MAKE_DIRECTORY ${CMAKE_CURRENT_BINARY_DIR}/Dt)
if (ENABLE_DISTRIBUTED)
+ file(MAKE_DIRECTORY ${CMAKE_CURRENT_BINARY_DIR}/AnalyzeDistributed)
file(MAKE_DIRECTORY ${CMAKE_CURRENT_BINARY_DIR}/DDistributed)
file(MAKE_DIRECTORY ${CMAKE_CURRENT_BINARY_DIR}/DtDistributed)
endif(ENABLE_DISTRIBUTED)
diff --git a/cli/tests/command_executor/D.test b/cli/tests/command_executor/D.test
index 2d96b47..1001ee8 100644
--- a/cli/tests/command_executor/D.test
+++ b/cli/tests/command_executor/D.test
@@ -58,7 +58,6 @@
--
==
-
\d foo
--
Table "foo"
@@ -139,23 +138,17 @@
==
-\d
---
- List of relations
-
- Name | Type | Blocks
-+--------------------------------------+-------+---------+
- foo | table | 1
- foo2 | table | 1
- foo3 | table | 1
- foo4 | table | 0
- foo_hash_part | table | 4
- averylongtablenamethatseemstoneverend | table | 1
-
-==
-
\d invalidtable
--
ERROR: Unrecognized relation invalidtable (1 : 4)
\d invalidtable
^
+==
+
+DROP TABLE foo;
+DROP TABLE foo2;
+DROP TABLE foo3;
+DROP TABLE foo4;
+DROP TABLE foo_hash_part;
+--
+==
diff --git a/cli/tests/command_executor/Dt.test b/cli/tests/command_executor/Dt.test
index f735fd3..046040c 100644
--- a/cli/tests/command_executor/Dt.test
+++ b/cli/tests/command_executor/Dt.test
@@ -80,3 +80,11 @@
ERROR: Unrecognized relation invalidtable (1 : 5)
\dt invalidtable
^
+==
+
+DROP TABLE foo;
+DROP TABLE foo2;
+DROP TABLE foo3;
+DROP TABLE foo4;
+DROP TABLE averylongtablenamethatseemstoneverend;
+--
