jihoonson commented on a change in pull request #10082:
URL: https://github.com/apache/druid/pull/10082#discussion_r448135388
##########
File path:
processing/src/main/java/org/apache/druid/query/context/ResponseContext.java
##########
@@ -112,6 +114,30 @@
"uncoveredIntervalsOverflowed",
(oldValue, newValue) -> (boolean) oldValue || (boolean) newValue
),
+ /**
+ * Map of most relevant query ID to remaining number of responses from
query nodes.
+ * The value is initialized in {@code CachingClusteredClient} when it
initializes the connection to the query nodes,
+ * and is updated whenever they respond (@code DirectDruidClient). {@code
RetryQueryRunner} uses this value to
+ * check if the {@link #MISSING_SEGMENTS} is valid.
+ *
+ * Currently, the broker doesn't run subqueries in parallel, the remaining
number of responses will be updated
+ * one by one per subquery. However, since we are planning to parallelize
running subqueries, we store them
Review comment:
> I'm not sure if we are, but, we also haven't decided definitely _not_
to.
Heh, rephrased to sound more chill.
> I think it would also be okay to not handle parallel queries here, but
instead build in a sanity check that verifies the subqueries are issued in
series. Maybe by verifying that when the id changes, the previous number of
responses remaining must be down to zero. It's up to you. I think your current
code is okay too.
Hmm, how could it be used? Is it to make sure that the subqueries will be
issued in series until we parallelize running them?
##########
File path: server/src/main/java/org/apache/druid/client/DirectDruidClient.java
##########
@@ -231,7 +237,19 @@ private InputStream dequeue() throws InterruptedException
final boolean continueReading;
try {
+ if (query.getId() != null && query.getSubQueryId() != null) {
Review comment:
Changed to always print trace-level log.
##########
File path: server/src/main/java/org/apache/druid/query/RetryQueryRunner.java
##########
@@ -0,0 +1,250 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.query;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
+import org.apache.druid.java.util.common.ISE;
+import org.apache.druid.java.util.common.guava.BaseSequence;
+import org.apache.druid.java.util.common.guava.BaseSequence.IteratorMaker;
+import org.apache.druid.java.util.common.guava.MergeSequence;
+import org.apache.druid.java.util.common.guava.Sequence;
+import org.apache.druid.java.util.common.guava.Yielder;
+import org.apache.druid.java.util.common.guava.YieldingAccumulator;
+import org.apache.druid.java.util.common.guava.YieldingSequenceBase;
+import org.apache.druid.java.util.common.logger.Logger;
+import org.apache.druid.query.context.ResponseContext;
+import org.apache.druid.query.context.ResponseContext.Key;
+import org.apache.druid.segment.SegmentMissingException;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.NoSuchElementException;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.function.BiFunction;
+
+public class RetryQueryRunner<T> implements QueryRunner<T>
+{
+ private static final Logger LOG = new Logger(RetryQueryRunner.class);
+
+ private final QueryRunner<T> baseRunner;
+ private final BiFunction<Query<T>, List<SegmentDescriptor>, QueryRunner<T>>
retryRunnerCreateFn;
+ private final RetryQueryRunnerConfig config;
+ private final ObjectMapper jsonMapper;
+
+ /**
+ * Runnable executed after the broker creates query distribution tree for
the first attempt. This is only
+ * for testing and must not be used in production code.
+ */
+ private final Runnable runnableAfterFirstAttempt;
+
+ private int totalNumRetries;
+
+ public RetryQueryRunner(
+ QueryRunner<T> baseRunner,
+ BiFunction<Query<T>, List<SegmentDescriptor>, QueryRunner<T>>
retryRunnerCreateFn,
+ RetryQueryRunnerConfig config,
+ ObjectMapper jsonMapper
+ )
+ {
+ this(baseRunner, retryRunnerCreateFn, config, jsonMapper, () -> {});
+ }
+
+ /**
+ * Constructor only for testing.
+ */
+ @VisibleForTesting
+ RetryQueryRunner(
+ QueryRunner<T> baseRunner,
+ BiFunction<Query<T>, List<SegmentDescriptor>, QueryRunner<T>>
retryRunnerCreateFn,
+ RetryQueryRunnerConfig config,
+ ObjectMapper jsonMapper,
+ Runnable runnableAfterFirstAttempt
+ )
+ {
+ this.baseRunner = baseRunner;
+ this.retryRunnerCreateFn = retryRunnerCreateFn;
+ this.config = config;
+ this.jsonMapper = jsonMapper;
+ this.runnableAfterFirstAttempt = runnableAfterFirstAttempt;
+ }
+
+ @VisibleForTesting
+ int getTotalNumRetries()
+ {
+ return totalNumRetries;
+ }
+
+ @Override
+ public Sequence<T> run(final QueryPlus<T> queryPlus, final ResponseContext
context)
+ {
+ return new YieldingSequenceBase<T>()
+ {
+ @Override
+ public <OutType> Yielder<OutType> toYielder(OutType initValue,
YieldingAccumulator<OutType, T> accumulator)
+ {
+ final Sequence<Sequence<T>> retryingSequence = new BaseSequence<>(
+ new IteratorMaker<Sequence<T>, RetryingSequenceIterator>()
+ {
+ @Override
+ public RetryingSequenceIterator make()
+ {
+ return new RetryingSequenceIterator(queryPlus, context,
baseRunner, runnableAfterFirstAttempt);
+ }
+
+ @Override
+ public void cleanup(RetryingSequenceIterator iterFromMake)
+ {
+ totalNumRetries = iterFromMake.retryCount;
+ }
+ }
+ );
+ return new MergeSequence<>(queryPlus.getQuery().getResultOrdering(),
retryingSequence)
+ .toYielder(initValue, accumulator);
+ }
+ };
+ }
+
+ private List<SegmentDescriptor> getMissingSegments(QueryPlus<T> queryPlus,
final ResponseContext context)
+ {
+ // Sanity check before retrieving missingSegments from responseContext.
+ // The missingSegments in the responseContext is only valid when all
servers have responded to the broker.
+ // The remainingResponses must be not null but 0 in the responseContext at
this point.
+ final ConcurrentHashMap<String, Integer> idToRemainingResponses =
+ (ConcurrentHashMap<String, Integer>) Preconditions.checkNotNull(
+ context.get(Key.REMAINING_RESPONSES_FROM_QUERY_SERVERS),
+ "%s in responseContext",
+ Key.REMAINING_RESPONSES_FROM_QUERY_SERVERS.getName()
+ );
+
+ final int remainingResponses = Preconditions.checkNotNull(
+ idToRemainingResponses.get(queryPlus.getQuery().getMostSpecificId()),
+ "Number of remaining responses for query[%s]",
+ queryPlus.getQuery().getMostSpecificId()
+ );
+ if (remainingResponses > 0) {
+ throw new ISE("Failed to check missing segments due to missing responds
from [%d] servers", remainingResponses);
Review comment:
Oops thanks. For the comment, I wrote some comments
[here](https://github.com/apache/druid/pull/10082/files#diff-661e59db1678430f72450fd2fa5ce061R133).
Do you think it's not clear?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
