xiangfu0 commented on code in PR #19944:
URL: https://github.com/apache/pinot/pull/19944#discussion_r4343867923
##########
pinot-query-planner/src/main/java/org/apache/pinot/query/planner/logical/EquivalentStagesReplacer.java:
##########
@@ -55,33 +77,58 @@ public interface OnSubstitution {
}
private static class Replacer extends
PlanNodeVisitor.DepthFirstVisitor<Void, GroupedStages> {
+ private final PlanNode _root;
private final OnSubstitution _listener;
+ private final boolean _guarded;
+ private final List<SpoolSubstitution> _substitutions = new ArrayList<>();
+ /// A stage can be visited more than once (for example the leader of a
group), so a receive is decided only once.
+ private final Set<MailboxReceiveNode> _visitedReceives =
Collections.newSetFromMap(new IdentityHashMap<>());
- public Replacer(OnSubstitution listener) {
+ public Replacer(PlanNode root, OnSubstitution listener, boolean guarded) {
+ _root = root;
_listener = listener;
+ _guarded = guarded;
}
@Override
public Void visitMailboxReceive(MailboxReceiveNode node, GroupedStages
equivalenceGroups) {
+ if (!_visitedReceives.add(node)) {
+ return null;
+ }
MailboxSendNode sender = node.getSender();
MailboxSendNode leader = equivalenceGroups.getGroup(sender).first();
- if (canSubstitute(sender, leader)) {
- // we don't want to visit the children of the node given it is going
to be pruned
- node.setSender(leader);
- leader.addReceiver(node);
- _listener.onSubstitution(node.getStageId(), sender.getStageId(),
leader.getStageId());
- } else {
+ if (sender == leader) {
visitMailboxSend(leader, equivalenceGroups);
Review Comment:
MINOR (perf): under `auto`, the guard re-runs `findRisk` over the whole plan
after every accepted substitution, so this is roughly O(substitutions x
plan-size), and `findRisk` itself builds an O(deferrals^2) matrix where
deferrals can grow with the cartesian product of receives per join/union. For
plans with many equivalent stages that could add noticeable planning latency.
It's gated behind spools being present and only on `auto`, so probably fine in
practice — but worth confirming on a worst-case plan (many shared stages
feeding several joins) before this is the default.
##########
pinot-query-planner/src/main/java/org/apache/pinot/query/planner/logical/EquivalentStagesReplacer.java:
##########
@@ -55,33 +77,58 @@ public interface OnSubstitution {
}
private static class Replacer extends
PlanNodeVisitor.DepthFirstVisitor<Void, GroupedStages> {
+ private final PlanNode _root;
private final OnSubstitution _listener;
+ private final boolean _guarded;
+ private final List<SpoolSubstitution> _substitutions = new ArrayList<>();
+ /// A stage can be visited more than once (for example the leader of a
group), so a receive is decided only once.
+ private final Set<MailboxReceiveNode> _visitedReceives =
Collections.newSetFromMap(new IdentityHashMap<>());
- public Replacer(OnSubstitution listener) {
+ public Replacer(PlanNode root, OnSubstitution listener, boolean guarded) {
+ _root = root;
_listener = listener;
+ _guarded = guarded;
}
@Override
public Void visitMailboxReceive(MailboxReceiveNode node, GroupedStages
equivalenceGroups) {
+ if (!_visitedReceives.add(node)) {
+ return null;
+ }
MailboxSendNode sender = node.getSender();
MailboxSendNode leader = equivalenceGroups.getGroup(sender).first();
- if (canSubstitute(sender, leader)) {
- // we don't want to visit the children of the node given it is going
to be pruned
- node.setSender(leader);
- leader.addReceiver(node);
- _listener.onSubstitution(node.getStageId(), sender.getStageId(),
leader.getStageId());
- } else {
+ if (sender == leader) {
visitMailboxSend(leader, equivalenceGroups);
+ return null;
+ }
+ if (leader.sharesReceiverStages(sender)) {
+ // The leader is already sending to this stage. Given we don't have
the ability to send to multiple
+ // receivers in the same stage, we cannot optimize this case right now.
+ _substitutions.add(new SpoolSubstitution(node.getStageId(),
sender.getStageId(), leader.getStageId(),
+ SpoolSubstitution.Outcome.REJECTED_SAME_RECEIVER_STAGE, null));
+ // The original sender stays in the plan, so the stages below it can
still be shared.
+ visitMailboxSend(sender, equivalenceGroups);
+ return null;
}
+ node.setSender(leader);
+ leader.addReceiver(node);
+ if (_guarded) {
+ SpoolDeadlockGuard.DeadlockRisk risk =
SpoolDeadlockGuard.findRisk(_root);
Review Comment:
MINOR (perf): under `auto`, the guard re-runs `findRisk` over the whole plan
after every accepted substitution, so this is roughly O(substitutions x
plan-size), and `findRisk` itself builds an O(deferrals^2) matrix where the
deferral count grows with the cartesian product of receives per join/union. For
plans with many equivalent stages that could add noticeable planning latency.
It's gated behind spools being present and only on `auto`, so probably fine in
practice — but worth confirming on a worst-case plan (many shared stages
feeding several joins) before this becomes the default.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
