gortiz commented on code in PR #19944:
URL: https://github.com/apache/pinot/pull/19944#discussion_r4466218503
##########
pinot-query-planner/src/main/java/org/apache/pinot/query/QueryEnvironment.java:
##########
@@ -726,12 +728,26 @@ public static ImmutableQueryEnvironment.Config.Builder
configBuilder() {
return ImmutableQueryEnvironment.Config.builder();
}
- public boolean useSpools(Map<String, String> options) {
+ /// Returns the spool mode of the query: the
[CommonConstants.Broker.Request.QueryOptionKey#USE_SPOOLS] query option
+ /// when it is set, or [Config#defaultSpoolMode()] otherwise.
+ ///
+ /// @throws QueryException with [QueryErrorCode#QUERY_VALIDATION] if the
query option is not a valid [SpoolMode].
+ public SpoolMode getSpoolMode(Map<String, String> options) {
String optionValue =
options.get(CommonConstants.Broker.Request.QueryOptionKey.USE_SPOOLS);
if (optionValue == null) {
- return _envConfig.defaultUseSpools();
+ return _envConfig.defaultSpoolMode();
}
- return Boolean.parseBoolean(optionValue);
+ try {
+ return SpoolMode.fromString(optionValue);
+ } catch (IllegalArgumentException e) {
+ throw QueryErrorCode.QUERY_VALIDATION.asException(e.getMessage(), e);
+ }
+ }
+
+ /// Whether the response must describe the spool decisions of the planner.
See
+ /// [CommonConstants.Broker.Request.QueryOptionKey#SPOOLS_FEEDBACK].
+ public static boolean isSpoolsFeedback(Map<String, String> options) {
+ return
Boolean.parseBoolean(options.get(CommonConstants.Broker.Request.QueryOptionKey.SPOOLS_FEEDBACK));
Review Comment:
Duplicate of the other thread on `isSpoolsFeedback`, answered there. Fixed
in b88101fb60.
##########
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:
Good point, so I measured it. I wrote a JMH benchmark that times only the
stage planning (`PinotLogicalQueryPlanner.makePlan`) with spools off, on and
auto. The benchmark is not part of this PR.
On the TPC-H queries from the integration tests, only Q2, Q11 and Q23
produce spools, and auto accepts the same spools as on. On those three queries
the guard adds 7 µs, 9 µs and 84 µs. For the other queries, auto and on are
within noise, because the guard only runs when there is a candidate.
| query | spools | on (µs) | auto (µs) |
|---|---|---|---|
| Q2 | 1 | 33 | 40 |
| Q11 | 2 | 18 | 27 |
| Q23 | 4 | 123 | 207 |
I also tried synthetic worst cases, where one CTE is read up to 32 times.
They found a real problem. UNION created one deferral for each pair of inputs,
so the cost grew roughly as n^5, and planning a 32-way UNION took 82 ms. In
b88101fb60 the guard keeps one deferral per operator, with the waiting and
blocking receives grouped. This is exact, because the old pairs always formed a
full cartesian product. The 32-way UNION now plans in 1.6 ms, and the
32-consumer join shapes in about 1.2 ms.
##########
pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java:
##########
@@ -648,10 +648,19 @@ public static class Broker {
public static final String CONFIG_OF_INFER_PARTITION_HINT =
"pinot.broker.multistage.infer.partition.hint";
public static final boolean DEFAULT_INFER_PARTITION_HINT = false;
- /// Whether to use spools in multistage query engine by default.
+ /// The default spool mode of the multistage query engine: `off`, `auto`
or `on`. `false` and `no` are aliases of
+ /// `off`, and `true` and `yes` are aliases of `on`.
+ ///
+ /// - `off` never shares a stage between consumers.
+ /// - `auto` shares a stage only when the planner can prove that the
sharing cannot block the query.
+ /// - `on` shares every equivalent stage that the planner supports,
without that safety check.
+ ///
+ /// Only the default planner applies spools. With
[Request.QueryOptionKey#USE_PHYSICAL_OPTIMIZER] the mode is
+ /// ignored.
+ ///
/// This value can always be overridden by
[Request.QueryOptionKey#USE_SPOOLS] query option
public static final String CONFIG_OF_SPOOLS =
"pinot.broker.multistage.spools";
- public static final boolean DEFAULT_OF_SPOOLS = false;
+ public static final String DEFAULT_SPOOLS_MODE = "auto";
Review Comment:
Agreed that it deserves a release note, and I will add one. To be precise
about the semantics: with a non-deterministic CTE, both the old and the new
result are valid answers to the query, so this does not break correctness. The
real risk of the new default is a bug in the new path, or a blocking case that
the guard does not detect. That is why the plan is to merge it now, well before
1.7, so it gets a full release cycle of testing. Users can go back to the old
behavior with `pinot.broker.multistage.spools=off`.
--
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]
