You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@asterixdb.apache.org by AsterixDB Code Review <do...@asterix-gerrit.ics.uci.edu> on 2026/03/10 17:11:04 UTC
[M] Change in asterixdb[phoenix]: [ASTERIXDB-3713][COMP] Improve adding replicate operators
From Preetham Poluparthi <pr...@apache.org>:
Preetham Poluparthi has uploaded this change for review. ( https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/20988?usp=email )
Change subject: [ASTERIXDB-3713][COMP] Improve adding replicate operators
......................................................................
[ASTERIXDB-3713][COMP] Improve adding replicate operators
- user model changes: no
- storage format changes: no
- interface changes: no
Ext-ref: MB-70862
Change-Id: I9c95e889a6a85e6d3e4f6258c75aef8caa09acfd
---
M hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/logical/visitors/OperatorDeepCopyVisitor.java
M hyracks-fullstack/algebricks/algebricks-rewriter/src/main/java/org/apache/hyracks/algebricks/rewriter/rules/ExtractCommonOperatorsRule.java
2 files changed, 136 insertions(+), 19 deletions(-)
git pull ssh://asterix-gerrit.ics.uci.edu:29418/asterixdb refs/changes/88/20988/1
diff --git a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/logical/visitors/OperatorDeepCopyVisitor.java b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/logical/visitors/OperatorDeepCopyVisitor.java
index 8664acf..59cb2b0 100644
--- a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/logical/visitors/OperatorDeepCopyVisitor.java
+++ b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/logical/visitors/OperatorDeepCopyVisitor.java
@@ -438,7 +438,7 @@
@Override
public ILogicalOperator visitWindowOperator(WindowOperator op, Void arg) throws AlgebricksException {
List<Mutable<ILogicalExpression>> newPartitionExprs = new ArrayList<>();
- deepCopyExpressionRefs(op.getPartitionExpressions(), newPartitionExprs);
+ deepCopyExpressionRefs(newPartitionExprs, op.getPartitionExpressions());
List<Pair<IOrder, Mutable<ILogicalExpression>>> newOrderExprs =
deepCopyOrderAndExpression(op.getOrderExpressions());
List<Pair<IOrder, Mutable<ILogicalExpression>>> newFrameValueExprs =
diff --git a/hyracks-fullstack/algebricks/algebricks-rewriter/src/main/java/org/apache/hyracks/algebricks/rewriter/rules/ExtractCommonOperatorsRule.java b/hyracks-fullstack/algebricks/algebricks-rewriter/src/main/java/org/apache/hyracks/algebricks/rewriter/rules/ExtractCommonOperatorsRule.java
index 14a44ac..5d9ad50 100644
--- a/hyracks-fullstack/algebricks/algebricks-rewriter/src/main/java/org/apache/hyracks/algebricks/rewriter/rules/ExtractCommonOperatorsRule.java
+++ b/hyracks-fullstack/algebricks/algebricks-rewriter/src/main/java/org/apache/hyracks/algebricks/rewriter/rules/ExtractCommonOperatorsRule.java
@@ -64,6 +64,11 @@
new HashMap<>();
private final List<Mutable<ILogicalOperator>> roots = new ArrayList<>();
private final List<List<Mutable<ILogicalOperator>>> equivalenceClasses = new ArrayList<>();
+ private Boolean isPruned = false;
+
+ private final List<Mutable<ILogicalOperator>> emptyTuples = new ArrayList<>();
+ private final List<Mutable<ILogicalOperator>> replicates = new ArrayList<>();
+
private final HashMap<Mutable<ILogicalOperator>, BitSet> opToCandidateInputs = new HashMap<>();
private final HashMap<Mutable<ILogicalOperator>, MutableInt> clusterMap = new HashMap<>();
private final HashMap<Integer, BitSet> clusterWaitForMap = new HashMap<>();
@@ -101,6 +106,7 @@
changed = false;
// applying the rewriting until fixpoint
topDownMaterialization(roots);
+ makeEquivalenceClass();
genCandidates();
removeTrivialShare();
if (!equivalenceClasses.isEmpty()) {
@@ -110,6 +116,8 @@
rewritten = changed;
}
equivalenceClasses.clear();
+ emptyTuples.clear();
+ replicates.clear();
childrenToParents.clear();
opToCandidateInputs.clear();
resetPlanClusterState();
@@ -119,6 +127,21 @@
return rewritten;
}
+ private boolean isTrivialShare(Mutable<ILogicalOperator> candidate) {
+
+ AbstractLogicalOperator aop = (AbstractLogicalOperator) candidate.getValue();
+ if (aop.getOperatorTag() == LogicalOperatorTag.EXCHANGE) {
+ aop = (AbstractLogicalOperator) aop.getInputs().get(0).getValue();
+ } else {
+ return false;
+ }
+ if (aop.getOperatorTag() == LogicalOperatorTag.EMPTYTUPLESOURCE
+ || aop.getOperatorTag() == LogicalOperatorTag.REPLICATE) {
+ return true;
+ }
+ return false;
+ }
+
private void removeTrivialShare() {
for (List<Mutable<ILogicalOperator>> candidates : equivalenceClasses) {
for (int i = candidates.size() - 1; i >= 0; i--) {
@@ -414,14 +437,20 @@
private void genCandidates() throws AlgebricksException {
List<List<Mutable<ILogicalOperator>>> previousEquivalenceClasses = new ArrayList<>();
- while (equivalenceClasses.size() > 0) {
+ boolean grown = true;
+ while (equivalenceClasses.size() > 0 && grown) {
previousEquivalenceClasses.clear();
for (List<Mutable<ILogicalOperator>> candidates : equivalenceClasses) {
List<Mutable<ILogicalOperator>> candidatesCopy = new ArrayList<>(candidates);
previousEquivalenceClasses.add(candidatesCopy);
}
+
+ List<List<Mutable<ILogicalOperator>>> requiredEquivalenceClasses = new ArrayList<>();
+ List<List<Mutable<ILogicalOperator>>> changedEquivalenceClasses = new ArrayList<>();
+
List<Mutable<ILogicalOperator>> currentLevelOpRefs = new ArrayList<>();
- for (List<Mutable<ILogicalOperator>> candidates : equivalenceClasses) {
+ grown = false;
+ for (List<Mutable<ILogicalOperator>> candidates : previousEquivalenceClasses) {
if (candidates.size() > 0) {
for (Mutable<ILogicalOperator> opRef : candidates) {
List<Mutable<ILogicalOperator>> refs = childrenToParents.get(opRef);
@@ -433,21 +462,67 @@
if (currentLevelOpRefs.size() == 0) {
continue;
}
- candidatesGrow(currentLevelOpRefs, candidates);
+ List<Mutable<ILogicalOperator>> candidatesCopy = new ArrayList<>(candidates);
+ boolean isGrown = candidatesGrow(currentLevelOpRefs, candidates);
+ grown |= isGrown;
+
+ if (!isGrown) {
+ changedEquivalenceClasses.add(candidates);
+ continue;
+ }
+
+ List<List<Mutable<ILogicalOperator>>> newEquivalenceClasses = prune(candidates);
+ if (newEquivalenceClasses.isEmpty()) {
+ continue;
+ }
+ if (isPruned) {
+ requiredEquivalenceClasses.add(candidatesCopy);
+ }
+
+ changedEquivalenceClasses.addAll(newEquivalenceClasses);
}
if (currentLevelOpRefs.size() == 0) {
break;
}
- prune();
+
+ if (!requiredEquivalenceClasses.isEmpty() && !isTrivialShare(requiredEquivalenceClasses.get(0).get(0))) {
+ equivalenceClasses.clear();
+ equivalenceClasses.addAll(requiredEquivalenceClasses);
+ return;
+ } else {
+ equivalenceClasses.clear();
+ equivalenceClasses.addAll(changedEquivalenceClasses);
+ }
+
+ for (int i = equivalenceClasses.size() - 1; i >= 0; i--) {
+ for (int j = i - 1; j >= 0; j--) {
+ if (equivalenceClasses.get(i).equals(equivalenceClasses.get(j))) {
+ equivalenceClasses.remove(i);
+ break;
+ }
+ }
+ }
+
+ // prune();
}
- if (equivalenceClasses.size() < 1 && previousEquivalenceClasses.size() > 0) {
- equivalenceClasses.addAll(previousEquivalenceClasses);
- prune();
+ // if (equivalenceClasses.size() < 1 && previousEquivalenceClasses.size() > 0) {
+ // equivalenceClasses.addAll(previousEquivalenceClasses);
+ // prune();
+ // }
+ }
+
+ private void makeEquivalenceClass() {
+ if (!emptyTuples.isEmpty()) {
+ equivalenceClasses.add(emptyTuples);
+ }
+ if (!replicates.isEmpty()) {
+ for (Mutable<ILogicalOperator> opRef : replicates) {
+ equivalenceClasses.add(Collections.singletonList(opRef));
+ }
}
}
private void topDownMaterialization(List<Mutable<ILogicalOperator>> tops) {
- List<Mutable<ILogicalOperator>> candidates = new ArrayList<>();
List<Mutable<ILogicalOperator>> nextLevel = new ArrayList<>();
for (Mutable<ILogicalOperator> op : tops) {
for (Mutable<ILogicalOperator> opRef : op.getValue().getInputs()) {
@@ -455,33 +530,31 @@
if (opRefList == null) {
opRefList = new ArrayList<>();
childrenToParents.put(opRef, opRefList);
- nextLevel.add(opRef);
+ if (opRef.get().getOperatorTag() != LogicalOperatorTag.REPLICATE) {
+ nextLevel.add(opRef);
+ } else {
+ replicates.add(opRef);
+ }
}
opRefList.add(op);
}
if (op.getValue().getInputs().isEmpty()) {
- candidates.add(op);
+ emptyTuples.add(op);
}
}
- if (!equivalenceClasses.isEmpty()) {
- equivalenceClasses.get(0).addAll(candidates);
- } else {
- equivalenceClasses.add(candidates);
- }
if (!nextLevel.isEmpty()) {
topDownMaterialization(nextLevel);
}
}
- private void candidatesGrow(List<Mutable<ILogicalOperator>> opList, List<Mutable<ILogicalOperator>> candidates) {
+ private boolean candidatesGrow(List<Mutable<ILogicalOperator>> opList, List<Mutable<ILogicalOperator>> candidates) {
List<Mutable<ILogicalOperator>> previousCandidates = new ArrayList<>(candidates);
candidates.clear();
- boolean validCandidate = false;
for (Mutable<ILogicalOperator> op : opList) {
List<Mutable<ILogicalOperator>> inputs = op.getValue().getInputs();
+ boolean validCandidate = false;
for (int i = 0; i < inputs.size(); i++) {
Mutable<ILogicalOperator> inputRef = inputs.get(i);
- validCandidate = false;
for (Mutable<ILogicalOperator> candidate : previousCandidates) {
// if current input is in candidates
if (inputRef.getValue().equals(candidate.getValue())) {
@@ -509,6 +582,50 @@
candidates.add(op);
}
}
+ if (candidates.isEmpty()) {
+ candidates.addAll(previousCandidates);
+ return false;
+ }
+ return true;
+ }
+
+ private List<List<Mutable<ILogicalOperator>>> prune(List<Mutable<ILogicalOperator>> candidates)
+ throws AlgebricksException {
+ isPruned = false;
+ List<List<Mutable<ILogicalOperator>>> result = new ArrayList<>();
+ boolean[] reserved = new boolean[candidates.size()];
+ for (int i = candidates.size() - 1; i >= 0; i--) {
+ if (!reserved[i]) {
+ List<Mutable<ILogicalOperator>> equivalentClass = new ArrayList<>();
+ List<Integer> indexes = new ArrayList<>();
+ ILogicalOperator candidate = candidates.get(i).getValue();
+ equivalentClass.add(candidates.get(i));
+ indexes.add(i);
+ for (int j = i - 1; j >= 0; j--) {
+ ILogicalOperator peer = candidates.get(j).getValue();
+ boolean isomorphic = candidate.getInputs().size() > 1
+ ? IsomorphismUtilities.isOperatorIsomorphicPlanSegment(candidate, peer)
+ : IsomorphismUtilities.isOperatorIsomorphic(candidate, peer);
+ if (isomorphic) {
+ reserved[i] = true;
+ reserved[j] = true;
+ equivalentClass.add(candidates.get(j));
+ indexes.add(j);
+ } else {
+ isPruned = true;
+ }
+ }
+
+ if (equivalentClass.size() == 1) {
+ continue;
+ }
+
+ result.add(equivalentClass);
+ Collections.reverse(indexes);
+ Collections.reverse(equivalentClass);
+ }
+ }
+ return result;
}
private void prune() throws AlgebricksException {
--
To view, visit https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/20988?usp=email
To unsubscribe, or for help writing mail filters, visit https://asterix-gerrit.ics.uci.edu/settings?usp=email
Gerrit-MessageType: newchange
Gerrit-Project: asterixdb
Gerrit-Branch: phoenix
Gerrit-Change-Id: I9c95e889a6a85e6d3e4f6258c75aef8caa09acfd
Gerrit-Change-Number: 20988
Gerrit-PatchSet: 1
Gerrit-Owner: Preetham Poluparthi <pr...@apache.org>