StarTreeSegmentCreator implements SegmentCreator { @Override public void init(SegmentGeneratorConfig config, Map<String, ColumnIndexCreationInfo> columnInfo, Schema schema, int totalDocs, File outDir) throws Exception { this.config = config; this.columnInfo = columnInfo; this.schema = schema; this.outDir = outDir; this.starTreeDimensionDictionary = new HashMap<String, Integer>(); this.starTreeMetricDictionary = new HashMap<String, Integer>(); initializeAndBuildDictionaries(schema, columnInfo, outDir); for (int i = 0; i < schema.getDimensionNames().size(); i++) { starTreeDimensionDictionary.put(schema.getDimensionNames().get(i), i); } LOG.info("StarTree dimension dictionary: {}", starTreeDimensionDictionary); for (int i = 0; i < schema.getMetricNames().size(); i++) { starTreeMetricDictionary.put(schema.getMetricNames().get(i), i); } LOG.info("StarTree metric dictionary: {}", starTreeDimensionDictionary); splitOrder = computeSplitOrder(columnInfo); LOG.info("Computed split order {} (excluded: {})", splitOrder, starTreeIndexSpec.getSplitExcludes()); List<Integer> splitOrderIndexes = new ArrayList<Integer>(); for (String dimensionName : splitOrder) { Integer dimensionId = starTreeDimensionDictionary.get(dimensionName); splitOrderIndexes.add(dimensionId); } List<FieldSpec.DataType> dimensionTypes = new ArrayList<>(); for (DimensionFieldSpec spec : schema.getDimensionFieldSpecs()) { if (starTreeDimensionDictionary.containsKey(spec.getName())) { dimensionTypes.add(spec.getDataType()); } } List<FieldSpec.DataType> metricTypes = new ArrayList<>(); for (MetricFieldSpec spec : schema.getMetricFieldSpecs()) { if (starTreeMetricDictionary.containsKey(spec.getName())) { metricTypes.add(spec.getDataType()); } } starTreeTableFile = new File(System.getProperty("java.io.tmpdir"), STAR_TREE_TABLE_FILE_PREFIX + System.currentTimeMillis()); StarTreeTable table = new MmapLinkedListStarTreeTable( dimensionTypes, metricTypes, starTreeTableFile, DEFAULT_STAR_TREE_TABLE_INCREMENT); starTreeBuilder.init(splitOrderIndexes, starTreeIndexSpec.getMaxLeafRecords(), table); LOG.info("Building StarTree table..."); int count = 0; long startMillis = System.currentTimeMillis(); recordReader.rewind(); while (recordReader.hasNext()) { GenericRow row = recordReader.next(); StarTreeTableRow starTreeTableRow = extractValues(row); starTreeBuilder.append(starTreeTableRow); count++; } long endMillis = System.currentTimeMillis(); LOG.info("Finished building StarTree table ({} documents, took {} ms)", count, endMillis - startMillis); LOG.info("Building StarTree (computing aggregates)..."); startMillis = System.currentTimeMillis(); starTreeBuilder.build(); endMillis = System.currentTimeMillis(); LOG.info("Finished building StarTree, took {} ms", endMillis - startMillis); LOG.info("Re-computing unique metric values for dictionary encoding..."); startMillis = System.currentTimeMillis(); Map<String, Set<Object>> uniqueMetricValues = computeUniqueMetricValues(); resetMetricDictionaries(uniqueMetricValues); endMillis = System.currentTimeMillis(); LOG.info("Finished re-computing unique metric values (took {} ms)", endMillis - startMillis); int totalCombinedDocs = starTreeBuilder.getTotalRawDocumentCount() + starTreeBuilder.getTotalAggregateDocumentCount(); for (final String column : dictionaryCreatorMap.keySet()) { ColumnIndexCreationInfo indexCreationInfo = columnInfo.get(column); int uniqueValueCount = indexCreationInfo.getDistinctValueCount(); if (schema.getMetricNames().contains(column)) { uniqueValueCount = uniqueMetricValues.get(column).toArray().length; } if (schema.getFieldSpecFor(column).isSingleValueField()) { if (indexCreationInfo.isSorted()) { forwardIndexCreatorMap.put(column, new SingleValueSortedForwardIndexCreator(outDir, uniqueValueCount, schema.getFieldSpecFor(column))); } else { forwardIndexCreatorMap.put( column, new SingleValueUnsortedForwardIndexCreator(schema.getFieldSpecFor(column), outDir, uniqueValueCount, totalCombinedDocs, indexCreationInfo.getTotalNumberOfEntries(), indexCreationInfo.hasNulls())); } } else { forwardIndexCreatorMap.put( column, new MultiValueUnsortedForwardIndexCreator(schema.getFieldSpecFor(column), outDir, uniqueValueCount, totalCombinedDocs, indexCreationInfo.getTotalNumberOfEntries(), indexCreationInfo.hasNulls())); } if (config.isCreateInvertedIndexEnabled()) { invertedIndexCreatorMap.put( column, new BitmapInvertedIndexCreator(outDir, uniqueValueCount, schema.getFieldSpecFor(column))); } } } StarTreeSegmentCreator(StarTreeIndexSpec starTreeIndexSpec, RecordReader recordReader); @Override void init(SegmentGeneratorConfig config,
                   Map<String, ColumnIndexCreationInfo> columnInfo,
                   Schema schema,
                   int totalDocs,
                   File outDir); @Override void indexRow(GenericRow row); @Override void setSegmentName(String segmentName); @Override void seal();  }
@Test(enabled = true) public void testCreation() throws Exception { BasicConfigurator.configure(); final SegmentGeneratorConfig config = SegmentTestUtils.getSegmentGenSpecWithSchemAndProjectedColumns(avroFile, indexDir, "daysSinceEpoch", TimeUnit.DAYS, "testTable"); config.setSegmentNamePostfix("1"); config.setTimeColumnName("daysSinceEpoch"); StarTreeIndexSpec starTreeIndexSpec = new StarTreeIndexSpec(); starTreeIndexSpec.setSplitExcludes(Arrays.asList("daysSinceEpoch")); starTreeIndexSpec.setMaxLeafRecords(4); config.getSchema().setStarTreeIndexSpec(starTreeIndexSpec); final SegmentIndexCreationDriver driver = SegmentCreationDriverFactory.get(null); driver.init(config); driver.build(); }