diff --git a/CMakeLists.txt b/CMakeLists.txt
index 882d1da..2b76e06 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -708,3 +708,88 @@
 
 # Link against other required system and third-party libraries.
 target_link_libraries(quickstep_cli_shell ${LIBS})
+
+if (ENABLE_NATIVENET)
+  include_directories(${CMAKE_CURRENT_BINARY_DIR}/third_party/tmb/include)
+
+  # Build the quickstep_distributed_cli_shell executable.
+  add_executable (quickstep_distributed_cli_shell cli/QuickstepCliDistributed.cpp)
+  # Link against direct deps (will transitively pull in everything needed).
+  target_link_libraries(quickstep_distributed_cli_shell
+                        gflags_nothreads-static
+                        glog
+                        quickstep_catalog_CatalogRelation
+                        quickstep_catalog_Catalog_proto
+                        quickstep_cli_LineReader
+                        quickstep_cli_PrintToScreen
+                        quickstep_parser_ParseStatement
+                        quickstep_parser_SqlParserWrapper
+                        quickstep_queryexecution_QueryExecutionMessages_proto
+                        quickstep_queryexecution_QueryExecutionTypedefs
+                        quickstep_queryexecution_QueryExecutionUtil
+                        quickstep_storage_StorageBlockInfo
+                        quickstep_storage_StorageManager
+                        quickstep_utility_Macros
+                        tmb
+                        ${GRPCPLUSPLUS_LIBRARIES})
+  if (ENABLE_HDFS)
+    target_link_libraries(quickstep_distributed_cli_shell
+                          quickstep_storage_FileManagerHdfs)
+  endif()
+
+  # Link against other required system and third-party libraries.
+  target_link_libraries(quickstep_distributed_cli_shell
+                        ${LIBS})
+
+  # Build the quickstep_distributed_conductor executable.
+  add_executable (quickstep_distributed_conductor cli/ConductorDistributed.cpp)
+  # Link against direct deps (will transitively pull in everything needed).
+  target_link_libraries(quickstep_distributed_conductor
+                        gflags_nothreads-static
+                        glog
+                        quickstep_catalog_CatalogDatabase
+                        quickstep_catalog_CatalogRelation
+                        quickstep_catalog_CatalogTypedefs
+                        quickstep_catalog_Catalog_proto
+                        quickstep_parser_SqlParserWrapper
+                        quickstep_queryexecution_ForemanDistributed
+                        quickstep_queryexecution_QueryExecutionMessages_proto
+                        quickstep_queryexecution_QueryExecutionTypedefs
+                        quickstep_queryexecution_QueryExecutionUtil
+                        quickstep_queryoptimizer_QueryHandle
+                        quickstep_queryoptimizer_QueryProcessor
+                        quickstep_utility_SqlError
+                        tmb
+                        ${GRPCPLUSPLUS_LIBRARIES})
+  if (ENABLE_HDFS)
+    target_link_libraries(quickstep_distributed_conductor
+                          quickstep_storage_FileManagerHdfs)
+  endif()
+
+  # Link against other required system and third-party libraries.
+  target_link_libraries(quickstep_distributed_conductor ${LIBS})
+
+  # Build the quickstep_distributed_worker executable.
+  add_executable (quickstep_distributed_worker cli/WorkerDistributed.cpp)
+  # Link against direct deps (will transitively pull in everything needed).
+  target_link_libraries(quickstep_distributed_worker
+                        gflags_nothreads-static
+                        glog
+                        quickstep_catalog_CatalogDatabase
+                        quickstep_cli_InputParserUtil
+                        quickstep_queryexecution_Shiftboss
+                        quickstep_queryexecution_Worker
+                        quickstep_queryexecution_WorkerDirectory
+                        quickstep_storage_PreloaderThread
+                        quickstep_storage_StorageManager
+                        quickstep_utility_PtrVector
+                        tmb
+                        ${GRPCPLUSPLUS_LIBRARIES})
+  if (ENABLE_HDFS)
+    target_link_libraries(quickstep_distributed_worker
+                          quickstep_storage_FileManagerHdfs)
+  endif()
+
+  # Link against other required system and third-party libraries.
+  target_link_libraries(quickstep_distributed_worker ${LIBS})
+endif()
diff --git a/cli/ConductorDistributed.cpp b/cli/ConductorDistributed.cpp
new file mode 100644
index 0000000..111a9a0
--- /dev/null
+++ b/cli/ConductorDistributed.cpp
@@ -0,0 +1,282 @@
+/**
+ *   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 Conductor */
+
+#include <cstddef>
+#include <cstdint>
+#include <cstdio>
+#include <cstdlib>
+#include <exception>
+#include <memory>
+#include <string>
+#include <unordered_map>
+#include <utility>
+
+#include "catalog/Catalog.pb.h"
+#include "catalog/CatalogDatabase.hpp"
+#include "catalog/CatalogRelation.hpp"
+#include "catalog/CatalogTypedefs.hpp"
+#include "parser/SqlParserWrapper.hpp"
+#include "query_execution/ForemanDistributed.hpp"
+#include "query_execution/QueryExecutionMessages.pb.h"
+#include "query_execution/QueryExecutionTypedefs.hpp"
+#include "query_execution/QueryExecutionUtil.hpp"
+#include "query_optimizer/QueryHandle.hpp"
+#include "query_optimizer/QueryProcessor.hpp"
+#include "utility/SqlError.hpp"
+
+#include "gflags/gflags.h"
+#include "glog/logging.h"
+#include "grpc/grpc.h"
+
+#include "tmb/id_typedefs.h"
+#include "tmb/message_bus.h"
+#include "tmb/native_net_client_message_bus.h"
+#include "tmb/tagged_message.h"
+
+using std::free;
+using std::malloc;
+using std::move;
+using std::size_t;
+using std::string;
+using std::unique_ptr;
+using std::unordered_map;
+
+using quickstep::QueryExecutionUtil;
+using quickstep::QueryProcessor;
+using quickstep::kDistributedCliRegistrationResponseMessage;
+using quickstep::kDistributedCliRegistrationMessage;
+using quickstep::kQueryExecutionErrorMessage;
+using quickstep::kQueryExecutionSuccessMessage;
+using quickstep::kSqlQueryMessage;
+
+using tmb::AnnotatedMessage;
+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(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);
+
+DEFINE_string(storage_path, kDefaultStoragePath,
+              "Filesystem path for quickstep database storage.");
+
+}  // namespace quickstep
+
+int main(int argc, char* argv[]) {
+  gflags::ParseCommandLineFlags(&argc, &argv, true);
+  google::InitGoogleLogging(argv[0]);
+  grpc_init();
+
+  quickstep::SqlParserWrapper parser_wrapper;
+
+  // 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);
+  }
+
+  string catalog_path(fixed_storage_path);
+  catalog_path.append("catalog.pb.bin");
+
+  // Setup QueryProcessor, including CatalogDatabase and StorageManager.
+  std::unique_ptr<QueryProcessor> query_processor;
+  try {
+    query_processor.reset(new QueryProcessor(catalog_path, fixed_storage_path));
+  } catch (const std::exception &e) {
+    LOG(FATAL) << "FATAL ERROR DURING STARTUP: " << e.what();
+  } catch (...) {
+    LOG(FATAL) << "NON-STANDARD EXCEPTION DURING STARTUP";
+  }
+
+  tmb::NativeNetClientMessageBus bus;
+  bus.AddServer(quickstep::FLAGS_tmb_server_ip.c_str(), quickstep::FLAGS_tmb_server_port);
+  bus.Initialize();
+  bus.ResetBus();
+
+  const client_id conductor_id = bus.Connect();
+  LOG(INFO) << "Conductor TMB Client ID: " << conductor_id;
+
+  bus.RegisterClientAsReceiver(conductor_id, kDistributedCliRegistrationMessage);
+  bus.RegisterClientAsSender(conductor_id, kDistributedCliRegistrationResponseMessage);
+
+  bus.RegisterClientAsReceiver(conductor_id, kSqlQueryMessage);
+  bus.RegisterClientAsSender(conductor_id, kQueryExecutionSuccessMessage);
+  bus.RegisterClientAsSender(conductor_id, kQueryExecutionErrorMessage);
+
+  quickstep::ForemanDistributed foreman(&bus, query_processor->getDefaultDatabase());
+
+  for (;;) {
+    AnnotatedMessage annotated_message(bus.Receive(conductor_id, 0, true));
+    const TaggedMessage &tagged_message = annotated_message.tagged_message;
+    const client_id sender = annotated_message.sender;
+
+    LOG(INFO) << "Conductor received typed '" << tagged_message.message_type()
+              << "' message from client " << sender;
+    switch (tagged_message.message_type()) {
+      case kDistributedCliRegistrationMessage: {
+        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 message(static_cast<const void*>(proto_bytes),
+                              proto_length,
+                              kDistributedCliRegistrationResponseMessage);
+        free(proto_bytes);
+
+        LOG(INFO) << "Conductor sent DistributedCliRegistrationResponseMessage (typed '"
+                  << kDistributedCliRegistrationResponseMessage
+                  << "') to Distributed CLI " << sender;
+        QueryExecutionUtil::SendTMBMessage(&bus,
+                                           conductor_id,
+                                           sender,
+                                           move(message));
+        break;
+      }
+      case kSqlQueryMessage: {
+        quickstep::serialization::SqlQueryMessage proto;
+        CHECK(proto.ParseFromArray(tagged_message.message(), tagged_message.message_bytes()));
+
+        LOG(INFO) << "Conductor received the following SQL query: " << proto.sql_query();
+
+        string *command_string = new string(move(proto.sql_query()));
+
+        parser_wrapper.feedNextBuffer(command_string);
+        quickstep::ParseResult parse_result = parser_wrapper.getNextStatement();
+
+        unique_ptr<quickstep::QueryHandle> query_handle;
+        try {
+          query_handle.reset(query_processor->generateQueryHandle(*parse_result.parsed_statement));
+        } catch (const quickstep::SqlError &sql_error) {
+          // Set the query execution status along with the error message.
+          quickstep::serialization::QueryExecutionErrorMessage proto;
+          proto.set_error_message(sql_error.formatMessage(*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 message(static_cast<const void*>(proto_bytes),
+                                proto_length,
+                                kQueryExecutionErrorMessage);
+          free(proto_bytes);
+
+          LOG(INFO) << "Conductor (on behalf of Optimizer) sent QueryExecutionErrorMessage (typed '"
+                    << kQueryExecutionErrorMessage
+                    << "') to Distributed CLI " << sender;
+          QueryExecutionUtil::SendTMBMessage(&bus,
+                                             conductor_id,
+                                             sender,
+                                             move(message));
+          break;
+        }
+
+        try {
+          foreman.setQueryHandle(query_handle.get());
+          foreman.start();
+          foreman.join();
+
+          quickstep::serialization::QueryExecutionSuccessMessage proto;
+
+          const quickstep::CatalogRelation *query_result_relation = query_handle->getQueryResultRelation();
+          if (query_result_relation) {
+            proto.mutable_query_result_relation()->MergeFrom(query_result_relation->getProto());
+          }
+
+          const size_t 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,
+                                kQueryExecutionSuccessMessage);
+          free(proto_bytes);
+
+          LOG(INFO) << "Conductor sent QueryExecutionSuccessMessage (typed '" << kQueryExecutionSuccessMessage
+                    << "') to Distributed CLI " << sender;
+          QueryExecutionUtil::SendTMBMessage(&bus,
+                                             conductor_id,
+                                             sender,
+                                             move(message));
+
+          if (query_result_relation) {
+            const quickstep::relation_id query_result_relation_id = query_result_relation->getID();
+            DCHECK(query_processor->getDefaultDatabase()->hasRelationWithId(query_result_relation_id));
+            query_processor->getDefaultDatabase()->dropRelationById(query_result_relation_id);
+          }
+
+          query_processor->saveCatalog();
+        } catch (const std::exception &e) {
+          string error_message("QUERY EXECUTION ERROR: ");
+          error_message.append(e.what());
+          error_message.push_back('\n');
+
+          quickstep::serialization::QueryExecutionErrorMessage proto;
+          proto.set_error_message(error_message);
+
+          const size_t 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,
+                                kQueryExecutionErrorMessage);
+          free(proto_bytes);
+
+          LOG(INFO) << "Conductor (on behalf of Executor) sent QueryExecutionErrorMessage (typed '"
+                    << kQueryExecutionErrorMessage
+                    << "') to Distributed CLI " << sender;
+          QueryExecutionUtil::SendTMBMessage(&bus,
+                                             conductor_id,
+                                             sender,
+                                             move(message));
+        }
+        break;
+      }
+      default: {
+        LOG(FATAL) << "Unknown TMB message type";
+      }
+    }
+  }
+
+  return 0;
+}
diff --git a/cli/QuickstepCliDistributed.cpp b/cli/QuickstepCliDistributed.cpp
new file mode 100644
index 0000000..5bc6ccf
--- /dev/null
+++ b/cli/QuickstepCliDistributed.cpp
@@ -0,0 +1,293 @@
+/**
+ *   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;
+}
diff --git a/cli/WorkerDistributed.cpp b/cli/WorkerDistributed.cpp
new file mode 100644
index 0000000..ff49aec
--- /dev/null
+++ b/cli/WorkerDistributed.cpp
@@ -0,0 +1,210 @@
+/**
+ *   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 Worker */
+
+#include <cstddef>
+#include <cstdint>
+#include <cstdio>
+#include <string>
+#include <vector>
+
+#include "catalog/CatalogDatabase.hpp"
+#include "cli/InputParserUtil.hpp"
+#include "query_execution/Shiftboss.hpp"
+#include "query_execution/Worker.hpp"
+#include "query_execution/WorkerDirectory.hpp"
+#include "storage/StorageConfig.h"  // For QUICKSTEP_HAVE_FILE_MANAGER_HDFS.
+
+#ifdef QUICKSTEP_HAVE_FILE_MANAGER_HDFS
+#include "storage/FileManagerHdfs.hpp"
+#endif
+
+#include "storage/PreloaderThread.hpp"
+#include "storage/StorageManager.hpp"
+#include "utility/PtrVector.hpp"
+
+#include "gflags/gflags.h"
+#include "glog/logging.h"
+#include "grpc/grpc.h"
+
+#include "tmb/id_typedefs.h"
+#include "tmb/message_bus.h"
+#include "tmb/native_net_client_message_bus.h"
+
+using std::fflush;
+using std::fprintf;
+using std::printf;
+using std::size_t;
+using std::string;
+using std::vector;
+
+using quickstep::InputParserUtil;
+using quickstep::Worker;
+
+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_int32(num_workers, 1, "Number of worker threads");
+
+DEFINE_bool(preload_buffer_pool, false,
+            "If true, pre-load all known blocks into buffer pool before "
+            "accepting queries (should also set --buffer_pool_slots to be "
+            "large enough to accomodate the entire database).");
+
+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 {
+    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);
+
+DEFINE_string(worker_affinities, "",
+              "A comma-separated list of CPU IDs to pin worker threads to "
+              "(leaving this empty will cause all worker threads to inherit "
+              "the affinity mask of the Quickstep process, which typically "
+              "means that they will all be runable on any CPU according to "
+              "the kernel's own scheduling policy).");
+
+}  // namespace quickstep
+
+int main(int argc, char* argv[]) {
+  gflags::ParseCommandLineFlags(&argc, &argv, true);
+  google::InitGoogleLogging(argv[0]);
+  grpc_init();
+
+  if (quickstep::FLAGS_num_workers > 0) {
+    printf("Starting Quickstep with %d worker thread(s)\n",
+           quickstep::FLAGS_num_workers);
+  } else {
+    LOG(FATAL) << "Quickstep needs at least one worker thread";
+  }
+
+#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();
+
+  // Parse the CPU affinities for workers and the preloader thread, if enabled
+  // to warm up the buffer pool.
+  const vector<int> worker_cpu_affinities =
+      InputParserUtil::ParseWorkerAffinities(quickstep::FLAGS_num_workers,
+                                             quickstep::FLAGS_worker_affinities);
+
+  if (quickstep::FLAGS_preload_buffer_pool) {
+    // TODO(quickstep-team): Enable Preloading by reconstructing CatalogDatabase
+    // from Catalog in Conductor.
+    quickstep::CatalogDatabase database(nullptr, "default");
+    quickstep::PreloaderThread preloader(database,
+                                         &storage_manager,
+                                         worker_cpu_affinities.front());
+    printf("Preloading buffer pool... ");
+    fflush(stdout);
+    preloader.start();
+    preloader.join();
+    printf("DONE\n");
+  }
+
+  // Get the NUMA affinities for workers.
+  vector<int> cpu_numa_nodes = InputParserUtil::GetNUMANodesForCPUs();
+  if (cpu_numa_nodes.empty()) {
+    // libnuma is not present. Assign -1 as the NUMA node for every worker.
+    cpu_numa_nodes.assign(worker_cpu_affinities.size(), -1);
+  }
+
+  vector<int> worker_numa_nodes;
+  quickstep::PtrVector<Worker> workers;
+  vector<client_id> worker_client_ids;
+
+  // Initialize the worker threads.
+  DCHECK_EQ(static_cast<size_t>(quickstep::FLAGS_num_workers),
+            worker_cpu_affinities.size());
+  for (size_t worker_idx = 0;
+       worker_idx < worker_cpu_affinities.size();
+       ++worker_idx) {
+    int numa_node_id = -1;
+    if (worker_cpu_affinities[worker_idx] >= 0) {
+      // This worker can be NUMA affinitized.
+      numa_node_id = cpu_numa_nodes[worker_cpu_affinities[worker_idx]];
+    }
+    worker_numa_nodes.push_back(numa_node_id);
+
+    workers.push_back(new Worker(worker_idx,
+                                 &bus,
+                                 worker_cpu_affinities[worker_idx]));
+    worker_client_ids.push_back(workers.back().getBusClientID());
+  }
+
+  // TODO(zuyu): Move WorkerDirectory within Shiftboss once the latter is added.
+  quickstep::WorkerDirectory worker_directory(worker_cpu_affinities.size(),
+                                              worker_client_ids,
+                                              worker_numa_nodes);
+
+  quickstep::Shiftboss shiftboss(&bus,
+                                 &storage_manager,
+                                 &worker_directory);
+  shiftboss.start();
+
+  // Start the worker threads.
+  for (Worker &worker : workers) {
+    worker.start();
+  }
+
+  for (Worker &worker : workers) {
+    worker.join();
+  }
+  shiftboss.join();
+
+  return 0;
+}
diff --git a/query_execution/QueryExecutionMessages.proto b/query_execution/QueryExecutionMessages.proto
index 95807d4..b9a964c 100644
--- a/query_execution/QueryExecutionMessages.proto
+++ b/query_execution/QueryExecutionMessages.proto
@@ -20,6 +20,10 @@
 import "query_execution/QueryContext.proto";
 import "relational_operators/WorkOrder.proto";
 
+// Used for any messages that do not carry payloads.
+message EmptyMessage {
+}
+
 message ShiftbossRegistrationMessage {
   // The total Work Order processing capacity in Shiftboss, which equals to the
   // sum of the capacity of each worker managed by Shiftboss.
@@ -87,7 +91,7 @@
 }
 
 message QueryExecutionSuccessMessage {
-  optional CatalogRelation query_result_relation = 1;
+  optional CatalogRelationSchema query_result_relation = 1;
 }
 
 message QueryExecutionErrorMessage {