ngsg commented on code in PR #3998:
URL: https://github.com/apache/hive/pull/3998#discussion_r1959290514
##########
ql/src/java/org/apache/hadoop/hive/ql/optimizer/graph/OperatorGraph.java:
##########
@@ -81,183 +76,202 @@ public EdgeType getEdgeType() {
}
- public static interface OperatorEdgePredicate {
+ public interface OperatorEdgePredicate {
boolean accept(Operator<?> s, Operator<?> t, OpEdge opEdge);
}
- Map<Operator<?>, Cluster> nodeCluster = new HashMap<>();
-
public class Cluster {
- Set<Operator<?>> members = new LinkedHashSet<>();
+ private final Set<Operator<?>> members = new HashSet<>();
- protected void merge(Cluster o) {
- if (o == this) {
- return;
- }
- for (Operator<?> node : o.members) {
- add(node);
- }
- o.members.clear();
+ public void add(Operator<?> operator) {
+ members.add(operator);
}
- protected void add(Operator<?> curr) {
- nodeCluster.put(curr, this);
- members.add(curr);
+ public void merge(Cluster cluster) {
+ members.addAll(cluster.getMembers());
}
public Set<Cluster> parentClusters(OperatorEdgePredicate traverseEdge) {
Set<Cluster> ret = new HashSet<Cluster>();
for (Operator<?> operator : members) {
- for (Operator<? extends OperatorDesc> p :
operator.getParentOperators()) {
- if (members.contains(p)) {
- continue;
- }
- Optional<OpEdge> e = g.getEdge(p, operator);
- if (traverseEdge.accept(p, operator, e.get())) {
- ret.add(nodeCluster.get(p));
+ Stream<Operator<?>> foreignParentOperators =
+ dagGraph.predecessors(operator).stream()
+ .filter(pOp -> !members.contains(pOp))
+ .filter(pOp -> traverseEdge.accept(pOp, operator,
dagGraph.getEdge(pOp, operator).get()));
+ foreignParentOperators.forEach(parentOperator -> {
+ for (Cluster parentCluster: operatorToCluster.get(parentOperator)) {
+ if (!parentCluster.getMembers().contains(operator)) {
+ ret.add(parentCluster);
+ }
}
- }
+ });
}
return ret;
}
public Set<Cluster> childClusters(OperatorEdgePredicate traverseEdge) {
Set<Cluster> ret = new HashSet<Cluster>();
for (Operator<?> operator : members) {
- for (Operator<? extends OperatorDesc> p :
operator.getChildOperators()) {
- if (members.contains(p)) {
- continue;
- }
- Optional<OpEdge> e = g.getEdge(operator, p);
- if (traverseEdge.accept(operator, p, e.get())) {
- ret.add(nodeCluster.get(p));
+ Stream<Operator<?>> foreignChildOperators =
+ dagGraph.successors(operator).stream()
+ .filter(cOp -> !members.contains(cOp))
+ .filter(cOp -> traverseEdge.accept(operator, cOp,
dagGraph.getEdge(operator, cOp).get()));
+ foreignChildOperators.forEach(childOperator -> {
+ for (Cluster childCluster: operatorToCluster.get(childOperator)) {
+ if (!childCluster.getMembers().contains(operator)) {
+ ret.add(childCluster);
+ }
}
- }
+ });
}
return ret;
}
public Set<Operator<?>> getMembers() {
return Collections.unmodifiableSet(members);
}
+
}
+ private Cluster createCluster(Operator<?> rootOperator) {
+ Cluster cluster = new Cluster();
+ Queue<Operator<?>> remainingOperators = new LinkedList<>();
+ remainingOperators.add(rootOperator);
- public OperatorGraph(ParseContext pctx) {
- g = new DagGraph<Operator<?>, OperatorGraph.OpEdge>();
- Set<Operator<?>> visited = Sets.newIdentityHashSet();
- Set<Operator<?>> seen = Sets.newIdentityHashSet();
-
- seen.addAll(pctx.getTopOps().values());
- while (!seen.isEmpty()) {
- Operator<?> curr = seen.iterator().next();
- seen.remove(curr);
- if (visited.contains(curr)) {
- continue;
+ while (!remainingOperators.isEmpty()) {
+ Operator<?> currentOperator = remainingOperators.poll();
+ if (!cluster.getMembers().contains(currentOperator)) {
+ cluster.add(currentOperator);
+
+ // TODO: write about DummyStoreOperator and FileSinkOperator
+ if (!(currentOperator instanceof ReduceSinkOperator)) {
+ remainingOperators.addAll(currentOperator.getChildOperators());
+ }
}
+ }
- visited.add(curr);
+ return cluster;
+ }
- Cluster currentCluster = nodeCluster.get(curr);
- if (currentCluster == null) {
- currentCluster=new Cluster();
- currentCluster.add(curr);
+ private Set<Cluster> createClusterSet(ParseContext pctx) {
Review Comment:
You're right.
The original implementation assumes that every operator belongs to a
single-merged cluster, so it uses `nodeCluster: Map[Operator, Cluster]`.
However, the assumption is incorrect.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscr...@hive.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscr...@hive.apache.org
For additional commands, e-mail: gitbox-h...@hive.apache.org
