Fixed the incorrect exchange merging introduced by the previous commit;
updated the IntroHashPartitionMergeExchange rule to handle the
hash-merge-exchange operator.
diff --git a/algebricks/algebricks-rewriter/src/main/java/edu/uci/ics/hyracks/algebricks/rewriter/rules/EnforceStructuralPropertiesRule.java b/algebricks/algebricks-rewriter/src/main/java/edu/uci/ics/hyracks/algebricks/rewriter/rules/EnforceStructuralPropertiesRule.java
index 11122d7..810defc 100644
--- a/algebricks/algebricks-rewriter/src/main/java/edu/uci/ics/hyracks/algebricks/rewriter/rules/EnforceStructuralPropertiesRule.java
+++ b/algebricks/algebricks-rewriter/src/main/java/edu/uci/ics/hyracks/algebricks/rewriter/rules/EnforceStructuralPropertiesRule.java
@@ -540,10 +540,6 @@
}
}
Mutable<ILogicalOperator> ci = op.getInputs().get(i);
- if (((AbstractLogicalOperator) ci.getValue()).getOperatorTag() == LogicalOperatorTag.EXCHANGE) {
- ci = ci.getValue().getInputs().get(0);
- op.getInputs().set(i, ci);
- }
ExchangeOperator exchg = new ExchangeOperator();
exchg.setPhysicalOperator(pop);
setNewOp(ci, exchg, context);
diff --git a/algebricks/algebricks-rewriter/src/main/java/edu/uci/ics/hyracks/algebricks/rewriter/rules/IntroHashPartitionMergeExchange.java b/algebricks/algebricks-rewriter/src/main/java/edu/uci/ics/hyracks/algebricks/rewriter/rules/IntroHashPartitionMergeExchange.java
index 82e6970..e702d9f 100644
--- a/algebricks/algebricks-rewriter/src/main/java/edu/uci/ics/hyracks/algebricks/rewriter/rules/IntroHashPartitionMergeExchange.java
+++ b/algebricks/algebricks-rewriter/src/main/java/edu/uci/ics/hyracks/algebricks/rewriter/rules/IntroHashPartitionMergeExchange.java
@@ -43,7 +43,8 @@
throws AlgebricksException {
AbstractLogicalOperator op1 = (AbstractLogicalOperator) opRef.getValue();
if (op1.getPhysicalOperator() == null
- || op1.getPhysicalOperator().getOperatorTag() != PhysicalOperatorTag.HASH_PARTITION_EXCHANGE) {
+ || (op1.getPhysicalOperator().getOperatorTag() != PhysicalOperatorTag.HASH_PARTITION_EXCHANGE && op1
+ .getPhysicalOperator().getOperatorTag() != PhysicalOperatorTag.HASH_PARTITION_MERGE_EXCHANGE)) {
return false;
}
AbstractLogicalOperator op2 = (AbstractLogicalOperator) op1.getInputs().get(0).getValue();
@@ -51,6 +52,12 @@
|| op2.getPhysicalOperator().getOperatorTag() != PhysicalOperatorTag.SORT_MERGE_EXCHANGE) {
return false;
}
+ if (op1.getPhysicalOperator().getOperatorTag() == PhysicalOperatorTag.HASH_PARTITION_MERGE_EXCHANGE) {
+ // if it is a hash_partition_merge_exchange, the sort_merge_exchange can be simply removed
+ op1.getInputs().get(0).setValue(op2.getInputs().get(0).getValue());
+ op1.computeDeliveredPhysicalProperties(context);
+ return true;
+ }
HashPartitionExchangePOperator hpe = (HashPartitionExchangePOperator) op1.getPhysicalOperator();
SortMergeExchangePOperator sme = (SortMergeExchangePOperator) op2.getPhysicalOperator();
List<OrderColumn> ocList = new ArrayList<OrderColumn>();