Cube implements ICube, OTState<CubeDiff>, Initializable<Cube>, EventloopJmxBeanEx { public Promise<CubeDiff> consolidate(Function<Aggregation, Promise<AggregationDiff>> strategy) { logger.info("Launching consolidation"); Map<String, AggregationDiff> map = new HashMap<>(); List<AsyncSupplier<Void>> runnables = new ArrayList<>(); for (Entry<String, AggregationContainer> entry : aggregations.entrySet()) { String aggregationId = entry.getKey(); Aggregation aggregation = entry.getValue().aggregation; runnables.add(() -> strategy.apply(aggregation) .whenResult(aggregationDiff -> { if (!aggregationDiff.isEmpty()) { map.put(aggregationId, aggregationDiff); } }) .toVoid()); } return Promises.sequence(runnables).map($ -> CubeDiff.of(map)); } Cube(Eventloop eventloop, Executor executor, DefiningClassLoader classLoader,
			AggregationChunkStorage aggregationChunkStorage); static Cube create(@NotNull Eventloop eventloop, @NotNull Executor executor, @NotNull DefiningClassLoader classLoader,
			@NotNull AggregationChunkStorage aggregationChunkStorage); Cube withAttribute(String attribute, AttributeResolver resolver); Cube withClassLoaderCache(CubeClassLoaderCache classLoaderCache); Cube withDimension(String dimensionId, FieldType type); Cube withMeasure(String measureId, Measure measure); Cube withComputedMeasure(String measureId, ComputedMeasure computedMeasure); Cube withRelation(String child, String parent); Cube withTemporarySortDir(Path temporarySortDir); Cube withAggregation(AggregationConfig aggregationConfig); void addMeasure(String measureId, Measure measure); void addComputedMeasure(String measureId, ComputedMeasure computedMeasure); void addRelation(String child, String parent); void addDimension(String dimensionId, FieldType type); Cube addAggregation(AggregationConfig config); @NotNull Class<?> getAttributeInternalType(String attribute); @NotNull Class<?> getMeasureInternalType(String field); @NotNull Type getAttributeType(String attribute); @NotNull Type getMeasureType(String field); @Override Map<String, Type> getAttributeTypes(); @Override Map<String, Type> getMeasureTypes(); Aggregation getAggregation(String aggregationId); Set<String> getAggregationIds(); @Override void init(); @Override void apply(CubeDiff op); LogDataConsumer<T, CubeDiff> logStreamConsumer(Class<T> inputClass); LogDataConsumer<T, CubeDiff> logStreamConsumer(Class<T> inputClass,
			AggregationPredicate predicate); LogDataConsumer<T, CubeDiff> logStreamConsumer(Class<T> inputClass, Map<String, String> dimensionFields, Map<String, String> measureFields); LogDataConsumer<T, CubeDiff> logStreamConsumer(Class<T> inputClass, Map<String, String> dimensionFields, Map<String, String> measureFields,
			AggregationPredicate predicate); StreamConsumerWithResult<T, CubeDiff> consume(Class<T> inputClass); StreamConsumerWithResult<T, CubeDiff> consume(Class<T> inputClass, AggregationPredicate predicate); StreamConsumerWithResult<T, CubeDiff> consume(Class<T> inputClass, Map<String, String> dimensionFields, Map<String, String> measureFields,
			AggregationPredicate dataPredicate); StreamSupplier<T> queryRawStream(List<String> dimensions, List<String> storedMeasures, AggregationPredicate where,
			Class<T> resultClass); StreamSupplier<T> queryRawStream(List<String> dimensions, List<String> storedMeasures, AggregationPredicate where,
			Class<T> resultClass, DefiningClassLoader queryClassLoader); boolean containsExcessiveNumberOfOverlappingChunks(); Promise<CubeDiff> consolidate(Function<Aggregation, Promise<AggregationDiff>> strategy); Set<Object> getAllChunks(); Map<String, List<ConsolidationDebugInfo>> getConsolidationDebugInfo(); DefiningClassLoader getClassLoader(); @Override Promise<QueryResult> query(CubeQuery cubeQuery); @Override String toString(); @JmxAttribute int getAggregationsChunkSize(); @JmxAttribute void setAggregationsChunkSize(int aggregationsChunkSize); Cube withAggregationsChunkSize(int aggregationsChunkSize); Cube withAggregationsReducerBufferSize(int aggregationsReducerBufferSize); @JmxAttribute int getAggregationsSorterItemsInMemory(); @JmxAttribute void setAggregationsSorterItemsInMemory(int aggregationsSorterItemsInMemory); Cube withAggregationsSorterItemsInMemory(int aggregationsSorterItemsInMemory); @JmxAttribute int getAggregationsMaxChunksToConsolidate(); @JmxAttribute void setAggregationsMaxChunksToConsolidate(int aggregationsMaxChunksToConsolidate); Cube withAggregationsMaxChunksToConsolidate(int aggregationsMaxChunksToConsolidate); @JmxAttribute boolean getAggregationsIgnoreChunkReadingExceptions(); @JmxAttribute void setAggregationsIgnoreChunkReadingExceptions(boolean aggregationsIgnoreChunkReadingExceptions); Cube withAggregationsIgnoreChunkReadingExceptions(boolean aggregationsIgnoreChunkReadingExceptions); @JmxAttribute int getMaxOverlappingChunksToProcessLogs(); @JmxAttribute void setMaxOverlappingChunksToProcessLogs(int maxOverlappingChunksToProcessLogs); Cube withMaxOverlappingChunksToProcessLogs(int maxOverlappingChunksToProcessLogs); @JmxAttribute Duration getMaxIncrementalReloadPeriod(); @JmxAttribute void setMaxIncrementalReloadPeriod(Duration maxIncrementalReloadPeriod); Cube withMaxIncrementalReloadPeriod(Duration maxIncrementalReloadPeriod); @JmxAttribute ValueStats getQueryTimes(); @JmxAttribute long getQueryErrors(); @JmxAttribute Throwable getQueryLastError(); @JmxAttribute AggregationStats getAggregationStats(); @NotNull @Override Eventloop getEventloop(); static final int DEFAULT_OVERLAPPING_CHUNKS_THRESHOLD; }
@Test public void testConsolidate() throws QueryException { List<DataItemResult> expected = singletonList(new DataItemResult(1, 4, 0, 30, 60)); await( consume(cube, chunkStorage, new DataItem1(1, 2, 10, 20), new DataItem1(1, 3, 10, 20)), consume(cube, chunkStorage, new DataItem2(1, 3, 10, 20), new DataItem2(1, 4, 10, 20)), consume(cube, chunkStorage, new DataItem2(1, 2, 10, 20), new DataItem2(1, 4, 10, 20)), consume(cube, chunkStorage, new DataItem2(1, 4, 10, 20), new DataItem2(1, 5, 100, 200)) ); CubeDiff diff = await(cube.consolidate(Aggregation::consolidateHotSegment)); assertFalse(diff.isEmpty()); diff = await(cube.consolidate(Aggregation::consolidateHotSegment)); assertFalse(diff.isEmpty()); List<DataItemResult> list = await(cube.queryRawStream( asList("key1", "key2"), asList("metric1", "metric2", "metric3"), and(eq("key1", 1), eq("key2", 4)), DataItemResult.class, classLoader ).toList()); assertEquals(expected, list); }