DataUpdateManager { public void persistTree(String collection, String schedule, DateTime minTime, DateTime maxTime, final StarTree starTree) throws Exception { Lock lock = collectionLocks.get(collection); if (lock == null) { collectionLocks.putIfAbsent(collection, new ReentrantLock()); lock = collectionLocks.get(collection); } lock.lock(); LOGGER.info("Locked collection {} using lock {} for persist tree", collection, lock); try { File collectionDir = new File(rootDir, collection); if (!collectionDir.exists()) { FileUtils.forceMkdir(collectionDir); LOGGER.info("Created {}", collectionDir); } if (schedule.contains("_")) { throw new IOException("schedule cannot contain '_'"); } String persistId = "persist_" + UUID.randomUUID(); final File tmpDir = new File(new File(rootDir, collection), persistId); final File leafBufferDir = new File(tmpDir, "leafBuffers"); final File segmentBufferDir = new File(tmpDir, "segmentBuffers"); LOGGER.info("Beginning persist {}", persistId); LOGGER.info("Creating leaf buffer files in {}", leafBufferDir); final AtomicLong minDataTime = new AtomicLong(-1); final AtomicLong maxDataTime = new AtomicLong(-1); starTree.eachLeaf(new StarTreeCallback() { @Override public void call(StarTreeNode node) { try { Map<DimensionKey, MetricTimeSeries> records = new HashMap<>(); for (StarTreeRecord record : node.getRecordStore()) { MetricTimeSeries timeSeries = record.getMetricTimeSeries(); for (Long time : timeSeries.getTimeWindowSet()) { if (minDataTime.get() == -1 || minDataTime.get() > time) { minDataTime.set(time); } if (maxDataTime.get() == -1 || maxDataTime.get() < time) { maxDataTime.set(time); } } records.put(record.getDimensionKey(), timeSeries); } List<DimensionSpec> dimensions = starTree.getConfig().getDimensions(); String[] catchAll = new String[dimensions.size()]; Arrays.fill(catchAll, StarTreeConstants.OTHER); for (int i = 0; i < dimensions.size(); i++) { String name = dimensions.get(i).getName(); String value = node.getAncestorDimensionValues().get(name); if (value != null) { catchAll[i] = value; } } DimensionKey catchAllKey = new DimensionKey(catchAll); List<MetricSpec> metrics = starTree.getConfig().getMetrics(); MetricTimeSeries timeSeries = new MetricTimeSeries(MetricSchema.fromMetricSpecs(metrics)); if (!records.containsKey(catchAllKey)) { records.put(catchAllKey, timeSeries); } DimensionDictionary dictionary = new DimensionDictionary(node.getRecordStore().getForwardIndex()); VariableSizeBufferUtil.createLeafBufferFiles(leafBufferDir, node.getId().toString(), starTree.getConfig(), records, dictionary); } catch (Exception e) { LOGGER.error("Error creating leaf buffer files for {}", node.getId(), e); } } }); LOGGER.info("Serializing star tree"); ByteArrayOutputStream baos = new ByteArrayOutputStream(); ObjectOutputStream oos = new ObjectOutputStream(baos); oos.writeObject(starTree.getRoot()); oos.flush(); InputStream treeStream = new ByteArrayInputStream(baos.toByteArray()); LOGGER.info("Combining data files into {}", segmentBufferDir); VariableSizeBufferUtil.combineDataFiles(treeStream, leafBufferDir, segmentBufferDir); File metadataFile = new File(segmentBufferDir, StarTreeConstants.METADATA_FILE_NAME); LOGGER.info("Creating index metadata {}", metadataFile); TimeUnit aggregationGranularity = starTree.getConfig().getTime().getBucket().getUnit(); int bucketSize = starTree.getConfig().getTime().getBucket().getSize(); Long minDataTimeMillis = TimeUnit.MILLISECONDS.convert(minDataTime.get() * bucketSize, aggregationGranularity); Long maxDataTimeMillis = TimeUnit.MILLISECONDS.convert(maxDataTime.get() * bucketSize, aggregationGranularity); Long startTime = aggregationGranularity.convert(minTime.getMillis(), TimeUnit.MILLISECONDS) / bucketSize; Long endTime = aggregationGranularity.convert(maxTime.getMillis(), TimeUnit.MILLISECONDS) / bucketSize; IndexMetadata metadata = new IndexMetadata(minDataTime.get(), maxDataTime.get(), minDataTimeMillis, maxDataTimeMillis, startTime, endTime, minTime.getMillis(), maxTime.getMillis(), schedule, aggregationGranularity.toString(), bucketSize, IndexFormat.VARIABLE_SIZE); OutputStream metadataStream = new FileOutputStream(metadataFile); metadata.toProperties().store(metadataStream, "This segment was created via DataUpdateManager#persistTree"); metadataStream.close(); File configFile = new File(segmentBufferDir, StarTreeConstants.CONFIG_FILE_NAME); LOGGER.info("Creating config.yml {}", configFile); ObjectMapper objectMapper = new ObjectMapper(); objectMapper.writeValue(configFile, starTree.getConfig()); String treeId = starTree.getRoot().getId().toString(); File dataDir = new File(collectionDir, StorageUtils.getDataDirName(treeId, schedule, minTime, maxTime)); if (dataDir.exists()) { throw new Exception("Data is already persisted for timerange:" + minTime + " to " + maxTime); } LOGGER.info("Moving segments into {}", dataDir); FileUtils.forceMkdir(dataDir); StorageUtils.moveAllFiles(segmentBufferDir, dataDir); if (!dataDir.setLastModified(System.currentTimeMillis())) { LOGGER.warn("setLastModified on dataDir failed - watch service will not be triggered!"); } LOGGER.info("Removing tmp directory {}", tmpDir); FileUtils.forceDelete(tmpDir); } finally { lock.unlock(); LOGGER.info("Unlocked collection {} using lock {} for persist tree", collection, lock); } } DataUpdateManager(File rootDir, boolean autoExpire); void deleteCollection(String collection); void deleteData(String collection,
                         String schedule,
                         DateTime minTime,
                         DateTime maxTime); void deleteData(String collection, String schedule, DateTime maxTime); void updateData(String collection,
                         String schedule,
                         DateTime minTime,
                         DateTime maxTime,
                         byte[] data); void updateData(String collection,
                         String schedule,
                         DateTime minTime,
                         DateTime maxTime,
                         InputStream data); void expireSegments(File collectionDir, DateTime minTime, DateTime maxTime, String schedule); void persistTree(String collection,
                          String schedule,
                          DateTime minTime,
                          DateTime maxTime,
                          final StarTree starTree); StarTree mergeTrees(Collection<StarTree> starTrees);  }
