Implemented a workflow to enrich the result with additional data

This was problematic, because a processor can have multiple result
receivers and every result receiver could add additional data. So I
implemented an additional result data builder, that handles the conflicts,
when a member of the additional data is set multiple times.
There's a remaining problem with the StringMessagesForTests, because it
isn't as rich as the string messages used in the productive environment.
When this is resolved, the query tests should be successful.
This commit is contained in:
Lennart Hensler committed 2014-03-07 14:46:51 +01:00
1 parent 4c7a1706d6
commit 0350a54dec
34 files changed
+426 -81

No files matched your search

@@ -7,6 +7,7 @@ import java.util.concurrent.ExecutorService;
import com.sap.sailing.datamining.data.GPSFixWithContext;
import com.sap.sailing.domain.leaderboard.LeaderboardGroup;
import com.sap.sailing.server.RacingEventService;
import com.sap.sse.datamining.AdditionalResultDataBuilder;
import com.sap.sse.datamining.components.Processor;
import com.sap.sse.datamining.impl.components.AbstractPartitioningParallelProcessor;
@@ -19,13 +20,17 @@ public class GPSFixRetrievalProcessor extends
@Override
protected Callable<GPSFixWithContext> createInstruction(LeaderboardGroup partialElement) {
// TODO Auto-generated method stub
return null;
throw new UnsupportedOperationException();
}
@Override
protected Iterable<LeaderboardGroup> partitionElement(RacingEventService element) {
return null;
throw new UnsupportedOperationException();
}
@Override
protected void setAdditionalData(AdditionalResultDataBuilder additionalDataBuilder) {
throw new UnsupportedOperationException();
}
}
@@ -0,0 +1,17 @@
package com.sap.sse.datamining.shared;
public interface AdditionalResultData {
public int getRetrievedDataAmount();
public int getFilteredDataAmount();
public double getCalculationTimeInSeconds();
public String getResultSignifier();
public Unit getUnit();
public int getValueDecimals();
}
@@ -4,7 +4,7 @@ import java.io.Serializable;
import java.util.Map;
public interface QueryResult<AggregatedType> extends Serializable {
int getRetrievedDataAmount();
public int getFilteredDataAmount();
public double getCalculationTimeInSeconds();
@@ -0,0 +1,55 @@
package com.sap.sse.datamining.shared.impl;
import com.sap.sse.datamining.shared.AdditionalResultData;
import com.sap.sse.datamining.shared.Unit;
public class AdditionalResultDataImpl implements AdditionalResultData {
private final int retrievedDataAmount;
private int filteredDataAmount;
private String resultSignifier;
private Unit unit;
private int valueDecimals;
private long calculationTimeInNanos;
public AdditionalResultDataImpl(int retrievedDataAmount, int filteredDataAmount, String resultSignifier, Unit unit, int valueDecimals,
long calculationTimeInNanos) {
this.retrievedDataAmount = retrievedDataAmount;
this.filteredDataAmount = filteredDataAmount;
this.resultSignifier = resultSignifier;
this.unit = unit;
this.valueDecimals = valueDecimals;
this.calculationTimeInNanos = calculationTimeInNanos;
}
@Override
public int getRetrievedDataAmount() {
return retrievedDataAmount;
}
@Override
public int getFilteredDataAmount() {
return filteredDataAmount;
}
@Override
public double getCalculationTimeInSeconds() {
return calculationTimeInNanos;
}
@Override
public String getResultSignifier() {
return resultSignifier;
}
@Override
public Unit getUnit() {
return unit;
}
@Override
public int getValueDecimals() {
return valueDecimals;
}
}
@@ -3,6 +3,7 @@ package com.sap.sse.datamining.shared.impl;
import java.util.HashMap;
import java.util.Map;
import com.sap.sse.datamining.shared.AdditionalResultData;
import com.sap.sse.datamining.shared.GroupKey;
import com.sap.sse.datamining.shared.QueryResult;
import com.sap.sse.datamining.shared.Unit;
@@ -10,15 +11,8 @@ import com.sap.sse.datamining.shared.Unit;
public class QueryResultImpl<AggregatedType> implements QueryResult<AggregatedType> {
private static final long serialVersionUID = -8347731622150585715L;
private int retrievedDataAmount;
private int filteredDataAmount;
private long calculationTimeInNanos;
private String resultSignifier;
private Unit unit;
private int valueDecimals;
private Map<GroupKey, AggregatedType> results;
private AdditionalResultData additionalData;
/**
* Constructor for the GWT-Serialization. Don't use this!
@@ -26,32 +20,24 @@ public class QueryResultImpl<AggregatedType> implements QueryResult<AggregatedTy
@Deprecated
QueryResultImpl() { }
public QueryResultImpl(int retrievedDataAmount, int filteredDataAmount, String resultSiginifier, Unit unit, int valueDecimals) {
this.retrievedDataAmount = retrievedDataAmount;
this.filteredDataAmount = filteredDataAmount;
results = new HashMap<GroupKey, AggregatedType>();
this.resultSignifier = resultSiginifier;
this.unit = unit;
this.valueDecimals = valueDecimals;
public QueryResultImpl(Map<GroupKey, AggregatedType> results, AdditionalResultData additionalData) {
this.results = new HashMap<GroupKey, AggregatedType>(results);
this.additionalData = additionalData;
}
@Override
public int getRetrievedDataAmount() {
return retrievedDataAmount;
return additionalData.getRetrievedDataAmount();
}
@Override
public int getFilteredDataAmount() {
return filteredDataAmount;
return additionalData.getFilteredDataAmount();
}
@Override
public double getCalculationTimeInSeconds() {
return calculationTimeInNanos / 1000000000.0;
}
public void setCalculationTimeInNanos(long calculationTimeInNanos) {
this.calculationTimeInNanos = calculationTimeInNanos;
return additionalData.getCalculationTimeInSeconds();
}
@Override
@@ -61,17 +47,17 @@ public class QueryResultImpl<AggregatedType> implements QueryResult<AggregatedTy
@Override
public String getResultSignifier() {
return resultSignifier;
return additionalData.getResultSignifier();
}
@Override
public Unit getUnit() {
return unit;
return additionalData.getUnit();
}
@Override
public int getValueDecimals() {
return valueDecimals;
return additionalData.getValueDecimals();
}
@Override
@@ -11,6 +11,7 @@ import java.util.concurrent.Callable;
import org.junit.Before;
import org.junit.Test;
import com.sap.sse.datamining.AdditionalResultDataBuilder;
import com.sap.sse.datamining.components.Processor;
import com.sap.sse.datamining.test.util.ConcurrencyTestsUtil;
@@ -56,6 +57,10 @@ public class TestAbstractParallelProcessorFinishing {
}
@Override
public void abort() { }
@Override
public AdditionalResultDataBuilder getAdditionalResultData(AdditionalResultDataBuilder additionalDataBuilder) {
return additionalDataBuilder;
}
};
}
@@ -77,6 +82,9 @@ public class TestAbstractParallelProcessorFinishing {
}
};
}
@Override
protected void setAdditionalData(AdditionalResultDataBuilder additionalDataBuilder) {
}
};
}
@@ -10,6 +10,7 @@ import java.util.concurrent.Callable;
import org.junit.Before;
import org.junit.Test;
import com.sap.sse.datamining.AdditionalResultDataBuilder;
import com.sap.sse.datamining.components.Processor;
import com.sap.sse.datamining.test.util.ConcurrencyTestsUtil;
@@ -34,6 +35,10 @@ public class TestAbstractParallelProcessorWithManySimpleInstructions {
@Override
public void abort() {
}
@Override
public AdditionalResultDataBuilder getAdditionalResultData(AdditionalResultDataBuilder additionalDataBuilder) {
return additionalDataBuilder;
}
};
Collection<Processor<Integer>> receivers = new ArrayList<>();
@@ -48,6 +53,9 @@ public class TestAbstractParallelProcessorWithManySimpleInstructions {
}
};
}
@Override
protected void setAdditionalData(AdditionalResultDataBuilder additionalDataBuilder) {
}
};
}
@@ -9,6 +9,7 @@ import java.util.HashSet;
import org.junit.Before;
import org.junit.Test;
import com.sap.sse.datamining.AdditionalResultDataBuilder;
import com.sap.sse.datamining.components.Processor;
public class TestAbstractProcessor {
@@ -29,6 +30,10 @@ public class TestAbstractProcessor {
@Override
public void abort() {
}
@Override
public AdditionalResultDataBuilder getAdditionalResultData(AdditionalResultDataBuilder additionalDataBuilder) {
return additionalDataBuilder;
}
};
receivers = new HashSet<>();
@@ -45,6 +50,9 @@ public class TestAbstractProcessor {
@Override
public void abort() {
}
@Override
protected void setAdditionalData(AdditionalResultDataBuilder additionalDataBuilder) {
}
};
processor.finish();
assertThat(receiverWasToldToFinish, is(true));
@@ -12,6 +12,7 @@ import java.util.Map;
import org.junit.Before;
import org.junit.Test;
import com.sap.sse.datamining.AdditionalResultDataBuilder;
import com.sap.sse.datamining.components.FilterCriteria;
import com.sap.sse.datamining.components.NonFilteringProcessor;
import com.sap.sse.datamining.components.Processor;
@@ -101,6 +102,10 @@ public class TestFilteringProcessors {
@Override
public void abort() {
}
@Override
public AdditionalResultDataBuilder getAdditionalResultData(AdditionalResultDataBuilder additionalDataBuilder) {
return additionalDataBuilder;
}
};
receivers = new HashSet<>();
@@ -13,6 +13,7 @@ import java.util.Map;
import org.junit.Before;
import org.junit.Test;
import com.sap.sse.datamining.AdditionalResultDataBuilder;
import com.sap.sse.datamining.components.Processor;
import com.sap.sse.datamining.factories.FunctionFactory;
import com.sap.sse.datamining.functions.Function;
@@ -49,6 +50,10 @@ public class TestParallelExtractionProcessor {
@Override
public void abort() {
}
@Override
public AdditionalResultDataBuilder getAdditionalResultData(AdditionalResultDataBuilder additionalDataBuilder) {
return additionalDataBuilder;
}
};
receivers = new HashSet<>();
@@ -11,6 +11,7 @@ import java.util.Collection;
import org.junit.Before;
import org.junit.Test;
import com.sap.sse.datamining.AdditionalResultDataBuilder;
import com.sap.sse.datamining.components.Processor;
import com.sap.sse.datamining.factories.FunctionFactory;
import com.sap.sse.datamining.functions.Function;
@@ -40,6 +41,10 @@ public class TestParallelMultiDimensionalGroupingProcessor {
public void finish() throws InterruptedException { }
@Override
public void abort() { }
@Override
public AdditionalResultDataBuilder getAdditionalResultData(AdditionalResultDataBuilder additionalDataBuilder) {
return additionalDataBuilder;
}
};
receivers = new ArrayList<>();
@@ -9,6 +9,7 @@ import java.util.concurrent.Callable;
import org.junit.Test;
import com.sap.sse.datamining.AdditionalResultDataBuilder;
import com.sap.sse.datamining.components.Processor;
import com.sap.sse.datamining.test.util.ConcurrencyTestsUtil;
import com.sap.sse.datamining.test.util.FunctionTestsUtil;
@@ -35,6 +36,9 @@ public class TestPartitioningOfAbstractSimpleParallelProcessor {
}
};
}
@Override
protected void setAdditionalData(AdditionalResultDataBuilder additionalDataBuilder) {
}
};
processor.onElement(1);
@@ -9,6 +9,7 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.HashMap;
import java.util.Locale;
import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
@@ -18,22 +19,28 @@ import java.util.concurrent.TimeoutException;
import org.junit.Test;
import com.sap.sse.datamining.AdditionalResultDataBuilder;
import com.sap.sse.datamining.Query;
import com.sap.sse.datamining.components.Processor;
import com.sap.sse.datamining.factories.FunctionFactory;
import com.sap.sse.datamining.functions.Function;
import com.sap.sse.datamining.i18n.DataMiningStringMessages;
import com.sap.sse.datamining.impl.components.aggregators.ParallelGroupedDoubleDataSumAggregationProcessor;
import com.sap.sse.datamining.shared.GroupKey;
import com.sap.sse.datamining.shared.QueryResult;
import com.sap.sse.datamining.shared.Unit;
import com.sap.sse.datamining.shared.impl.AdditionalResultDataImpl;
import com.sap.sse.datamining.shared.impl.GenericGroupKey;
import com.sap.sse.datamining.shared.impl.QueryResultImpl;
import com.sap.sse.datamining.test.components.util.BlockingProcessor;
import com.sap.sse.datamining.test.components.util.Number;
import com.sap.sse.datamining.test.util.ConcurrencyTestsUtil;
import com.sap.sse.datamining.test.util.FunctionTestsUtil;
import com.sap.sse.datamining.test.util.StringMessagesForTests;
public class TestProcessorQuery {
private final static DataMiningStringMessages stringMessages = new StringMessagesForTests();
private boolean receivedElementOrFinished;
private boolean receivedAbort;
@@ -86,7 +93,8 @@ public class TestProcessorQuery {
*/
private Query<Double> createQueryWithStandardWorkflow(Collection<Number> dataSource) {
ThreadPoolExecutor executor = ConcurrencyTestsUtil.getExecutor();
ProcessorQuery<Double, Iterable<Number>> query = new ProcessorQuery<Double, Iterable<Number>>(executor, dataSource);
ProcessorQuery<Double, Iterable<Number>> query = new ProcessorQuery<Double, Iterable<Number>>(executor,
dataSource, stringMessages, Locale.ENGLISH);
Collection<Processor<Map<GroupKey, Double>>> aggregationResultReceivers = asCollection(query.getResultReceiver());
Processor<GroupedDataEntry<Double>> sumAggregator =
@@ -115,26 +123,29 @@ public class TestProcessorQuery {
}
private QueryResult<Double> buildExpectedResult(Collection<Number> dataSource) {
QueryResultImpl<Double> result = new QueryResultImpl<>(dataSource.size(), 2, "Cross sum (Sum)", Unit.None, 0);
result.addResult(new GenericGroupKey<Integer>(1), 8.0);
result.addResult(new GenericGroupKey<Integer>(2), 5.0);
result.addResult(new GenericGroupKey<Integer>(3), 3.0);
result.addResult(new GenericGroupKey<Integer>(4), 10.0);
Map<GroupKey, Double> results = new HashMap<>();
results.put(new GenericGroupKey<Integer>(1), 8.0);
results.put(new GenericGroupKey<Integer>(2), 5.0);
results.put(new GenericGroupKey<Integer>(3), 3.0);
results.put(new GenericGroupKey<Integer>(4), 10.0);
QueryResultImpl<Double> result = new QueryResultImpl<>(results, new AdditionalResultDataImpl(dataSource.size(), 2, "Cross sum (Sum)", Unit.None, 0, 0));
return result;
}
private void verifyResult(QueryResult<Double> result, QueryResult<Double> expectedResult) {
assertThat("Result values aren't correct.", result.getResults(), is(expectedResult.getResults()));
// assertThat("Retrieved data amount isn't correct.", result.getRetrievedDataAmount(), is(expectedResult.getRetrievedDataAmount()));
// assertThat("Filtered data amount isn't correct.", result.getFilteredDataAmount(), is(expectedResult.getFilteredDataAmount()));
// assertThat("Result signifier isn't correct.", result.getResultSignifier(), is(expectedResult.getResultSignifier()));
// assertThat("Unit isn't correct.", result.getUnit(), is(expectedResult.getUnit()));
// assertThat("Value decimals aren't correct.", result.getValueDecimals(), is(expectedResult.getValueDecimals()));
assertThat("Filtered data amount isn't correct.", result.getFilteredDataAmount(), is(expectedResult.getFilteredDataAmount()));
assertThat("Result signifier isn't correct.", result.getResultSignifier(), is(expectedResult.getResultSignifier()));
assertThat("Unit isn't correct.", result.getUnit(), is(expectedResult.getUnit()));
assertThat("Value decimals aren't correct.", result.getValueDecimals(), is(expectedResult.getValueDecimals()));
}
@Test(timeout=2000)
public void testQueryTimeouting() throws TimeoutException {
ProcessorQuery<Double, Iterable<Number>> query = new ProcessorQuery<Double, Iterable<Number>>(ConcurrencyTestsUtil.getExecutor(), createDataSource());
ProcessorQuery<Double, Iterable<Number>> query = new ProcessorQuery<Double, Iterable<Number>>(
ConcurrencyTestsUtil.getExecutor(), createDataSource(), stringMessages, Locale.ENGLISH);
Processor<Double> resultReceiver = new Processor<Double>() {
@Override
public void onElement(Double element) {
@@ -148,6 +159,10 @@ public class TestProcessorQuery {
public void abort() {
receivedAbort = true;
}
@Override
public AdditionalResultDataBuilder getAdditionalResultData(AdditionalResultDataBuilder additionalDataBuilder) {
return additionalDataBuilder;
}
};
query.setFirstProcessor(new BlockingProcessor<Iterable<Number>, Double>(ConcurrencyTestsUtil.getExecutor(), Arrays.asList(resultReceiver), (long) 1000));
@@ -165,7 +180,8 @@ public class TestProcessorQuery {
@Test
public void testQueryWithTimeoutAndNonBlockingProcess() throws TimeoutException {
ProcessorQuery<Double, Iterable<Number>> query = new ProcessorQuery<Double, Iterable<Number>>(ConcurrencyTestsUtil.getExecutor(), createDataSource());
ProcessorQuery<Double, Iterable<Number>> query = new ProcessorQuery<Double, Iterable<Number>>(
ConcurrencyTestsUtil.getExecutor(), createDataSource(), stringMessages, Locale.ENGLISH);
String keyValue = "Sum";
query.setFirstProcessor(createSumBuildingProcessor(query, keyValue));
@@ -193,6 +209,10 @@ public class TestProcessorQuery {
}
};
}
@Override
protected void setAdditionalData(AdditionalResultDataBuilder additionalDataBuilder) {
}
};
}
@@ -10,7 +10,9 @@ import java.util.HashSet;
import org.junit.Before;
import org.junit.Test;
import com.sap.sse.datamining.AdditionalResultDataBuilder;
import com.sap.sse.datamining.components.Processor;
import com.sap.sse.datamining.shared.Message;
import com.sap.sse.datamining.test.util.ConcurrencyTestsUtil;
public class TestAbstractStoringParallelAggregationProcessor {
@@ -35,6 +37,10 @@ public class TestAbstractStoringParallelAggregationProcessor {
@Override
public void abort() {
}
@Override
public AdditionalResultDataBuilder getAdditionalResultData(AdditionalResultDataBuilder additionalDataBuilder) {
return additionalDataBuilder;
}
};
receivers = new HashSet<>();
@@ -43,7 +49,7 @@ public class TestAbstractStoringParallelAggregationProcessor {
@Test
public void testAbstractAggregationHandling() throws InterruptedException {
Processor<Integer> processor = new AbstractParallelStoringAggregationProcessor<Integer, Integer>(ConcurrencyTestsUtil.getExecutor(), receivers) {
Processor<Integer> processor = new AbstractParallelStoringAggregationProcessor<Integer, Integer>(ConcurrencyTestsUtil.getExecutor(), receivers, Message.Sum.toString()) {
@Override
protected void storeElement(Integer element) {
elementStore.add(element);
@@ -75,7 +81,7 @@ public class TestAbstractStoringParallelAggregationProcessor {
@Test(timeout=5000)
public void testThatTheLockIsReleasedAfterStoringFailed() throws InterruptedException {
Processor<Integer> processor = new AbstractParallelStoringAggregationProcessor<Integer, Integer>(ConcurrencyTestsUtil.getExecutor(), receivers) {
Processor<Integer> processor = new AbstractParallelStoringAggregationProcessor<Integer, Integer>(ConcurrencyTestsUtil.getExecutor(), receivers, Message.Sum.toString()) {
@Override
protected void storeElement(Integer element) {
if (element < 0) {
@@ -15,6 +15,7 @@ import java.util.Map.Entry;
import org.junit.Before;
import org.junit.Test;
import com.sap.sse.datamining.AdditionalResultDataBuilder;
import com.sap.sse.datamining.components.Processor;
import com.sap.sse.datamining.impl.components.GroupedDataEntry;
import com.sap.sse.datamining.shared.GroupKey;
@@ -188,6 +189,10 @@ public class TestParallelDoubleAggregationProcessors {
@Override
public void abort() {
}
@Override
public AdditionalResultDataBuilder getAdditionalResultData(AdditionalResultDataBuilder additionalDataBuilder) {
return additionalDataBuilder;
}
};
receivers = new ArrayList<>();
@@ -4,6 +4,7 @@ import java.util.Collection;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import com.sap.sse.datamining.AdditionalResultDataBuilder;
import com.sap.sse.datamining.components.Processor;
import com.sap.sse.datamining.impl.components.AbstractSimpleParallelProcessor;
@@ -25,4 +26,9 @@ public class BlockingProcessor<InputType, ResultType> extends AbstractSimplePara
}
};
}
@Override
protected void setAdditionalData(AdditionalResultDataBuilder additionalDataBuilder) {
}
}
@@ -0,0 +1,30 @@
package com.sap.sse.datamining;
import java.util.Locale;
import com.sap.sse.datamining.i18n.DataMiningStringMessages;
import com.sap.sse.datamining.shared.AdditionalResultData;
import com.sap.sse.datamining.shared.Unit;
/**
* These builders are used to collect the additional result data from different sources (e.g. the Processors)
* and resolves conflicts, if a member of the additional data gets set multiple times.
*/
public interface AdditionalResultDataBuilder {
public AdditionalResultData build(long calculationTimeInNanos, DataMiningStringMessages stringMessages, Locale locale);
public void setRetrievedDataAmount(int retrievedDataAmount);
public void setFilteredDataAmount(int filteredDataAmount);
public void setExtractedStatisticNameMessageKey(String extractedStatisticNameMessageKey);
public void setAggregationNameMessageKey(String aggregationNameMessageKey);
public void setUnit(Unit unit);
public void setValueDecimals(int valueDecimals);
}
@@ -2,6 +2,7 @@ package com.sap.sse.datamining.components;
import java.util.Collection;
import com.sap.sse.datamining.AdditionalResultDataBuilder;
import com.sap.sse.datamining.impl.components.AbstractProcessor;
public class NonFilteringProcessor<InputType> extends AbstractProcessor<InputType, InputType> {
@@ -19,4 +20,9 @@ public class NonFilteringProcessor<InputType> extends AbstractProcessor<InputTyp
public void abort() {
}
@Override
protected void setAdditionalData(AdditionalResultDataBuilder additionalDataBuilder) {
additionalDataBuilder.setFilteredDataAmount(0);
}
}
@@ -1,5 +1,7 @@
package com.sap.sse.datamining.components;
import com.sap.sse.datamining.AdditionalResultDataBuilder;
public interface Processor<InputType> {
public void onElement(InputType element);
@@ -14,4 +16,11 @@ public interface Processor<InputType> {
public void abort();
/**
* Takes a result builder and fills it with its additional data and the data of its result receivers.
* @return The builder filled with the additional data of all processors in the chain. It can be used to
* construct the additional data of the executed processor chain.
*/
public AdditionalResultDataBuilder getAdditionalResultData(AdditionalResultDataBuilder additionalDataBuilder);
}
@@ -45,5 +45,7 @@ public interface Function<ReturnType> {
* If the function has no message key, the function name is used as display name.
*/
public FunctionDTO asDTO(Locale locale, DataMiningStringMessages stringMessages);
public String getMessageKey();
}
@@ -0,0 +1,80 @@
package com.sap.sse.datamining.impl;
import java.util.Locale;
import com.sap.sse.datamining.AdditionalResultDataBuilder;
import com.sap.sse.datamining.i18n.DataMiningStringMessages;
import com.sap.sse.datamining.shared.AdditionalResultData;
import com.sap.sse.datamining.shared.Message;
import com.sap.sse.datamining.shared.Unit;
import com.sap.sse.datamining.shared.impl.AdditionalResultDataImpl;
/**
* For general information see {@link AdditionalResultDataBuilder}.<br />
* Conflict resolving strategy:
* <ul>
* <li>Builds the sum of all given retrieved/filtered data amounts</li>
* <li>Other values are overwritten</li>
* </ul>
*/
public class SumBuildingAndOverwritingResultDataBuilder implements AdditionalResultDataBuilder {
private int retrievedDataAmount;
private int filteredDataAmount;
private String extractedStatisticNameMessageKey;
private String aggregationNameMessageKey;
private Unit unit;
private int valueDecimals;
/**
* Creates a new builder with standard values for the additional data.
*/
public SumBuildingAndOverwritingResultDataBuilder() {
retrievedDataAmount = 0;
filteredDataAmount = 0;
unit = Unit.None;
valueDecimals = 0;
}
@Override
public AdditionalResultData build(long calculationTimeInNanos, DataMiningStringMessages stringMessages, Locale locale) {
return new AdditionalResultDataImpl(retrievedDataAmount, filteredDataAmount, buildResultSignifier(stringMessages, locale), unit, valueDecimals, calculationTimeInNanos);
}
private String buildResultSignifier(DataMiningStringMessages stringMessages, Locale locale) {
String extractedStatisticName = stringMessages.get(locale, extractedStatisticNameMessageKey);
String aggregationName = stringMessages.get(locale, aggregationNameMessageKey);
return stringMessages.get(locale, Message.ResultSignifier, extractedStatisticName, aggregationName);
}
@Override
public void setRetrievedDataAmount(int retrievedDataAmount) {
this.retrievedDataAmount += retrievedDataAmount;
}
@Override
public void setFilteredDataAmount(int filteredDataAmount) {
this.filteredDataAmount += filteredDataAmount;
}
@Override
public void setExtractedStatisticNameMessageKey(String extractedStatisticNameMessageKey) {
this.extractedStatisticNameMessageKey = extractedStatisticNameMessageKey;
}
@Override
public void setAggregationNameMessageKey(String aggregationNameMessageKey) {
this.aggregationNameMessageKey = aggregationNameMessageKey;
}
@Override
public void setUnit(Unit unit) {
this.unit = unit;
}
@Override
public void setValueDecimals(int valueDecimals) {
this.valueDecimals = valueDecimals;
}
}
@@ -9,6 +9,7 @@ import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.logging.Level;
import java.util.logging.Logger;
import com.sap.sse.datamining.AdditionalResultDataBuilder;
import com.sap.sse.datamining.components.Processor;
public abstract class AbstractPartitioningParallelProcessor<InputType, WorkingType, ResultType>
@@ -126,6 +127,17 @@ public abstract class AbstractPartitioningParallelProcessor<InputType, WorkingTy
}
}
@Override
public AdditionalResultDataBuilder getAdditionalResultData(AdditionalResultDataBuilder additionalDataBuilder) {
setAdditionalData(additionalDataBuilder);
for (Processor<ResultType> resultReceiver : getResultReceivers()) {
additionalDataBuilder = resultReceiver.getAdditionalResultData(additionalDataBuilder);
}
return additionalDataBuilder;
}
protected abstract void setAdditionalData(AdditionalResultDataBuilder additionalDataBuilder);
/**
* Thread safe class to manage, if there are unfinished instructions.
*/
@@ -4,6 +4,7 @@ import java.util.Collection;
import java.util.HashSet;
import java.util.Set;
import com.sap.sse.datamining.AdditionalResultDataBuilder;
import com.sap.sse.datamining.components.Processor;
public abstract class AbstractProcessor<InputType, ResultType> implements Processor<InputType> {
@@ -34,5 +35,16 @@ public abstract class AbstractProcessor<InputType, ResultType> implements Proces
resultReceiver.finish();
}
}
@Override
public AdditionalResultDataBuilder getAdditionalResultData(AdditionalResultDataBuilder additionalDataBuilder) {
setAdditionalData(additionalDataBuilder);
for (Processor<ResultType> resultReceiver : resultReceivers) {
additionalDataBuilder = resultReceiver.getAdditionalResultData(additionalDataBuilder);
}
return additionalDataBuilder;
}
protected abstract void setAdditionalData(AdditionalResultDataBuilder additionalDataBuilder);
}
@@ -4,16 +4,20 @@ import java.util.Collection;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import com.sap.sse.datamining.AdditionalResultDataBuilder;
import com.sap.sse.datamining.components.FilterCriteria;
import com.sap.sse.datamining.components.Processor;
public class ParallelFilteringProcessor<InputType> extends AbstractSimpleParallelProcessor<InputType, InputType> {
private final FilterCriteria<InputType> filterCriteria;
private int filteredDataAmount;
public ParallelFilteringProcessor(ExecutorService executor, Collection<Processor<InputType>> resultReceivers, FilterCriteria<InputType> filterCriteria) {
super(executor, resultReceivers);
this.filterCriteria = filterCriteria;
filteredDataAmount = 0;
}
@Override
@@ -21,10 +25,23 @@ public class ParallelFilteringProcessor<InputType> extends AbstractSimpleParalle
return new Callable<InputType>() {
@Override
public InputType call() throws Exception {
InputType input = element;
return filterCriteria.matches(input) ? input : ParallelFilteringProcessor.super.createInvalidResult();
if (filterCriteria.matches(element)) {
return element;
} else {
increaseFilteredDataAmount();
return ParallelFilteringProcessor.super.createInvalidResult();
}
}
};
}
private synchronized void increaseFilteredDataAmount() {
filteredDataAmount++;
}
@Override
protected void setAdditionalData(AdditionalResultDataBuilder additionalDataBuilder) {
additionalDataBuilder.setFilteredDataAmount(filteredDataAmount);
}
}
@@ -4,6 +4,7 @@ import java.util.Collection;
import java.util.concurrent.Callable;
import java.util.concurrent.ThreadPoolExecutor;
import com.sap.sse.datamining.AdditionalResultDataBuilder;
import com.sap.sse.datamining.components.Processor;
import com.sap.sse.datamining.functions.Function;
@@ -31,4 +32,9 @@ public class ParallelGroupedElementsValueExtractionProcessor<ElementType, Functi
};
}
@Override
protected void setAdditionalData(AdditionalResultDataBuilder additionalDataBuilder) {
additionalDataBuilder.setExtractedStatisticNameMessageKey(extractionFunction.getMessageKey());
}
}
@@ -5,6 +5,7 @@ import java.util.Iterator;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import com.sap.sse.datamining.AdditionalResultDataBuilder;
import com.sap.sse.datamining.components.Processor;
import com.sap.sse.datamining.functions.Function;
import com.sap.sse.datamining.shared.GroupKey;
@@ -70,4 +71,8 @@ public class ParallelMultiDimensionalGroupingProcessor<DataType>
return element;
}
@Override
protected void setAdditionalData(AdditionalResultDataBuilder additionalDataBuilder) {
}
}
@@ -1,20 +1,22 @@
package com.sap.sse.datamining.impl.components;
import java.util.Locale;
import java.util.Map;
import java.util.Timer;
import java.util.TimerTask;
import java.util.Map.Entry;
import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.logging.Level;
import java.util.logging.Logger;
import com.sap.sse.datamining.AdditionalResultDataBuilder;
import com.sap.sse.datamining.Query;
import com.sap.sse.datamining.components.Processor;
import com.sap.sse.datamining.i18n.DataMiningStringMessages;
import com.sap.sse.datamining.impl.SumBuildingAndOverwritingResultDataBuilder;
import com.sap.sse.datamining.shared.GroupKey;
import com.sap.sse.datamining.shared.QueryResult;
import com.sap.sse.datamining.shared.Unit;
import com.sap.sse.datamining.shared.impl.QueryResultImpl;
public class ProcessorQuery<AggregatedType, DataSourceType> implements Query<AggregatedType> {
@@ -26,14 +28,19 @@ public class ProcessorQuery<AggregatedType, DataSourceType> implements Query<Agg
private Executor executor;
private final ProcessResultReceiver resultReceiver;
private final DataMiningStringMessages stringMessages;
private final Locale locale;
private final Object monitorObject = new Object();
private boolean workIsDone = false;
private boolean processorTimedOut = false;
public ProcessorQuery(Executor executor, DataSourceType dataSource) {
public ProcessorQuery(Executor executor, DataSourceType dataSource, DataMiningStringMessages stringMessages, Locale locale) {
this.executor = executor;
this.dataSource = dataSource;
this.stringMessages = stringMessages;
this.locale = locale;
resultReceiver = new ProcessResultReceiver();
}
@@ -65,6 +72,7 @@ public class ProcessorQuery<AggregatedType, DataSourceType> implements Query<Agg
}
private QueryResult<AggregatedType> processQuery(long timeoutInMillis) throws InterruptedException, TimeoutException {
final long startTime = System.nanoTime();
executor.execute(new Runnable() {
@Override
public void run() {
@@ -77,7 +85,13 @@ public class ProcessorQuery<AggregatedType, DataSourceType> implements Query<Agg
}
});
waitTillWorkIsDone(timeoutInMillis);
return resultReceiver.getResult();
final long endTime = System.nanoTime();
long calculationTimeInNanos = endTime - startTime;
AdditionalResultDataBuilder additionalDataBuilder = new SumBuildingAndOverwritingResultDataBuilder();
additionalDataBuilder = firstProcessor.getAdditionalResultData(additionalDataBuilder);
Map<GroupKey, AggregatedType> results = resultReceiver.getResult();
return new QueryResultImpl<>(results, additionalDataBuilder.build(calculationTimeInNanos, stringMessages, locale));
}
private void waitTillWorkIsDone(long timeoutInMillis) throws InterruptedException, TimeoutException {
@@ -116,19 +130,11 @@ public class ProcessorQuery<AggregatedType, DataSourceType> implements Query<Agg
private class ProcessResultReceiver implements Processor<Map<GroupKey, AggregatedType>> {
private QueryResult<AggregatedType> result;
private Map<GroupKey, AggregatedType> results;
@Override
public void onElement(Map<GroupKey, AggregatedType> groupedAggregations) {
result = constructResult(groupedAggregations);
}
private QueryResult<AggregatedType> constructResult(Map<GroupKey, AggregatedType> groupedAggregations) {
QueryResultImpl<AggregatedType> result = new QueryResultImpl<>(0, 0, "", Unit.None, 0);
for (Entry<GroupKey, AggregatedType> groupedAggregationsEntry : groupedAggregations.entrySet()) {
result.addResult(groupedAggregationsEntry.getKey(), groupedAggregationsEntry.getValue());
}
return result;
results = groupedAggregations;
}
@Override
@@ -141,11 +147,16 @@ public class ProcessorQuery<AggregatedType, DataSourceType> implements Query<Agg
@Override
public void abort() {
result = null;
results = null;
}
public QueryResult<AggregatedType> getResult() {
return result;
public Map<GroupKey, AggregatedType> getResult() {
return results;
}
@Override
public AdditionalResultDataBuilder getAdditionalResultData(AdditionalResultDataBuilder additionalDataBuilder) {
return additionalDataBuilder;
}
}
@@ -5,6 +5,7 @@ import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import com.sap.sse.datamining.AdditionalResultDataBuilder;
import com.sap.sse.datamining.components.Processor;
import com.sap.sse.datamining.impl.components.AbstractSimpleParallelProcessor;
@@ -12,10 +13,13 @@ public abstract class AbstractParallelStoringAggregationProcessor<InputType, Agg
extends AbstractSimpleParallelProcessor<InputType, AggregatedType> {
private final ReentrantReadWriteLock storeLock;
private final String aggregationNameMessageKey;
public AbstractParallelStoringAggregationProcessor(ExecutorService executor, Collection<Processor<AggregatedType>> resultReceivers) {
public AbstractParallelStoringAggregationProcessor(ExecutorService executor, Collection<Processor<AggregatedType>> resultReceivers,
String aggregationNameMessageKey) {
super(executor, resultReceivers);
storeLock = new ReentrantReadWriteLock();
this.aggregationNameMessageKey = aggregationNameMessageKey;
}
@Override
@@ -48,5 +52,10 @@ public abstract class AbstractParallelStoringAggregationProcessor<InputType, Agg
}
protected abstract AggregatedType aggregateResult();
@Override
protected void setAdditionalData(AdditionalResultDataBuilder additionalDataBuilder) {
additionalDataBuilder.setAggregationNameMessageKey(aggregationNameMessageKey);
}
}
@@ -9,6 +9,7 @@ import java.util.concurrent.ExecutorService;
import com.sap.sse.datamining.components.Processor;
import com.sap.sse.datamining.impl.components.GroupedDataEntry;
import com.sap.sse.datamining.shared.GroupKey;
import com.sap.sse.datamining.shared.Message;
public class ParallelGroupedDoubleDataAverageAggregationProcessor extends
AbstractParallelStoringAggregationProcessor<GroupedDataEntry<Double>, Map<GroupKey, Double>> {
@@ -18,7 +19,7 @@ public class ParallelGroupedDoubleDataAverageAggregationProcessor extends
public ParallelGroupedDoubleDataAverageAggregationProcessor(ExecutorService executor,
Collection<Processor<Map<GroupKey, Double>>> resultReceivers) {
super(executor, resultReceivers);
super(executor, resultReceivers, Message.Average.toString());
elementAmountPerKey = new HashMap<>();
sumAggregationProcessor = new ParallelGroupedDoubleDataSumAggregationProcessor(executor, resultReceivers);
}
@@ -12,6 +12,7 @@ import java.util.concurrent.ExecutorService;
import com.sap.sse.datamining.components.Processor;
import com.sap.sse.datamining.impl.components.GroupedDataEntry;
import com.sap.sse.datamining.shared.GroupKey;
import com.sap.sse.datamining.shared.Message;
public class ParallelGroupedDoubleDataMedianAggregationProcessor
extends AbstractParallelStoringAggregationProcessor<GroupedDataEntry<Double>, Map<GroupKey, Double>> {
@@ -20,7 +21,7 @@ public class ParallelGroupedDoubleDataMedianAggregationProcessor
public ParallelGroupedDoubleDataMedianAggregationProcessor(ExecutorService executor,
Collection<Processor<Map<GroupKey, Double>>> resultReceivers) {
super(executor, resultReceivers);
super(executor, resultReceivers, Message.Median.toString());
groupedValues = new HashMap<>();
}
@@ -9,6 +9,7 @@ import java.util.concurrent.ExecutorService;
import com.sap.sse.datamining.components.Processor;
import com.sap.sse.datamining.impl.components.GroupedDataEntry;
import com.sap.sse.datamining.shared.GroupKey;
import com.sap.sse.datamining.shared.Message;
public class ParallelGroupedDoubleDataSumAggregationProcessor
extends AbstractParallelStoringAggregationProcessor<GroupedDataEntry<Double>, Map<GroupKey, Double>> {
@@ -17,7 +18,7 @@ public class ParallelGroupedDoubleDataSumAggregationProcessor
public ParallelGroupedDoubleDataSumAggregationProcessor(ExecutorService executor,
Collection<Processor<Map<GroupKey, Double>>> resultReceivers) {
super(executor, resultReceivers);
super(executor, resultReceivers, Message.Sum.toString());
elementAmountMap = new HashMap<>();
}
@@ -3,7 +3,6 @@ package com.sap.sse.datamining.impl.deprecated;
import java.util.Collection;
import java.util.Locale;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
@@ -15,9 +14,11 @@ import com.sap.sse.datamining.components.ParallelExtractor;
import com.sap.sse.datamining.components.ParallelFilter;
import com.sap.sse.datamining.components.ParallelGrouper;
import com.sap.sse.datamining.i18n.DataMiningStringMessages;
import com.sap.sse.datamining.shared.AdditionalResultData;
import com.sap.sse.datamining.shared.GroupKey;
import com.sap.sse.datamining.shared.Message;
import com.sap.sse.datamining.shared.QueryResult;
import com.sap.sse.datamining.shared.impl.AdditionalResultDataImpl;
import com.sap.sse.datamining.shared.impl.QueryResultImpl;
public class QueryImpl<DataType, ExtractedType, AggregatedType> implements Query<AggregatedType> {
@@ -46,20 +47,18 @@ public class QueryImpl<DataType, ExtractedType, AggregatedType> implements Query
@Override
public QueryResult<AggregatedType> run() throws InterruptedException, ExecutionException {
final long startTime = System.nanoTime();
Collection<DataType> retrievedData = retriever.start(null).get();
Collection<DataType> filteredData = filter.start(retrievedData).get();
Map<GroupKey, Collection<DataType>> groupedData = grouper.start(filteredData).get();
Map<GroupKey, Collection<ExtractedType>> extractedData = extractor.start(groupedData).get();
Map<GroupKey, AggregatedType> aggregatedData = aggregator.start(extractedData).get();
QueryResultImpl<AggregatedType> result = new QueryResultImpl<AggregatedType>(retrievedData.size(), filteredData.size(), createResultSignifier(), extractor.getUnit(), extractor.getValueDecimals());
for (Entry<GroupKey, AggregatedType> resultEntry : aggregatedData.entrySet()) {
result.addResult(resultEntry.getKey(), resultEntry.getValue());
}
final long endTime = System.nanoTime();
result.setCalculationTimeInNanos(endTime - startTime);
long calculationTimeInNanos = endTime - startTime;
AdditionalResultData additionalData = new AdditionalResultDataImpl(retrievedData.size(), filteredData.size(),
createResultSignifier(), extractor.getUnit(), extractor.getValueDecimals(), calculationTimeInNanos);
QueryResult<AggregatedType> result = new QueryResultImpl<AggregatedType>(aggregatedData, additionalData);
return result;
}
@@ -94,7 +94,8 @@ public class MethodWrappingFunction<ReturnType> extends AbstractFunction<ReturnT
return method.getName();
}
private String getMessageKey() {
@Override
public String getMessageKey() {
return messageKey;
}
@@ -30,6 +30,11 @@ public class DataMiningResourceBundleManager implements DataMiningStringMessages
return get(locale, messageKey, new String[0]);
}
@Override
public String get(Locale locale, Message message, String... parameters) {
return get(locale, message.toString(), parameters);
}
@Override
public String get(Locale locale, String messageKey, String... parameters) {
String result = getResourceBundle(locale).getString(messageKey);
@@ -41,11 +46,6 @@ public class DataMiningResourceBundleManager implements DataMiningStringMessages
return result;
}
@Override
public String get(Locale locale, Message message, String... parameters) {
return get(locale, message.toString(), parameters);
}
@Override
public String get(Locale locale, Message message, Message... parameters) {
String[] parametersAsString = new String[parameters.length];