groupBy v2 failing intermittently with complex columns

[akashdw]

We are on druid 0.10 and seeing groupBy v2 failing intermittently with complex columns, we don't see those failures with groupByStrategy=v1.
Stacktrace :
com.yahoo.sketches.SketchesArgumentException: Possible Corruption: Illegal Family ID: 0
at com.yahoo.sketches.Family.idToFamily(Family.java:184) ~[sketches-core-0.8.4.jar:?]
at com.yahoo.sketches.theta.SetOperation.wrap(SetOperation.java:105) ~[sketches-core-0.8.4.jar:?]
at com.yahoo.sketches.theta.SetOperation.wrap(SetOperation.java:91) ~[sketches-core-0.8.4.jar:?]
at io.druid.query.aggregation.datasketches.theta.SketchBufferAggregator.getUnion(SketchBufferAggregator.java:92) ~[druid-datasketches-0.10.0-SNAPSHOT.jar:0.10.0-SNAPSHOT]
at io.druid.query.aggregation.datasketches.theta.SketchBufferAggregator.aggregate(SketchBufferAggregator.java:71) ~[druid-datasketches-0.10.0-SNAPSHOT.jar:0.10.0-SNAPSHOT]
at io.druid.query.groupby.epinephelinae.BufferGrouper.aggregate(BufferGrouper.java:203) ~[druid-processing-0.10.0-0a471fb.jar:0.10.0-SNAPSHOT]
at io.druid.query.groupby.epinephelinae.SpillingGrouper.aggregate(SpillingGrouper.java:111) ~[druid-processing-0.10.0-0a471fb.jar:0.10.0-SNAPSHOT]
at io.druid.query.groupby.epinephelinae.ConcurrentGrouper.aggregate(ConcurrentGrouper.java:163) ~[druid-processing-0.10.0-0a471fb.jar:0.10.0-SNAPSHOT]
at io.druid.query.groupby.epinephelinae.ConcurrentGrouper.aggregate(ConcurrentGrouper.java:184) ~[druid-processing-0.10.0-0a471fb.jar:0.10.0-SNAPSHOT]
at io.druid.query.groupby.epinephelinae.RowBasedGrouperHelper$1.accumulate(RowBasedGrouperHelper.java:173) ~[druid-processing-0.10.0-0a471fb.jar:0.10.0-SNAPSHOT]
at io.druid.query.groupby.epinephelinae.RowBasedGrouperHelper$1.accumulate(RowBasedGrouperHelper.java:148) ~[druid-processing-0.10.0-0a471fb.jar:0.10.0-SNAPSHOT]
at io.druid.java.util.common.guava.BaseSequence.accumulate(BaseSequence.java:46) ~[java-util-0.10.0-0a471fb.jar:0.10.0-SNAPSHOT]
at io.druid.java.util.common.guava.LazySequence.accumulate(LazySequence.java:40) ~[java-util-0.10.0-0a471fb.jar:0.10.0-SNAPSHOT]
at io.druid.java.util.common.guava.WrappingSequence$1.get(WrappingSequence.java:50) ~[java-util-0.10.0-0a471fb.jar:0.10.0-SNAPSHOT]
at io.druid.java.util.common.guava.SequenceWrapper.wrap(SequenceWrapper.java:55) ~[java-util-0.10.0-0a471fb.jar:0.10.0-SNAPSHOT]
at io.druid.java.util.common.guava.WrappingSequence.accumulate(WrappingSequence.java:45) ~[java-util-0.10.0-0a471fb.jar:0.10.0-SNAPSHOT]
at io.druid.java.util.common.guava.LazySequence.accumulate(LazySequence.java:40) ~[java-util-0.10.0-0a471fb.jar:0.10.0-SNAPSHOT]
at io.druid.java.util.common.guava.WrappingSequence$1.get(WrappingSequence.java:50) ~[java-util-0.10.0-0a471fb.jar:0.10.0-SNAPSHOT]
at io.druid.java.util.common.guava.SequenceWrapper.wrap(SequenceWrapper.java:55) ~[java-util-0.10.0-0a471fb.jar:0.10.0-SNAPSHOT]
at io.druid.java.util.common.guava.WrappingSequence.accumulate(WrappingSequence.java:45) ~[java-util-0.10.0-0a471fb.jar:0.10.0-SNAPSHOT]
at io.druid.java.util.common.guava.WrappingSequence$1.get(WrappingSequence.java:50) ~[java-util-0.10.0-0a471fb.jar:0.10.0-SNAPSHOT]
at io.druid.query.CPUTimeMetricQueryRunner$1.wrap(CPUTimeMetricQueryRunner.java:78) ~[druid-processing-0.10.0-0a471fb.jar:0.10.0-SNAPSHOT]
at io.druid.java.util.common.guava.WrappingSequence.accumulate(WrappingSequence.java:45) ~[java-util-0.10.0-0a471fb.jar:0.10.0-SNAPSHOT]
at io.druid.query.spec.SpecificSegmentQueryRunner$2.accumulate(SpecificSegmentQueryRunner.java:83) ~[druid-processing-0.10.0-0a471fb.jar:0.10.0-SNAPSHOT]
at io.druid.java.util.common.guava.WrappingSequence$1.get(WrappingSequence.java:50) ~[java-util-0.10.0-0a471fb.jar:0.10.0-SNAPSHOT]
at io.druid.query.spec.SpecificSegmentQueryRunner.doNamed(SpecificSegmentQueryRunner.java:169) ~[druid-processing-0.10.0-0a471fb.jar:0.10.0-SNAPSHOT]
at io.druid.query.spec.SpecificSegmentQueryRunner.access$200(SpecificSegmentQueryRunner.java:43) ~[druid-processing-0.10.0-0a471fb.jar:0.10.0-SNAPSHOT]
at io.druid.query.spec.SpecificSegmentQueryRunner$3.wrap(SpecificSegmentQueryRunner.java:149) ~[druid-processing-0.10.0-0a471fb.jar:0.10.0-SNAPSHOT]
at io.druid.java.util.common.guava.WrappingSequence.accumulate(WrappingSequence.java:45) ~[java-util-0.10.0-0a471fb.jar:0.10.0-SNAPSHOT]
at io.druid.query.groupby.epinephelinae.GroupByMergingQueryRunnerV2$1$1$1.call(GroupByMergingQueryRunnerV2.java:228) ~[druid-processing-0.10.0-0a471fb.jar:0.10.0-SNAPSHOT]
at io.druid.query.groupby.epinephelinae.GroupByMergingQueryRunnerV2$1$1$1.call(GroupByMergingQueryRunnerV2.java:219) ~[druid-processing-0.10.0-0a471fb.jar:0.10.0-SNAPSHOT]
at java.util.concurrent.FutureTask.run(FutureTask.java:266) ~[?:1.8.0_112]
at io.druid.query.PrioritizedListenableFutureTask.run(PrioritizedExecutorService.java:271) ~[druid-processing-0.10.0-0a471fb.jar:0.10.0-SNAPSHOT]
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) ~[?:1.8.0_112]
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) ~[?:1.8.0_112]
at java.lang.Thread.run(Thread.java:745) [?:1.8.0_112]

