@Test public void testOneValueAggregation() { genEvents(80000); StreamingCubeDataSearcher searcher = streamingSegmentManager.getSearcher(); String startTimeStr = "2018-07-30 20:00:00"; long startTime = DateFormat.stringToMillis(startTimeStr); String endTimeStr = "2018-07-30 20:04:00"; long endTime = DateFormat.stringToMillis(endTimeStr); CompareTupleFilter filter1 = testHelper.buildCompareFilter("STREAMING_V2_TABLE.MINUTE_START", FilterOperatorEnum.GTE, startTimeStr); CompareTupleFilter filter2 = testHelper.buildCompareFilter("STREAMING_V2_TABLE.MINUTE_START", FilterOperatorEnum.LT, endTimeStr); TupleFilter filter = testHelper.buildAndFilter(filter1, filter2); Set<FunctionDesc> metrics = Sets.newHashSet(testHelper.simulateCountMetric()); Set<TblColRef> dimensions = testHelper.simulateDimensions("STREAMING_V2_TABLE.MINUTE_START"); Set<TblColRef> groups = Sets.newHashSet(); StreamingSearchContext searchRequest = new StreamingSearchContext(cubeDesc, dimensions, groups, metrics, filter, null); IStreamingSearchResult segmentResults1 = searcher.doSearch(searchRequest, -1, true); for (Record record : segmentResults1) { long minStart = Long.valueOf(record.getDimensions()[0]); assertTrue(startTime <= minStart && minStart < endTime); System.out.println(record); } }
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; }