Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Limit pages size to a configurable limit #14994

Merged
merged 12 commits into from
Oct 12, 2023
Merged

Limit pages size to a configurable limit #14994

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
12 commits
Select commit Hold shift + click to select a range
90564dc
Initial patch for page size limiting
cryptoe Sep 15, 2023
1624db3
Merge remote-tracking branch 'apache/master' into pageSizeLimiting
cryptoe Oct 1, 2023
6cd5b36
Upstream merge things.
cryptoe Oct 9, 2023
1cf1aa0
Merge remote-tracking branch 'apache/master' into pageSizeLimiting
cryptoe Oct 10, 2023
be9b736
Addressing review comments.
cryptoe Oct 11, 2023
65cea5c
Addressing review comments.
cryptoe Oct 11, 2023
cb4101f
Fixing forbidden check
cryptoe Oct 11, 2023
79a4eb0
Fixing checkstyle.
cryptoe Oct 11, 2023
6766b34
Adding some more tests.
cryptoe Oct 11, 2023
0def1aa
Review
cryptoe Oct 11, 2023
00e0829
Adding multiWorker test case.
cryptoe Oct 12, 2023
2f0fba8
Adding more tests.
cryptoe Oct 12, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion docs/multi-stage-query/reference.md
View file
Open in desktop
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 |
LakshSingla marked this conversation as resolved.
Show resolved Hide resolved
## 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
Open in desktop
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())
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I might be missing something in somewhere else in this PR, but doesn't GlobalSortTargetSizeShuffleSpec enforce the limit on the total partition size summed across all workers? Since we create a new page for each worker parition combination, would the limit be enforced?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

GlobalSortTargetSizeShuffleSpec enforces a limit on partition size globally yes.

So there can be 2 cases:

  1. If the last stage is group by post shuffle, then we know that each partition will only be present on distinct worker only. Hence the page size will control the number of rows in that partition.

  2. If the last stage is scanStage, then we add a new QueryResultFrameProcessor since data needs to be sorted on the boost column. The queryResultFrameProcessor will merge the result in the same partition and write out a single partition. Since the partition cuts on sizes are done globally, in the controller, we would have the final partition equal to the page size configured.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I have added a new testcase testExternSelectWithMultipleWorkers. You can look at the counter checks to see whats happening with a scan query.

);
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
Open in desktop
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
Open in desktop
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
Open in desktop
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
Open in desktop
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
Loading