@Test public void testSearchBasicCuboid() throws Exception { StreamingQueryProfile profile = new StreamingQueryProfile("test-query-id", System.currentTimeMillis()); StreamingQueryProfile.set(profile); int eventCnt = 50000; prepareDataToMemoryStore(eventCnt); Set<TblColRef> dimensions = testHelper.simulateDimensions("DAY_START", "SITE"); Set<TblColRef> groups = testHelper.simulateDimensions("DAY_START", "SITE"); Set<FunctionDesc> metrics = Sets.newHashSet(testHelper.simulateCountMetric()); StreamingSearchContext searchRequest = new StreamingSearchContext(cubeDesc, dimensions, groups, metrics, null, null); ResultCollector resultCollector = new SingleThreadResultCollector(); memoryStore.search(searchRequest, resultCollector); int returnRecordCnt = 0; int returnColNum = 0; for (Record record : resultCollector) { returnRecordCnt++; returnColNum = record.getDimensions().length + record.getMetrics().length; } assertEquals(eventCnt, returnRecordCnt); assertEquals(3, returnColNum); }
@Override public void search(StreamingSearchContext searchContext, ResultCollector collector) throws IOException { ResponseResultSchema schema = searchContext.getRespResultSchema(); TblColRef[] selectedDimensions = schema.getDimensions(); FunctionDesc[] selectedMetrics = schema.getMetrics(); collector.collectSearchResult(new AggregationBufferSearchResult(searchContext, selectedDimensions, selectedMetrics)); }
SegmentMemoryStore implements IStreamingGTSearcher { @Override public void search(StreamingSearchContext searchContext, ResultCollector collector) throws IOException { ResponseResultSchema schema = searchContext.getRespResultSchema(); TblColRef[] selectedDimensions = schema.getDimensions(); FunctionDesc[] selectedMetrics = schema.getMetrics(); collector.collectSearchResult(new AggregationBufferSearchResult(searchContext, selectedDimensions, selectedMetrics)); } }
SegmentMemoryStore implements IStreamingGTSearcher { @Override public void search(StreamingSearchContext searchContext, ResultCollector collector) throws IOException { ResponseResultSchema schema = searchContext.getRespResultSchema(); TblColRef[] selectedDimensions = schema.getDimensions(); FunctionDesc[] selectedMetrics = schema.getMetrics(); collector.collectSearchResult(new AggregationBufferSearchResult(searchContext, selectedDimensions, selectedMetrics)); } SegmentMemoryStore(ParsedStreamingCubeInfo parsedStreamingCubeInfo, String segmentName); }
SegmentMemoryStore implements IStreamingGTSearcher { @Override public void search(StreamingSearchContext searchContext, ResultCollector collector) throws IOException { ResponseResultSchema schema = searchContext.getRespResultSchema(); TblColRef[] selectedDimensions = schema.getDimensions(); FunctionDesc[] selectedMetrics = schema.getMetrics(); collector.collectSearchResult(new AggregationBufferSearchResult(searchContext, selectedDimensions, selectedMetrics)); } SegmentMemoryStore(ParsedStreamingCubeInfo parsedStreamingCubeInfo, String segmentName); void setDictionaryMap(Map<TblColRef, Dictionary<String>> dictionaryMap); int index(StreamingMessage event); int getRowCount(); int getOriginRowCount(); ConcurrentSkipListMap<String[], MeasureAggregator[]> getBasicCuboidData(); Map<CuboidInfo, ConcurrentSkipListMap<String[], MeasureAggregator[]>> getAdditionalCuboidsData(); ConcurrentSkipListMap<String[], MeasureAggregator[]> getCuboidData(long cuboidID); long getMinEventTime(); long getMaxEventTime(); @Override void search(StreamingSearchContext searchContext, ResultCollector collector); }
SegmentMemoryStore implements IStreamingGTSearcher { @Override public void search(StreamingSearchContext searchContext, ResultCollector collector) throws IOException { ResponseResultSchema schema = searchContext.getRespResultSchema(); TblColRef[] selectedDimensions = schema.getDimensions(); FunctionDesc[] selectedMetrics = schema.getMetrics(); collector.collectSearchResult(new AggregationBufferSearchResult(searchContext, selectedDimensions, selectedMetrics)); } SegmentMemoryStore(ParsedStreamingCubeInfo parsedStreamingCubeInfo, String segmentName); void setDictionaryMap(Map<TblColRef, Dictionary<String>> dictionaryMap); int index(StreamingMessage event); int getRowCount(); int getOriginRowCount(); ConcurrentSkipListMap<String[], MeasureAggregator[]> getBasicCuboidData(); Map<CuboidInfo, ConcurrentSkipListMap<String[], MeasureAggregator[]>> getAdditionalCuboidsData(); ConcurrentSkipListMap<String[], MeasureAggregator[]> getCuboidData(long cuboidID); long getMinEventTime(); long getMaxEventTime(); @Override void search(StreamingSearchContext searchContext, ResultCollector collector);  }