Subtotal queries give incorrect results if query has a limit spec by loquisgon · Pull Request #10685 · apache/druid

loquisgon · 2020-12-16T18:46:23Z

Description
For calculating the results for a subtotal spec, we reuse the buffer iterator that was used in the base query. However, not all the implementations of such iterators are reusable. Specifically, LimitedBufferHashGrouper#iterator is one of those non-reusable iterators. This results in subtotal specs not working if there is also a limit used as part of the query.

This PR deeps copy some shared state in the iterator makeHeapIteratormethod to make sure that it is reusable. It also added a unit test to verify that it can be reused now.

To summarize:

Issue is that makeHeapIteratormethod copies overwrite some shared state
Solution in this ticket is to actually copy the state so that the iterator is reusable

This PR has:

been self-reviewed.
added documentation for new or modified features or behaviors.
added unit tests or modified existing tests to cover new code paths, ensuring the threshold for code coverage is met.

…rator()

loquisgon · 2020-12-16T20:44:47Z

sql/src/main/java/org/apache/druid/sql/SqlLifecycle.java

      setParameters(SqlQuery.getParameterList(parameters));
      planAndAuthorize(authenticationResult);
-      result = execute();
+      result = execute(); //agt


I will remove this debug comment

Fixed...along with some style cleanup

abhishekagarwal87

Looks mostly good except for unit tests.

abhishekagarwal87 · 2020-12-17T09:29:52Z

...ing/src/main/java/org/apache/druid/query/groupby/epinephelinae/LimitedBufferHashGrouper.java

  private CloseableIterator<Entry<KeyType>> makeHeapIterator()
  {
    final int initialHeapSize = offsetHeap.getHeapSize();
+    ByteBufferMinMaxOffsetHeap newHeap = offsetHeap.copy();


can you add a comment here on why we are creating another copy?

Done in next commit

abhishekagarwal87 · 2020-12-17T09:30:10Z

...g/src/main/java/org/apache/druid/query/groupby/epinephelinae/ByteBufferMinMaxOffsetHeap.java

    this.heapIndexUpdater = heapIndexUpdater;
  }

+  public ByteBufferMinMaxOffsetHeap copy()


Let's also add some unit tests for this method.

Done in next commit

abhishekagarwal87 · 2020-12-17T09:31:11Z

...ing/src/main/java/org/apache/druid/query/groupby/epinephelinae/LimitedBufferHashGrouper.java

@@ -318,6 +336,8 @@ public void close()
  private CloseableIterator<Entry<KeyType>> makeHeapIterator()


Please add some unit tests in the LimitedBufferHashGrouperTest where an iterator is created multiple times and still returns the same results.

Done in next commit

abhishekagarwal87 · 2020-12-23T06:14:50Z

...src/test/java/org/apache/druid/query/groupby/epinephelinae/LimitedBufferHashGrouperTest.java

+    // With minimum buffer size, after the first swap, every new key added will result in a swap
+    if (NullHandling.replaceWithDefault()) {
+      Assert.assertEquals(224, grouper.getGrowthCount());
+      Assert.assertEquals(104, grouper.getSize());
+      Assert.assertEquals(209, grouper.getBuckets());
+      Assert.assertEquals(104, grouper.getMaxSize());
+    } else {
+      Assert.assertEquals(899, grouper.getGrowthCount());
+      Assert.assertEquals(101, grouper.getSize());
+      Assert.assertEquals(202, grouper.getBuckets());
+      Assert.assertEquals(101, grouper.getMaxSize());
+    }
+    Assert.assertEquals(100, grouper.getLimit());
+
+    // Aggregate slightly different row
+    // Since these keys are smaller, they will evict the previous 100 top entries
+    // First 100 of these new rows will be the expected results.
+    columnSelectorFactory.setRow(new MapBasedRow(0, ImmutableMap.of("value", 11L)));
+    for (int i = 0; i < numRows; i++) {
+      Assert.assertTrue(String.valueOf(i), grouper.aggregate(i).isOk());
+    }
+    if (NullHandling.replaceWithDefault()) {
+      Assert.assertEquals(474, grouper.getGrowthCount());
+      Assert.assertEquals(104, grouper.getSize());
+      Assert.assertEquals(209, grouper.getBuckets());
+      Assert.assertEquals(104, grouper.getMaxSize());
+    } else {
+      Assert.assertEquals(1899, grouper.getGrowthCount());
+      Assert.assertEquals(101, grouper.getSize());
+      Assert.assertEquals(202, grouper.getBuckets());
+      Assert.assertEquals(101, grouper.getMaxSize());
+    }
+    Assert.assertEquals(100, grouper.getLimit());


Suggested change

// With minimum buffer size, after the first swap, every new key added will result in a swap

if (NullHandling.replaceWithDefault()) {

Assert.assertEquals(224, grouper.getGrowthCount());

Assert.assertEquals(104, grouper.getSize());

Assert.assertEquals(209, grouper.getBuckets());

Assert.assertEquals(104, grouper.getMaxSize());

} else {

Assert.assertEquals(899, grouper.getGrowthCount());

Assert.assertEquals(101, grouper.getSize());

Assert.assertEquals(202, grouper.getBuckets());

Assert.assertEquals(101, grouper.getMaxSize());

}

Assert.assertEquals(100, grouper.getLimit());

// Aggregate slightly different row

// Since these keys are smaller, they will evict the previous 100 top entries

