Added ParallelGroupedNumberPairCollectingProcessor. Renamed AveragePairWithStats to PairWithStats to express that this class is now not only used for aggregating an average but also to collect all individual pairs of a group when combined with ParallelGroupedNumberPairCollectingProcessor. Adjusted NumberPairResultsPresenter. Changed default series name to groupKey.

2026-11-07 19:42:17 +00:00 · 2018-04-26 16:39:20 +10:00
1 parent 4495e68367
commit edf922dfb6
@@ -1,59 +1,63 @@
 package com.google.gwt.user.client.rpc.core.com.sap.sse.datamining.shared.impl;

+import java.util.HashSet;
+
 import com.google.gwt.user.client.rpc.CustomFieldSerializer;
 import com.google.gwt.user.client.rpc.SerializationException;
 import com.google.gwt.user.client.rpc.SerializationStreamReader;
 import com.google.gwt.user.client.rpc.SerializationStreamWriter;
 import com.sap.sse.common.Util.Pair;
-import com.sap.sse.datamining.shared.impl.AveragePairWithStatsImpl;
+import com.sap.sse.datamining.shared.impl.PairWithStatsImpl;

@SuppressWarnings("rawtypes")
-public class AveragePairWithStatsImpl_CustomFieldSerializer extends CustomFieldSerializer<AveragePairWithStatsImpl> {
+public class PairWithStatsImpl_CustomFieldSerializer extends CustomFieldSerializer<PairWithStatsImpl> {
    @Override
    public boolean hasCustomInstantiateInstance() {
        return true;
    }

    @Override
-    public AveragePairWithStatsImpl instantiateInstance(SerializationStreamReader streamReader) throws SerializationException {
+    public PairWithStatsImpl instantiateInstance(SerializationStreamReader streamReader) throws SerializationException {
        return instantiate(streamReader);
    }

    @SuppressWarnings("unchecked")
-    public static AveragePairWithStatsImpl instantiate(SerializationStreamReader streamReader) throws SerializationException {
+    public static PairWithStatsImpl instantiate(SerializationStreamReader streamReader) throws SerializationException {
        final Pair average = (Pair) streamReader.readObject();
        final Pair min = (Pair) streamReader.readObject();
        final Pair max = (Pair) streamReader.readObject();
        final Pair median = (Pair) streamReader.readObject();
        final Pair standardDeviation = (Pair) streamReader.readObject();
+        final HashSet individualPairs = (HashSet) streamReader.readObject();
        final String resultType = streamReader.readString();
        final long count = streamReader.readLong();
-        return new AveragePairWithStatsImpl<Object>(average, min, max, median, standardDeviation, count, resultType);
+        return new PairWithStatsImpl<Object>(average, min, max, median, standardDeviation, individualPairs, count, resultType);
    }

    @Override
-    public void deserializeInstance(SerializationStreamReader streamReader, AveragePairWithStatsImpl instance)
+    public void deserializeInstance(SerializationStreamReader streamReader, PairWithStatsImpl instance)
            throws SerializationException {
        deserialize(streamReader, instance);
    }

-    public static void deserialize(SerializationStreamReader streamReader, AveragePairWithStatsImpl instance) {
+    public static void deserialize(SerializationStreamReader streamReader, PairWithStatsImpl instance) {
        // handled by instantiate
    }

    @Override
-    public void serializeInstance(SerializationStreamWriter streamWriter, AveragePairWithStatsImpl instance)
+    public void serializeInstance(SerializationStreamWriter streamWriter, PairWithStatsImpl instance)
            throws SerializationException {
        serialize(streamWriter, instance);
    }

-    public static void serialize(SerializationStreamWriter streamWriter, AveragePairWithStatsImpl instance)
+    public static void serialize(SerializationStreamWriter streamWriter, PairWithStatsImpl instance)
            throws SerializationException {
        streamWriter.writeObject(instance.getAverage());
        streamWriter.writeObject(instance.getMin());
        streamWriter.writeObject(instance.getMax());
        streamWriter.writeObject(instance.getMedian());
        streamWriter.writeObject(instance.getStandardDeviation());
+        streamWriter.writeObject(instance.getIndividualPairs());
        streamWriter.writeString(instance.getResultType());
        streamWriter.writeLong(instance.getCount());
    }
@@ -22,9 +22,10 @@ import org.moxieapps.gwt.highcharts.client.plotOptions.SeriesPlotOptions;
 import com.google.gwt.user.client.ui.SimpleLayoutPanel;
 import com.google.gwt.user.client.ui.Widget;
 import com.sap.sailing.gwt.ui.client.StringMessages;
+import com.sap.sse.common.Util.Pair;
 import com.sap.sse.common.settings.Settings;
 import com.sap.sse.datamining.shared.GroupKey;
-import com.sap.sse.datamining.shared.data.AveragePairWithStats;
+import com.sap.sse.datamining.shared.data.PairWithStats;
 import com.sap.sse.datamining.shared.impl.CompoundGroupKey;
 import com.sap.sse.datamining.shared.impl.GenericGroupKey;
 import com.sap.sse.datamining.shared.impl.dto.QueryResultDTO;
@@ -87,18 +88,30 @@ public class NumberPairResultsPresenter extends AbstractResultsPresenter<Setting
    protected void internalShowResults(QueryResultDTO<?> res) {
        result = res;
        createAndAddSeriesToChart();
-
+        
        for (Entry<GroupKey, ?> resultEntry : result.getResults().entrySet()) {
            @SuppressWarnings("unchecked")
-            AveragePairWithStats<Number> value = (AveragePairWithStats<Number>) resultEntry.getValue();
-            Point point = new Point(value.getAverage().getA(), value.getAverage().getB());
-            point.setName(resultEntry.getKey().asString());
-            seriesMappedByGroupKey.get(groupKeyToSeriesKey(resultEntry.getKey()))
-            .addPoint(point, false, false, false);   
+            PairWithStats<Number> value = (PairWithStats<Number>) resultEntry.getValue();
+            
+            if(value.getIndividualPairs() != null) {
+                for(Pair<Number, Number> pair : value.getIndividualPairs()) { 
+                    createAndAddPoint(resultEntry.getKey(), pair.getA(), pair.getB());
+                }
+            }
+            else {
+                createAndAddPoint(resultEntry.getKey(), value.getAverage().getA(), value.getAverage().getB());
+            }
+
        }
        chart.getXAxis().setAxisTitleText(result.getResultSignifier());
    }
    
+    private void createAndAddPoint(GroupKey key, Number x, Number y) {
+        Point point = new Point(x, y);
+        point.setName(key.asString());
+        seriesMappedByGroupKey.get(groupKeyToSeriesKey(key)).addPoint(point, false, false, false);     
+    }
+    
    private void createAndAddSeriesToChart() {
        for (GroupKey groupKey : result.getResults().keySet()) {
            GroupKey seriesKey = groupKeyToSeriesKey(groupKey);
@@ -115,7 +128,7 @@ public class NumberPairResultsPresenter extends AbstractResultsPresenter<Setting
            List<? extends GroupKey> subKeys = GroupKey.Util.getSubKeys(groupKey);
            return subKeys.size() == 1 ? subKeys.get(0) : new CompoundGroupKey(subKeys);
        } else {
-            return new GenericGroupKey<>(stringMessages.results());
+            return groupKey;
        }
    }

@@ -90,7 +90,7 @@ public class TabbedResultsPresenter extends AbstractComponent<Settings> implemen
                removeTab(oldHeader);
            } 
            else if (result.getResultType()
-                    .equals("com.sap.sse.datamining.shared.data.AveragePairWithStats")) {
+                    .equals("com.sap.sse.datamining.shared.data.PairWithStats")) {
                CloseableTabHeader oldHeader = getSelectedHeader();
                addTabAndFocus(new NumberPairResultsPresenter(TabbedResultsPresenter.this,
                        getComponentContext(), stringMessages));
@@ -1,15 +1,17 @@
 package com.sap.sse.datamining.shared.data;

 import java.io.Serializable;
+import java.util.HashSet;

 import com.sap.sse.common.Util.Pair;

-public interface AveragePairWithStats<T> extends Serializable {
+public interface PairWithStats<T> extends Serializable {
    Pair<T, T> getAverage();
    Pair<T, T> getMin();
    Pair<T, T> getMax();
    Pair<T, T> getMedian();
    Pair<T, T> getStandardDeviation();
+    HashSet<Pair<T,T>> getIndividualPairs();
    long getCount();
    
    /**
@@ -1,25 +1,29 @@
 package com.sap.sse.datamining.shared.impl;

-import com.sap.sse.common.Util.Pair;
-import com.sap.sse.datamining.shared.data.AveragePairWithStats;
+import java.util.HashSet;

-public class AveragePairWithStatsImpl<T> implements AveragePairWithStats<T> {
+import com.sap.sse.common.Util.Pair;
+import com.sap.sse.datamining.shared.data.PairWithStats;
+
+public class PairWithStatsImpl<T> implements PairWithStats<T> {
    private static final long serialVersionUID = -2643300186690471524L;
    private final Pair<T, T> average;
    private final Pair<T, T> min;
    private final Pair<T, T> max;
    private final Pair<T, T> median;
    private final Pair<T, T> standardDeviation;
+    private final HashSet<Pair<T,T>> individualPairs;
    private final long count;
    private final String resultType;

-    public AveragePairWithStatsImpl(Pair<T, T> average, Pair<T, T> min, Pair<T, T> max, Pair<T, T> median, Pair<T, T> standardDeviation, long count, String resultType) {
+    public PairWithStatsImpl(Pair<T, T> average, Pair<T, T> min, Pair<T, T> max, Pair<T, T> median, Pair<T, T> standardDeviation, HashSet<Pair<T,T>> individualPairs, long count, String resultType) {
        super();
        this.average = average;
        this.min = min;
        this.max = max;
        this.median = median;
        this.standardDeviation = standardDeviation;
+        this.individualPairs = individualPairs;
        this.count = count;
        this.resultType = resultType;
    }
@@ -58,11 +62,16 @@ public class AveragePairWithStatsImpl<T> implements AveragePairWithStats<T> {
    public String getResultType() {
        return resultType;
    }
+    
+    @Override
+    public HashSet<Pair<T, T>> getIndividualPairs() {
+        return individualPairs;
+    }

    @Override
    public String toString() {
-        return "AveragePairWithStatsImpl [average=" + average + ", min=" + min + ", max=" + max + ", median=" + median
-                + ", standardDeviation=" + standardDeviation + ", count=" + count + ", resultType=" + resultType + "]";
+        return "PairWithStatsImpl [average=" + average + ", min=" + min + ", max=" + max + ", median=" + median
+                + ", standardDeviation=" + standardDeviation + ", individualPairs=" + individualPairs + ", count=" + count + ", resultType=" + resultType + "]";
    }

    @Override
@@ -76,6 +85,7 @@ public class AveragePairWithStatsImpl<T> implements AveragePairWithStats<T> {
        result = prime * result + ((min == null) ? 0 : min.hashCode());
        result = prime * result + ((resultType == null) ? 0 : resultType.hashCode());
        result = prime * result + ((standardDeviation == null) ? 0 : standardDeviation.hashCode());
+        result = prime * result + ((individualPairs == null) ? 0 : individualPairs.hashCode());
        return result;
    }

@@ -87,7 +97,7 @@ public class AveragePairWithStatsImpl<T> implements AveragePairWithStats<T> {
            return false;
        if (getClass() != obj.getClass())
            return false;
-        AveragePairWithStatsImpl<?> other = (AveragePairWithStatsImpl<?>) obj;
+        PairWithStatsImpl<?> other = (PairWithStatsImpl<?>) obj;
        if (average == null) {
            if (other.average != null)
                return false;
@@ -120,6 +130,11 @@ public class AveragePairWithStatsImpl<T> implements AveragePairWithStats<T> {
                return false;
        } else if (!standardDeviation.equals(other.standardDeviation))
            return false;
+        if (individualPairs == null) {
+            if (other.individualPairs != null)
+                return false;
+        } else if (!individualPairs.equals(other.individualPairs))
+            return false;
        return true;
    }
 }
@@ -1,6 +1,7 @@
 Average=Average
 PairAverage=Average
 Median=Median
+PairCollecting=Identity
 Sum=Sum
 Maximum=Maximum
 Minimum=Minimum
@@ -1,5 +1,6 @@
 Average=Durchschnitt
 PairAverage=Durchschnitt
+PairCollecting=Identität
 Median=Median
 Sum=Summe
 Maximum=Maximum
@@ -28,12 +28,13 @@ import com.sap.sse.datamining.components.management.FunctionRegistry;
 import com.sap.sse.datamining.components.management.QueryDefinitionDTORegistry;
 import com.sap.sse.datamining.impl.components.aggregators.ParallelGroupedDataCountAggregationProcessor;
 import com.sap.sse.datamining.impl.components.aggregators.ParallelGroupedDataCountDistinctAggregationProcessor;
+import com.sap.sse.datamining.impl.components.aggregators.ParallelGroupedNumberPairCollectingProcessor;
 import com.sap.sse.datamining.impl.components.aggregators.ParallelGroupedNumberDataAverageAggregationProcessor;
 import com.sap.sse.datamining.impl.components.aggregators.ParallelGroupedNumberDataMaxAggregationProcessor;
 import com.sap.sse.datamining.impl.components.aggregators.ParallelGroupedNumberDataMedianAggregationProcessor;
 import com.sap.sse.datamining.impl.components.aggregators.ParallelGroupedNumberDataMinAggregationProcessor;
 import com.sap.sse.datamining.impl.components.aggregators.ParallelGroupedNumberDataSumAggregationProcessor;
-import com.sap.sse.datamining.impl.components.aggregators.ParallelGroupedNumberPairDataAverageAggregationProcessor;
+import com.sap.sse.datamining.impl.components.aggregators.ParallelGroupedNumberPairAverageAggregationProcessor;
 import com.sap.sse.datamining.impl.components.management.AggregationProcessorDefinitionManager;
 import com.sap.sse.datamining.impl.components.management.DataRetrieverChainDefinitionManager;
 import com.sap.sse.datamining.impl.components.management.DataSourceProviderManager;
@@ -87,7 +88,8 @@ public class DataMiningFrameworkActivator implements BundleActivator {
        defaultAggregationProcessors.add(ParallelGroupedDataCountDistinctAggregationProcessor.getDefinition());
        defaultAggregationProcessors.add(ParallelGroupedDataCountAggregationProcessor.getDefinition());
        defaultAggregationProcessors.add(ParallelGroupedNumberDataAverageAggregationProcessor.getDefinition());
-        defaultAggregationProcessors.add(ParallelGroupedNumberPairDataAverageAggregationProcessor.getDefinition());
+        defaultAggregationProcessors.add(ParallelGroupedNumberPairAverageAggregationProcessor.getDefinition());
+        defaultAggregationProcessors.add(ParallelGroupedNumberPairCollectingProcessor.getDefinition());
        defaultAggregationProcessors.add(ParallelGroupedNumberDataMaxAggregationProcessor.getDefinition());
        defaultAggregationProcessors.add(ParallelGroupedNumberDataMedianAggregationProcessor.getDefinition());
        defaultAggregationProcessors.add(ParallelGroupedNumberDataMinAggregationProcessor.getDefinition());
@@ -13,21 +13,21 @@ import com.sap.sse.datamining.components.Processor;
 import com.sap.sse.datamining.impl.components.GroupedDataEntry;
 import com.sap.sse.datamining.impl.components.SimpleAggregationProcessorDefinition;
 import com.sap.sse.datamining.shared.GroupKey;
-import com.sap.sse.datamining.shared.data.AveragePairWithStats;
-import com.sap.sse.datamining.shared.impl.AveragePairWithStatsImpl;
+import com.sap.sse.datamining.shared.data.PairWithStats;
+import com.sap.sse.datamining.shared.impl.PairWithStatsImpl;

@SuppressWarnings("rawtypes")
-public class ParallelGroupedNumberPairDataAverageAggregationProcessor
-            extends AbstractParallelGroupedDataStoringAggregationProcessor<Pair, AveragePairWithStats<Number>> {
-    private static final Class<?> _c = AveragePairWithStats.class;
+public class ParallelGroupedNumberPairAverageAggregationProcessor
+            extends AbstractParallelGroupedDataStoringAggregationProcessor<Pair, PairWithStats<Number>> {
+    private static final Class<?> _c = PairWithStats.class;
    @SuppressWarnings("unchecked")
-    private static final Class<AveragePairWithStats<Number>> _cc = (Class<AveragePairWithStats<Number>>) _c;
+    private static final Class<PairWithStats<Number>> _cc = (Class<PairWithStats<Number>>) _c;
    
-    private static final AggregationProcessorDefinition<Pair, AveragePairWithStats<Number>> DEFINITION =
+    private static final AggregationProcessorDefinition<Pair, PairWithStats<Number>> DEFINITION =
            new SimpleAggregationProcessorDefinition<>(Pair.class,
-                    _cc, "PairAverage", ParallelGroupedNumberPairDataAverageAggregationProcessor.class);
+                    _cc, "PairAverage", ParallelGroupedNumberPairAverageAggregationProcessor.class);
    
-    public static AggregationProcessorDefinition<Pair, AveragePairWithStats<Number>> getDefinition() {
+    public static AggregationProcessorDefinition<Pair, PairWithStats<Number>> getDefinition() {
        return DEFINITION;
    }

@@ -36,8 +36,8 @@ public class ParallelGroupedNumberPairDataAverageAggregationProcessor
    private final Map<GroupKey, Pair<Number, Number>> maxPerKey;
    private final Map<GroupKey, AtomicLong> elementAmountPerKey;

-    public ParallelGroupedNumberPairDataAverageAggregationProcessor(ExecutorService executor,
-            Collection<Processor<Map<GroupKey, AveragePairWithStats<Number>>, ?>> resultReceivers) {
+    public ParallelGroupedNumberPairAverageAggregationProcessor(ExecutorService executor,
+            Collection<Processor<Map<GroupKey, PairWithStats<Number>>, ?>> resultReceivers) {
        super(executor, resultReceivers, "PairAverage");
        elementAmountPerKey = new HashMap<>();
        sumPerKey = new HashMap<>();
@@ -87,14 +87,15 @@ public class ParallelGroupedNumberPairDataAverageAggregationProcessor
    }

    @Override
-    protected Map<GroupKey, AveragePairWithStats<Number>> aggregateResult() {
-        Map<GroupKey, AveragePairWithStats<Number>> result = new HashMap<>();
+    protected Map<GroupKey, PairWithStats<Number>> aggregateResult() {
+        Map<GroupKey, PairWithStats<Number>> result = new HashMap<>();
        for (Entry<GroupKey, Pair<Number, Number>> sumAggregationEntry : sumPerKey.entrySet()) {
            GroupKey key = sumAggregationEntry.getKey();
-            result.put(key, new AveragePairWithStatsImpl<Number>(new Pair<>(sumAggregationEntry.getValue().getA() != null ? sumAggregationEntry.getValue().getA().doubleValue() / elementAmountPerKey.get(key).get() : null, sumAggregationEntry.getValue().getB() != null ? sumAggregationEntry.getValue().getB().doubleValue() / elementAmountPerKey.get(key).get() : null) ,
+            result.put(key, new PairWithStatsImpl<Number>(new Pair<>(sumAggregationEntry.getValue().getA() != null ? sumAggregationEntry.getValue().getA().doubleValue() / elementAmountPerKey.get(key).get() : null, sumAggregationEntry.getValue().getB() != null ? sumAggregationEntry.getValue().getB().doubleValue() / elementAmountPerKey.get(key).get() : null) ,
                    minPerKey.get(key), maxPerKey.get(key),
                    /* median */ null,
                    /* standardDeviation */ null,
+                    /* individualPairs */ null,
                    /* count */ elementAmountPerKey.get(key).get(),
                    Pair.class.getName()));
        }
@@ -0,0 +1,99 @@
+package com.sap.sse.datamining.impl.components.aggregators;
+
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.atomic.AtomicLong;
+
+import com.sap.sse.common.Util.Pair;
+import com.sap.sse.datamining.components.AggregationProcessorDefinition;
+import com.sap.sse.datamining.components.Processor;
+import com.sap.sse.datamining.impl.components.GroupedDataEntry;
+import com.sap.sse.datamining.impl.components.SimpleAggregationProcessorDefinition;
+import com.sap.sse.datamining.shared.GroupKey;
+import com.sap.sse.datamining.shared.data.PairWithStats;
+import com.sap.sse.datamining.shared.impl.PairWithStatsImpl;
+
+
+@SuppressWarnings("rawtypes")
+public class ParallelGroupedNumberPairCollectingProcessor
+             extends AbstractParallelGroupedDataStoringAggregationProcessor<Pair, PairWithStats<Number>> {
+    
+    private static final Class<?> _c = PairWithStats.class;
+    @SuppressWarnings("unchecked")
+    private static final Class<PairWithStats<Number>> _cc = (Class<PairWithStats<Number>>) _c;
+    
+    private static final AggregationProcessorDefinition<Pair, PairWithStats<Number>> DEFINITION =
+            new SimpleAggregationProcessorDefinition<>(Pair.class,
+                    _cc, "PairCollecting", ParallelGroupedNumberPairCollectingProcessor.class);
+    
+    public static AggregationProcessorDefinition<Pair, PairWithStats<Number>> getDefinition() {
+        return DEFINITION;
+    }
+    
+    private final Map<GroupKey, HashSet<Pair<Number, Number>>> individualPairs;
+    private final Map<GroupKey, AtomicLong> elementAmountPerKey;
+
+    public ParallelGroupedNumberPairCollectingProcessor(ExecutorService executor,
+            Collection<Processor<Map<GroupKey, PairWithStats<Number>>, ?>> resultReceivers) {
+        super(executor, resultReceivers, "PairCollecting");
+        individualPairs = new HashMap<>();
+        elementAmountPerKey = new HashMap<>();
+    }
+
+    @Override
+    protected void storeElement(GroupedDataEntry<Pair> element) {
+        if (element.getDataEntry() != null) {
+            incrementElementAmount(element);
+            HashSet<Pair<Number, Number>> aggregate = individualPairs.get(element.getKey());
+            Number a = ((Number) element.getDataEntry().getA());
+            Number b = ((Number) element.getDataEntry().getB());
+
+            final Double doubleValueA = a == null ? null : a.doubleValue();
+            final Double doubleValueB = b == null ? null : b.doubleValue();
+            
+            if (doubleValueA != null && doubleValueB != null) {
+                if(aggregate == null) {
+                    aggregate = new HashSet<>();
+                    aggregate.add(new Pair<>(doubleValueA, doubleValueB));
+                    individualPairs.put(element.getKey(), aggregate);
+                }
+                else {
+                    individualPairs.get(element.getKey()).add(new Pair<>(doubleValueA, doubleValueB));
+                }
+            } 
+        }
+    }
+    
+    private void incrementElementAmount(GroupedDataEntry<Pair> element) {
+        GroupKey key = element.getKey();
+        // concurrency is not an issue here; needsSynchronization() returns true
+        AtomicLong currentAmount = elementAmountPerKey.get(key);
+        if (currentAmount == null) {
+            elementAmountPerKey.put(key, new AtomicLong(1));
+        } else {
+            currentAmount.incrementAndGet();
+        }
+    }
+
+    @Override
+    protected Map<GroupKey, PairWithStats<Number>> aggregateResult() {
+        Map<GroupKey, PairWithStats<Number>> result = new HashMap<>();
+        for (Entry<GroupKey, HashSet<Pair<Number, Number>>> sumAggregationEntry : individualPairs.entrySet()) {
+            GroupKey key = sumAggregationEntry.getKey();
+            result.put(key, new PairWithStatsImpl<Number>(null,
+                    /* min */ null, 
+                    /* max */ null,
+                    /* median */ null,
+                    /* standardDeviation */ null,
+                    /* individualPairs */ individualPairs.get(key),
+                    /* count */  elementAmountPerKey.get(key).get(),
+                    Pair.class.getName()));
+        }
+        return result;
+    }
+
+}