Skip to content

Commit

Permalink
Limit pages size to a configurable limit (apache#14994)
Browse files Browse the repository at this point in the history 
Adding the ability to limit the pages sizes of select queries.

    We piggyback on the same machinery that is used to control the numRowsPerSegment.
    This patch introduces a new context parameter rowsPerPage for which the default value is set to 100000 rows.
    This patch also optimizes adding the last selectResults stage only when the previous stages have sorted outputs. Currently for each select query with selectDestination=durableStorage, we used to add this extra selectResults stage.
  • Loading branch information
@cryptoe
cryptoe authored and ycp2 committed Nov 17, 2023
1 parent e3255c4 commit b89b93a
Show file tree
Hide file tree
Showing 13 changed files with 802 additions and 101 deletions.
2 changes: 1 addition & 1 deletion docs/multi-stage-query/reference.md
View file
Original file line number Diff line number Diff line change
Expand Up @@ -248,7 +248,7 @@ The following table lists the context parameters for the MSQ task engine:
| `selectDestination` | SELECT<br /><br /> Controls where the final result of the select query is written. <br />Use `taskReport`(the default) to write select results to the task report. <b> This is not scalable since task reports size explodes for large results </b> <br/>Use `durableStorage` to write results to durable storage location. <b>For large results sets, its recommended to use `durableStorage` </b>. To configure durable storage see [`this`](#durable-storage) section. | `taskReport` |
| `waitTillSegmentsLoad` | INSERT, REPLACE<br /><br /> If set, the ingest query waits for the generated segment to be loaded before exiting, else the ingest query exits without waiting. The task and live reports contain the information about the status of loading segments if this flag is set. This will ensure that any future queries made after the ingestion exits will include results from the ingestion. The drawback is that the controller task will stall till the segments are loaded. | `false` |
| `includeSegmentSource` | SELECT, INSERT, REPLACE<br /><br /> Controls the sources, which will be queried for results in addition to the segments present on deep storage. Can be `NONE` or `REALTIME`. If this value is `NONE`, only non-realtime (published and used) segments will be downloaded from deep storage. If this value is `REALTIME`, results will also be included from realtime tasks. | `NONE` |

| `rowsPerPage` | SELECT<br /><br />The number of rows per page to target. The actual number of rows per page may be somewhat higher or lower than this number. In most cases, use the default.<br /> This property comes into effect only when `selectDestination` is set to `durableStorage` | 100000 |
## Joins

Joins in multi-stage queries use one of two algorithms based on what you set the [context parameter](#context-parameters) `sqlJoinAlgorithm` to:
Expand Down
43 changes: 20 additions & 23 deletions ...nsions-core/multi-stage-query/src/main/java/org/apache/druid/msq/exec/ControllerImpl.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -145,7 +145,6 @@
import org.apache.druid.msq.input.table.DataSegmentWithLocation;
import org.apache.druid.msq.input.table.TableInputSpec;
import org.apache.druid.msq.input.table.TableInputSpecSlicer;
import org.apache.druid.msq.kernel.GlobalSortTargetSizeShuffleSpec;
import org.apache.druid.msq.kernel.QueryDefinition;
import org.apache.druid.msq.kernel.QueryDefinitionBuilder;
import org.apache.druid.msq.kernel.StageDefinition;
Expand Down Expand Up @@ -1663,12 +1662,7 @@ private static QueryDefinition makeQueryDefinition(
final ShuffleSpecFactory shuffleSpecFactory;

if (MSQControllerTask.isIngestion(querySpec)) {
shuffleSpecFactory = (clusterBy, aggregate) ->
new GlobalSortTargetSizeShuffleSpec(
clusterBy,
tuningConfig.getRowsPerSegment(),
aggregate
);
shuffleSpecFactory = ShuffleSpecFactories.getGlobalSortWithTargetSize(tuningConfig.getRowsPerSegment());

if (!columnMappings.hasUniqueOutputColumnNames()) {
// We do not expect to hit this case in production, because the SQL validator checks that column names
Expand All @@ -1693,8 +1687,9 @@ private static QueryDefinition makeQueryDefinition(
shuffleSpecFactory = ShuffleSpecFactories.singlePartition();
queryToPlan = querySpec.getQuery();
} else if (querySpec.getDestination() instanceof DurableStorageMSQDestination) {
// we add a final stage which generates one partition per worker.
shuffleSpecFactory = ShuffleSpecFactories.globalSortWithMaxPartitionCount(tuningConfig.getMaxNumWorkers());
shuffleSpecFactory = ShuffleSpecFactories.getGlobalSortWithTargetSize(
MultiStageQueryContext.getRowsPerPage(querySpec.getQuery().context())
);
queryToPlan = querySpec.getQuery();
} else {
throw new ISE("Unsupported destination [%s]", querySpec.getDestination());
Expand Down Expand Up @@ -1772,27 +1767,29 @@ private static QueryDefinition makeQueryDefinition(
return queryDef;
} else if (querySpec.getDestination() instanceof DurableStorageMSQDestination) {

// attaching new query results stage always.
// attaching new query results stage if the final stage does sort during shuffle so that results are ordered.
StageDefinition finalShuffleStageDef = queryDef.getFinalStageDefinition();
final QueryDefinitionBuilder builder = QueryDefinition.builder();
for (final StageDefinition stageDef : queryDef.getStageDefinitions()) {
builder.add(StageDefinition.builder(stageDef));
if (finalShuffleStageDef.doesSortDuringShuffle()) {
final QueryDefinitionBuilder builder = QueryDefinition.builder();
builder.addAll(queryDef);
builder.add(StageDefinition.builder(queryDef.getNextStageNumber())
.inputs(new StageInputSpec(queryDef.getFinalStageDefinition().getStageNumber()))
.maxWorkerCount(tuningConfig.getMaxNumWorkers())
.signature(finalShuffleStageDef.getSignature())
.shuffleSpec(null)
.processorFactory(new QueryResultFrameProcessorFactory())
);
return builder.build();
} else {
return queryDef;
}

builder.add(StageDefinition.builder(queryDef.getNextStageNumber())
.inputs(new StageInputSpec(queryDef.getFinalStageDefinition().getStageNumber()))
.maxWorkerCount(tuningConfig.getMaxNumWorkers())
.signature(finalShuffleStageDef.getSignature())
.shuffleSpec(null)
.processorFactory(new QueryResultFrameProcessorFactory())
);

return builder.build();
} else {
throw new ISE("Unsupported destination [%s]", querySpec.getDestination());
}
}



private static DataSchema generateDataSchema(
MSQSpec querySpec,
RowSignature querySignature,
Expand Down
14 changes: 14 additions & 0 deletions ...e/multi-stage-query/src/main/java/org/apache/druid/msq/querykit/ShuffleSpecFactories.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@
package org.apache.druid.msq.querykit;

import org.apache.druid.msq.kernel.GlobalSortMaxCountShuffleSpec;
import org.apache.druid.msq.kernel.GlobalSortTargetSizeShuffleSpec;
import org.apache.druid.msq.kernel.MixShuffleSpec;

/**
Expand Down Expand Up @@ -53,4 +54,17 @@ public static ShuffleSpecFactory globalSortWithMaxPartitionCount(final int parti
{
return (clusterBy, aggregate) -> new GlobalSortMaxCountShuffleSpec(clusterBy, partitions, aggregate);
}

/**
* Factory that produces globally sorted partitions of a target size.
*/
public static ShuffleSpecFactory getGlobalSortWithTargetSize(int targetSize)
{
return (clusterBy, aggregate) ->
new GlobalSortTargetSizeShuffleSpec(
clusterBy,
targetSize,
aggregate
);
}
}
63 changes: 46 additions & 17 deletions ...core/multi-stage-query/src/main/java/org/apache/druid/msq/sql/entity/PageInformation.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -21,11 +21,11 @@


import com.fasterxml.jackson.annotation.JsonCreator;
import com.fasterxml.jackson.annotation.JsonIgnore;
import com.fasterxml.jackson.annotation.JsonInclude;
import com.fasterxml.jackson.annotation.JsonProperty;

import javax.annotation.Nullable;
import java.util.Comparator;
import java.util.Objects;

/**
Expand All @@ -39,6 +39,14 @@ public class PageInformation
@Nullable
private final Long sizeInBytes;

// Worker field should not flow to the users of SqlStatementResource API since users should not care about worker
@Nullable
private final Integer worker;

// Partition field should not flow to the users of SqlStatementResource API since users should not care about partitions
@Nullable
private final Integer partition;

@JsonCreator
public PageInformation(
@JsonProperty("id") long id,
Expand All @@ -49,8 +57,27 @@ public PageInformation(
this.id = id;
this.numRows = numRows;
this.sizeInBytes = sizeInBytes;
this.worker = null;
this.partition = null;
}


public PageInformation(
long id,
Long numRows,
Long sizeInBytes,
Integer worker,
Integer partition
)
{
this.id = id;
this.numRows = numRows;
this.sizeInBytes = sizeInBytes;
this.worker = worker;
this.partition = partition;
}


@JsonProperty
public long getId()
{
Expand All @@ -74,6 +101,20 @@ public Long getSizeInBytes()
}


@Nullable
@JsonIgnore
public Integer getWorker()
{
return worker;
}

@Nullable
@JsonIgnore
public Integer getPartition()
{
return partition;
}

@Override
public boolean equals(Object o)
{
Expand All @@ -87,13 +128,13 @@ public boolean equals(Object o)
return id == that.id && Objects.equals(numRows, that.numRows) && Objects.equals(
sizeInBytes,
that.sizeInBytes
);
) && Objects.equals(worker, that.worker) && Objects.equals(partition, that.partition);
}

@Override
public int hashCode()
{
return Objects.hash(id, numRows, sizeInBytes);
return Objects.hash(id, numRows, sizeInBytes, worker, partition);
}

@Override
Expand All @@ -103,20 +144,8 @@ public String toString()
"id=" + id +
", numRows=" + numRows +
", sizeInBytes=" + sizeInBytes +
", worker=" + worker +
", partition=" + partition +
'}';
}

public static Comparator<PageInformation> getIDComparator()
{
return new PageComparator();
}

public static class PageComparator implements Comparator<PageInformation>
{
@Override
public int compare(PageInformation s1, PageInformation s2)
{
return Long.compare(s1.getId(), s2.getId());
}
}
}
15 changes: 10 additions & 5 deletions ...ti-stage-query/src/main/java/org/apache/druid/msq/sql/resources/SqlStatementResource.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -504,7 +504,7 @@ private Response buildNonOkResponse(DruidException exception)
}

@SuppressWarnings("ReassignedVariable")
private Optional<ResultSetInformation> getSampleResults(
private Optional<ResultSetInformation> getResultSetInformation(
String queryId,
String dataSource,
SqlStatementState sqlStatementState,
Expand Down Expand Up @@ -617,7 +617,7 @@ private Optional<SqlStatementResult> getStatementStatus(
taskResponse.getStatus().getCreatedTime(),
signature.orElse(null),
taskResponse.getStatus().getDuration(),
withResults ? getSampleResults(
withResults ? getResultSetInformation(
queryId,
msqControllerTask.getDataSource(),
sqlStatementState,
Expand Down Expand Up @@ -782,11 +782,16 @@ private Optional<Yielder<Object[]>> getResultYielder(
|| selectedPageId.equals(pageInformation.getId()))
.map(pageInformation -> {
try {
if (pageInformation.getWorker() == null || pageInformation.getPartition() == null) {
throw DruidException.defensive(
"Worker or partition number is null for page id [%d]",
pageInformation.getId()
);
}
return new FrameChannelSequence(standardImplementation.openChannel(
finalStage.getId(),
(int) pageInformation.getId(),
(int) pageInformation.getId()
// we would always have partition number == worker number
pageInformation.getWorker(),
pageInformation.getPartition()
));
}
catch (Exception e) {
Expand Down
12 changes: 12 additions & 0 deletions ...ore/multi-stage-query/src/main/java/org/apache/druid/msq/util/MultiStageQueryContext.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -115,6 +115,9 @@ public class MultiStageQueryContext
public static final String CTX_ROWS_PER_SEGMENT = "rowsPerSegment";
static final int DEFAULT_ROWS_PER_SEGMENT = 3000000;

public static final String CTX_ROWS_PER_PAGE = "rowsPerPage";
static final int DEFAULT_ROWS_PER_PAGE = 100000;

public static final String CTX_ROWS_IN_MEMORY = "rowsInMemory";
// Lower than the default to minimize the impact of per-row overheads that are not accounted for by
// OnheapIncrementalIndex. For example: overheads related to creating bitmaps during persist.
Expand Down Expand Up @@ -238,6 +241,15 @@ public static int getRowsPerSegment(final QueryContext queryContext)
);
}

public static int getRowsPerPage(final QueryContext queryContext)
{
return queryContext.getInt(
CTX_ROWS_PER_PAGE,
DEFAULT_ROWS_PER_PAGE
);
}


public static MSQSelectDestination getSelectDestination(final QueryContext queryContext)
{
return QueryContexts.getAsEnum(
Expand Down
Loading

0 comments on commit b89b93a

Please sign in to comment.