You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@druid.apache.org by "gianm (via GitHub)" <gi...@apache.org> on 2024/08/16 17:32:33 UTC
[PR] MSQ: Add limitHint to global-sort shuffles. (druid)
gianm opened a new pull request, #16911:
URL: https://github.com/apache/druid/pull/16911
This allows pushing down limits into the SuperSorter.
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] MSQ: Add limitHint to global-sort shuffles. (druid)
Posted by "gianm (via GitHub)" <gi...@apache.org>.
gianm commented on code in PR #16911:
URL: https://github.com/apache/druid/pull/16911#discussion_r1722765106
##########
processing/src/main/java/org/apache/druid/frame/processor/SuperSorter.java:
##########
@@ -382,29 +388,42 @@ private void runWorkersIfPossible()
@GuardedBy("runWorkersLock")
private void setAllDoneIfPossible()
{
- if (totalInputFrames == 0 && outputPartitionsFuture.isDone()) {
- // No input data -- generate empty output channels.
- final ClusterByPartitions partitions = getOutputPartitions();
- final List<OutputChannel> channels = new ArrayList<>(partitions.size());
+ try {
+ if (totalInputFrames == 0 && outputPartitionsFuture.isDone()) {
+ // No input data -- generate empty output channels.
+ final ClusterByPartitions partitions = getOutputPartitions();
+ final List<OutputChannel> channels = new ArrayList<>(partitions.size());
- for (int partitionNum = 0; partitionNum < partitions.size(); partitionNum++) {
- channels.add(outputChannelFactory.openNilChannel(partitionNum));
- }
+ for (int partitionNum = 0; partitionNum < partitions.size(); partitionNum++) {
+ channels.add(outputChannelFactory.openNilChannel(partitionNum));
+ }
+
+ // OK to use wrap, not wrapReadOnly, because nil channels are already read-only.
+ allDone.set(OutputChannels.wrap(channels));
+ } else if (rowLimit == 0 && activeProcessors == 0) {
+ // We had a row limit, and got it all the way down to zero.
+ // Generate empty output channels for any partitions that we haven't written yet.
+ superSorterProgressTracker.markTriviallyComplete();
Review Comment:
Oh, yeah, you're right. I changed this to instead call `addMergedBatchesForLevel` to "fill in" the empty batches.
##########
processing/src/main/java/org/apache/druid/frame/processor/SuperSorter.java:
##########
@@ -382,29 +388,42 @@ private void runWorkersIfPossible()
@GuardedBy("runWorkersLock")
private void setAllDoneIfPossible()
{
- if (totalInputFrames == 0 && outputPartitionsFuture.isDone()) {
- // No input data -- generate empty output channels.
- final ClusterByPartitions partitions = getOutputPartitions();
- final List<OutputChannel> channels = new ArrayList<>(partitions.size());
+ try {
+ if (totalInputFrames == 0 && outputPartitionsFuture.isDone()) {
+ // No input data -- generate empty output channels.
+ final ClusterByPartitions partitions = getOutputPartitions();
+ final List<OutputChannel> channels = new ArrayList<>(partitions.size());
- for (int partitionNum = 0; partitionNum < partitions.size(); partitionNum++) {
- channels.add(outputChannelFactory.openNilChannel(partitionNum));
- }
+ for (int partitionNum = 0; partitionNum < partitions.size(); partitionNum++) {
+ channels.add(outputChannelFactory.openNilChannel(partitionNum));
+ }
+
+ // OK to use wrap, not wrapReadOnly, because nil channels are already read-only.
+ allDone.set(OutputChannels.wrap(channels));
+ } else if (rowLimit == 0 && activeProcessors == 0) {
+ // We had a row limit, and got it all the way down to zero.
+ // Generate empty output channels for any partitions that we haven't written yet.
+ superSorterProgressTracker.markTriviallyComplete();
Review Comment:
Oh, yeah, you're right. I changed this to instead call `addMergedBatchesForLevel` to "fill in" the empty channels.
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] MSQ: Add limitHint to global-sort shuffles. (druid)
Posted by "LakshSingla (via GitHub)" <gi...@apache.org>.
LakshSingla commented on code in PR #16911:
URL: https://github.com/apache/druid/pull/16911#discussion_r1721005643
##########
processing/src/main/java/org/apache/druid/frame/processor/SuperSorter.java:
##########
@@ -382,29 +388,42 @@ private void runWorkersIfPossible()
@GuardedBy("runWorkersLock")
private void setAllDoneIfPossible()
{
- if (totalInputFrames == 0 && outputPartitionsFuture.isDone()) {
- // No input data -- generate empty output channels.
- final ClusterByPartitions partitions = getOutputPartitions();
- final List<OutputChannel> channels = new ArrayList<>(partitions.size());
+ try {
+ if (totalInputFrames == 0 && outputPartitionsFuture.isDone()) {
+ // No input data -- generate empty output channels.
+ final ClusterByPartitions partitions = getOutputPartitions();
+ final List<OutputChannel> channels = new ArrayList<>(partitions.size());
- for (int partitionNum = 0; partitionNum < partitions.size(); partitionNum++) {
- channels.add(outputChannelFactory.openNilChannel(partitionNum));
- }
+ for (int partitionNum = 0; partitionNum < partitions.size(); partitionNum++) {
+ channels.add(outputChannelFactory.openNilChannel(partitionNum));
+ }
+
+ // OK to use wrap, not wrapReadOnly, because nil channels are already read-only.
+ allDone.set(OutputChannels.wrap(channels));
+ } else if (rowLimit == 0 && activeProcessors == 0) {
+ // We had a row limit, and got it all the way down to zero.
+ // Generate empty output channels for any partitions that we haven't written yet.
+ superSorterProgressTracker.markTriviallyComplete();
Review Comment:
If there was a row limit initially and we brought it down to zero, should it be called trivially completed? Trivially completed means that there wasn't any data to begin with.
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/querykit/groupby/GroupByQueryKit.java:
##########
Review Comment:
Can similar optimisation be applied for ScanQueryKit as well?
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] MSQ: Add limitHint to global-sort shuffles. (druid)
Posted by "github-advanced-security[bot] (via GitHub)" <gi...@apache.org>.
github-advanced-security[bot] commented on code in PR #16911:
URL: https://github.com/apache/druid/pull/16911#discussion_r1720150988
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/kernel/LimitHintJsonIncludeFilter.java:
##########
@@ -0,0 +1,38 @@
+/*
+ * 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.kernel;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+
+/**
+ * {@link JsonInclude} filter for {@link ShuffleSpec#limitHint()}.
+ *
+ * This API works by "creative" use of equals. It requires warnings to be suppressed
+ * and also requires spotbugs exclusions (see spotbugs-exclude.xml).
+ */
+@SuppressWarnings({"EqualsAndHashcode", "EqualsHashCode"})
+public class LimitHintJsonIncludeFilter
Review Comment:
## Inconsistent equals and hashCode
Class LimitHintJsonIncludeFilter overrides [equals](1) but not hashCode.
[Show more details](https://github.com/apache/druid/security/code-scanning/7723)
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] MSQ: Add limitHint to global-sort shuffles. (druid)
Posted by "LakshSingla (via GitHub)" <gi...@apache.org>.
LakshSingla commented on code in PR #16911:
URL: https://github.com/apache/druid/pull/16911#discussion_r1723270853
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/kernel/ShuffleSpec.java:
##########
@@ -68,4 +70,17 @@ public interface ShuffleSpec
* @throws IllegalStateException if kind is {@link ShuffleKind#GLOBAL_SORT} with more than one target partition
*/
int partitionCount();
+
+ /**
+ * Limit that can be applied during shuffling. This is provided to enable performance optimizations.
+ *
+ * Implementations may apply this limit to each partition individually, or may apply it to the entire resultset
+ * (across all partitions). Either approach is valid, so downstream logic must handle either one.
+ *
+ * Implementations may also ignore this hint completely, or may apply a limit that is somewhat higher than this hint.
+ */
+ default long limitHint()
Review Comment:
I was wondering if there's any merit in separating the limit hint into a POJO, and having both limit and offset baked into it. Something like `DefaultLimitSpec` without the columns. It's up to the reader to calculate the combined limit and use that, instead of baking that knowledge into a long.
I was thinking of use cases when we want to pushdown this limit into other portions of MSQ's stack and want to distinguish between rows [0..offset) (thrown away) and rows [offset, offset + limit) (kept)
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] MSQ: Add limitHint to global-sort shuffles. (druid)
Posted by "LakshSingla (via GitHub)" <gi...@apache.org>.
LakshSingla commented on code in PR #16911:
URL: https://github.com/apache/druid/pull/16911#discussion_r1737889909
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/kernel/LimitHintJsonIncludeFilter.java:
##########
Review Comment:
For my understanding, how does this work
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] MSQ: Add limitHint to global-sort shuffles. (druid)
Posted by "gianm (via GitHub)" <gi...@apache.org>.
gianm merged PR #16911:
URL: https://github.com/apache/druid/pull/16911
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] MSQ: Add limitHint to global-sort shuffles. (druid)
Posted by "gianm (via GitHub)" <gi...@apache.org>.
gianm commented on PR #16911:
URL: https://github.com/apache/druid/pull/16911#issuecomment-2298101662
> WDYT of pushing this limit hint into the `InputChannelsImpl#openSorted` (perhaps in a future patch)? This will also reduce the amount of data a single worker would have to read per partition.
I do think that'd make sense as a followup.
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] MSQ: Add limitHint to global-sort shuffles. (druid)
Posted by "gianm (via GitHub)" <gi...@apache.org>.
gianm commented on code in PR #16911:
URL: https://github.com/apache/druid/pull/16911#discussion_r1722783495
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/querykit/groupby/GroupByQueryKit.java:
##########
Review Comment:
Good point. I just added it. It's only there for the case where the `scan` query requires a sort. In theory it could be there for the non-sort case too, but in this patch only the sorting shuffle spec has a limit hint. I think we could add `limitHint` to other specs in the future.
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] MSQ: Add limitHint to global-sort shuffles. (druid)
Posted by "LakshSingla (via GitHub)" <gi...@apache.org>.
LakshSingla commented on code in PR #16911:
URL: https://github.com/apache/druid/pull/16911#discussion_r1723270853
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/kernel/ShuffleSpec.java:
##########
@@ -68,4 +70,17 @@ public interface ShuffleSpec
* @throws IllegalStateException if kind is {@link ShuffleKind#GLOBAL_SORT} with more than one target partition
*/
int partitionCount();
+
+ /**
+ * Limit that can be applied during shuffling. This is provided to enable performance optimizations.
+ *
+ * Implementations may apply this limit to each partition individually, or may apply it to the entire resultset
+ * (across all partitions). Either approach is valid, so downstream logic must handle either one.
+ *
+ * Implementations may also ignore this hint completely, or may apply a limit that is somewhat higher than this hint.
+ */
+ default long limitHint()
Review Comment:
I was wondering if there's any merit to separate the limit hint into a separate object, and have both limit and offset baked into it. Something like `DefaultLimitSpec` without the columns. It's up to the reader to calculate the combined limit and use that, instead of baking that knowledge into a long.
I was thinking of use cases when we want to pushdown this limit into other portions of MSQ's stack and want to distinguish between rows [0..offset) (thrown away) and rows [offset, offset + limit) (kept)
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] MSQ: Add limitHint to global-sort shuffles. (druid)
Posted by "gianm (via GitHub)" <gi...@apache.org>.
gianm commented on code in PR #16911:
URL: https://github.com/apache/druid/pull/16911#discussion_r1728013632
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/kernel/ShuffleSpec.java:
##########
@@ -68,4 +70,17 @@ public interface ShuffleSpec
* @throws IllegalStateException if kind is {@link ShuffleKind#GLOBAL_SORT} with more than one target partition
*/
int partitionCount();
+
+ /**
+ * Limit that can be applied during shuffling. This is provided to enable performance optimizations.
+ *
+ * Implementations may apply this limit to each partition individually, or may apply it to the entire resultset
+ * (across all partitions). Either approach is valid, so downstream logic must handle either one.
+ *
+ * Implementations may also ignore this hint completely, or may apply a limit that is somewhat higher than this hint.
+ */
+ default long limitHint()
Review Comment:
I don't think offset can be pushed down? Limit can be pushed down when sorting because if some row is in the top N globally, it must also be in the first N rows of whichever partition it appears in.
But with offset, if for example you have `LIMIT N OFFSET M`, we can't push down `OFFSET M` (i.e. skip `M` rows). It is possible that some of the first `M` rows in partition A still need to appear in the final resultset, perhaps because some of them are greater than any of the first `M` rows in the globally-sorted result. So the best we can do is push down `LIMIT N + M`.
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] MSQ: Add limitHint to global-sort shuffles. (druid)
Posted by "LakshSingla (via GitHub)" <gi...@apache.org>.
LakshSingla commented on code in PR #16911:
URL: https://github.com/apache/druid/pull/16911#discussion_r1737889909
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/kernel/LimitHintJsonIncludeFilter.java:
##########
Review Comment:
For my understanding, how does this work?
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] MSQ: Add limitHint to global-sort shuffles. (druid)
Posted by "gianm (via GitHub)" <gi...@apache.org>.
gianm commented on code in PR #16911:
URL: https://github.com/apache/druid/pull/16911#discussion_r1742323730
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/kernel/LimitHintJsonIncludeFilter.java:
##########
Review Comment:
It's this API: https://fasterxml.github.io/jackson-annotations/javadoc/2.9/com/fasterxml/jackson/annotation/JsonInclude.Include.html#CUSTOM
> Value that indicates that separate `filter` Object (specified by [JsonInclude.valueFilter()](https://fasterxml.github.io/jackson-annotations/javadoc/2.9/com/fasterxml/jackson/annotation/JsonInclude.html#valueFilter()) for value itself, and/or [JsonInclude.contentFilter()](https://fasterxml.github.io/jackson-annotations/javadoc/2.9/com/fasterxml/jackson/annotation/JsonInclude.html#contentFilter()) for contents of structured types) is to be used for determining inclusion criteria. Filter object's equals() method is called with value to serialize; if it returns true value is excluded (that is, filtered out); if false value is included.
It's kind of goofy, but it's the only tool Jackson provides us for keeping the serialized JSON clean other than "include non-null", "include non-default", and "include non-empty".
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org