You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@quickstep.apache.org by ji...@apache.org on 2017/08/26 02:59:00 UTC
incubator-quickstep git commit: Some refactorings
Repository: incubator-quickstep
Updated Branches:
refs/heads/copy-to 1f9bc9147 -> c82144197
Some refactorings
Project: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/commit/c8214419
Tree: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/tree/c8214419
Diff: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/diff/c8214419
Branch: refs/heads/copy-to
Commit: c82144197b4969c63aacd2087fd69e056d979610
Parents: 1f9bc91
Author: Jianqiao Zhu <ji...@cs.wisc.edu>
Authored: Fri Aug 25 21:58:56 2017 -0500
Committer: Jianqiao Zhu <ji...@cs.wisc.edu>
Committed: Fri Aug 25 21:58:56 2017 -0500
----------------------------------------------------------------------
query_execution/QueryContext.hpp | 34 ------
query_optimizer/physical/CopyFrom.hpp | 4 +-
query_optimizer/resolver/Resolver.cpp | 9 +-
.../tests/execution_generator/Copy.test | 35 +++---
relational_operators/TableExportOperator.cpp | 107 +++++++++++--------
relational_operators/TableExportOperator.hpp | 89 +++++++++++----
6 files changed, 153 insertions(+), 125 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/c8214419/query_execution/QueryContext.hpp
----------------------------------------------------------------------
diff --git a/query_execution/QueryContext.hpp b/query_execution/QueryContext.hpp
index b146e13..7876821 100644
--- a/query_execution/QueryContext.hpp
+++ b/query_execution/QueryContext.hpp
@@ -23,7 +23,6 @@
#include <cstddef>
#include <cstdint>
#include <memory>
-#include <string>
#include <unordered_map>
#include <vector>
@@ -34,7 +33,6 @@
#include "storage/AggregationOperationState.hpp"
#include "storage/HashTable.hpp"
#include "storage/InsertDestination.hpp"
-#include "storage/StorageBlockInfo.hpp"
#include "storage/WindowAggregationOperationState.hpp"
#include "threading/SpinSharedMutex.hpp"
#include "types/containers/Tuple.hpp"
@@ -573,36 +571,6 @@ class QueryContext {
}
/**
- * @brief Release the output text buffer for the specified block.
- *
- * @param id The id of the block to release its output text buffer.
- *
- * @return The output text buffer for the specified block. Caller should take
- * ownership of the returned object.
- **/
- inline std::string* releaseBlockOutputTextBuffer(const block_id id) {
- SpinSharedMutexExclusiveLock<false> lock(block_output_text_buffer_mutex_);
- auto it = block_output_text_buffers_.find(id);
- DCHECK(it != block_output_text_buffers_.end());
- std::unique_ptr<std::string> output = std::move(it->second);
- block_output_text_buffers_.erase(it);
- return output.release();
- }
-
- /**
- * @brief Set the output text buffer for the specified block.
- *
- * @param id The id of the block.
- * @param output The output buffer to set. Caller relinquishes ownership of
- * \p output by calling this method.
- **/
- inline void setBlockOutputTextBuffer(const block_id id, std::string *output) {
- SpinSharedMutexExclusiveLock<false> lock(block_output_text_buffer_mutex_);
- DCHECK(block_output_text_buffers_.find(id) == block_output_text_buffers_.end());
- block_output_text_buffers_.emplace(id, std::unique_ptr<std::string>(output));
- }
-
- /**
* @brief Get the total memory footprint of the temporary data structures
* used for query execution (e.g. join hash tables, aggregation hash
* tables) in bytes.
@@ -666,12 +634,10 @@ class QueryContext {
std::vector<std::unique_ptr<Tuple>> tuples_;
std::vector<std::unordered_map<attribute_id, std::unique_ptr<const Scalar>>> update_groups_;
std::vector<std::unique_ptr<WindowAggregationOperationState>> window_aggregation_states_;
- std::unordered_map<block_id, std::unique_ptr<std::string>> block_output_text_buffers_;
mutable SpinSharedMutex<false> hash_tables_mutex_;
mutable SpinSharedMutex<false> aggregation_states_mutex_;
mutable SpinSharedMutex<false> insert_destinations_mutex_;
- mutable SpinSharedMutex<false> block_output_text_buffer_mutex_;
DISALLOW_COPY_AND_ASSIGN(QueryContext);
};
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/c8214419/query_optimizer/physical/CopyFrom.hpp
----------------------------------------------------------------------
diff --git a/query_optimizer/physical/CopyFrom.hpp b/query_optimizer/physical/CopyFrom.hpp
index 6aa60cb..b100eaa 100644
--- a/query_optimizer/physical/CopyFrom.hpp
+++ b/query_optimizer/physical/CopyFrom.hpp
@@ -76,9 +76,7 @@ class CopyFrom : public Physical {
PhysicalPtr copyWithNewChildren(
const std::vector<PhysicalPtr> &new_children) const override {
DCHECK(new_children.empty());
- return Create(catalog_relation_,
- file_name_,
- options_);
+ return Create(catalog_relation_, file_name_, options_);
}
std::vector<expressions::AttributeReferencePtr> getOutputAttributes() const override {
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/c8214419/query_optimizer/resolver/Resolver.cpp
----------------------------------------------------------------------
diff --git a/query_optimizer/resolver/Resolver.cpp b/query_optimizer/resolver/Resolver.cpp
index 45ccc56..3215707 100644
--- a/query_optimizer/resolver/Resolver.cpp
+++ b/query_optimizer/resolver/Resolver.cpp
@@ -147,8 +147,9 @@ namespace S = ::quickstep::serialization;
namespace {
-attribute_id GetAttributeIdFromName(const PtrList<ParseAttributeDefinition> &attribute_definition_list,
- const std::string &attribute_name) {
+static attribute_id GetAttributeIdFromName(
+ const PtrList<ParseAttributeDefinition> &attribute_definition_list,
+ const std::string &attribute_name) {
const std::string lower_attribute_name = ToLower(attribute_name);
attribute_id attr_id = 0;
@@ -163,7 +164,7 @@ attribute_id GetAttributeIdFromName(const PtrList<ParseAttributeDefinition> &att
return kInvalidAttributeID;
}
-const ParseString* GetKeyValueString(const ParseKeyValue &key_value) {
+static const ParseString* GetKeyValueString(const ParseKeyValue &key_value) {
if (key_value.getKeyValueType() != ParseKeyValue::kStringString) {
THROW_SQL_ERROR_AT(&key_value)
<< "Invalid value type for " << key_value.key()->value()
@@ -172,7 +173,7 @@ const ParseString* GetKeyValueString(const ParseKeyValue &key_value) {
return static_cast<const ParseKeyStringValue&>(key_value).value();
}
-bool GetKeyValueBool(const ParseKeyValue &key_value) {
+static bool GetKeyValueBool(const ParseKeyValue &key_value) {
if (key_value.getKeyValueType() != ParseKeyValue::kStringBool) {
THROW_SQL_ERROR_AT(&key_value)
<< "Invalid value for " << key_value.key()->value()
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/c8214419/query_optimizer/tests/execution_generator/Copy.test
----------------------------------------------------------------------
diff --git a/query_optimizer/tests/execution_generator/Copy.test b/query_optimizer/tests/execution_generator/Copy.test
index 4233c0f..be924df 100644
--- a/query_optimizer/tests/execution_generator/Copy.test
+++ b/query_optimizer/tests/execution_generator/Copy.test
@@ -52,24 +52,26 @@ cc 9
COPY
SELECT * FROM (
- SELECT -int_col, 'Negative' FROM source
+ SELECT -int_col * 1000, 'Negative' FROM source
UNION ALL
- SELECT int_col, 'Positive' FROM source
- ) AS t(value, sign) ORDER BY value
+ SELECT int_col * 1000, 'Positive' FROM source
+ ) AS t(VALUE, SIGN)
+ ORDER BY value
TO stderr WITH (FORMAT 'CSV', DELIMITER e'\t');
--
--6 Negative
--5 Negative
--4 Negative
--3 Negative
--2 Negative
--1 Negative
-1 Positive
-2 Positive
-3 Positive
-4 Positive
-5 Positive
-6 Positive
+VALUE SIGN
+-6000 Negative
+-5000 Negative
+-4000 Negative
+-3000 Negative
+-2000 Negative
+-1000 Negative
+1000 Positive
+2000 Positive
+3000 Positive
+4000 Positive
+5000 Positive
+6000 Positive
==
# Test handling of NULL values and special characters.
@@ -85,7 +87,7 @@ COPY source TO stderr;
\N \N b\nb """
==
-COPY source TO stderr WITH (FORMAT 'CSV');
+COPY source TO stderr WITH (FORMAT 'CSV', HEADER FALSE);
--
1,2000-01-01,abc,def
2,2000-02-02,a a,"|,|"
@@ -95,6 +97,7 @@ b",""""""""
COPY source TO stderr WITH (FORMAT 'CSV', DELIMITER '|', NULL_STRING '.na', HEADER TRUE);
--
+int_col|date_col|char_col|varchar_col
1|2000-01-01|abc|def
2|2000-02-02|a a|"|,|"
.na|.na|"b
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/c8214419/relational_operators/TableExportOperator.cpp
----------------------------------------------------------------------
diff --git a/relational_operators/TableExportOperator.cpp b/relational_operators/TableExportOperator.cpp
index 809e34c..45170c8 100644
--- a/relational_operators/TableExportOperator.cpp
+++ b/relational_operators/TableExportOperator.cpp
@@ -49,41 +49,42 @@ bool TableExportOperator::getAllWorkOrders(
StorageManager *storage_manager,
const tmb::client_id scheduler_client_id,
tmb::MessageBus *bus) {
- if (query_context_ == nullptr) {
- query_context_ = query_context;
- }
-
- const auto add_work_order = [&](const block_id input_block_id) -> void {
+ const auto add_work_order = [&](const block_id input_block_id,
+ const bool is_first_work_order) -> void {
std::unique_ptr<std::string> output_buffer = std::make_unique<std::string>();
container->addNormalWorkOrder(
- new TableExportWorkOrder(query_id_,
- input_relation_,
- input_block_id,
- options_->getFormat(),
- options_->getDelimiter(),
- options_->escapeStrings(),
- options_->getQuoteCharacter(),
- options_->getNullString(),
- op_index_,
- scheduler_client_id,
- storage_manager,
- bus,
- output_buffer.get()),
+ new TableExportToStringWorkOrder(query_id_,
+ input_relation_,
+ input_block_id,
+ options_->getFormat(),
+ is_first_work_order && options_->hasHeader(),
+ options_->getDelimiter(),
+ options_->escapeStrings(),
+ options_->getQuoteCharacter(),
+ options_->getNullString(),
+ op_index_,
+ scheduler_client_id,
+ storage_manager,
+ bus,
+ output_buffer.get()),
op_index_);
- query_context->setBlockOutputTextBuffer(input_block_id, output_buffer.release());
+
+ SpinMutexLock lock(output_buffers_mutex_);
+ output_buffers_.emplace(input_block_id, output_buffer.release());
};
if (input_relation_is_stored_) {
if (!started_) {
- for (const block_id input_block_id : input_relation_block_ids_) {
- add_work_order(input_block_id);
+ for (std::size_t i = 0; i < input_relation_block_ids_.size(); ++i) {
+ add_work_order(input_relation_block_ids_[i], i == 0);
}
started_ = true;
}
return true;
} else {
while (num_workorders_generated_ < input_relation_block_ids_.size()) {
- add_work_order(input_relation_block_ids_[num_workorders_generated_]);
+ add_work_order(input_relation_block_ids_[num_workorders_generated_],
+ num_workorders_generated_ == 0);
++num_workorders_generated_;
}
return done_feeding_input_relation_;
@@ -114,15 +115,15 @@ void TableExportOperator::receiveFeedbackMessage(
throw std::runtime_error("Can not open file " + file_name_ + " for writing");
}
}
-
- if (options_->hasHeader()) {
-
- }
}
- block_id completed_block_id = *static_cast<const block_id*>(msg.payload());
- outputs_.emplace(completed_block_id, std::unique_ptr<std::string>(
- query_context_->releaseBlockOutputTextBuffer(completed_block_id)));
+ // Mark block done.
+ const block_id done_block_id = *static_cast<const block_id*>(msg.payload());
+ {
+ SpinMutexLock lock(output_buffers_mutex_);
+ DCHECK(output_buffers_.find(done_block_id) != output_buffers_.end());
+ output_buffers_.at(done_block_id).done = true;
+ }
while (true) {
block_id next_block_id;
@@ -130,13 +131,18 @@ void TableExportOperator::receiveFeedbackMessage(
SpinMutexLock lock(block_ids_mutex_);
next_block_id = input_relation_block_ids_[num_blocks_written_];
}
- auto it = outputs_.find(next_block_id);
- if (it == outputs_.end()) {
- break;
+ std::unique_ptr<std::string> output_buffer;
+ {
+ SpinMutexLock lock(output_buffers_mutex_);
+ auto it = output_buffers_.find(next_block_id);
+ if (it == output_buffers_.end() || !it->second.done) {
+ break;
+ }
+ output_buffer = std::move(it->second.buffer);
+ output_buffers_.erase(it);
}
- std::fwrite(it->second->c_str(), 1, it->second->length(), file_);
+ std::fwrite(output_buffer->c_str(), 1, output_buffer->length(), file_);
++num_blocks_written_;
- outputs_.erase(it);
}
}
@@ -147,7 +153,7 @@ void TableExportOperator::updateCatalogOnCompletion() {
file_ = nullptr;
}
-void TableExportWorkOrder::execute() {
+void TableExportToStringWorkOrder::execute() {
BlockReference block(
storage_manager_->getBlock(input_block_id_, input_relation_));
std::unique_ptr<ValueAccessor> accessor(
@@ -155,11 +161,11 @@ void TableExportWorkOrder::execute() {
switch (format_) {
case BulkIOFormat::kCSV:
- writeToString<&TableExportWorkOrder::quoteCSVField>(
+ writeToString<&TableExportToStringWorkOrder::quoteCSVField>(
accessor.get(), output_buffer_);
break;
case BulkIOFormat::kText:
- writeToString<&TableExportWorkOrder::escapeTextField>(
+ writeToString<&TableExportToStringWorkOrder::escapeTextField>(
accessor.get(), output_buffer_);
break;
default:
@@ -176,7 +182,7 @@ void TableExportWorkOrder::execute() {
bus_, ClientIDMap::Instance()->getValue(), scheduler_client_id_, msg);
}
-inline std::string TableExportWorkOrder::quoteCSVField(std::string &&field) const {
+inline std::string TableExportToStringWorkOrder::quoteCSVField(std::string &&field) const {
bool need_quote = false;
for (const char c : field) {
if (c == column_delimiter_ || c == quote_character_ || c == '\n') {
@@ -201,7 +207,7 @@ inline std::string TableExportWorkOrder::quoteCSVField(std::string &&field) cons
}
-inline std::string TableExportWorkOrder::escapeTextField(std::string &&field) const {
+inline std::string TableExportToStringWorkOrder::escapeTextField(std::string &&field) const {
if (escape_strings_ == false || field == "\\N") {
return std::move(field);
}
@@ -263,11 +269,11 @@ inline std::string TableExportWorkOrder::escapeTextField(std::string &&field) co
return escaped;
}
-template <std::string (TableExportWorkOrder::*transform)(std::string&&) const,
+template <std::string (TableExportToStringWorkOrder::*transform)(std::string&&) const,
typename Container, typename Functor>
-inline void TableExportWorkOrder::writeEachToString(const Container &container,
- std::string *output,
- const Functor &functor) const {
+inline void TableExportToStringWorkOrder::writeEachToString(const Container &container,
+ std::string *output,
+ const Functor &functor) const {
auto it = container.begin();
if (it != container.end()) {
std::size_t idx = 0;
@@ -279,15 +285,24 @@ inline void TableExportWorkOrder::writeEachToString(const Container &container,
}
}
-template <std::string (TableExportWorkOrder::*transform)(std::string&&) const>
-void TableExportWorkOrder::writeToString(
- ValueAccessor *accessor, std::string *output) const {
+template <std::string (TableExportToStringWorkOrder::*transform)(std::string&&) const>
+void TableExportToStringWorkOrder::writeToString(ValueAccessor *accessor,
+ std::string *output) const {
std::vector<const Type*> value_types;
value_types.reserve(input_relation_.size());
for (const CatalogAttribute &attribute : input_relation_) {
value_types.emplace_back(&attribute.getType());
}
+ if (print_header_) {
+ writeEachToString<transform>(
+ input_relation_, output,
+ [&](const CatalogAttribute &attr, const std::size_t idx) -> std::string {
+ return attr.getDisplayName();
+ });
+ output->push_back('\n');
+ }
+
accessor->beginIterationVirtual();
while (accessor->nextVirtual()) {
std::unique_ptr<Tuple> tuple(accessor->getTupleVirtual());
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/c8214419/relational_operators/TableExportOperator.hpp
----------------------------------------------------------------------
diff --git a/relational_operators/TableExportOperator.hpp b/relational_operators/TableExportOperator.hpp
index a8152c8..159b28c 100644
--- a/relational_operators/TableExportOperator.hpp
+++ b/relational_operators/TableExportOperator.hpp
@@ -22,8 +22,10 @@
#include <cstddef>
#include <cstdio>
+#include <memory>
#include <string>
#include <unordered_map>
+#include <utility>
#include <vector>
#include "catalog/CatalogRelation.hpp"
@@ -78,7 +80,6 @@ class TableExportOperator : public RelationalOperator {
: std::vector<block_id>()),
num_workorders_generated_(0),
started_(false),
- query_context_(nullptr),
num_blocks_written_(0),
file_(nullptr) {}
@@ -118,6 +119,15 @@ class TableExportOperator : public RelationalOperator {
void updateCatalogOnCompletion() override;
private:
+ // Buffer for storing a block's exported string.
+ struct BlockBuffer {
+ BlockBuffer(std::string *buffer_in)
+ : done(false),
+ buffer(buffer_in) {}
+ bool done;
+ std::unique_ptr<std::string> buffer;
+ };
+
const CatalogRelation &input_relation_;
const bool input_relation_is_stored_;
const std::string file_name_;
@@ -125,38 +135,40 @@ class TableExportOperator : public RelationalOperator {
std::vector<block_id> input_relation_block_ids_;
std::size_t num_workorders_generated_;
+ SpinMutex block_ids_mutex_;
bool started_;
- QueryContext *query_context_;
std::size_t num_blocks_written_;
- std::unordered_map<block_id, std::unique_ptr<std::string>> outputs_;
- SpinMutex block_ids_mutex_;
+ std::unordered_map<block_id, BlockBuffer> output_buffers_;
+ SpinMutex output_buffers_mutex_;
FILE *file_;
DISALLOW_COPY_AND_ASSIGN(TableExportOperator);
};
-class TableExportWorkOrder : public WorkOrder {
+class TableExportToStringWorkOrder : public WorkOrder {
public:
- TableExportWorkOrder(const std::size_t query_id,
- const CatalogRelationSchema &input_relation,
- const block_id input_block_id,
- const BulkIOFormat format,
- const char column_delimiter,
- const bool escape_strings,
- const char quote_character,
- const std::string null_string,
- const std::size_t operator_index,
- const tmb::client_id scheduler_client_id,
- StorageManager *storage_manager,
- MessageBus *bus,
- std::string *output_buffer)
+ TableExportToStringWorkOrder(const std::size_t query_id,
+ const CatalogRelationSchema &input_relation,
+ const block_id input_block_id,
+ const BulkIOFormat format,
+ const bool print_header,
+ const char column_delimiter,
+ const bool escape_strings,
+ const char quote_character,
+ const std::string null_string,
+ const std::size_t operator_index,
+ const tmb::client_id scheduler_client_id,
+ StorageManager *storage_manager,
+ MessageBus *bus,
+ std::string *output_buffer)
: WorkOrder(query_id),
input_relation_(input_relation),
input_block_id_(input_block_id),
format_(format),
+ print_header_(print_header),
column_delimiter_(column_delimiter),
escape_strings_(escape_strings),
quote_character_(quote_character),
@@ -168,7 +180,7 @@ class TableExportWorkOrder : public WorkOrder {
output_buffer_(output_buffer) {
}
- ~TableExportWorkOrder() override {}
+ ~TableExportToStringWorkOrder() override {}
void execute() override;
@@ -176,19 +188,20 @@ class TableExportWorkOrder : public WorkOrder {
inline std::string quoteCSVField(std::string &&field) const;
inline std::string escapeTextField(std::string &&field) const;
- template <std::string (TableExportWorkOrder::*transform)(std::string&&) const,
+ template <std::string (TableExportToStringWorkOrder::*transform)(std::string&&) const,
typename Container, typename Functor>
inline void writeEachToString(const Container &container,
std::string *output,
const Functor &functor) const;
- template <std::string (TableExportWorkOrder::*transform)(std::string&&) const>
+ template <std::string (TableExportToStringWorkOrder::*transform)(std::string&&) const>
void writeToString(ValueAccessor *accessor, std::string *output) const;
const CatalogRelationSchema &input_relation_;
const block_id input_block_id_;
const BulkIOFormat format_;
+ const bool print_header_;
const char column_delimiter_;
const bool escape_strings_;
const char quote_character_;
@@ -201,7 +214,39 @@ class TableExportWorkOrder : public WorkOrder {
std::string *output_buffer_;
- DISALLOW_COPY_AND_ASSIGN(TableExportWorkOrder);
+ DISALLOW_COPY_AND_ASSIGN(TableExportToStringWorkOrder);
+};
+
+class TableStringSaveToFileWorkOrder : public WorkOrder {
+ public:
+ TableStringSaveToFileWorkOrder(const std::size_t query_id,
+ FILE *file,
+ std::vector<std::string> &&output,
+ const std::size_t operator_index,
+ const tmb::client_id scheduler_client_id,
+ StorageManager *storage_manager,
+ MessageBus *bus)
+ : WorkOrder(query_id),
+ file_(file),
+ output_(std::move(output)),
+ operator_index_(operator_index),
+ scheduler_client_id_(scheduler_client_id),
+ storage_manager_(storage_manager),
+ bus_(bus) {
+ }
+
+ void execute() override;
+
+ private:
+ FILE *file_;
+ const std::vector<std::string> output_;
+
+ const std::size_t operator_index_;
+ const tmb::client_id scheduler_client_id_;
+ StorageManager *storage_manager_;
+ MessageBus *bus_;
+
+ DISALLOW_COPY_AND_ASSIGN(TableStringSaveToFileWorkOrder);
};
/** @} */