Added output_num_partitions in RelationalOperators.
diff --git a/relational_operators/DeleteOperator.hpp b/relational_operators/DeleteOperator.hpp
index b084560..cc6f6cf 100644
--- a/relational_operators/DeleteOperator.hpp
+++ b/relational_operators/DeleteOperator.hpp
@@ -74,7 +74,8 @@
const CatalogRelation &relation,
const QueryContext::predicate_id predicate_index,
const bool relation_is_stored)
- : RelationalOperator(query_id, relation.getNumPartitions()),
+ : RelationalOperator(query_id, relation.getNumPartitions(), false /* has_repartition */,
+ relation.getNumPartitions()),
relation_(relation),
predicate_index_(predicate_index),
relation_is_stored_(relation_is_stored),
diff --git a/relational_operators/FinalizeAggregationOperator.hpp b/relational_operators/FinalizeAggregationOperator.hpp
index b344c13..5931ca2 100644
--- a/relational_operators/FinalizeAggregationOperator.hpp
+++ b/relational_operators/FinalizeAggregationOperator.hpp
@@ -74,12 +74,14 @@
const std::size_t aggr_state_num_partitions,
const CatalogRelation &output_relation,
const QueryContext::insert_destination_id output_destination_index)
- : RelationalOperator(query_id, num_partitions, has_repartition),
+ : RelationalOperator(query_id, num_partitions, has_repartition, output_relation.getNumPartitions()),
aggr_state_index_(aggr_state_index),
aggr_state_num_partitions_(aggr_state_num_partitions),
output_relation_(output_relation),
output_destination_index_(output_destination_index),
- started_(false) {}
+ started_(false) {
+ DCHECK(has_repartition || num_partitions == output_num_partitions_);
+ }
~FinalizeAggregationOperator() override {}
diff --git a/relational_operators/HashJoinOperator.hpp b/relational_operators/HashJoinOperator.hpp
index a5e6eb4..306875e 100644
--- a/relational_operators/HashJoinOperator.hpp
+++ b/relational_operators/HashJoinOperator.hpp
@@ -139,7 +139,7 @@
const QueryContext::scalar_group_id selection_index,
const std::vector<bool> *is_selection_on_build = nullptr,
const JoinType join_type = JoinType::kInnerJoin)
- : RelationalOperator(query_id, num_partitions, has_repartition),
+ : RelationalOperator(query_id, num_partitions, has_repartition, output_relation.getNumPartitions()),
build_relation_(build_relation),
probe_relation_(probe_relation),
probe_relation_is_stored_(probe_relation_is_stored),
@@ -157,6 +157,7 @@
probe_relation_block_ids_(num_partitions),
num_workorders_generated_(num_partitions),
started_(false) {
+ DCHECK(has_repartition || num_partitions == output_num_partitions_);
DCHECK(join_type != JoinType::kLeftOuterJoin ||
(is_selection_on_build != nullptr &&
residual_predicate_index == QueryContext::kInvalidPredicateId));
diff --git a/relational_operators/InsertOperator.hpp b/relational_operators/InsertOperator.hpp
index b0bdf87..b103538 100644
--- a/relational_operators/InsertOperator.hpp
+++ b/relational_operators/InsertOperator.hpp
@@ -68,7 +68,7 @@
const CatalogRelation &output_relation,
const QueryContext::insert_destination_id output_destination_index,
const QueryContext::tuple_id tuple_index)
- : RelationalOperator(query_id),
+ : RelationalOperator(query_id, 1u, false, output_relation.getNumPartitions()),
output_relation_(output_relation),
output_destination_index_(output_destination_index),
tuple_index_(tuple_index),
diff --git a/relational_operators/NestedLoopsJoinOperator.hpp b/relational_operators/NestedLoopsJoinOperator.hpp
index 777a737..daca91f 100644
--- a/relational_operators/NestedLoopsJoinOperator.hpp
+++ b/relational_operators/NestedLoopsJoinOperator.hpp
@@ -102,7 +102,7 @@
const QueryContext::scalar_group_id selection_index,
const bool left_relation_is_stored,
const bool right_relation_is_stored)
- : RelationalOperator(query_id, num_partitions, has_repartition),
+ : RelationalOperator(query_id, num_partitions, has_repartition, output_relation.getNumPartitions()),
nested_loops_join_index_(nested_loops_join_index),
left_input_relation_(left_input_relation),
right_input_relation_(right_input_relation),
diff --git a/relational_operators/RelationalOperator.hpp b/relational_operators/RelationalOperator.hpp
index 0541bbd..9307b5a 100644
--- a/relational_operators/RelationalOperator.hpp
+++ b/relational_operators/RelationalOperator.hpp
@@ -278,6 +278,15 @@
}
/**
+ * @brief Get the number of partitions of the output relation in the operator.
+ *
+ * @return The number of partitions of the output relation.
+ */
+ std::size_t getOutputNumPartitions() const {
+ return output_num_partitions_;
+ }
+
+ /**
* @brief Deploy a group of LIPFilters to this operator.
*/
void deployLIPFilters(const QueryContext::lip_deployment_id lip_deployment_index,
@@ -294,18 +303,22 @@
* @param num_partitions The number of partitions of the input relation.
* If create table / index, return zero. If no partitions, return one.
* @param has_repartition Whether this operator does repartition.
+ * @param output_num_partitions The number of partitions of the output
+ * relation. If no output, return zero. If no partitions, return one.
**/
explicit RelationalOperator(const std::size_t query_id,
const std::size_t num_partitions = 1u,
- const bool has_repartition = false)
+ const bool has_repartition = false,
+ const std::size_t output_num_partitions = 0u)
: query_id_(query_id),
num_partitions_(num_partitions),
+ output_num_partitions_(output_num_partitions),
has_repartition_(has_repartition),
done_feeding_input_relation_(false),
lip_deployment_index_(QueryContext::kInvalidLIPDeploymentId) {}
const std::size_t query_id_;
- const std::size_t num_partitions_;
+ const std::size_t num_partitions_, output_num_partitions_;
const bool has_repartition_;
bool done_feeding_input_relation_;
diff --git a/relational_operators/SelectOperator.hpp b/relational_operators/SelectOperator.hpp
index 71dcfca..e7cdcd0 100644
--- a/relational_operators/SelectOperator.hpp
+++ b/relational_operators/SelectOperator.hpp
@@ -96,7 +96,8 @@
const QueryContext::predicate_id predicate_index,
const QueryContext::scalar_group_id selection_index,
const bool input_relation_is_stored)
- : RelationalOperator(query_id, input_relation.getNumPartitions(), has_repartition),
+ : RelationalOperator(query_id, input_relation.getNumPartitions(), has_repartition,
+ output_relation.getNumPartitions()),
input_relation_(input_relation),
output_relation_(output_relation),
output_destination_index_(output_destination_index),
@@ -107,6 +108,7 @@
simple_projection_(false),
input_relation_is_stored_(input_relation_is_stored),
started_(false) {
+ DCHECK(has_repartition || num_partitions_ == output_num_partitions_);
#ifdef QUICKSTEP_HAVE_LIBNUMA
placement_scheme_ = input_relation.getNUMAPlacementSchemePtr();
#endif
@@ -153,7 +155,8 @@
const QueryContext::predicate_id predicate_index,
std::vector<attribute_id> &&selection,
const bool input_relation_is_stored)
- : RelationalOperator(query_id, input_relation.getNumPartitions(), has_repartition),
+ : RelationalOperator(query_id, input_relation.getNumPartitions(), has_repartition,
+ output_relation.getNumPartitions()),
input_relation_(input_relation),
output_relation_(output_relation),
output_destination_index_(output_destination_index),
@@ -165,6 +168,7 @@
simple_projection_(true),
input_relation_is_stored_(input_relation_is_stored),
started_(false) {
+ DCHECK(has_repartition || num_partitions_ == output_num_partitions_);
#ifdef QUICKSTEP_HAVE_LIBNUMA
placement_scheme_ = input_relation.getNUMAPlacementSchemePtr();
#endif
diff --git a/relational_operators/SortMergeRunOperator.hpp b/relational_operators/SortMergeRunOperator.hpp
index a8ecb88..6287068 100644
--- a/relational_operators/SortMergeRunOperator.hpp
+++ b/relational_operators/SortMergeRunOperator.hpp
@@ -107,7 +107,8 @@
const std::size_t merge_factor,
const std::size_t top_k,
const bool input_relation_is_stored)
- : RelationalOperator(query_id),
+ : RelationalOperator(query_id, 1u /* input_num_partitions */, false /* has_repartition */,
+ 1u /* output_num_partition */),
input_relation_(input_relation),
output_relation_(output_relation),
output_destination_index_(output_destination_index),
@@ -124,6 +125,8 @@
input_relation_is_stored_(input_relation_is_stored),
input_stream_done_(input_relation_is_stored),
started_(false) {
+ DCHECK_EQ(1u, input_relation.getNumPartitions());
+ DCHECK_EQ(1u, output_relation.getNumPartitions());
DCHECK_GT(merge_factor_, 1u);
}
diff --git a/relational_operators/SortRunGenerationOperator.hpp b/relational_operators/SortRunGenerationOperator.hpp
index 716d4f7..aaac1c3 100644
--- a/relational_operators/SortRunGenerationOperator.hpp
+++ b/relational_operators/SortRunGenerationOperator.hpp
@@ -98,7 +98,8 @@
const QueryContext::insert_destination_id output_destination_index,
const QueryContext::sort_config_id sort_config_index,
bool input_relation_is_stored)
- : RelationalOperator(query_id),
+ : RelationalOperator(query_id, 1u /* input_num_partitions */, false /* has_repartition */,
+ 1u /* output_num_partition */),
input_relation_(input_relation),
output_relation_(output_relation),
output_destination_index_(output_destination_index),
@@ -108,7 +109,9 @@
: std::vector<block_id>()),
num_workorders_generated_(0),
started_(false),
- input_relation_is_stored_(input_relation_is_stored) {}
+ input_relation_is_stored_(input_relation_is_stored) {
+ DCHECK_EQ(1u, output_relation.getNumPartitions());
+ }
~SortRunGenerationOperator() {}
diff --git a/relational_operators/TableGeneratorOperator.hpp b/relational_operators/TableGeneratorOperator.hpp
index adfe2bb..c6a6b7d 100644
--- a/relational_operators/TableGeneratorOperator.hpp
+++ b/relational_operators/TableGeneratorOperator.hpp
@@ -70,7 +70,8 @@
const CatalogRelation &output_relation,
const QueryContext::insert_destination_id output_destination_index,
const QueryContext::generator_function_id generator_function_index)
- : RelationalOperator(query_id),
+ : RelationalOperator(query_id, 1u, output_relation.getNumPartitions() != 1u /* has_repartition */,
+ output_relation.getNumPartitions()),
output_relation_(output_relation),
output_destination_index_(output_destination_index),
generator_function_index_(generator_function_index),
diff --git a/relational_operators/TextScanOperator.hpp b/relational_operators/TextScanOperator.hpp
index f6be8c8..4dbeb92 100644
--- a/relational_operators/TextScanOperator.hpp
+++ b/relational_operators/TextScanOperator.hpp
@@ -125,7 +125,8 @@
const bool process_escape_sequences,
const CatalogRelation &output_relation,
const QueryContext::insert_destination_id output_destination_index)
- : RelationalOperator(query_id),
+ : RelationalOperator(query_id, 1u, output_relation.getNumPartitions() != 1u /* has_repartition */,
+ output_relation.getNumPartitions()),
file_pattern_(file_pattern),
field_terminator_(field_terminator),
process_escape_sequences_(process_escape_sequences),
diff --git a/relational_operators/UpdateOperator.hpp b/relational_operators/UpdateOperator.hpp
index fd2096b..7b7fad6 100644
--- a/relational_operators/UpdateOperator.hpp
+++ b/relational_operators/UpdateOperator.hpp
@@ -88,7 +88,8 @@
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.getNumPartitions()),
+ : RelationalOperator(query_id, relation.getNumPartitions(), false /* has_repartition */,
+ relation.getNumPartitions()),
relation_(relation),
relocation_destination_index_(relocation_destination_index),
predicate_index_(predicate_index),
diff --git a/relational_operators/WindowAggregationOperator.hpp b/relational_operators/WindowAggregationOperator.hpp
index 10546b4..beaf1a3 100644
--- a/relational_operators/WindowAggregationOperator.hpp
+++ b/relational_operators/WindowAggregationOperator.hpp
@@ -70,7 +70,10 @@
const CatalogRelation &output_relation,
const QueryContext::window_aggregation_state_id window_aggregation_state_index,
const QueryContext::insert_destination_id output_destination_index)
- : RelationalOperator(query_id),
+ : RelationalOperator(query_id, input_relation.getNumPartitions(),
+ input_relation.getNumPartitions() !=
+ output_relation.getNumPartitions() /* has_repartition */,
+ output_relation.getNumPartitions()),
input_relation_(input_relation),
output_relation_(output_relation),
window_aggregation_state_index_(window_aggregation_state_index),