@himanshug @gianm

[gianm]

@akashdw Is this on a historical or on realtime or on a broker?

If it's on realtime, maybe it's due to non-thread-safe use of aggregators by incremental index (see #3956, #3578)?

[akashdw]

@gianm historical and realtime

[himanshug]

@gianm note that SketchBufferAggregator.aggregate() is not threadsafe while SketchAggregator.aggregator() is threadsafe to handle concurrency from IncrementalIndex.

with v2, is it possible that BufferAggregator.aggregate() would get called concurrently ?

[gianm]

@himanshug, groupBy v2 still won't call aggregator methods concurrently.

I think this is because this map in SketchBufferAggregator assumes that sketches will never be relocated:

private final Map<Integer, Union> unions = new HashMap<>(); //position in BB -> Union Object

But groupBy v2 does relocate aggregation buffers when it expands its hash table. Can you try working around this by adding "bufferGrouperInitialBuckets" : 2147483647 in your query context and confirm if that fixes the problem? That will start out the hash table at max size and so it won't expand. Note that this will slow down queries that only need a small hash table (it takes time to allocate a big one) so you don't want to do this blindly for every query.

If that's what's going on, then for a real fix, groupBy could close and re-open BufferAggregators when it resizes its hash table. We could do this for 0.10.0.

[himanshug]

@gianm yeah, in case of relocation that might not work.
they are cached for performance , creating Union objects from buffers is expensive.

@akashdw is gonna try max size and verify if this is indeed the issue.

[akashdw]

@gianm Yes, groupBy V2 query worked with "bufferGrouperInitialBuckets" : 2147483647

[himanshug]

close/re-open aggregators on hashtable growth sounds good to me.

[gianm]

great, let's do that patch then. @akashdw @himanshug will one of you work on that or shall I?

[akashdw]

@gianm Yes, Working on it.

[gianm]

thanks 🌟

[cheddar]

I disagree with that fix. The aggregator is breaking contract by caching the position... The aggregator should change to not cache objects anymore. We can try to get the union object to not be expensive to make on a buffer.

[cheddar]

also, open/close I don't believe will work because the current implementation assumes that hte object in its local cache is the "good" object. which is broken.

[gianm]

That sounds like a better fix if feasible.

@akashdw and @himanshug what do you think?

@cheddar by close/open I mean close, throw away, and re-create with factorizeBuffered. That should work around this, I think, although it'd be better for the aggregator not to store state inside itself.

[himanshug]

@cheddar @gianm @akashdw I'm good with removing caching of Unions and that would be ideal, however that would require some level of performance testing at least. also, we can check with @leerho if things have changed in Union creation since that code was written.