From f3b2b7e97446f9977479c529856867113fe76943 Mon Sep 17 00:00:00 2001 From: Adarsh Sanjeev Date: Thu, 15 Dec 2022 09:35:11 +0530 Subject: [PATCH] Multiple fixes for the MSQ stats merging piece which (#13463) * Add validation checks to worker chat handler apis * Merge things and polishing the error messages. * Minor error message change * Fixing race and adding some tests * Fixing controller fetching stats from wrong workers. Fixing race Changing default mode to Parallel Adding logging. Fixing exceptions not propagated properly. * Changing to kernel worker count * Added a better logic to figure out assigned worker for a stage. * Nits * Moving to existing kernel methods * Adding more coverage Co-authored-by: cryptoe (cherry picked from commit 2b605aa9cf987ccb8e9ace04f71214aa6b508ec4) --- docs/multi-stage-query/reference.md | 2 +- .../apache/druid/msq/exec/ControllerImpl.java | 12 +- .../exec/ExceptionWrappingWorkerClient.java | 14 +- .../org/apache/druid/msq/exec/WorkerImpl.java | 29 +- .../druid/msq/exec/WorkerSketchFetcher.java | 54 +++- .../druid/msq/indexing/WorkerChatHandler.java | 49 +++- .../msq/kernel/worker/WorkerStageKernel.java | 3 +- .../ClusterByStatisticsSnapshot.java | 4 + .../msq/util/MultiStageQueryContext.java | 2 +- .../apache/druid/msq/exec/MSQInsertTest.java | 26 ++ .../apache/druid/msq/exec/WorkerImplTest.java | 54 ++++ .../exec/WorkerSketchFetcherAutoModeTest.java | 39 ++- .../msq/exec/WorkerSketchFetcherTest.java | 37 ++- .../msq/indexing/WorkerChatHandlerTest.java | 254 ++++++++++++++++++ 14 files changed, 529 insertions(+), 50 deletions(-) create mode 100644 extensions-core/multi-stage-query/src/test/java/org/apache/druid/msq/exec/WorkerImplTest.java create mode 100644 extensions-core/multi-stage-query/src/test/java/org/apache/druid/msq/indexing/WorkerChatHandlerTest.java diff --git a/docs/multi-stage-query/reference.md b/docs/multi-stage-query/reference.md index ae9bc106ca8c..86f869924120 100644 --- a/docs/multi-stage-query/reference.md +++ b/docs/multi-stage-query/reference.md @@ -203,7 +203,7 @@ The following table lists the context parameters for the MSQ task engine: | `maxParseExceptions`| SELECT, INSERT, REPLACE

Maximum number of parse exceptions that are ignored while executing the query before it stops with `TooManyWarningsFault`. To ignore all the parse exceptions, set the value to -1.| 0 | | `rowsPerSegment` | INSERT or REPLACE

The number of rows per segment to target. The actual number of rows per segment may be somewhat higher or lower than this number. In most cases, use the default. For general information about sizing rows per segment, see [Segment Size Optimization](../operations/segment-optimization.md). | 3,000,000 | | `indexSpec` | INSERT or REPLACE

An [`indexSpec`](../ingestion/ingestion-spec.md#indexspec) to use when generating segments. May be a JSON string or object. See [Front coding](../ingestion/ingestion-spec.md#front-coding) for details on configuring an `indexSpec` with front coding. | See [`indexSpec`](../ingestion/ingestion-spec.md#indexspec). | -| `clusterStatisticsMergeMode` | Whether to use parallel or sequential mode for merging of the worker sketches. Can be `PARALLEL`, `SEQUENTIAL` or `AUTO`. See [Sketch Merging Mode](#sketch-merging-mode) for more information. | `AUTO` | +| `clusterStatisticsMergeMode` | Whether to use parallel or sequential mode for merging of the worker sketches. Can be `PARALLEL`, `SEQUENTIAL` or `AUTO`. See [Sketch Merging Mode](#sketch-merging-mode) for more information. | `PARALLEL` | ## Sketch Merging Mode This section details the advantages and performance of various Cluster By Statistics Merge Modes. diff --git a/extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/exec/ControllerImpl.java b/extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/exec/ControllerImpl.java index 528baa4c27d1..f36b91e55eb9 100644 --- a/extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/exec/ControllerImpl.java +++ b/extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/exec/ControllerImpl.java @@ -263,6 +263,7 @@ public class ControllerImpl implements Controller // For live reports. Written by the main controller thread, read by HTTP threads. private final ConcurrentHashMap stagePartitionCountsForLiveReports = new ConcurrentHashMap<>(); + private WorkerSketchFetcher workerSketchFetcher; // Time at which the query started. // For live reports. Written by the main controller thread, read by HTTP threads. @@ -624,14 +625,21 @@ public void updatePartialKeyStatisticsInformation(int stageNumber, int workerNum workerSketchFetcher.submitFetcherTask( completeKeyStatisticsInformation, workerTaskIds, - stageDef + stageDef, + queryKernel.getWorkerInputsForStage(stageId).workers() + // we only need tasks which are active for this stage. ); // Add the listener to handle completion. clusterByPartitionsCompletableFuture.whenComplete((clusterByPartitionsEither, throwable) -> { addToKernelManipulationQueue(holder -> { if (throwable != null) { - holder.failStageForReason(stageId, UnknownFault.forException(throwable)); + log.error("Error while fetching stats for stageId[%s]", stageId); + if (throwable instanceof MSQException) { + holder.failStageForReason(stageId, ((MSQException) throwable).getFault()); + } else { + holder.failStageForReason(stageId, UnknownFault.forException(throwable)); + } } else if (clusterByPartitionsEither.isError()) { holder.failStageForReason(stageId, new TooManyPartitionsFault(stageDef.getMaxPartitionCount())); } else { diff --git a/extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/exec/ExceptionWrappingWorkerClient.java b/extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/exec/ExceptionWrappingWorkerClient.java index 3d78b7c9ced4..eb6b1af529e5 100644 --- a/extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/exec/ExceptionWrappingWorkerClient.java +++ b/extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/exec/ExceptionWrappingWorkerClient.java @@ -57,9 +57,13 @@ public ListenableFuture postWorkOrder(String workerTaskId, WorkOrder workO } @Override - public ListenableFuture fetchClusterByStatisticsSnapshot(String workerTaskId, String queryId, int stageNumber) + public ListenableFuture fetchClusterByStatisticsSnapshot( + String workerTaskId, + String queryId, + int stageNumber + ) { - return client.fetchClusterByStatisticsSnapshot(workerTaskId, queryId, stageNumber); + return wrap(workerTaskId, client, c -> c.fetchClusterByStatisticsSnapshot(workerTaskId, queryId, stageNumber)); } @Override @@ -70,7 +74,11 @@ public ListenableFuture fetchClusterByStatisticsSna long timeChunk ) { - return client.fetchClusterByStatisticsSnapshotForTimeChunk(workerTaskId, queryId, stageNumber, timeChunk); + return wrap( + workerTaskId, + client, + c -> c.fetchClusterByStatisticsSnapshotForTimeChunk(workerTaskId, queryId, stageNumber, timeChunk) + ); } @Override diff --git a/extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/exec/WorkerImpl.java b/extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/exec/WorkerImpl.java index 49d6f9080d7c..8c5a782f5368 100644 --- a/extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/exec/WorkerImpl.java +++ b/extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/exec/WorkerImpl.java @@ -571,16 +571,37 @@ public void postFinish() @Override public ClusterByStatisticsSnapshot fetchStatisticsSnapshot(StageId stageId) { - return stageKernelMap.get(stageId).getResultKeyStatisticsSnapshot(); + if (stageKernelMap.get(stageId) == null) { + throw new ISE("Requested statistics snapshot for non-existent stageId %s.", stageId); + } else if (stageKernelMap.get(stageId).getResultKeyStatisticsSnapshot() == null) { + throw new ISE( + "Requested statistics snapshot is not generated yet for stageId[%s]", + stageId + ); + } else { + return stageKernelMap.get(stageId).getResultKeyStatisticsSnapshot(); + } } @Override public ClusterByStatisticsSnapshot fetchStatisticsSnapshotForTimeChunk(StageId stageId, long timeChunk) { - ClusterByStatisticsSnapshot snapshot = stageKernelMap.get(stageId).getResultKeyStatisticsSnapshot(); - return snapshot.getSnapshotForTimeChunk(timeChunk); + if (stageKernelMap.get(stageId) == null) { + throw new ISE("Requested statistics snapshot for non-existent stageId[%s].", stageId); + } else if (stageKernelMap.get(stageId).getResultKeyStatisticsSnapshot() == null) { + throw new ISE( + "Requested statistics snapshot is not generated yet for stageId[%s]", + stageId + ); + } else { + return stageKernelMap.get(stageId) + .getResultKeyStatisticsSnapshot() + .getSnapshotForTimeChunk(timeChunk); + } + } + @Override public CounterSnapshotsTree getCounters() { @@ -643,7 +664,7 @@ private OutputChannelFactory makeStageOutputChannelFactory(final FrameContext fr /** * Decorates the server-wide {@link QueryProcessingPool} such that any Callables and Runnables, not just * {@link PrioritizedCallable} and {@link PrioritizedRunnable}, may be added to it. - * + *

* In production, the underlying {@link QueryProcessingPool} pool is set up by * {@link org.apache.druid.guice.DruidProcessingModule}. */ diff --git a/extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/exec/WorkerSketchFetcher.java b/extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/exec/WorkerSketchFetcher.java index dc6f21990587..2eba0c409d28 100644 --- a/extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/exec/WorkerSketchFetcher.java +++ b/extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/exec/WorkerSketchFetcher.java @@ -20,6 +20,7 @@ package org.apache.druid.msq.exec; import com.google.common.util.concurrent.ListenableFuture; +import it.unimi.dsi.fastutil.ints.IntSet; import org.apache.druid.frame.key.ClusterBy; import org.apache.druid.frame.key.ClusterByPartition; import org.apache.druid.frame.key.ClusterByPartitions; @@ -40,7 +41,7 @@ import java.util.Set; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutorService; -import java.util.stream.IntStream; +import java.util.stream.Collectors; /** * Queues up fetching sketches from workers and progressively generates partitions boundaries. @@ -78,7 +79,8 @@ public WorkerSketchFetcher( public CompletableFuture> submitFetcherTask( CompleteKeyStatisticsInformation completeKeyStatisticsInformation, List workerTaskIds, - StageDefinition stageDefinition + StageDefinition stageDefinition, + IntSet workersForStage ) { ClusterBy clusterBy = stageDefinition.getClusterBy(); @@ -87,18 +89,31 @@ public CompletableFuture> submitFetcherTask( case SEQUENTIAL: return sequentialTimeChunkMerging(completeKeyStatisticsInformation, stageDefinition, workerTaskIds); case PARALLEL: - return inMemoryFullSketchMerging(stageDefinition, workerTaskIds); + return inMemoryFullSketchMerging(stageDefinition, workerTaskIds, workersForStage); case AUTO: if (clusterBy.getBucketByCount() == 0) { - log.info("Query [%s] AUTO mode: chose PARALLEL mode to merge key statistics", stageDefinition.getId().getQueryId()); + log.info( + "Query[%s] stage[%d] for AUTO mode: chose PARALLEL mode to merge key statistics", + stageDefinition.getId().getQueryId(), + stageDefinition.getStageNumber() + ); // If there is no time clustering, there is no scope for sequential merge - return inMemoryFullSketchMerging(stageDefinition, workerTaskIds); - } else if (stageDefinition.getMaxWorkerCount() > WORKER_THRESHOLD || completeKeyStatisticsInformation.getBytesRetained() > BYTES_THRESHOLD) { - log.info("Query [%s] AUTO mode: chose SEQUENTIAL mode to merge key statistics", stageDefinition.getId().getQueryId()); + return inMemoryFullSketchMerging(stageDefinition, workerTaskIds, workersForStage); + } else if (stageDefinition.getMaxWorkerCount() > WORKER_THRESHOLD + || completeKeyStatisticsInformation.getBytesRetained() > BYTES_THRESHOLD) { + log.info( + "Query[%s] stage[%d] for AUTO mode: chose SEQUENTIAL mode to merge key statistics", + stageDefinition.getId().getQueryId(), + stageDefinition.getStageNumber() + ); return sequentialTimeChunkMerging(completeKeyStatisticsInformation, stageDefinition, workerTaskIds); } - log.info("Query [%s] AUTO mode: chose PARALLEL mode to merge key statistics", stageDefinition.getId().getQueryId()); - return inMemoryFullSketchMerging(stageDefinition, workerTaskIds); + log.info( + "Query[%s] stage[%d] for AUTO mode: chose PARALLEL mode to merge key statistics", + stageDefinition.getId().getQueryId(), + stageDefinition.getStageNumber() + ); + return inMemoryFullSketchMerging(stageDefinition, workerTaskIds, workersForStage); default: throw new IllegalStateException("No fetching strategy found for mode: " + clusterStatisticsMergeMode); } @@ -111,7 +126,8 @@ public CompletableFuture> submitFetcherTask( */ CompletableFuture> inMemoryFullSketchMerging( StageDefinition stageDefinition, - List workerTaskIds + List workerTaskIds, + IntSet workersForStage ) { CompletableFuture> partitionFuture = new CompletableFuture<>(); @@ -119,12 +135,19 @@ CompletableFuture> inMemoryFullSketchMerging( // Create a new key statistics collector to merge worker sketches into final ClusterByStatisticsCollector mergedStatisticsCollector = stageDefinition.createResultKeyStatisticsCollector(statisticsMaxRetainedBytes); - final int workerCount = workerTaskIds.size(); + final int workerCount = workersForStage.size(); // Guarded by synchronized mergedStatisticsCollector final Set finishedWorkers = new HashSet<>(); + log.info( + "Fetching stats using %s for stage[%d] for workers[%s] ", + ClusterStatisticsMergeMode.PARALLEL, + stageDefinition.getStageNumber(), + workersForStage.stream().map(Object::toString).collect(Collectors.joining(",")) + ); + // Submit a task for each worker to fetch statistics - IntStream.range(0, workerCount).forEach(workerNo -> { + workersForStage.forEach(workerNo -> { executorService.submit(() -> { ListenableFuture snapshotFuture = workerClient.fetchClusterByStatisticsSnapshot( @@ -177,6 +200,13 @@ CompletableFuture> sequentialTimeChunkMerging( workerTaskIds, completeKeyStatisticsInformation.getTimeSegmentVsWorkerMap().entrySet().iterator() ); + + log.info( + "Fetching stats using %s for stage[%d] for tasks[%s]", + ClusterStatisticsMergeMode.SEQUENTIAL, + stageDefinition.getStageNumber(), + String.join("", workerTaskIds) + ); sequentialFetchStage.submitFetchingTasksForNextTimeChunk(); return sequentialFetchStage.getPartitionFuture(); } diff --git a/extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/indexing/WorkerChatHandler.java b/extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/indexing/WorkerChatHandler.java index dd6ea7cb7124..3eae3b05ccf4 100644 --- a/extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/indexing/WorkerChatHandler.java +++ b/extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/indexing/WorkerChatHandler.java @@ -19,11 +19,13 @@ package org.apache.druid.msq.indexing; +import com.google.common.collect.ImmutableMap; import it.unimi.dsi.fastutil.bytes.ByteArrays; import org.apache.commons.lang.mutable.MutableLong; import org.apache.druid.frame.file.FrameFileHttpResponseHandler; import org.apache.druid.frame.key.ClusterByPartitions; import org.apache.druid.indexing.common.TaskToolbox; +import org.apache.druid.java.util.common.StringUtils; import org.apache.druid.java.util.common.logger.Logger; import org.apache.druid.msq.exec.Worker; import org.apache.druid.msq.kernel.StageId; @@ -71,7 +73,7 @@ public WorkerChatHandler(TaskToolbox toolbox, Worker worker) /** * Returns up to {@link #CHANNEL_DATA_CHUNK_SIZE} bytes of stage output data. - * + *

* See {@link org.apache.druid.msq.exec.WorkerClient#fetchChannelData} for the client-side code that calls this API. */ @GET @@ -193,17 +195,30 @@ public Response httpFetchKeyStatistics( ChatHandlers.authorizationCheck(req, Action.READ, task.getDataSource(), toolbox.getAuthorizerMapper()); ClusterByStatisticsSnapshot clusterByStatisticsSnapshot; StageId stageId = new StageId(queryId, stageNumber); - clusterByStatisticsSnapshot = worker.fetchStatisticsSnapshot(stageId); - return Response.status(Response.Status.ACCEPTED) - .entity(clusterByStatisticsSnapshot) - .build(); + try { + clusterByStatisticsSnapshot = worker.fetchStatisticsSnapshot(stageId); + return Response.status(Response.Status.ACCEPTED) + .entity(clusterByStatisticsSnapshot) + .build(); + } + catch (Exception e) { + String errorMessage = StringUtils.format( + "Invalid request for key statistics for query[%s] and stage[%d]", + queryId, + stageNumber + ); + log.error(e, errorMessage); + return Response.status(Response.Status.BAD_REQUEST) + .entity(ImmutableMap.of("error", errorMessage)) + .build(); + } } @POST @Path("/keyStatisticsForTimeChunk/{queryId}/{stageNumber}/{timeChunk}") @Produces(MediaType.APPLICATION_JSON) @Consumes(MediaType.APPLICATION_JSON) - public Response httpSketch( + public Response httpFetchKeyStatisticsWithSnapshot( @PathParam("queryId") final String queryId, @PathParam("stageNumber") final int stageNumber, @PathParam("timeChunk") final long timeChunk, @@ -213,10 +228,24 @@ public Response httpSketch( ChatHandlers.authorizationCheck(req, Action.READ, task.getDataSource(), toolbox.getAuthorizerMapper()); ClusterByStatisticsSnapshot snapshotForTimeChunk; StageId stageId = new StageId(queryId, stageNumber); - snapshotForTimeChunk = worker.fetchStatisticsSnapshotForTimeChunk(stageId, timeChunk); - return Response.status(Response.Status.ACCEPTED) - .entity(snapshotForTimeChunk) - .build(); + try { + snapshotForTimeChunk = worker.fetchStatisticsSnapshotForTimeChunk(stageId, timeChunk); + return Response.status(Response.Status.ACCEPTED) + .entity(snapshotForTimeChunk) + .build(); + } + catch (Exception e) { + String errorMessage = StringUtils.format( + "Invalid request for key statistics for query[%s], stage[%d] and timeChunk[%d]", + queryId, + stageNumber, + timeChunk + ); + log.error(e, errorMessage); + return Response.status(Response.Status.BAD_REQUEST) + .entity(ImmutableMap.of("error", errorMessage)) + .build(); + } } /** diff --git a/extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/kernel/worker/WorkerStageKernel.java b/extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/kernel/worker/WorkerStageKernel.java index b0ed8e5c19da..00a49656be48 100644 --- a/extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/kernel/worker/WorkerStageKernel.java +++ b/extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/kernel/worker/WorkerStageKernel.java @@ -48,8 +48,9 @@ public class WorkerStageKernel private WorkerStagePhase phase = WorkerStagePhase.NEW; + // We read this variable in the main thread and the netty threads @Nullable - private ClusterByStatisticsSnapshot resultKeyStatisticsSnapshot; + private volatile ClusterByStatisticsSnapshot resultKeyStatisticsSnapshot; @Nullable private ClusterByPartitions resultPartitionBoundaries; diff --git a/extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/statistics/ClusterByStatisticsSnapshot.java b/extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/statistics/ClusterByStatisticsSnapshot.java index e54253ad2186..16a4c1656b06 100644 --- a/extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/statistics/ClusterByStatisticsSnapshot.java +++ b/extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/statistics/ClusterByStatisticsSnapshot.java @@ -25,6 +25,7 @@ import com.google.common.base.Preconditions; import com.google.common.collect.ImmutableMap; import org.apache.druid.frame.key.RowKey; +import org.apache.druid.java.util.common.ISE; import javax.annotation.Nullable; import java.util.Collections; @@ -61,6 +62,9 @@ Map getBuckets() public ClusterByStatisticsSnapshot getSnapshotForTimeChunk(long timeChunk) { Bucket bucket = buckets.get(timeChunk); + if (bucket == null) { + throw new ISE("ClusterByStatistics not present for requested timechunk %s", timeChunk); + } return new ClusterByStatisticsSnapshot(ImmutableMap.of(timeChunk, bucket), null); } diff --git a/extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/util/MultiStageQueryContext.java b/extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/util/MultiStageQueryContext.java index 7c589f2326f1..3dc622870f43 100644 --- a/extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/util/MultiStageQueryContext.java +++ b/extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/util/MultiStageQueryContext.java @@ -60,7 +60,7 @@ public class MultiStageQueryContext public static final String CTX_ENABLE_DURABLE_SHUFFLE_STORAGE = "durableShuffleStorage"; public static final String CTX_CLUSTER_STATISTICS_MERGE_MODE = "clusterStatisticsMergeMode"; - public static final String DEFAULT_CLUSTER_STATISTICS_MERGE_MODE = ClusterStatisticsMergeMode.AUTO.toString(); + public static final String DEFAULT_CLUSTER_STATISTICS_MERGE_MODE = ClusterStatisticsMergeMode.PARALLEL.toString(); private static final boolean DEFAULT_ENABLE_DURABLE_SHUFFLE_STORAGE = false; public static final String CTX_DESTINATION = "destination"; diff --git a/extensions-core/multi-stage-query/src/test/java/org/apache/druid/msq/exec/MSQInsertTest.java b/extensions-core/multi-stage-query/src/test/java/org/apache/druid/msq/exec/MSQInsertTest.java index 64b9a8067240..2f2f1d8f6be3 100644 --- a/extensions-core/multi-stage-query/src/test/java/org/apache/druid/msq/exec/MSQInsertTest.java +++ b/extensions-core/multi-stage-query/src/test/java/org/apache/druid/msq/exec/MSQInsertTest.java @@ -132,6 +132,32 @@ public void testInsertOnFoo1WithTimeFunction() } + @Test + public void testInsertOnFoo1WithTimeFunctionWithSequential() + { + RowSignature rowSignature = RowSignature.builder() + .add("__time", ColumnType.LONG) + .add("dim1", ColumnType.STRING) + .add("cnt", ColumnType.LONG).build(); + Map context = ImmutableMap.builder() + .putAll(DEFAULT_MSQ_CONTEXT) + .put( + MultiStageQueryContext.CTX_CLUSTER_STATISTICS_MERGE_MODE, + ClusterStatisticsMergeMode.SEQUENTIAL.toString() + ) + .build(); + + testIngestQuery().setSql( + "insert into foo1 select floor(__time to day) as __time , dim1 , count(*) as cnt from foo where dim1 is not null group by 1, 2 PARTITIONED by day clustered by dim1") + .setQueryContext(context) + .setExpectedDataSource("foo1") + .setExpectedRowSignature(rowSignature) + .setExpectedSegment(expectedFooSegments()) + .setExpectedResultRows(expectedFooRows()) + .verifyResults(); + + } + @Test public void testInsertOnFoo1WithMultiValueDim() { diff --git a/extensions-core/multi-stage-query/src/test/java/org/apache/druid/msq/exec/WorkerImplTest.java b/extensions-core/multi-stage-query/src/test/java/org/apache/druid/msq/exec/WorkerImplTest.java new file mode 100644 index 000000000000..52231a116b6a --- /dev/null +++ b/extensions-core/multi-stage-query/src/test/java/org/apache/druid/msq/exec/WorkerImplTest.java @@ -0,0 +1,54 @@ +/* + * 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.msq.exec; + +import org.apache.druid.java.util.common.ISE; +import org.apache.druid.msq.indexing.MSQWorkerTask; +import org.apache.druid.msq.kernel.StageId; +import org.junit.Assert; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.Mock; +import org.mockito.junit.MockitoJUnitRunner; + +import java.util.HashMap; + + +@RunWith(MockitoJUnitRunner.class) +public class WorkerImplTest +{ + @Mock + WorkerContext workerContext; + + @Test + public void testFetchStatsThrows() + { + WorkerImpl worker = new WorkerImpl(new MSQWorkerTask("controller", "ds", 1, new HashMap<>()), workerContext); + Assert.assertThrows(ISE.class, () -> worker.fetchStatisticsSnapshot(new StageId("xx", 1))); + } + + @Test + public void testFetchStatsWithTimeChunkThrows() + { + WorkerImpl worker = new WorkerImpl(new MSQWorkerTask("controller", "ds", 1, new HashMap<>()), workerContext); + Assert.assertThrows(ISE.class, () -> worker.fetchStatisticsSnapshotForTimeChunk(new StageId("xx", 1), 1L)); + } + +} diff --git a/extensions-core/multi-stage-query/src/test/java/org/apache/druid/msq/exec/WorkerSketchFetcherAutoModeTest.java b/extensions-core/multi-stage-query/src/test/java/org/apache/druid/msq/exec/WorkerSketchFetcherAutoModeTest.java index 42f6f0437f59..02be2876f98b 100644 --- a/extensions-core/multi-stage-query/src/test/java/org/apache/druid/msq/exec/WorkerSketchFetcherAutoModeTest.java +++ b/extensions-core/multi-stage-query/src/test/java/org/apache/druid/msq/exec/WorkerSketchFetcherAutoModeTest.java @@ -19,6 +19,7 @@ package org.apache.druid.msq.exec; +import it.unimi.dsi.fastutil.ints.IntSet; import org.apache.druid.frame.key.ClusterBy; import org.apache.druid.msq.kernel.StageDefinition; import org.apache.druid.msq.kernel.StageId; @@ -56,7 +57,7 @@ public void setUp() target = spy(new WorkerSketchFetcher(mock(WorkerClient.class), ClusterStatisticsMergeMode.AUTO, 300_000_000)); // Don't actually try to fetch sketches - doReturn(null).when(target).inMemoryFullSketchMerging(any(), any()); + doReturn(null).when(target).inMemoryFullSketchMerging(any(), any(), any()); doReturn(null).when(target).sequentialTimeChunkMerging(any(), any(), any()); doReturn(StageId.fromString("1_1")).when(stageDefinition).getId(); @@ -81,8 +82,13 @@ public void test_submitFetcherTask_belowThresholds_ShouldBeParallel() // Worker count below threshold doReturn(1).when(stageDefinition).getMaxWorkerCount(); - target.submitFetcherTask(completeKeyStatisticsInformation, Collections.emptyList(), stageDefinition); - verify(target, times(1)).inMemoryFullSketchMerging(any(), any()); + target.submitFetcherTask( + completeKeyStatisticsInformation, + Collections.emptyList(), + stageDefinition, + IntSet.of() + ); + verify(target, times(1)).inMemoryFullSketchMerging(any(), any(), any()); verify(target, times(0)).sequentialTimeChunkMerging(any(), any(), any()); } @@ -98,8 +104,13 @@ public void test_submitFetcherTask_workerCountAboveThreshold_shouldBeSequential( // Worker count below threshold doReturn((int) WorkerSketchFetcher.WORKER_THRESHOLD + 1).when(stageDefinition).getMaxWorkerCount(); - target.submitFetcherTask(completeKeyStatisticsInformation, Collections.emptyList(), stageDefinition); - verify(target, times(0)).inMemoryFullSketchMerging(any(), any()); + target.submitFetcherTask( + completeKeyStatisticsInformation, + Collections.emptyList(), + stageDefinition, + IntSet.of() + ); + verify(target, times(0)).inMemoryFullSketchMerging(any(), any(), any()); verify(target, times(1)).sequentialTimeChunkMerging(any(), any(), any()); } @@ -115,8 +126,13 @@ public void test_submitFetcherTask_noClusterByColumns_shouldBeParallel() // Worker count above threshold doReturn((int) WorkerSketchFetcher.WORKER_THRESHOLD + 1).when(stageDefinition).getMaxWorkerCount(); - target.submitFetcherTask(completeKeyStatisticsInformation, Collections.emptyList(), stageDefinition); - verify(target, times(1)).inMemoryFullSketchMerging(any(), any()); + target.submitFetcherTask( + completeKeyStatisticsInformation, + Collections.emptyList(), + stageDefinition, + IntSet.of() + ); + verify(target, times(1)).inMemoryFullSketchMerging(any(), any(), any()); verify(target, times(0)).sequentialTimeChunkMerging(any(), any(), any()); } @@ -132,8 +148,13 @@ public void test_submitFetcherTask_bytesRetainedAboveThreshold_shouldBeSequentia // Worker count below threshold doReturn(1).when(stageDefinition).getMaxWorkerCount(); - target.submitFetcherTask(completeKeyStatisticsInformation, Collections.emptyList(), stageDefinition); - verify(target, times(0)).inMemoryFullSketchMerging(any(), any()); + target.submitFetcherTask( + completeKeyStatisticsInformation, + Collections.emptyList(), + stageDefinition, + IntSet.of() + ); + verify(target, times(0)).inMemoryFullSketchMerging(any(), any(), any()); verify(target, times(1)).sequentialTimeChunkMerging(any(), any(), any()); } } diff --git a/extensions-core/multi-stage-query/src/test/java/org/apache/druid/msq/exec/WorkerSketchFetcherTest.java b/extensions-core/multi-stage-query/src/test/java/org/apache/druid/msq/exec/WorkerSketchFetcherTest.java index 83fb73043bd9..fc2449003614 100644 --- a/extensions-core/multi-stage-query/src/test/java/org/apache/druid/msq/exec/WorkerSketchFetcherTest.java +++ b/extensions-core/multi-stage-query/src/test/java/org/apache/druid/msq/exec/WorkerSketchFetcherTest.java @@ -23,6 +23,8 @@ import com.google.common.collect.ImmutableSet; import com.google.common.collect.ImmutableSortedMap; import com.google.common.util.concurrent.Futures; +import it.unimi.dsi.fastutil.ints.IntAVLTreeSet; +import it.unimi.dsi.fastutil.ints.IntSet; import org.apache.druid.frame.key.ClusterBy; import org.apache.druid.frame.key.ClusterByPartition; import org.apache.druid.frame.key.ClusterByPartitions; @@ -88,11 +90,19 @@ public void setUp() doReturn(clusterBy).when(stageDefinition).getClusterBy(); doReturn(25_000).when(stageDefinition).getMaxPartitionCount(); - expectedPartitions1 = new ClusterByPartitions(ImmutableList.of(new ClusterByPartition(mock(RowKey.class), mock(RowKey.class)))); - expectedPartitions2 = new ClusterByPartitions(ImmutableList.of(new ClusterByPartition(mock(RowKey.class), mock(RowKey.class)))); + expectedPartitions1 = new ClusterByPartitions(ImmutableList.of(new ClusterByPartition( + mock(RowKey.class), + mock(RowKey.class) + ))); + expectedPartitions2 = new ClusterByPartitions(ImmutableList.of(new ClusterByPartition( + mock(RowKey.class), + mock(RowKey.class) + ))); - doReturn(Either.value(expectedPartitions1)).when(stageDefinition).generatePartitionsForShuffle(eq(mergedClusterByStatisticsCollector1)); - doReturn(Either.value(expectedPartitions2)).when(stageDefinition).generatePartitionsForShuffle(eq(mergedClusterByStatisticsCollector2)); + doReturn(Either.value(expectedPartitions1)).when(stageDefinition) + .generatePartitionsForShuffle(eq(mergedClusterByStatisticsCollector1)); + doReturn(Either.value(expectedPartitions2)).when(stageDefinition) + .generatePartitionsForShuffle(eq(mergedClusterByStatisticsCollector2)); doReturn( mergedClusterByStatisticsCollector1, @@ -128,10 +138,14 @@ public void test_submitFetcherTask_parallelFetch_mergePerformedCorrectly() return Futures.immediateFuture(snapshot); }).when(workerClient).fetchClusterByStatisticsSnapshot(any(), any(), anyInt()); + IntSet workersForStage = new IntAVLTreeSet(); + workersForStage.addAll(ImmutableSet.of(0, 1, 2, 3, 4)); + CompletableFuture> eitherCompletableFuture = target.submitFetcherTask( completeKeyStatisticsInformation, workerIds, - stageDefinition + stageDefinition, + workersForStage ); // Assert that the final result is complete and all other sketches returned have been merged. @@ -154,7 +168,12 @@ public void test_submitFetcherTask_sequentialFetch_mergePerformedCorrectly() // Store snapshots in a queue final Queue snapshotQueue = new ConcurrentLinkedQueue<>(); - SortedMap> timeSegmentVsWorkerMap = ImmutableSortedMap.of(1L, ImmutableSet.of(0, 1, 2), 2L, ImmutableSet.of(0, 1, 4)); + SortedMap> timeSegmentVsWorkerMap = ImmutableSortedMap.of( + 1L, + ImmutableSet.of(0, 1, 2), + 2L, + ImmutableSet.of(0, 1, 4) + ); doReturn(timeSegmentVsWorkerMap).when(completeKeyStatisticsInformation).getTimeSegmentVsWorkerMap(); final CyclicBarrier barrier = new CyclicBarrier(3); @@ -168,10 +187,14 @@ public void test_submitFetcherTask_sequentialFetch_mergePerformedCorrectly() return Futures.immediateFuture(snapshot); }).when(workerClient).fetchClusterByStatisticsSnapshotForTimeChunk(any(), any(), anyInt(), anyLong()); + IntSet workersForStage = new IntAVLTreeSet(); + workersForStage.addAll(ImmutableSet.of(0, 1, 2, 3, 4)); + CompletableFuture> eitherCompletableFuture = target.submitFetcherTask( completeKeyStatisticsInformation, ImmutableList.of("0", "1", "2", "3", "4"), - stageDefinition + stageDefinition, + workersForStage ); // Assert that the final result is complete and all other sketches returned have been merged. diff --git a/extensions-core/multi-stage-query/src/test/java/org/apache/druid/msq/indexing/WorkerChatHandlerTest.java b/extensions-core/multi-stage-query/src/test/java/org/apache/druid/msq/indexing/WorkerChatHandlerTest.java new file mode 100644 index 000000000000..5b9d6e497aa0 --- /dev/null +++ b/extensions-core/multi-stage-query/src/test/java/org/apache/druid/msq/indexing/WorkerChatHandlerTest.java @@ -0,0 +1,254 @@ +/* + * 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.msq.indexing; + +import com.fasterxml.jackson.databind.ObjectMapper; +import org.apache.druid.frame.key.ClusterByPartitions; +import org.apache.druid.indexer.TaskStatus; +import org.apache.druid.indexing.common.TaskReport; +import org.apache.druid.indexing.common.TaskReportFileWriter; +import org.apache.druid.indexing.common.TaskToolbox; +import org.apache.druid.jackson.DefaultObjectMapper; +import org.apache.druid.java.util.common.ISE; +import org.apache.druid.msq.counters.CounterSnapshotsTree; +import org.apache.druid.msq.exec.Worker; +import org.apache.druid.msq.kernel.StageId; +import org.apache.druid.msq.kernel.WorkOrder; +import org.apache.druid.msq.statistics.ClusterByStatisticsSnapshot; +import org.apache.druid.segment.IndexIO; +import org.apache.druid.segment.IndexMergerV9; +import org.apache.druid.segment.writeout.OffHeapMemorySegmentWriteOutMediumFactory; +import org.apache.druid.server.security.AuthConfig; +import org.apache.druid.server.security.AuthenticationResult; +import org.apache.druid.sql.calcite.util.CalciteTests; +import org.junit.After; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; +import org.mockito.Mock; +import org.mockito.Mockito; +import org.mockito.MockitoAnnotations; + +import javax.annotation.Nullable; +import javax.servlet.http.HttpServletRequest; +import javax.ws.rs.core.Response; +import java.io.InputStream; +import java.util.HashMap; +import java.util.Map; + +public class WorkerChatHandlerTest +{ + private static final StageId TEST_STAGE = new StageId("123", 0); + @Mock + private HttpServletRequest req; + + private TaskToolbox toolbox; + private AutoCloseable mocks; + + private final TestWorker worker = new TestWorker(); + + @Before + public void setUp() + { + ObjectMapper mapper = new DefaultObjectMapper(); + IndexIO indexIO = new IndexIO(mapper, () -> 0); + IndexMergerV9 indexMerger = new IndexMergerV9( + mapper, + indexIO, + OffHeapMemorySegmentWriteOutMediumFactory.instance() + ); + + mocks = MockitoAnnotations.openMocks(this); + Mockito.when(req.getAttribute(AuthConfig.DRUID_AUTHENTICATION_RESULT)) + .thenReturn(new AuthenticationResult("druid", "druid", null, null)); + TaskToolbox.Builder builder = new TaskToolbox.Builder(); + toolbox = builder.authorizerMapper(CalciteTests.TEST_AUTHORIZER_MAPPER) + .indexIO(indexIO) + .indexMergerV9(indexMerger) + .taskReportFileWriter( + new TaskReportFileWriter() + { + @Override + public void write(String taskId, Map reports) + { + + } + + @Override + public void setObjectMapper(ObjectMapper objectMapper) + { + + } + } + ) + .build(); + } + + @Test + public void testFetchSnapshot() + { + WorkerChatHandler chatHandler = new WorkerChatHandler(toolbox, worker); + Assert.assertEquals( + ClusterByStatisticsSnapshot.empty(), + chatHandler.httpFetchKeyStatistics(TEST_STAGE.getQueryId(), TEST_STAGE.getStageNumber(), req) + .getEntity() + ); + } + + @Test + public void testFetchSnapshot404() + { + WorkerChatHandler chatHandler = new WorkerChatHandler(toolbox, worker); + Assert.assertEquals( + Response.Status.BAD_REQUEST.getStatusCode(), + chatHandler.httpFetchKeyStatistics("123", 2, req) + .getStatus() + ); + } + + @Test + public void testFetchSnapshotWithTimeChunk() + { + WorkerChatHandler chatHandler = new WorkerChatHandler(toolbox, worker); + Assert.assertEquals( + ClusterByStatisticsSnapshot.empty(), + chatHandler.httpFetchKeyStatisticsWithSnapshot(TEST_STAGE.getQueryId(), TEST_STAGE.getStageNumber(), 1, req) + .getEntity() + ); + } + + @Test + public void testFetchSnapshotWithTimeChunk404() + { + WorkerChatHandler chatHandler = new WorkerChatHandler(toolbox, worker); + Assert.assertEquals( + Response.Status.BAD_REQUEST.getStatusCode(), + chatHandler.httpFetchKeyStatisticsWithSnapshot("123", 2, 1, req) + .getStatus() + ); + } + + + private static class TestWorker implements Worker + { + + @Override + public String id() + { + return TEST_STAGE.getQueryId() + "task"; + } + + @Override + public MSQWorkerTask task() + { + return new MSQWorkerTask("controller", "ds", 1, new HashMap<>()); + } + + @Override + public TaskStatus run() + { + return null; + } + + @Override + public void stopGracefully() + { + + } + + @Override + public void controllerFailed() + { + + } + + @Override + public void postWorkOrder(WorkOrder workOrder) + { + + } + + @Override + public ClusterByStatisticsSnapshot fetchStatisticsSnapshot(StageId stageId) + { + if (TEST_STAGE.equals(stageId)) { + return ClusterByStatisticsSnapshot.empty(); + } else { + throw new ISE("stage not found %s", stageId); + } + } + + @Override + public ClusterByStatisticsSnapshot fetchStatisticsSnapshotForTimeChunk(StageId stageId, long timeChunk) + { + if (TEST_STAGE.equals(stageId)) { + return ClusterByStatisticsSnapshot.empty(); + } else { + throw new ISE("stage not found %s", stageId); + } + } + + @Override + public boolean postResultPartitionBoundaries( + ClusterByPartitions stagePartitionBoundaries, + String queryId, + int stageNumber + ) + { + return false; + } + + @Nullable + @Override + public InputStream readChannel(String queryId, int stageNumber, int partitionNumber, long offset) + { + return null; + } + + @Override + public CounterSnapshotsTree getCounters() + { + return null; + } + + @Override + public void postCleanupStage(StageId stageId) + { + + } + + @Override + public void postFinish() + { + + } + } + + @After + public void tearDown() + { + try { + mocks.close(); + } + catch (Exception ignored) { + // ignore tear down exceptions + } + } +}