Druid: Extension: oak-incremental-index: Low resource (RAM and CPU) incremental-index implementation using off-heap key/value map (OakMap)

[liran-funaro]

Motivation

The current incremental-index implementations (on-heap and off-heap) suffer from poor memory utilization and sub-optimal performance. In some ingestion scenarios, we observed 200% memory overhead and 70% runtime overhead that both are attributed to the GC mechanism. This is mainly due to the large number of metadata objects created by Java’s ConcurrentSkipList (CSL).

Proposed changes

We implemented an alternative incremental-index (OakIncrementalIndex) that has two main attributes that are different from the current implementations:

1. It stores both keys and values off-heap (as opposed to the off-heap implementation that stores only the values off-heap).
2. It is based on OakMap [1] instead of Java’s ConcurrentSkipList (CSL).

These two changes significantly reduce the number of heap-objects and thus decrease dramatically the GC’s memory and performance overhead.

This implementation was proposed before (#5698 and #7676). This issue expands on these with system-level experiments results, as well as more comprehensive component-level benchmarks results (as requested by the community). In addition to improved performance compared to older versions.

[1] Oak: a Scalable Off-Heap Allocated Key-Value Map. _ACM Conference on Principles and Practices of Parallel Programming (PPoPP) ‘2020_.

Rationale

Our implementation (OakIncrementalIndex) instantiates a sub-linear number of objects with respect to the number of rows in the incremental-index, as opposed to a linear number of metadata objects that are instantiated by CSL. For typical Incremental-Index sizes (e.g., the current flush threshold is 1M rows), this overhead is millions of Java metadata objects just for internal CSL use. In addition, an on-heap multi-dimensional key might include many small objects that increase the memory overhead even further, as opposed to OakIncrementalIndex that needs only one buffer object for many multi-dimensional keys.

Our experiments show that when using OnHeapIncrementalIndex and OffHeapIncrementalIndex, Java GC requires roughly 200% memory compared to the raw data size to achieve reasonable ingestion speed. Furthermore, this large number of objects also incur longer GC pauses (about 40% of the runtime in our experiments) as there are many long-living objects to traverse. OakIncrementalIndex has only 2% memory overhead and negligible GC runtime overhead.

We evaluated OakIncrementalIndex with comparison to OnHeapIncrementalIndex and OffHeapIncrementalIndex via system-level experiments and component-level benchmarks. The experimental setup and the results are depicted here.

The system-level experiments show improved ingestion memory and CPU efficiency. It uses 60% less memory and 50% less CPU-time to achieve the same performance. This translates to nearly double the system's ingestion-throughput with the same memory budget, and a 75% increase in throughput with the same CPU-time budget. The component-level benchmarks show almost 33% of the memory usage and 60% of the runtime (1.7x ingestion throughput) compared to the on-heap and off-heap implementations.

Test plan

We modified all the unit-test and benchmarks to test all the available incremental-index implementations (on-heap, off-heap, and Oak). All the unit tests passed successfully.

Operational impact

This change will not affect any existing clusters. It will work seamlessly and interchangeably with existing incremental index implementations. See our wiki’s usage section for more details.

[ebortnikov]

Hey Druid community,

Could we start discussing this issue please?

Thanks!

[liran-funaro]

We noticed a lot of Druid users run their workload on Amazon EC2. We want to point out that OakIncrementalIndex will not only improve performance but will also reduce operational costs by allowing the users to choose more affordable EC2 instances without sacrificing performance.

The figure below shows the operational cost of different required ingestion throughput on Amazon EC2.

[yuanlihan]

Hi @liran-funaro, thanks for your work on this!

I am so excited about this improvement that I tried to apply the patch and rebuilt modules(exclude the benchmark mudule) on on-prem cluster. And I got a OOM error related with direct memory allocation.

2020-06-17T14:51:12,290 ERROR [[index_kafka_ds_name_a4e6ce8abc9e093_ofeaphml]-threading-task-runner-executor-4] org.apache.d
ruid.indexing.overlord.ThreadingTaskRunner - Exception caught while running the task.
java.lang.OutOfMemoryError: Direct buffer memory
        at java.nio.Bits.reserveMemory(Bits.java:695) ~[?:1.8.0_181]
        at java.nio.DirectByteBuffer.<init>(DirectByteBuffer.java:123) ~[?:1.8.0_181]
        at java.nio.ByteBuffer.allocateDirect(ByteBuffer.java:311) ~[?:1.8.0_181]
        at com.yahoo.oak.Block.<init>(Block.java:37) ~[oak-0.2.1.jar:?]
        at com.yahoo.oak.BlocksPool.prealloc(BlocksPool.java:143) ~[oak-0.2.1.jar:?]
        at com.yahoo.oak.BlocksPool.getBlock(BlocksPool.java:100) ~[oak-0.2.1.jar:?]
        at com.yahoo.oak.NativeMemoryAllocator.allocateNewCurrentBlock(NativeMemoryAllocator.java:213) ~[oak-0.2.1.jar:?]
        at com.yahoo.oak.NativeMemoryAllocator.<init>(NativeMemoryAllocator.java:64) ~[oak-0.2.1.jar:?]
        at com.yahoo.oak.NativeMemoryAllocator.<init>(NativeMemoryAllocator.java:53) ~[oak-0.2.1.jar:?]
        at com.yahoo.oak.OakMapBuilder.build(OakMapBuilder.java:84) ~[oak-0.2.1.jar:?]
        at org.apache.druid.segment.incremental.OakIncrementalIndex$OakFactsHolder.<init>(OakIncrementalIndex.java:418) ~[druid-processing-0.18.1.jar:0
.18.1]
        at org.apache.druid.segment.incremental.OakIncrementalIndex.<init>(OakIncrementalIndex.java:89) ~[druid-processing-0.18.1.jar:0.18.1]
        at org.apache.druid.segment.incremental.IncrementalIndex$Builder.buildOak(IncrementalIndex.java:512) ~[druid-processing-0.18.1.jar:0.18.1]
        at org.apache.druid.segment.incremental.IncrementalIndex$Builder.build(IncrementalIndex.java:458) ~[druid-processing-0.18.1.jar:0.18.1]
        at org.apache.druid.segment.incremental.IncrementalIndex$Builder.build(IncrementalIndex.java:445) ~[druid-processing-0.18.1.jar:0.18.1]
        at org.apache.druid.segment.realtime.plumber.Sink.makeNewCurrIndex(Sink.java:375) ~[druid-server-0.18.1.jar:0.18.1]
        at org.apache.druid.segment.realtime.plumber.Sink.<init>(Sink.java:179) ~[druid-server-0.18.1.jar:0.18.1]
        at org.apache.druid.segment.realtime.plumber.Sink.<init>(Sink.java:122) ~[druid-server-0.18.1.jar:0.18.1]
        at org.apache.druid.segment.realtime.appenderator.AppenderatorImpl.getOrCreateSink(AppenderatorImpl.java:406) ~[druid-server-0.18.1.jar:0.18.1]
        at org.apache.druid.segment.realtime.appenderator.AppenderatorImpl.add(AppenderatorImpl.java:250) ~[druid-server-0.18.1.jar:0.18.1]
        at org.apache.druid.segment.realtime.appenderator.BaseAppenderatorDriver.append(BaseAppenderatorDriver.java:406) ~[druid-server-0.18.1.jar:0.18
.1]
        at org.apache.druid.segment.realtime.appenderator.StreamAppenderatorDriver.add(StreamAppenderatorDriver.java:187) ~[druid-server-0.18.1.jar:0.1
8.1]
        at org.apache.druid.indexing.seekablestream.SeekableStreamIndexTaskRunner.runInternal(SeekableStreamIndexTaskRunner.java:654) ~[druid-indexing-
service-0.18.1.jar:0.18.1]
        at org.apache.druid.indexing.seekablestream.SeekableStreamIndexTaskRunner.run(SeekableStreamIndexTaskRunner.java:277) ~[druid-indexing-service-
0.18.1.jar:0.18.1]
        at org.apache.druid.indexing.seekablestream.SeekableStreamIndexTask.run(SeekableStreamIndexTask.java:164) ~[druid-indexing-service-0.18.1.jar:0
.18.1]
        at org.apache.druid.indexing.overlord.ThreadingTaskRunner$1.call(ThreadingTaskRunner.java:209) [druid-indexing-service-0.18.1.jar:0.18.1]
        at org.apache.druid.indexing.overlord.ThreadingTaskRunner$1.call(ThreadingTaskRunner.java:149) [druid-indexing-service-0.18.1.jar:0.18.1]
        at java.util.concurrent.FutureTask.run(FutureTask.java:266) [?:1.8.0_181]
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) [?:1.8.0_181]
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) [?:1.8.0_181]
        at java.lang.Thread.run(Thread.java:748) [?:1.8.0_181]

