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 fc33de4b499 fix: LimitedBufferHashGrouper heapIndexUpdater used
incorrect table. (#20861)
fc33de4b499 is described below
commit fc33de4b499b100ac26b9b2478669c4c7472322f
Author: Gian Merlino <[email protected]>
AuthorDate: Fri Nov 6 14:15:48 2026 -0800
fix: LimitedBufferHashGrouper heapIndexUpdater used incorrect table.
(#20861)
When LimitedBufferHashGrouper swaps hash tables, it needs to also
update the hash table buffer reference in the heapIndexUpdater.
Failing to do so means that the pointers from the hash table to the
heap are not properly updated, which could cause rows to be skipped
during aggregation.
---
.../epinephelinae/LimitedBufferHashGrouper.java | 1 +
.../LimitedBufferHashGrouperTest.java | 54 ++++++++++++++++++++++
2 files changed, 55 insertions(+)
diff --git
a/processing/src/main/java/org/apache/druid/query/groupby/epinephelinae/LimitedBufferHashGrouper.java
b/processing/src/main/java/org/apache/druid/query/groupby/epinephelinae/LimitedBufferHashGrouper.java
index 167b21ee77c..737b0cafed0 100644
---
a/processing/src/main/java/org/apache/druid/query/groupby/epinephelinae/LimitedBufferHashGrouper.java
+++
b/processing/src/main/java/org/apache/druid/query/groupby/epinephelinae/LimitedBufferHashGrouper.java
@@ -585,6 +585,7 @@ public class LimitedBufferHashGrouper<KeyType> extends
AbstractBufferHashGrouper
size = numCopied;
updateMaxMergeBufferUsedBytes();
tableBuffer = newTableBuffer;
+ heapIndexUpdater.setHashTableBuffer(newTableBuffer);
growthCount++;
}
diff --git
a/processing/src/test/java/org/apache/druid/query/groupby/epinephelinae/LimitedBufferHashGrouperTest.java
b/processing/src/test/java/org/apache/druid/query/groupby/epinephelinae/LimitedBufferHashGrouperTest.java
index aa3884cbcce..34ef1753cf1 100644
---
a/processing/src/test/java/org/apache/druid/query/groupby/epinephelinae/LimitedBufferHashGrouperTest.java
+++
b/processing/src/test/java/org/apache/druid/query/groupby/epinephelinae/LimitedBufferHashGrouperTest.java
@@ -43,8 +43,10 @@ import org.junit.jupiter.api.Test;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Collections;
import java.util.Iterator;
import java.util.List;
+import java.util.Random;
public class LimitedBufferHashGrouperTest extends InitializedNullHandlingTest
{
@@ -450,6 +452,58 @@ public class LimitedBufferHashGrouperTest extends
InitializedNullHandlingTest
grouper.close();
}
+ @Test
+ public void testRepeatedKeysAcrossBufferSwaps()
+ {
+ for (final byte fill : new byte[]{0, (byte) 0xFF, 0x7F}) {
+ final GroupByTestColumnSelectorFactory columnSelectorFactory =
GrouperTestUtil.newColumnSelectorFactory();
+ final ByteBuffer buffer = ByteBuffer.allocate(20000);
+ for (int i = 0; i < buffer.capacity(); i++) {
+ buffer.put(i, fill);
+ }
+ final LimitedBufferHashGrouper<IntKey> grouper = new
LimitedBufferHashGrouper<>(
+ Suppliers.ofInstance(buffer),
+ GrouperTestUtil.intKeySerde(),
+ AggregatorAdapters.factorizeBuffered(
+ columnSelectorFactory,
+ ImmutableList.of(
+ new LongSumAggregatorFactory("valueSum", "value"),
+ new CountAggregatorFactory("count")
+ )
+ ),
+ Integer.MAX_VALUE,
+ 0.5f,
+ 2,
+ LIMIT,
+ false
+ );
+ grouper.init();
+
+ // Shuffle so that keys recur after buffer swaps, which is when the
stale heap index matters.
+ final int numKeys = 1000;
+ final int repeats = 10;
+ final List<Integer> rows = new ArrayList<>();
+ for (int r = 0; r < repeats; r++) {
+ for (int k = 0; k < numKeys; k++) {
+ rows.add(k);
+ }
+ }
+ Collections.shuffle(rows, new Random(0));
+
+ columnSelectorFactory.setRow(new MapBasedRow(0, ImmutableMap.of("value",
1L)));
+ for (final int k : rows) {
+ Assertions.assertTrue(grouper.aggregate(new IntKey(k)).isOk());
+ }
+ Assertions.assertTrue(grouper.getGrowthCount() > 1, "growthCount " +
grouper.getGrowthCount());
+
+ final List<Pair<Integer, List<Object>>> expected = new ArrayList<>();
+ for (int i = 0; i < LIMIT; i++) {
+ expected.add(Pair.of(i, ImmutableList.of((long) repeats, (long)
repeats)));
+ }
+ Assertions.assertEquals(expected, entriesToList(grouper.iterator(true)),
"fill " + fill);
+ }
+ }
+
private static LimitedBufferHashGrouper<IntKey> makeGrouper(
GroupByTestColumnSelectorFactory columnSelectorFactory,
int bufferSize
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