@Test public void testPersistTree() throws Exception { File collectionDir = new File(rootDir, collection); File configFile = new File(collectionDir, StarTreeConstants.CONFIG_FILE_NAME); FileUtils.forceMkdir(collectionDir); ObjectMapper objectMapper = new ObjectMapper(); objectMapper.writeValue(configFile, baseConfig); StarTree starTree = new StarTreeImpl(config); starTree.open(); for (int i = 0; i < 10000; i++) { String[] combination = new String[] { "A" + (i % 2), "B" + (i % 4), "C" + (i % 8) }; MetricTimeSeries timeSeries = new MetricTimeSeries(metricSchema); for (int j = 0; j < 4; j++) { timeSeries.increment(j, "M", 1); } StarTreeRecord record = new StarTreeRecordImpl(config, new DimensionKey(combination), timeSeries); starTree.add(record); } StarTreeQuery query = new StarTreeQueryImpl(config, new DimensionKey(new String[] { "*", "*", "*" }), new TimeRange(0L, 0L)); MetricTimeSeries result = starTree.getTimeSeries(query); Assert.assertEquals(result.get(0, "M"), 10000); String collection = config.getCollection(); String schedule = "ONCE"; DateTime minTime = new DateTime(0); DateTime maxTime = new DateTime(4); String treeId = starTree.getRoot().getId().toString(); dataUpdateManager.persistTree(collection, schedule, minTime, maxTime, starTree); File dataDir = new File(collectionDir, StorageUtils.getDataDirName(treeId, schedule, minTime, maxTime)); Assert.assertTrue(dataDir.exists()); StarTreeManager manager = new StarTreeManagerImpl(); manager.restore(rootDir, collection); Map<File, StarTree> starTrees = manager.getStarTrees(collection); Assert.assertEquals(starTrees.size(), 1); StarTree restoredTree = starTrees.values().iterator().next(); result = restoredTree.getTimeSeries(query); Assert.assertEquals(result.get(0, "M"), 10000); }