This is an automated email from the ASF dual-hosted git repository.
gianm pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/druid.git
The following commit(s) were added to refs/heads/master by this push:
new e78e3b56452 fix: ScanQueryFrameProcessor should not send offset to
data servers. (#20588)
e78e3b56452 is described below
commit e78e3b56452d2f3c9283eb807831288beba975c0
Author: Gian Merlino <[email protected]>
AuthorDate: Wed Oct 14 11:23:05 2026 -0700
fix: ScanQueryFrameProcessor should not send offset to data servers.
(#20588)
ScanQueryFrameProcessor currently includes the original offset and
limit when sending queries to data servers. It should instead send
an offset of zero and add the offset to the limit.
---
.../msq/querykit/scan/ScanQueryFrameProcessor.java | 18 +--
.../druid/msq/exec/MSQLoadedSegmentTests.java | 145 +++++++++++++++++++++
2 files changed, 155 insertions(+), 8 deletions(-)
diff --git
a/multi-stage-query/src/main/java/org/apache/druid/msq/querykit/scan/ScanQueryFrameProcessor.java
b/multi-stage-query/src/main/java/org/apache/druid/msq/querykit/scan/ScanQueryFrameProcessor.java
index b2dac19e4b1..5ce7b01580a 100644
---
a/multi-stage-query/src/main/java/org/apache/druid/msq/querykit/scan/ScanQueryFrameProcessor.java
+++
b/multi-stage-query/src/main/java/org/apache/druid/msq/querykit/scan/ScanQueryFrameProcessor.java
@@ -188,7 +188,7 @@ public class ScanQueryFrameProcessor extends
BaseLeafFrameProcessor
public ReturnOrAwait<Object> runIncrementally(final IntSet readableInputs)
throws IOException
{
if (sharedRunningCountForLimit != null
- && sharedRunningCountForLimit.get() >= getQueryOffsetPlusLimit()) {
+ && sharedRunningCountForLimit.get() >= getOffsetPlusLimit(query)) {
return ReturnOrAwait.returnObject(Unit.instance());
}
@@ -223,14 +223,16 @@ public class ScanQueryFrameProcessor extends
BaseLeafFrameProcessor
*/
private static ScanQuery prepareScanQueryForDataServer(@NotNull ScanQuery
scanQuery)
{
+ // Add the offset to the limit, and remove the offset, when pushing down a
query.
+ final Druids.ScanQueryBuilder builder =
Druids.ScanQueryBuilder.copy(scanQuery).offset(0);
+
if (Order.NONE.equals(scanQuery.getTimeOrder()) &&
!scanQuery.getOrderBys().isEmpty()) {
- return Druids.ScanQueryBuilder.copy(scanQuery)
- .orderBy(ImmutableList.of())
- .limit(0)
- .build();
+ builder.orderBy(ImmutableList.of()).limit(0);
} else {
- return scanQuery;
+ builder.limit(getOffsetPlusLimit(scanQuery));
}
+
+ return builder.build();
}
@Override
@@ -557,7 +559,7 @@ public class ScanQueryFrameProcessor extends
BaseLeafFrameProcessor
if (query.isLimited()
&& ordering.size() >= query.getOrderBys().size()
&& query.getOrderBys().equals(ordering.subList(0,
query.getOrderBys().size()))) {
- cursorPushDownLimit = getQueryOffsetPlusLimit();
+ cursorPushDownLimit = getOffsetPlusLimit(query);
} else {
cursorPushDownLimit = NO_LIMIT;
}
@@ -569,7 +571,7 @@ public class ScanQueryFrameProcessor extends
BaseLeafFrameProcessor
* Returns the {@link ScanQuery#getScanRowsOffset()} plus {@link
ScanQuery#getScanRowsLimit()}, or
* {@link Long#MAX_VALUE} if that addition would overflow.
*/
- private long getQueryOffsetPlusLimit()
+ private static long getOffsetPlusLimit(final ScanQuery query)
{
final long sum = query.getScanRowsOffset() + query.getScanRowsLimit();
if (sum < 0) {
diff --git
a/multi-stage-query/src/test/java/org/apache/druid/msq/exec/MSQLoadedSegmentTests.java
b/multi-stage-query/src/test/java/org/apache/druid/msq/exec/MSQLoadedSegmentTests.java
index fd629551e0a..0b13ab04676 100644
---
a/multi-stage-query/src/test/java/org/apache/druid/msq/exec/MSQLoadedSegmentTests.java
+++
b/multi-stage-query/src/test/java/org/apache/druid/msq/exec/MSQLoadedSegmentTests.java
@@ -25,9 +25,11 @@ import com.google.common.collect.ImmutableSet;
import com.google.common.util.concurrent.Futures;
import org.apache.druid.client.ImmutableSegmentLoadInfo;
import org.apache.druid.error.ThrowableMatcher;
+import org.apache.druid.java.util.common.DateTimes;
import org.apache.druid.java.util.common.ISE;
import org.apache.druid.java.util.common.Intervals;
import org.apache.druid.java.util.common.granularity.Granularities;
+import org.apache.druid.java.util.common.guava.Sequence;
import org.apache.druid.java.util.common.guava.Sequences;
import org.apache.druid.java.util.common.guava.Yielders;
import org.apache.druid.msq.indexing.LegacyMSQSpec;
@@ -35,12 +37,18 @@ import org.apache.druid.msq.indexing.MSQTuningConfig;
import org.apache.druid.msq.indexing.destination.TaskReportMSQDestination;
import org.apache.druid.msq.test.MSQTestBase;
import org.apache.druid.msq.util.MultiStageQueryContext;
+import org.apache.druid.query.DefaultGenericQueryMetricsFactory;
import org.apache.druid.query.OrderBy;
+import org.apache.druid.query.QueryPlus;
+import org.apache.druid.query.QueryRunner;
import org.apache.druid.query.aggregation.CountAggregatorFactory;
+import org.apache.druid.query.context.ResponseContext;
import org.apache.druid.query.dimension.DefaultDimensionSpec;
import org.apache.druid.query.groupby.GroupByQuery;
import org.apache.druid.query.groupby.ResultRow;
import org.apache.druid.query.scan.ScanQuery;
+import org.apache.druid.query.scan.ScanQueryQueryToolChest;
+import org.apache.druid.query.scan.ScanResultValue;
import org.apache.druid.query.spec.MultipleIntervalSegmentSpec;
import org.apache.druid.segment.column.ColumnType;
import org.apache.druid.segment.column.RowSignature;
@@ -56,7 +64,10 @@ import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import java.util.List;
import java.util.Map;
+import java.util.function.Function;
+import java.util.stream.Collectors;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.doAnswer;
@@ -222,6 +233,102 @@ public class MSQLoadedSegmentTests extends MSQTestBase
.verifyResults();
}
+ @Test
+ public void testSelectWithLoadedSegmentsOnFooWithOrderByLimitOffset()
+ {
+ RowSignature resultSignature = RowSignature.builder()
+ .add("cnt", ColumnType.LONG)
+ .add("dim1", ColumnType.STRING)
+ .build();
+
+ stubDataServerScanResults(
+ ImmutableList.of("cnt", "dim1"),
+ ImmutableList.of(
+ ImmutableList.of(1L, "qwe"),
+ ImmutableList.of(1L, "tyu")
+ )
+ );
+
+ testSelectQuery()
+ .setSql("select cnt, dim1 from foo order by dim1 limit 2 offset 6")
+ .setExpectedMSQSpec(
+ LegacyMSQSpec.builder()
+ .query(
+ newScanQueryBuilder()
+ .dataSource(CalciteTests.DATASOURCE1)
+ .intervals(querySegmentSpec(Filtration.eternity()))
+ .columns("cnt", "dim1")
+ .columnTypes(resultSignature.getColumnTypes())
+
.orderBy(ImmutableList.of(OrderBy.ascending("dim1")))
+ .offset(6)
+ .limit(2)
+ .context(REALTIME_QUERY_CTX)
+ .build()
+ )
+ .columnMappings(ColumnMappings.identity(resultSignature))
+ .tuningConfig(MSQTuningConfig.defaultConfig())
+ .destination(TaskReportMSQDestination.INSTANCE)
+ .build()
+ )
+ .setQueryContext(REALTIME_QUERY_CTX)
+ .setExpectedRowSignature(resultSignature)
+ .setExpectedResultRows(ImmutableList.of(
+ new Object[]{1L, "qwe"},
+ new Object[]{1L, "tyu"}
+ ))
+ .verifyResults();
+ }
+
+ @Test
+ public void testSelectWithLoadedSegmentsOnFooWithTimeOrderLimitOffset()
+ {
+ RowSignature resultSignature = RowSignature.builder()
+ .add("__time", ColumnType.LONG)
+ .add("cnt", ColumnType.LONG)
+ .add("dim1", ColumnType.STRING)
+ .build();
+
+ final long time1 = DateTimes.of("2003-01-01").getMillis();
+ final long time2 = DateTimes.of("2003-01-02").getMillis();
+
+ stubDataServerScanResults(
+ ImmutableList.of("__time", "cnt", "dim1"),
+ ImmutableList.of(
+ ImmutableList.of(time1, 1L, "qwe"),
+ ImmutableList.of(time2, 1L, "tyu")
+ )
+ );
+
+ testSelectQuery()
+ .setSql("select __time, cnt, dim1 from foo order by __time limit 2
offset 6")
+ .setExpectedMSQSpec(
+ LegacyMSQSpec.builder()
+ .query(
+ newScanQueryBuilder()
+ .dataSource(CalciteTests.DATASOURCE1)
+ .intervals(querySegmentSpec(Filtration.eternity()))
+ .columns("__time", "cnt", "dim1")
+ .columnTypes(resultSignature.getColumnTypes())
+
.orderBy(ImmutableList.of(OrderBy.ascending("__time")))
+ .offset(6)
+ .limit(2)
+ .context(REALTIME_QUERY_CTX)
+ .build()
+ )
+ .columnMappings(ColumnMappings.identity(resultSignature))
+ .tuningConfig(MSQTuningConfig.defaultConfig())
+ .destination(TaskReportMSQDestination.INSTANCE)
+ .build()
+ )
+ .setQueryContext(REALTIME_QUERY_CTX)
+ .setExpectedRowSignature(resultSignature)
+ .setExpectedResultRows(ImmutableList.of(
+ new Object[]{time1, 1L, "qwe"},
+ new Object[]{time2, 1L, "tyu"}
+ ))
+ .verifyResults();
+ }
+
@Test
public void testGroupByWithLoadedSegmentsOnFoo()
{
@@ -394,4 +501,42 @@ public class MSQLoadedSegmentTests extends MSQTestBase
.setExpectedExecutionErrorMatcher(ThrowableMatcher.of(ISE.class))
.verifyExecutionError();
}
+
+ /**
+ * Stubs {@link #dataServerQueryHandler} to serve scan queries from the
given rows, passing them through
+ * {@link ScanQueryQueryToolChest#mergeResults} so that offset and limit are
applied the way a realtime server
+ * applies them.
+ */
+ private void stubDataServerScanResults(final List<String> columns, final
List<List<Object>> rows)
+ {
+ doAnswer(
+ invocationOnMock -> {
+ final ScanQuery query = invocationOnMock.getArgument(0);
+ final Function<Sequence<ScanResultValue>, Sequence<Object[]>>
mappingFunction =
+ invocationOnMock.getArgument(2);
+
+ // One row per batch, since that is what the time-ordered merge
expects.
+ final QueryRunner<ScanResultValue> baseRunner = (queryPlus,
responseContext) -> Sequences.simple(
+ rows.stream()
+ .map(row -> new ScanResultValue(null, columns,
ImmutableList.of(row)))
+ .collect(Collectors.toList())
+ );
+
+ final Sequence<ScanResultValue> results =
+ new
ScanQueryQueryToolChest(DefaultGenericQueryMetricsFactory.instance())
+ .mergeResults(baseRunner)
+ .run(QueryPlus.wrap(query), ResponseContext.createEmpty());
+
+ return Futures.immediateFuture(
+ new DataServerQueryResult<>(
+
ImmutableList.of(Yielders.each(mappingFunction.apply(results))),
+ ImmutableList.of(),
+ "foo"
+ )
+ );
+ }
+ )
+ .when(dataServerQueryHandler)
+ .fetchRowsFromDataServer(any(), any(), any(), any());
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