It seems that the direct memory allocator of oak incremental index exceeds the MaxDirectMemorySize of the JVM process. Found that the maxBytesInMemory is calculated from heap memory by default while the oak incremental index will take it as available size of direct memory.

I solved this OOM issue by increasing MaxDirectMemorySize. In my opinion, it will be nice to have additional docs about direct memory calculation when users choose oak type incremental index. And the maxBytesInMemory seems a little ambiguous.

[liran-funaro]

Hi @yuanlihan. Thanks for trying our implementation.
We can't wait to hear about your experience with OakIncrementalIndex.
Did you witness ingestion speedup and/or resource utilization improvement? Please share your findings.

Regarding the issues you mentioned, thank you for pointing out the mismatch in the default configuration of maxBytesInMemory. We did not notice it was configured according to the on-heap memory.
We are working on a solution to this configuration problem and we will publish an update soon.
We will also address the OOM issue in this update.

[yuanlihan]

Hi @liran-funaro, I would like to post feedback about performance improvement here later.

But currently before performing further performance testing, I suffer from NPE errors for timeseries queries which scan memory data of latest 10 minutes. Note that I applied the patch to branch 0.18.1 and built successfully except the benchmark module.

"timeseries_ds_name_[2020-06-18T09:07:46.039Z/2020-06-18T09:17:46.039Z]@12302" daemon prio=5 tid=0x152 nid=NA runnable
  java.lang.Thread.State: RUNNABLE
      at com.yahoo.oak.Slice.getAddress(Slice.java:172)
      at com.yahoo.oak.KeyBuffer.getAddress(KeyBuffer.java:9)
      at com.yahoo.oak.UnscopedBuffer.getAddress(UnscopedBuffer.java:132)
      at org.apache.druid.segment.incremental.OakIncrementalIndexRow.<init>(OakIncrementalIndexRow.java:52)
      at org.apache.druid.segment.incremental.OakIncrementalIndex$OakFactsHolder.lambda$transformStreamIterator$1(OakIncrementalIndex.java:482)
      at org.apache.druid.segment.incremental.OakIncrementalIndex$OakFactsHolder$$Lambda$389.770801012.apply(Unknown Source:-1)
      at com.google.common.collect.Iterators$8.transform(Iterators.java:794)
      at com.google.common.collect.TransformedIterator.next(TransformedIterator.java:48)
      at org.apache.druid.segment.incremental.IncrementalIndexStorageAdapter$IncrementalIndexCursor.reset(IncrementalIndexStorageAdapter.java:360)
      at org.apache.druid.segment.incremental.IncrementalIndexStorageAdapter$IncrementalIndexCursor.<init>(IncrementalIndexStorageAdapter.java:264)
      at org.apache.druid.segment.incremental.IncrementalIndexStorageAdapter.lambda$makeCursors$0(IncrementalIndexStorageAdapter.java:213)
      at org.apache.druid.segment.incremental.IncrementalIndexStorageAdapter$$Lambda$381.1128483784.apply(Unknown Source:-1)
      at org.apache.druid.java.util.common.guava.MappingAccumulator.accumulate(MappingAccumulator.java:40)
      at org.apache.druid.java.util.common.guava.BaseSequence.accumulate(BaseSequence.java:44)
      at org.apache.druid.java.util.common.guava.MappedSequence.accumulate(MappedSequence.java:43)
      at org.apache.druid.java.util.common.guava.MappedSequence.accumulate(MappedSequence.java:43)
      at org.apache.druid.java.util.common.guava.FilteredSequence.accumulate(FilteredSequence.java:45)
      at org.apache.druid.java.util.common.guava.WrappingSequence$1.get(WrappingSequence.java:50)
      at org.apache.druid.java.util.common.guava.SequenceWrapper.wrap(SequenceWrapper.java:55)
      at org.apache.druid.java.util.common.guava.WrappingSequence.accumulate(WrappingSequence.java:45)
      at org.apache.druid.java.util.common.guava.Sequence.toList(Sequence.java:85)
      at org.apache.druid.query.ChainedExecutionQueryRunner$1$1.call(ChainedExecutionQueryRunner.java:124)
      at org.apache.druid.query.ChainedExecutionQueryRunner$1$1.call(ChainedExecutionQueryRunner.java:114)
      at java.util.concurrent.FutureTask.run(FutureTask.java:266)
      at org.apache.druid.java.util.common.concurrent.DirectExecutorService.execute(DirectExecutorService.java:81)
      at java.util.concurrent.AbstractExecutorService.submit(AbstractExecutorService.java:134)
      at com.google.common.util.concurrent.AbstractListeningExecutorService.submit(AbstractListeningExecutorService.java:58)
      at org.apache.druid.query.ChainedExecutionQueryRunner$1.lambda$make$0(ChainedExecutionQueryRunner.java:112)
      at org.apache.druid.query.ChainedExecutionQueryRunner$1$$Lambda$379.804059870.apply(Unknown Source:-1)
      at com.google.common.collect.Iterators$8.transform(Iterators.java:794)
      at com.google.common.collect.TransformedIterator.next(TransformedIterator.java:48)
      at com.google.common.collect.Iterators.addAll(Iterators.java:357)
      at com.google.common.collect.Lists.newArrayList(Lists.java:147)
      at com.google.common.collect.Lists.newArrayList(Lists.java:129)
      at org.apache.druid.query.ChainedExecutionQueryRunner$1.make(ChainedExecutionQueryRunner.java:104)
      at org.apache.druid.java.util.common.guava.BaseSequence.accumulate(BaseSequence.java:39)
      at org.apache.druid.java.util.common.guava.LazySequence.accumulate(LazySequence.java:40)
      at org.apache.druid.java.util.common.guava.WrappingSequence$1.get(WrappingSequence.java:50)
      at org.apache.druid.java.util.common.guava.SequenceWrapper.wrap(SequenceWrapper.java:55)
      at org.apache.druid.java.util.common.guava.WrappingSequence.accumulate(WrappingSequence.java:45)
      at org.apache.druid.java.util.common.guava.LazySequence.accumulate(LazySequence.java:40)
      at org.apache.druid.java.util.common.guava.WrappingSequence$1.get(WrappingSequence.java:50)
      at org.apache.druid.java.util.common.guava.SequenceWrapper.wrap(SequenceWrapper.java:55)
      at org.apache.druid.java.util.common.guava.WrappingSequence.accumulate(WrappingSequence.java:45)
      at org.apache.druid.java.util.common.guava.WrappingSequence$1.get(WrappingSequence.java:50)
      at org.apache.druid.query.CPUTimeMetricQueryRunner$1.wrap(CPUTimeMetricQueryRunner.java:74)
      at org.apache.druid.java.util.common.guava.WrappingSequence.accumulate(WrappingSequence.java:45)
      at org.apache.druid.query.spec.SpecificSegmentQueryRunner$1.accumulate(SpecificSegmentQueryRunner.java:87)
      at org.apache.druid.java.util.common.guava.WrappingSequence$1.get(WrappingSequence.java:50)
      at org.apache.druid.query.spec.SpecificSegmentQueryRunner.doNamed(SpecificSegmentQueryRunner.java:171)
      at org.apache.druid.query.spec.SpecificSegmentQueryRunner.access$100(SpecificSegmentQueryRunner.java:44)
      at org.apache.druid.query.spec.SpecificSegmentQueryRunner$2.wrap(SpecificSegmentQueryRunner.java:153)
      at org.apache.druid.java.util.common.guava.WrappingSequence.accumulate(WrappingSequence.java:45)
      at org.apache.druid.java.util.common.guava.Sequence.toList(Sequence.java:85)
      at org.apache.druid.query.ChainedExecutionQueryRunner$1$1.call(ChainedExecutionQueryRunner.java:124)
      at org.apache.druid.query.ChainedExecutionQueryRunner$1$1.call(ChainedExecutionQueryRunner.java:114)
      at java.util.concurrent.FutureTask.run(FutureTask.java:266)
      at org.apache.druid.query.PrioritizedListenableFutureTask.run(PrioritizedExecutorService.java:247)
      at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
      at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
      at java.lang.Thread.run(Thread.java:748)

