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 b3184fbcee4 perf: Improve ScanQueryFrameProcessor's
sharedRunningCountForLimit. (#20589)
b3184fbcee4 is described below
commit b3184fbcee47e54e3c107105c5b191df5ce28653
Author: Gian Merlino <[email protected]>
AuthorDate: Wed Oct 21 09:11:20 2026 -0700
perf: Improve ScanQueryFrameProcessor's sharedRunningCountForLimit. (#20589)
Three changes:
1) Check the shared count on every row. The cost of an atomic read
should be negligible next to the cost of reading and writing a row.
2) Provide a mechanism for ReadableInputQueue to stop loading and
handing out segments once the limit has been reached.
3) Fix a bug where ChainedProcessorManager did not close its first
delegate ProcessorManager. This in turn could cause segments to leak
from ReadableInputQueue on early exit.
---
.../druid/msq/querykit/BaseLeafStageProcessor.java | 12 +++++-
.../msq/querykit/ChainedProcessorManager.java | 18 +++++----
.../druid/msq/querykit/ReadableInputQueue.java | 22 ++++++++++-
.../msq/querykit/scan/ScanQueryFrameProcessor.java | 39 ++++++++++++------
.../msq/querykit/scan/ScanQueryStageProcessor.java | 7 ++++
.../msq/querykit/ChainedProcessorManagerTest.java | 46 ++++++++++++++++++++++
.../druid/msq/querykit/ReadableInputQueueTest.java | 46 +++++++++++++++++++++-
.../querykit/scan/ScanQueryFrameProcessorTest.java | 39 ++++++++++++++++--
8 files changed, 202 insertions(+), 27 deletions(-)
diff --git
a/multi-stage-query/src/main/java/org/apache/druid/msq/querykit/BaseLeafStageProcessor.java
b/multi-stage-query/src/main/java/org/apache/druid/msq/querykit/BaseLeafStageProcessor.java
index 68b0015f2fe..3d0df2ee338 100644
---
a/multi-stage-query/src/main/java/org/apache/druid/msq/querykit/BaseLeafStageProcessor.java
+++
b/multi-stage-query/src/main/java/org/apache/druid/msq/querykit/BaseLeafStageProcessor.java
@@ -307,6 +307,15 @@ public abstract class BaseLeafStageProcessor extends
BasicStageProcessor
return AcquireMode.FULL;
}
+ /**
+ * Whether this stage's processors need no more base inputs, because they
have already produced all the output the
+ * stage needs. Once this returns true, the {@link ReadableInputQueue} stops
loading and handing out inputs.
+ */
+ protected boolean canSkipRemainingInputs()
+ {
+ return false;
+ }
+
/**
* Read base inputs, where "base" is meant in the same sense as in {@link
ExecutionVertex}: the primary datasource
* that drives query processing.
@@ -349,7 +358,8 @@ public abstract class BaseLeafStageProcessor extends
BasicStageProcessor
inputNumbers,
filteredSlices,
context.segmentLoadAheadCount(),
-
context.workOrder().getWorkerContext().getSegmentAcquireMode(getDefaultAcquireMode())
+
context.workOrder().getWorkerContext().getSegmentAcquireMode(getDefaultAcquireMode()),
+ this::canSkipRemainingInputs
);
}
diff --git
a/multi-stage-query/src/main/java/org/apache/druid/msq/querykit/ChainedProcessorManager.java
b/multi-stage-query/src/main/java/org/apache/druid/msq/querykit/ChainedProcessorManager.java
index 22be0f3e64d..e337cc4b62b 100644
---
a/multi-stage-query/src/main/java/org/apache/druid/msq/querykit/ChainedProcessorManager.java
+++
b/multi-stage-query/src/main/java/org/apache/druid/msq/querykit/ChainedProcessorManager.java
@@ -28,7 +28,6 @@ import
org.apache.druid.frame.processor.manager.ProcessorAndCallback;
import org.apache.druid.frame.processor.manager.ProcessorManager;
import org.apache.druid.utils.CloseableUtils;
-import javax.annotation.Nullable;
import java.util.List;
import java.util.Optional;
import java.util.concurrent.CopyOnWriteArrayList;
@@ -43,10 +42,13 @@ public class ChainedProcessorManager<A, B, R> implements
ProcessorManager<Object
{
/**
* First processor manager. FrameProcessors created by this runs before all
the others.
- * The reference is set to null once all the processors have been returned
by the channel.
*/
- @Nullable
- private ProcessorManager<A, List<A>> first;
+ private final ProcessorManager<A, List<A>> first;
+
+ /**
+ * Whether {@link #first} has returned all its processors.
+ */
+ private boolean firstExhausted;
/**
* Produces {@link #restFuture}.
@@ -90,7 +92,7 @@ public class ChainedProcessorManager<A, B, R> implements
ProcessorManager<Object
{
if (closed) {
throw new IllegalStateException();
- } else if (first != null) {
+ } else if (!firstExhausted) {
Optional<ProcessorAndCallback<A>> processorAndCallbackOptional =
Futures.getUnchecked(first.next());
if (processorAndCallbackOptional.isPresent()) {
// More processors left to run.
@@ -99,7 +101,7 @@ public class ChainedProcessorManager<A, B, R> implements
ProcessorManager<Object
//noinspection unchecked
return
Futures.immediateFuture(Optional.of((ProcessorAndCallback<Object>)
aProcessorAndCallback));
} else {
- first = null;
+ firstExhausted = true;
checkFirstProcessorComplete();
}
}
@@ -113,7 +115,7 @@ public class ChainedProcessorManager<A, B, R> implements
ProcessorManager<Object
private void checkFirstProcessorComplete()
{
- if (first == null && (firstProcessorResult.size() ==
firstProcessorCount.get())) {
+ if (firstExhausted && (firstProcessorResult.size() ==
firstProcessorCount.get())) {
restFuture.set(restFactory.apply(firstProcessorResult));
}
}
@@ -130,7 +132,7 @@ public class ChainedProcessorManager<A, B, R> implements
ProcessorManager<Object
if (!closed) {
closed = true;
CloseableUtils.closeAndWrapExceptions(() -> CloseableUtils.closeAll(
- first != null ? first : null,
+ first,
restFuture.isDone() ? FutureUtils.getUnchecked(restFuture, false) :
null
));
}
diff --git
a/multi-stage-query/src/main/java/org/apache/druid/msq/querykit/ReadableInputQueue.java
b/multi-stage-query/src/main/java/org/apache/druid/msq/querykit/ReadableInputQueue.java
index 520c08417ea..b071a5baada 100644
---
a/multi-stage-query/src/main/java/org/apache/druid/msq/querykit/ReadableInputQueue.java
+++
b/multi-stage-query/src/main/java/org/apache/druid/msq/querykit/ReadableInputQueue.java
@@ -54,6 +54,7 @@ import java.util.Optional;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.BooleanSupplier;
/**
* Queue for returning {@link ReadableInput} from a list of {@link
PhysicalInputSlice}.
@@ -125,6 +126,12 @@ public class ReadableInputQueue implements Closeable
* {@link LoadableSegment#acquireIfCached(AcquireMode)}.
*/
private final AcquireMode acquireMode;
+
+ /**
+ * Whether the caller needs no more inputs. Once true, {@link #nextInput()}
returns null and no new segment loads
+ * start; inputs that were already acquired are released by {@link #close()}.
+ */
+ private final BooleanSupplier canSkipRemaining;
private final AtomicBoolean started = new AtomicBoolean(false);
/**
@@ -135,18 +142,21 @@ public class ReadableInputQueue implements Closeable
* @param slices array of slices, one per input (same length as inputNumbers)
* @param loadahead number of items to load ahead of the queue
* @param acquireMode acquire mode for {@link SegmentCacheManager}
+ * @param canSkipRemaining whether the caller needs no more inputs; see
{@link #canSkipRemaining}
*/
public ReadableInputQueue(
final StandardPartitionReader partitionReader,
final IntList inputNumbers,
final List<PhysicalInputSlice> slices,
final int loadahead,
- final AcquireMode acquireMode
+ final AcquireMode acquireMode,
+ final BooleanSupplier canSkipRemaining
)
{
this.partitionReader = partitionReader;
this.loadahead = loadahead;
this.acquireMode = acquireMode;
+ this.canSkipRemaining = canSkipRemaining;
if (inputNumbers.size() != slices.size()) {
throw DruidException.defensive("Got input numbers[%s] for slices[%s]",
inputNumbers, slices);
@@ -235,6 +245,10 @@ public class ReadableInputQueue implements Closeable
throw DruidException.defensive("Not started, must call start() first");
}
+ if (canSkipRemaining.getAsBoolean()) {
+ return null;
+ }
+
ListenableFuture<NumberedReadableInput> future;
future = nextServerInput();
@@ -343,11 +357,15 @@ public class ReadableInputQueue implements Closeable
/**
* Load the next segment from {@link #loadableSegments} and return a future
to its reference. Returns null
- * if {@link #loadableSegments} is empty.
+ * if {@link #loadableSegments} is empty, or if {@link #canSkipRemaining} is
true.
*/
@Nullable
private ListenableFuture<NumberedReadableInput> loadNextSegment()
{
+ if (canSkipRemaining.getAsBoolean()) {
+ return null;
+ }
+
synchronized (this) {
final NumberedItem<LoadableSegment> nextItem = loadableSegments.poll();
if (nextItem == null) {
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 5ce7b01580a..c425f9dc07f 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
@@ -114,6 +114,10 @@ public class ScanQueryFrameProcessor extends
BaseLeafFrameProcessor
* Only set when {@link ScanQuery#getOrderBys()} is empty.
*/
private final AtomicLong sharedRunningCountForLimit;
+ /**
+ * Cached result of {@link #getOffsetPlusLimit(ScanQuery)}.
+ */
+ private final long queryOffsetPlusLimit;
private final ObjectMapper jsonMapper;
private final SettableLongVirtualColumn partitionBoostVirtualColumn;
private final VirtualColumns frameWriterVirtualColumns;
@@ -168,6 +172,7 @@ public class ScanQueryFrameProcessor extends
BaseLeafFrameProcessor
);
this.query = query;
this.sharedRunningCountForLimit = sharedRunningCountForLimit;
+ this.queryOffsetPlusLimit = getOffsetPlusLimit(query);
this.jsonMapper = jsonMapper;
this.partitionBoostVirtualColumn = new
SettableLongVirtualColumn(QueryKitUtils.PARTITION_BOOST_COLUMN);
@@ -187,8 +192,7 @@ public class ScanQueryFrameProcessor extends
BaseLeafFrameProcessor
@Override
public ReturnOrAwait<Object> runIncrementally(final IntSet readableInputs)
throws IOException
{
- if (sharedRunningCountForLimit != null
- && sharedRunningCountForLimit.get() >= getOffsetPlusLimit(query)) {
+ if (isSharedLimitReached(0)) {
return ReturnOrAwait.returnObject(Unit.instance());
}
@@ -485,8 +489,9 @@ public class ScanQueryFrameProcessor extends
BaseLeafFrameProcessor
cursorOffset.increment();
partitionBoostVirtualColumn.setValue(partitionBoostVirtualColumn.getValue() +
1);
- // Flush if we reached cursorPushDownLimit.
- flush = cursorPushDownLimit >= 0 && cursorRowsRead >=
cursorPushDownLimit;
+ // Flush if we reached cursorPushDownLimit or the shared limit.
+ flush = (cursorPushDownLimit >= 0 && cursorRowsRead >=
cursorPushDownLimit)
+ || isSharedLimitReached(frameWriter.getNumRows());
} else {
// addSelection failed because the frame is full.
if (frameWriter.getNumRows() == 0) {
@@ -497,12 +502,7 @@ public class ScanQueryFrameProcessor extends
BaseLeafFrameProcessor
}
if (flush) {
- final long numRowsWritten = flushFrameWriter();
-
- if (sharedRunningCountForLimit != null) {
- sharedRunningCountForLimit.addAndGet(numRowsWritten);
- }
-
+ flushFrameWriter();
break;
}
}
@@ -526,6 +526,11 @@ public class ScanQueryFrameProcessor extends
BaseLeafFrameProcessor
Iterables.getOnlyElement(outputChannels()).write(frame);
frameWriter.close();
frameWriter = null;
+
+ if (sharedRunningCountForLimit != null) {
+ sharedRunningCountForLimit.addAndGet(frame.numRows());
+ }
+
return frame.numRows();
} else {
if (frameWriter != null) {
@@ -559,7 +564,7 @@ public class ScanQueryFrameProcessor extends
BaseLeafFrameProcessor
if (query.isLimited()
&& ordering.size() >= query.getOrderBys().size()
&& query.getOrderBys().equals(ordering.subList(0,
query.getOrderBys().size()))) {
- cursorPushDownLimit = getOffsetPlusLimit(query);
+ cursorPushDownLimit = queryOffsetPlusLimit;
} else {
cursorPushDownLimit = NO_LIMIT;
}
@@ -567,11 +572,21 @@ public class ScanQueryFrameProcessor extends
BaseLeafFrameProcessor
return rowsFlushed;
}
+ /**
+ * Whether the limit in {@link #sharedRunningCountForLimit} has been
reached, or would be reached if
+ * {@code pendingRows} were added.
+ */
+ private boolean isSharedLimitReached(final long pendingRows)
+ {
+ return sharedRunningCountForLimit != null
+ && sharedRunningCountForLimit.get() + pendingRows >=
queryOffsetPlusLimit;
+ }
+
/**
* Returns the {@link ScanQuery#getScanRowsOffset()} plus {@link
ScanQuery#getScanRowsLimit()}, or
* {@link Long#MAX_VALUE} if that addition would overflow.
*/
- private static long getOffsetPlusLimit(final ScanQuery query)
+ static long getOffsetPlusLimit(final ScanQuery query)
{
final long sum = query.getScanRowsOffset() + query.getScanRowsLimit();
if (sum < 0) {
diff --git
a/multi-stage-query/src/main/java/org/apache/druid/msq/querykit/scan/ScanQueryStageProcessor.java
b/multi-stage-query/src/main/java/org/apache/druid/msq/querykit/scan/ScanQueryStageProcessor.java
index 52991e5a3be..715481036c3 100644
---
a/multi-stage-query/src/main/java/org/apache/druid/msq/querykit/scan/ScanQueryStageProcessor.java
+++
b/multi-stage-query/src/main/java/org/apache/druid/msq/querykit/scan/ScanQueryStageProcessor.java
@@ -123,6 +123,13 @@ public class ScanQueryStageProcessor extends
BaseLeafStageProcessor
);
}
+ @Override
+ protected boolean canSkipRemainingInputs()
+ {
+ return runningCountForLimit != null
+ && runningCountForLimit.get() >=
ScanQueryFrameProcessor.getOffsetPlusLimit(query);
+ }
+
@Override
protected AcquireMode getDefaultAcquireMode()
{
diff --git
a/multi-stage-query/src/test/java/org/apache/druid/msq/querykit/ChainedProcessorManagerTest.java
b/multi-stage-query/src/test/java/org/apache/druid/msq/querykit/ChainedProcessorManagerTest.java
index 04ca695c70c..62ceacef658 100644
---
a/multi-stage-query/src/test/java/org/apache/druid/msq/querykit/ChainedProcessorManagerTest.java
+++
b/multi-stage-query/src/test/java/org/apache/druid/msq/querykit/ChainedProcessorManagerTest.java
@@ -33,6 +33,7 @@ import org.apache.druid.frame.processor.Bouncer;
import org.apache.druid.frame.processor.FrameProcessorExecutorTest;
import org.apache.druid.frame.processor.FrameProcessors;
import org.apache.druid.frame.processor.manager.NilFrameProcessor;
+import org.apache.druid.frame.processor.manager.ProcessorAndCallback;
import org.apache.druid.frame.processor.manager.ProcessorManager;
import org.apache.druid.frame.processor.manager.ProcessorManagers;
import org.apache.druid.frame.processor.test.SimpleReturningFrameProcessor;
@@ -61,8 +62,10 @@ import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
+import java.util.Optional;
import java.util.Set;
import java.util.concurrent.ExecutionException;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.util.stream.Collectors;
import java.util.stream.LongStream;
@@ -191,6 +194,49 @@ public class ChainedProcessorManagerTest extends
FrameProcessorExecutorTest.Base
Assertions.assertEquals(expectedValues, actualValues);
}
+ @Test
+ public void test_first_processor_manager_closed_after_exhaustion() throws
ExecutionException, InterruptedException
+ {
+ final BlockingQueueFrameChannel outputChannel = new
BlockingQueueFrameChannel(3);
+ final ProcessorManager<List<Long>, Long> firstManager =
+ ProcessorManagers.of(() -> new
SimpleReturningFrameProcessor<>(ImmutableList.of(1L, 2L, 3L)));
+ final AtomicBoolean firstClosed = new AtomicBoolean(false);
+
+ final ChainedProcessorManager<List<Long>, Long, Long>
chainedProcessorManager = new ChainedProcessorManager<>(
+ new ProcessorManager<List<Long>, Long>()
+ {
+ @Override
+ public ListenableFuture<Optional<ProcessorAndCallback<List<Long>>>>
next()
+ {
+ return firstManager.next();
+ }
+
+ @Override
+ public Long result()
+ {
+ return firstManager.result();
+ }
+
+ @Override
+ public void close()
+ {
+ firstClosed.set(true);
+ firstManager.close();
+ }
+ },
+ (values) -> createNextProcessors(outputChannel.writable(),
values.stream().flatMap(List::stream).collect(Collectors.toList()))
+ );
+
+ exec.runAllFully(
+ chainedProcessorManager,
+ maxOutstandingProcessors,
+ bouncer,
+ null
+ ).get();
+
+ Assertions.assertTrue(firstClosed.get(), "first manager must be closed
even after returning all its processors");
+ }
+
@Test
public void test_failing_processor_manager()
{
diff --git
a/multi-stage-query/src/test/java/org/apache/druid/msq/querykit/ReadableInputQueueTest.java
b/multi-stage-query/src/test/java/org/apache/druid/msq/querykit/ReadableInputQueueTest.java
index ea243c65bab..66bb3123234 100644
---
a/multi-stage-query/src/test/java/org/apache/druid/msq/querykit/ReadableInputQueueTest.java
+++
b/multi-stage-query/src/test/java/org/apache/druid/msq/querykit/ReadableInputQueueTest.java
@@ -46,7 +46,9 @@ import java.util.List;
import java.util.Optional;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.BooleanSupplier;
/**
* Tests for {@link ReadableInputQueue}'s segment-load lifecycle:
close-while-loading must promptly fail the futures
@@ -152,13 +154,23 @@ class ReadableInputQueueTest
}
private static ReadableInputQueue makeQueue(int loadahead,
LoadableSegment... segments)
+ {
+ return makeQueue(loadahead, () -> false, segments);
+ }
+
+ private static ReadableInputQueue makeQueue(
+ int loadahead,
+ BooleanSupplier canSkipRemaining,
+ LoadableSegment... segments
+ )
{
return new ReadableInputQueue(
null,
IntList.of(0),
List.of(new PhysicalInputSlice(ReadablePartitions.empty(),
List.of(segments), List.of())),
loadahead,
- AcquireMode.FULL
+ AcquireMode.FULL,
+ canSkipRemaining
);
}
@@ -410,6 +422,38 @@ class ReadableInputQueueTest
Assertions.assertNull(queue.nextInput(), "no inputs may be handed out
after close");
}
+ @Test
+ void testCanSkipRemainingStopsHandingOutInputs()
+ {
+ final CountingSegment cachedSegment = new CountingSegment();
+ final TestLoadableSegment loadable = new TestLoadableSegment(new
AcquireSegmentAction(), Optional.of(cachedSegment));
+ final AtomicBoolean canSkipRemaining = new AtomicBoolean(false);
+
+ final ReadableInputQueue queue = makeQueue(0, canSkipRemaining::get,
loadable);
+ queue.start();
+ Assertions.assertEquals(1, queue.remaining());
+
+ canSkipRemaining.set(true);
+ Assertions.assertNull(queue.nextInput(), "no inputs may be handed out once
remaining inputs can be skipped");
+
+ queue.close();
+ Assertions.assertEquals(1, cachedSegment.closes.get(), "skipped cached
segment must be closed on close");
+ }
+
+ @Test
+ void testCanSkipRemainingStopsLoadahead()
+ {
+ final AtomicInteger cancels = new AtomicInteger();
+ final TestLoadableSegment loadable = new TestLoadableSegment(new
AcquireSegmentAction(cancels::incrementAndGet));
+
+ final ReadableInputQueue queue = makeQueue(1, () -> true, loadable);
+ queue.start();
+ Assertions.assertNull(queue.nextInput());
+
+ queue.close();
+ Assertions.assertEquals(0, cancels.get(), "loadahead must not start a load
once remaining inputs can be skipped");
+ }
+
@Test
void testRemainingCountsPendingInputs()
{
diff --git
a/multi-stage-query/src/test/java/org/apache/druid/msq/querykit/scan/ScanQueryFrameProcessorTest.java
b/multi-stage-query/src/test/java/org/apache/druid/msq/querykit/scan/ScanQueryFrameProcessorTest.java
index 66b1f4b220e..5f2d7021d7d 100644
---
a/multi-stage-query/src/test/java/org/apache/druid/msq/querykit/scan/ScanQueryFrameProcessorTest.java
+++
b/multi-stage-query/src/test/java/org/apache/druid/msq/querykit/scan/ScanQueryFrameProcessorTest.java
@@ -79,6 +79,7 @@ import javax.annotation.Nullable;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
public class ScanQueryFrameProcessorTest extends FrameProcessorTestBase
{
@@ -112,6 +113,28 @@ public class ScanQueryFrameProcessorTest extends
FrameProcessorTestBase
);
}
+ @Test
+ public void test_runWithSegments_sharedLimitReachedMidFrame() throws
Exception
+ {
+ final QueryableIndex queryableIndex = TestIndex.getMMappedTestIndex();
+
+ final ScanQuery query =
+ Druids.newScanQueryBuilder()
+ .dataSource("test")
+ .intervals(new
MultipleIntervalSegmentSpec(ImmutableList.of(Intervals.ETERNITY)))
+ .columns("__time", "market")
+ .limit(13)
+ .build();
+
+ // Other processors have already written 10 rows, so this one must stop
after 3, even though a frame can hold more.
+ final AtomicLong sharedRunningCountForLimit = new AtomicLong(10);
+ final List<List<Object>> rows =
+ runScanOverSegment(queryableIndex, query, sharedRunningCountForLimit,
Integer.MAX_VALUE);
+
+ Assertions.assertEquals(3, rows.size());
+ Assertions.assertEquals(13, sharedRunningCountForLimit.get());
+ }
+
@Test
public void test_runWithSegments_descendingMatchesNonVectorized() throws
Exception
{
@@ -387,13 +410,23 @@ public class ScanQueryFrameProcessorTest extends
FrameProcessorTestBase
private List<List<Object>> runScanOverSegment(final QueryableIndex
queryableIndex, final ScanQuery query)
throws Exception
+ {
+ // Limit output frames to 1 row to ensure we test edge cases
+ return runScanOverSegment(queryableIndex, query, null, 1);
+ }
+
+ private List<List<Object>> runScanOverSegment(
+ final QueryableIndex queryableIndex,
+ final ScanQuery query,
+ @Nullable final AtomicLong sharedRunningCountForLimit,
+ final int maxRowsPerFrame
+ ) throws Exception
{
final CursorFactory cursorFactory = new
QueryableIndexCursorFactory(queryableIndex);
final RowSignature signature = cursorFactory.getRowSignature();
final BlockingQueueFrameChannel outputChannel =
BlockingQueueFrameChannel.minimal();
- // Limit output frames to 1 row to ensure we test edge cases
final FrameWriterFactory frameWriterFactory = new
LimitedFrameWriterFactory(
FrameWriters.makeFrameWriterFactory(
FrameType.latestRowBased(),
@@ -402,14 +435,14 @@ public class ScanQueryFrameProcessorTest extends
FrameProcessorTestBase
Collections.emptyList(),
false
),
- 1
+ maxRowsPerFrame
);
final ReferenceCountedSegmentProvider segmentReferenceProvider =
new ReferenceCountedSegmentProvider(new
QueryableIndexSegment(queryableIndex, SegmentId.dummy("test")));
final ScanQueryFrameProcessor processor = new ScanQueryFrameProcessor(
query,
- null,
+ sharedRunningCountForLimit,
new DefaultObjectMapper(),
ReadableInput.segment(
new SegmentReferenceHolder(
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
