@Test public void testAddEventAndScan() { genEvents(80000); StreamingCubeDataSearcher searcher = streamingSegmentManager.getSearcher(); Set<TblColRef> dimensions = testHelper.simulateDimensions("STREAMING_V2_TABLE.SITE"); Set<TblColRef> groups = testHelper.simulateDimensions("STREAMING_V2_TABLE.SITE"); Set<FunctionDesc> metrics = Sets.newHashSet(testHelper.simulateMetric("STREAMING_V2_TABLE.GMV", "SUM", "decimal(19,6)")); StreamingSearchContext searchRequest = new StreamingSearchContext(cubeDesc, dimensions, groups, metrics, null, null); IStreamingSearchResult segmentResults1 = searcher.doSearch(searchRequest, -1, true); int recordNum = 0; for (Record record : segmentResults1) { recordNum++; } assertEquals(10, recordNum); dimensions = testHelper.simulateDimensions("STREAMING_V2_TABLE.ITM"); groups = testHelper.simulateDimensions("STREAMING_V2_TABLE.ITM"); searchRequest = new StreamingSearchContext(cubeDesc, dimensions, groups, metrics, null, null); segmentResults1 = searcher.doSearch(searchRequest, -1, true); recordNum = 0; for (Record record : segmentResults1) { recordNum++; } assertEquals(80000, recordNum); dimensions = testHelper.simulateDimensions("STREAMING_V2_TABLE.MINUTE_START", "STREAMING_V2_TABLE.SITE"); groups = testHelper.simulateDimensions("STREAMING_V2_TABLE.SITE"); CompareTupleFilter filter1 = testHelper.buildCompareFilter("STREAMING_V2_TABLE.MINUTE_START", FilterOperatorEnum.GTE, "2018-07-30 20:00:00"); CompareTupleFilter filter2 = testHelper.buildCompareFilter("STREAMING_V2_TABLE.MINUTE_START", FilterOperatorEnum.LT, "2018-07-30 20:04:00"); TupleFilter filter = testHelper.buildAndFilter(filter1, filter2); metrics = Sets.newHashSet(testHelper.simulateCountMetric()); searchRequest = new StreamingSearchContext(cubeDesc, dimensions, groups, metrics, filter, null); segmentResults1 = searcher.doSearch(searchRequest, -1, true); recordNum = 0; for (Record record : segmentResults1) { recordNum++; long cnt = (Long)record.getMetrics()[0]; assertEquals(4000, cnt); } assertEquals(10, recordNum); CompareTupleFilter filter3 = testHelper.buildCompareFilter("STREAMING_V2_TABLE.MINUTE_START", FilterOperatorEnum.GTE, String.valueOf(DateFormat.stringToMillis("2018-07-30 20:04:00"))); CompareTupleFilter filter4 = testHelper.buildCompareFilter("STREAMING_V2_TABLE.MINUTE_START", FilterOperatorEnum.LTE, String.valueOf(DateFormat.stringToMillis("2018-07-30 20:06:00"))); filter = testHelper.buildAndFilter(filter3, filter4); groups = testHelper.simulateDimensions("STREAMING_V2_TABLE.MINUTE_START"); dimensions = groups; searchRequest = new StreamingSearchContext(cubeDesc, dimensions, groups, metrics, filter, null); segmentResults1 = searcher.doSearch(searchRequest, -1, true); recordNum = 0; for (Record record : segmentResults1) { recordNum++; System.out.println(record); } assertEquals(3, recordNum); }
public StreamingCubeDataSearcher getSearcher() { return new StreamingCubeDataSearcher(this); }
StreamingSegmentManager implements Closeable { public StreamingCubeDataSearcher getSearcher() { return new StreamingCubeDataSearcher(this); } }
StreamingSegmentManager implements Closeable { public StreamingCubeDataSearcher getSearcher() { return new StreamingCubeDataSearcher(this); } StreamingSegmentManager(String baseStorePath, CubeInstance cubeInstance, ISourcePositionHandler sourcePosHandler, IConsumerProvider consumerProvider); }
StreamingSegmentManager implements Closeable { public StreamingCubeDataSearcher getSearcher() { return new StreamingCubeDataSearcher(this); } StreamingSegmentManager(String baseStorePath, CubeInstance cubeInstance, ISourcePositionHandler sourcePosHandler, IConsumerProvider consumerProvider); void addEvent(StreamingMessage event); void restoreConsumerStates(ISourcePosition sourcePosition); void restoreSegmentsFromLocal(); String getCheckPointSourcePosition(); Collection<StreamingCubeSegment> getAllSegments(); StreamingCubeSegment getSegmentByName(String segmentName); Collection<StreamingCubeSegment> getActiveSegments(); Collection<StreamingCubeSegment> getRequireRemotePersistSegments(); ISourcePosition getSmallestSourcePosition(StreamingCubeSegment currSegment); void purgeSegment(String segmentName); void makeAllSegmentsImmutable(); void makeSegmentImmutable(String segmentName); List<String> remoteSegmentBuildComplete(String segmentName); void purgeAllSegments(); String getConsumePositionStr(); ISourcePosition getConsumePosition(); long getIngestCount(); long getLatestEventTime(); long getLatestEventIngestTime(); LongLatencyInfo getLongLatencyInfo(); CubeInstance getCubeInstance(); StreamingCubeDataSearcher getSearcher(); Map<String, SegmentStats> getSegmentStats(); @Override void close(); synchronized void checkpoint(); static long resetTimestampByTimeZone(long ts); }
StreamingSegmentManager implements Closeable { public StreamingCubeDataSearcher getSearcher() { return new StreamingCubeDataSearcher(this); } StreamingSegmentManager(String baseStorePath, CubeInstance cubeInstance, ISourcePositionHandler sourcePosHandler, IConsumerProvider consumerProvider); void addEvent(StreamingMessage event); void restoreConsumerStates(ISourcePosition sourcePosition); void restoreSegmentsFromLocal(); String getCheckPointSourcePosition(); Collection<StreamingCubeSegment> getAllSegments(); StreamingCubeSegment getSegmentByName(String segmentName); Collection<StreamingCubeSegment> getActiveSegments(); Collection<StreamingCubeSegment> getRequireRemotePersistSegments(); ISourcePosition getSmallestSourcePosition(StreamingCubeSegment currSegment); void purgeSegment(String segmentName); void makeAllSegmentsImmutable(); void makeSegmentImmutable(String segmentName); List<String> remoteSegmentBuildComplete(String segmentName); void purgeAllSegments(); String getConsumePositionStr(); ISourcePosition getConsumePosition(); long getIngestCount(); long getLatestEventTime(); long getLatestEventIngestTime(); LongLatencyInfo getLongLatencyInfo(); CubeInstance getCubeInstance(); StreamingCubeDataSearcher getSearcher(); Map<String, SegmentStats> getSegmentStats(); @Override void close(); synchronized void checkpoint(); static long resetTimestampByTimeZone(long ts); final long cubeDuration; }