and log

2020-06-18T17:32:49,884 ERROR [timeseries_shopee_sto_apm__tracking_http_live_[2020-06-18T09:23:46.039Z/2020-06-18T09:33:46.039Z]] org.apache.druid.quer
y.ChainedExecutionQueryRunner - Exception with one of the sequences! (java.lang.NullPointerException)

The NPE error occurs when the writeBuffer is null. And I would like to provide more info to address this issue.

[liran-funaro]

Thanks, @yuanlihan.
We are looking into this NPE bug. We'll update once we will understand more.

[liran-funaro]

@yuanlihan I found the reason for the NPE. It is due to a bug in the Oak library.
For now, until we update this PR, you can solve it by upgrading Oak version.
Simply modify Oak's version in druid/processing/pom.xml file as follows, then recompile using maven.
Note: If maven repository is not automatically updated for you, run the mvn command with -U flag.

diff --git a/processing/pom.xml b/processing/pom.xml
index 50ccc9eb6a..ebe850e480 100644
--- a/processing/pom.xml
+++ b/processing/pom.xml
@@ -35,7 +35,7 @@
         <dependency>
             <groupId>com.yahoo.oak</groupId>
             <artifactId>oak</artifactId>
-            <version>0.2.1</version>
+            <version>0.2.2</version>
         </dependency>
         <dependency>
             <groupId>org.apache.druid</groupId>

[yuanlihan]

Thanks @liran-funaro, I have solved the NPE error after upgrading Oak's jar.

[ebortnikov]

Hi Community,

Anybody up to reviewing this PR? Hopefully, the feature motivation is pretty clear.

Thanks!

[liran-funaro]

@yuanlihan Did you have time to continue evaluating our optimization.
Our latest updates of #10001 solved all know issues in addition to addressing your review notes.
We will also appreciate if you can check out our new design proposal (#10321) that will allow embedding this optimization as a Druid extension.