adds float support by more buil-in aggregation functions

vmzakharov · Mar 26, 2024 · 3734ae4 · 3734ae4
1 parent 2584ce2
commit 3734ae4
Show file tree

Hide file tree

Showing 7 changed files with 178 additions and 90 deletions.
diff --git a/src/main/java/io/github/vmzakharov/ecdataframe/dataframe/AggregateFunction.java b/src/main/java/io/github/vmzakharov/ecdataframe/dataframe/AggregateFunction.java
@@ -231,7 +231,7 @@ protected double doubleAccumulator(double currentAggregate, double newValue)
         throw this.unsupportedAccumulatorException(ValueType.DOUBLE.toString());
     }
 
-    protected int floatAccumulator(float currentAggregate, float newValue)
+    protected float floatAccumulator(float currentAggregate, float newValue)
     {
         throw this.unsupportedAccumulatorException(ValueType.FLOAT.toString());
     }

diff --git a/src/main/java/io/github/vmzakharov/ecdataframe/dataframe/aggregation/Avg.java b/src/main/java/io/github/vmzakharov/ecdataframe/dataframe/aggregation/Avg.java
@@ -8,13 +8,15 @@
 import io.github.vmzakharov.ecdataframe.dataframe.DfDoubleColumn;
 import io.github.vmzakharov.ecdataframe.dataframe.DfDoubleColumnStored;
 import io.github.vmzakharov.ecdataframe.dataframe.DfFloatColumn;
+import io.github.vmzakharov.ecdataframe.dataframe.DfFloatColumnStored;
 import io.github.vmzakharov.ecdataframe.dataframe.DfIntColumn;
 import io.github.vmzakharov.ecdataframe.dataframe.DfIntColumnStored;
 import io.github.vmzakharov.ecdataframe.dataframe.DfLongColumn;
 import io.github.vmzakharov.ecdataframe.dataframe.DfLongColumnStored;
 import io.github.vmzakharov.ecdataframe.dataframe.DfObjectColumn;
 import io.github.vmzakharov.ecdataframe.dsl.value.ValueType;
 import io.github.vmzakharov.ecdataframe.util.ExceptionFactory;
+import org.eclipse.collections.api.block.procedure.primitive.IntProcedure;
 import org.eclipse.collections.api.list.ListIterable;
 import org.eclipse.collections.impl.factory.Lists;
 
@@ -24,7 +26,7 @@
 
 import static io.github.vmzakharov.ecdataframe.dsl.value.ValueType.*;
 
-// TODO - avoid overflows
+// TODO - change the agg avoid overflows
 public class Avg
 extends AggregateFunction
 {
@@ -105,6 +107,12 @@ protected double doubleAccumulator(double currentAggregate, double newValue)
         return currentAggregate + newValue;
     }
 
+    @Override
+    protected float floatAccumulator(float currentAggregate, float newValue)
+    {
+        return currentAggregate + newValue;
+    }
+
     @Override
     protected Object objectAccumulator(Object currentAggregate, Object newValue)
     {
@@ -129,6 +137,12 @@ public double doubleInitialValue()
         return 0;
     }
 
