@Test public void testInvertIndexSearch() throws Exception { Set<TblColRef> dimensions = testHelper.simulateDimensions("STREAMING_V2_TABLE.SITE", "STREAMING_V2_TABLE.ITM", "STREAMING_V2_TABLE.MINUTE_START"); Set<TblColRef> groups = testHelper.simulateDimensions("STREAMING_V2_TABLE.SITE"); Set<FunctionDesc> metrics = testHelper.simulateMetrics(); String itmValue = "ITM0000009000"; CompareTupleFilter itemFilter = testHelper.buildEQFilter("STREAMING_V2_TABLE.ITM", itmValue); StreamingSearchContext searchRequest = new StreamingSearchContext(parsedStreamingCubeInfo.cubeDesc, dimensions, groups, metrics, itemFilter, null); ResponseResultSchema schema = searchRequest.getRespResultSchema(); int itmDimIdx = schema.getIndexOfDimension(itemFilter.getColumn()); ResultCollector resultCollector = new SingleThreadResultCollector(); fragmentFileSearcher.search(searchRequest, resultCollector); for (Record record : resultCollector) { assertEquals(itmValue, record.getDimensions()[itmDimIdx]); Object[] values = record.getMetrics(); PercentileCounter counter = (PercentileCounter) values[values.length - 1]; values[values.length - 1] = estPercentileValue(counter); System.out.println(Lists.newArrayList(values)); } String siteValue = "SITE0"; CompareTupleFilter siteFilter = testHelper.buildEQFilter("STREAMING_V2_TABLE.SITE", siteValue); int siteDimIdx = schema.getIndexOfDimension(siteFilter.getColumn()); TupleFilter andFilter = testHelper.buildAndFilter(itemFilter, siteFilter); searchRequest = new StreamingSearchContext(parsedStreamingCubeInfo.cubeDesc, dimensions, groups, metrics, andFilter, null); resultCollector = new SingleThreadResultCollector(); fragmentFileSearcher.search(searchRequest, resultCollector); for (Record record : resultCollector) { assertEquals(itmValue, record.getDimensions()[itmDimIdx]); assertEquals(siteValue, record.getDimensions()[siteDimIdx]); Object[] values = record.getMetrics(); PercentileCounter counter = (PercentileCounter) values[values.length - 1]; values[values.length - 1] = estPercentileValue(counter); System.out.println(Lists.newArrayList(values)); } TupleFilter likeFilter = testHelper.buildLikeFilter("STREAMING_V2_TABLE.ITM", "ITM000001%"); metrics = Sets.newHashSet(testHelper.simulateCountMetric()); searchRequest = new StreamingSearchContext(parsedStreamingCubeInfo.cubeDesc, dimensions, groups, metrics, likeFilter, null); resultCollector = new SingleThreadResultCollector(); fragmentFileSearcher.search(searchRequest, resultCollector); long count = 0; int rowsNum = 0; for (Record record : resultCollector) { count += (Long)record.getMetrics()[0]; rowsNum ++; } assertEquals(10000, count); assertEquals(10, rowsNum); }
@Override public void search(StreamingSearchContext searchContext, ResultCollector collector) throws IOException { String timezone = searchContext.getCubeDesc().getConfig().getStreamingDerivedTimeTimezone(); long timezoneOffset = 0; if (timezone != null && timezone.length() > 0) { timezoneOffset = TimeZone.getTimeZone(timezone).getRawOffset(); } FragmentMetaInfo fragmentMetaInfo = fragmentData.getFragmentMetaInfo(); CuboidMetaInfo cuboidMetaInfo; if (searchContext.hitBasicCuboid()) { cuboidMetaInfo = fragmentMetaInfo.getBasicCuboidMetaInfo(); } else { cuboidMetaInfo = fragmentMetaInfo.getCuboidMetaInfo(searchContext.getHitCuboid()); if (cuboidMetaInfo == null) { logger.warn("the cuboid:{} is not exist in the fragment:{}, use basic cuboid instead", searchContext.getHitCuboid(), fragment.getFragmentId()); cuboidMetaInfo = fragmentMetaInfo.getBasicCuboidMetaInfo(); } } ResponseResultSchema responseSchema = searchContext.getRespResultSchema(); TblColRef[] dimensions = responseSchema.getDimensions(); FunctionDesc[] metrics = responseSchema.getMetrics(); Map<TblColRef, Dictionary<String>> dictMap = fragmentData.getDimensionDictionaries(dimensions); CubeDesc cubeDesc = responseSchema.getCubeDesc(); List<MeasureDesc> allMeasures = cubeDesc.getMeasures(); Map<FunctionDesc, MeasureDesc> funcMeasureMap = Maps.newHashMap(); for (MeasureDesc measure : allMeasures) { funcMeasureMap.put(measure.getFunction(), measure); } MeasureDesc[] measures = new MeasureDesc[metrics.length]; for (int i = 0; i < measures.length; i++) { measures[i] = funcMeasureMap.get(metrics[i]); } DimensionEncoding[] dimensionEncodings = ParsedStreamingCubeInfo.getDimensionEncodings(cubeDesc, dimensions, dictMap); ColumnarMetricsEncoding[] metricsEncodings = ParsedStreamingCubeInfo.getMetricsEncodings(measures); ColumnarRecordCodec recordCodec = new ColumnarRecordCodec(dimensionEncodings, metricsEncodings); Set<TblColRef> unEvaluateDims = Sets.newHashSet(); TupleFilter fragmentFilter = null; if (searchContext.getFilter() != null) { fragmentFilter = convertFilter(fragmentMetaInfo, searchContext.getFilter(), recordCodec, dimensions, new CubeDimEncMap(cubeDesc, dictMap), unEvaluateDims, timezoneOffset); } if (ConstantTupleFilter.TRUE == fragmentFilter) { fragmentFilter = null; } else if (ConstantTupleFilter.FALSE == fragmentFilter) { collector.collectSearchResult(IStreamingSearchResult.EMPTY_RESULT); } Set<TblColRef> groups = searchContext.getGroups(); if (!unEvaluateDims.isEmpty()) { searchContext.addNewGroups(unEvaluateDims); groups = Sets.union(groups, unEvaluateDims); } collector.collectSearchResult(new FragmentSearchResult(fragment, fragmentData, cuboidMetaInfo, responseSchema, fragmentFilter, groups, searchContext.getHavingFilter(), recordCodec)); }
FragmentFileSearcher implements IStreamingGTSearcher { @Override public void search(StreamingSearchContext searchContext, ResultCollector collector) throws IOException { String timezone = searchContext.getCubeDesc().getConfig().getStreamingDerivedTimeTimezone(); long timezoneOffset = 0; if (timezone != null && timezone.length() > 0) { timezoneOffset = TimeZone.getTimeZone(timezone).getRawOffset(); } FragmentMetaInfo fragmentMetaInfo = fragmentData.getFragmentMetaInfo(); CuboidMetaInfo cuboidMetaInfo; if (searchContext.hitBasicCuboid()) { cuboidMetaInfo = fragmentMetaInfo.getBasicCuboidMetaInfo(); } else { cuboidMetaInfo = fragmentMetaInfo.getCuboidMetaInfo(searchContext.getHitCuboid()); if (cuboidMetaInfo == null) { logger.warn("the cuboid:{} is not exist in the fragment:{}, use basic cuboid instead", searchContext.getHitCuboid(), fragment.getFragmentId()); cuboidMetaInfo = fragmentMetaInfo.getBasicCuboidMetaInfo(); } } ResponseResultSchema responseSchema = searchContext.getRespResultSchema(); TblColRef[] dimensions = responseSchema.getDimensions(); FunctionDesc[] metrics = responseSchema.getMetrics(); Map<TblColRef, Dictionary<String>> dictMap = fragmentData.getDimensionDictionaries(dimensions); CubeDesc cubeDesc = responseSchema.getCubeDesc(); List<MeasureDesc> allMeasures = cubeDesc.getMeasures(); Map<FunctionDesc, MeasureDesc> funcMeasureMap = Maps.newHashMap(); for (MeasureDesc measure : allMeasures) { funcMeasureMap.put(measure.getFunction(), measure); } MeasureDesc[] measures = new MeasureDesc[metrics.length]; for (int i = 0; i < measures.length; i++) { measures[i] = funcMeasureMap.get(metrics[i]); } DimensionEncoding[] dimensionEncodings = ParsedStreamingCubeInfo.getDimensionEncodings(cubeDesc, dimensions, dictMap); ColumnarMetricsEncoding[] metricsEncodings = ParsedStreamingCubeInfo.getMetricsEncodings(measures); ColumnarRecordCodec recordCodec = new ColumnarRecordCodec(dimensionEncodings, metricsEncodings); Set<TblColRef> unEvaluateDims = Sets.newHashSet(); TupleFilter fragmentFilter = null; if (searchContext.getFilter() != null) { fragmentFilter = convertFilter(fragmentMetaInfo, searchContext.getFilter(), recordCodec, dimensions, new CubeDimEncMap(cubeDesc, dictMap), unEvaluateDims, timezoneOffset); } if (ConstantTupleFilter.TRUE == fragmentFilter) { fragmentFilter = null; } else if (ConstantTupleFilter.FALSE == fragmentFilter) { collector.collectSearchResult(IStreamingSearchResult.EMPTY_RESULT); } Set<TblColRef> groups = searchContext.getGroups(); if (!unEvaluateDims.isEmpty()) { searchContext.addNewGroups(unEvaluateDims); groups = Sets.union(groups, unEvaluateDims); } collector.collectSearchResult(new FragmentSearchResult(fragment, fragmentData, cuboidMetaInfo, responseSchema, fragmentFilter, groups, searchContext.getHavingFilter(), recordCodec)); } }
FragmentFileSearcher implements IStreamingGTSearcher { @Override public void search(StreamingSearchContext searchContext, ResultCollector collector) throws IOException { String timezone = searchContext.getCubeDesc().getConfig().getStreamingDerivedTimeTimezone(); long timezoneOffset = 0; if (timezone != null && timezone.length() > 0) { timezoneOffset = TimeZone.getTimeZone(timezone).getRawOffset(); } FragmentMetaInfo fragmentMetaInfo = fragmentData.getFragmentMetaInfo(); CuboidMetaInfo cuboidMetaInfo; if (searchContext.hitBasicCuboid()) { cuboidMetaInfo = fragmentMetaInfo.getBasicCuboidMetaInfo(); } else { cuboidMetaInfo = fragmentMetaInfo.getCuboidMetaInfo(searchContext.getHitCuboid()); if (cuboidMetaInfo == null) { logger.warn("the cuboid:{} is not exist in the fragment:{}, use basic cuboid instead", searchContext.getHitCuboid(), fragment.getFragmentId()); cuboidMetaInfo = fragmentMetaInfo.getBasicCuboidMetaInfo(); } } ResponseResultSchema responseSchema = searchContext.getRespResultSchema(); TblColRef[] dimensions = responseSchema.getDimensions(); FunctionDesc[] metrics = responseSchema.getMetrics(); Map<TblColRef, Dictionary<String>> dictMap = fragmentData.getDimensionDictionaries(dimensions); CubeDesc cubeDesc = responseSchema.getCubeDesc(); List<MeasureDesc> allMeasures = cubeDesc.getMeasures(); Map<FunctionDesc, MeasureDesc> funcMeasureMap = Maps.newHashMap(); for (MeasureDesc measure : allMeasures) { funcMeasureMap.put(measure.getFunction(), measure); } MeasureDesc[] measures = new MeasureDesc[metrics.length]; for (int i = 0; i < measures.length; i++) { measures[i] = funcMeasureMap.get(metrics[i]); } DimensionEncoding[] dimensionEncodings = ParsedStreamingCubeInfo.getDimensionEncodings(cubeDesc, dimensions, dictMap); ColumnarMetricsEncoding[] metricsEncodings = ParsedStreamingCubeInfo.getMetricsEncodings(measures); ColumnarRecordCodec recordCodec = new ColumnarRecordCodec(dimensionEncodings, metricsEncodings); Set<TblColRef> unEvaluateDims = Sets.newHashSet(); TupleFilter fragmentFilter = null; if (searchContext.getFilter() != null) { fragmentFilter = convertFilter(fragmentMetaInfo, searchContext.getFilter(), recordCodec, dimensions, new CubeDimEncMap(cubeDesc, dictMap), unEvaluateDims, timezoneOffset); } if (ConstantTupleFilter.TRUE == fragmentFilter) { fragmentFilter = null; } else if (ConstantTupleFilter.FALSE == fragmentFilter) { collector.collectSearchResult(IStreamingSearchResult.EMPTY_RESULT); } Set<TblColRef> groups = searchContext.getGroups(); if (!unEvaluateDims.isEmpty()) { searchContext.addNewGroups(unEvaluateDims); groups = Sets.union(groups, unEvaluateDims); } collector.collectSearchResult(new FragmentSearchResult(fragment, fragmentData, cuboidMetaInfo, responseSchema, fragmentFilter, groups, searchContext.getHavingFilter(), recordCodec)); } FragmentFileSearcher(DataSegmentFragment fragment, FragmentData fragmentData); }
FragmentFileSearcher implements IStreamingGTSearcher { @Override public void search(StreamingSearchContext searchContext, ResultCollector collector) throws IOException { String timezone = searchContext.getCubeDesc().getConfig().getStreamingDerivedTimeTimezone(); long timezoneOffset = 0; if (timezone != null && timezone.length() > 0) { timezoneOffset = TimeZone.getTimeZone(timezone).getRawOffset(); } FragmentMetaInfo fragmentMetaInfo = fragmentData.getFragmentMetaInfo(); CuboidMetaInfo cuboidMetaInfo; if (searchContext.hitBasicCuboid()) { cuboidMetaInfo = fragmentMetaInfo.getBasicCuboidMetaInfo(); } else { cuboidMetaInfo = fragmentMetaInfo.getCuboidMetaInfo(searchContext.getHitCuboid()); if (cuboidMetaInfo == null) { logger.warn("the cuboid:{} is not exist in the fragment:{}, use basic cuboid instead", searchContext.getHitCuboid(), fragment.getFragmentId()); cuboidMetaInfo = fragmentMetaInfo.getBasicCuboidMetaInfo(); } } ResponseResultSchema responseSchema = searchContext.getRespResultSchema(); TblColRef[] dimensions = responseSchema.getDimensions(); FunctionDesc[] metrics = responseSchema.getMetrics(); Map<TblColRef, Dictionary<String>> dictMap = fragmentData.getDimensionDictionaries(dimensions); CubeDesc cubeDesc = responseSchema.getCubeDesc(); List<MeasureDesc> allMeasures = cubeDesc.getMeasures(); Map<FunctionDesc, MeasureDesc> funcMeasureMap = Maps.newHashMap(); for (MeasureDesc measure : allMeasures) { funcMeasureMap.put(measure.getFunction(), measure); } MeasureDesc[] measures = new MeasureDesc[metrics.length]; for (int i = 0; i < measures.length; i++) { measures[i] = funcMeasureMap.get(metrics[i]); } DimensionEncoding[] dimensionEncodings = ParsedStreamingCubeInfo.getDimensionEncodings(cubeDesc, dimensions, dictMap); ColumnarMetricsEncoding[] metricsEncodings = ParsedStreamingCubeInfo.getMetricsEncodings(measures); ColumnarRecordCodec recordCodec = new ColumnarRecordCodec(dimensionEncodings, metricsEncodings); Set<TblColRef> unEvaluateDims = Sets.newHashSet(); TupleFilter fragmentFilter = null; if (searchContext.getFilter() != null) { fragmentFilter = convertFilter(fragmentMetaInfo, searchContext.getFilter(), recordCodec, dimensions, new CubeDimEncMap(cubeDesc, dictMap), unEvaluateDims, timezoneOffset); } if (ConstantTupleFilter.TRUE == fragmentFilter) { fragmentFilter = null; } else if (ConstantTupleFilter.FALSE == fragmentFilter) { collector.collectSearchResult(IStreamingSearchResult.EMPTY_RESULT); } Set<TblColRef> groups = searchContext.getGroups(); if (!unEvaluateDims.isEmpty()) { searchContext.addNewGroups(unEvaluateDims); groups = Sets.union(groups, unEvaluateDims); } collector.collectSearchResult(new FragmentSearchResult(fragment, fragmentData, cuboidMetaInfo, responseSchema, fragmentFilter, groups, searchContext.getHavingFilter(), recordCodec)); } FragmentFileSearcher(DataSegmentFragment fragment, FragmentData fragmentData); @Override void search(StreamingSearchContext searchContext, ResultCollector collector); }
FragmentFileSearcher implements IStreamingGTSearcher { @Override public void search(StreamingSearchContext searchContext, ResultCollector collector) throws IOException { String timezone = searchContext.getCubeDesc().getConfig().getStreamingDerivedTimeTimezone(); long timezoneOffset = 0; if (timezone != null && timezone.length() > 0) { timezoneOffset = TimeZone.getTimeZone(timezone).getRawOffset(); } FragmentMetaInfo fragmentMetaInfo = fragmentData.getFragmentMetaInfo(); CuboidMetaInfo cuboidMetaInfo; if (searchContext.hitBasicCuboid()) { cuboidMetaInfo = fragmentMetaInfo.getBasicCuboidMetaInfo(); } else { cuboidMetaInfo = fragmentMetaInfo.getCuboidMetaInfo(searchContext.getHitCuboid()); if (cuboidMetaInfo == null) { logger.warn("the cuboid:{} is not exist in the fragment:{}, use basic cuboid instead", searchContext.getHitCuboid(), fragment.getFragmentId()); cuboidMetaInfo = fragmentMetaInfo.getBasicCuboidMetaInfo(); } } ResponseResultSchema responseSchema = searchContext.getRespResultSchema(); TblColRef[] dimensions = responseSchema.getDimensions(); FunctionDesc[] metrics = responseSchema.getMetrics(); Map<TblColRef, Dictionary<String>> dictMap = fragmentData.getDimensionDictionaries(dimensions); CubeDesc cubeDesc = responseSchema.getCubeDesc(); List<MeasureDesc> allMeasures = cubeDesc.getMeasures(); Map<FunctionDesc, MeasureDesc> funcMeasureMap = Maps.newHashMap(); for (MeasureDesc measure : allMeasures) { funcMeasureMap.put(measure.getFunction(), measure); } MeasureDesc[] measures = new MeasureDesc[metrics.length]; for (int i = 0; i < measures.length; i++) { measures[i] = funcMeasureMap.get(metrics[i]); } DimensionEncoding[] dimensionEncodings = ParsedStreamingCubeInfo.getDimensionEncodings(cubeDesc, dimensions, dictMap); ColumnarMetricsEncoding[] metricsEncodings = ParsedStreamingCubeInfo.getMetricsEncodings(measures); ColumnarRecordCodec recordCodec = new ColumnarRecordCodec(dimensionEncodings, metricsEncodings); Set<TblColRef> unEvaluateDims = Sets.newHashSet(); TupleFilter fragmentFilter = null; if (searchContext.getFilter() != null) { fragmentFilter = convertFilter(fragmentMetaInfo, searchContext.getFilter(), recordCodec, dimensions, new CubeDimEncMap(cubeDesc, dictMap), unEvaluateDims, timezoneOffset); } if (ConstantTupleFilter.TRUE == fragmentFilter) { fragmentFilter = null; } else if (ConstantTupleFilter.FALSE == fragmentFilter) { collector.collectSearchResult(IStreamingSearchResult.EMPTY_RESULT); } Set<TblColRef> groups = searchContext.getGroups(); if (!unEvaluateDims.isEmpty()) { searchContext.addNewGroups(unEvaluateDims); groups = Sets.union(groups, unEvaluateDims); } collector.collectSearchResult(new FragmentSearchResult(fragment, fragmentData, cuboidMetaInfo, responseSchema, fragmentFilter, groups, searchContext.getHavingFilter(), recordCodec)); } FragmentFileSearcher(DataSegmentFragment fragment, FragmentData fragmentData); @Override void search(StreamingSearchContext searchContext, ResultCollector collector);  }