http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/1be47dcb/relational_operators/NestedLoopsJoinOperator.hpp ---------------------------------------------------------------------- diff --git a/relational_operators/NestedLoopsJoinOperator.hpp b/relational_operators/NestedLoopsJoinOperator.hpp index f165442..0b13842 100644 --- a/relational_operators/NestedLoopsJoinOperator.hpp +++ b/relational_operators/NestedLoopsJoinOperator.hpp @@ -59,6 +59,7 @@ class NestedLoopsJoinOperator : public RelationalOperator { /** * @brief Constructor. * + * @param query_id The ID of the query to which this operator belongs. * @param left_input_relation The first relation in the join (order is not * actually important). * @param right_input_relation The second relation in the join (order is not @@ -76,17 +77,17 @@ class NestedLoopsJoinOperator : public RelationalOperator { * @param left_relation_is_stored If left_input_relation is a stored relation. * @param right_relation_is_stored If right_input_relation is a stored * relation. - * @param query_id The ID of the query to which this operator belongs. **/ - NestedLoopsJoinOperator(const CatalogRelation &left_input_relation, - const CatalogRelation &right_input_relation, - const CatalogRelation &output_relation, - const QueryContext::insert_destination_id output_destination_index, - const QueryContext::predicate_id join_predicate_index, - const QueryContext::scalar_group_id selection_index, - bool left_relation_is_stored, - bool right_relation_is_stored, - const std::size_t query_id) + NestedLoopsJoinOperator( + const std::size_t query_id, + const CatalogRelation &left_input_relation, + const CatalogRelation &right_input_relation, + const CatalogRelation &output_relation, + const QueryContext::insert_destination_id output_destination_index, + const QueryContext::predicate_id join_predicate_index, + const QueryContext::scalar_group_id selection_index, + bool left_relation_is_stored, + bool right_relation_is_stored) : RelationalOperator(query_id), left_input_relation_(left_input_relation), right_input_relation_(right_input_relation), @@ -96,10 +97,12 @@ class NestedLoopsJoinOperator : public RelationalOperator { selection_index_(selection_index), left_relation_is_stored_(left_relation_is_stored), right_relation_is_stored_(right_relation_is_stored), - left_relation_block_ids_(left_relation_is_stored ? left_input_relation.getBlocksSnapshot() - : std::vector<block_id>()), - right_relation_block_ids_(right_relation_is_stored ? right_input_relation.getBlocksSnapshot() - : std::vector<block_id>()), + left_relation_block_ids_(left_relation_is_stored + ? left_input_relation.getBlocksSnapshot() + : std::vector<block_id>()), + right_relation_block_ids_(right_relation_is_stored + ? right_input_relation.getBlocksSnapshot() + : std::vector<block_id>()), num_left_workorders_generated_(0), num_right_workorders_generated_(0), done_feeding_left_relation_(false), @@ -222,6 +225,7 @@ class NestedLoopsJoinWorkOrder : public WorkOrder { /** * @brief Constructor. * + * @param query_id The ID of the query to which this operator belongs. * @param left_input_relation The first relation in the join (order is not * actually important). * @param right_input_relation The second relation in the join (order is not @@ -233,19 +237,19 @@ class NestedLoopsJoinWorkOrder : public WorkOrder { * @param selection A list of Scalars corresponding to the relation attributes * in \c output_destination. Each Scalar is evaluated for the joined * tuples, and the resulting value is inserted into the join result. - * @param query_id The ID of the query to which this operator belongs. * @param output_destination The InsertDestination to insert the join results. * @param storage_manager The StorageManager to use. **/ - NestedLoopsJoinWorkOrder(const CatalogRelationSchema &left_input_relation, - const CatalogRelationSchema &right_input_relation, - const block_id left_block_id, - const block_id right_block_id, - const Predicate *join_predicate, - const std::vector<std::unique_ptr<const Scalar>> &selection, - const std::size_t query_id, - InsertDestination *output_destination, - StorageManager *storage_manager) + NestedLoopsJoinWorkOrder( + const std::size_t query_id, + const CatalogRelationSchema &left_input_relation, + const CatalogRelationSchema &right_input_relation, + const block_id left_block_id, + const block_id right_block_id, + const Predicate *join_predicate, + const std::vector<std::unique_ptr<const Scalar>> &selection, + InsertDestination *output_destination, + StorageManager *storage_manager) : WorkOrder(query_id), left_input_relation_(left_input_relation), right_input_relation_(right_input_relation),
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/1be47dcb/relational_operators/RebuildWorkOrder.hpp ---------------------------------------------------------------------- diff --git a/relational_operators/RebuildWorkOrder.hpp b/relational_operators/RebuildWorkOrder.hpp index ae876ba..86f8eaf 100644 --- a/relational_operators/RebuildWorkOrder.hpp +++ b/relational_operators/RebuildWorkOrder.hpp @@ -48,6 +48,7 @@ class RebuildWorkOrder : public WorkOrder { /** * @brief Constructor. * + * @param query_id The ID of the query to which this RebuildWorkOrder belongs. * @param block_ref A MutableBlockReference to the storage block which needs * to be rebuilt. * @param input_operator_index The index of the relational operator in the @@ -55,15 +56,15 @@ class RebuildWorkOrder : public WorkOrder { * @param input_relation_id The ID of the CatalogRelation to which the given * storage block belongs to. * @param scheduler_client_id The TMB client ID of the scheduler thread. - * @param query_id The ID of the query to which this RebuildWorkOrder belongs. * @param bus A pointer to the TMB. **/ - RebuildWorkOrder(MutableBlockReference &&block_ref, - const std::size_t input_operator_index, - const relation_id input_relation_id, - const client_id scheduler_client_id, - const std::size_t query_id, - MessageBus *bus) + RebuildWorkOrder( + const std::size_t query_id, + MutableBlockReference &&block_ref, // NOLINT(whitespace/operators) + const std::size_t input_operator_index, + const relation_id input_relation_id, + const client_id scheduler_client_id, + MessageBus *bus) : WorkOrder(query_id), block_ref_(std::move(block_ref)), input_operator_index_(input_operator_index), http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/1be47dcb/relational_operators/RelationalOperator.hpp ---------------------------------------------------------------------- diff --git a/relational_operators/RelationalOperator.hpp b/relational_operators/RelationalOperator.hpp index aa93018..c173a0a 100644 --- a/relational_operators/RelationalOperator.hpp +++ b/relational_operators/RelationalOperator.hpp @@ -212,16 +212,17 @@ class RelationalOperator { * @param blocking_dependencies_met If those dependencies which break the * pipeline have been met. **/ - explicit RelationalOperator(const std::size_t query_id = 0, + explicit RelationalOperator(const std::size_t query_id, const bool blocking_dependencies_met = false) - : blocking_dependencies_met_(blocking_dependencies_met), - done_feeding_input_relation_(false), - query_id_(query_id) {} + : query_id_(query_id), + blocking_dependencies_met_(blocking_dependencies_met), + done_feeding_input_relation_(false) {} + + const std::size_t query_id_; bool blocking_dependencies_met_; bool done_feeding_input_relation_; std::size_t op_index_; - const std::size_t query_id_; private: DISALLOW_COPY_AND_ASSIGN(RelationalOperator); http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/1be47dcb/relational_operators/SampleOperator.cpp ---------------------------------------------------------------------- diff --git a/relational_operators/SampleOperator.cpp b/relational_operators/SampleOperator.cpp index b318ce4..8d5fade 100644 --- a/relational_operators/SampleOperator.cpp +++ b/relational_operators/SampleOperator.cpp @@ -54,11 +54,11 @@ bool SampleOperator::getAllWorkOrders( for (const block_id input_block_id : input_relation_block_ids_) { if (distribution(generator) <= probability) { container->addNormalWorkOrder( - new SampleWorkOrder(input_relation_, + new SampleWorkOrder(query_id_, + input_relation_, input_block_id, is_block_sample_, percentage_, - query_id_, output_destination, storage_manager), op_index_); @@ -68,15 +68,14 @@ bool SampleOperator::getAllWorkOrders( // Add all the blocks for tuple sampling which would handle // the sampling from each block for (const block_id input_block_id : input_relation_block_ids_) { - container->addNormalWorkOrder( - new SampleWorkOrder(input_relation_, - input_block_id, - is_block_sample_, - percentage_, - query_id_, - output_destination, - storage_manager), - op_index_); + container->addNormalWorkOrder(new SampleWorkOrder(query_id_, + input_relation_, + input_block_id, + is_block_sample_, + percentage_, + output_destination, + storage_manager), + op_index_); } } started_ = true; @@ -87,13 +86,14 @@ bool SampleOperator::getAllWorkOrders( while (num_workorders_generated_ < input_relation_block_ids_.size()) { if (distribution(generator) <= probability) { container->addNormalWorkOrder( - new SampleWorkOrder(input_relation_, - input_relation_block_ids_[num_workorders_generated_], - is_block_sample_, - percentage_, - query_id_, - output_destination, - storage_manager), + new SampleWorkOrder( + query_id_, + input_relation_, + input_relation_block_ids_[num_workorders_generated_], + is_block_sample_, + percentage_, + output_destination, + storage_manager), op_index_); ++num_workorders_generated_; } @@ -101,13 +101,14 @@ bool SampleOperator::getAllWorkOrders( } else { while (num_workorders_generated_ < input_relation_block_ids_.size()) { container->addNormalWorkOrder( - new SampleWorkOrder(input_relation_, - input_relation_block_ids_[num_workorders_generated_], - is_block_sample_, - percentage_, - query_id_, - output_destination, - storage_manager), + new SampleWorkOrder( + query_id_, + input_relation_, + input_relation_block_ids_[num_workorders_generated_], + is_block_sample_, + percentage_, + output_destination, + storage_manager), op_index_); ++num_workorders_generated_; } @@ -115,6 +116,7 @@ bool SampleOperator::getAllWorkOrders( return done_feeding_input_relation_; } } + void SampleWorkOrder::execute() { BlockReference block( storage_manager_->getBlock(input_block_id_, input_relation_)); http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/1be47dcb/relational_operators/SampleOperator.hpp ---------------------------------------------------------------------- diff --git a/relational_operators/SampleOperator.hpp b/relational_operators/SampleOperator.hpp index f65f28a..505daa2 100644 --- a/relational_operators/SampleOperator.hpp +++ b/relational_operators/SampleOperator.hpp @@ -54,8 +54,9 @@ class WorkOrdersContainer; class SampleOperator : public RelationalOperator { public: /** - * @brief Constructor for SampleOperator with the sampling percentage and type of sampling. + * @brief Constructor for SampleOperator with the sampling percentage and type of sampling. * + * @param query_id The ID of the query to which this operator belongs. * @param input_relation The relation to perform sampling over. * @param output_relation The output relation. * @param output_destination_index The index of the InsertDestination in the @@ -65,15 +66,15 @@ class SampleOperator : public RelationalOperator { * workorders. * @param is_block_sample Flag indicating whether the sample type is block or tuple. * @param percentage The percentage of data to be sampled. - * @param query_id The ID of the query to which this operator belongs. **/ - SampleOperator(const CatalogRelation &input_relation, - const CatalogRelationSchema &output_relation, - const QueryContext::insert_destination_id output_destination_index, - const bool input_relation_is_stored, - const bool is_block_sample, - const int percentage, - const std::size_t query_id) + SampleOperator( + const std::size_t query_id, + const CatalogRelation &input_relation, + const CatalogRelationSchema &output_relation, + const QueryContext::insert_destination_id output_destination_index, + const bool input_relation_is_stored, + const bool is_block_sample, + const int percentage) : RelationalOperator(query_id), input_relation_(input_relation), output_relation_(output_relation), @@ -81,8 +82,9 @@ class SampleOperator : public RelationalOperator { input_relation_is_stored_(input_relation_is_stored), is_block_sample_(is_block_sample), percentage_(percentage), - input_relation_block_ids_(input_relation_is_stored ? input_relation.getBlocksSnapshot() - : std::vector<block_id>()), + input_relation_block_ids_(input_relation_is_stored + ? input_relation.getBlocksSnapshot() + : std::vector<block_id>()), num_workorders_generated_(0), started_(false) {} @@ -133,11 +135,22 @@ class SampleOperator : public RelationalOperator { **/ class SampleWorkOrder : public WorkOrder { public: - SampleWorkOrder(const CatalogRelationSchema &input_relation, + /** + * @brief Constructor. + * + * @param query_id The ID of the query to which this WorkOrder belongs. + * @param input_relation The relation to perform sampling over. + * @param input_block_id The block to sample. + * @param is_block_sample Flag indicating whether the sample type is block or tuple. + * @param percentage The percentage of data to be sampled. + * @param output_destination The InsertDestination to insert the sample results. + * @param storage_manager The StorageManager to use. + **/ + SampleWorkOrder(const std::size_t query_id, + const CatalogRelationSchema &input_relation, const block_id input_block_id, const bool is_block_sample, const int percentage, - const std::size_t query_id, InsertDestination *output_destination, StorageManager *storage_manager) : WorkOrder(query_id), http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/1be47dcb/relational_operators/SaveBlocksOperator.cpp ---------------------------------------------------------------------- diff --git a/relational_operators/SaveBlocksOperator.cpp b/relational_operators/SaveBlocksOperator.cpp index 3581090..8127d88 100644 --- a/relational_operators/SaveBlocksOperator.cpp +++ b/relational_operators/SaveBlocksOperator.cpp @@ -36,9 +36,9 @@ bool SaveBlocksOperator::getAllWorkOrders( while (num_workorders_generated_ < destination_block_ids_.size()) { container->addNormalWorkOrder( new SaveBlocksWorkOrder( + query_id_, destination_block_ids_[num_workorders_generated_], force_, - query_id_, storage_manager), op_index_); ++num_workorders_generated_; http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/1be47dcb/relational_operators/SaveBlocksOperator.hpp ---------------------------------------------------------------------- diff --git a/relational_operators/SaveBlocksOperator.hpp b/relational_operators/SaveBlocksOperator.hpp index f4650bb..6e2c72b 100644 --- a/relational_operators/SaveBlocksOperator.hpp +++ b/relational_operators/SaveBlocksOperator.hpp @@ -55,7 +55,8 @@ class SaveBlocksOperator : public RelationalOperator { * @param force If true, force writing of all blocks to disk, otherwise only * write dirty blocks. **/ - explicit SaveBlocksOperator(const std::size_t query_id, bool force = false) + explicit SaveBlocksOperator(const std::size_t query_id, + const bool force = false) : RelationalOperator(query_id), force_(force), num_workorders_generated_(0) {} @@ -96,15 +97,15 @@ class SaveBlocksWorkOrder : public WorkOrder { /** * @brief Constructor. * + * @param query_id The ID of the query to which this operator belongs. * @param save_block_id The id of the block to save. * @param force If true, force writing of all blocks to disk, otherwise only * write dirty blocks. - * @param query_id The ID of the query to which this operator belongs. * @param storage_manager The StorageManager to use. **/ - SaveBlocksWorkOrder(const block_id save_block_id, + SaveBlocksWorkOrder(const std::size_t query_id, + const block_id save_block_id, const bool force, - const std::size_t query_id, StorageManager *storage_manager) : WorkOrder(query_id), save_block_id_(save_block_id), http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/1be47dcb/relational_operators/SelectOperator.cpp ---------------------------------------------------------------------- diff --git a/relational_operators/SelectOperator.cpp b/relational_operators/SelectOperator.cpp index 350890d..e9a96f3 100644 --- a/relational_operators/SelectOperator.cpp +++ b/relational_operators/SelectOperator.cpp @@ -42,29 +42,28 @@ void SelectOperator::addWorkOrders(WorkOrdersContainer *container, InsertDestination *output_destination) { if (input_relation_is_stored_) { for (const block_id input_block_id : input_relation_block_ids_) { - container->addNormalWorkOrder( - new SelectWorkOrder(input_relation_, - input_block_id, - predicate, - simple_projection_, - simple_selection_, - selection, - query_id_, - output_destination, - storage_manager), - op_index_); + container->addNormalWorkOrder(new SelectWorkOrder(query_id_, + input_relation_, + input_block_id, + predicate, + simple_projection_, + simple_selection_, + selection, + output_destination, + storage_manager), + op_index_); } } else { while (num_workorders_generated_ < input_relation_block_ids_.size()) { container->addNormalWorkOrder( new SelectWorkOrder( + query_id_, input_relation_, input_relation_block_ids_[num_workorders_generated_], predicate, simple_projection_, simple_selection_, selection, - query_id_, output_destination, storage_manager), op_index_); @@ -87,13 +86,13 @@ void SelectOperator::addPartitionAwareWorkOrders(WorkOrdersContainer *container, input_relation_block_ids_in_partition_[part_id]) { container->addNormalWorkOrder( new SelectWorkOrder( + query_id_, input_relation_, input_block_id, predicate, simple_projection_, simple_selection_, selection, - query_id_, output_destination, storage_manager, placement_scheme_->getNUMANodeForBlock(input_block_id)), @@ -108,13 +107,13 @@ void SelectOperator::addPartitionAwareWorkOrders(WorkOrdersContainer *container, = input_relation_block_ids_in_partition_[part_id][num_workorders_generated_in_partition_[part_id]]; container->addNormalWorkOrder( new SelectWorkOrder( + query_id_, input_relation_, block_in_partition, predicate, simple_projection_, simple_selection_, selection, - query_id_, output_destination, storage_manager, placement_scheme_->getNUMANodeForBlock(block_in_partition)), http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/1be47dcb/relational_operators/SelectOperator.hpp ---------------------------------------------------------------------- diff --git a/relational_operators/SelectOperator.hpp b/relational_operators/SelectOperator.hpp index 4f5b8ca..ac7b038 100644 --- a/relational_operators/SelectOperator.hpp +++ b/relational_operators/SelectOperator.hpp @@ -64,6 +64,7 @@ class SelectOperator : public RelationalOperator { * @brief Constructor for selection with arbitrary expressions in projection * list. * + * @param query_id The ID of the query to which this operator belongs. * @param input_relation The relation to perform selection over. * @param output_relation The output relation. * @param output_destination_index The index of the InsertDestination in the @@ -76,23 +77,24 @@ class SelectOperator : public RelationalOperator { * @param input_relation_is_stored If input_relation is a stored relation and * is fully available to the operator before it can start generating * workorders. - * @param query_id The ID of the query to which this operator belongs. **/ - SelectOperator(const CatalogRelation &input_relation, - const CatalogRelation &output_relation, - const QueryContext::insert_destination_id output_destination_index, - const QueryContext::predicate_id predicate_index, - const QueryContext::scalar_group_id selection_index, - const bool input_relation_is_stored, - const std::size_t query_id) + SelectOperator( + const std::size_t query_id, + const CatalogRelation &input_relation, + const CatalogRelation &output_relation, + const QueryContext::insert_destination_id output_destination_index, + const QueryContext::predicate_id predicate_index, + const QueryContext::scalar_group_id selection_index, + const bool input_relation_is_stored) : RelationalOperator(query_id), input_relation_(input_relation), output_relation_(output_relation), output_destination_index_(output_destination_index), predicate_index_(predicate_index), selection_index_(selection_index), - input_relation_block_ids_(input_relation_is_stored ? input_relation.getBlocksSnapshot() - : std::vector<block_id>()), + input_relation_block_ids_(input_relation_is_stored + ? input_relation.getBlocksSnapshot() + : std::vector<block_id>()), num_workorders_generated_(0), simple_projection_(false), input_relation_is_stored_(input_relation_is_stored), @@ -124,6 +126,7 @@ class SelectOperator : public RelationalOperator { * * @note selection_index_ is invalid, and will not be used for projection. * + * @param query_id The ID of the query to which this operator belongs. * @param input_relation The relation to perform selection over. * @param output_relation The output relation. * @param output_destination_index The index of the InsertDestination in the @@ -136,15 +139,15 @@ class SelectOperator : public RelationalOperator { * @param input_relation_is_stored If input_relation is a stored relation and * is fully available to the operator before it can start generating * workorders. - * @param query_id The ID of the query to which this operator belongs. **/ - SelectOperator(const CatalogRelation &input_relation, - const CatalogRelation &output_relation, - const QueryContext::insert_destination_id output_destination_index, - const QueryContext::predicate_id predicate_index, - std::vector<attribute_id> &&selection, - const bool input_relation_is_stored, - const std::size_t query_id) + SelectOperator( + const std::size_t query_id, + const CatalogRelation &input_relation, + const CatalogRelation &output_relation, + const QueryContext::insert_destination_id output_destination_index, + const QueryContext::predicate_id predicate_index, + std::vector<attribute_id> &&selection, + const bool input_relation_is_stored) : RelationalOperator(query_id), input_relation_(input_relation), output_relation_(output_relation), @@ -152,8 +155,9 @@ class SelectOperator : public RelationalOperator { predicate_index_(predicate_index), selection_index_(QueryContext::kInvalidScalarGroupId), simple_selection_(std::move(selection)), - input_relation_block_ids_(input_relation_is_stored ? input_relation.getBlocksSnapshot() - : std::vector<block_id>()), + input_relation_block_ids_(input_relation_is_stored + ? input_relation.getBlocksSnapshot() + : std::vector<block_id>()), num_workorders_generated_(0), simple_projection_(true), input_relation_is_stored_(input_relation_is_stored), @@ -278,6 +282,7 @@ class SelectWorkOrder : public WorkOrder { * @note Reference parameter selection is NOT owned by this class and must * remain valid until after execute() is called. * + * @param query_id The ID of the query to which this WorkOrder belongs. * @param input_relation The relation to perform selection over. * @param input_block_id The block id. * @param predicate All tuples matching \c predicate will be selected (or NULL @@ -287,18 +292,17 @@ class SelectWorkOrder : public WorkOrder { * simple_projection is true. * @param selection A list of Scalars which will be evaluated to project * input tuples, used if \c simple_projection is false. - * @param query_id The ID of the query to which this WorkOrder belongs. * @param output_destination The InsertDestination to insert the selection * results. * @param storage_manager The StorageManager to use. **/ - SelectWorkOrder(const CatalogRelationSchema &input_relation, + SelectWorkOrder(const std::size_t query_id, + const CatalogRelationSchema &input_relation, const block_id input_block_id, const Predicate *predicate, const bool simple_projection, const std::vector<attribute_id> &simple_selection, const std::vector<std::unique_ptr<const Scalar>> *selection, - const std::size_t query_id, InsertDestination *output_destination, StorageManager *storage_manager, const numa_node_id numa_node = 0) @@ -320,6 +324,7 @@ class SelectWorkOrder : public WorkOrder { * @note Reference parameter selection is NOT owned by this class and must * remain valid until after execute() is called. * + * @param query_id The ID of the query to which this WorkOrder belongs. * @param input_relation The relation to perform selection over. * @param input_block_id The block id. * @param predicate All tuples matching \c predicate will be selected (or NULL @@ -329,18 +334,17 @@ class SelectWorkOrder : public WorkOrder { * simple_projection is true. * @param selection A list of Scalars which will be evaluated to project * input tuples, used if \c simple_projection is false. - * @param query_id The ID of the query to which this WorkOrder belongs. * @param output_destination The InsertDestination to insert the selection * results. * @param storage_manager The StorageManager to use. **/ - SelectWorkOrder(const CatalogRelationSchema &input_relation, + SelectWorkOrder(const std::size_t query_id, + const CatalogRelationSchema &input_relation, const block_id input_block_id, const Predicate *predicate, const bool simple_projection, std::vector<attribute_id> &&simple_selection, const std::vector<std::unique_ptr<const Scalar>> *selection, - const std::size_t query_id, InsertDestination *output_destination, StorageManager *storage_manager, const numa_node_id numa_node = 0) http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/1be47dcb/relational_operators/SortMergeRunOperator.cpp ---------------------------------------------------------------------- diff --git a/relational_operators/SortMergeRunOperator.cpp b/relational_operators/SortMergeRunOperator.cpp index 9db8de1..6bf5719 100644 --- a/relational_operators/SortMergeRunOperator.cpp +++ b/relational_operators/SortMergeRunOperator.cpp @@ -85,12 +85,12 @@ WorkOrder *SortMergeRunOperator::createWorkOrder( // Create a work order from the merge job from merge tree. return new SortMergeRunWorkOrder( + query_id_, query_context->getSortConfig(sort_config_index_), job->level > 0 ? run_relation_ : input_relation_, std::move(job->runs), top_k_, job->level, - query_id_, output_destination, storage_manager, op_index_, http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/1be47dcb/relational_operators/SortMergeRunOperator.hpp ---------------------------------------------------------------------- diff --git a/relational_operators/SortMergeRunOperator.hpp b/relational_operators/SortMergeRunOperator.hpp index f54e925..cfff8b9 100644 --- a/relational_operators/SortMergeRunOperator.hpp +++ b/relational_operators/SortMergeRunOperator.hpp @@ -72,6 +72,7 @@ class SortMergeRunOperator : public RelationalOperator { /** * @brief Constructor for merging sorted runs to generate a sorted relation. * + * @param query_id The ID of the query to which this operator belongs. * @param input_relation The relation to merge sorted blocks. * @param output_relation The output relation. * @param output_destination_index The index of the InsertDestination in the @@ -88,18 +89,18 @@ class SortMergeRunOperator : public RelationalOperator { * \c top_k is 0. * @param input_relation_is_stored Boolean to indicate is input relation is * stored or streamed. - * @param query_id The ID of the query to which this operator belongs. **/ - SortMergeRunOperator(const CatalogRelation &input_relation, - const CatalogRelation &output_relation, - const QueryContext::insert_destination_id output_destination_index, - const CatalogRelation &run_relation, - const QueryContext::insert_destination_id run_block_destination_index, - const QueryContext::sort_config_id sort_config_index, - const std::size_t merge_factor, - const std::size_t top_k, - const bool input_relation_is_stored, - const std::size_t query_id) + SortMergeRunOperator( + const std::size_t query_id, + const CatalogRelation &input_relation, + const CatalogRelation &output_relation, + const QueryContext::insert_destination_id output_destination_index, + const CatalogRelation &run_relation, + const QueryContext::insert_destination_id run_block_destination_index, + const QueryContext::sort_config_id sort_config_index, + const std::size_t merge_factor, + const std::size_t top_k, + const bool input_relation_is_stored) : RelationalOperator(query_id), input_relation_(input_relation), output_relation_(output_relation), @@ -214,12 +215,12 @@ class SortMergeRunWorkOrder : public WorkOrder { /** * @brief Constructor. * + * @param query_id The ID of the query to which this WorkOrder belongs. * @param sort_config The Sort configuration. * @param run_relation The relation to which the run blocks belong to. * @param input_runs Input runs to merge. * @param top_k If non-zero will merge only \c top_k tuples. * @param merge_level Merge level in the merge tree. - * @param query_id The ID of the query to which this WorkOrder belongs. * @param output_destination The InsertDestination to create new blocks. * @param storage_manager The StorageManager to use. * @param operator_index Merge-run operator index to send feedback messages @@ -228,12 +229,12 @@ class SortMergeRunWorkOrder : public WorkOrder { * @param bus TMB to send the feedback message on. **/ SortMergeRunWorkOrder( + const std::size_t query_id, const SortConfiguration &sort_config, const CatalogRelationSchema &run_relation, std::vector<merge_run_operator::Run> &&input_runs, const std::size_t top_k, const std::size_t merge_level, - const std::size_t query_id, InsertDestination *output_destination, StorageManager *storage_manager, const std::size_t operator_index, http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/1be47dcb/relational_operators/SortRunGenerationOperator.cpp ---------------------------------------------------------------------- diff --git a/relational_operators/SortRunGenerationOperator.cpp b/relational_operators/SortRunGenerationOperator.cpp index e352f9e..37b8fb8 100644 --- a/relational_operators/SortRunGenerationOperator.cpp +++ b/relational_operators/SortRunGenerationOperator.cpp @@ -51,10 +51,10 @@ bool SortRunGenerationOperator::getAllWorkOrders( if (!started_) { for (const block_id input_block_id : input_relation_block_ids_) { container->addNormalWorkOrder( - new SortRunGenerationWorkOrder(input_relation_, + new SortRunGenerationWorkOrder(query_id_, + input_relation_, input_block_id, sort_config, - query_id_, output_destination, storage_manager), op_index_); @@ -67,10 +67,10 @@ bool SortRunGenerationOperator::getAllWorkOrders( while (num_workorders_generated_ < input_relation_block_ids_.size()) { container->addNormalWorkOrder( new SortRunGenerationWorkOrder( + query_id_, input_relation_, input_relation_block_ids_[num_workorders_generated_], sort_config, - query_id_, output_destination, storage_manager), op_index_); http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/1be47dcb/relational_operators/SortRunGenerationOperator.hpp ---------------------------------------------------------------------- diff --git a/relational_operators/SortRunGenerationOperator.hpp b/relational_operators/SortRunGenerationOperator.hpp index 3da9813..f96e6a6 100644 --- a/relational_operators/SortRunGenerationOperator.hpp +++ b/relational_operators/SortRunGenerationOperator.hpp @@ -73,6 +73,7 @@ class SortRunGenerationOperator : public RelationalOperator { * @brief Constructor for sorting tuples in blocks based on the sort * configuration and writing to output destination. * + * @param query_id The ID of the query to which this operator belongs. * @param input_relation The relation to generate sorted runs of. * @param output_relation The output relation. * @param output_destination_index The index of the InsertDestination in the @@ -83,21 +84,22 @@ class SortRunGenerationOperator : public RelationalOperator { * @param input_relation_is_stored Does the input relation contain the blocks * to sort. If \c false, the blocks are * streamed. - * @param query_id The ID of the query to which this operator belongs. **/ - SortRunGenerationOperator(const CatalogRelation &input_relation, - const CatalogRelation &output_relation, - const QueryContext::insert_destination_id output_destination_index, - const QueryContext::sort_config_id sort_config_index, - bool input_relation_is_stored, - const std::size_t query_id) + SortRunGenerationOperator( + const std::size_t query_id, + const CatalogRelation &input_relation, + const CatalogRelation &output_relation, + const QueryContext::insert_destination_id output_destination_index, + const QueryContext::sort_config_id sort_config_index, + bool input_relation_is_stored) : RelationalOperator(query_id), input_relation_(input_relation), output_relation_(output_relation), output_destination_index_(output_destination_index), sort_config_index_(sort_config_index), - input_relation_block_ids_(input_relation_is_stored ? input_relation.getBlocksSnapshot() - : std::vector<block_id>()), + input_relation_block_ids_(input_relation_is_stored + ? input_relation.getBlocksSnapshot() + : std::vector<block_id>()), num_workorders_generated_(0), started_(false), input_relation_is_stored_(input_relation_is_stored) {} @@ -151,19 +153,19 @@ class SortRunGenerationWorkOrder : public WorkOrder { /** * @brief Constructor. * + * @param query_id The ID of the query to which this WorkOrder belongs. * @param input_relation The relation to generate sorted runs of. * @param input_block_id The block id. * @param sort_config The Sort configuration specifying ORDER BY, ordering, * and null ordering. - * @param query_id The ID of the query to which this WorkOrder belongs. * @param output_destination The InsertDestination to store the sorted blocks * of runs. * @param storage_manager The StorageManager to use. **/ - SortRunGenerationWorkOrder(const CatalogRelationSchema &input_relation, + SortRunGenerationWorkOrder(const std::size_t query_id, + const CatalogRelationSchema &input_relation, const block_id input_block_id, const SortConfiguration &sort_config, - const std::size_t query_id, InsertDestination *output_destination, StorageManager *storage_manager) : WorkOrder(query_id), http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/1be47dcb/relational_operators/TableGeneratorOperator.cpp ---------------------------------------------------------------------- diff --git a/relational_operators/TableGeneratorOperator.cpp b/relational_operators/TableGeneratorOperator.cpp index fb1f743..a3f9340 100644 --- a/relational_operators/TableGeneratorOperator.cpp +++ b/relational_operators/TableGeneratorOperator.cpp @@ -43,9 +43,9 @@ bool TableGeneratorOperator::getAllWorkOrders( // so just produce one work order. container->addNormalWorkOrder( new TableGeneratorWorkOrder( + query_id_, query_context->getGeneratorFunctionHandle( generator_function_index_), - query_id_, query_context->getInsertDestination(output_destination_index_)), op_index_); started_ = true; http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/1be47dcb/relational_operators/TableGeneratorOperator.hpp ---------------------------------------------------------------------- diff --git a/relational_operators/TableGeneratorOperator.hpp b/relational_operators/TableGeneratorOperator.hpp index bfc70c5..6a6af4b 100644 --- a/relational_operators/TableGeneratorOperator.hpp +++ b/relational_operators/TableGeneratorOperator.hpp @@ -55,23 +55,23 @@ class TableGeneratorOperator : public RelationalOperator { /** * @brief Constructor * + * @param query_id The ID of the query to which this operator belongs. * @param output_relation The output relation. * @param output_destination_index The index of the InsertDestination in the * QueryContext to insert the generated output. * @param generator_function_index The index of the GeneratorFunctionHandle in * the QueryContext. - * @param query_id The ID of the query to which this operator belongs. **/ - TableGeneratorOperator(const CatalogRelation &output_relation, - const QueryContext::insert_destination_id output_destination_index, - const QueryContext::generator_function_id generator_function_index, - const std::size_t query_id) + TableGeneratorOperator( + const std::size_t query_id, + const CatalogRelation &output_relation, + const QueryContext::insert_destination_id output_destination_index, + const QueryContext::generator_function_id generator_function_index) : RelationalOperator(query_id), output_relation_(output_relation), output_destination_index_(output_destination_index), generator_function_index_(generator_function_index), - started_(false) { - } + started_(false) {} ~TableGeneratorOperator() override {} @@ -113,13 +113,13 @@ class TableGeneratorWorkOrder : public WorkOrder { /** * @brief Constructor. * - * @param generator_function The GeneratorFunctionHandle to use. * @param query_id The ID of the query to which this WorkOrder belongs. + * @param generator_function The GeneratorFunctionHandle to use. * @param output_destination The InsertDestination to insert the generated * output. **/ - TableGeneratorWorkOrder(const GeneratorFunctionHandle &function_handle, - const std::size_t query_id, + TableGeneratorWorkOrder(const std::size_t query_id, + const GeneratorFunctionHandle &function_handle, InsertDestination *output_destination) : WorkOrder(query_id), function_handle_(function_handle), http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/1be47dcb/relational_operators/TextScanOperator.cpp ---------------------------------------------------------------------- diff --git a/relational_operators/TextScanOperator.cpp b/relational_operators/TextScanOperator.cpp index 8db5ef1..5acecbf 100644 --- a/relational_operators/TextScanOperator.cpp +++ b/relational_operators/TextScanOperator.cpp @@ -165,9 +165,9 @@ bool TextScanOperator::getAllWorkOrders( // First, generate text-split work orders. for (const auto &file : files) { container->addNormalWorkOrder( - new TextSplitWorkOrder(file, + new TextSplitWorkOrder(query_id_, + file, process_escape_sequences_, - query_id_, storage_manager, op_index_, scheduler_client_id, @@ -182,11 +182,11 @@ bool TextScanOperator::getAllWorkOrders( while (!text_blob_queue_.empty()) { const TextBlob blob_work = text_blob_queue_.popOne(); container->addNormalWorkOrder( - new TextScanWorkOrder(blob_work.blob_id, + new TextScanWorkOrder(query_id_, + blob_work.blob_id, blob_work.size, field_terminator_, process_escape_sequences_, - query_id_, output_destination, storage_manager), op_index_); @@ -203,10 +203,10 @@ bool TextScanOperator::getAllWorkOrders( if (blocking_dependencies_met_ && !work_generated_) { for (const auto &file : files) { container->addNormalWorkOrder( - new TextScanWorkOrder(file, + new TextScanWorkOrder(query_id_, + file, field_terminator_, process_escape_sequences_, - query_id_, output_destination, storage_manager), op_index_); @@ -234,11 +234,10 @@ void TextScanOperator::receiveFeedbackMessage(const WorkOrder::FeedbackMessage & } } - -TextScanWorkOrder::TextScanWorkOrder(const std::string &filename, +TextScanWorkOrder::TextScanWorkOrder(const std::size_t query_id, + const std::string &filename, const char field_terminator, const bool process_escape_sequences, - const std::size_t query_id, InsertDestination *output_destination, StorageManager *storage_manager) : WorkOrder(query_id), @@ -254,11 +253,11 @@ TextScanWorkOrder::TextScanWorkOrder(const std::string &filename, DCHECK(storage_manager_ != nullptr); } -TextScanWorkOrder::TextScanWorkOrder(const block_id text_blob, +TextScanWorkOrder::TextScanWorkOrder(const std::size_t query_id, + const block_id text_blob, const std::size_t text_size, const char field_terminator, const bool process_escape_sequences, - const std::size_t query_id, InsertDestination *output_destination, StorageManager *storage_manager) : WorkOrder(query_id), http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/1be47dcb/relational_operators/TextScanOperator.hpp ---------------------------------------------------------------------- diff --git a/relational_operators/TextScanOperator.hpp b/relational_operators/TextScanOperator.hpp index 1d0c04f..3cda65b 100644 --- a/relational_operators/TextScanOperator.hpp +++ b/relational_operators/TextScanOperator.hpp @@ -121,6 +121,7 @@ class TextScanOperator : public RelationalOperator { /** * @brief Constructor * + * @param query_id The ID of the query to which this operator belongs. * @param file_pattern The glob-like file pattern of the sources to load. The * pattern could include * (wildcard for multiple chars) and ? * (wildcard for single char). It defaults to single file load, if a @@ -134,15 +135,15 @@ class TextScanOperator : public RelationalOperator { * @param output_relation The output relation. * @param output_destination_index The index of the InsertDestination in the * QueryContext to insert tuples. - * @param query_id The ID of the query to which this operator belongs. **/ - TextScanOperator(const std::string &file_pattern, - const char field_terminator, - const bool process_escape_sequences, - const bool parallelize_load, - const CatalogRelation &output_relation, - const QueryContext::insert_destination_id output_destination_index, - const std::size_t query_id) + TextScanOperator( + const std::size_t query_id, + const std::string &file_pattern, + const char field_terminator, + const bool process_escape_sequences, + const bool parallelize_load, + const CatalogRelation &output_relation, + const QueryContext::insert_destination_id output_destination_index) : RelationalOperator(query_id), file_pattern_(file_pattern), field_terminator_(field_terminator), @@ -200,42 +201,42 @@ class TextScanWorkOrder : public WorkOrder { /** * @brief Constructor * + * @param query_id The ID of the query to which this WorkOrder belongs. * @param filename The name of the text file to bulk insert. * @param field_terminator The string which separates attribute values in * the text file. * @param process_escape_sequences Whether to decode escape sequences in the * text file. - * @param query_id The ID of the query to which this WorkOrder belongs. * @param output_destination The InsertDestination to insert tuples. * @param storage_manager The StorageManager to use. **/ TextScanWorkOrder( + const std::size_t query_id, const std::string &filename, const char field_terminator, const bool process_escape_sequences, - const std::size_t query_id, InsertDestination *output_destination, StorageManager *storage_manager); /** * @brief Constructor. * + * @param query_id The ID of the query to which this WorkOrder belongs. * @param text_blob Blob ID containing the data to be scanned. * @param text_size Size of the data in the blob. * @param field_terminator The character which separates attribute values in * the text file. * @param process_escape_sequences Whether to decode escape sequences in the * text file. - * @param query_id The ID of the query to which this WorkOrder belongs. * @param output_destination The InsertDestination to write the read tuples. * @param storage_manager The StorageManager to use. */ TextScanWorkOrder( + const std::size_t query_id, const block_id text_blob, const std::size_t text_size, const char field_terminator, const bool process_escape_sequences, - const std::size_t query_id, InsertDestination *output_destination, StorageManager *storage_manager); @@ -322,19 +323,20 @@ class TextSplitWorkOrder : public WorkOrder { public: /** * @brief Constructor. + * + * @param query_id The ID of the query to which this WorkOrder belongs. * @param filename File to split into row-aligned blobs. * @param process_escape_sequences Whether to decode escape sequences in the * text file. - * @param query_id The ID of the query to which this WorkOrder belongs. * @param storage_manager The StorageManager to use. * @param operator_index Operator index of the current operator. This is used * to send new-work available message to Foreman. * @param scheduler_client_id The TMB client ID of the scheduler thread. * @param bus A pointer to the TMB. */ - TextSplitWorkOrder(const std::string &filename, + TextSplitWorkOrder(const std::size_t query_id, + const std::string &filename, const bool process_escape_sequences, - const std::size_t query_id, StorageManager *storage_manager, const std::size_t operator_index, const tmb::client_id scheduler_client_id, http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/1be47dcb/relational_operators/UpdateOperator.cpp ---------------------------------------------------------------------- diff --git a/relational_operators/UpdateOperator.cpp b/relational_operators/UpdateOperator.cpp index b331a9c..2130563 100644 --- a/relational_operators/UpdateOperator.cpp +++ b/relational_operators/UpdateOperator.cpp @@ -53,16 +53,18 @@ bool UpdateOperator::getAllWorkOrders( for (const block_id input_block_id : input_blocks_) { container->addNormalWorkOrder( - new UpdateWorkOrder(relation_, - input_block_id, - query_context->getPredicate(predicate_index_), - query_context->getUpdateGroup(update_group_index_), - query_id_, - query_context->getInsertDestination(relocation_destination_index_), - storage_manager, - op_index_, - scheduler_client_id, - bus), + new UpdateWorkOrder( + query_id_, + relation_, + input_block_id, + query_context->getPredicate(predicate_index_), + query_context->getUpdateGroup(update_group_index_), + query_context->getInsertDestination( + relocation_destination_index_), + storage_manager, + op_index_, + scheduler_client_id, + bus), op_index_); } started_ = true; http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/1be47dcb/relational_operators/UpdateOperator.hpp ---------------------------------------------------------------------- diff --git a/relational_operators/UpdateOperator.hpp b/relational_operators/UpdateOperator.hpp index ba2d6cf..cebb9b5 100644 --- a/relational_operators/UpdateOperator.hpp +++ b/relational_operators/UpdateOperator.hpp @@ -62,6 +62,7 @@ class UpdateOperator : public RelationalOperator { /** * @brief Constructor * + * @param query_id The ID of the query to which this operator belongs. * @param relation The relation to perform the UPDATE over. * @param relocation_destination_index The index of the InsertDestination in * the QueryContext to relocate tuples which can not be updated @@ -72,16 +73,16 @@ class UpdateOperator : public RelationalOperator { * @param update_group_index The index of a update group (the map of * attribute_ids to Scalars) which should be evaluated to get the new * value for the corresponding attribute. - * @param query_id The ID of the query to which this operator belongs. * * @warning The constructed InsertDestination should belong to relation, but * must NOT contain any pre-existing blocks. **/ - UpdateOperator(const CatalogRelation &relation, - const QueryContext::insert_destination_id relocation_destination_index, - const QueryContext::predicate_id predicate_index, - const QueryContext::update_group_id update_group_index, - const std::size_t query_id) + UpdateOperator( + const std::size_t query_id, + const CatalogRelation &relation, + const QueryContext::insert_destination_id relocation_destination_index, + const QueryContext::predicate_id predicate_index, + const QueryContext::update_group_id update_group_index) : RelationalOperator(query_id), relation_(relation), relocation_destination_index_(relocation_destination_index), @@ -127,13 +128,13 @@ class UpdateWorkOrder : public WorkOrder { /** * @brief Constructor * + * @param query_id The ID of the query to which this WorkOrder belongs. * @param relation The relation to perform the UPDATE over. * @param predicate All tuples matching \c predicate will be updated (or NULL * to update all tuples). * @param assignments The assignments (the map of attribute_ids to Scalars) * which should be evaluated to get the new value for the corresponding * attribute. - * @param query_id The ID of the query to which this WorkOrder belongs. * @param input_block_id The block id. * @param relocation_destination The InsertDestination to relocate tuples * which can not be updated in-place. @@ -143,16 +144,18 @@ class UpdateWorkOrder : public WorkOrder { * @param scheduler_client_id The TMB client ID of the scheduler thread. * @param bus A pointer to the TMB. **/ - UpdateWorkOrder(const CatalogRelationSchema &relation, - const block_id input_block_id, - const Predicate *predicate, - const std::unordered_map<attribute_id, std::unique_ptr<const Scalar>> &assignments, - const std::size_t query_id, - InsertDestination *relocation_destination, - StorageManager *storage_manager, - const std::size_t update_operator_index, - const tmb::client_id scheduler_client_id, - MessageBus *bus) + UpdateWorkOrder( + const std::size_t query_id, + const CatalogRelationSchema &relation, + const block_id input_block_id, + const Predicate *predicate, + const std::unordered_map<attribute_id, std::unique_ptr<const Scalar>> + &assignments, + InsertDestination *relocation_destination, + StorageManager *storage_manager, + const std::size_t update_operator_index, + const tmb::client_id scheduler_client_id, + MessageBus *bus) : WorkOrder(query_id), relation_(relation), input_block_id_(input_block_id), http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/1be47dcb/relational_operators/WorkOrderFactory.cpp ---------------------------------------------------------------------- diff --git a/relational_operators/WorkOrderFactory.cpp b/relational_operators/WorkOrderFactory.cpp index fdd694f..489b666 100644 --- a/relational_operators/WorkOrderFactory.cpp +++ b/relational_operators/WorkOrderFactory.cpp @@ -75,8 +75,8 @@ WorkOrder* WorkOrderFactory::ReconstructFromProto(const serialization::WorkOrder case serialization::AGGREGATION: { LOG(INFO) << "Creating AggregationWorkOrder"; return new AggregationWorkOrder( - proto.GetExtension(serialization::AggregationWorkOrder::block_id), proto.query_id(), + proto.GetExtension(serialization::AggregationWorkOrder::block_id), query_context->getAggregationState( proto.GetExtension(serialization::AggregationWorkOrder::aggr_state_index))); } @@ -89,12 +89,12 @@ WorkOrder* WorkOrderFactory::ReconstructFromProto(const serialization::WorkOrder } return new BuildHashWorkOrder( + proto.query_id(), catalog_database->getRelationSchemaById( proto.GetExtension(serialization::BuildHashWorkOrder::relation_id)), move(join_key_attributes), proto.GetExtension(serialization::BuildHashWorkOrder::any_join_key_attributes_nullable), proto.GetExtension(serialization::BuildHashWorkOrder::block_id), - proto.query_id(), query_context->getJoinHashTable( proto.GetExtension(serialization::BuildHashWorkOrder::join_hash_table_index)), storage_manager); @@ -102,6 +102,7 @@ WorkOrder* WorkOrderFactory::ReconstructFromProto(const serialization::WorkOrder case serialization::DELETE: { LOG(INFO) << "Creating DeleteWorkOrder"; return new DeleteWorkOrder( + proto.query_id(), catalog_database->getRelationSchemaById( proto.GetExtension(serialization::DeleteWorkOrder::relation_id)), proto.GetExtension(serialization::DeleteWorkOrder::block_id), @@ -110,15 +111,14 @@ WorkOrder* WorkOrderFactory::ReconstructFromProto(const serialization::WorkOrder storage_manager, proto.GetExtension(serialization::DeleteWorkOrder::operator_index), shiftboss_client_id, - proto.query_id(), bus); } case serialization::DESTROY_HASH: { LOG(INFO) << "Creating DestroyHashWorkOrder"; return new DestroyHashWorkOrder( + proto.query_id(), proto.GetExtension( serialization::DestroyHashWorkOrder::join_hash_table_index), - proto.query_id(), query_context); } case serialization::DROP_TABLE: { @@ -193,6 +193,7 @@ WorkOrder* WorkOrderFactory::ReconstructFromProto(const serialization::WorkOrder case serialization::HashJoinWorkOrder::HASH_ANTI_JOIN: { LOG(INFO) << "Creating HashAntiJoinWorkOrder"; return new HashAntiJoinWorkOrder( + proto.query_id(), build_relation, probe_relation, move(join_key_attributes), @@ -201,13 +202,13 @@ WorkOrder* WorkOrderFactory::ReconstructFromProto(const serialization::WorkOrder residual_predicate, selection, hash_table, - proto.query_id(), output_destination, storage_manager); } case serialization::HashJoinWorkOrder::HASH_INNER_JOIN: { LOG(INFO) << "Creating HashInnerJoinWorkOrder"; return new HashInnerJoinWorkOrder( + proto.query_id(), build_relation, probe_relation, move(join_key_attributes), @@ -216,7 +217,6 @@ WorkOrder* WorkOrderFactory::ReconstructFromProto(const serialization::WorkOrder residual_predicate, selection, hash_table, - proto.query_id(), output_destination, storage_manager); } @@ -231,6 +231,7 @@ WorkOrder* WorkOrderFactory::ReconstructFromProto(const serialization::WorkOrder LOG(INFO) << "Creating HashOuterJoinWorkOrder"; return new HashOuterJoinWorkOrder( + proto.query_id(), build_relation, probe_relation, move(join_key_attributes), @@ -239,13 +240,13 @@ WorkOrder* WorkOrderFactory::ReconstructFromProto(const serialization::WorkOrder selection, move(is_selection_on_build), hash_table, - proto.query_id(), output_destination, storage_manager); } case serialization::HashJoinWorkOrder::HASH_SEMI_JOIN: { LOG(INFO) << "Creating HashSemiJoinWorkOrder"; return new HashSemiJoinWorkOrder( + proto.query_id(), build_relation, probe_relation, move(join_key_attributes), @@ -254,7 +255,6 @@ WorkOrder* WorkOrderFactory::ReconstructFromProto(const serialization::WorkOrder residual_predicate, selection, hash_table, - proto.query_id(), output_destination, storage_manager); } @@ -274,6 +274,7 @@ WorkOrder* WorkOrderFactory::ReconstructFromProto(const serialization::WorkOrder case serialization::NESTED_LOOP_JOIN: { LOG(INFO) << "Creating NestedLoopsJoinWorkOrder"; return new NestedLoopsJoinWorkOrder( + proto.query_id(), catalog_database->getRelationSchemaById( proto.GetExtension(serialization::NestedLoopsJoinWorkOrder::left_relation_id)), catalog_database->getRelationSchemaById( @@ -284,7 +285,6 @@ WorkOrder* WorkOrderFactory::ReconstructFromProto(const serialization::WorkOrder proto.GetExtension(serialization::NestedLoopsJoinWorkOrder::join_predicate_index)), query_context->getScalarGroup( proto.GetExtension(serialization::NestedLoopsJoinWorkOrder::selection_index)), - proto.query_id(), query_context->getInsertDestination( proto.GetExtension(serialization::NestedLoopsJoinWorkOrder::insert_destination_index)), storage_manager); @@ -292,12 +292,12 @@ WorkOrder* WorkOrderFactory::ReconstructFromProto(const serialization::WorkOrder case serialization::SAMPLE: { LOG(INFO) << "Creating SampleWorkOrder"; return new SampleWorkOrder( + proto.query_id(), catalog_database->getRelationSchemaById( proto.GetExtension(serialization::SampleWorkOrder::relation_id)), proto.GetExtension(serialization::SampleWorkOrder::block_id), proto.GetExtension(serialization::SampleWorkOrder::is_block_sample), proto.GetExtension(serialization::SampleWorkOrder::percentage), - proto.query_id(), query_context->getInsertDestination( proto.GetExtension(serialization::SampleWorkOrder::insert_destination_index)), storage_manager); @@ -305,9 +305,9 @@ WorkOrder* WorkOrderFactory::ReconstructFromProto(const serialization::WorkOrder case serialization::SAVE_BLOCKS: { LOG(INFO) << "Creating SaveBlocksWorkOrder"; return new SaveBlocksWorkOrder( + proto.query_id(), proto.GetExtension(serialization::SaveBlocksWorkOrder::block_id), proto.GetExtension(serialization::SaveBlocksWorkOrder::force), - proto.query_id(), storage_manager); } case serialization::SELECT: { @@ -321,6 +321,7 @@ WorkOrder* WorkOrderFactory::ReconstructFromProto(const serialization::WorkOrder } return new SelectWorkOrder( + proto.query_id(), catalog_database->getRelationSchemaById( proto.GetExtension(serialization::SelectWorkOrder::relation_id)), proto.GetExtension(serialization::SelectWorkOrder::block_id), @@ -331,7 +332,6 @@ WorkOrder* WorkOrderFactory::ReconstructFromProto(const serialization::WorkOrder simple_projection ? nullptr : &query_context->getScalarGroup( proto.GetExtension(serialization::SelectWorkOrder::selection_index)), - proto.query_id(), query_context->getInsertDestination( proto.GetExtension(serialization::SelectWorkOrder::insert_destination_index)), storage_manager); @@ -350,6 +350,7 @@ WorkOrder* WorkOrderFactory::ReconstructFromProto(const serialization::WorkOrder } return new SortMergeRunWorkOrder( + proto.query_id(), query_context->getSortConfig( proto.GetExtension(serialization::SortMergeRunWorkOrder::sort_config_index)), catalog_database->getRelationSchemaById( @@ -357,7 +358,6 @@ WorkOrder* WorkOrderFactory::ReconstructFromProto(const serialization::WorkOrder move(runs), proto.GetExtension(serialization::SortMergeRunWorkOrder::top_k), proto.GetExtension(serialization::SortMergeRunWorkOrder::merge_level), - proto.query_id(), query_context->getInsertDestination( proto.GetExtension(serialization::SortMergeRunWorkOrder::insert_destination_index)), storage_manager, @@ -368,12 +368,12 @@ WorkOrder* WorkOrderFactory::ReconstructFromProto(const serialization::WorkOrder case serialization::SORT_RUN_GENERATION: { LOG(INFO) << "Creating SortRunGenerationWorkOrder"; return new SortRunGenerationWorkOrder( + proto.query_id(), catalog_database->getRelationSchemaById( proto.GetExtension(serialization::SortRunGenerationWorkOrder::relation_id)), proto.GetExtension(serialization::SortRunGenerationWorkOrder::block_id), query_context->getSortConfig( proto.GetExtension(serialization::SortRunGenerationWorkOrder::sort_config_index)), - proto.query_id(), query_context->getInsertDestination( proto.GetExtension(serialization::SortRunGenerationWorkOrder::insert_destination_index)), storage_manager); @@ -381,9 +381,9 @@ WorkOrder* WorkOrderFactory::ReconstructFromProto(const serialization::WorkOrder case serialization::TABLE_GENERATOR: { LOG(INFO) << "Creating SortRunGenerationWorkOrder"; return new TableGeneratorWorkOrder( + proto.query_id(), query_context->getGeneratorFunctionHandle( proto.GetExtension(serialization::TableGeneratorWorkOrder::generator_function_index)), - proto.query_id(), query_context->getInsertDestination( proto.GetExtension(serialization::TableGeneratorWorkOrder::insert_destination_index))); } @@ -391,10 +391,10 @@ WorkOrder* WorkOrderFactory::ReconstructFromProto(const serialization::WorkOrder LOG(INFO) << "Creating TextScanWorkOrder"; if (proto.HasExtension(serialization::TextScanWorkOrder::filename)) { return new TextScanWorkOrder( + proto.query_id(), proto.GetExtension(serialization::TextScanWorkOrder::filename), proto.GetExtension(serialization::TextScanWorkOrder::field_terminator), proto.GetExtension(serialization::TextScanWorkOrder::process_escape_sequences), - proto.query_id(), query_context->getInsertDestination( proto.GetExtension(serialization::TextScanWorkOrder::insert_destination_index)), storage_manager); @@ -403,11 +403,11 @@ WorkOrder* WorkOrderFactory::ReconstructFromProto(const serialization::WorkOrder proto.GetExtension(serialization::TextScanWorkOrder::text_blob); return new TextScanWorkOrder( + proto.query_id(), text_blob_proto.blob_id(), text_blob_proto.size(), proto.GetExtension(serialization::TextScanWorkOrder::field_terminator), proto.GetExtension(serialization::TextScanWorkOrder::process_escape_sequences), - proto.query_id(), query_context->getInsertDestination( proto.GetExtension(serialization::TextScanWorkOrder::insert_destination_index)), storage_manager); @@ -416,9 +416,9 @@ WorkOrder* WorkOrderFactory::ReconstructFromProto(const serialization::WorkOrder case serialization::TEXT_SPLIT: { LOG(INFO) << "Creating TextSplitWorkOrder"; return new TextSplitWorkOrder( + proto.query_id(), proto.GetExtension(serialization::TextSplitWorkOrder::filename), proto.GetExtension(serialization::TextSplitWorkOrder::process_escape_sequences), - proto.query_id(), storage_manager, proto.GetExtension(serialization::TextSplitWorkOrder::operator_index), shiftboss_client_id, @@ -427,6 +427,7 @@ WorkOrder* WorkOrderFactory::ReconstructFromProto(const serialization::WorkOrder case serialization::UPDATE: { LOG(INFO) << "Creating UpdateWorkOrder"; return new UpdateWorkOrder( + proto.query_id(), catalog_database->getRelationSchemaById( proto.GetExtension(serialization::UpdateWorkOrder::relation_id)), proto.GetExtension(serialization::UpdateWorkOrder::block_id), @@ -434,7 +435,6 @@ WorkOrder* WorkOrderFactory::ReconstructFromProto(const serialization::WorkOrder proto.GetExtension(serialization::UpdateWorkOrder::predicate_index)), query_context->getUpdateGroup( proto.GetExtension(serialization::UpdateWorkOrder::update_group_index)), - proto.query_id(), query_context->getInsertDestination( proto.GetExtension(serialization::UpdateWorkOrder::insert_destination_index)), storage_manager, http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/1be47dcb/relational_operators/tests/AggregationOperator_unittest.cpp ---------------------------------------------------------------------- diff --git a/relational_operators/tests/AggregationOperator_unittest.cpp b/relational_operators/tests/AggregationOperator_unittest.cpp index ace7951..fdcc54f 100644 --- a/relational_operators/tests/AggregationOperator_unittest.cpp +++ b/relational_operators/tests/AggregationOperator_unittest.cpp @@ -77,6 +77,7 @@ using std::unique_ptr; namespace quickstep { namespace { +constexpr std::size_t kQueryId = 0; constexpr int kOpIndex = 0; } // namespace @@ -270,7 +271,7 @@ class AggregationOperatorTest : public ::testing::Test { aggr_state_proto->set_estimated_num_entries(estimated_entries); // Create Operators. - op_.reset(new AggregationOperator(*table_, true, aggr_state_index, 0)); + op_.reset(new AggregationOperator(0, *table_, true, aggr_state_index)); // Setup the InsertDestination proto in the query context proto. const QueryContext::insert_destination_id insert_destination_index = @@ -281,10 +282,11 @@ class AggregationOperatorTest : public ::testing::Test { insert_destination_proto->set_relation_id(result_table_->getID()); insert_destination_proto->set_relational_op_index(kOpIndex); - finalize_op_.reset(new FinalizeAggregationOperator(aggr_state_index, - *result_table_, - insert_destination_index, - 0 /* dummy query ID */)); + finalize_op_.reset( + new FinalizeAggregationOperator(kQueryId, + aggr_state_index, + *result_table_, + insert_destination_index)); // Set up the QueryContext. query_context_.reset(new QueryContext(query_context_proto, @@ -354,7 +356,7 @@ class AggregationOperatorTest : public ::testing::Test { serialization::HashTableImplType::LINEAR_OPEN_ADDRESSING); // Create Operators. - op_.reset(new AggregationOperator(*table_, true, aggr_state_index, 0)); + op_.reset(new AggregationOperator(0, *table_, true, aggr_state_index)); // Setup the InsertDestination proto in the query context proto. const QueryContext::insert_destination_id insert_destination_index = @@ -365,10 +367,11 @@ class AggregationOperatorTest : public ::testing::Test { insert_destination_proto->set_relation_id(result_table_->getID()); insert_destination_proto->set_relational_op_index(kOpIndex); - finalize_op_.reset(new FinalizeAggregationOperator(aggr_state_index, - *result_table_, - insert_destination_index, - 0 /* dummy query ID */)); + finalize_op_.reset( + new FinalizeAggregationOperator(kQueryId, + aggr_state_index, + *result_table_, + insert_destination_index)); // Set up the QueryContext. query_context_.reset(new QueryContext(query_context_proto,