+    @Override
+    public float floatInitialValue()
+    {
+        return 0.0f;
+    }
+
     @Override
     public BigDecimal objectInitialValue()
     {
@@ -141,73 +155,80 @@ public void finishAggregating(DataFrame aggregatedDataFrame, int[] countsByRow)
         DfColumn aggregatedColumn = aggregatedDataFrame.getColumnNamed(this.getTargetColumnName());
         int columnSize = aggregatedColumn.getSize();
 
+        IntProcedure avgValueSetter = null;
+
         if (aggregatedColumn.getType().isLong())
         {
             DfLongColumnStored longColumn = (DfLongColumnStored) aggregatedColumn;
 
-            for (int rowIndex = 0; rowIndex < columnSize; rowIndex++)
-            {
-                if (!longColumn.isNull(rowIndex))
-                {
-                    long aggregateValue = longColumn.getLong(rowIndex);
+            avgValueSetter = (int rowIndex) -> {
+                long aggregateValue = longColumn.getLong(rowIndex);
 
-                    if (this.zeroContributorCheck(countsByRow[rowIndex], aggregateValue != 0, rowIndex))
-                    {
-                        longColumn.setLong(rowIndex, aggregateValue / countsByRow[rowIndex]);
-                    }
+                if (this.zeroContributorCheck(countsByRow[rowIndex], aggregateValue != 0, rowIndex))
+                {
+                    longColumn.setLong(rowIndex, aggregateValue / countsByRow[rowIndex]);
                 }
-            }
+            };
         }
         else if (aggregatedColumn.getType().isInt())
         {
-            DfIntColumnStored longColumn = (DfIntColumnStored) aggregatedColumn;
+            DfIntColumnStored intColumn = (DfIntColumnStored) aggregatedColumn;
 
-            for (int rowIndex = 0; rowIndex < columnSize; rowIndex++)
-            {
-                if (!longColumn.isNull(rowIndex))
-                {
-                    int aggregateValue = longColumn.getInt(rowIndex);
+            avgValueSetter = (int rowIndex) -> {
+                int aggregateValue = intColumn.getInt(rowIndex);
 
-                    if (this.zeroContributorCheck(countsByRow[rowIndex], aggregateValue != 0, rowIndex))
-                    {
-                        longColumn.setInt(rowIndex, aggregateValue / countsByRow[rowIndex]);
-                    }
+                if (this.zeroContributorCheck(countsByRow[rowIndex], aggregateValue != 0, rowIndex))
+                {
+                    intColumn.setInt(rowIndex, aggregateValue / countsByRow[rowIndex]);
                 }
-            }
+            };
         }
         else if (aggregatedColumn.getType().isDouble())
         {
             DfDoubleColumnStored doubleColumn = (DfDoubleColumnStored) aggregatedColumn;
 
-            for (int rowIndex = 0; rowIndex < columnSize; rowIndex++)
-            {
-                if (!doubleColumn.isNull(rowIndex))
+            avgValueSetter = (int rowIndex) -> {
+                double aggregateValue = doubleColumn.getDouble(rowIndex);
+
+                if (this.zeroContributorCheck(countsByRow[rowIndex], aggregateValue != 0, rowIndex))
                 {
-                    double aggregateValue = doubleColumn.getDouble(rowIndex);
+                    doubleColumn.setDouble(rowIndex, aggregateValue / countsByRow[rowIndex]);
+                }
+            };
+        }
+        else if (aggregatedColumn.getType().isFloat())
+        {
+            DfFloatColumnStored floatColumn = (DfFloatColumnStored) aggregatedColumn;
 
-                    if (this.zeroContributorCheck(countsByRow[rowIndex], aggregateValue != 0, rowIndex))
-                    {
-                        doubleColumn.setDouble(rowIndex, aggregateValue / countsByRow[rowIndex]);
-                    }
+            avgValueSetter = (int rowIndex) -> {
+                float aggregateValue = floatColumn.getFloat(rowIndex);
+
+                if (this.zeroContributorCheck(countsByRow[rowIndex], aggregateValue != 0, rowIndex))
+                {
+                    floatColumn.setFloat(rowIndex, aggregateValue / countsByRow[rowIndex]);
                 }
-            }
+            };
         }
         else if (aggregatedColumn.getType().isDecimal())
         {
             DfDecimalColumnStored decimalColumn = (DfDecimalColumnStored) aggregatedColumn;
 
-            for (int rowIndex = 0; rowIndex < columnSize; rowIndex++)
-            {
-                if (!decimalColumn.isNull(rowIndex))
-                {
-                    BigDecimal aggregateValue = decimalColumn.getTypedObject(rowIndex);
+            avgValueSetter = (int rowIndex) -> {
+                BigDecimal aggregateValue = decimalColumn.getTypedObject(rowIndex);
 
-                    if (this.zeroContributorCheck(countsByRow[rowIndex], aggregateValue.signum() != 0, rowIndex))
-                    {
-                        decimalColumn.setObject(rowIndex,
-                                aggregateValue.divide(BigDecimal.valueOf(countsByRow[rowIndex]), MathContext.DECIMAL128));
-                    }
+                if (this.zeroContributorCheck(countsByRow[rowIndex], aggregateValue.signum() != 0, rowIndex))
+                {
+                    decimalColumn.setObject(rowIndex,
+                            aggregateValue.divide(BigDecimal.valueOf(countsByRow[rowIndex]), MathContext.DECIMAL128));
                 }
+            };
+        }
+
+        for (int rowIndex = 0; rowIndex < columnSize; rowIndex++)
+        {
+            if (!aggregatedColumn.isNull(rowIndex))
+            {
+                avgValueSetter.value(rowIndex);
             }
         }
     }

diff --git a/src/main/java/io/github/vmzakharov/ecdataframe/dataframe/aggregation/Max.java b/src/main/java/io/github/vmzakharov/ecdataframe/dataframe/aggregation/Max.java
@@ -88,6 +88,12 @@ public double doubleInitialValue()
         return -Double.MAX_VALUE;
     }
 
+    @Override
+    public float floatInitialValue()
+    {
+        return -Float.MAX_VALUE;
+    }
+
     @Override
     public Object applyToObjectColumn(DfObjectColumn<?> objectColumn)
     {
@@ -114,6 +120,12 @@ protected double doubleAccumulator(double currentAggregate, double newValue)
         return Math.max(currentAggregate, newValue);
     }
 
+    @Override
+    protected float floatAccumulator(float currentAggregate, float newValue)
+    {
+        return Math.max(currentAggregate, newValue);
+    }
+
     @Override
     protected int intAccumulator(int currentAggregate, int newValue)
     {

diff --git a/src/main/java/io/github/vmzakharov/ecdataframe/dataframe/aggregation/Min.java b/src/main/java/io/github/vmzakharov/ecdataframe/dataframe/aggregation/Min.java
@@ -97,6 +97,12 @@ public double doubleInitialValue()
         return Double.MAX_VALUE;
     }
 
+    @Override
+    public float floatInitialValue()
+    {
+        return Float.MAX_VALUE;
+    }
+
     @Override
     public BigDecimal objectInitialValue()
     {
@@ -115,6 +121,12 @@ protected double doubleAccumulator(double currentAggregate, double newValue)
         return Math.min(currentAggregate, newValue);
     }
 
+    @Override
+    protected float floatAccumulator(float currentAggregate, float newValue)
+    {
+        return Math.min(currentAggregate, newValue);
+    }
+
     @Override
     protected int intAccumulator(int currentAggregate, int newValue)
     {

diff --git a/src/test/java/io/github/vmzakharov/ecdataframe/dataframe/DataFrameAggregationTest.java b/src/test/java/io/github/vmzakharov/ecdataframe/dataframe/DataFrameAggregationTest.java
@@ -97,7 +97,7 @@ public void averageEmptyIntThrowsException()
     @Test(expected = RuntimeException.class)
     public void averageEmptyFloatThrowsException()
     {
-        new DataFrame("FrameOfData").addStringColumn("Name").addIntColumn("Thud")
+        new DataFrame("FrameOfData").addStringColumn("Name").addFloatColumn("Thud")
                                     .aggregate(Lists.immutable.of(avg("Thud")));
     }