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/07/17 00:23:40 UTC
[1/2] incubator-quickstep git commit: Added Comparisons for the Date
Type. [Forced Update!]
Repository: incubator-quickstep
Updated Branches:
refs/heads/refactor-process-msg 2e0e00191 -> db0e7e3cc (forced update)
Added Comparisons for the Date Type.
Project: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/commit/cf81b5e6
Tree: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/tree/cf81b5e6
Diff: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/diff/cf81b5e6
Branch: refs/heads/refactor-process-msg
Commit: cf81b5e62ddf7e96f3fcdb1bcc97ee0768a6997a
Parents: 9a398d3
Author: Zuyu Zhang <zu...@twitter.com>
Authored: Fri Jul 15 16:56:14 2016 -0700
Committer: Zuyu Zhang <zu...@twitter.com>
Committed: Sat Jul 16 15:51:21 2016 -0700
----------------------------------------------------------------------
.../operations/comparisons/BasicComparison.cpp | 6 ++-
.../operations/comparisons/BasicComparison.hpp | 29 ++++++++++--
types/operations/comparisons/CMakeLists.txt | 2 -
types/operations/comparisons/ComparisonUtil.hpp | 34 +++++++++++++++
.../comparisons/LiteralComparators.hpp | 24 +++++-----
.../comparisons/tests/Comparison_unittest.cpp | 46 +++++++++++++++++---
6 files changed, 115 insertions(+), 26 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/cf81b5e6/types/operations/comparisons/BasicComparison.cpp
----------------------------------------------------------------------
diff --git a/types/operations/comparisons/BasicComparison.cpp b/types/operations/comparisons/BasicComparison.cpp
index 21b92ca..ed0ce62 100644
--- a/types/operations/comparisons/BasicComparison.cpp
+++ b/types/operations/comparisons/BasicComparison.cpp
@@ -23,7 +23,6 @@
#include "types/Type.hpp"
#include "types/TypeID.hpp"
#include "utility/EqualsAnyConstant.hpp"
-#include "utility/Macros.hpp"
#include "glog/logging.h"
@@ -49,6 +48,9 @@ bool BasicComparison::canCompareTypes(const Type &left, const Type &right) const
return false;
}
}
+ case kDate: {
+ return right.getTypeID() == kDate;
+ }
case kDatetime: {
return right.getTypeID() == kDatetime;
}
@@ -89,7 +91,7 @@ bool BasicComparison::canComparePartialTypes(const Type *left_type,
const Type *known_type = (left_type != nullptr) ? left_type : right_type;
return QUICKSTEP_EQUALS_ANY_CONSTANT(known_type->getTypeID(),
kInt, kLong, kFloat, kDouble,
- kDatetime, kDatetimeInterval, kYearMonthInterval,
+ kDate, kDatetime, kDatetimeInterval, kYearMonthInterval,
kChar, kVarChar);
}
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/cf81b5e6/types/operations/comparisons/BasicComparison.hpp
----------------------------------------------------------------------
diff --git a/types/operations/comparisons/BasicComparison.hpp b/types/operations/comparisons/BasicComparison.hpp
index 9741c32..c95f774 100644
--- a/types/operations/comparisons/BasicComparison.hpp
+++ b/types/operations/comparisons/BasicComparison.hpp
@@ -22,8 +22,6 @@
#include <cstdint>
#include <string>
-#include "types/DatetimeLit.hpp"
-#include "types/IntervalLit.hpp"
#include "types/Type.hpp"
#include "types/TypeErrors.hpp"
#include "types/TypeFactory.hpp"
@@ -37,6 +35,11 @@
namespace quickstep {
+struct DateLit;
+struct DatetimeIntervalLit;
+struct DatetimeLit;
+struct YearMonthIntervalLit;
+
/** \addtogroup Types
* @{
*/
@@ -161,6 +164,18 @@ bool BasicComparison::compareTypedValuesCheckedHelper(const TypedValue &left,
}
break;
}
+ case kDate: {
+ if (right.getTypeID() == kDate) {
+ if (left.isNull() || right.isNull()) {
+ return false;
+ }
+
+ ComparisonFunctor<DateLit> comparison_functor;
+ return comparison_functor(left.getLiteral<DateLit>(), right.getLiteral<DateLit>());
+ }
+ LOG(FATAL) << "Comparison " << getName() << " can not be applied to types "
+ << left_type.getName() << " and " << right_type.getName();
+ }
case kDatetime: {
if (right.getTypeID() == kDatetime) {
if (left.isNull() || right.isNull()) {
@@ -269,7 +284,8 @@ UncheckedComparator* BasicComparison::makeUncheckedComparatorForTypesHelper(cons
const Type &right) const {
if (left.getSuperTypeID() == Type::kNumeric && right.getSuperTypeID() == Type::kNumeric) {
return makeNumericComparatorOuterHelper<LiteralComparator>(left, right);
- } else if ((left.getTypeID() == kDatetime && right.getTypeID() == kDatetime) ||
+ } else if ((left.getTypeID() == kDate && right.getTypeID() == kDate) ||
+ (left.getTypeID() == kDatetime && right.getTypeID() == kDatetime) ||
(left.getTypeID() == kDatetimeInterval && right.getTypeID() == kDatetimeInterval) ||
(left.getTypeID() == kYearMonthInterval && right.getTypeID() == kYearMonthInterval)) {
return makeDateComparatorOuterHelper<LiteralComparator>(left, right);
@@ -358,6 +374,12 @@ UncheckedComparator* BasicComparison::makeDateComparatorOuterHelper(
const Type &left,
const Type &right) const {
switch (left.getTypeID()) {
+ case kDate:
+ if (left.isNullable()) {
+ return makeDateComparatorInnerHelper<ComparatorType, DateLit, true>(left, right);
+ } else {
+ return makeDateComparatorInnerHelper<ComparatorType, DateLit, false>(left, right);
+ }
case kDatetime:
if (left.isNullable()) {
return makeDateComparatorInnerHelper<ComparatorType, DatetimeLit, true>(left, right);
@@ -389,6 +411,7 @@ UncheckedComparator* BasicComparison::makeDateComparatorInnerHelper(
const Type &left,
const Type &right) const {
switch (right.getTypeID()) {
+ case kDate:
case kDatetime:
case kDatetimeInterval:
case kYearMonthInterval:
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/cf81b5e6/types/operations/comparisons/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/types/operations/comparisons/CMakeLists.txt b/types/operations/comparisons/CMakeLists.txt
index d2dfecf..d48cdc2 100644
--- a/types/operations/comparisons/CMakeLists.txt
+++ b/types/operations/comparisons/CMakeLists.txt
@@ -64,8 +64,6 @@ target_link_libraries(quickstep_types_operations_comparisons_AsciiStringComparat
quickstep_utility_Macros)
target_link_libraries(quickstep_types_operations_comparisons_BasicComparison
glog
- quickstep_types_DatetimeLit
- quickstep_types_IntervalLit
quickstep_types_Type
quickstep_types_TypeErrors
quickstep_types_TypeFactory
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/cf81b5e6/types/operations/comparisons/ComparisonUtil.hpp
----------------------------------------------------------------------
diff --git a/types/operations/comparisons/ComparisonUtil.hpp b/types/operations/comparisons/ComparisonUtil.hpp
index d7da3fd..d589192 100644
--- a/types/operations/comparisons/ComparisonUtil.hpp
+++ b/types/operations/comparisons/ComparisonUtil.hpp
@@ -163,6 +163,11 @@ auto InvokeOnLessComparatorForTypeIgnoreNullability(const Type &type,
comp(0, 0);
return functor(comp);
}
+ case kDate: {
+ LessLiteralUncheckedComparator<DateLit, false,
+ DateLit, false> comp;
+ return functor(comp);
+ }
case kDatetime: {
LessLiteralUncheckedComparator<DatetimeLit, false,
DatetimeLit, false> comp;
@@ -429,6 +434,18 @@ auto InvokeOnLessComparatorForDifferentTypesIgnoreNullability(
left_type, right_type, functor);
}
}
+ case kDate: {
+ switch (right_type.getTypeID()) {
+ case kDate: {
+ LessLiteralUncheckedComparator<DateLit, false,
+ DateLit, false> comp;
+ return functor(comp);
+ }
+ default:
+ return comparison_util_detail::InvokeOnLessComparatorForDifferentTypesFallback(
+ left_type, right_type, functor);
+ }
+ }
case kDatetime: {
switch (right_type.getTypeID()) {
case kDatetime: {
@@ -766,6 +783,18 @@ auto InvokeOnBothLessComparatorsForDifferentTypesIgnoreNullability(
left_type, right_type, functor);
}
}
+ case kDate: {
+ switch (right_type.getTypeID()) {
+ case kDate: {
+ LessLiteralUncheckedComparator<DateLit, false,
+ DateLit, false> comp;
+ return functor(comp, comp);
+ }
+ default:
+ return comparison_util_detail::InvokeOnBothLessComparatorsForDifferentTypesFallback(
+ left_type, right_type, functor);
+ }
+ }
case kDatetime: {
switch (right_type.getTypeID()) {
case kDatetime: {
@@ -963,6 +992,8 @@ inline bool CheckUntypedValuesEqual(const Type &type, const void *left, const vo
return STLCharEqual(static_cast<const AsciiStringSuperType&>(type).getStringLength())(left, right);
case kVarChar:
return STLVarCharEqual()(left, right);
+ case kDate:
+ return STLLiteralEqual<DateLit>()(left, right);
case kDatetime:
return STLLiteralEqual<DatetimeLit>()(left, right);
case kDatetimeInterval:
@@ -1220,6 +1251,9 @@ inline TypedValue GetLastValueForType(const Type &type) {
return TypedValue(kChar, kLastString, 2);
case kVarChar:
return TypedValue(kVarChar, kLastString, 2);
+ case kDate: {
+ return TypedValue(DateLit::Create(99999, 12, 31));
+ }
case kDatetime: {
DatetimeLit lit;
lit.ticks = std::numeric_limits<std::int64_t>::max();
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/cf81b5e6/types/operations/comparisons/LiteralComparators.hpp
----------------------------------------------------------------------
diff --git a/types/operations/comparisons/LiteralComparators.hpp b/types/operations/comparisons/LiteralComparators.hpp
index 3bf6b00..406ae40 100644
--- a/types/operations/comparisons/LiteralComparators.hpp
+++ b/types/operations/comparisons/LiteralComparators.hpp
@@ -257,8 +257,8 @@ class LiteralUncheckedComparator : public UncheckedComparator {
};
/**
- * @brief The equal UncheckedComparator for seven literals
- * (int, std::int64_t, float, double, DatetimeLit, DatetimeIntervalLit, and YearMonthIntervalLit).
+ * @brief The equal UncheckedComparator for the following literals
+ * (int, std::int64_t, float, double, DateLit, DatetimeLit, DatetimeIntervalLit, and YearMonthIntervalLit).
**/
template <typename LeftCppType, bool left_nullable,
typename RightCppType, bool right_nullable>
@@ -267,8 +267,8 @@ using EqualLiteralUncheckedComparator = LiteralUncheckedComparator<EqualFunctor,
RightCppType, right_nullable>;
/**
- * @brief The not-equal UncheckedComparator for seven literals
- * (int, std::int64_t, float, double, DatetimeLit, DatetimeIntervalLit, and YearMonthIntervalLit).
+ * @brief The not-equal UncheckedComparator for the following literals
+ * (int, std::int64_t, float, double, DateLit, DatetimeLit, DatetimeIntervalLit, and YearMonthIntervalLit).
**/
template <typename LeftCppType, bool left_nullable,
typename RightCppType, bool right_nullable>
@@ -277,8 +277,8 @@ using NotEqualLiteralUncheckedComparator = LiteralUncheckedComparator<NotEqualFu
RightCppType, right_nullable>;
/**
- * @brief The less-than UncheckedComparator for seven literals
- * (int, std::int64_t, float, double, DatetimeLit, DatetimeIntervalLit, and YearMonthIntervalLit).
+ * @brief The less-than UncheckedComparator for the following literals
+ * (int, std::int64_t, float, double, DateLit, DatetimeLit, DatetimeIntervalLit, and YearMonthIntervalLit).
**/
template <typename LeftCppType, bool left_nullable,
typename RightCppType, bool right_nullable>
@@ -287,8 +287,8 @@ using LessLiteralUncheckedComparator = LiteralUncheckedComparator<LessFunctor,
RightCppType, right_nullable>;
/**
- * @brief The less-than-or-equal UncheckedComparator for seven literals
- * (int, std::int64_t, float, double, DatetimeLit, DatetimeIntervalLit, and YearMonthIntervalLit).
+ * @brief The less-than-or-equal UncheckedComparator for the following literals
+ * (int, std::int64_t, float, double, DateLit, DatetimeLit, DatetimeIntervalLit, and YearMonthIntervalLit).
**/
template <typename LeftCppType, bool left_nullable,
typename RightCppType, bool right_nullable>
@@ -297,8 +297,8 @@ using LessOrEqualLiteralUncheckedComparator = LiteralUncheckedComparator<LessOrE
RightCppType, right_nullable>;
/**
- * @brief The greater-than UncheckedComparator for seven literals
- * (int, std::int64_t, float, double, DatetimeLit, DatetimeIntervalLit, and YearMonthIntervalLit).
+ * @brief The greater-than UncheckedComparator for the following literals
+ * (int, std::int64_t, float, double, DateLit, DatetimeLit, DatetimeIntervalLit, and YearMonthIntervalLit).
**/
template <typename LeftCppType, bool left_nullable,
typename RightCppType, bool right_nullable>
@@ -307,8 +307,8 @@ using GreaterLiteralUncheckedComparator = LiteralUncheckedComparator<GreaterFunc
RightCppType, right_nullable>;
/**
- * @brief The greater-than-or-equal UncheckedComparator for seven literals
- * (int, std::int64_t, float, double, DatetimeLit, DatetimeIntervalLit, and YearMonthIntervalLit).
+ * @brief The greater-than-or-equal UncheckedComparator for the following literals
+ * (int, std::int64_t, float, double, DateLit, DatetimeLit, DatetimeIntervalLit, and YearMonthIntervalLit).
**/
template <typename LeftCppType, bool left_nullable,
typename RightCppType, bool right_nullable>
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/cf81b5e6/types/operations/comparisons/tests/Comparison_unittest.cpp
----------------------------------------------------------------------
diff --git a/types/operations/comparisons/tests/Comparison_unittest.cpp b/types/operations/comparisons/tests/Comparison_unittest.cpp
index 0d5e387..d09aab7 100644
--- a/types/operations/comparisons/tests/Comparison_unittest.cpp
+++ b/types/operations/comparisons/tests/Comparison_unittest.cpp
@@ -25,8 +25,6 @@
#include <utility>
#include <vector>
-#include "gtest/gtest.h"
-
#include "types/DatetimeLit.hpp"
#include "types/IntervalLit.hpp"
#include "types/Type.hpp"
@@ -40,6 +38,8 @@
#include "utility/Macros.hpp"
#include "utility/ScopedBuffer.hpp"
+#include "gtest/gtest.h"
+
using std::int64_t;
using std::min;
using std::numeric_limits;
@@ -147,6 +147,23 @@ class ComparisonTest : public ::testing::Test {
numeric_typed_values_.emplace_back(double_min_.get(),
&TypeFactory::GetType(kDouble, false));
+ date_null_.reset(new TypedValue(kDate));
+ date_positive_.reset(new TypedValue(DateLit::Create(2016, 7, 15)));
+ date_negative_.reset(new TypedValue(DateLit::Create(-18017, 4, 13)));
+ date_max_.reset(new TypedValue(DateLit::Create(99999, 12, 31)));
+ date_min_.reset(new TypedValue(DateLit::Create(-99999, 1, 1)));
+
+ date_typed_values_.emplace_back(date_null_.get(),
+ &TypeFactory::GetType(kDate, true));
+ date_typed_values_.emplace_back(date_positive_.get(),
+ &TypeFactory::GetType(kDate, false));
+ date_typed_values_.emplace_back(date_negative_.get(),
+ &TypeFactory::GetType(kDate, false));
+ date_typed_values_.emplace_back(date_max_.get(),
+ &TypeFactory::GetType(kDate, false));
+ date_typed_values_.emplace_back(date_min_.get(),
+ &TypeFactory::GetType(kDate, false));
+
datetime_null_.reset(new TypedValue(kDatetime));
DatetimeLit datetime;
datetime.ticks = 0;
@@ -347,6 +364,7 @@ class ComparisonTest : public ::testing::Test {
void checkComparison(const Comparison &comparison) {
checkNumericComparison(comparison);
checkStringComparison(comparison);
+ checkDateComparison(comparison);
checkDatetimeComparison(comparison);
checkDatetimeIntervalComparison(comparison);
checkYearMonthIntervalComparison(comparison);
@@ -524,6 +542,19 @@ class ComparisonTest : public ::testing::Test {
}
}
+ void checkDateComparison(const Comparison &comparison) {
+ for (const std::pair<const TypedValue*, const Type*> &left_item : date_typed_values_) {
+ for (const std::pair<const TypedValue*, const Type*> &right_item : date_typed_values_) {
+ checkDateComparisonChecked<DateLit>(comparison,
+ *left_item.first, *left_item.second,
+ *right_item.first, *right_item.second);
+ checkDateComparisonUnchecked<DateLit>(comparison,
+ *left_item.first, *left_item.second,
+ *right_item.first, *right_item.second);
+ }
+ }
+ }
+
void checkDatetimeComparison(const Comparison &comparison) {
for (const std::pair<const TypedValue*, const Type*> &left_item : datetime_typed_values_) {
for (const std::pair<const TypedValue*, const Type*> &right_item : datetime_typed_values_) {
@@ -719,6 +750,7 @@ class ComparisonTest : public ::testing::Test {
char_ref_extended_short_, char_lit_extended_short_,
char_ref_extended_long_, char_lit_extended_long_,
varchar_null_, varchar_ref_short_, varchar_lit_short_, varchar_ref_long_, varchar_lit_long_,
+ date_null_, date_positive_, date_negative_, date_max_, date_min_,
datetime_null_, datetime_zero_, datetime_positive_, datetime_negative_, datetime_max_, datetime_min_,
datetime_interval_null_, datetime_interval_zero_, datetime_interval_positive_, datetime_interval_negative_,
year_month_interval_null_, year_month_interval_zero_,
@@ -727,7 +759,7 @@ class ComparisonTest : public ::testing::Test {
unique_ptr<ScopedBuffer> extended_short_char_buffer_, extended_long_char_buffer_;
vector<pair<const TypedValue*, const Type*>>
- numeric_typed_values_, datetime_typed_values_, datetime_interval_typed_values_,
+ numeric_typed_values_, date_typed_values_, datetime_typed_values_, datetime_interval_typed_values_,
year_month_interval_typed_values_, string_typed_values_;
};
@@ -803,7 +835,7 @@ void CheckBasicComparisonApplicableToPartialTypes(const Comparison &comparison)
// If only one argument is known, any orderable Type is OK.
for (const TypeID type_id
- : {kInt, kLong, kFloat, kDouble, kDatetime, kDatetimeInterval, kYearMonthInterval}) {
+ : {kInt, kLong, kFloat, kDouble, kDate, kDatetime, kDatetimeInterval, kYearMonthInterval}) {
EXPECT_TRUE(comparison.canComparePartialTypes(
&TypeFactory::GetType(type_id, false), nullptr));
EXPECT_TRUE(comparison.canComparePartialTypes(
@@ -827,12 +859,12 @@ void CheckBasicComparisonApplicableToPartialTypes(const Comparison &comparison)
// If both types are known, expect the same thing as canCompareTypes().
for (const TypeID first_type_id
- : {kInt, kLong, kFloat, kDouble, kDatetime, kDatetimeInterval, kYearMonthInterval}) {
+ : {kInt, kLong, kFloat, kDouble, kDate, kDatetime, kDatetimeInterval, kYearMonthInterval}) {
const Type &first_type = TypeFactory::GetType(first_type_id, false);
const Type &first_type_nullable = TypeFactory::GetType(first_type_id, true);
for (const TypeID second_type_id
- : {kInt, kLong, kFloat, kDouble, kDatetime, kDatetimeInterval, kYearMonthInterval}) {
+ : {kInt, kLong, kFloat, kDouble, kDate, kDatetime, kDatetimeInterval, kYearMonthInterval}) {
const Type &second_type = TypeFactory::GetType(second_type_id, false);
const Type &second_type_nullable = TypeFactory::GetType(second_type_id, true);
EXPECT_EQ(comparison.canCompareTypes(first_type, second_type),
@@ -864,7 +896,7 @@ void CheckBasicComparisonApplicableToPartialTypes(const Comparison &comparison)
const Type &first_type_nullable = TypeFactory::GetType(first_type_id, 10, true);
for (const TypeID second_type_id
- : {kInt, kLong, kFloat, kDouble, kDatetime, kDatetimeInterval, kYearMonthInterval}) {
+ : {kInt, kLong, kFloat, kDouble, kDate, kDatetime, kDatetimeInterval, kYearMonthInterval}) {
const Type &second_type = TypeFactory::GetType(second_type_id, false);
const Type &second_type_nullable = TypeFactory::GetType(second_type_id, true);
EXPECT_EQ(comparison.canCompareTypes(first_type, second_type),
[2/2] incubator-quickstep git commit: Refactored messages processing
in both PolicyEnforcer and QueryManager.
Posted by zu...@apache.org.
Refactored messages processing in both PolicyEnforcer and QueryManager.
Project: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/commit/db0e7e3c
Tree: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/tree/db0e7e3c
Diff: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/diff/db0e7e3c
Branch: refs/heads/refactor-process-msg
Commit: db0e7e3ccf4c2d631670a4dc0cd2499f9b0294b4
Parents: cf81b5e
Author: Zuyu Zhang <zu...@apache.org>
Authored: Sat Jul 9 14:54:05 2016 -0700
Committer: Zuyu Zhang <zu...@twitter.com>
Committed: Sat Jul 16 17:23:05 2016 -0700
----------------------------------------------------------------------
query_execution/CMakeLists.txt | 10 +--
query_execution/PolicyEnforcer.cpp | 61 ++++++++++---
query_execution/PolicyEnforcer.hpp | 10 ++-
query_execution/QueryManagerBase.cpp | 91 ++------------------
query_execution/QueryManagerBase.hpp | 90 ++++++++++---------
query_execution/QueryManagerSingleNode.cpp | 4 +-
query_execution/QueryManagerSingleNode.hpp | 4 +-
.../tests/QueryManagerSingleNode_unittest.cpp | 79 +++--------------
8 files changed, 126 insertions(+), 223 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/db0e7e3c/query_execution/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/query_execution/CMakeLists.txt b/query_execution/CMakeLists.txt
index 5a9189c..028531d 100644
--- a/query_execution/CMakeLists.txt
+++ b/query_execution/CMakeLists.txt
@@ -91,8 +91,12 @@ target_link_libraries(quickstep_queryexecution_ForemanSingleNode
${GFLAGS_LIB_NAME})
target_link_libraries(quickstep_queryexecution_PolicyEnforcer
glog
+ quickstep_catalog_CatalogDatabase
+ quickstep_catalog_CatalogRelation
quickstep_catalog_CatalogTypedefs
+ quickstep_catalog_PartitionScheme
quickstep_queryexecution_QueryExecutionMessages_proto
+ quickstep_queryexecution_QueryExecutionState
quickstep_queryexecution_QueryExecutionTypedefs
quickstep_queryexecution_QueryManagerBase
quickstep_queryexecution_QueryManagerSingleNode
@@ -100,6 +104,7 @@ target_link_libraries(quickstep_queryexecution_PolicyEnforcer
quickstep_queryexecution_WorkerMessage
quickstep_queryoptimizer_QueryHandle
quickstep_relationaloperators_WorkOrder
+ quickstep_storage_StorageBlockInfo
quickstep_utility_Macros
tmb
${GFLAGS_LIB_NAME})
@@ -152,14 +157,9 @@ target_link_libraries(quickstep_queryexecution_QueryExecutionUtil
quickstep_utility_Macros
tmb)
target_link_libraries(quickstep_queryexecution_QueryManagerBase
- quickstep_catalog_CatalogDatabase
- quickstep_catalog_CatalogRelation
quickstep_catalog_CatalogTypedefs
- quickstep_catalog_PartitionScheme
quickstep_queryexecution_QueryContext
- quickstep_queryexecution_QueryExecutionMessages_proto
quickstep_queryexecution_QueryExecutionState
- quickstep_queryexecution_QueryExecutionTypedefs
quickstep_queryoptimizer_QueryHandle
quickstep_queryoptimizer_QueryPlan
quickstep_relationaloperators_RelationalOperator
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/db0e7e3c/query_execution/PolicyEnforcer.cpp
----------------------------------------------------------------------
diff --git a/query_execution/PolicyEnforcer.cpp b/query_execution/PolicyEnforcer.cpp
index f310ee1..4cba8c5 100644
--- a/query_execution/PolicyEnforcer.cpp
+++ b/query_execution/PolicyEnforcer.cpp
@@ -24,12 +24,19 @@
#include <unordered_map>
#include <vector>
+#include "catalog/CatalogDatabase.hpp"
+#include "catalog/CatalogRelation.hpp"
#include "catalog/CatalogTypedefs.hpp"
+#include "catalog/PartitionScheme.hpp"
#include "query_execution/QueryExecutionMessages.pb.h"
+#include "query_execution/QueryExecutionState.hpp"
+#include "query_execution/QueryManagerBase.hpp"
#include "query_execution/QueryManagerSingleNode.hpp"
#include "query_execution/WorkerDirectory.hpp"
+#include "query_execution/WorkerMessage.hpp"
#include "query_optimizer/QueryHandle.hpp"
#include "relational_operators/WorkOrder.hpp"
+#include "storage/StorageBlockInfo.hpp"
#include "gflags/gflags.h"
#include "glog/logging.h"
@@ -62,10 +69,9 @@ bool PolicyEnforcer::admitQuery(QueryHandle *query_handle) {
}
void PolicyEnforcer::processMessage(const TaggedMessage &tagged_message) {
- // TODO(harshad) : Provide processXMessage() public functions in
- // QueryManager, so that we need to extract message from the
- // TaggedMessage only once.
std::size_t query_id;
+ QueryManagerBase::dag_node_index op_index;
+
switch (tagged_message.message_type()) {
case kWorkOrderCompleteMessage: {
serialization::NormalWorkOrderCompletionMessage proto;
@@ -73,12 +79,17 @@ void PolicyEnforcer::processMessage(const TaggedMessage &tagged_message) {
// WorkOrder. It can be accessed in this scope.
CHECK(proto.ParseFromArray(tagged_message.message(),
tagged_message.message_bytes()));
- query_id = proto.query_id();
worker_directory_->decrementNumQueuedWorkOrders(
proto.worker_thread_index());
if (profile_individual_workorders_) {
recordTimeForWorkOrder(proto);
}
+
+ query_id = proto.query_id();
+ DCHECK(admitted_queries_.find(query_id) != admitted_queries_.end());
+
+ op_index = proto.operator_index();
+ admitted_queries_[query_id]->processWorkOrderCompleteMessage(op_index);
break;
}
case kRebuildWorkOrderCompleteMessage: {
@@ -87,23 +98,43 @@ void PolicyEnforcer::processMessage(const TaggedMessage &tagged_message) {
// rebuild WorkOrder. It can be accessed in this scope.
CHECK(proto.ParseFromArray(tagged_message.message(),
tagged_message.message_bytes()));
- query_id = proto.query_id();
worker_directory_->decrementNumQueuedWorkOrders(
proto.worker_thread_index());
+
+ query_id = proto.query_id();
+ DCHECK(admitted_queries_.find(query_id) != admitted_queries_.end());
+
+ op_index = proto.operator_index();
+ admitted_queries_[query_id]->processRebuildWorkOrderCompleteMessage(op_index);
break;
}
case kCatalogRelationNewBlockMessage: {
serialization::CatalogRelationNewBlockMessage proto;
CHECK(proto.ParseFromArray(tagged_message.message(),
tagged_message.message_bytes()));
- query_id = proto.query_id();
- break;
+
+ const block_id block = proto.block_id();
+
+ CatalogRelation *relation =
+ static_cast<CatalogDatabase*>(catalog_database_)->getRelationByIdMutable(proto.relation_id());
+ relation->addBlock(block);
+
+ if (proto.has_partition_id()) {
+ relation->getPartitionSchemeMutable()->addBlockToPartition(
+ proto.partition_id(), block);
+ }
+ return;
}
case kDataPipelineMessage: {
serialization::DataPipelineMessage proto;
CHECK(proto.ParseFromArray(tagged_message.message(),
tagged_message.message_bytes()));
query_id = proto.query_id();
+ DCHECK(admitted_queries_.find(query_id) != admitted_queries_.end());
+
+ op_index = proto.operator_index();
+ admitted_queries_[query_id]->processDataPipelineMessage(
+ op_index, proto.block_id(), proto.relation_id());
break;
}
case kWorkOrdersAvailableMessage: {
@@ -111,6 +142,12 @@ void PolicyEnforcer::processMessage(const TaggedMessage &tagged_message) {
CHECK(proto.ParseFromArray(tagged_message.message(),
tagged_message.message_bytes()));
query_id = proto.query_id();
+ DCHECK(admitted_queries_.find(query_id) != admitted_queries_.end());
+
+ op_index = proto.operator_index();
+
+ // Check if new work orders are available.
+ admitted_queries_[query_id]->fetchNormalWorkOrders(op_index);
break;
}
case kWorkOrderFeedbackMessage: {
@@ -118,15 +155,17 @@ void PolicyEnforcer::processMessage(const TaggedMessage &tagged_message) {
const_cast<void *>(tagged_message.message()),
tagged_message.message_bytes());
query_id = msg.header().query_id;
+ DCHECK(admitted_queries_.find(query_id) != admitted_queries_.end());
+
+ op_index = msg.header().rel_op_index;
+ admitted_queries_[query_id]->processFeedbackMessage(op_index, msg);
break;
}
default:
LOG(FATAL) << "Unknown message type found in PolicyEnforcer";
}
- DCHECK(admitted_queries_.find(query_id) != admitted_queries_.end());
- const QueryManagerBase::QueryStatusCode return_code =
- admitted_queries_[query_id]->processMessage(tagged_message);
- if (return_code == QueryManagerBase::QueryStatusCode::kQueryExecuted) {
+ if (admitted_queries_[query_id]->queryStatus(op_index) ==
+ QueryManagerBase::QueryStatusCode::kQueryExecuted) {
removeQuery(query_id);
if (!waiting_queries_.empty()) {
// Admit the earliest waiting query.
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/db0e7e3c/query_execution/PolicyEnforcer.hpp
----------------------------------------------------------------------
diff --git a/query_execution/PolicyEnforcer.hpp b/query_execution/PolicyEnforcer.hpp
index 79e61d1..8bd6d92 100644
--- a/query_execution/PolicyEnforcer.hpp
+++ b/query_execution/PolicyEnforcer.hpp
@@ -33,8 +33,8 @@
#include "glog/logging.h"
#include "tmb/id_typedefs.h"
-#include "tmb/message_bus.h"
-#include "tmb/tagged_message.h"
+
+namespace tmb { class MessageBus; }
namespace quickstep {
@@ -43,6 +43,12 @@ class QueryHandle;
class StorageManager;
class WorkerDirectory;
+namespace serialization { class NormalWorkOrderCompletionMessage; }
+
+/** \addtogroup QueryExecution
+ * @{
+ */
+
/**
* @brief A class that ensures that a high level policy is maintained
* in sharing resources among concurrent queries.
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/db0e7e3c/query_execution/QueryManagerBase.cpp
----------------------------------------------------------------------
diff --git a/query_execution/QueryManagerBase.cpp b/query_execution/QueryManagerBase.cpp
index f7e183f..37beb02 100644
--- a/query_execution/QueryManagerBase.cpp
+++ b/query_execution/QueryManagerBase.cpp
@@ -21,13 +21,8 @@
#include <utility>
#include <vector>
-#include "catalog/CatalogDatabase.hpp"
-#include "catalog/CatalogRelation.hpp"
#include "catalog/CatalogTypedefs.hpp"
-#include "catalog/PartitionScheme.hpp"
#include "query_execution/QueryContext.hpp"
-#include "query_execution/QueryExecutionMessages.pb.h"
-#include "query_execution/QueryExecutionTypedefs.hpp"
#include "query_optimizer/QueryHandle.hpp"
#include "query_optimizer/QueryPlan.hpp"
#include "relational_operators/WorkOrder.hpp"
@@ -39,10 +34,8 @@ using std::pair;
namespace quickstep {
-QueryManagerBase::QueryManagerBase(QueryHandle *query_handle,
- CatalogDatabaseLite *catalog_database)
+QueryManagerBase::QueryManagerBase(QueryHandle *query_handle)
: query_id_(DCHECK_NOTNULL(query_handle)->query_id()),
- catalog_database_(DCHECK_NOTNULL(catalog_database)),
query_dag_(DCHECK_NOTNULL(
DCHECK_NOTNULL(query_handle->getQueryPlanMutable())->getQueryPlanDAGMutable())),
num_operators_in_dag_(query_dag_->size()),
@@ -76,82 +69,8 @@ QueryManagerBase::QueryManagerBase(QueryHandle *query_handle,
}
}
-QueryManagerBase::QueryStatusCode QueryManagerBase::processMessage(
- const TaggedMessage &tagged_message) {
- dag_node_index op_index;
- switch (tagged_message.message_type()) {
- case kWorkOrderCompleteMessage: {
- serialization::NormalWorkOrderCompletionMessage proto;
- CHECK(proto.ParseFromArray(tagged_message.message(),
- tagged_message.message_bytes()));
-
- op_index = proto.operator_index();
- processWorkOrderCompleteMessage(proto.operator_index());
- break;
- }
- case kRebuildWorkOrderCompleteMessage: {
- serialization::RebuildWorkOrderCompletionMessage proto;
- CHECK(proto.ParseFromArray(tagged_message.message(),
- tagged_message.message_bytes()));
-
- op_index = proto.operator_index();
- processRebuildWorkOrderCompleteMessage(proto.operator_index());
- break;
- }
- case kCatalogRelationNewBlockMessage: {
- serialization::CatalogRelationNewBlockMessage proto;
- CHECK(proto.ParseFromArray(tagged_message.message(),
- tagged_message.message_bytes()));
-
- const block_id block = proto.block_id();
-
- CatalogRelation *relation =
- static_cast<CatalogDatabase*>(catalog_database_)->getRelationByIdMutable(proto.relation_id());
- relation->addBlock(block);
-
- if (proto.has_partition_id()) {
- relation->getPartitionSchemeMutable()->addBlockToPartition(
- proto.partition_id(), block);
- }
- return QueryStatusCode::kNone;
- }
- case kDataPipelineMessage: {
- // Possible message senders include InsertDestinations and some
- // operators which modify existing blocks.
- serialization::DataPipelineMessage proto;
- CHECK(proto.ParseFromArray(tagged_message.message(),
- tagged_message.message_bytes()));
-
- op_index = proto.operator_index();
- processDataPipelineMessage(proto.operator_index(),
- proto.block_id(),
- proto.relation_id());
- break;
- }
- case kWorkOrdersAvailableMessage: {
- serialization::WorkOrdersAvailableMessage proto;
- CHECK(proto.ParseFromArray(tagged_message.message(),
- tagged_message.message_bytes()));
-
- op_index = proto.operator_index();
-
- // Check if new work orders are available.
- fetchNormalWorkOrders(op_index);
- break;
- }
- case kWorkOrderFeedbackMessage: {
- WorkOrder::FeedbackMessage msg(
- const_cast<void *>(tagged_message.message()),
- tagged_message.message_bytes());
-
- op_index = msg.header().rel_op_index;
- processFeedbackMessage(msg);
- break;
- }
- default:
- LOG(FATAL) << "Unknown message type found in QueryManager";
- }
-
+QueryManagerBase::QueryStatusCode QueryManagerBase::queryStatus(
+ const dag_node_index op_index) {
if (query_exec_state_->hasExecutionFinished(op_index)) {
return QueryStatusCode::kOperatorExecuted;
}
@@ -165,9 +84,9 @@ QueryManagerBase::QueryStatusCode QueryManagerBase::processMessage(
}
void QueryManagerBase::processFeedbackMessage(
- const WorkOrder::FeedbackMessage &msg) {
+ const dag_node_index op_index, const WorkOrder::FeedbackMessage &msg) {
RelationalOperator *op =
- query_dag_->getNodePayloadMutable(msg.header().rel_op_index);
+ query_dag_->getNodePayloadMutable(op_index);
op->receiveFeedbackMessage(msg);
}
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/db0e7e3c/query_execution/QueryManagerBase.hpp
----------------------------------------------------------------------
diff --git a/query_execution/QueryManagerBase.hpp b/query_execution/QueryManagerBase.hpp
index 9e192c8..6edfd5c 100644
--- a/query_execution/QueryManagerBase.hpp
+++ b/query_execution/QueryManagerBase.hpp
@@ -24,7 +24,6 @@
#include "catalog/CatalogTypedefs.hpp"
#include "query_execution/QueryExecutionState.hpp"
-#include "query_execution/QueryExecutionTypedefs.hpp"
#include "relational_operators/RelationalOperator.hpp"
#include "relational_operators/WorkOrder.hpp"
#include "storage/StorageBlockInfo.hpp"
@@ -33,7 +32,6 @@
namespace quickstep {
-class CatalogDatabaseLite;
class QueryHandle;
/** \addtogroup QueryExecution
@@ -50,7 +48,7 @@ class QueryManagerBase {
typedef DAG<RelationalOperator, bool>::size_type_nodes dag_node_index;
/**
- * @brief Return codes for processMessage() function.
+ * @brief Return codes for queryStatus() function.
*
* @note When both operator and query get executed, kQueryExecuted takes
* precedence over kOperatorExecuted.
@@ -65,10 +63,8 @@ class QueryManagerBase {
* @brief Constructor.
*
* @param query_handle The QueryHandle object for this query.
- * @param catalog_database The CatalogDatabse used by the query.
**/
- QueryManagerBase(QueryHandle *query_handle,
- CatalogDatabaseLite *catalog_database);
+ explicit QueryManagerBase(QueryHandle *query_handle);
/**
* @brief Virtual destructor.
@@ -76,26 +72,16 @@ class QueryManagerBase {
virtual ~QueryManagerBase() {}
/**
- * @brief Process a message sent to the QueryManager.
- *
- * @param tagged_message TaggedMessage sent to the QueryManager.
- *
- * @return QueryStatusCode as determined after the message is processed.
- **/
- QueryStatusCode processMessage(const TaggedMessage &tagged_message);
-
- /**
* @brief Get the QueryExecutionState for this query.
**/
inline const QueryExecutionState& getQueryExecutionState() const {
return *query_exec_state_;
}
- protected:
/**
* @brief Process the received WorkOrder complete message.
*
- * @param node_index The index of the specified operator node in the query DAG
+ * @param op_index The index of the specified operator node in the query DAG
* for the completed WorkOrder.
**/
void processWorkOrderCompleteMessage(const dag_node_index op_index);
@@ -103,28 +89,15 @@ class QueryManagerBase {
/**
* @brief Process the received RebuildWorkOrder complete message.
*
- * @param node_index The index of the specified operator node in the query DAG
+ * @param op_index The index of the specified operator node in the query DAG
* for the completed RebuildWorkOrder.
**/
void processRebuildWorkOrderCompleteMessage(const dag_node_index op_index);
/**
- * @brief Process a current relational operator: Get its workorders and store
- * them in the WorkOrdersContainer for this query. If the operator can
- * be marked as done, do so.
- *
- * @param index The index of the relational operator to be processed in the
- * query plan DAG.
- * @param recursively_check_dependents If an operator is done, should we
- * call processOperator on its dependents recursively.
- **/
- void processOperator(const dag_node_index index,
- const bool recursively_check_dependents);
-
- /**
* @brief Process the received data pipeline message.
*
- * @param node_index The index of the specified operator node in the query DAG
+ * @param op_index The index of the specified operator node in the query DAG
* for the pipelining block.
* @param block The block id.
* @param rel_id The ID of the relation that produced 'block'.
@@ -134,12 +107,50 @@ class QueryManagerBase {
const relation_id rel_id);
/**
+ * @brief Fetch all work orders currently available in relational operator and
+ * store them internally.
+ *
+ * @param index The index of the relational operator to be processed in the
+ * query plan DAG.
+ *
+ * @return Whether any work order was generated by op.
+ **/
+ virtual bool fetchNormalWorkOrders(const dag_node_index index) = 0;
+
+ /**
* @brief Process the received work order feedback message and notify
* relational operator.
*
+ * @param op_index The index of the specified operator node in the query DAG
+ * for the feedback message.
* @param message Feedback message from work order.
**/
- void processFeedbackMessage(const WorkOrder::FeedbackMessage &message);
+ void processFeedbackMessage(const dag_node_index op_index,
+ const WorkOrder::FeedbackMessage &message);
+
+ /**
+ * @brief Get the query status after processing an incoming message.
+ *
+ * @param op_index The index of the specified operator node in the query DAG
+ * for the incoming message.
+ *
+ * @return QueryStatusCode as determined after the message is processed.
+ **/
+ QueryStatusCode queryStatus(const dag_node_index op_index);
+
+ protected:
+ /**
+ * @brief Process a current relational operator: Get its workorders and store
+ * them in the WorkOrdersContainer for this query. If the operator can
+ * be marked as done, do so.
+ *
+ * @param index The index of the relational operator to be processed in the
+ * query plan DAG.
+ * @param recursively_check_dependents If an operator is done, should we
+ * call processOperator on its dependents recursively.
+ **/
+ void processOperator(const dag_node_index index,
+ const bool recursively_check_dependents);
/**
* @brief This function does the following things:
@@ -241,8 +252,6 @@ class QueryManagerBase {
const std::size_t query_id_;
- CatalogDatabaseLite *catalog_database_;
-
DAG<RelationalOperator, bool> *query_dag_;
const dag_node_index num_operators_in_dag_;
@@ -256,17 +265,6 @@ class QueryManagerBase {
private:
/**
- * @brief Fetch all work orders currently available in relational operator and
- * store them internally.
- *
- * @param index The index of the relational operator to be processed in the
- * query plan DAG.
- *
- * @return Whether any work order was generated by op.
- **/
- virtual bool fetchNormalWorkOrders(const dag_node_index index) = 0;
-
- /**
* @brief Check if the given operator's normal execution is over.
*
* @note The conditions for a given operator's normal execution to get over:
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/db0e7e3c/query_execution/QueryManagerSingleNode.cpp
----------------------------------------------------------------------
diff --git a/query_execution/QueryManagerSingleNode.cpp b/query_execution/QueryManagerSingleNode.cpp
index 193b188..12f8ff5 100644
--- a/query_execution/QueryManagerSingleNode.cpp
+++ b/query_execution/QueryManagerSingleNode.cpp
@@ -46,12 +46,12 @@ QueryManagerSingleNode::QueryManagerSingleNode(
CatalogDatabaseLite *catalog_database,
StorageManager *storage_manager,
tmb::MessageBus *bus)
- : QueryManagerBase(query_handle, catalog_database),
+ : QueryManagerBase(query_handle),
foreman_client_id_(foreman_client_id),
storage_manager_(DCHECK_NOTNULL(storage_manager)),
bus_(DCHECK_NOTNULL(bus)),
query_context_(new QueryContext(query_handle->getQueryContextProto(),
- *catalog_database_,
+ *catalog_database,
storage_manager_,
foreman_client_id_,
bus_)),
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/db0e7e3c/query_execution/QueryManagerSingleNode.hpp
----------------------------------------------------------------------
diff --git a/query_execution/QueryManagerSingleNode.hpp b/query_execution/QueryManagerSingleNode.hpp
index 5533f06..b9bad2e 100644
--- a/query_execution/QueryManagerSingleNode.hpp
+++ b/query_execution/QueryManagerSingleNode.hpp
@@ -68,6 +68,8 @@ class QueryManagerSingleNode final : public QueryManagerBase {
~QueryManagerSingleNode() override {}
+ bool fetchNormalWorkOrders(const dag_node_index index) override;
+
/**
* @brief Get the next workorder to be excuted, wrapped in a WorkerMessage.
*
@@ -91,8 +93,6 @@ class QueryManagerSingleNode final : public QueryManagerBase {
}
private:
- bool fetchNormalWorkOrders(const dag_node_index index) override;
-
bool checkNormalExecutionOver(const dag_node_index index) const override {
return (checkAllDependenciesMet(index) &&
!workorders_container_->hasNormalWorkOrder(index) &&
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/db0e7e3c/query_execution/tests/QueryManagerSingleNode_unittest.cpp
----------------------------------------------------------------------
diff --git a/query_execution/tests/QueryManagerSingleNode_unittest.cpp b/query_execution/tests/QueryManagerSingleNode_unittest.cpp
index 52cee20..39ca58c 100644
--- a/query_execution/tests/QueryManagerSingleNode_unittest.cpp
+++ b/query_execution/tests/QueryManagerSingleNode_unittest.cpp
@@ -16,7 +16,6 @@
**/
#include <climits>
-#include <cstdlib>
#include <memory>
#include <utility>
#include <vector>
@@ -26,7 +25,6 @@
#include "catalog/CatalogTypedefs.hpp"
#include "query_execution/QueryContext.hpp"
#include "query_execution/QueryContext.pb.h"
-#include "query_execution/QueryExecutionMessages.pb.h"
#include "query_execution/QueryExecutionState.hpp"
#include "query_execution/QueryExecutionTypedefs.hpp"
#include "query_execution/QueryManagerSingleNode.hpp"
@@ -49,7 +47,6 @@
#include "gtest/gtest.h"
#include "tmb/id_typedefs.h"
-#include "tmb/tagged_message.h"
namespace tmb { class MessageBus; }
@@ -254,89 +251,33 @@ class QueryManagerTest : public ::testing::Test {
inline bool placeDataPipelineMessage(const QueryPlan::DAGNodeIndex source_operator_index) {
VLOG(3) << "Place DataPipeline message for Op[" << source_operator_index << "]";
- serialization::DataPipelineMessage proto;
- proto.set_operator_index(source_operator_index);
-
- proto.set_block_id(0); // dummy block ID
- proto.set_relation_id(0); // dummy relation ID.
- proto.set_query_id(0); // dummy query ID.
-
- // NOTE(zuyu): Using the heap memory to serialize proto as a c-like string.
- const std::size_t proto_length = proto.ByteSize();
- char *proto_bytes = static_cast<char*>(std::malloc(proto_length));
- CHECK(proto.SerializeToArray(proto_bytes, proto_length));
-
- tmb::TaggedMessage tagged_message(static_cast<const void *>(proto_bytes),
- proto_length,
- kDataPipelineMessage);
- std::free(proto_bytes);
- query_manager_->processMessage(tagged_message);
+
+ query_manager_->processDataPipelineMessage(source_operator_index,
+ 0 /* dummy block ID */,
+ 0 /* dummy relation ID */);
return query_manager_->getQueryExecutionState().hasQueryExecutionFinished();
}
inline bool placeWorkOrderCompleteMessage(const QueryPlan::DAGNodeIndex index) {
VLOG(3) << "Place WorkOrderComplete message for Op[" << index << "]";
- TaggedMessage tagged_message;
- serialization::NormalWorkOrderCompletionMessage proto;
- proto.set_operator_index(index);
- proto.set_worker_thread_index(1); // dummy worker ID.
- proto.set_query_id(0); // dummy query ID.
-
- // NOTE(zuyu): Using the heap memory to serialize proto as a c-like string.
- const size_t proto_length = proto.ByteSize();
- char *proto_bytes = static_cast<char*>(std::malloc(proto_length));
- CHECK(proto.SerializeToArray(proto_bytes, proto_length));
-
- TaggedMessage message(static_cast<const void*>(proto_bytes),
- proto_length,
- kWorkOrderCompleteMessage);
- std::free(proto_bytes);
- query_manager_->processMessage(message);
+ query_manager_->processWorkOrderCompleteMessage(index);
return query_manager_->getQueryExecutionState().hasQueryExecutionFinished();
}
inline bool placeRebuildWorkOrderCompleteMessage(const QueryPlan::DAGNodeIndex index) {
VLOG(3) << "Place RebuildWorkOrderComplete message for Op[" << index << "]";
- serialization::RebuildWorkOrderCompletionMessage proto;
- proto.set_operator_index(index);
- proto.set_worker_thread_index(1); // dummy worker thread ID.
- proto.set_query_id(0); // dummy query ID.
-
- // NOTE(zuyu): Using the heap memory to serialize proto as a c-like string.
- const size_t proto_length = proto.ByteSize();
- char *proto_bytes = static_cast<char*>(std::malloc(proto_length));
- CHECK(proto.SerializeToArray(proto_bytes, proto_length));
-
- TaggedMessage message(static_cast<const void*>(proto_bytes),
- proto_length,
- kRebuildWorkOrderCompleteMessage);
-
- std::free(proto_bytes);
- query_manager_->processMessage(message);
+ query_manager_->processRebuildWorkOrderCompleteMessage(index);
return query_manager_->getQueryExecutionState().hasQueryExecutionFinished();
}
inline bool placeOutputBlockMessage(const QueryPlan::DAGNodeIndex index) {
VLOG(3) << "Place OutputBlock message for Op[" << index << "]";
- serialization::DataPipelineMessage proto;
- proto.set_operator_index(index);
-
- proto.set_block_id(0); // dummy block ID
- proto.set_relation_id(0); // dummy relation ID.
- proto.set_query_id(0); // dummy query ID.
-
- // NOTE(zuyu): Using the heap memory to serialize proto as a c-like string.
- const std::size_t proto_length = proto.ByteSize();
- char *proto_bytes = static_cast<char*>(std::malloc(proto_length));
- CHECK(proto.SerializeToArray(proto_bytes, proto_length));
-
- tmb::TaggedMessage tagged_message(static_cast<const void *>(proto_bytes),
- proto_length,
- kDataPipelineMessage);
- std::free(proto_bytes);
- query_manager_->processMessage(tagged_message);
+
+ query_manager_->processDataPipelineMessage(index,
+ 0 /* dummy block ID */,
+ 0 /* dummy relation ID */);
return query_manager_->getQueryExecutionState().hasQueryExecutionFinished();
}