Filtered aggregators should work at ingestion time

[dclim]

Right now including a filtered aggregator in the metricSpec throws an NPE but there are likely additional issues that need to be addressed to support this:

2015-12-07T23:00:54,620 ERROR [task-runner-0] io.druid.indexing.common.task.RealtimeIndexTask - Exception aborted realtime processing[twitter4]: {class=io.druid.indexing.common.task.RealtimeIndexTask, exceptionType=class java.lang.NullPointerException, exceptionMessage=null}
java.lang.NullPointerException
at io.druid.segment.incremental.IncrementalIndex$1$6.lookupId(IncrementalIndex.java:238) ~[classes/:?]
at io.druid.segment.filter.SelectorFilter.makeMatcher(SelectorFilter.java:67) ~[classes/:?]
at io.druid.query.aggregation.FilteredAggregatorFactory.factorize(FilteredAggregatorFactory.java:53) ~[classes/:?]
at io.druid.segment.incremental.OnheapIncrementalIndex.addToFacts(OnheapIncrementalIndex.java:144) ~[classes/:?]
at io.druid.segment.incremental.IncrementalIndex.add(IncrementalIndex.java:462) ~[classes/:?]
at io.druid.segment.realtime.plumber.Sink.add(Sink.java:125) ~[classes/:?]
at io.druid.segment.realtime.plumber.RealtimePlumber.add(RealtimePlumber.java:232) ~[classes/:?]
at io.druid.indexing.common.task.RealtimeIndexTask.run(RealtimeIndexTask.java:332) [classes/:?]
at io.druid.indexing.overlord.ThreadPoolTaskRunner$ThreadPoolTaskRunnerCallable.call(ThreadPoolTaskRunner.java:286) [classes/:?]
at io.druid.indexing.overlord.ThreadPoolTaskRunner$ThreadPoolTaskRunnerCallable.call(ThreadPoolTaskRunner.java:266) [classes/:?]
at java.util.concurrent.FutureTask.run(FutureTask.java:266) [?:1.8.0_60]
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) [?:1.8.0_60]
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) [?:1.8.0_60]
at java.lang.Thread.run(Thread.java:745) [?:1.8.0_60]
2015-12-07T23:00:54,624 ERROR [task-runner-0] io.druid.indexing.overlord.ThreadPoolTaskRunner - Exception while running task[RealtimeIndexTask{id=index_realtime_twitter4_2015-12-07T15:00:00.000-07:00_0_0, type=index_realtime, dataSource=twitter4}]
java.lang.NullPointerException
at io.druid.segment.incremental.IncrementalIndex$1$6.lookupId(IncrementalIndex.java:238) ~[classes/:?]
at io.druid.segment.filter.SelectorFilter.makeMatcher(SelectorFilter.java:67) ~[classes/:?]
at io.druid.query.aggregation.FilteredAggregatorFactory.factorize(FilteredAggregatorFactory.java:53) ~[classes/:?]
at io.druid.segment.incremental.OnheapIncrementalIndex.addToFacts(OnheapIncrementalIndex.java:144) ~[classes/:?]
at io.druid.segment.incremental.IncrementalIndex.add(IncrementalIndex.java:462) ~[classes/:?]
at io.druid.segment.realtime.plumber.Sink.add(Sink.java:125) ~[classes/:?]
at io.druid.segment.realtime.plumber.RealtimePlumber.add(RealtimePlumber.java:232) ~[classes/:?]
at io.druid.indexing.common.task.RealtimeIndexTask.run(RealtimeIndexTask.java:332) ~[classes/:?]
at io.druid.indexing.overlord.ThreadPoolTaskRunner$ThreadPoolTaskRunnerCallable.call(ThreadPoolTaskRunner.java:286) [classes/:?]
at io.druid.indexing.overlord.ThreadPoolTaskRunner$ThreadPoolTaskRunnerCallable.call(ThreadPoolTaskRunner.java:266) [classes/:?]
at java.util.concurrent.FutureTask.run(FutureTask.java:266) [?:1.8.0_60]
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) [?:1.8.0_60]
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) [?:1.8.0_60]
at java.lang.Thread.run(Thread.java:745) [?:1.8.0_60]
2015-12-07T23:00:54,628 INFO [task-runner-0] io.druid.indexing.worker.executor.ExecutorLifecycle - Task completed with status: {
"id" : "index_realtime_twitter4_2015-12-07T15:00:00.000-07:00_0_0",
"status" : "FAILED",
"duration" : 644
}

[se7entyse7en]

I was trying to add a filtered aggregator at ingestion time too and then I bumped into this issue. Are there any updates on this?

[se7entyse7en]

Here's also my situation when trying to use a filtered count aggregator: https://groups.google.com/forum/#!topic/druid-user/3-fdTDzZF3s.