You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@druid.apache.org by "maytasm (via GitHub)" <gi...@apache.org> on 2026/04/21 00:24:26 UTC
[PR] Optimizes SpillingGrouper for high cardinality group by dimensions (druid)
maytasm opened a new pull request, #19357:
URL: https://github.com/apache/druid/pull/19357
<!-- Thanks for trying to help us make Apache Druid be the best it can be! Please fill out as much of the following information as is possible (where relevant, and remove it when irrelevant) to help make the intention and scope of this PR clear in order to ease review. -->
<!-- Please read the doc for contribution (https://github.com/apache/druid/blob/master/CONTRIBUTING.md) before making this PR. Also, once you open a PR, please _avoid using force pushes and rebasing_ since these make it difficult for reviewers to see what you've changed in response to their reviews. See [the 'If your pull request shows conflicts with master' section](https://github.com/apache/druid/blob/master/CONTRIBUTING.md#if-your-pull-request-shows-conflicts-with-master) for more details. -->
Fixes #XXXX.
<!-- Replace XXXX with the id of the issue fixed in this PR. Remove this section if there is no corresponding issue. Don't reference the issue in the title of this pull-request. -->
<!-- If you are a committer, follow the PR action item checklist for committers:
https://github.com/apache/druid/blob/master/dev/committer-instructions.md#pr-and-issue-action-item-checklist-for-committers. -->
### Description
<!-- Describe the goal of this PR, what problem are you fixing. If there is a corresponding issue (referenced above), it's not necessary to repeat the description here, however, you may choose to keep one summary sentence. -->
<!-- Describe your patch: what did you change in code? How did you fix the problem? -->
<!-- If there are several relatively logically separate changes in this PR, create a mini-section for each of them. For example: -->
#### Fixed the bug ...
#### Renamed the class ...
#### Added a forbidden-apis entry ...
<!--
In each section, please describe design decisions made, including:
- Choice of algorithms
- Behavioral aspects. What configuration values are acceptable? How are corner cases and error conditions handled, such as when there are insufficient resources?
- Class organization and design (how the logic is split between classes, inheritance, composition, design patterns)
- Method organization and design (how the logic is split between methods, parameters and return types)
- Naming (class, method, API, configuration, HTTP endpoint, names of emitted metrics)
-->
<!-- It's good to describe an alternative design (or mention an alternative name) for every design (or naming) decision point and compare the alternatives with the designs that you've implemented (or the names you've chosen) to highlight the advantages of the chosen designs and names. -->
<!-- If there was a discussion of the design of the feature implemented in this PR elsewhere (e. g. a "Proposal" issue, any other issue, or a thread in the development mailing list), link to that discussion from this PR description and explain what have changed in your final design compared to your original proposal or the consensus version in the end of the discussion. If something hasn't changed since the original discussion, you can omit a detailed discussion of those aspects of the design here, perhaps apart from brief mentioning for the sake of readability of this PR description. -->
<!-- Some of the aspects mentioned above may be omitted for simple and small changes. -->
#### Release note
<!-- Give your best effort to summarize your changes in a couple of sentences aimed toward Druid users.
If your change doesn't have end user impact, you can skip this section.
For tips about how to write a good release note, see [Release notes](https://github.com/apache/druid/blob/master/CONTRIBUTING.md#release-notes).
-->
<hr>
##### Key changed/added classes in this PR
* `MyFoo`
* `OurBar`
* `TheirBaz`
<hr>
<!-- Check the items by putting "x" in the brackets for the done things. Not all of these items apply to every PR. Remove the items which are not done or not relevant to the PR. None of the items from the checklist below are strictly necessary, but it would be very helpful if you at least self-review the PR. -->
This PR has:
- [ ] been self-reviewed.
- [ ] using the [concurrency checklist](https://github.com/apache/druid/blob/master/dev/code-review/concurrency.md) (Remove this item if the PR doesn't have any relation to concurrency.)
- [ ] added documentation for new or modified features or behaviors.
- [ ] a release note entry in the PR description.
- [ ] added Javadocs for most classes and all non-trivial methods. Linked related entities via Javadoc links.
- [ ] added or updated version, license, or notice information in [licenses.yaml](https://github.com/apache/druid/blob/master/dev/license.md)
- [ ] added comments explaining the "why" and the intent of the code wherever would not be obvious for an unfamiliar reader.
- [ ] added unit tests or modified existing tests to cover new code paths, ensuring the threshold for [code coverage](https://github.com/apache/druid/blob/master/dev/code-review/code-coverage.md) is met.
- [ ] added integration tests.
- [ ] been tested in a test Druid cluster.
--
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] fix: Optimizes SpillingGrouper for high cardinality dimension(s) GroupBy with large memory footprint aggregators (druid)
Posted by "maytasm (via GitHub)" <gi...@apache.org>.
maytasm commented on code in PR #19357:
URL: https://github.com/apache/druid/pull/19357#discussion_r3192703516
##########
processing/src/main/java/org/apache/druid/query/groupby/epinephelinae/SpillingGrouper.java:
##########
Review Comment:
Im not sure I follow. Delete dictionaryFiles where/why?
--
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] fix: Optimizes SpillingGrouper for high cardinality dimension(s) GroupBy with large memory footprint aggregators (druid)
Posted by "jtuglu1 (via GitHub)" <gi...@apache.org>.
jtuglu1 commented on code in PR #19357:
URL: https://github.com/apache/druid/pull/19357#discussion_r3192095332
##########
processing/src/main/java/org/apache/druid/query/groupby/epinephelinae/LimitedTemporaryStorage.java:
##########
@@ -121,13 +122,15 @@ public void delete(final File file)
{
synchronized (files) {
if (files.contains(file)) {
+ final long fileSize = file.length();
try {
Files.delete(file.toPath());
}
catch (IOException e) {
log.warn(e, "Cannot delete file: %s", file);
}
files.remove(file);
+ bytesUsed.addAndGet(-fileSize);
Review Comment:
if this delete fails, I'd rather not have our accounting also be inaccurate – can we put this in the try after the delete? IMO, our accounting should always be an overestimate to avoid actual disk space issues, etc. where worse exceptions can happen (e.g. I'd rather exit/throw an error cleanly and debug that then see historicals/peons start crashing due to disk space errors).
##########
processing/src/main/java/org/apache/druid/query/groupby/epinephelinae/LimitedTemporaryStorage.java:
##########
@@ -121,13 +122,15 @@ public void delete(final File file)
{
synchronized (files) {
if (files.contains(file)) {
+ final long fileSize = file.length();
try {
Files.delete(file.toPath());
}
catch (IOException e) {
log.warn(e, "Cannot delete file: %s", file);
}
files.remove(file);
+ bytesUsed.addAndGet(-fileSize);
Review Comment:
if this delete fails, I'd rather not have our accounting also be inaccurate – can we put this in the try after the delete? IMO, our accounting should always be an overestimate to avoid actual disk space issues, etc. where worse exceptions can happen (e.g. I'd rather throw a query error cleanly and debug that then see historicals/peons start crashing due to disk space errors).
--
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] fix: Optimizes SpillingGrouper for high cardinality dimension(s) GroupBy with large memory footprint aggregators (druid)
Posted by "jtuglu1 (via GitHub)" <gi...@apache.org>.
jtuglu1 commented on code in PR #19357:
URL: https://github.com/apache/druid/pull/19357#discussion_r3192789245
##########
processing/src/main/java/org/apache/druid/query/groupby/epinephelinae/SpillingGrouper.java:
##########
@@ -345,12 +354,103 @@ public Entry<KeyType> apply(Entry<KeyType> entry)
private void spill() throws IOException
{
+ // Stream directly to a temp file first, then check the file size. If the file is small
+ // (serialized size much smaller than the pre-allocated buffer, e.g. HLL sketches in List mode),
+ // read it back into memory for batching to avoid creating thousands of tiny disk files.
+ // If the file is already large enough, keep it on disk as-is.
+ final File file;
try (CloseableIterator<Entry<KeyType>> iterator = grouper.iterator(true)) {
- files.add(spill(iterator));
- dictionaryFiles.add(spill(keySerde.getDictionary().iterator()));
+ file = spill(iterator);
+ }
+ pendingDictionaryEntries.addAll(keySerde.getDictionary());
+ grouper.reset();
+
+ final long fileSize = file.length();
+ if (fileSize < minSpillFileSize) {
+ pendingSpillRuns.add(Files.readAllBytes(file.toPath()));
+ pendingSpillBytes += fileSize;
+ temporaryStorage.delete(file);
+
+ if (pendingSpillBytes >= minSpillFileSize) {
+ flushPendingRunsToDisk();
+ }
+ } else {
+ files.add(file);
+ dictionaryFiles.add(spill(pendingDictionaryEntries.iterator()));
+ pendingDictionaryEntries.clear();
+ }
+ }
+
+ /**
+ * Merge-sorts all pending in-memory spill runs and writes them as a single sorted file to disk.
+ * Each run is already individually sorted (from grouper.iterator(true)); this method merges them
+ * so the output file is fully sorted, as required by iterator()'s mergeSorted across files.
+ */
+ private void flushPendingRunsToDisk() throws IOException
+ {
+ if (pendingSpillRuns.isEmpty()) {
+ return;
+ }
+
+ final Comparator<Entry<KeyType>> sortComparator =
+ sortHasNonGroupingFields ? defaultOrderKeyObjComparator : keyObjComparator;
- grouper.reset();
+ final List<MappingIterator<Entry<KeyType>>> readers = new ArrayList<>(pendingSpillRuns.size());
+ try {
+ for (final byte[] runBytes : pendingSpillRuns) {
+ readers.add(spillMapper.readValues(
+ spillMapper.getFactory().createParser(new LZ4BlockInputStream(new ByteArrayInputStream(runBytes))),
+ spillMapper.getTypeFactory().constructParametricType(ReusableEntry.class, keySerde.keyClazz())
+ ));
+ }
+ final List<CloseableIterator<Entry<KeyType>>> iterators = new ArrayList<>(readers.size());
+ for (final MappingIterator<Entry<KeyType>> reader : readers) {
+ iterators.add(deserializeIterator(reader));
+ }
+ files.add(spill(CloseableIterators.mergeSorted(iterators, sortComparator)));
+ dictionaryFiles.add(spill(pendingDictionaryEntries.iterator()));
}
+ finally {
+ for (final MappingIterator<Entry<KeyType>> reader : readers) {
+ try {
+ reader.close();
+ }
+ catch (IOException e) {
+ log.warn(e, "Failed to close reader while flushing pending spill runs");
+ }
+ }
+ pendingSpillRuns.clear();
+ pendingSpillBytes = 0;
+ pendingDictionaryEntries.clear();
+ }
+ }
+
+ private CloseableIterator<Entry<KeyType>> deserializeIterator(final Iterator<Entry<KeyType>> iterator)
+ {
+ return CloseableIterators.withEmptyBaggage(
+ Iterators.transform(
+ iterator,
+ new Function<>()
+ {
+ final ReusableEntry<KeyType> reusableEntry =
+ ReusableEntry.create(keySerde, aggregatorFactories.length);
+
+ @Override
+ public Entry<KeyType> apply(Entry<KeyType> entry)
+ {
+ final Object[] deserializedValues = reusableEntry.getValues();
+ for (int i = 0; i < deserializedValues.length; i++) {
+ deserializedValues[i] = aggregatorFactories[i].deserialize(entry.getValues()[i]);
+ if (deserializedValues[i] instanceof Integer) {
Review Comment:
Why are we coercing to long values here?
##########
processing/src/main/java/org/apache/druid/query/groupby/epinephelinae/SpillingGrouper.java:
##########
@@ -345,12 +354,103 @@ public Entry<KeyType> apply(Entry<KeyType> entry)
private void spill() throws IOException
{
+ // Stream directly to a temp file first, then check the file size. If the file is small
+ // (serialized size much smaller than the pre-allocated buffer, e.g. HLL sketches in List mode),
+ // read it back into memory for batching to avoid creating thousands of tiny disk files.
+ // If the file is already large enough, keep it on disk as-is.
+ final File file;
try (CloseableIterator<Entry<KeyType>> iterator = grouper.iterator(true)) {
- files.add(spill(iterator));
- dictionaryFiles.add(spill(keySerde.getDictionary().iterator()));
+ file = spill(iterator);
+ }
+ pendingDictionaryEntries.addAll(keySerde.getDictionary());
+ grouper.reset();
+
+ final long fileSize = file.length();
Review Comment:
I wonder if there might be an optimization here to reserve entire needed capacity for the array upfront to prevent it needing to double up to sufficient capacity. Since we only .clear() and not something like `trimToSize()` this resizing might be amortized across all the files, especially if they are of similar size.
##########
processing/src/test/java/org/apache/druid/query/groupby/epinephelinae/SpillingGrouperTest.java:
##########
@@ -0,0 +1,385 @@
+/*
+ * 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.query.groupby.epinephelinae;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.base.Suppliers;
+import com.google.common.collect.ImmutableMap;
+import org.apache.druid.data.input.MapBasedRow;
+import org.apache.druid.java.util.common.parsers.CloseableIterator;
+import org.apache.druid.query.aggregation.AggregatorFactory;
+import org.apache.druid.query.aggregation.CountAggregatorFactory;
+import org.apache.druid.query.aggregation.LongSumAggregatorFactory;
+import org.apache.druid.query.groupby.GroupByStatsProvider;
+import org.apache.druid.testing.InitializedNullHandlingTest;
+import org.junit.Assert;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+public class SpillingGrouperTest extends InitializedNullHandlingTest
+{
+ private static final AggregatorFactory[] AGGREGATOR_FACTORIES = new AggregatorFactory[]{
+ new LongSumAggregatorFactory("valueSum", "value"),
+ new CountAggregatorFactory("count")
+ };
+ private static final int KEY_SIZE = new IntKeySerde().keySize();
+ private static final float MAX_LOAD_FACTOR = 0.75f;
+ private static final int INITIAL_BUCKETS = 4;
+
+ @Rule
+ public TemporaryFolder temporaryFolder = new TemporaryFolder();
+
+ @Test
+ public void testNoSpilling() throws IOException
+ {
+ final File storageDir = temporaryFolder.newFolder();
+ // Only 3 keys with a 10,000-byte buffer. Everything fits in memory
+ try (SpillingGrouper<IntKey> grouper = makeGrouper(10000, storageDir, 1024 * 1024, 100)) {
+ for (int i = 0; i < 3; i++) {
+ Assert.assertTrue(grouper.aggregate(new IntKey(i)).isOk());
+ }
+
+ assertResultsCorrect(grouper, 3, 1);
+ Assert.assertEquals(0, storageDir.listFiles().length);
+ }
+ }
+
+ @Test
+ public void testSpillAndIterateSorted() throws IOException
+ {
+ final File storageDir = temporaryFolder.newFolder();
+ final int numKeys = 100;
+ // 100 unique keys force many spills since buffer is only 50 bytes. With iterator(true), results should be sorted ascending by key.
+ try (SpillingGrouper<IntKey> grouper = makeGrouper(50, storageDir, 1024 * 1024, 100)) {
+ for (int i = 0; i < numKeys; i++) {
+ Assert.assertTrue(grouper.aggregate(new IntKey(i)).isOk());
+ }
+
+ try (CloseableIterator<Grouper.Entry<IntKey>> iterator = grouper.iterator(true)) {
+ Assert.assertTrue("spilling should have occurred", storageDir.listFiles().length > 0);
+ int prevKey = -1;
+ int count = 0;
+ while (iterator.hasNext()) {
+ Grouper.Entry<IntKey> entry = iterator.next();
+ Assert.assertTrue(
+ "keys should be sorted ascending",
+ entry.getKey().intValue() > prevKey
+ );
+ prevKey = entry.getKey().intValue();
+ Assert.assertEquals(1L, entry.getValues()[0]);
+ Assert.assertEquals(1L, entry.getValues()[1]);
+ count++;
+ }
+ Assert.assertEquals(numKeys, count);
+ }
+ }
+ }
+
+ @Test
+ public void testSpillAndIterateUnsorted() throws IOException
+ {
+ final File storageDir = temporaryFolder.newFolder();
+ final int numKeys = 100;
+ // 100 unique keys force many spills since buffer is only 50 bytes. With iterator(false), results may be in any order, but all keys should be present with correct values.
+ try (SpillingGrouper<IntKey> grouper = makeGrouper(50, storageDir, 1024 * 1024, 100)) {
+ for (int i = 0; i < numKeys; i++) {
+ Assert.assertTrue(grouper.aggregate(new IntKey(i)).isOk());
+ }
+
+ assertResultsCorrect(grouper, numKeys, 1);
+ Assert.assertTrue("spilling should have occurred", storageDir.listFiles().length > 0);
+ }
+ }
+
+ @Test
+ public void testAggregatesDuplicateKeys() throws IOException
+ {
+ // SpillingGrouper doesn't combine across spills — duplicate keys from different spill files
+ // appear as separate entries in the sorted iterator. Verify that the total aggregate values
+ // per key sum to the expected amount across all entries.
+ final File storageDir = temporaryFolder.newFolder();
+ final int numKeys = 20;
+ final int duplicates = 5;
+ try (SpillingGrouper<IntKey> grouper = makeGrouper(50, storageDir, 1024 * 1024, 100)) {
+ for (int round = 0; round < duplicates; round++) {
+ for (int i = 0; i < numKeys; i++) {
+ Assert.assertTrue(grouper.aggregate(new IntKey(i)).isOk());
+ }
+ }
+
+ int totalEntries = 0;
+ final Map<Integer, Long> totalCounts = new HashMap<>();
+ try (CloseableIterator<Grouper.Entry<IntKey>> iterator = grouper.iterator(true)) {
+ Assert.assertTrue("spilling should have occurred", storageDir.listFiles().length > 0);
+ while (iterator.hasNext()) {
+ Grouper.Entry<IntKey> entry = iterator.next();
+ totalCounts.merge(entry.getKey().intValue(), (Long) entry.getValues()[1], Long::sum);
+ totalEntries++;
+ }
+ }
+ Assert.assertTrue(
+ "duplicate keys should exist across spills, so total entries (" + totalEntries
+ + ") should exceed unique key count (" + numKeys + ")",
+ totalEntries > numKeys
+ );
+ Assert.assertEquals(numKeys, totalCounts.size());
+ for (Map.Entry<Integer, Long> e : totalCounts.entrySet()) {
+ Assert.assertEquals(
+ "total count for key " + e.getKey(),
+ (long) duplicates,
+ (long) e.getValue()
+ );
+ }
+ }
+ }
+
+ @Test
+ public void testSmallSpillsAreBatched() throws IOException
+ {
+ final File storageDir = temporaryFolder.newFolder();
+ final int bufferSize = 50;
+ final int numKeys = 100;
+
+ int maxUsableEntries = computeMaxUsableEntries(bufferSize);
+ Assert.assertEquals(
+ "buffer should hold at most 1 entry, guaranteeing a spill on every key",
+ 1,
+ maxUsableEntries
+ );
+
+ try (SpillingGrouper<IntKey> grouper = makeGrouper(bufferSize, storageDir, 1024 * 1024, 100)) {
+ for (int i = 0; i < numKeys; i++) {
+ Assert.assertTrue(grouper.aggregate(new IntKey(i)).isOk());
+ }
+
+ assertResultsCorrect(grouper, numKeys, 1);
+
+ File[] files = storageDir.listFiles();
+ Assert.assertNotNull(files);
+ Assert.assertEquals(
+ "all spills are tiny and should batch into a single data + dictionary file pair",
+ 2,
+ files.length
+ );
+ }
+ }
+
+ @Test
+ public void testResetClearsPendingState() throws IOException
+ {
+ try (SpillingGrouper<IntKey> grouper = makeGrouper(50, temporaryFolder.newFolder(), 1024 * 1024, 100)) {
+ for (int i = 0; i < 50; i++) {
+ Assert.assertTrue(grouper.aggregate(new IntKey(i)).isOk());
+ }
+
+ grouper.reset();
+
+ for (int i = 1000; i < 1010; i++) {
+ Assert.assertTrue(grouper.aggregate(new IntKey(i)).isOk());
+ }
+
+ try (CloseableIterator<Grouper.Entry<IntKey>> iterator = grouper.iterator(true)) {
+ int count = 0;
+ while (iterator.hasNext()) {
+ Grouper.Entry<IntKey> entry = iterator.next();
+ Assert.assertTrue(
+ "keys should be >= 1000 after reset",
+ entry.getKey().intValue() >= 1000
+ );
+ count++;
+ }
+ Assert.assertEquals(10, count);
+ }
+ }
+ }
+
+ @Test
+ public void testDiskFull() throws IOException
+ {
+ try (SpillingGrouper<IntKey> grouper = makeGrouper(50, temporaryFolder.newFolder(), 10, 100)) {
+ AggregateResult lastResult = AggregateResult.ok();
+ for (int i = 0; i < 10000 && lastResult.isOk(); i++) {
+ lastResult = grouper.aggregate(new IntKey(i));
+ }
+
+ Assert.assertFalse("should have hit disk full", lastResult.isOk());
+ Assert.assertTrue(
+ "reason should mention disk space",
+ lastResult.getReason().contains("Not enough disk space")
+ );
Review Comment:
Could we maybe add some asserts/another test which verify the statistics on file delete errors? Something that would catch regressions for the bug that was patched in this PR.
--
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] fix: Optimizes SpillingGrouper for high cardinality dimension(s) GroupBy with large memory footprint aggregators (druid)
Posted by "maytasm (via GitHub)" <gi...@apache.org>.
maytasm commented on code in PR #19357:
URL: https://github.com/apache/druid/pull/19357#discussion_r3191687469
##########
processing/src/main/java/org/apache/druid/query/groupby/epinephelinae/SpillingGrouper.java:
##########
@@ -345,14 +362,105 @@ public Entry<KeyType> apply(Entry<KeyType> entry)
private void spill() throws IOException
{
+ // Stream directly to a temp file first, then check the file size. If the file is small
+ // (serialized size much smaller than the pre-allocated buffer, e.g. HLL sketches in List mode),
+ // read it back into memory for batching to avoid creating thousands of tiny disk files.
+ // If the file is already large enough, keep it on disk as-is.
+ final File file;
try (CloseableIterator<Entry<KeyType>> iterator = grouper.iterator(true)) {
- files.add(spill(iterator));
- dictionaryFiles.add(spill(keySerde.getDictionary().iterator()));
+ file = spill(iterator);
+ }
+ pendingDictionaryEntries.addAll(keySerde.getDictionary());
+ grouper.reset();
+
+ final long fileSize = file.length();
+ if (fileSize < MIN_SPILL_FILE_BYTES) {
+ pendingSpillRuns.add(Files.readAllBytes(file.toPath()));
Review Comment:
Fixed
--
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] fix: Optimizes SpillingGrouper for high cardinality dimension(s) GroupBy with large memory footprint aggregators (druid)
Posted by "jtuglu1 (via GitHub)" <gi...@apache.org>.
jtuglu1 commented on code in PR #19357:
URL: https://github.com/apache/druid/pull/19357#discussion_r3192789245
##########
processing/src/main/java/org/apache/druid/query/groupby/epinephelinae/SpillingGrouper.java:
##########
@@ -345,12 +354,103 @@ public Entry<KeyType> apply(Entry<KeyType> entry)
private void spill() throws IOException
{
+ // Stream directly to a temp file first, then check the file size. If the file is small
+ // (serialized size much smaller than the pre-allocated buffer, e.g. HLL sketches in List mode),
+ // read it back into memory for batching to avoid creating thousands of tiny disk files.
+ // If the file is already large enough, keep it on disk as-is.
+ final File file;
try (CloseableIterator<Entry<KeyType>> iterator = grouper.iterator(true)) {
- files.add(spill(iterator));
- dictionaryFiles.add(spill(keySerde.getDictionary().iterator()));
+ file = spill(iterator);
+ }
+ pendingDictionaryEntries.addAll(keySerde.getDictionary());
+ grouper.reset();
+
+ final long fileSize = file.length();
+ if (fileSize < minSpillFileSize) {
+ pendingSpillRuns.add(Files.readAllBytes(file.toPath()));
+ pendingSpillBytes += fileSize;
+ temporaryStorage.delete(file);
+
+ if (pendingSpillBytes >= minSpillFileSize) {
+ flushPendingRunsToDisk();
+ }
+ } else {
+ files.add(file);
+ dictionaryFiles.add(spill(pendingDictionaryEntries.iterator()));
+ pendingDictionaryEntries.clear();
+ }
+ }
+
+ /**
+ * Merge-sorts all pending in-memory spill runs and writes them as a single sorted file to disk.
+ * Each run is already individually sorted (from grouper.iterator(true)); this method merges them
+ * so the output file is fully sorted, as required by iterator()'s mergeSorted across files.
+ */
+ private void flushPendingRunsToDisk() throws IOException
+ {
+ if (pendingSpillRuns.isEmpty()) {
+ return;
+ }
+
+ final Comparator<Entry<KeyType>> sortComparator =
+ sortHasNonGroupingFields ? defaultOrderKeyObjComparator : keyObjComparator;
- grouper.reset();
+ final List<MappingIterator<Entry<KeyType>>> readers = new ArrayList<>(pendingSpillRuns.size());
+ try {
+ for (final byte[] runBytes : pendingSpillRuns) {
+ readers.add(spillMapper.readValues(
+ spillMapper.getFactory().createParser(new LZ4BlockInputStream(new ByteArrayInputStream(runBytes))),
+ spillMapper.getTypeFactory().constructParametricType(ReusableEntry.class, keySerde.keyClazz())
+ ));
+ }
+ final List<CloseableIterator<Entry<KeyType>>> iterators = new ArrayList<>(readers.size());
+ for (final MappingIterator<Entry<KeyType>> reader : readers) {
+ iterators.add(deserializeIterator(reader));
+ }
+ files.add(spill(CloseableIterators.mergeSorted(iterators, sortComparator)));
+ dictionaryFiles.add(spill(pendingDictionaryEntries.iterator()));
}
+ finally {
+ for (final MappingIterator<Entry<KeyType>> reader : readers) {
+ try {
+ reader.close();
+ }
+ catch (IOException e) {
+ log.warn(e, "Failed to close reader while flushing pending spill runs");
+ }
+ }
+ pendingSpillRuns.clear();
+ pendingSpillBytes = 0;
+ pendingDictionaryEntries.clear();
+ }
+ }
+
+ private CloseableIterator<Entry<KeyType>> deserializeIterator(final Iterator<Entry<KeyType>> iterator)
+ {
+ return CloseableIterators.withEmptyBaggage(
+ Iterators.transform(
+ iterator,
+ new Function<>()
+ {
+ final ReusableEntry<KeyType> reusableEntry =
+ ReusableEntry.create(keySerde, aggregatorFactories.length);
+
+ @Override
+ public Entry<KeyType> apply(Entry<KeyType> entry)
+ {
+ final Object[] deserializedValues = reusableEntry.getValues();
+ for (int i = 0; i < deserializedValues.length; i++) {
+ deserializedValues[i] = aggregatorFactories[i].deserialize(entry.getValues()[i]);
+ if (deserializedValues[i] instanceof Integer) {
Review Comment:
Curious why are we coercing to long values here?
--
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] fix: Optimizes SpillingGrouper for high cardinality dimension(s) GroupBy with large memory footprint aggregators (druid)
Posted by "maytasm (via GitHub)" <gi...@apache.org>.
maytasm merged PR #19357:
URL: https://github.com/apache/druid/pull/19357
--
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] fix: Optimizes SpillingGrouper for high cardinality dimension(s) GroupBy with large memory footprint aggregators (druid)
Posted by "jtuglu1 (via GitHub)" <gi...@apache.org>.
jtuglu1 commented on code in PR #19357:
URL: https://github.com/apache/druid/pull/19357#discussion_r3192785270
##########
processing/src/main/java/org/apache/druid/query/groupby/epinephelinae/SpillingGrouper.java:
##########
@@ -345,12 +354,103 @@ public Entry<KeyType> apply(Entry<KeyType> entry)
private void spill() throws IOException
{
+ // Stream directly to a temp file first, then check the file size. If the file is small
+ // (serialized size much smaller than the pre-allocated buffer, e.g. HLL sketches in List mode),
+ // read it back into memory for batching to avoid creating thousands of tiny disk files.
+ // If the file is already large enough, keep it on disk as-is.
+ final File file;
try (CloseableIterator<Entry<KeyType>> iterator = grouper.iterator(true)) {
- files.add(spill(iterator));
- dictionaryFiles.add(spill(keySerde.getDictionary().iterator()));
+ file = spill(iterator);
+ }
+ pendingDictionaryEntries.addAll(keySerde.getDictionary());
+ grouper.reset();
+
+ final long fileSize = file.length();
Review Comment:
I wonder if there might be an optimization here to reserve entire needed capacity for the array list upfront to prevent it needing to repeatedly double up to sufficient capacity. I think it should noop on subsequent calls too once capacity is allocated. Since we only .clear() and not something like `trimToSize()` this resizing might just be amortized across all the files, especially if they are of similar size and don't amount to much.
--
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] fix: Optimizes SpillingGrouper for high cardinality dimension(s) GroupBy with large memory footprint aggregators (druid)
Posted by "maytasm (via GitHub)" <gi...@apache.org>.
maytasm commented on code in PR #19357:
URL: https://github.com/apache/druid/pull/19357#discussion_r3193243756
##########
processing/src/main/java/org/apache/druid/query/groupby/epinephelinae/SpillingGrouper.java:
##########
@@ -293,6 +320,22 @@ public void setSpillingAllowed(final boolean spillingAllowed)
@Override
public CloseableIterator<Entry<KeyType>> iterator(final boolean sorted)
{
+ // Flush any runs that did not reach MIN_SPILL_FILE_BYTES during the spill phase.
+ try {
+ flushPendingRunsToDisk();
Review Comment:
Done
--
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] fix: Optimizes SpillingGrouper for high cardinality dimension(s) GroupBy with large memory footprint aggregators (druid)
Posted by "maytasm (via GitHub)" <gi...@apache.org>.
maytasm commented on code in PR #19357:
URL: https://github.com/apache/druid/pull/19357#discussion_r3193316749
##########
processing/src/main/java/org/apache/druid/query/groupby/epinephelinae/SpillingGrouper.java:
##########
@@ -345,12 +354,103 @@ public Entry<KeyType> apply(Entry<KeyType> entry)
private void spill() throws IOException
{
+ // Stream directly to a temp file first, then check the file size. If the file is small
+ // (serialized size much smaller than the pre-allocated buffer, e.g. HLL sketches in List mode),
+ // read it back into memory for batching to avoid creating thousands of tiny disk files.
+ // If the file is already large enough, keep it on disk as-is.
+ final File file;
try (CloseableIterator<Entry<KeyType>> iterator = grouper.iterator(true)) {
- files.add(spill(iterator));
- dictionaryFiles.add(spill(keySerde.getDictionary().iterator()));
+ file = spill(iterator);
+ }
+ pendingDictionaryEntries.addAll(keySerde.getDictionary());
+ grouper.reset();
+
+ final long fileSize = file.length();
Review Comment:
I don't think it's worth optimizing.
1. ArrayList.clear() sets size to 0 but keeps the backing array, so after the first batch flushes, the capacity is already there for subsequent batches.
2. Even on the first batch, the cost is trivial. Say we have ~170 small spills per 1MB batch — that's about 7-8 doublings from the default capacity of 10. Each doubling copies an array of object references (8 bytes each), so the total copy overhead is on the order of a few KB. Negligible compared to the MB of actual spill data being handled.
3. Pre-allocating would require estimating how many spills fit in a batch (minSpillFileSize / avgSpillSize), but the spill size isn't known ahead of time — it depends on the data. So the estimate would be a guess anyway.
--
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] fix: Optimizes SpillingGrouper for high cardinality dimension(s) GroupBy with large memory footprint aggregators (druid)
Posted by "maytasm (via GitHub)" <gi...@apache.org>.
maytasm commented on code in PR #19357:
URL: https://github.com/apache/druid/pull/19357#discussion_r3192696172
##########
processing/src/main/java/org/apache/druid/query/groupby/epinephelinae/SpillingGrouper.java:
##########
@@ -293,6 +320,22 @@ public void setSpillingAllowed(final boolean spillingAllowed)
@Override
public CloseableIterator<Entry<KeyType>> iterator(final boolean sorted)
{
+ // Flush any runs that did not reach MIN_SPILL_FILE_BYTES during the spill phase.
+ try {
+ flushPendingRunsToDisk();
Review Comment:
Btw I wanted to benchmark `SpillingGrouper.iterator(false)` code path but turns out that SpillingGrouper.iterator(false) is never reachable through any production query path without a code change.
The only place SpillingGrouper.iterator() is called is:
- RowBasedGrouperHelper:634 → iterator(true) (always)
- ConcurrentGrouper:430/470 → iterator(true) or iterator(sorted) (where sorted comes from ConcurrentGrouper.iterator(sorted), which is always called with true from
RowBasedGrouperHelper)
--
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] fix: Optimizes SpillingGrouper for high cardinality dimension(s) GroupBy with large memory footprint aggregators (druid)
Posted by "maytasm (via GitHub)" <gi...@apache.org>.
maytasm commented on code in PR #19357:
URL: https://github.com/apache/druid/pull/19357#discussion_r3192703516
##########
processing/src/main/java/org/apache/druid/query/groupby/epinephelinae/SpillingGrouper.java:
##########
Review Comment:
Im not sure I follow. Delete dictionaryFiles where/why?
--
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] fix: Optimizes SpillingGrouper for high cardinality dimension(s) GroupBy with large memory footprint aggregators (druid)
Posted by "maytasm (via GitHub)" <gi...@apache.org>.
maytasm commented on PR #19357:
URL: https://github.com/apache/druid/pull/19357#issuecomment-4385679482
> LGTM once the remaining comments addressed and/or resolved
Thanks for the review. All followup comments addressed.
--
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] fix: Optimizes SpillingGrouper for high cardinality dimension(s) GroupBy with large memory footprint aggregators (druid)
Posted by "maytasm (via GitHub)" <gi...@apache.org>.
maytasm commented on code in PR #19357:
URL: https://github.com/apache/druid/pull/19357#discussion_r3192458687
##########
processing/src/main/java/org/apache/druid/query/groupby/epinephelinae/SpillingGrouper.java:
##########
@@ -72,6 +74,17 @@ public class SpillingGrouper<KeyType> implements Grouper<KeyType>
"Maximum number of spill files reached for this query. Try raising druid.query.groupBy.maxSpillFileCount."
);
+ /**
+ * Minimum number of serialized bytes that must accumulate across pending in-memory spill runs before they are
+ * flushed as a single file to disk. Aggregators like ThetaSketch pre-allocate a large fixed buffer per row
+ * (e.g. ~131KB for ThetaSketch(K=16384)), causing the in-memory grouper to flush frequently. However, when
+ * each key has been seen only a few times, the sketch serializes to just a handful of bytes in compact form.
+ * Without batching, this produces thousands of tiny spill files. By accumulating runs in heap memory first
+ * and writing to disk only once this threshold is reached, we avoid that explosion in file count without any
+ * extra disk I/O for small spills.
+ */
+ private static final long MIN_SPILL_FILE_BYTES = 1024 * 1024L; // 1MB
Review Comment:
Done
##########
processing/src/main/java/org/apache/druid/query/groupby/epinephelinae/LimitedTemporaryStorage.java:
##########
@@ -121,13 +122,15 @@ public void delete(final File file)
{
synchronized (files) {
if (files.contains(file)) {
+ final long fileSize = file.length();
try {
Files.delete(file.toPath());
}
catch (IOException e) {
log.warn(e, "Cannot delete file: %s", file);
}
files.remove(file);
+ bytesUsed.addAndGet(-fileSize);
Review Comment:
Done
--
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] fix: Optimizes SpillingGrouper for high cardinality dimension(s) GroupBy with large memory footprint aggregators (druid)
Posted by "maytasm (via GitHub)" <gi...@apache.org>.
maytasm commented on code in PR #19357:
URL: https://github.com/apache/druid/pull/19357#discussion_r3193294430
##########
processing/src/test/java/org/apache/druid/query/groupby/epinephelinae/SpillingGrouperTest.java:
##########
@@ -0,0 +1,385 @@
+/*
+ * 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.query.groupby.epinephelinae;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.base.Suppliers;
+import com.google.common.collect.ImmutableMap;
+import org.apache.druid.data.input.MapBasedRow;
+import org.apache.druid.java.util.common.parsers.CloseableIterator;
+import org.apache.druid.query.aggregation.AggregatorFactory;
+import org.apache.druid.query.aggregation.CountAggregatorFactory;
+import org.apache.druid.query.aggregation.LongSumAggregatorFactory;
+import org.apache.druid.query.groupby.GroupByStatsProvider;
+import org.apache.druid.testing.InitializedNullHandlingTest;
+import org.junit.Assert;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+public class SpillingGrouperTest extends InitializedNullHandlingTest
+{
+ private static final AggregatorFactory[] AGGREGATOR_FACTORIES = new AggregatorFactory[]{
+ new LongSumAggregatorFactory("valueSum", "value"),
+ new CountAggregatorFactory("count")
+ };
+ private static final int KEY_SIZE = new IntKeySerde().keySize();
+ private static final float MAX_LOAD_FACTOR = 0.75f;
+ private static final int INITIAL_BUCKETS = 4;
+
+ @Rule
+ public TemporaryFolder temporaryFolder = new TemporaryFolder();
+
+ @Test
+ public void testNoSpilling() throws IOException
+ {
+ final File storageDir = temporaryFolder.newFolder();
+ // Only 3 keys with a 10,000-byte buffer. Everything fits in memory
+ try (SpillingGrouper<IntKey> grouper = makeGrouper(10000, storageDir, 1024 * 1024, 100)) {
+ for (int i = 0; i < 3; i++) {
+ Assert.assertTrue(grouper.aggregate(new IntKey(i)).isOk());
+ }
+
+ assertResultsCorrect(grouper, 3, 1);
+ Assert.assertEquals(0, storageDir.listFiles().length);
+ }
+ }
+
+ @Test
+ public void testSpillAndIterateSorted() throws IOException
+ {
+ final File storageDir = temporaryFolder.newFolder();
+ final int numKeys = 100;
+ // 100 unique keys force many spills since buffer is only 50 bytes. With iterator(true), results should be sorted ascending by key.
+ try (SpillingGrouper<IntKey> grouper = makeGrouper(50, storageDir, 1024 * 1024, 100)) {
+ for (int i = 0; i < numKeys; i++) {
+ Assert.assertTrue(grouper.aggregate(new IntKey(i)).isOk());
+ }
+
+ try (CloseableIterator<Grouper.Entry<IntKey>> iterator = grouper.iterator(true)) {
+ Assert.assertTrue("spilling should have occurred", storageDir.listFiles().length > 0);
+ int prevKey = -1;
+ int count = 0;
+ while (iterator.hasNext()) {
+ Grouper.Entry<IntKey> entry = iterator.next();
+ Assert.assertTrue(
+ "keys should be sorted ascending",
+ entry.getKey().intValue() > prevKey
+ );
+ prevKey = entry.getKey().intValue();
+ Assert.assertEquals(1L, entry.getValues()[0]);
+ Assert.assertEquals(1L, entry.getValues()[1]);
+ count++;
+ }
+ Assert.assertEquals(numKeys, count);
+ }
+ }
+ }
+
+ @Test
+ public void testSpillAndIterateUnsorted() throws IOException
+ {
+ final File storageDir = temporaryFolder.newFolder();
+ final int numKeys = 100;
+ // 100 unique keys force many spills since buffer is only 50 bytes. With iterator(false), results may be in any order, but all keys should be present with correct values.
+ try (SpillingGrouper<IntKey> grouper = makeGrouper(50, storageDir, 1024 * 1024, 100)) {
+ for (int i = 0; i < numKeys; i++) {
+ Assert.assertTrue(grouper.aggregate(new IntKey(i)).isOk());
+ }
+
+ assertResultsCorrect(grouper, numKeys, 1);
+ Assert.assertTrue("spilling should have occurred", storageDir.listFiles().length > 0);
+ }
+ }
+
+ @Test
+ public void testAggregatesDuplicateKeys() throws IOException
+ {
+ // SpillingGrouper doesn't combine across spills — duplicate keys from different spill files
+ // appear as separate entries in the sorted iterator. Verify that the total aggregate values
+ // per key sum to the expected amount across all entries.
+ final File storageDir = temporaryFolder.newFolder();
+ final int numKeys = 20;
+ final int duplicates = 5;
+ try (SpillingGrouper<IntKey> grouper = makeGrouper(50, storageDir, 1024 * 1024, 100)) {
+ for (int round = 0; round < duplicates; round++) {
+ for (int i = 0; i < numKeys; i++) {
+ Assert.assertTrue(grouper.aggregate(new IntKey(i)).isOk());
+ }
+ }
+
+ int totalEntries = 0;
+ final Map<Integer, Long> totalCounts = new HashMap<>();
+ try (CloseableIterator<Grouper.Entry<IntKey>> iterator = grouper.iterator(true)) {
+ Assert.assertTrue("spilling should have occurred", storageDir.listFiles().length > 0);
+ while (iterator.hasNext()) {
+ Grouper.Entry<IntKey> entry = iterator.next();
+ totalCounts.merge(entry.getKey().intValue(), (Long) entry.getValues()[1], Long::sum);
+ totalEntries++;
+ }
+ }
+ Assert.assertTrue(
+ "duplicate keys should exist across spills, so total entries (" + totalEntries
+ + ") should exceed unique key count (" + numKeys + ")",
+ totalEntries > numKeys
+ );
+ Assert.assertEquals(numKeys, totalCounts.size());
+ for (Map.Entry<Integer, Long> e : totalCounts.entrySet()) {
+ Assert.assertEquals(
+ "total count for key " + e.getKey(),
+ (long) duplicates,
+ (long) e.getValue()
+ );
+ }
+ }
+ }
+
+ @Test
+ public void testSmallSpillsAreBatched() throws IOException
+ {
+ final File storageDir = temporaryFolder.newFolder();
+ final int bufferSize = 50;
+ final int numKeys = 100;
+
+ int maxUsableEntries = computeMaxUsableEntries(bufferSize);
+ Assert.assertEquals(
+ "buffer should hold at most 1 entry, guaranteeing a spill on every key",
+ 1,
+ maxUsableEntries
+ );
+
+ try (SpillingGrouper<IntKey> grouper = makeGrouper(bufferSize, storageDir, 1024 * 1024, 100)) {
+ for (int i = 0; i < numKeys; i++) {
+ Assert.assertTrue(grouper.aggregate(new IntKey(i)).isOk());
+ }
+
+ assertResultsCorrect(grouper, numKeys, 1);
+
+ File[] files = storageDir.listFiles();
+ Assert.assertNotNull(files);
+ Assert.assertEquals(
+ "all spills are tiny and should batch into a single data + dictionary file pair",
+ 2,
+ files.length
+ );
+ }
+ }
+
+ @Test
+ public void testResetClearsPendingState() throws IOException
+ {
+ try (SpillingGrouper<IntKey> grouper = makeGrouper(50, temporaryFolder.newFolder(), 1024 * 1024, 100)) {
+ for (int i = 0; i < 50; i++) {
+ Assert.assertTrue(grouper.aggregate(new IntKey(i)).isOk());
+ }
+
+ grouper.reset();
+
+ for (int i = 1000; i < 1010; i++) {
+ Assert.assertTrue(grouper.aggregate(new IntKey(i)).isOk());
+ }
+
+ try (CloseableIterator<Grouper.Entry<IntKey>> iterator = grouper.iterator(true)) {
+ int count = 0;
+ while (iterator.hasNext()) {
+ Grouper.Entry<IntKey> entry = iterator.next();
+ Assert.assertTrue(
+ "keys should be >= 1000 after reset",
+ entry.getKey().intValue() >= 1000
+ );
+ count++;
+ }
+ Assert.assertEquals(10, count);
+ }
+ }
+ }
+
+ @Test
+ public void testDiskFull() throws IOException
+ {
+ try (SpillingGrouper<IntKey> grouper = makeGrouper(50, temporaryFolder.newFolder(), 10, 100)) {
+ AggregateResult lastResult = AggregateResult.ok();
+ for (int i = 0; i < 10000 && lastResult.isOk(); i++) {
+ lastResult = grouper.aggregate(new IntKey(i));
+ }
+
+ Assert.assertFalse("should have hit disk full", lastResult.isOk());
+ Assert.assertTrue(
+ "reason should mention disk space",
+ lastResult.getReason().contains("Not enough disk space")
+ );
Review Comment:
done
--
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] fix: Optimizes SpillingGrouper for high cardinality dimension(s) GroupBy with large memory footprint aggregators (druid)
Posted by "maytasm (via GitHub)" <gi...@apache.org>.
maytasm commented on code in PR #19357:
URL: https://github.com/apache/druid/pull/19357#discussion_r3193321843
##########
processing/src/main/java/org/apache/druid/query/groupby/epinephelinae/SpillingGrouper.java:
##########
@@ -345,12 +354,103 @@ public Entry<KeyType> apply(Entry<KeyType> entry)
private void spill() throws IOException
{
+ // Stream directly to a temp file first, then check the file size. If the file is small
+ // (serialized size much smaller than the pre-allocated buffer, e.g. HLL sketches in List mode),
+ // read it back into memory for batching to avoid creating thousands of tiny disk files.
+ // If the file is already large enough, keep it on disk as-is.
+ final File file;
try (CloseableIterator<Entry<KeyType>> iterator = grouper.iterator(true)) {
- files.add(spill(iterator));
- dictionaryFiles.add(spill(keySerde.getDictionary().iterator()));
+ file = spill(iterator);
+ }
+ pendingDictionaryEntries.addAll(keySerde.getDictionary());
+ grouper.reset();
+
+ final long fileSize = file.length();
+ if (fileSize < minSpillFileSize) {
+ pendingSpillRuns.add(Files.readAllBytes(file.toPath()));
+ pendingSpillBytes += fileSize;
+ temporaryStorage.delete(file);
+
+ if (pendingSpillBytes >= minSpillFileSize) {
+ flushPendingRunsToDisk();
+ }
+ } else {
+ files.add(file);
+ dictionaryFiles.add(spill(pendingDictionaryEntries.iterator()));
+ pendingDictionaryEntries.clear();
+ }
+ }
+
+ /**
+ * Merge-sorts all pending in-memory spill runs and writes them as a single sorted file to disk.
+ * Each run is already individually sorted (from grouper.iterator(true)); this method merges them
+ * so the output file is fully sorted, as required by iterator()'s mergeSorted across files.
+ */
+ private void flushPendingRunsToDisk() throws IOException
+ {
+ if (pendingSpillRuns.isEmpty()) {
+ return;
+ }
+
+ final Comparator<Entry<KeyType>> sortComparator =
+ sortHasNonGroupingFields ? defaultOrderKeyObjComparator : keyObjComparator;
- grouper.reset();
+ final List<MappingIterator<Entry<KeyType>>> readers = new ArrayList<>(pendingSpillRuns.size());
+ try {
+ for (final byte[] runBytes : pendingSpillRuns) {
+ readers.add(spillMapper.readValues(
+ spillMapper.getFactory().createParser(new LZ4BlockInputStream(new ByteArrayInputStream(runBytes))),
+ spillMapper.getTypeFactory().constructParametricType(ReusableEntry.class, keySerde.keyClazz())
+ ));
+ }
+ final List<CloseableIterator<Entry<KeyType>>> iterators = new ArrayList<>(readers.size());
+ for (final MappingIterator<Entry<KeyType>> reader : readers) {
+ iterators.add(deserializeIterator(reader));
+ }
+ files.add(spill(CloseableIterators.mergeSorted(iterators, sortComparator)));
+ dictionaryFiles.add(spill(pendingDictionaryEntries.iterator()));
}
+ finally {
+ for (final MappingIterator<Entry<KeyType>> reader : readers) {
+ try {
+ reader.close();
+ }
+ catch (IOException e) {
+ log.warn(e, "Failed to close reader while flushing pending spill runs");
+ }
+ }
+ pendingSpillRuns.clear();
+ pendingSpillBytes = 0;
+ pendingDictionaryEntries.clear();
+ }
+ }
+
+ private CloseableIterator<Entry<KeyType>> deserializeIterator(final Iterator<Entry<KeyType>> iterator)
+ {
+ return CloseableIterators.withEmptyBaggage(
+ Iterators.transform(
+ iterator,
+ new Function<>()
+ {
+ final ReusableEntry<KeyType> reusableEntry =
+ ReusableEntry.create(keySerde, aggregatorFactories.length);
+
+ @Override
+ public Entry<KeyType> apply(Entry<KeyType> entry)
+ {
+ final Object[] deserializedValues = reusableEntry.getValues();
+ for (int i = 0; i < deserializedValues.length; i++) {
+ deserializedValues[i] = aggregatorFactories[i].deserialize(entry.getValues()[i]);
+ if (deserializedValues[i] instanceof Integer) {
Review Comment:
This is existing code. https://github.com/apache/druid/blob/master/processing/src/main/java/org/apache/druid/query/groupby/epinephelinae/SpillingGrouper.java#L319
I just refactored it to a common method to avoid repeating code / sharing between flushPendingRunsToDisk() and iterator().
--
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] fix: Optimizes SpillingGrouper for high cardinality dimension(s) GroupBy with large memory footprint aggregators (druid)
Posted by "maytasm (via GitHub)" <gi...@apache.org>.
maytasm commented on code in PR #19357:
URL: https://github.com/apache/druid/pull/19357#discussion_r3193321843
##########
processing/src/main/java/org/apache/druid/query/groupby/epinephelinae/SpillingGrouper.java:
##########
@@ -345,12 +354,103 @@ public Entry<KeyType> apply(Entry<KeyType> entry)
private void spill() throws IOException
{
+ // Stream directly to a temp file first, then check the file size. If the file is small
+ // (serialized size much smaller than the pre-allocated buffer, e.g. HLL sketches in List mode),
+ // read it back into memory for batching to avoid creating thousands of tiny disk files.
+ // If the file is already large enough, keep it on disk as-is.
+ final File file;
try (CloseableIterator<Entry<KeyType>> iterator = grouper.iterator(true)) {
- files.add(spill(iterator));
- dictionaryFiles.add(spill(keySerde.getDictionary().iterator()));
+ file = spill(iterator);
+ }
+ pendingDictionaryEntries.addAll(keySerde.getDictionary());
+ grouper.reset();
+
+ final long fileSize = file.length();
+ if (fileSize < minSpillFileSize) {
+ pendingSpillRuns.add(Files.readAllBytes(file.toPath()));
+ pendingSpillBytes += fileSize;
+ temporaryStorage.delete(file);
+
+ if (pendingSpillBytes >= minSpillFileSize) {
+ flushPendingRunsToDisk();
+ }
+ } else {
+ files.add(file);
+ dictionaryFiles.add(spill(pendingDictionaryEntries.iterator()));
+ pendingDictionaryEntries.clear();
+ }
+ }
+
+ /**
+ * Merge-sorts all pending in-memory spill runs and writes them as a single sorted file to disk.
+ * Each run is already individually sorted (from grouper.iterator(true)); this method merges them
+ * so the output file is fully sorted, as required by iterator()'s mergeSorted across files.
+ */
+ private void flushPendingRunsToDisk() throws IOException
+ {
+ if (pendingSpillRuns.isEmpty()) {
+ return;
+ }
+
+ final Comparator<Entry<KeyType>> sortComparator =
+ sortHasNonGroupingFields ? defaultOrderKeyObjComparator : keyObjComparator;
- grouper.reset();
+ final List<MappingIterator<Entry<KeyType>>> readers = new ArrayList<>(pendingSpillRuns.size());
+ try {
+ for (final byte[] runBytes : pendingSpillRuns) {
+ readers.add(spillMapper.readValues(
+ spillMapper.getFactory().createParser(new LZ4BlockInputStream(new ByteArrayInputStream(runBytes))),
+ spillMapper.getTypeFactory().constructParametricType(ReusableEntry.class, keySerde.keyClazz())
+ ));
+ }
+ final List<CloseableIterator<Entry<KeyType>>> iterators = new ArrayList<>(readers.size());
+ for (final MappingIterator<Entry<KeyType>> reader : readers) {
+ iterators.add(deserializeIterator(reader));
+ }
+ files.add(spill(CloseableIterators.mergeSorted(iterators, sortComparator)));
+ dictionaryFiles.add(spill(pendingDictionaryEntries.iterator()));
}
+ finally {
+ for (final MappingIterator<Entry<KeyType>> reader : readers) {
+ try {
+ reader.close();
+ }
+ catch (IOException e) {
+ log.warn(e, "Failed to close reader while flushing pending spill runs");
+ }
+ }
+ pendingSpillRuns.clear();
+ pendingSpillBytes = 0;
+ pendingDictionaryEntries.clear();
+ }
+ }
+
+ private CloseableIterator<Entry<KeyType>> deserializeIterator(final Iterator<Entry<KeyType>> iterator)
+ {
+ return CloseableIterators.withEmptyBaggage(
+ Iterators.transform(
+ iterator,
+ new Function<>()
+ {
+ final ReusableEntry<KeyType> reusableEntry =
+ ReusableEntry.create(keySerde, aggregatorFactories.length);
+
+ @Override
+ public Entry<KeyType> apply(Entry<KeyType> entry)
+ {
+ final Object[] deserializedValues = reusableEntry.getValues();
+ for (int i = 0; i < deserializedValues.length; i++) {
+ deserializedValues[i] = aggregatorFactories[i].deserialize(entry.getValues()[i]);
+ if (deserializedValues[i] instanceof Integer) {
Review Comment:
This is existing code. https://github.com/apache/druid/blob/master/processing/src/main/java/org/apache/druid/query/groupby/epinephelinae/SpillingGrouper.java#L319
I just refactored it to a common method to avoid repeating code.
--
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] fix: Optimizes SpillingGrouper for high cardinality dimension(s) GroupBy with large memory footprint aggregators (druid)
Posted by "jtuglu1 (via GitHub)" <gi...@apache.org>.
jtuglu1 commented on code in PR #19357:
URL: https://github.com/apache/druid/pull/19357#discussion_r3196625224
##########
processing/src/main/java/org/apache/druid/query/groupby/epinephelinae/SpillingGrouper.java:
##########
@@ -345,12 +354,103 @@ public Entry<KeyType> apply(Entry<KeyType> entry)
private void spill() throws IOException
{
+ // Stream directly to a temp file first, then check the file size. If the file is small
+ // (serialized size much smaller than the pre-allocated buffer, e.g. HLL sketches in List mode),
+ // read it back into memory for batching to avoid creating thousands of tiny disk files.
+ // If the file is already large enough, keep it on disk as-is.
+ final File file;
try (CloseableIterator<Entry<KeyType>> iterator = grouper.iterator(true)) {
- files.add(spill(iterator));
- dictionaryFiles.add(spill(keySerde.getDictionary().iterator()));
+ file = spill(iterator);
+ }
+ pendingDictionaryEntries.addAll(keySerde.getDictionary());
+ grouper.reset();
+
+ final long fileSize = file.length();
Review Comment:
That's fine – I mention point #1 already as another reason against this, but I think #2 is really holding the weight here. #3 isn't really too relevant IMO because you can just do `pendingSpillRuns.ensureCapacity(fileSize);` (with the heuristic that files will be approximately similarly sized).
--
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] Optimizes SpillingGrouper for high cardinality group by dimensions (druid)
Posted by "github-advanced-security[bot] (via GitHub)" <gi...@apache.org>.
github-advanced-security[bot] commented on code in PR #19357:
URL: https://github.com/apache/druid/pull/19357#discussion_r3114503272
##########
processing/src/main/java/org/apache/druid/query/groupby/epinephelinae/SpillingGrouper.java:
##########
@@ -345,12 +380,122 @@
private void spill() throws IOException
{
+ // Serialize the buffer flush to heap memory rather than directly to disk. When the pre-allocated
+ // in-memory size per row (e.g. ~131KB for ThetaSketch) is much larger than the compact serialized
+ // size (e.g. ~24 bytes when each key has been seen only a few times), each flush produces a tiny
+ // byte array. Batching these in memory until MIN_SPILL_FILE_BYTES is reached avoids creating
+ // thousands of tiny disk files, which would each require a live file handle during the merge phase.
+ final byte[] runBytes;
try (CloseableIterator<Entry<KeyType>> iterator = grouper.iterator(true)) {
- files.add(spill(iterator));
- dictionaryFiles.add(spill(keySerde.getDictionary().iterator()));
+ runBytes = serializeToBytes(iterator);
+ }
+ pendingSpillRuns.add(runBytes);
+ pendingSpillBytes += runBytes.length;
+ pendingDictionaryEntries.addAll(keySerde.getDictionary());
+ grouper.reset();
- grouper.reset();
+ if (pendingSpillBytes >= MIN_SPILL_FILE_BYTES) {
+ flushPendingRunsToDisk(true);
+ }
+ }
+
+ /**
+ * Merge-sorts all pending in-memory spill runs and writes them as a single sorted file to disk.
+ * Reading from heap memory (ByteArrayInputStream) is orders of magnitude faster than reading
+ * from disk, so the merge cost here is negligible even for many accumulated runs.
+ */
+ private void flushPendingRunsToDisk(final boolean sorted) throws IOException
+ {
+ if (pendingSpillRuns.isEmpty()) {
+ return;
+ }
+
+ final List<MappingIterator<Entry<KeyType>>> readers = new ArrayList<>(pendingSpillRuns.size());
+ try {
+ for (final byte[] runBytes : pendingSpillRuns) {
+ readers.add(spillMapper.readValues(
+ spillMapper.getFactory().createParser(new LZ4BlockInputStream(new ByteArrayInputStream(runBytes))),
Review Comment:
## CodeQL / Deprecated method or constructor invocation
Invoking [LZ4BlockInputStream.LZ4BlockInputStream](1) should be avoided because it has been deprecated.
[Show more details](https://github.com/apache/druid/security/code-scanning/11101)
--
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] fix: Optimizes SpillingGrouper for high cardinality dimension(s) GroupBy with large memory footprint aggregators (druid)
Posted by "jtuglu1 (via GitHub)" <gi...@apache.org>.
jtuglu1 commented on code in PR #19357:
URL: https://github.com/apache/druid/pull/19357#discussion_r3196625224
##########
processing/src/main/java/org/apache/druid/query/groupby/epinephelinae/SpillingGrouper.java:
##########
@@ -345,12 +354,103 @@ public Entry<KeyType> apply(Entry<KeyType> entry)
private void spill() throws IOException
{
+ // Stream directly to a temp file first, then check the file size. If the file is small
+ // (serialized size much smaller than the pre-allocated buffer, e.g. HLL sketches in List mode),
+ // read it back into memory for batching to avoid creating thousands of tiny disk files.
+ // If the file is already large enough, keep it on disk as-is.
+ final File file;
try (CloseableIterator<Entry<KeyType>> iterator = grouper.iterator(true)) {
- files.add(spill(iterator));
- dictionaryFiles.add(spill(keySerde.getDictionary().iterator()));
+ file = spill(iterator);
+ }
+ pendingDictionaryEntries.addAll(keySerde.getDictionary());
+ grouper.reset();
+
+ final long fileSize = file.length();
Review Comment:
I mention point #1 already, but I think #2 is really holding the weight here. #3 isn't really too relevant IMO because you can just do `pendingSpillRuns.ensureCapacity(fileSize);` (with the heuristic that files will be approximately similarly sized).
--
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] fix: Optimizes SpillingGrouper for high cardinality dimension(s) GroupBy with large memory footprint aggregators (druid)
Posted by "maytasm (via GitHub)" <gi...@apache.org>.
maytasm commented on code in PR #19357:
URL: https://github.com/apache/druid/pull/19357#discussion_r3192727929
##########
processing/src/main/java/org/apache/druid/query/groupby/epinephelinae/SpillingGrouper.java:
##########
@@ -293,6 +320,22 @@ public void setSpillingAllowed(final boolean spillingAllowed)
@Override
public CloseableIterator<Entry<KeyType>> iterator(final boolean sorted)
{
+ // Flush any runs that did not reach MIN_SPILL_FILE_BYTES during the spill phase.
+ try {
+ flushPendingRunsToDisk();
Review Comment:
Looks like the reason iterator(true) is hardcoded in RowBasedGrouperHelper:634 is that the merge layer above it relies on sorted input — CombiningIterator in ConcurrentGrouper and the broker merge both detect duplicate keys by comparing consecutive sorted entries. So sorted=true is for merge correctness, not output ordering.
--
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] fix: Optimizes SpillingGrouper for high cardinality dimension(s) GroupBy with large memory footprint aggregators (druid)
Posted by "jtuglu1 (via GitHub)" <gi...@apache.org>.
jtuglu1 commented on PR #19357:
URL: https://github.com/apache/druid/pull/19357#issuecomment-4375284397
Before I review, can we please include benchmarks before/after for this change?
--
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] fix: Optimizes SpillingGrouper for high cardinality dimension(s) GroupBy with large memory footprint aggregators (druid)
Posted by "jtuglu1 (via GitHub)" <gi...@apache.org>.
jtuglu1 commented on code in PR #19357:
URL: https://github.com/apache/druid/pull/19357#discussion_r3192193001
##########
processing/src/main/java/org/apache/druid/query/groupby/epinephelinae/SpillingGrouper.java:
##########
Review Comment:
Now that we are properly tracking bytes during deletion, I wonder if we should also look at deleting `dictionaryFiles` as well here.
##########
processing/src/main/java/org/apache/druid/query/groupby/epinephelinae/SpillingGrouper.java:
##########
@@ -72,6 +74,17 @@ public class SpillingGrouper<KeyType> implements Grouper<KeyType>
"Maximum number of spill files reached for this query. Try raising druid.query.groupBy.maxSpillFileCount."
);
+ /**
+ * Minimum number of serialized bytes that must accumulate across pending in-memory spill runs before they are
+ * flushed as a single file to disk. Aggregators like ThetaSketch pre-allocate a large fixed buffer per row
+ * (e.g. ~131KB for ThetaSketch(K=16384)), causing the in-memory grouper to flush frequently. However, when
+ * each key has been seen only a few times, the sketch serializes to just a handful of bytes in compact form.
+ * Without batching, this produces thousands of tiny spill files. By accumulating runs in heap memory first
+ * and writing to disk only once this threshold is reached, we avoid that explosion in file count without any
+ * extra disk I/O for small spills.
+ */
+ private static final long MIN_SPILL_FILE_BYTES = 1024 * 1024L; // 1MB
Review Comment:
let's make this a config. sketch sizes can vary with the columns they cover and the k-values.
##########
processing/src/main/java/org/apache/druid/query/groupby/epinephelinae/LimitedTemporaryStorage.java:
##########
@@ -121,13 +122,15 @@ public void delete(final File file)
{
synchronized (files) {
if (files.contains(file)) {
+ final long fileSize = file.length();
try {
Files.delete(file.toPath());
}
catch (IOException e) {
log.warn(e, "Cannot delete file: %s", file);
}
files.remove(file);
+ bytesUsed.addAndGet(-fileSize);
Review Comment:
if this delete fails, I'd rather not have our accounting also be inaccurate – can we put this in the try after the delete?
##########
processing/src/main/java/org/apache/druid/query/groupby/epinephelinae/SpillingGrouper.java:
##########
@@ -293,6 +320,22 @@ public void setSpillingAllowed(final boolean spillingAllowed)
@Override
public CloseableIterator<Entry<KeyType>> iterator(final boolean sorted)
{
+ // Flush any runs that did not reach MIN_SPILL_FILE_BYTES during the spill phase.
+ try {
+ flushPendingRunsToDisk();
Review Comment:
we should see what the overhead of `sorted=false` is here. If we don't need a sorted run as the end result, we can just do a simple concat to avoid the decompress/re-sort overhead of merge sort. I think we might need to condition this on `sortHasNonGroupingFields=false` too.
--
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] fix: Optimizes SpillingGrouper for high cardinality dimension(s) GroupBy with large memory footprint aggregators (druid)
Posted by "jtuglu1 (via GitHub)" <gi...@apache.org>.
jtuglu1 commented on code in PR #19357:
URL: https://github.com/apache/druid/pull/19357#discussion_r3192095332
##########
processing/src/main/java/org/apache/druid/query/groupby/epinephelinae/LimitedTemporaryStorage.java:
##########
@@ -121,13 +122,15 @@ public void delete(final File file)
{
synchronized (files) {
if (files.contains(file)) {
+ final long fileSize = file.length();
try {
Files.delete(file.toPath());
}
catch (IOException e) {
log.warn(e, "Cannot delete file: %s", file);
}
files.remove(file);
+ bytesUsed.addAndGet(-fileSize);
Review Comment:
if this delete fails, I'd rather not have our accounting also be inaccurate – can we put this in the try after the delete? IMO, our accounting should always be an overestimate to avoid actual disk space issues where worse exceptions can happen (e.g. I'd rather throw a query error cleanly and debug that then see historicals/peons start crashing due to underestimated statistics leading to disk space errors).
--
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] fix: Optimizes SpillingGrouper for high cardinality dimension(s) GroupBy with large memory footprint aggregators (druid)
Posted by "jtuglu1 (via GitHub)" <gi...@apache.org>.
jtuglu1 commented on code in PR #19357:
URL: https://github.com/apache/druid/pull/19357#discussion_r3192761913
##########
processing/src/main/java/org/apache/druid/query/groupby/epinephelinae/SpillingGrouper.java:
##########
@@ -293,6 +320,22 @@ public void setSpillingAllowed(final boolean spillingAllowed)
@Override
public CloseableIterator<Entry<KeyType>> iterator(final boolean sorted)
{
+ // Flush any runs that did not reach MIN_SPILL_FILE_BYTES during the spill phase.
+ try {
+ flushPendingRunsToDisk();
Review Comment:
could we add a comment for this on the method? Without looking at all the call-sites, it's tough to see why we ignore the sort param.
--
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] fix: Optimizes SpillingGrouper for high cardinality dimension(s) GroupBy with large memory footprint aggregators (druid)
Posted by "jtuglu1 (via GitHub)" <gi...@apache.org>.
jtuglu1 commented on code in PR #19357:
URL: https://github.com/apache/druid/pull/19357#discussion_r3192785270
##########
processing/src/main/java/org/apache/druid/query/groupby/epinephelinae/SpillingGrouper.java:
##########
@@ -345,12 +354,103 @@ public Entry<KeyType> apply(Entry<KeyType> entry)
private void spill() throws IOException
{
+ // Stream directly to a temp file first, then check the file size. If the file is small
+ // (serialized size much smaller than the pre-allocated buffer, e.g. HLL sketches in List mode),
+ // read it back into memory for batching to avoid creating thousands of tiny disk files.
+ // If the file is already large enough, keep it on disk as-is.
+ final File file;
try (CloseableIterator<Entry<KeyType>> iterator = grouper.iterator(true)) {
- files.add(spill(iterator));
- dictionaryFiles.add(spill(keySerde.getDictionary().iterator()));
+ file = spill(iterator);
+ }
+ pendingDictionaryEntries.addAll(keySerde.getDictionary());
+ grouper.reset();
+
+ final long fileSize = file.length();
Review Comment:
I wonder if there might be an optimization here to reserve entire needed capacity for the array upfront to prevent it needing to double up to sufficient capacity. Since we only .clear() and not something like `trimToSize()` this resizing might just be amortized across all the files, especially if they are of similar size and not amount to much.
--
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] fix: Optimizes SpillingGrouper for high cardinality dimension(s) GroupBy with large memory footprint aggregators (druid)
Posted by "jtuglu1 (via GitHub)" <gi...@apache.org>.
jtuglu1 commented on code in PR #19357:
URL: https://github.com/apache/druid/pull/19357#discussion_r3192840232
##########
processing/src/main/java/org/apache/druid/query/groupby/epinephelinae/LimitedTemporaryStorage.java:
##########
Review Comment:
@gianm might be able to comment on why this not put inside the try 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] fix: Optimizes SpillingGrouper for high cardinality dimension(s) GroupBy with large memory footprint aggregators (druid)
Posted by "FrankChen021 (via GitHub)" <gi...@apache.org>.
FrankChen021 commented on code in PR #19357:
URL: https://github.com/apache/druid/pull/19357#discussion_r3189545609
##########
processing/src/main/java/org/apache/druid/query/groupby/epinephelinae/SpillingGrouper.java:
##########
@@ -345,14 +362,105 @@ public Entry<KeyType> apply(Entry<KeyType> entry)
private void spill() throws IOException
{
+ // Stream directly to a temp file first, then check the file size. If the file is small
+ // (serialized size much smaller than the pre-allocated buffer, e.g. HLL sketches in List mode),
+ // read it back into memory for batching to avoid creating thousands of tiny disk files.
+ // If the file is already large enough, keep it on disk as-is.
+ final File file;
try (CloseableIterator<Entry<KeyType>> iterator = grouper.iterator(true)) {
- files.add(spill(iterator));
- dictionaryFiles.add(spill(keySerde.getDictionary().iterator()));
+ file = spill(iterator);
+ }
+ pendingDictionaryEntries.addAll(keySerde.getDictionary());
+ grouper.reset();
+
+ final long fileSize = file.length();
+ if (fileSize < MIN_SPILL_FILE_BYTES) {
+ pendingSpillRuns.add(Files.readAllBytes(file.toPath()));
Review Comment:
[P1] Deleted staging spills still consume the disk quota
This path writes every small spill through LimitedTemporaryStorage, reads it back into heap, then deletes the temp file. LimitedTemporaryStorage.delete only removes the file from the file set; it does not decrement bytesUsed, and LimitedOutputStream.grab has already charged those bytes against maxOnDiskStorage. As a result, high-cardinality small-spill queries can hit TemporaryStorageFullException even though those staging files were deleted and no persistent spill file exists yet, and later flushes double-charge the same data when writing the merged file. This undermines the batching optimization and can fail queries well below their configured on-disk limit; small runs should avoid charging LimitedTemporaryStorage or the accounting needs to refund deleted staging bytes.
--
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] fix: Optimizes SpillingGrouper for high cardinality dimension(s) GroupBy with large memory footprint aggregators (druid)
Posted by "jtuglu1 (via GitHub)" <gi...@apache.org>.
jtuglu1 commented on code in PR #19357:
URL: https://github.com/apache/druid/pull/19357#discussion_r3192095332
##########
processing/src/main/java/org/apache/druid/query/groupby/epinephelinae/LimitedTemporaryStorage.java:
##########
@@ -121,13 +122,15 @@ public void delete(final File file)
{
synchronized (files) {
if (files.contains(file)) {
+ final long fileSize = file.length();
try {
Files.delete(file.toPath());
}
catch (IOException e) {
log.warn(e, "Cannot delete file: %s", file);
}
files.remove(file);
+ bytesUsed.addAndGet(-fileSize);
Review Comment:
if this delete fails, I'd rather not have our accounting also be inaccurate – can we put this in the try after the delete? IMO, our accounting should always be an overestimate to avoid actual disk space issues where worse exceptions can happen (e.g. I'd rather throw a query error cleanly and debug that then see historicals/peons start crashing due to disk space errors).
--
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] fix: Optimizes SpillingGrouper for high cardinality dimension(s) GroupBy with large memory footprint aggregators (druid)
Posted by "maytasm (via GitHub)" <gi...@apache.org>.
maytasm commented on code in PR #19357:
URL: https://github.com/apache/druid/pull/19357#discussion_r3192535630
##########
processing/src/main/java/org/apache/druid/query/groupby/epinephelinae/SpillingGrouper.java:
##########
@@ -293,6 +320,22 @@ public void setSpillingAllowed(final boolean spillingAllowed)
@Override
public CloseableIterator<Entry<KeyType>> iterator(final boolean sorted)
{
+ // Flush any runs that did not reach MIN_SPILL_FILE_BYTES during the spill phase.
+ try {
+ flushPendingRunsToDisk();
Review Comment:
Overhead breakdown of flushPendingRunsToDisk():
1. LZ4 decompress — fast (~GB/s)
2. JSON parse — moderate (dominant cost)
3. Merge-sort comparison — cheap (O(N log K), K = few pending runs)
4. JSON serialize — moderate (dominant cost)
5. LZ4 compress + write — fast
Replacing mergeSorted with concat (step 3) saves very little — the JSON serde in steps 2+4 dominates.
The other approach is to write each pending run's raw byte[] sequentially into one file (each is already a complete LZ4+JSON stream). At read time, create one iterator per sub-stream. The catch with this approach is that LZ4BlockInputStream stops at each stream boundary, so reading N streams from one file requires creating N LZ4BlockInputStream instances on the same underlying FileInputStream. LZ4BlockInputStream allocates a single decompression buffer (default 64KB, matching LZ4BlockOutputStream's default block size). With a lot of spills (the scenario is are trying to fix with large aggregators + high cardinality group bys), these LZ4BlockInputStream will adds up resulting in OOM like before.
The merge-sort serde cost in flushPendingRunsToDisk() is the price we pay for keeping both file count and read-time memory bounded. And as noted earlier, replacing mergeSorted with concat alone saves very little since JSON serde dominates the cost.
--
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] fix: Optimizes SpillingGrouper for high cardinality dimension(s) GroupBy with large memory footprint aggregators (druid)
Posted by "maytasm (via GitHub)" <gi...@apache.org>.
maytasm commented on code in PR #19357:
URL: https://github.com/apache/druid/pull/19357#discussion_r3192711641
##########
processing/src/main/java/org/apache/druid/query/groupby/epinephelinae/SpillingGrouper.java:
##########
Review Comment:
Right — deleteFiles() only deletes data files but not dictionaryFiles. Those are also tracked by LimitedTemporaryStorage and consume disk quota. Fixed.
--
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] fix: Optimizes SpillingGrouper for high cardinality dimension(s) GroupBy with large memory footprint aggregators (druid)
Posted by "maytasm (via GitHub)" <gi...@apache.org>.
maytasm commented on PR #19357:
URL: https://github.com/apache/druid/pull/19357#issuecomment-4383170063
> Before I review, can we please include benchmarks before/after for this change?
@jtuglu1 Added benchmark results.
--
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] fix: Optimizes SpillingGrouper for high cardinality dimension(s) GroupBy with large memory footprint aggregators (druid)
Posted by "jtuglu1 (via GitHub)" <gi...@apache.org>.
jtuglu1 commented on code in PR #19357:
URL: https://github.com/apache/druid/pull/19357#discussion_r3192785270
##########
processing/src/main/java/org/apache/druid/query/groupby/epinephelinae/SpillingGrouper.java:
##########
@@ -345,12 +354,103 @@ public Entry<KeyType> apply(Entry<KeyType> entry)
private void spill() throws IOException
{
+ // Stream directly to a temp file first, then check the file size. If the file is small
+ // (serialized size much smaller than the pre-allocated buffer, e.g. HLL sketches in List mode),
+ // read it back into memory for batching to avoid creating thousands of tiny disk files.
+ // If the file is already large enough, keep it on disk as-is.
+ final File file;
try (CloseableIterator<Entry<KeyType>> iterator = grouper.iterator(true)) {
- files.add(spill(iterator));
- dictionaryFiles.add(spill(keySerde.getDictionary().iterator()));
+ file = spill(iterator);
+ }
+ pendingDictionaryEntries.addAll(keySerde.getDictionary());
+ grouper.reset();
+
+ final long fileSize = file.length();
Review Comment:
I wonder if there might be an optimization here to reserve entire needed capacity for the array upfront to prevent it needing to double up to sufficient capacity. I think it should noop on subsequent calls too once capacity is allocated. Since we only .clear() and not something like `trimToSize()` this resizing might just be amortized across all the files, especially if they are of similar size and not amount to much.
##########
processing/src/main/java/org/apache/druid/query/groupby/epinephelinae/SpillingGrouper.java:
##########
@@ -345,12 +354,103 @@ public Entry<KeyType> apply(Entry<KeyType> entry)
private void spill() throws IOException
{
+ // Stream directly to a temp file first, then check the file size. If the file is small
+ // (serialized size much smaller than the pre-allocated buffer, e.g. HLL sketches in List mode),
+ // read it back into memory for batching to avoid creating thousands of tiny disk files.
+ // If the file is already large enough, keep it on disk as-is.
+ final File file;
try (CloseableIterator<Entry<KeyType>> iterator = grouper.iterator(true)) {
- files.add(spill(iterator));
- dictionaryFiles.add(spill(keySerde.getDictionary().iterator()));
+ file = spill(iterator);
+ }
+ pendingDictionaryEntries.addAll(keySerde.getDictionary());
+ grouper.reset();
+
+ final long fileSize = file.length();
Review Comment:
I wonder if there might be an optimization here to reserve entire needed capacity for the array list upfront to prevent it needing to double up to sufficient capacity. I think it should noop on subsequent calls too once capacity is allocated. Since we only .clear() and not something like `trimToSize()` this resizing might just be amortized across all the files, especially if they are of similar size and not amount to much.
--
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] fix: Optimizes SpillingGrouper for high cardinality dimension(s) GroupBy with large memory footprint aggregators (druid)
Posted by "jtuglu1 (via GitHub)" <gi...@apache.org>.
jtuglu1 commented on code in PR #19357:
URL: https://github.com/apache/druid/pull/19357#discussion_r3192785270
##########
processing/src/main/java/org/apache/druid/query/groupby/epinephelinae/SpillingGrouper.java:
##########
@@ -345,12 +354,103 @@ public Entry<KeyType> apply(Entry<KeyType> entry)
private void spill() throws IOException
{
+ // Stream directly to a temp file first, then check the file size. If the file is small
+ // (serialized size much smaller than the pre-allocated buffer, e.g. HLL sketches in List mode),
+ // read it back into memory for batching to avoid creating thousands of tiny disk files.
+ // If the file is already large enough, keep it on disk as-is.
+ final File file;
try (CloseableIterator<Entry<KeyType>> iterator = grouper.iterator(true)) {
- files.add(spill(iterator));
- dictionaryFiles.add(spill(keySerde.getDictionary().iterator()));
+ file = spill(iterator);
+ }
+ pendingDictionaryEntries.addAll(keySerde.getDictionary());
+ grouper.reset();
+
+ final long fileSize = file.length();
Review Comment:
I wonder if there might be an optimization here to reserve entire needed capacity for the array list upfront to prevent it needing to repeatedly double up to sufficient capacity. I think it should noop on subsequent calls too once capacity is allocated. Since we only .clear() and not something like `trimToSize()` this resizing might just be amortized across all the files, especially if they are of similar size and not amount to much.
--
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