You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@quickstep.apache.org by zu...@apache.org on 2016/11/27 18:51:30 UTC
[01/12] incubator-quickstep git commit: Moved FLAGS_num_workers into
a file for shared flags. [Forced Update!]
Repository: incubator-quickstep
Updated Branches:
refs/heads/multi-shiftboss-test d4f4b4717 -> d2862e578 (forced update)
Moved FLAGS_num_workers into a file for shared flags.
Project: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/commit/e666a81f
Tree: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/tree/e666a81f
Diff: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/diff/e666a81f
Branch: refs/heads/multi-shiftboss-test
Commit: e666a81f472e4c8ed81e98995c1914ce655e247a
Parents: 212e6cd
Author: Zuyu Zhang <zu...@apache.org>
Authored: Sun Nov 20 22:42:49 2016 -0800
Committer: Zuyu Zhang <zu...@apache.org>
Committed: Sun Nov 20 22:42:49 2016 -0800
----------------------------------------------------------------------
cli/CMakeLists.txt | 1 +
cli/Flags.cpp | 29 +++++++++++++++++++++++++++++
cli/Flags.hpp | 2 ++
cli/QuickstepCli.cpp | 31 +++++--------------------------
4 files changed, 37 insertions(+), 26 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/e666a81f/cli/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/cli/CMakeLists.txt b/cli/CMakeLists.txt
index 9928f3f..9847787 100644
--- a/cli/CMakeLists.txt
+++ b/cli/CMakeLists.txt
@@ -112,6 +112,7 @@ if(QUICKSTEP_HAVE_LIBNUMA)
${LIBNUMA_LIBRARY})
endif()
target_link_libraries(quickstep_cli_Flags
+ quickstep_cli_DefaultsConfigurator
quickstep_storage_StorageConstants
${GFLAGS_LIB_NAME})
target_link_libraries(quickstep_cli_InputParserUtil
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/e666a81f/cli/Flags.cpp
----------------------------------------------------------------------
diff --git a/cli/Flags.cpp b/cli/Flags.cpp
index 384ef71..87f9f73 100644
--- a/cli/Flags.cpp
+++ b/cli/Flags.cpp
@@ -19,14 +19,43 @@
#include "cli/Flags.hpp"
+#include <cstddef>
+#include <cstdio>
#include <string>
+#include "cli/DefaultsConfigurator.hpp"
#include "storage/StorageConstants.hpp"
#include "gflags/gflags.h"
+using std::fprintf;
+
namespace quickstep {
+static bool ValidateNumWorkers(const char *flagname, int value) {
+ if (value > 0) {
+ return true;
+ }
+
+ // Detect the hardware concurrency level.
+ const std::size_t num_hw_threads =
+ DefaultsConfigurator::GetNumHardwareThreads();
+
+ // Use the command-line value if that was supplied, else use the value
+ // that we computed above, provided it did return a valid value.
+ // TODO(jmp): May need to change this at some point to keep one thread
+ // available for the OS if the hardware concurrency level is high.
+ FLAGS_num_workers = num_hw_threads != 0 ? num_hw_threads : 1;
+
+ return FLAGS_num_workers > 0;
+}
+DEFINE_int32(num_workers, 0, "Number of worker threads. If this value is "
+ "specified and is greater than 0, then this user-supplied value is "
+ "used. Else (i.e. the default case), we examine the reported "
+ "hardware concurrency level, and use that.");
+static const volatile bool num_workers_dummy
+ = gflags::RegisterFlagValidator(&FLAGS_num_workers, &ValidateNumWorkers);
+
static bool ValidateStoragePath(const char *flagname,
const std::string &value) {
if (!value.empty() && value.back() != kPathSeparator) {
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/e666a81f/cli/Flags.hpp
----------------------------------------------------------------------
diff --git a/cli/Flags.hpp b/cli/Flags.hpp
index a623448..b020a3e 100644
--- a/cli/Flags.hpp
+++ b/cli/Flags.hpp
@@ -32,6 +32,8 @@ namespace quickstep {
* @brief A collection of common flags shared by Quickstep CLIs in both the
* single-node and the distributed version.
**/
+DECLARE_int32(num_workers);
+
DECLARE_string(storage_path);
/** @} */
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/e666a81f/cli/QuickstepCli.cpp
----------------------------------------------------------------------
diff --git a/cli/QuickstepCli.cpp b/cli/QuickstepCli.cpp
index 33a9ff4..6a7c500 100644
--- a/cli/QuickstepCli.cpp
+++ b/cli/QuickstepCli.cpp
@@ -109,6 +109,7 @@ using quickstep::AdmitRequestMessage;
using quickstep::CatalogRelation;
using quickstep::DefaultsConfigurator;
using quickstep::DropRelation;
+using quickstep::FLAGS_num_workers;
using quickstep::FLAGS_storage_path;
using quickstep::ForemanSingleNode;
using quickstep::InputParserUtil;
@@ -136,11 +137,6 @@ using tmb::client_id;
namespace quickstep {
-DEFINE_int32(num_workers, 0, "Number of worker threads. If this value is "
- "specified and is greater than 0, then this "
- "user-supplied value is used. Else (i.e. the"
- "default case), we examine the reported "
- "hardware concurrency level, and use that.");
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 "
@@ -189,25 +185,8 @@ int main(int argc, char* argv[]) {
google::InitGoogleLogging(argv[0]);
gflags::ParseCommandLineFlags(&argc, &argv, true);
- // Detect the hardware concurrency level.
- const std::size_t num_hw_threads =
- DefaultsConfigurator::GetNumHardwareThreads();
-
- // Use the command-line value if that was supplied, else use the value
- // that we computed above, provided it did return a valid value.
- // TODO(jmp): May need to change this at some point to keep one thread
- // available for the OS if the hardware concurrency level is high.
- if (quickstep::FLAGS_num_workers <= 0) {
- LOG(INFO) << "Quickstep expects at least one worker thread, switching to "
- "the default number of worker threads";
- }
- const int real_num_workers = quickstep::FLAGS_num_workers > 0
- ? quickstep::FLAGS_num_workers
- : (num_hw_threads != 0 ? num_hw_threads : 1);
-
- DCHECK_GT(real_num_workers, 0);
- printf("Starting Quickstep with %d worker thread(s) and a %.2f GB buffer pool\n",
- real_num_workers,
+ printf("Starting Quickstep with %d worker thread(s) and a %.2f GB buffer pool.\n",
+ FLAGS_num_workers,
(static_cast<double>(quickstep::FLAGS_buffer_pool_slots) * quickstep::kSlotSizeBytes)/quickstep::kAGigaByte);
#ifdef QUICKSTEP_HAVE_FILE_MANAGER_HDFS
@@ -290,7 +269,7 @@ int main(int argc, char* argv[]) {
// 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(real_num_workers,
+ InputParserUtil::ParseWorkerAffinities(FLAGS_num_workers,
quickstep::FLAGS_worker_affinities);
const std::size_t num_numa_nodes_system = DefaultsConfigurator::GetNumNUMANodes();
@@ -323,7 +302,7 @@ int main(int argc, char* argv[]) {
vector<client_id> worker_client_ids;
// Initialize the worker threads.
- DCHECK_EQ(static_cast<std::size_t>(real_num_workers),
+ DCHECK_EQ(static_cast<std::size_t>(FLAGS_num_workers),
worker_cpu_affinities.size());
for (std::size_t worker_thread_index = 0;
worker_thread_index < worker_cpu_affinities.size();
[08/12] incubator-quickstep git commit: Added back the debug build
with llvm.
Posted by zu...@apache.org.
Added back the debug build with llvm.
Project: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/commit/a31dde4e
Tree: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/tree/a31dde4e
Diff: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/diff/a31dde4e
Branch: refs/heads/multi-shiftboss-test
Commit: a31dde4eaab6122768e21812b958a05d3048e50f
Parents: 3047d89
Author: Zuyu Zhang <zu...@apache.org>
Authored: Fri Nov 25 21:52:43 2016 -0800
Committer: Zuyu Zhang <zu...@apache.org>
Committed: Sat Nov 26 23:49:57 2016 -0800
----------------------------------------------------------------------
.travis.yml | 11 ++++++++++-
1 file changed, 10 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/a31dde4e/.travis.yml
----------------------------------------------------------------------
diff --git a/.travis.yml b/.travis.yml
index 9b43b16..8915eeb 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -1,4 +1,4 @@
-# NOTE(quickstep-team): In Travis-CI, jobs timeout if they take more than 120
+# NOTE(quickstep-team): In Travis-CI, jobs timeout if they take more than 50
# mins or if there is no log output for more than 10 mins. Hence, we use -O0 to
# speed up compilation in release build. Also, jobs can only use upto 20GB of
# disk space. Hence, we minimize the amount of debug symbol using -g0 (DEBUG
@@ -15,9 +15,18 @@ compiler:
- clang
env:
+ - BUILD_TYPE=Debug VECTOR_COPY_ELISION_LEVEL=joinwithbinaryexpressions
+ - BUILD_TYPE=Debug VECTOR_COPY_ELISION_LEVEL=selection
- BUILD_TYPE=Release VECTOR_COPY_ELISION_LEVEL=joinwithbinaryexpressions
- BUILD_TYPE=Release VECTOR_COPY_ELISION_LEVEL=selection
+matrix:
+ exclude: # Due to time-out.
+ - compiler: gcc
+ env: BUILD_TYPE=Debug VECTOR_COPY_ELISION_LEVEL=joinwithbinaryexpressions
+ - compiler: gcc
+ env: BUILD_TYPE=Debug VECTOR_COPY_ELISION_LEVEL=selection
+
install:
- export TEST_JOBS=2;
- if [ "$CC" = "gcc" ]; then
[03/12] incubator-quickstep git commit: Refactored the database init
code.
Posted by zu...@apache.org.
Refactored the database init code.
Project: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/commit/f673f945
Tree: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/tree/f673f945
Diff: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/diff/f673f945
Branch: refs/heads/multi-shiftboss-test
Commit: f673f94561e104c62d0441f1ded04737e23b0ba0
Parents: 54e0654
Author: Zuyu Zhang <zu...@apache.org>
Authored: Sun Nov 20 23:55:22 2016 -0800
Committer: Zuyu Zhang <zu...@apache.org>
Committed: Sun Nov 20 23:55:22 2016 -0800
----------------------------------------------------------------------
cli/CMakeLists.txt | 8 +++-
cli/DefaultsConfigurator.cpp | 77 +++++++++++++++++++++++++++++++++++++++
cli/DefaultsConfigurator.hpp | 10 +++++
cli/QuickstepCli.cpp | 36 +-----------------
4 files changed, 94 insertions(+), 37 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/f673f945/cli/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/cli/CMakeLists.txt b/cli/CMakeLists.txt
index 9847787..9b62af9 100644
--- a/cli/CMakeLists.txt
+++ b/cli/CMakeLists.txt
@@ -75,7 +75,7 @@ else()
quickstep_utility_Macros)
endif()
-add_library(quickstep_cli_DefaultsConfigurator ../empty_src.cpp DefaultsConfigurator.hpp)
+add_library(quickstep_cli_DefaultsConfigurator DefaultsConfigurator.cpp DefaultsConfigurator.hpp)
add_library(quickstep_cli_InputParserUtil InputParserUtil.cpp InputParserUtil.hpp)
add_library(quickstep_cli_PrintToScreen PrintToScreen.cpp PrintToScreen.hpp)
@@ -106,10 +106,14 @@ target_link_libraries(quickstep_cli_CommandExecutor
quickstep_utility_SqlError)
target_link_libraries(quickstep_cli_DefaultsConfigurator
+ glog
+ quickstep_catalog_Catalog
+ quickstep_catalog_Catalog_proto
+ quickstep_catalog_CatalogDatabase
quickstep_utility_Macros)
if(QUICKSTEP_HAVE_LIBNUMA)
target_link_libraries(quickstep_cli_DefaultsConfigurator
- ${LIBNUMA_LIBRARY})
+ ${LIBNUMA_LIBRARY})
endif()
target_link_libraries(quickstep_cli_Flags
quickstep_cli_DefaultsConfigurator
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/f673f945/cli/DefaultsConfigurator.cpp
----------------------------------------------------------------------
diff --git a/cli/DefaultsConfigurator.cpp b/cli/DefaultsConfigurator.cpp
new file mode 100644
index 0000000..355031c
--- /dev/null
+++ b/cli/DefaultsConfigurator.cpp
@@ -0,0 +1,77 @@
+/**
+ * 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 "cli/DefaultsConfigurator.hpp"
+
+#include "cli/CliConfig.h" // For QUICKSTEP_OS_WINDOWS.
+
+#ifdef QUICKSTEP_OS_WINDOWS
+#include <experimental/filesystem>
+#else
+#include <cstdlib>
+#endif // QUICKSTEP_OS_WINDOWS
+
+#include <fstream>
+#include <string>
+
+#include "catalog/Catalog.hpp"
+#include "catalog/Catalog.pb.h"
+#include "catalog/CatalogDatabase.hpp"
+
+#include "glog/logging.h"
+
+using std::string;
+
+namespace quickstep {
+
+void DefaultsConfigurator::InitializeDefaultDatabase(const string &storage_path, const string &catalog_path) {
+ // TODO(jmp): Refactor the code in this file!
+ LOG(INFO) << "Initializing the database, creating a new catalog file and storage directory";
+
+ // Create the directory
+ // TODO(jmp): At some point, likely in C++-17, we will just have the
+ // filesystem path, and we can clean this up
+#ifdef QUICKSTEP_OS_WINDOWS
+ CHECK(std::experimental::filesystem::create_directories(storage_path))
+ << "Failed when attempting to create the directory: " << storage_path
+ << "\nCheck if the directory already exists. If so, delete it or move it before initializing";
+#else
+ {
+ const string path_name = "mkdir " + storage_path;
+ CHECK(std::system(path_name.c_str()))
+ << "Failed when attempting to create the directory: " << storage_path;
+ }
+#endif // QUICKSTEP_OS_WINDOWS
+
+ // Create the default catalog file.
+ std::ofstream catalog_file(catalog_path.c_str());
+ CHECK(catalog_file.good())
+ << "ERROR: Unable to open " << catalog_path << " for writing.";
+
+ Catalog catalog;
+ catalog.addDatabase(new CatalogDatabase(nullptr, "default"));
+
+ CHECK(catalog.getProto().SerializeToOstream(&catalog_file))
+ << "ERROR: Unable to serialize catalog proto to file " << catalog_path;
+
+ // Close the catalog file - it will be reopened below by the QueryProcessor.
+ catalog_file.close();
+}
+
+} // namespace quickstep
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/f673f945/cli/DefaultsConfigurator.hpp
----------------------------------------------------------------------
diff --git a/cli/DefaultsConfigurator.hpp b/cli/DefaultsConfigurator.hpp
index 4fd62ac..4b534d6 100644
--- a/cli/DefaultsConfigurator.hpp
+++ b/cli/DefaultsConfigurator.hpp
@@ -27,6 +27,7 @@
#endif // QUICKSTEP_HAVE_LIBNUMA
#include <cstddef>
+#include <string>
#include <thread> // NOLINT(build/c++11)
#ifdef QUICKSTEP_HAVE_LIBNUMA
@@ -106,6 +107,15 @@ class DefaultsConfigurator {
return 1;
}
+ /**
+ * @brief Initialize the default database with no relations.
+ *
+ * @param storage_path The filesystem directory to store catalog.
+ * @param catalog_path The full path of the catalog file.
+ **/
+ static void InitializeDefaultDatabase(const std::string &storage_path,
+ const std::string &catalog_path);
+
private:
/**
* @brief Private constructor to disable instantiation of the class.
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/f673f945/cli/QuickstepCli.cpp
----------------------------------------------------------------------
diff --git a/cli/QuickstepCli.cpp b/cli/QuickstepCli.cpp
index 6a7c500..465793c 100644
--- a/cli/QuickstepCli.cpp
+++ b/cli/QuickstepCli.cpp
@@ -216,41 +216,7 @@ int main(int argc, char* argv[]) {
string catalog_path(FLAGS_storage_path);
catalog_path.append(kCatalogFilename);
if (quickstep::FLAGS_initialize_db) { // Initialize the database
- // TODO(jmp): Refactor the code in this file!
- LOG(INFO) << "Initializing the database, creating a new catalog file and storage directory\n";
-
- // Create the directory
- // TODO(jmp): At some point, likely in C++-17, we will just have the
- // filesystem path, and we can clean this up
-#ifdef QUICKSTEP_OS_WINDOWS
- std::filesystem::create_directories(FLAGS_storage_path);
- LOG(FATAL) << "Failed when attempting to create the directory: " << FLAGS_storage_path << "\n";
- LOG(FATAL) << "Check if the directory already exists. If so, delete it or move it before initializing \n";
-#else
- {
- string path_name = "mkdir " + FLAGS_storage_path;
- if (std::system(path_name.c_str())) {
- LOG(FATAL) << "Failed when attempting to create the directory: " << FLAGS_storage_path << "\n";
- }
- }
-#endif
-
- // Create the default catalog file.
- std::ofstream catalog_file(catalog_path);
- if (!catalog_file.good()) {
- LOG(FATAL) << "ERROR: Unable to open " << kCatalogFilename << " for writing.\n";
- }
-
- quickstep::Catalog catalog;
- catalog.addDatabase(new quickstep::CatalogDatabase(nullptr, "default"));
-
- if (!catalog.getProto().SerializeToOstream(&catalog_file)) {
- LOG(FATAL) << "ERROR: Unable to serialize catalog proto to file " << kCatalogFilename;
- return 1;
- }
-
- // Close the catalog file - it will be reopened below by the QueryProcessor.
- catalog_file.close();
+ DefaultsConfigurator::InitializeDefaultDatabase(FLAGS_storage_path, catalog_path);
}
// Setup QueryProcessor, including CatalogDatabase.
[10/12] incubator-quickstep git commit: Used multiple Shiftbosses in
the distributed unit tests.
Posted by zu...@apache.org.
Used multiple Shiftbosses in the distributed unit tests.
Project: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/commit/76cf1897
Tree: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/tree/76cf1897
Diff: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/diff/76cf1897
Branch: refs/heads/multi-shiftboss-test
Commit: 76cf1897ab795caa3034800cbc67a312d6ceac22
Parents: a31dde4
Author: Zuyu Zhang <zu...@apache.org>
Authored: Sat Nov 12 22:31:42 2016 -0800
Committer: Zuyu Zhang <zu...@apache.org>
Committed: Sun Nov 27 10:50:04 2016 -0800
----------------------------------------------------------------------
query_execution/ForemanDistributed.cpp | 14 +++---
query_execution/PolicyEnforcerDistributed.cpp | 6 ++-
query_execution/QueryExecutionState.hpp | 53 ++++++++++++++++++++++
query_execution/QueryManagerDistributed.cpp | 35 ++++++++------
query_execution/QueryManagerDistributed.hpp | 4 +-
5 files changed, 88 insertions(+), 24 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/76cf1897/query_execution/ForemanDistributed.cpp
----------------------------------------------------------------------
diff --git a/query_execution/ForemanDistributed.cpp b/query_execution/ForemanDistributed.cpp
index d619657..7dccce4 100644
--- a/query_execution/ForemanDistributed.cpp
+++ b/query_execution/ForemanDistributed.cpp
@@ -238,17 +238,19 @@ void ForemanDistributed::run() {
bool ForemanDistributed::canCollectNewMessages(const tmb::message_type_id message_type) {
return !QUICKSTEP_EQUALS_ANY_CONSTANT(message_type,
kCatalogRelationNewBlockMessage,
- kWorkOrderFeedbackMessage) &&
- // TODO(zuyu): Multiple Shiftbosses support.
- !shiftboss_directory_.hasReachedCapacity(0);
+ kWorkOrderFeedbackMessage);
}
void ForemanDistributed::dispatchWorkOrderMessages(const vector<unique_ptr<S::WorkOrderMessage>> &messages) {
+ const size_t num_shiftbosses = shiftboss_directory_.size();
+ size_t shiftboss_index = 0u;
for (const auto &message : messages) {
DCHECK(message != nullptr);
- // TODO(zuyu): Multiple Shiftbosses support.
- sendWorkOrderMessage(0, *message);
- shiftboss_directory_.incrementNumQueuedWorkOrders(0);
+ sendWorkOrderMessage(shiftboss_index, *message);
+ shiftboss_directory_.incrementNumQueuedWorkOrders(shiftboss_index);
+
+ // TO(zuyu): Take data-locality into account for scheduling.
+ shiftboss_index = (shiftboss_index + 1) % num_shiftbosses;
}
}
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/76cf1897/query_execution/PolicyEnforcerDistributed.cpp
----------------------------------------------------------------------
diff --git a/query_execution/PolicyEnforcerDistributed.cpp b/query_execution/PolicyEnforcerDistributed.cpp
index c06fd86..6e09ea8 100644
--- a/query_execution/PolicyEnforcerDistributed.cpp
+++ b/query_execution/PolicyEnforcerDistributed.cpp
@@ -140,8 +140,10 @@ void PolicyEnforcerDistributed::processInitiateRebuildResponseMessage(const tmb:
QueryManagerDistributed *query_manager = static_cast<QueryManagerDistributed*>(admitted_queries_[query_id].get());
const std::size_t num_rebuild_work_orders = proto.num_rebuild_work_orders();
- query_manager->processInitiateRebuildResponseMessage(proto.operator_index(), num_rebuild_work_orders);
- shiftboss_directory_->addNumQueuedWorkOrders(proto.shiftboss_index(), num_rebuild_work_orders);
+ const size_t shiftboss_index = proto.shiftboss_index();
+ query_manager->processInitiateRebuildResponseMessage(
+ proto.operator_index(), num_rebuild_work_orders, shiftboss_index);
+ shiftboss_directory_->addNumQueuedWorkOrders(shiftboss_index, num_rebuild_work_orders);
if (query_manager->getQueryExecutionState().hasQueryExecutionFinished()) {
onQueryCompletion(query_manager);
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/76cf1897/query_execution/QueryExecutionState.hpp
----------------------------------------------------------------------
diff --git a/query_execution/QueryExecutionState.hpp b/query_execution/QueryExecutionState.hpp
index f5281d5..a4273dc 100644
--- a/query_execution/QueryExecutionState.hpp
+++ b/query_execution/QueryExecutionState.hpp
@@ -22,6 +22,12 @@
#include <cstddef>
#include <unordered_map>
+
+#include "query_optimizer/QueryOptimizerConfig.h" // For QUICKSTEP_DISTRIBUTED.
+#ifdef QUICKSTEP_DISTRIBUTED
+#include <unordered_set>
+#endif // QUICKSTEP_DISTRIBUTED
+
#include <utility>
#include <vector>
@@ -103,6 +109,49 @@ class QueryExecutionState {
}
}
+#ifdef QUICKSTEP_DISTRIBUTED
+ /**
+ * @brief Update the rebuild status of the given operator the number of
+ * pending rebuild work orders, after the rebuild has been initiated.
+ *
+ * @param operator_index The index of the given operator.
+ * @param num_rebuild_workorders The number of rebuild workorders of the given
+ * operator.
+ * @param shiftboss_index The index of the Shiftboss that rebuilt.
+ **/
+ void updateRebuildStatus(const std::size_t operator_index,
+ const std::size_t num_rebuild_workorders,
+ const std::size_t shiftboss_index) {
+ DCHECK_LT(operator_index, num_operators_);
+ auto search_res = rebuild_status_.find(operator_index);
+ DCHECK(search_res != rebuild_status_.end() && search_res->second.has_initiated);
+ search_res->second.num_pending_workorders += num_rebuild_workorders;
+ search_res->second.rebuilt_shiftboss_indexes.insert(shiftboss_index);
+ }
+
+ /**
+ * @brief Check if the rebuild has been finished for the given operator.
+ *
+ * @param operator_index The index of the given operator.
+ * @param num_shiftbosses The number of the Shiftbosses for rebuilt.
+ *
+ * @return True if the rebuild has been finished, false otherwise.
+ **/
+ inline bool hasRebuildFinished(const std::size_t operator_index,
+ const std::size_t num_shiftbosses) const {
+ DCHECK_LT(operator_index, num_operators_);
+ const auto search_res = rebuild_status_.find(operator_index);
+ DCHECK(search_res != rebuild_status_.end());
+
+ const auto &rebuild_status = search_res->second;
+ DCHECK(rebuild_status.has_initiated);
+
+ return rebuild_status.rebuilt_shiftboss_indexes.size() == num_shiftbosses &&
+ rebuild_status.num_pending_workorders == 0u;
+ }
+
+#endif // QUICKSTEP_DISTRIBUTED
+
/**
* @brief Check if the rebuild has been initiated for the given operator.
*
@@ -314,6 +363,10 @@ class QueryExecutionState {
// The number of pending rebuild workorders for the operator.
// Valid if and only if 'has_initiated' is true.
std::size_t num_pending_workorders;
+
+#ifdef QUICKSTEP_DISTRIBUTED
+ std::unordered_set<std::size_t> rebuilt_shiftboss_indexes;
+#endif // QUICKSTEP_DISTRIBUTED
};
// Key is dag_node_index for which rebuild is required.
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/76cf1897/query_execution/QueryManagerDistributed.cpp
----------------------------------------------------------------------
diff --git a/query_execution/QueryManagerDistributed.cpp b/query_execution/QueryManagerDistributed.cpp
index 7d45933..20650d0 100644
--- a/query_execution/QueryManagerDistributed.cpp
+++ b/query_execution/QueryManagerDistributed.cpp
@@ -36,7 +36,9 @@
#include "glog/logging.h"
+#include "tmb/address.h"
#include "tmb/id_typedefs.h"
+#include "tmb/tagged_message.h"
using std::free;
using std::malloc;
@@ -125,16 +127,16 @@ bool QueryManagerDistributed::fetchNormalWorkOrders(const dag_node_index index)
}
void QueryManagerDistributed::processInitiateRebuildResponseMessage(const dag_node_index op_index,
- const std::size_t num_rebuild_work_orders) {
- // TODO(zuyu): Multiple Shiftbosses support.
- query_exec_state_->setRebuildStatus(op_index, num_rebuild_work_orders, true);
+ const std::size_t num_rebuild_work_orders,
+ const std::size_t shiftboss_index) {
+ query_exec_state_->updateRebuildStatus(op_index, num_rebuild_work_orders, shiftboss_index);
- if (num_rebuild_work_orders != 0u) {
+ if (!query_exec_state_->hasRebuildFinished(op_index, shiftboss_directory_->size())) {
// Wait for the rebuild work orders to finish.
return;
}
- // No needs for rebuilds.
+ // No needs for rebuilds, or the rebuild has finished.
markOperatorFinished(op_index);
for (const std::pair<dag_node_index, bool> &dependent_link :
@@ -168,17 +170,20 @@ bool QueryManagerDistributed::initiateRebuild(const dag_node_index index) {
kInitiateRebuildMessage);
free(proto_bytes);
+ // TODO(quickstep-team): Dynamically scale-up/down Shiftbosses.
+ tmb::Address shiftboss_addresses;
+ for (std::size_t i = 0; i < shiftboss_directory_->size(); ++i) {
+ shiftboss_addresses.AddRecipient(shiftboss_directory_->getClientId(i));
+ }
+
LOG(INFO) << "ForemanDistributed sent InitiateRebuildMessage (typed '" << kInitiateRebuildMessage
- << "') to Shiftboss";
- // TODO(zuyu): Multiple workers support.
- QueryExecutionUtil::SendTMBMessage(bus_,
- foreman_client_id_,
- shiftboss_directory_->getClientId(0),
- move(tagged_msg));
-
- // The negative value indicates that the number of rebuild work orders is to be
- // determined.
- query_exec_state_->setRebuildStatus(index, -1, true);
+ << "') to all Shiftbosses";
+ QueryExecutionUtil::BroadcastMessage(foreman_client_id_,
+ shiftboss_addresses,
+ move(tagged_msg),
+ bus_);
+
+ query_exec_state_->setRebuildStatus(index, 0, true);
// Wait for Shiftbosses to report the number of rebuild work orders.
return false;
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/76cf1897/query_execution/QueryManagerDistributed.hpp
----------------------------------------------------------------------
diff --git a/query_execution/QueryManagerDistributed.hpp b/query_execution/QueryManagerDistributed.hpp
index e609ab8..f8ac53c 100644
--- a/query_execution/QueryManagerDistributed.hpp
+++ b/query_execution/QueryManagerDistributed.hpp
@@ -73,9 +73,11 @@ class QueryManagerDistributed final : public QueryManagerBase {
* for initiating the rebuild work order.
* @param num_rebuild_work_orders The number of the rebuild work orders
* generated for the operator indexed by 'op_index'.
+ * @param shiftboss_index The index of the Shiftboss that sends the message.
**/
void processInitiateRebuildResponseMessage(const dag_node_index op_index,
- const std::size_t num_rebuild_work_orders);
+ const std::size_t num_rebuild_work_orders,
+ const std::size_t shiftboss_index);
/**
* @brief Get the next normal workorder to be excuted, wrapped in a
[09/12] incubator-quickstep git commit: Scheduled HashJoinWorkOrder
on the same Shiftboss of BuildHashWorkOrder.
Posted by zu...@apache.org.
Scheduled HashJoinWorkOrder on the same Shiftboss of BuildHashWorkOrder.
Project: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/commit/4d935f40
Tree: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/tree/4d935f40
Diff: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/diff/4d935f40
Branch: refs/heads/multi-shiftboss-test
Commit: 4d935f40bbe7a913c84eadfdaa6a63ceaf55b34d
Parents: 76cf189
Author: Zuyu Zhang <zu...@apache.org>
Authored: Sun Nov 13 20:20:56 2016 -0800
Committer: Zuyu Zhang <zu...@apache.org>
Committed: Sun Nov 27 10:50:04 2016 -0800
----------------------------------------------------------------------
query_execution/CMakeLists.txt | 3 ++
query_execution/ForemanDistributed.cpp | 47 ++++++++++++++++++++--
query_execution/ForemanDistributed.hpp | 4 ++
query_execution/PolicyEnforcerDistributed.cpp | 12 ++++++
query_execution/PolicyEnforcerDistributed.hpp | 19 +++++++++
query_execution/QueryManagerDistributed.hpp | 25 ++++++++++++
6 files changed, 106 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/4d935f40/query_execution/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/query_execution/CMakeLists.txt b/query_execution/CMakeLists.txt
index 719d9f3..1f7add8 100644
--- a/query_execution/CMakeLists.txt
+++ b/query_execution/CMakeLists.txt
@@ -108,10 +108,12 @@ if (ENABLE_DISTRIBUTED)
quickstep_queryexecution_ForemanBase
quickstep_queryexecution_PolicyEnforcerBase
quickstep_queryexecution_PolicyEnforcerDistributed
+ quickstep_queryexecution_QueryContext
quickstep_queryexecution_QueryExecutionMessages_proto
quickstep_queryexecution_QueryExecutionTypedefs
quickstep_queryexecution_QueryExecutionUtil
quickstep_queryexecution_ShiftbossDirectory
+ quickstep_relationaloperators_WorkOrder_proto
quickstep_threading_ThreadUtil
quickstep_utility_EqualsAnyConstant
quickstep_utility_Macros
@@ -153,6 +155,7 @@ if (ENABLE_DISTRIBUTED)
quickstep_catalog_CatalogRelation
quickstep_catalog_Catalog_proto
quickstep_queryexecution_PolicyEnforcerBase
+ quickstep_queryexecution_QueryContext
quickstep_queryexecution_QueryContext_proto
quickstep_queryexecution_QueryExecutionMessages_proto
quickstep_queryexecution_QueryExecutionState
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/4d935f40/query_execution/ForemanDistributed.cpp
----------------------------------------------------------------------
diff --git a/query_execution/ForemanDistributed.cpp b/query_execution/ForemanDistributed.cpp
index 7dccce4..900a71f 100644
--- a/query_execution/ForemanDistributed.cpp
+++ b/query_execution/ForemanDistributed.cpp
@@ -30,10 +30,12 @@
#include "query_execution/AdmitRequestMessage.hpp"
#include "query_execution/PolicyEnforcerBase.hpp"
#include "query_execution/PolicyEnforcerDistributed.hpp"
+#include "query_execution/QueryContext.hpp"
#include "query_execution/QueryExecutionMessages.pb.h"
#include "query_execution/QueryExecutionTypedefs.hpp"
#include "query_execution/QueryExecutionUtil.hpp"
#include "query_execution/ShiftbossDirectory.hpp"
+#include "relational_operators/WorkOrder.pb.h"
#include "threading/ThreadUtil.hpp"
#include "utility/EqualsAnyConstant.hpp"
@@ -241,16 +243,53 @@ bool ForemanDistributed::canCollectNewMessages(const tmb::message_type_id messag
kWorkOrderFeedbackMessage);
}
+bool ForemanDistributed::isHashJoinRelatedWorkOrder(const unique_ptr<S::WorkOrderMessage> &message,
+ const size_t next_shiftboss_index_to_schedule,
+ size_t *shiftboss_index_for_hash_join) {
+ const S::WorkOrder &work_order_proto = message->work_order();
+ QueryContext::join_hash_table_id join_hash_table_index;
+
+ switch (work_order_proto.work_order_type()) {
+ case S::BUILD_HASH:
+ join_hash_table_index = work_order_proto.GetExtension(serialization::BuildHashWorkOrder::join_hash_table_index);
+ break;
+ case S::DESTROY_HASH:
+ join_hash_table_index = work_order_proto.GetExtension(serialization::DestroyHashWorkOrder::join_hash_table_index);
+ break;
+ case S::HASH_JOIN:
+ join_hash_table_index = work_order_proto.GetExtension(serialization::HashJoinWorkOrder::join_hash_table_index);
+ break;
+ default:
+ return false;
+ }
+
+ policy_enforcer_->getShiftbossIndexForHashJoin(
+ message->query_id(), join_hash_table_index, next_shiftboss_index_to_schedule,
+ shiftboss_index_for_hash_join);
+
+ return true;
+}
+
void ForemanDistributed::dispatchWorkOrderMessages(const vector<unique_ptr<S::WorkOrderMessage>> &messages) {
const size_t num_shiftbosses = shiftboss_directory_.size();
size_t shiftboss_index = 0u;
for (const auto &message : messages) {
DCHECK(message != nullptr);
- sendWorkOrderMessage(shiftboss_index, *message);
- shiftboss_directory_.incrementNumQueuedWorkOrders(shiftboss_index);
+ size_t shiftboss_index_for_hash_join;
+ if (isHashJoinRelatedWorkOrder(message, shiftboss_index, &shiftboss_index_for_hash_join)) {
+ sendWorkOrderMessage(shiftboss_index_for_hash_join, *message);
+ shiftboss_directory_.incrementNumQueuedWorkOrders(shiftboss_index_for_hash_join);
+
+ if (shiftboss_index == shiftboss_index_for_hash_join) {
+ shiftboss_index = (shiftboss_index + 1) % num_shiftbosses;
+ }
+ } else {
+ sendWorkOrderMessage(shiftboss_index, *message);
+ shiftboss_directory_.incrementNumQueuedWorkOrders(shiftboss_index);
- // TO(zuyu): Take data-locality into account for scheduling.
- shiftboss_index = (shiftboss_index + 1) % num_shiftbosses;
+ // TODO(zuyu): Take data-locality into account for scheduling.
+ shiftboss_index = (shiftboss_index + 1) % num_shiftbosses;
+ }
}
}
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/4d935f40/query_execution/ForemanDistributed.hpp
----------------------------------------------------------------------
diff --git a/query_execution/ForemanDistributed.hpp b/query_execution/ForemanDistributed.hpp
index ccdd0ae..55e747f 100644
--- a/query_execution/ForemanDistributed.hpp
+++ b/query_execution/ForemanDistributed.hpp
@@ -71,6 +71,10 @@ class ForemanDistributed final : public ForemanBase {
void run() override;
private:
+ bool isHashJoinRelatedWorkOrder(const std::unique_ptr<serialization::WorkOrderMessage> &message,
+ const std::size_t next_shiftboss_index_to_schedule,
+ std::size_t *shiftboss_index_for_hash_join);
+
/**
* @brief Dispatch schedulable WorkOrders, wrapped in WorkOrderMessages to the
* worker threads.
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/4d935f40/query_execution/PolicyEnforcerDistributed.cpp
----------------------------------------------------------------------
diff --git a/query_execution/PolicyEnforcerDistributed.cpp b/query_execution/PolicyEnforcerDistributed.cpp
index 6e09ea8..86b36c8 100644
--- a/query_execution/PolicyEnforcerDistributed.cpp
+++ b/query_execution/PolicyEnforcerDistributed.cpp
@@ -158,6 +158,18 @@ void PolicyEnforcerDistributed::processInitiateRebuildResponseMessage(const tmb:
}
}
+void PolicyEnforcerDistributed::getShiftbossIndexForHashJoin(
+ const std::size_t query_id,
+ const QueryContext::join_hash_table_id join_hash_table_index,
+ const std::size_t next_shiftboss_index_to_schedule,
+ std::size_t *shiftboss_index) {
+ DCHECK(admitted_queries_.find(query_id) != admitted_queries_.end());
+ QueryManagerDistributed *query_manager = static_cast<QueryManagerDistributed*>(admitted_queries_[query_id].get());
+ query_manager->getShiftbossIndexForHashJoin(join_hash_table_index,
+ next_shiftboss_index_to_schedule,
+ shiftboss_index);
+}
+
void PolicyEnforcerDistributed::initiateQueryInShiftboss(QueryHandle *query_handle) {
S::QueryInitiateMessage proto;
proto.set_query_id(query_handle->query_id());
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/4d935f40/query_execution/PolicyEnforcerDistributed.hpp
----------------------------------------------------------------------
diff --git a/query_execution/PolicyEnforcerDistributed.hpp b/query_execution/PolicyEnforcerDistributed.hpp
index 146e9af..37326bd 100644
--- a/query_execution/PolicyEnforcerDistributed.hpp
+++ b/query_execution/PolicyEnforcerDistributed.hpp
@@ -20,6 +20,7 @@
#include <vector>
#include "query_execution/PolicyEnforcerBase.hpp"
+#include "query_execution/QueryContext.hpp"
#include "query_execution/QueryExecutionMessages.pb.h"
#include "query_execution/ShiftbossDirectory.hpp"
#include "utility/Macros.hpp"
@@ -88,6 +89,24 @@ class PolicyEnforcerDistributed final : public PolicyEnforcerBase {
**/
void processInitiateRebuildResponseMessage(const tmb::TaggedMessage &tagged_message);
+ /**
+ * @brief Get or set the index of Shiftboss for a HashJoin related WorkOrder.
+ * If it is the first BuildHash on <join_hash_table_index>, <shiftboss_index>
+ * will be set to <next_shiftboss_index_to_schedule>. Otherwise,
+ * <shiftboss_index> will be set to the index of the Shiftboss that has
+ * executed the first BuildHash.
+ *
+ * @param query_id The query id.
+ * @param join_hash_table_index The Hash Table for the Join.
+ * @param next_shiftboss_index The index of Shiftboss to schedule a next WorkOrder.
+ * @param shiftboss_index The index of Shiftboss to schedule the WorkOrder.
+ **/
+ void getShiftbossIndexForHashJoin(
+ const std::size_t query_id,
+ const QueryContext::join_hash_table_id join_hash_table_index,
+ const std::size_t next_shiftboss_index_to_schedule,
+ std::size_t *shiftboss_index);
+
private:
void decrementNumQueuedWorkOrders(const serialization::WorkOrderCompletionMessage &proto) override {
shiftboss_directory_->decrementNumQueuedWorkOrders(proto.shiftboss_index());
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/4d935f40/query_execution/QueryManagerDistributed.hpp
----------------------------------------------------------------------
diff --git a/query_execution/QueryManagerDistributed.hpp b/query_execution/QueryManagerDistributed.hpp
index f8ac53c..2b21303 100644
--- a/query_execution/QueryManagerDistributed.hpp
+++ b/query_execution/QueryManagerDistributed.hpp
@@ -22,7 +22,9 @@
#include <cstddef>
#include <memory>
+#include <unordered_map>
+#include "query_execution/QueryContext.hpp"
#include "query_execution/QueryExecutionState.hpp"
#include "query_execution/QueryManagerBase.hpp"
#include "query_execution/WorkOrderProtosContainer.hpp"
@@ -92,6 +94,26 @@ class QueryManagerDistributed final : public QueryManagerBase {
serialization::WorkOrderMessage* getNextWorkOrderMessage(
const dag_node_index start_operator_index);
+ /**
+ * @brief Get the index of Shiftboss for a HashJoin related WorkOrder. If the
+ * Shiftboss index is not found, set using <next_shiftboss_index_to_schedule>.
+ *
+ * @param join_hash_table_index The Hash Table for the Join.
+ * @param next_shiftboss_index The index of Shiftboss to schedule a next WorkOrder.
+ * @param shiftboss_index The index of Shiftboss to schedule the WorkOrder.
+ **/
+ void getShiftbossIndexForHashJoin(const QueryContext::join_hash_table_id join_hash_table_index,
+ const std::size_t next_shiftboss_index_to_schedule,
+ std::size_t *shiftboss_index) {
+ const auto cit = shiftboss_indexes_for_hash_joins_.find(join_hash_table_index);
+ if (cit != shiftboss_indexes_for_hash_joins_.end()) {
+ *shiftboss_index = cit->second;
+ } else {
+ shiftboss_indexes_for_hash_joins_.emplace(join_hash_table_index, next_shiftboss_index_to_schedule);
+ *shiftboss_index = next_shiftboss_index_to_schedule;
+ }
+ }
+
private:
bool checkNormalExecutionOver(const dag_node_index index) const override {
return (checkAllDependenciesMet(index) &&
@@ -114,6 +136,9 @@ class QueryManagerDistributed final : public QueryManagerBase {
std::unique_ptr<WorkOrderProtosContainer> normal_workorder_protos_container_;
+ // A map from a join hash table to its scheduled Shiftboss index.
+ std::unordered_map<QueryContext::join_hash_table_id, std::size_t> shiftboss_indexes_for_hash_joins_;
+
DISALLOW_COPY_AND_ASSIGN(QueryManagerDistributed);
};
[06/12] incubator-quickstep git commit: Adds support for
PartialBulkInserts in StorageBlocks
Posted by zu...@apache.org.
Adds support for PartialBulkInserts in StorageBlocks
- Enables use of PartialBulkInserts in StorageBlocks
- Value accessor changes to allow use of 2 insert destinations
- Enables PartialInserts for SplitRow
- Changes HashJoin operator so that it can take advantage of the
PartialInserts code.
- This also cleans up code from Previous commit.
Project: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/commit/2d11ec58
Tree: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/tree/2d11ec58
Diff: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/diff/2d11ec58
Branch: refs/heads/multi-shiftboss-test
Commit: 2d11ec588e7e8d7a7a1a8adfc28fb30e5fe0852e
Parents: 172b51b
Author: navsan <na...@gmail.com>
Authored: Mon Nov 7 14:35:01 2016 -0600
Committer: cramja <ma...@gmail.com>
Committed: Mon Nov 21 14:28:02 2016 -0600
----------------------------------------------------------------------
relational_operators/HashJoinOperator.cpp | 150 ++++++++++++++++---
storage/InsertDestination.cpp | 84 +++++++++++
storage/InsertDestination.hpp | 16 ++
storage/InsertDestinationInterface.hpp | 22 +++
storage/SplitRowStoreTupleStorageSubBlock.hpp | 4 +-
storage/StorageBlock.cpp | 24 +++
storage/StorageBlock.hpp | 44 ++++++
storage/TupleStorageSubBlock.hpp | 50 +++++++
types/containers/ColumnVectorsValueAccessor.hpp | 4 +
9 files changed, 373 insertions(+), 25 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/2d11ec58/relational_operators/HashJoinOperator.cpp
----------------------------------------------------------------------
diff --git a/relational_operators/HashJoinOperator.cpp b/relational_operators/HashJoinOperator.cpp
index 4a91f86..2028046 100644
--- a/relational_operators/HashJoinOperator.cpp
+++ b/relational_operators/HashJoinOperator.cpp
@@ -65,10 +65,11 @@ namespace {
// Functor passed to HashTable::getAllFromValueAccessor() to collect matching
// tuples from the inner relation. It stores matching tuple ID pairs
-// in an unordered_map keyed by inner block ID.
-class MapBasedJoinedTupleCollector {
+// in an unordered_map keyed by inner block ID and a vector of
+// pairs of (build-tupleID, probe-tuple-ID).
+class VectorsOfPairsJoinedTuplesCollector {
public:
- MapBasedJoinedTupleCollector() {
+ VectorsOfPairsJoinedTuplesCollector() {
}
template <typename ValueAccessorT>
@@ -95,6 +96,34 @@ class MapBasedJoinedTupleCollector {
std::unordered_map<block_id, std::vector<std::pair<tuple_id, tuple_id>>> joined_tuples_;
};
+// Another collector using an unordered_map keyed on inner block just like above,
+// except that it uses of a pair of (build-tupleIDs-vector, probe-tuple-IDs-vector).
+class PairsOfVectorsJoinedTuplesCollector {
+ public:
+ PairsOfVectorsJoinedTuplesCollector() {
+ }
+
+ template <typename ValueAccessorT>
+ inline void operator()(const ValueAccessorT &accessor,
+ const TupleReference &tref) {
+ joined_tuples_[tref.block].first.push_back(tref.tuple);
+ joined_tuples_[tref.block].second.push_back(accessor.getCurrentPosition());
+ }
+
+ // Get a mutable pointer to the collected map of joined tuple ID pairs. The
+ // key is inner block_id, value is a pair consisting of
+ // inner block tuple IDs (first) and outer block tuple IDs (second).
+ inline std::unordered_map< block_id, std::pair<std::vector<tuple_id>, std::vector<tuple_id>>>*
+ getJoinedTuples() {
+ return &joined_tuples_;
+ }
+
+ private:
+ std::unordered_map<
+ block_id,
+ std::pair<std::vector<tuple_id>, std::vector<tuple_id>>> joined_tuples_;
+};
+
class SemiAntiJoinTupleCollector {
public:
explicit SemiAntiJoinTupleCollector(TupleIdSequence *filter)
@@ -432,7 +461,7 @@ void HashInnerJoinWorkOrder::execute() {
base_accessor->createSharedTupleIdSequenceAdapterVirtual(*existence_map));
}
- MapBasedJoinedTupleCollector collector;
+ PairsOfVectorsJoinedTuplesCollector collector;
if (join_key_attributes_.size() == 1) {
hash_table_.getAllFromValueAccessor(
probe_accessor.get(),
@@ -450,12 +479,14 @@ void HashInnerJoinWorkOrder::execute() {
const relation_id build_relation_id = build_relation_.getID();
const relation_id probe_relation_id = probe_relation_.getID();
- for (std::pair<const block_id, std::vector<std::pair<tuple_id, tuple_id>>>
+ for (std::pair<const block_id, std::pair<std::vector<tuple_id>, std::vector<tuple_id>>>
&build_block_entry : *collector.getJoinedTuples()) {
BlockReference build_block =
storage_manager_->getBlock(build_block_entry.first, build_relation_);
const TupleStorageSubBlock &build_store = build_block->getTupleStorageSubBlock();
std::unique_ptr<ValueAccessor> build_accessor(build_store.createValueAccessor());
+ const std::vector<tuple_id> &build_tids = build_block_entry.second.first;
+ const std::vector<tuple_id> &probe_tids = build_block_entry.second.second;
// Evaluate '*residual_predicate_', if any.
//
@@ -468,17 +499,16 @@ void HashInnerJoinWorkOrder::execute() {
// hash join is below a reasonable threshold so that we don't blow up
// temporary memory requirements to an unreasonable degree.
if (residual_predicate_ != nullptr) {
- std::vector<std::pair<tuple_id, tuple_id>> filtered_matches;
-
- for (const std::pair<tuple_id, tuple_id> &hash_match
- : build_block_entry.second) {
+ std::pair<std::vector<tuple_id>, std::vector<tuple_id>> filtered_matches;
+ for (std::size_t i = 0; i < build_tids.size(); ++i) {
if (residual_predicate_->matchesForJoinedTuples(*build_accessor,
build_relation_id,
- hash_match.first,
+ build_tids[i],
*probe_accessor,
probe_relation_id,
- hash_match.second)) {
- filtered_matches.emplace_back(hash_match);
+ probe_tids[i])) {
+ filtered_matches.first.push_back(build_tids[i]);
+ filtered_matches.second.push_back(probe_tids[i]);
}
}
@@ -501,22 +531,96 @@ void HashInnerJoinWorkOrder::execute() {
// benefit (probably only a real performance win when there are very few
// matching tuples in each individual inner block but very many inner
// blocks with at least one match).
+
+ // We now create ordered value accessors for both build and probe side,
+ // using the joined tuple TIDs. Note that we have to use this Lambda-based
+ // invocation method here because the accessors don't have a virtual
+ // function that creates such an OrderedTupleIdSequenceAdapterValueAccessor.
+ std::unique_ptr<ValueAccessor> ordered_build_accessor, ordered_probe_accessor;
+ InvokeOnValueAccessorNotAdapter(
+ build_accessor.get(),
+ [&](auto *accessor) -> void { // NOLINT(build/c++11)
+ ordered_build_accessor.reset(
+ accessor->createSharedOrderedTupleIdSequenceAdapter(build_tids));
+ });
+
+ if (probe_accessor->isTupleIdSequenceAdapter()) {
+ InvokeOnTupleIdSequenceAdapterValueAccessor(
+ probe_accessor.get(),
+ [&](auto *accessor) -> void { // NOLINT(build/c++11)
+ ordered_probe_accessor.reset(
+ accessor->createSharedOrderedTupleIdSequenceAdapter(probe_tids));
+ });
+ } else {
+ InvokeOnValueAccessorNotAdapter(
+ probe_accessor.get(),
+ [&](auto *accessor) -> void { // NOLINT(build/c++11)
+ ordered_probe_accessor.reset(
+ accessor->createSharedOrderedTupleIdSequenceAdapter(probe_tids));
+ });
+ }
+
+
+ // We also need a temp value accessor to store results of any scalar expressions.
ColumnVectorsValueAccessor temp_result;
- for (vector<unique_ptr<const Scalar>>::const_iterator selection_cit = selection_.begin();
- selection_cit != selection_.end();
- ++selection_cit) {
- temp_result.addColumn((*selection_cit)->getAllValuesForJoin(build_relation_id,
- build_accessor.get(),
- probe_relation_id,
- probe_accessor.get(),
- build_block_entry.second));
+
+ // Create a map of ValueAccessors and what attributes we want to pick from them
+ std::vector<std::pair<ValueAccessor *, std::vector<attribute_id>>> accessor_attribute_map;
+ const std::vector<ValueAccessor *> accessors{
+ ordered_build_accessor.get(), ordered_probe_accessor.get(), &temp_result};
+ const unsigned int build_index = 0, probe_index = 1, temp_index = 2;
+ for (auto &accessor : accessors) {
+ accessor_attribute_map.push_back(std::make_pair(
+ accessor,
+ std::vector<attribute_id>(selection_.size(), kInvalidCatalogId)));
+ }
+
+ attribute_id dest_attr = 0;
+ std::vector<std::pair<tuple_id, tuple_id>> zipped_joined_tuple_ids;
+
+ for (auto &selection_cit : selection_) {
+ // If the Scalar (column) is not an attribute in build/probe blocks, then
+ // insert it into a ColumnVectorsValueAccessor.
+ if (selection_cit->getDataSource() != Scalar::ScalarDataSource::kAttribute) {
+ // Current destination attribute maps to the column we'll create now.
+ accessor_attribute_map[temp_index].second[dest_attr] = temp_result.getNumColumns();
+
+ if (temp_result.getNumColumns() == 0) {
+ // The getAllValuesForJoin function below needs joined tuple IDs as
+ // a vector of pair of (build-tuple-ID, probe-tuple-ID), and we have
+ // a pair of (build-tuple-IDs-vector, probe-tuple-IDs-vector). So
+ // we'll have to zip our two vectors together. We do this inside
+ // the loop because most queries don't exercise this code since
+ // they don't have scalar expressions with attributes from both
+ // build and probe relations (other expressions would have been
+ // pushed down to before the join).
+ zipped_joined_tuple_ids.reserve(build_tids.size());
+ for (std::size_t i = 0; i < build_tids.size(); ++i) {
+ zipped_joined_tuple_ids.push_back(std::make_pair(build_tids[i], probe_tids[i]));
+ }
+ }
+ temp_result.addColumn(
+ selection_cit
+ ->getAllValuesForJoin(build_relation_id, build_accessor.get(),
+ probe_relation_id, probe_accessor.get(),
+ zipped_joined_tuple_ids));
+ } else {
+ auto scalar_attr = static_cast<const ScalarAttribute *>(selection_cit.get());
+ const attribute_id attr_id = scalar_attr->getAttribute().getID();
+ if (scalar_attr->getAttribute().getParent().getID() == build_relation_id) {
+ accessor_attribute_map[build_index].second[dest_attr] = attr_id;
+ } else {
+ accessor_attribute_map[probe_index].second[dest_attr] = attr_id;
+ }
+ }
+ ++dest_attr;
}
// NOTE(chasseur): calling the bulk-insert method of InsertDestination once
// for each pair of joined blocks incurs some extra overhead that could be
// avoided by keeping checked-out MutableBlockReferences across iterations
// of this loop, but that would get messy when combined with partitioning.
- output_destination_->bulkInsertTuples(&temp_result);
+ output_destination_->bulkInsertTuplesFromValueAccessors(accessor_attribute_map);
}
}
@@ -550,7 +654,7 @@ void HashSemiJoinWorkOrder::executeWithResidualPredicate() {
// We collect all the matching probe relation tuples, as there's a residual
// preidcate that needs to be applied after collecting these matches.
- MapBasedJoinedTupleCollector collector;
+ VectorsOfPairsJoinedTuplesCollector collector;
if (join_key_attributes_.size() == 1) {
hash_table_.getAllFromValueAccessor(
probe_accessor.get(),
@@ -759,7 +863,7 @@ void HashAntiJoinWorkOrder::executeWithResidualPredicate() {
base_accessor->createSharedTupleIdSequenceAdapterVirtual(*existence_map));
}
- MapBasedJoinedTupleCollector collector;
+ VectorsOfPairsJoinedTuplesCollector collector;
// We probe the hash table and get all the matches. Unlike
// executeWithoutResidualPredicate(), we have to collect all the matching
// tuples, because after this step we still have to evalute the residual
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/2d11ec58/storage/InsertDestination.cpp
----------------------------------------------------------------------
diff --git a/storage/InsertDestination.cpp b/storage/InsertDestination.cpp
index 5e83453..067edf6 100644
--- a/storage/InsertDestination.cpp
+++ b/storage/InsertDestination.cpp
@@ -247,6 +247,90 @@ void InsertDestination::bulkInsertTuplesWithRemappedAttributes(
});
}
+// A common case that we can optimize away is when the attribute_map
+// for an accessor only contains gaps. e.g. This happens for a join when
+// there are no attributes selected from one side.
+void removeGapOnlyAccessors(
+ const std::vector<std::pair<ValueAccessor *, std::vector<attribute_id>>>* accessor_attribute_map,
+ std::vector<std::pair<ValueAccessor *, const std::vector<attribute_id>>>* reduced_accessor_attribute_map) {
+ for (std::size_t i = 0; i < accessor_attribute_map->size(); ++i) {
+ bool all_gaps = true;
+ for (const auto &attr : (*accessor_attribute_map)[i].second)
+ if (attr != kInvalidCatalogId) {
+ all_gaps = false;
+ break;
+ }
+ if (all_gaps)
+ continue;
+ reduced_accessor_attribute_map->push_back((*accessor_attribute_map)[i]);
+ (*accessor_attribute_map)[i].first->beginIterationVirtual();
+ }
+}
+
+void InsertDestination::bulkInsertTuplesFromValueAccessors(
+ const std::vector<std::pair<ValueAccessor *, std::vector<attribute_id>>> &accessor_attribute_map,
+ bool always_mark_full) {
+ // Handle pathological corner case where there are no accessors
+ if (accessor_attribute_map.size() == 0)
+ return;
+
+ std::vector<std::pair<ValueAccessor *, const std::vector<attribute_id>>> reduced_accessor_attribute_map;
+ removeGapOnlyAccessors(&accessor_attribute_map, &reduced_accessor_attribute_map);
+
+ // We assume that all input accessors have the same number of tuples, so
+ // the iterations finish together. Therefore, we can just check the first one.
+ auto first_accessor = reduced_accessor_attribute_map[0].first;
+ while (!first_accessor->iterationFinishedVirtual()) {
+ tuple_id num_tuples_to_insert = kCatalogMaxID;
+ tuple_id num_tuples_inserted = 0;
+ MutableBlockReference output_block = this->getBlockForInsertion();
+
+ // Now iterate through all the accessors and do one round of bulk-insertion
+ // of partial tuples into the selected output_block.
+ // While inserting from the first ValueAccessor, space is reserved for
+ // all the columns including those coming from other ValueAccessors.
+ // Thereafter, in a given round, we only insert the remaining columns of the
+ // same tuples from the other ValueAccessors.
+ for (auto &p : reduced_accessor_attribute_map) {
+ ValueAccessor *accessor = p.first;
+ std::vector<attribute_id> attribute_map = p.second;
+
+
+ InvokeOnAnyValueAccessor(
+ accessor,
+ [&](auto *accessor) -> void { // NOLINT(build/c++11)
+ num_tuples_inserted = output_block->bulkInsertPartialTuples(
+ attribute_map, accessor, num_tuples_to_insert);
+ });
+
+ if (accessor == first_accessor) {
+ // Now we know how many full tuples can be inserted into this
+ // output_block (viz. number of tuples inserted from first ValueAccessor).
+ // We should only insert that many tuples from the remaining
+ // ValueAccessors as well.
+ num_tuples_to_insert = num_tuples_inserted;
+ } else {
+ // Since the bulk insertion of the first ValueAccessor should already
+ // have reserved the space for all the other ValueAccessors' columns,
+ // we must have been able to insert all the tuples we asked to insert.
+ DCHECK(num_tuples_inserted == num_tuples_to_insert);
+ }
+ }
+
+ // After one round of insertions, we have successfully inserted as many
+ // tuples as possible into the output_block. Strictly speaking, it's
+ // possible that there is more space for insertions because the size
+ // estimation of variable length columns is conservative. But we will ignore
+ // that case and proceed assuming that this output_block is full.
+
+ // Update the header for output_block and then return it.
+ output_block->bulkInsertPartialTuplesFinalize(num_tuples_inserted);
+ const bool mark_full = always_mark_full
+ || !first_accessor->iterationFinishedVirtual();
+ this->returnBlock(std::move(output_block), mark_full);
+ }
+}
+
void InsertDestination::insertTuplesFromVector(std::vector<Tuple>::const_iterator begin,
std::vector<Tuple>::const_iterator end) {
if (begin == end) {
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/2d11ec58/storage/InsertDestination.hpp
----------------------------------------------------------------------
diff --git a/storage/InsertDestination.hpp b/storage/InsertDestination.hpp
index 408e76b..3487638 100644
--- a/storage/InsertDestination.hpp
+++ b/storage/InsertDestination.hpp
@@ -152,6 +152,10 @@ class InsertDestination : public InsertDestinationInterface {
ValueAccessor *accessor,
bool always_mark_full = false) override;
+ void bulkInsertTuplesFromValueAccessors(
+ const std::vector<std::pair<ValueAccessor *, std::vector<attribute_id>>> &accessor_attribute_map,
+ bool always_mark_full = false) override;
+
void insertTuplesFromVector(std::vector<Tuple>::const_iterator begin,
std::vector<Tuple>::const_iterator end) override;
@@ -313,6 +317,12 @@ class AlwaysCreateBlockInsertDestination : public InsertDestination {
~AlwaysCreateBlockInsertDestination() override {
}
+ void bulkInsertTuplesFromValueAccessors(
+ const std::vector<std::pair<ValueAccessor *, std::vector<attribute_id>>> &accessor_attribute_map,
+ bool always_mark_full = false) override {
+ LOG(FATAL) << "bulkInsertTuplesFromValueAccessors is not implemented for AlwaysCreateBlockInsertDestination";
+ }
+
protected:
MutableBlockReference getBlockForInsertion() override;
@@ -517,6 +527,12 @@ class PartitionAwareInsertDestination : public InsertDestination {
ValueAccessor *accessor,
bool always_mark_full = false) override;
+ void bulkInsertTuplesFromValueAccessors(
+ const std::vector<std::pair<ValueAccessor *, std::vector<attribute_id>>> &accessor_attribute_map,
+ bool always_mark_full = false) override {
+ LOG(FATAL) << "bulkInsertTuplesFromValueAccessors is not implemented for PartitionAwareInsertDestination";
+ }
+
void insertTuplesFromVector(std::vector<Tuple>::const_iterator begin,
std::vector<Tuple>::const_iterator end) override;
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/2d11ec58/storage/InsertDestinationInterface.hpp
----------------------------------------------------------------------
diff --git a/storage/InsertDestinationInterface.hpp b/storage/InsertDestinationInterface.hpp
index 423dff1..b62d3e5 100644
--- a/storage/InsertDestinationInterface.hpp
+++ b/storage/InsertDestinationInterface.hpp
@@ -20,6 +20,7 @@
#ifndef QUICKSTEP_STORAGE_INSERT_DESTINATION_INTERFACE_HPP_
#define QUICKSTEP_STORAGE_INSERT_DESTINATION_INTERFACE_HPP_
+#include <utility>
#include <vector>
#include "catalog/CatalogTypedefs.hpp"
@@ -122,6 +123,27 @@ class InsertDestinationInterface {
bool always_mark_full = false) = 0;
/**
+ * @brief Bulk-insert tuples from one or more ValueAccessors
+ * into blocks managed by this InsertDestination.
+ *
+ * @warning It is implicitly assumed that all the input ValueAccessors have
+ * the same number of tuples in them.
+ *
+ * @param accessor_attribute_map A vector of pairs of ValueAccessor and
+ * corresponding attribute map
+ * The i-th attribute ID in the attr map for a value accessor is "n"
+ * if the attribute_id "i" in the output relation
+ * is the attribute_id "n" in corresponding input value accessor.
+ * Set the i-th element to kInvalidCatalogId if it doesn't come from
+ * the corresponding value accessor.
+ * @param always_mark_full If \c true, always mark the blocks full after
+ * insertion from ValueAccessor even when partially full.
+ **/
+ virtual void bulkInsertTuplesFromValueAccessors(
+ const std::vector<std::pair<ValueAccessor *, std::vector<attribute_id>>> &accessor_attribute_map,
+ bool always_mark_full = false) = 0;
+
+ /**
* @brief Insert tuples from a range of Tuples in a vector.
* @warning Unlike bulkInsertTuples(), this is not well-optimized and not
* intended for general use. It should only be used by
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/2d11ec58/storage/SplitRowStoreTupleStorageSubBlock.hpp
----------------------------------------------------------------------
diff --git a/storage/SplitRowStoreTupleStorageSubBlock.hpp b/storage/SplitRowStoreTupleStorageSubBlock.hpp
index 681001e..89c756d 100644
--- a/storage/SplitRowStoreTupleStorageSubBlock.hpp
+++ b/storage/SplitRowStoreTupleStorageSubBlock.hpp
@@ -304,9 +304,9 @@ class SplitRowStoreTupleStorageSubBlock: public TupleStorageSubBlock {
tuple_id bulkInsertPartialTuples(
const std::vector<attribute_id> &attribute_map,
ValueAccessor *accessor,
- const tuple_id max_num_tuples_to_insert);
+ const tuple_id max_num_tuples_to_insert) override;
- void bulkInsertPartialTuplesFinalize(const tuple_id num_tuples_inserted);
+ void bulkInsertPartialTuplesFinalize(const tuple_id num_tuples_inserted) override;
const void* getAttributeValue(const tuple_id tuple,
const attribute_id attr) const override;
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/2d11ec58/storage/StorageBlock.cpp
----------------------------------------------------------------------
diff --git a/storage/StorageBlock.cpp b/storage/StorageBlock.cpp
index ea74ee6..6267d6b 100644
--- a/storage/StorageBlock.cpp
+++ b/storage/StorageBlock.cpp
@@ -284,6 +284,30 @@ tuple_id StorageBlock::bulkInsertTuplesWithRemappedAttributes(
return num_inserted;
}
+tuple_id StorageBlock::bulkInsertPartialTuples(
+ const std::vector<attribute_id> &attribute_map,
+ ValueAccessor *accessor,
+ const tuple_id max_num_tuples_to_insert) {
+ const tuple_id num_inserted
+ = tuple_store_->bulkInsertPartialTuples(attribute_map,
+ accessor,
+ max_num_tuples_to_insert);
+ if (num_inserted != 0) {
+ invalidateAllIndexes();
+ dirty_ = true;
+ } else if (tuple_store_->isEmpty()) {
+ if (!accessor->iterationFinishedVirtual()) {
+ throw TupleTooLargeForBlock(0);
+ }
+ }
+ return num_inserted;
+}
+
+void StorageBlock::bulkInsertPartialTuplesFinalize(
+ const tuple_id num_tuples_inserted) {
+ tuple_store_->bulkInsertPartialTuplesFinalize(num_tuples_inserted);
+}
+
void StorageBlock::sample(const bool is_block_sample,
const int percentage,
InsertDestinationInterface *destination) const {
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/2d11ec58/storage/StorageBlock.hpp
----------------------------------------------------------------------
diff --git a/storage/StorageBlock.hpp b/storage/StorageBlock.hpp
index 56b3bdc..ed252c5 100644
--- a/storage/StorageBlock.hpp
+++ b/storage/StorageBlock.hpp
@@ -307,6 +307,7 @@ class StorageBlock : public StorageBlockBase {
* iteration will be advanced to the first non-inserted tuple or, if
* all accessible tuples were inserted in this block, to the end
* position.
+ * @param max_tuples_to_insert Insert at most these many tuples
* @return The number of tuples inserted from accessor.
**/
tuple_id bulkInsertTuplesWithRemappedAttributes(
@@ -314,6 +315,49 @@ class StorageBlock : public StorageBlockBase {
ValueAccessor *accessor);
/**
+ * @brief Insert up to max_num_tuples_to_insert tuples from a ValueAccessor
+ * as a single batch, using the attribute_map to project and reorder
+ * columns from the input ValueAccessor. Does not update header.
+ *
+ * @note Typical usage is where you want to bulk-insert columns from two
+ * or more value accessors. Instead of writing out the columns into
+ * one or more column vector value accessors, you can simply use this
+ * function with the appropriate attribute_map for each value
+ * accessor (InsertDestination::bulkInsertTuplesFromValueAccessors
+ * handles all the details) to insert tuples without an extra temp copy.
+ *
+ * @warning Must call bulkInsertPartialTuplesFinalize() to update the header,
+ * until which point, the insertion is not visible to others.
+ * @warning The inserted tuples may be placed in sub-optimal locations in this
+ * TupleStorageSubBlock.
+ *
+ * @param attribute_map A vector which maps the attributes of this
+ * TupleStorageSubBlock's relation (gaps indicated with kInvalidCatalogId)
+ * to the corresponding attributes which should be read from accessor.
+ * @param accessor A ValueAccessor to insert tuples from. The accessor's
+ * iteration will be advanced to the first non-inserted tuple or, if
+ * all accessible tuples were inserted in this sub-block, to the end
+ * position.
+ * @return The number of tuples inserted from accessor.
+ **/
+ tuple_id bulkInsertPartialTuples(
+ const std::vector<attribute_id> &attribute_map,
+ ValueAccessor *accessor,
+ const tuple_id max_num_tuples_to_insert);
+
+ /**
+ * @brief Update header after a bulkInsertPartialTuples.
+ *
+ * @warning Only call this after a bulkInsertPartialTuples, passing in the
+ * number of tuples that were inserted (return value of that function).
+ *
+ * @param num_tuples_inserted Number of tuples inserted (i.e., how much to
+ * advance the header.num_tuples by). Should be equal to the return
+ * value of bulkInsertPartialTuples.
+ **/
+ void bulkInsertPartialTuplesFinalize(tuple_id num_tuples_inserted);
+
+ /**
* @brief Get the IDs of tuples in this StorageBlock which match a given Predicate.
*
* @param predicate The predicate to match.
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/2d11ec58/storage/TupleStorageSubBlock.hpp
----------------------------------------------------------------------
diff --git a/storage/TupleStorageSubBlock.hpp b/storage/TupleStorageSubBlock.hpp
index aed6eea..26e8027 100644
--- a/storage/TupleStorageSubBlock.hpp
+++ b/storage/TupleStorageSubBlock.hpp
@@ -272,6 +272,56 @@ class TupleStorageSubBlock {
ValueAccessor *accessor) = 0;
/**
+ * @brief Insert up to max_num_tuples_to_insert tuples from a ValueAccessor
+ * as a single batch, using the attribute_map to project and reorder
+ * columns from the input ValueAccessor. Does not update header.
+ *
+ * @note Typical usage is where you want to bulk-insert columns from two
+ * or more value accessors. Instead of writing out the columns into
+ * one or more column vector value accessors, you can simply use this
+ * function with the appropriate attribute_map for each value
+ * accessor (InsertDestination::bulkInsertTuplesFromValueAccessors
+ * handles all the details) to insert tuples without an extra temp copy.
+ *
+ * @warning Must call bulkInsertPartialTuplesFinalize() to update the header,
+ * until which point, the insertion is not visible to others.
+ * @warning The inserted tuples may be placed in a suboptimal position in the
+ * block.
+ *
+ * @param attribute_map A vector which maps the attributes of this
+ * TupleStorageSubBlock's relation (gaps indicated with kInvalidCatalogId)
+ * to the corresponding attributes which should be read from accessor.
+ * @param accessor A ValueAccessor to insert tuples from. The accessor's
+ * iteration will be advanced to the first non-inserted tuple or, if
+ * all accessible tuples were inserted in this sub-block, to the end
+ * position.
+ * @return The number of tuples inserted from accessor.
+ **/
+ virtual tuple_id bulkInsertPartialTuples(
+ const std::vector<attribute_id> &attribute_map,
+ ValueAccessor *accessor,
+ const tuple_id max_num_tuples_to_insert) {
+ LOG(FATAL) << "Partial bulk insert is not supported for this TupleStorageBlock type ("
+ << getTupleStorageSubBlockType() << ").";
+ }
+
+ /**
+ * @brief Update header after a bulkInsertPartialTuples.
+ *
+ * @warning Only call this after a bulkInsertPartialTuples, passing in the
+ * number of tuples that were inserted (return value of that function).
+ *
+ * @param num_tuples_inserted Number of tuples inserted (i.e., how much to
+ * advance the header.num_tuples by). Should be equal to the return
+ * value of bulkInsertPartialTuples.
+ **/
+ virtual void bulkInsertPartialTuplesFinalize(
+ const tuple_id num_tuples_inserted) {
+ LOG(FATAL) << "Partial bulk insert is not supported for this TupleStorageBlock type ("
+ << getTupleStorageSubBlockType() << ").";
+ }
+
+ /**
* @brief Get the (untyped) value of an attribute in a tuple in this buffer.
* @warning This method may not be supported for all implementations of
* TupleStorageSubBlock. supportsUntypedGetAttributeValue() MUST be
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/2d11ec58/types/containers/ColumnVectorsValueAccessor.hpp
----------------------------------------------------------------------
diff --git a/types/containers/ColumnVectorsValueAccessor.hpp b/types/containers/ColumnVectorsValueAccessor.hpp
index fe413a0..fbbdc1b 100644
--- a/types/containers/ColumnVectorsValueAccessor.hpp
+++ b/types/containers/ColumnVectorsValueAccessor.hpp
@@ -139,6 +139,10 @@ class ColumnVectorsValueAccessor : public ValueAccessor {
return nullptr;
}
+ inline std::size_t getNumColumns() const {
+ return columns_.size();
+ }
+
template <bool check_null = true>
inline const void* getUntypedValue(const attribute_id attr_id) const {
return getUntypedValueAtAbsolutePosition<check_null>(attr_id, current_position_);
[07/12] incubator-quickstep git commit: Moved the init-db flag and
fixed a bug regarding db-init.
Posted by zu...@apache.org.
Moved the init-db flag and fixed a bug regarding db-init.
Project: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/commit/3047d89a
Tree: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/tree/3047d89a
Diff: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/diff/3047d89a
Branch: refs/heads/multi-shiftboss-test
Commit: 3047d89ab0e02bf0a30e914f55e6b49435c07e6a
Parents: 1b27888
Author: Zuyu Zhang <zu...@apache.org>
Authored: Mon Nov 21 23:44:13 2016 -0800
Committer: Zuyu Zhang <zu...@apache.org>
Committed: Mon Nov 21 23:44:13 2016 -0800
----------------------------------------------------------------------
cli/DefaultsConfigurator.cpp | 2 +-
cli/Flags.cpp | 2 ++
cli/Flags.hpp | 3 +++
cli/QuickstepCli.cpp | 1 -
4 files changed, 6 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/3047d89a/cli/DefaultsConfigurator.cpp
----------------------------------------------------------------------
diff --git a/cli/DefaultsConfigurator.cpp b/cli/DefaultsConfigurator.cpp
index 94280a7..21b0af6 100644
--- a/cli/DefaultsConfigurator.cpp
+++ b/cli/DefaultsConfigurator.cpp
@@ -55,7 +55,7 @@ void DefaultsConfigurator::InitializeDefaultDatabase(const string &storage_path,
#else
{
const string path_name = "mkdir " + storage_path;
- CHECK(std::system(path_name.c_str()))
+ CHECK(!std::system(path_name.c_str()))
<< "Failed when attempting to create the directory: " << storage_path;
}
#endif // QUICKSTEP_OS_WINDOWS
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/3047d89a/cli/Flags.cpp
----------------------------------------------------------------------
diff --git a/cli/Flags.cpp b/cli/Flags.cpp
index 87f9f73..1e3ce1d 100644
--- a/cli/Flags.cpp
+++ b/cli/Flags.cpp
@@ -32,6 +32,8 @@ using std::fprintf;
namespace quickstep {
+DEFINE_bool(initialize_db, false, "If true, initialize a database.");
+
static bool ValidateNumWorkers(const char *flagname, int value) {
if (value > 0) {
return true;
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/3047d89a/cli/Flags.hpp
----------------------------------------------------------------------
diff --git a/cli/Flags.hpp b/cli/Flags.hpp
index b020a3e..70aee98 100644
--- a/cli/Flags.hpp
+++ b/cli/Flags.hpp
@@ -32,6 +32,9 @@ namespace quickstep {
* @brief A collection of common flags shared by Quickstep CLIs in both the
* single-node and the distributed version.
**/
+
+DECLARE_bool(initialize_db);
+
DECLARE_int32(num_workers);
DECLARE_string(storage_path);
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/3047d89a/cli/QuickstepCli.cpp
----------------------------------------------------------------------
diff --git a/cli/QuickstepCli.cpp b/cli/QuickstepCli.cpp
index 656786a..f4816a8 100644
--- a/cli/QuickstepCli.cpp
+++ b/cli/QuickstepCli.cpp
@@ -128,7 +128,6 @@ DEFINE_string(worker_affinities, "",
"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).");
-DEFINE_bool(initialize_db, false, "If true, initialize a database.");
DEFINE_bool(print_query, false,
"Print each input query statement. This is useful when running a "
"large number of queries in a batch.");
[05/12] incubator-quickstep git commit: Set Default block type to
only be the splitrow store.
Posted by zu...@apache.org.
Set Default block type to only be the splitrow store.
Project: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/commit/1b27888c
Tree: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/tree/1b27888c
Diff: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/diff/1b27888c
Branch: refs/heads/multi-shiftboss-test
Commit: 1b27888c3ced7ff6b104893cc6fe1a02d5cb33b2
Parents: 2d11ec5
Author: cramja <ma...@gmail.com>
Authored: Thu Nov 17 16:12:55 2016 -0600
Committer: cramja <ma...@gmail.com>
Committed: Mon Nov 21 14:28:02 2016 -0600
----------------------------------------------------------------------
storage/StorageBlockLayout.cpp | 10 +++++-----
1 file changed, 5 insertions(+), 5 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/1b27888c/storage/StorageBlockLayout.cpp
----------------------------------------------------------------------
diff --git a/storage/StorageBlockLayout.cpp b/storage/StorageBlockLayout.cpp
index bdeb9cb..3b1bcc0 100644
--- a/storage/StorageBlockLayout.cpp
+++ b/storage/StorageBlockLayout.cpp
@@ -148,17 +148,17 @@ void StorageBlockLayout::copyHeaderTo(void *dest) const {
StorageBlockLayout* StorageBlockLayout::GenerateDefaultLayout(const CatalogRelationSchema &relation,
const bool relation_variable_length) {
+ // TODO(marc): In the future we may use relation_variable_length (columnstores as intermediate
+ // blocks), but for now, all we do is add (void) so that the compiler will not complain
+ // about an unused variable.
+ (void) relation_variable_length;
StorageBlockLayout *layout = new StorageBlockLayout(relation);
StorageBlockLayoutDescription *description = layout->getDescriptionMutable();
description->set_num_slots(1);
TupleStorageSubBlockDescription *tuple_store_description = description->mutable_tuple_store_description();
- if (relation_variable_length) {
- tuple_store_description->set_sub_block_type(TupleStorageSubBlockDescription::SPLIT_ROW_STORE);
- } else {
- tuple_store_description->set_sub_block_type(TupleStorageSubBlockDescription::PACKED_ROW_STORE);
- }
+ tuple_store_description->set_sub_block_type(TupleStorageSubBlockDescription::SPLIT_ROW_STORE);
layout->finalize();
return layout;
[04/12] incubator-quickstep git commit: IWYU for QuickstepCli.
Posted by zu...@apache.org.
IWYU for QuickstepCli.
Project: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/commit/172b51b5
Tree: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/tree/172b51b5
Diff: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/diff/172b51b5
Branch: refs/heads/multi-shiftboss-test
Commit: 172b51b5eaa9a2354792dc0a8a51fbad13ba10f2
Parents: f673f94
Author: Zuyu Zhang <zu...@apache.org>
Authored: Mon Nov 21 00:06:05 2016 -0800
Committer: Zuyu Zhang <zu...@apache.org>
Committed: Mon Nov 21 00:06:05 2016 -0800
----------------------------------------------------------------------
CMakeLists.txt | 3 ---
cli/DefaultsConfigurator.cpp | 1 +
cli/QuickstepCli.cpp | 21 +--------------------
3 files changed, 2 insertions(+), 23 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/172b51b5/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/CMakeLists.txt b/CMakeLists.txt
index cd53967..126b47b 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -758,16 +758,13 @@ target_link_libraries(quickstep_cli_shell
quickstep_cli_PrintToScreen
quickstep_parser_ParseStatement
quickstep_parser_SqlParserWrapper
- quickstep_queryexecution_AdmitRequestMessage
quickstep_queryexecution_ForemanSingleNode
quickstep_queryexecution_QueryContext
quickstep_queryexecution_QueryExecutionTypedefs
quickstep_queryexecution_QueryExecutionUtil
quickstep_queryexecution_Worker
quickstep_queryexecution_WorkerDirectory
- quickstep_queryexecution_WorkerMessage
quickstep_queryoptimizer_QueryHandle
- quickstep_queryoptimizer_QueryPlan
quickstep_queryoptimizer_QueryProcessor
quickstep_storage_PreloaderThread
quickstep_storage_StorageConstants
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/172b51b5/cli/DefaultsConfigurator.cpp
----------------------------------------------------------------------
diff --git a/cli/DefaultsConfigurator.cpp b/cli/DefaultsConfigurator.cpp
index 355031c..94280a7 100644
--- a/cli/DefaultsConfigurator.cpp
+++ b/cli/DefaultsConfigurator.cpp
@@ -22,6 +22,7 @@
#include "cli/CliConfig.h" // For QUICKSTEP_OS_WINDOWS.
#ifdef QUICKSTEP_OS_WINDOWS
+// TODO(jmp): If filesystem shows up in C++-17, we can switch to just using that.
#include <experimental/filesystem>
#else
#include <cstdlib>
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/172b51b5/cli/QuickstepCli.cpp
----------------------------------------------------------------------
diff --git a/cli/QuickstepCli.cpp b/cli/QuickstepCli.cpp
index 465793c..656786a 100644
--- a/cli/QuickstepCli.cpp
+++ b/cli/QuickstepCli.cpp
@@ -27,16 +27,8 @@
#include <string>
#include <utility>
#include <vector>
-#include <fstream>
-#include "cli/CliConfig.h" // For QUICKSTEP_USE_LINENOISE, QUICKSTEP_ENABLE_GOOGLE_PROFILER, and QUICKSTEP_OS_WINDOWS.
-
-// TODO(jmp): If filesystem shows up in C++-17, we can switch to just using that.
-#ifdef QUICKSTEP_OS_WINDOWS
-#include <filesystem>
-#else
-#include <stdlib.h>
-#endif
+#include "cli/CliConfig.h" // For QUICKSTEP_USE_LINENOISE, QUICKSTEP_ENABLE_GOOGLE_PROFILER.
#include "cli/CommandExecutor.hpp"
#include "cli/DropRelation.hpp"
@@ -59,15 +51,12 @@ typedef quickstep::LineReaderDumb LineReaderImpl;
#include "cli/PrintToScreen.hpp"
#include "parser/ParseStatement.hpp"
#include "parser/SqlParserWrapper.hpp"
-#include "query_execution/AdmitRequestMessage.hpp"
#include "query_execution/ForemanSingleNode.hpp"
#include "query_execution/QueryExecutionTypedefs.hpp"
#include "query_execution/QueryExecutionUtil.hpp"
#include "query_execution/Worker.hpp"
#include "query_execution/WorkerDirectory.hpp"
-#include "query_execution/WorkerMessage.hpp"
#include "query_optimizer/QueryHandle.hpp"
-#include "query_optimizer/QueryPlan.hpp"
#include "query_optimizer/QueryProcessor.hpp"
#include "storage/StorageConfig.h" // For QUICKSTEP_HAVE_FILE_MANAGER_HDFS.
@@ -89,10 +78,8 @@ typedef quickstep::LineReaderDumb LineReaderImpl;
#include "glog/logging.h"
-#include "tmb/address.h"
#include "tmb/id_typedefs.h"
#include "tmb/message_bus.h"
-#include "tmb/message_style.h"
namespace quickstep {
class CatalogRelation;
@@ -104,8 +91,6 @@ using std::printf;
using std::string;
using std::vector;
-using quickstep::Address;
-using quickstep::AdmitRequestMessage;
using quickstep::CatalogRelation;
using quickstep::DefaultsConfigurator;
using quickstep::DropRelation;
@@ -114,20 +99,16 @@ using quickstep::FLAGS_storage_path;
using quickstep::ForemanSingleNode;
using quickstep::InputParserUtil;
using quickstep::MessageBusImpl;
-using quickstep::MessageStyle;
-using quickstep::ParseCommand;
using quickstep::ParseResult;
using quickstep::ParseStatement;
using quickstep::PrintToScreen;
using quickstep::PtrVector;
using quickstep::QueryExecutionUtil;
using quickstep::QueryHandle;
-using quickstep::QueryPlan;
using quickstep::QueryProcessor;
using quickstep::SqlParserWrapper;
using quickstep::Worker;
using quickstep::WorkerDirectory;
-using quickstep::WorkerMessage;
using quickstep::kAdmitRequestMessage;
using quickstep::kCatalogFilename;
using quickstep::kPoisonMessage;
[11/12] incubator-quickstep git commit: Scheduled FinalizeAggr /
DestroyAggr WorkOrder on the same Shiftboss of AggrWorkOrder.
Posted by zu...@apache.org.
Scheduled FinalizeAggr / DestroyAggr WorkOrder on the same Shiftboss of AggrWorkOrder.
Project: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/commit/a98e4a14
Tree: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/tree/a98e4a14
Diff: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/diff/a98e4a14
Branch: refs/heads/multi-shiftboss-test
Commit: a98e4a147781ea883daeb8c6223b60ce0f839a26
Parents: 4d935f4
Author: Zuyu Zhang <zu...@apache.org>
Authored: Thu Nov 17 12:01:26 2016 -0800
Committer: Zuyu Zhang <zu...@apache.org>
Committed: Sun Nov 27 10:50:04 2016 -0800
----------------------------------------------------------------------
query_execution/ForemanDistributed.cpp | 71 +++++++++++++++++-----
query_execution/ForemanDistributed.hpp | 9 ++-
query_execution/PolicyEnforcerDistributed.cpp | 12 ++++
query_execution/PolicyEnforcerDistributed.hpp | 18 ++++++
query_execution/QueryManagerDistributed.hpp | 23 +++++++
5 files changed, 116 insertions(+), 17 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/a98e4a14/query_execution/ForemanDistributed.cpp
----------------------------------------------------------------------
diff --git a/query_execution/ForemanDistributed.cpp b/query_execution/ForemanDistributed.cpp
index 900a71f..7ca246d 100644
--- a/query_execution/ForemanDistributed.cpp
+++ b/query_execution/ForemanDistributed.cpp
@@ -243,28 +243,55 @@ bool ForemanDistributed::canCollectNewMessages(const tmb::message_type_id messag
kWorkOrderFeedbackMessage);
}
-bool ForemanDistributed::isHashJoinRelatedWorkOrder(const unique_ptr<S::WorkOrderMessage> &message,
+bool ForemanDistributed::isAggregationRelatedWorkOrder(const S::WorkOrderMessage &proto,
+ const size_t next_shiftboss_index_to_schedule,
+ size_t *shiftboss_index_for_aggregation) {
+ const S::WorkOrder &work_order_proto = proto.work_order();
+ QueryContext::aggregation_state_id aggr_state_index;
+
+ switch (work_order_proto.work_order_type()) {
+ case S::AGGREGATION:
+ aggr_state_index = work_order_proto.GetExtension(S::AggregationWorkOrder::aggr_state_index);
+ break;
+ case S::FINALIZE_AGGREGATION:
+ aggr_state_index = work_order_proto.GetExtension(S::FinalizeAggregationWorkOrder::aggr_state_index);
+ break;
+ case S::DESTROY_AGGREGATION_STATE:
+ aggr_state_index = work_order_proto.GetExtension(S::DestroyAggregationStateWorkOrder::aggr_state_index);
+ break;
+ default:
+ return false;
+ }
+
+ policy_enforcer_->getShiftbossIndexForAggregation(
+ proto.query_id(), aggr_state_index, next_shiftboss_index_to_schedule,
+ shiftboss_index_for_aggregation);
+
+ return true;
+}
+
+bool ForemanDistributed::isHashJoinRelatedWorkOrder(const S::WorkOrderMessage &proto,
const size_t next_shiftboss_index_to_schedule,
size_t *shiftboss_index_for_hash_join) {
- const S::WorkOrder &work_order_proto = message->work_order();
+ const S::WorkOrder &work_order_proto = proto.work_order();
QueryContext::join_hash_table_id join_hash_table_index;
switch (work_order_proto.work_order_type()) {
case S::BUILD_HASH:
- join_hash_table_index = work_order_proto.GetExtension(serialization::BuildHashWorkOrder::join_hash_table_index);
- break;
- case S::DESTROY_HASH:
- join_hash_table_index = work_order_proto.GetExtension(serialization::DestroyHashWorkOrder::join_hash_table_index);
+ join_hash_table_index = work_order_proto.GetExtension(S::BuildHashWorkOrder::join_hash_table_index);
break;
case S::HASH_JOIN:
- join_hash_table_index = work_order_proto.GetExtension(serialization::HashJoinWorkOrder::join_hash_table_index);
+ join_hash_table_index = work_order_proto.GetExtension(S::HashJoinWorkOrder::join_hash_table_index);
+ break;
+ case S::DESTROY_HASH:
+ join_hash_table_index = work_order_proto.GetExtension(S::DestroyHashWorkOrder::join_hash_table_index);
break;
default:
return false;
}
policy_enforcer_->getShiftbossIndexForHashJoin(
- message->query_id(), join_hash_table_index, next_shiftboss_index_to_schedule,
+ proto.query_id(), join_hash_table_index, next_shiftboss_index_to_schedule,
shiftboss_index_for_hash_join);
return true;
@@ -275,14 +302,11 @@ void ForemanDistributed::dispatchWorkOrderMessages(const vector<unique_ptr<S::Wo
size_t shiftboss_index = 0u;
for (const auto &message : messages) {
DCHECK(message != nullptr);
- size_t shiftboss_index_for_hash_join;
- if (isHashJoinRelatedWorkOrder(message, shiftboss_index, &shiftboss_index_for_hash_join)) {
- sendWorkOrderMessage(shiftboss_index_for_hash_join, *message);
- shiftboss_directory_.incrementNumQueuedWorkOrders(shiftboss_index_for_hash_join);
-
- if (shiftboss_index == shiftboss_index_for_hash_join) {
- shiftboss_index = (shiftboss_index + 1) % num_shiftbosses;
- }
+ size_t shiftboss_index_for_particular_work_order_type;
+ if (isAggregationRelatedWorkOrder(*message, shiftboss_index, &shiftboss_index_for_particular_work_order_type)) {
+ dispatchWorkOrderMessagesHelper(*message, shiftboss_index_for_particular_work_order_type, &shiftboss_index);
+ } else if (isHashJoinRelatedWorkOrder(*message, shiftboss_index, &shiftboss_index_for_particular_work_order_type)) {
+ dispatchWorkOrderMessagesHelper(*message, shiftboss_index_for_particular_work_order_type, &shiftboss_index);
} else {
sendWorkOrderMessage(shiftboss_index, *message);
shiftboss_directory_.incrementNumQueuedWorkOrders(shiftboss_index);
@@ -293,6 +317,21 @@ void ForemanDistributed::dispatchWorkOrderMessages(const vector<unique_ptr<S::Wo
}
}
+void ForemanDistributed::dispatchWorkOrderMessagesHelper(const S::WorkOrderMessage &proto,
+ const size_t shiftboss_index_for_particular_work_order_type,
+ size_t *shiftboss_index) {
+ sendWorkOrderMessage(shiftboss_index_for_particular_work_order_type, proto);
+ shiftboss_directory_.incrementNumQueuedWorkOrders(shiftboss_index_for_particular_work_order_type);
+
+ if (*shiftboss_index == shiftboss_index_for_particular_work_order_type) {
+ *shiftboss_index = (*shiftboss_index + 1) % shiftboss_directory_.size();
+ } else {
+ // NOTE(zuyu): This is not the exact round-robin scheduling, as in this case,
+ // <shiftboss_index_for_particular_work_order_type> will be scheduled one
+ // more WorkOrder.
+ }
+}
+
void ForemanDistributed::sendWorkOrderMessage(const size_t shiftboss_index,
const S::WorkOrderMessage &proto) {
const size_t proto_length = proto.ByteSize();
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/a98e4a14/query_execution/ForemanDistributed.hpp
----------------------------------------------------------------------
diff --git a/query_execution/ForemanDistributed.hpp b/query_execution/ForemanDistributed.hpp
index 55e747f..f6d8597 100644
--- a/query_execution/ForemanDistributed.hpp
+++ b/query_execution/ForemanDistributed.hpp
@@ -71,7 +71,11 @@ class ForemanDistributed final : public ForemanBase {
void run() override;
private:
- bool isHashJoinRelatedWorkOrder(const std::unique_ptr<serialization::WorkOrderMessage> &message,
+ bool isAggregationRelatedWorkOrder(const serialization::WorkOrderMessage &proto,
+ const std::size_t next_shiftboss_index_to_schedule,
+ std::size_t *shiftboss_index_for_aggregation);
+
+ bool isHashJoinRelatedWorkOrder(const serialization::WorkOrderMessage &proto,
const std::size_t next_shiftboss_index_to_schedule,
std::size_t *shiftboss_index_for_hash_join);
@@ -84,6 +88,9 @@ class ForemanDistributed final : public ForemanBase {
void dispatchWorkOrderMessages(
const std::vector<std::unique_ptr<serialization::WorkOrderMessage>> &messages);
+ void dispatchWorkOrderMessagesHelper(const serialization::WorkOrderMessage &proto,
+ const std::size_t shiftboss_index_for_particular_work_order_type,
+ std::size_t *shiftboss_index);
/**
* @brief Send the given message to the specified worker.
*
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/a98e4a14/query_execution/PolicyEnforcerDistributed.cpp
----------------------------------------------------------------------
diff --git a/query_execution/PolicyEnforcerDistributed.cpp b/query_execution/PolicyEnforcerDistributed.cpp
index 86b36c8..c5642bc 100644
--- a/query_execution/PolicyEnforcerDistributed.cpp
+++ b/query_execution/PolicyEnforcerDistributed.cpp
@@ -158,6 +158,18 @@ void PolicyEnforcerDistributed::processInitiateRebuildResponseMessage(const tmb:
}
}
+void PolicyEnforcerDistributed::getShiftbossIndexForAggregation(
+ const std::size_t query_id,
+ const QueryContext::aggregation_state_id aggr_state_index,
+ const std::size_t next_shiftboss_index_to_schedule,
+ std::size_t *shiftboss_index) {
+ DCHECK(admitted_queries_.find(query_id) != admitted_queries_.end());
+ QueryManagerDistributed *query_manager = static_cast<QueryManagerDistributed*>(admitted_queries_[query_id].get());
+ query_manager->getShiftbossIndexForAggregation(aggr_state_index,
+ next_shiftboss_index_to_schedule,
+ shiftboss_index);
+}
+
void PolicyEnforcerDistributed::getShiftbossIndexForHashJoin(
const std::size_t query_id,
const QueryContext::join_hash_table_id join_hash_table_index,
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/a98e4a14/query_execution/PolicyEnforcerDistributed.hpp
----------------------------------------------------------------------
diff --git a/query_execution/PolicyEnforcerDistributed.hpp b/query_execution/PolicyEnforcerDistributed.hpp
index 37326bd..e8bc394 100644
--- a/query_execution/PolicyEnforcerDistributed.hpp
+++ b/query_execution/PolicyEnforcerDistributed.hpp
@@ -90,6 +90,24 @@ class PolicyEnforcerDistributed final : public PolicyEnforcerBase {
void processInitiateRebuildResponseMessage(const tmb::TaggedMessage &tagged_message);
/**
+ * @brief Get or set the index of Shiftboss for an Aggregation related
+ * WorkOrder. If it is the first Aggregation on <aggr_state_index>,
+ * <shiftboss_index> will be set to <next_shiftboss_index_to_schedule>.
+ * Otherwise, <shiftboss_index> will be set to the index of the Shiftboss that
+ * has executed the first Aggregation.
+ *
+ * @param query_id The query id.
+ * @param aggr_state_index The Hash Table for the Aggregation.
+ * @param next_shiftboss_index The index of Shiftboss to schedule a next WorkOrder.
+ * @param shiftboss_index The index of Shiftboss to schedule the WorkOrder.
+ **/
+ void getShiftbossIndexForAggregation(
+ const std::size_t query_id,
+ const QueryContext::aggregation_state_id aggr_state_index,
+ const std::size_t next_shiftboss_index_to_schedule,
+ std::size_t *shiftboss_index);
+
+ /**
* @brief Get or set the index of Shiftboss for a HashJoin related WorkOrder.
* If it is the first BuildHash on <join_hash_table_index>, <shiftboss_index>
* will be set to <next_shiftboss_index_to_schedule>. Otherwise,
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/a98e4a14/query_execution/QueryManagerDistributed.hpp
----------------------------------------------------------------------
diff --git a/query_execution/QueryManagerDistributed.hpp b/query_execution/QueryManagerDistributed.hpp
index 2b21303..6455bf7 100644
--- a/query_execution/QueryManagerDistributed.hpp
+++ b/query_execution/QueryManagerDistributed.hpp
@@ -95,6 +95,26 @@ class QueryManagerDistributed final : public QueryManagerBase {
const dag_node_index start_operator_index);
/**
+ * @brief Get the index of Shiftboss for an Aggregation related WorkOrder. If
+ * the Shiftboss index is not found, set using <next_shiftboss_index_to_schedule>.
+ *
+ * @param aggr_state_index The Hash Table for the Aggregation.
+ * @param next_shiftboss_index The index of Shiftboss to schedule a next WorkOrder.
+ * @param shiftboss_index The index of Shiftboss to schedule the WorkOrder.
+ **/
+ void getShiftbossIndexForAggregation(const QueryContext::aggregation_state_id aggr_state_index,
+ const std::size_t next_shiftboss_index_to_schedule,
+ std::size_t *shiftboss_index) {
+ const auto cit = shiftboss_indexes_for_aggrs_.find(aggr_state_index);
+ if (cit != shiftboss_indexes_for_aggrs_.end()) {
+ *shiftboss_index = cit->second;
+ } else {
+ shiftboss_indexes_for_aggrs_.emplace(aggr_state_index, next_shiftboss_index_to_schedule);
+ *shiftboss_index = next_shiftboss_index_to_schedule;
+ }
+ }
+
+ /**
* @brief Get the index of Shiftboss for a HashJoin related WorkOrder. If the
* Shiftboss index is not found, set using <next_shiftboss_index_to_schedule>.
*
@@ -136,6 +156,9 @@ class QueryManagerDistributed final : public QueryManagerBase {
std::unique_ptr<WorkOrderProtosContainer> normal_workorder_protos_container_;
+ // A map from a aggregation to its scheduled Shiftboss index.
+ std::unordered_map<QueryContext::aggregation_state_id, std::size_t> shiftboss_indexes_for_aggrs_;
+
// A map from a join hash table to its scheduled Shiftboss index.
std::unordered_map<QueryContext::join_hash_table_id, std::size_t> shiftboss_indexes_for_hash_joins_;
[02/12] incubator-quickstep git commit: IWYU for DefaultsConfigurator.
Posted by zu...@apache.org.
IWYU for DefaultsConfigurator.
Project: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/commit/54e0654b
Tree: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/tree/54e0654b
Diff: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/diff/54e0654b
Branch: refs/heads/multi-shiftboss-test
Commit: 54e0654bfb9a42553129428fa94f78535e2781f4
Parents: e666a81
Author: Zuyu Zhang <zu...@apache.org>
Authored: Sun Nov 20 23:30:42 2016 -0800
Committer: Zuyu Zhang <zu...@apache.org>
Committed: Sun Nov 20 23:30:42 2016 -0800
----------------------------------------------------------------------
cli/DefaultsConfigurator.hpp | 28 +++++++++++++++++-----------
1 file changed, 17 insertions(+), 11 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/54e0654b/cli/DefaultsConfigurator.hpp
----------------------------------------------------------------------
diff --git a/cli/DefaultsConfigurator.hpp b/cli/DefaultsConfigurator.hpp
index 7450d88..4fd62ac 100644
--- a/cli/DefaultsConfigurator.hpp
+++ b/cli/DefaultsConfigurator.hpp
@@ -20,17 +20,23 @@
#ifndef QUICKSTEP_CLI_DEFAULTS_CONFIGURATOR_HPP_
#define QUICKSTEP_CLI_DEFAULTS_CONFIGURATOR_HPP_
+#include "storage/StorageConfig.h" // For QUICKSTEP_HAVE_LIBNUMA.
+
+#ifdef QUICKSTEP_HAVE_LIBNUMA
+#include <numa.h>
+#endif // QUICKSTEP_HAVE_LIBNUMA
+
+#include <cstddef>
#include <thread> // NOLINT(build/c++11)
+
+#ifdef QUICKSTEP_HAVE_LIBNUMA
#include <unordered_map>
+#endif // QUICKSTEP_HAVE_LIBNUMA
+
#include <vector>
-#include "storage/StorageConfig.h"
#include "utility/Macros.hpp"
-#ifdef QUICKSTEP_HAVE_LIBNUMA
-#include <numa.h>
-#endif
-
namespace quickstep {
/** \addtogroup CLI
@@ -60,12 +66,12 @@ class DefaultsConfigurator {
* have libnuma.
**/
static std::size_t GetNumNUMANodes() {
- #ifdef QUICKSTEP_HAVE_LIBNUMA
+#ifdef QUICKSTEP_HAVE_LIBNUMA
// Id of the maximum node.
return numa_max_node() + 1;
- #else
+#else
return 1;
- #endif
+#endif // QUICKSTEP_HAVE_LIBNUMA
}
/**
@@ -80,7 +86,7 @@ class DefaultsConfigurator {
**/
static std::size_t GetNumNUMANodesCoveredByWorkers(const std::vector<int> &worker_cpu_affinities) {
if (!worker_cpu_affinities.empty()) {
- #ifdef QUICKSTEP_HAVE_LIBNUMA
+#ifdef QUICKSTEP_HAVE_LIBNUMA
// Key = NUMA node, value = whether there is at least one worker whose
// affinity is set to a core on the given NUMA node.
std::unordered_map<int, bool> any_worker_on_numa_node;
@@ -93,7 +99,7 @@ class DefaultsConfigurator {
}
}
return any_worker_on_numa_node.size();
- #endif
+#endif // QUICKSTEP_HAVE_LIBNUMA
}
// When libnuma is not available, or worker affinities are not specified,
// the default return value is 1.
@@ -108,9 +114,9 @@ class DefaultsConfigurator {
DISALLOW_COPY_AND_ASSIGN(DefaultsConfigurator);
};
+
/** @} */
} // namespace quickstep
#endif // QUICKSTEP_CLI_DEFAULTS_CONFIGURATOR_HPP_
-
[12/12] incubator-quickstep git commit: A better debug log for worker.
Posted by zu...@apache.org.
A better debug log for worker.
Project: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/commit/d2862e57
Tree: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/tree/d2862e57
Diff: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/diff/d2862e57
Branch: refs/heads/multi-shiftboss-test
Commit: d2862e5783b5af3c096db3872a6820fcf08c23c5
Parents: a98e4a1
Author: Zuyu Zhang <zu...@apache.org>
Authored: Thu Nov 17 15:10:03 2016 -0800
Committer: Zuyu Zhang <zu...@apache.org>
Committed: Sun Nov 27 10:50:05 2016 -0800
----------------------------------------------------------------------
query_execution/Worker.cpp | 9 ++++++++-
1 file changed, 8 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/d2862e57/query_execution/Worker.cpp
----------------------------------------------------------------------
diff --git a/query_execution/Worker.cpp b/query_execution/Worker.cpp
index 31eb964..ec6c503 100644
--- a/query_execution/Worker.cpp
+++ b/query_execution/Worker.cpp
@@ -107,7 +107,14 @@ void Worker::sendWorkOrderCompleteMessage(const tmb::client_id receiver,
static_cast<const void *>(proto_bytes), proto_length, message_type);
std::free(proto_bytes);
- DLOG(INFO) << "Worker sent WorkOrderCompleteMessage (typed '" << message_type
+ DLOG(INFO) << "Worker"
+#ifdef QUICKSTEP_DISTRIBUTED
+ << " in Shiftboss " << shiftboss_index_
+#endif // QUICKSTEP_DISTRIBUTED
+ << " sent "
+ << (message_type == kWorkOrderCompleteMessage ? "WorkOrderCompleteMessage"
+ : "RebuildWorkOrderCompleteMessage")
+ << " (typed '" << message_type
<< "') to Scheduler with TMB client ID " << receiver;
const tmb::MessageBus::SendStatus send_status =
QueryExecutionUtil::SendTMBMessage(