@Test public void testSearchWithFilter() 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()); TupleFilter filter = testHelper.buildEQFilter("SITE", "SITE0"); StreamingSearchContext searchRequest = new StreamingSearchContext(cubeDesc, dimensions, groups, metrics, filter, 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/10, returnRecordCnt); assertEquals(3, returnColNum); dimensions = testHelper.simulateDimensions("DAY_START", "SITE", "ITM"); filter = testHelper.buildLikeFilter("ITM", "ITM000001%"); searchRequest = new StreamingSearchContext(cubeDesc, dimensions, groups, metrics, filter, null); resultCollector = new SingleThreadResultCollector(); memoryStore.search(searchRequest, resultCollector); returnRecordCnt = 0; for (Record record : resultCollector) { returnRecordCnt++; } assertEquals(10000, returnRecordCnt); filter = testHelper.buildLowerFilter("ITM", FilterOperatorEnum.EQ, "itm0000010000"); searchRequest = new StreamingSearchContext(cubeDesc, dimensions, groups, metrics, filter, null); resultCollector = new SingleThreadResultCollector(); memoryStore.search(searchRequest, resultCollector); returnRecordCnt = 0; for (Record record : resultCollector) { returnRecordCnt++; } assertEquals(1, returnRecordCnt); }
@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);  }