// First 100 of these new rows will be the expected results.

columnSelectorFactory.setRow(new MapBasedRow(0, ImmutableMap.of("value", 11L)));

for (int i = 0; i < numRows; i++) {

Assert.assertTrue(String.valueOf(i), grouper.aggregate(i).isOk());

}

if (NullHandling.replaceWithDefault()) {

Assert.assertEquals(474, grouper.getGrowthCount());

Assert.assertEquals(104, grouper.getSize());

Assert.assertEquals(209, grouper.getBuckets());

Assert.assertEquals(104, grouper.getMaxSize());

} else {

Assert.assertEquals(1899, grouper.getGrowthCount());

Assert.assertEquals(101, grouper.getSize());

Assert.assertEquals(202, grouper.getBuckets());

Assert.assertEquals(101, grouper.getMaxSize());

}

Assert.assertEquals(100, grouper.getLimit());

these look unrelated to the unit test and are already being verified in a different unit test.

abhishekagarwal87

LGTM. Thank you @loquisgon

suneet-s · 2021-01-05T00:53:00Z

...g/src/main/java/org/apache/druid/query/groupby/epinephelinae/ByteBufferMinMaxOffsetHeap.java

+    // deep copy buf
+    ByteBuffer buffer = ByteBuffer.allocateDirect(buf.capacity());
+
+    int bufPosition = buf.position();
+    int bufLimit = buf.limit();
+    buf.rewind();
+
+    buffer.put(buf);
+    buffer.position(bufPosition);
+    buffer.limit(bufLimit);
+
+    buf.position(bufPosition);
+    buf.limit(bufLimit);


nit: It seems like this should be moved in to a separate function since it's used in 2 places. I tried a little googling, and didn't find any libraries that implement deep copying of ByteBuffers which I thought was surprising.

Looking at an implementation I found in http://www.java2s.com/example/java/java.nio/performs-a-deep-copy-on-a-byte-buffer.html, it looks like this implementation does not copy the order of the original ByteBuffer. I can't tell if this is important or not. What do you think?

Yeah, agreed. Code should be refactored to abstract out the duplicated block.

It won't hurt to copy the order I guess. I will look more into it.

jihoonson · 2021-01-05T01:06:36Z

...g/src/main/java/org/apache/druid/query/groupby/epinephelinae/ByteBufferMinMaxOffsetHeap.java

+            .orElse(null);
+
+    // deep copy buf
+    ByteBuffer buffer = ByteBuffer.allocateDirect(buf.capacity());


GroupBy queries use both processing buffers and merge buffers. The former is used when you compute per-segment results while the later is for all other purposes (merging per-segment results, computing subtotals, etc). Especially the broker only uses the merge buffers to process subtotals and subqueries.

The merge buffers are maintained in a BlockingPool to manage the memory usage in brokers and historicals. This is important to avoid query failures due to OOM errors. Here, you should not allocate memory directly, but get one from the merge buffer pool. Check out how the merge buffer is currently acquired.

Yeah, I agree. My biggest concern with this code is that it is not obvious to the caller that creating new iterators of the type being changed here will allocate new off-heap memory in an unbounded fashion. This is ok if we think that not "too many" copies will be done but I cannot affirm that.

Yeah, I agree. My biggest concern with this code is that it is not obvious to the caller that creating new iterators of the type being changed here will allocate new off-heap memory in an unbounded fashion. This is ok if we think that not "too many" copies will be done but I cannot affirm that.

It seems pretty dangerous to me as users are not expected to be aware of this behavior. The blocking merge buffer pool is used to avoid unexpectedly using "too many" copies at the same time like in this problematic scenario.

By the way, on the second look, I'm wondering why we copy the buffer instead of fixing the iterator of LimitedBufferHashGrouper to be re-iterable. This seems a better fix to me.

Talked with @loganlinn offline, we agreed fixing LimitedBufferHashGrouper would be a better fix.

suneet-s · 2021-01-22T17:40:33Z

Fixed by #10743

Agustin Gonzalez added 2 commits December 16, 2020 11:44

Hides protected data member defined in base class

62d254b

Deep copy min-max related data structures in order to support new ite…

2b8161c

…rator()

loquisgon changed the title ~~Imply 5009 subtotal limit~~ Subtotal queries give incorrect results if query has a limit spec Dec 16, 2020

loquisgon commented Dec 16, 2020

View reviewed changes

Style cleanup

c0c1d94

abhishekagarwal87 reviewed Dec 17, 2020

View reviewed changes

clintropolis added Area - Querying Bug labels Dec 18, 2020

Addressing PR comments

319fd3a

abhishekagarwal87 reviewed Dec 23, 2020

View reviewed changes

Remove unnecessary lines according to PR comments

7e2f628

abhishekagarwal87 approved these changes Dec 29, 2020

View reviewed changes

suneet-s reviewed Jan 5, 2021

View reviewed changes

jihoonson reviewed Jan 5, 2021

View reviewed changes

clintropolis added the WIP label Jan 6, 2021

suneet-s closed this Jan 22, 2021

		@@ -318,6 +336,8 @@ public void close()
		private CloseableIterator<Entry<KeyType>> makeHeapIterator()

Conversation

loquisgon commented Dec 16, 2020 • edited by clintropolis Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

abhishekagarwal87 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

abhishekagarwal87 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

suneet-s commented Jan 22, 2021

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants

loquisgon commented Dec 16, 2020 •

edited by clintropolis

Loading