Make table.Running return integer typed columns for min/max

std-bits/table/src/main/java/org/enso/table/operations/AddRunning.java

@@ -5,6 +5,8 @@
 import org.enso.base.statistics.Statistic;
 import org.enso.table.data.column.storage.Storage;
 import org.enso.table.data.column.storage.numeric.DoubleStorage;
+import org.enso.table.data.column.storage.numeric.LongStorage;
+import org.enso.table.data.column.storage.type.IntegerType;
 import org.enso.table.data.table.Column;
 import org.enso.table.data.table.problems.IgnoredNaN;
 import org.enso.table.data.table.problems.IgnoredNothing;
@@ -31,7 +33,7 @@ public static Storage<?> create_running(
     return runningStatistic.getResult();
   }
 
-  private static RunningStatistic<Double> createRunningStatistic(
+  private static RunningStatistic<?> createRunningStatistic(
       Statistic statistic, Column sourceColumn, ProblemAggregator problemAggregator) {
     switch (statistic) {
       case Sum -> {
@@ -41,38 +43,98 @@ private static RunningStatistic<Double> createRunningStatistic(
         return new RunningMeanStatistic(sourceColumn, problemAggregator);
       }
       case Minimum -> {
+        if (sourceColumn.getStorage().getType() instanceof IntegerType type) {
+          return new RunningMinLongStatistic(sourceColumn, problemAggregator, type);
+        }
         return new RunningMinStatistic(sourceColumn, problemAggregator);
       }
       case Maximum -> {
+        if (sourceColumn.getStorage().getType() instanceof IntegerType type) {
+          return new RunningMaxLongStatistic(sourceColumn, problemAggregator, type);
+        }
         return new RunningMaxStatistic(sourceColumn, problemAggregator);
       }
       default -> throw new IllegalArgumentException("Unsupported statistic: " + statistic);
     }
   }
 
-  private abstract static class RunningStatisticBase implements RunningStatistic<Double> {
+  private interface TypeHandler<T> {
+
+    T tryConvertingToType(Object o);
+
+    long typeToRawLongBits(T t);
+
+    Storage<T> createStorage(long[] result, int size, BitSet isNothing);
+  }
+
+  private static class DoubleHandler implements TypeHandler<Double> {
+
+    @Override
+    public Double tryConvertingToType(Object o) {
+      return NumericConverter.tryConvertingToDouble(o);
+    }
+
+    @Override
+    public long typeToRawLongBits(Double d) {
+      return Double.doubleToRawLongBits(d);
+    }
+
+    @Override
+    public Storage<Double> createStorage(long[] result, int size, BitSet isNothing) {
+      return new DoubleStorage(result, size, isNothing);
+    }
+  }
+
+  private static class LongHandler implements TypeHandler<Long> {
+
+    IntegerType type;
+
+    LongHandler(IntegerType type) {
+      this.type = type;
+    }
+
+    @Override
+    public Long tryConvertingToType(Object o) {
+      return NumericConverter.tryConvertingToLong(o);
+    }
+
+    @Override
+    public long typeToRawLongBits(Long l) {
+      return l;
+    }
+
+    @Override
+    public Storage<Long> createStorage(long[] result, int size, BitSet isNothing) {
+      return new LongStorage(result, size, isNothing, type);
+    }
+  }
+
+  private abstract static class RunningStatisticBase<T> implements RunningStatistic<T> {
 
     long[] result;
     BitSet isNothing;
     ColumnAggregatedProblemAggregator columnAggregatedProblemAggregator;
     Column sourceColumn;
+    TypeHandler<T> typeHandler;
 
-    RunningStatisticBase(Column sourceColumn, ProblemAggregator problemAggregator) {
+    RunningStatisticBase(
+        Column sourceColumn, ProblemAggregator problemAggregator, TypeHandler<T> typeHandler) {
       result = new long[sourceColumn.getSize()];
       isNothing = new BitSet();
       columnAggregatedProblemAggregator = new ColumnAggregatedProblemAggregator(problemAggregator);
       this.sourceColumn = sourceColumn;
+      this.typeHandler = typeHandler;
     }
 
     @Override
-    public void calculateNextValue(int i, RunningIterator<Double> it) {
+    public void calculateNextValue(int i, RunningIterator<T> it) {
       Object value = sourceColumn.getStorage().getItemBoxed(i);
       if (value == null) {
         columnAggregatedProblemAggregator.reportColumnAggregatedProblem(
             new IgnoredNothing(sourceColumn.getName(), i));
       }
-      Double dValue = NumericConverter.tryConvertingToDouble(value);
-      Double dNextValue;
+      T dValue = typeHandler.tryConvertingToType(value);
+      T dNextValue;
       if (dValue != null && dValue.equals(Double.NaN)) {
         columnAggregatedProblemAggregator.reportColumnAggregatedProblem(
             new IgnoredNaN(sourceColumn.getName(), i));
@@ -83,13 +145,13 @@ public void calculateNextValue(int i, RunningIterator<Double> it) {
       if (dNextValue == null) {
         isNothing.set(i);
       } else {
-        result[i] = Double.doubleToRawLongBits(dNextValue);
+        result[i] = typeHandler.typeToRawLongBits(dNextValue);
       }
     }
 
     @Override
-    public Storage<Double> getResult() {
-      return new DoubleStorage(result, sourceColumn.getSize(), isNothing);
+    public Storage<T> getResult() {
+      return typeHandler.createStorage(result, sourceColumn.getSize(), isNothing);
     }
   }
 
@@ -127,10 +189,10 @@ protected double getCurrent() {
     }
   }
 
-  private static class RunningSumStatistic extends RunningStatisticBase {
+  private static class RunningSumStatistic extends RunningStatisticBase<Double> {
 
     RunningSumStatistic(Column sourceColumn, ProblemAggregator problemAggregator) {
-      super(sourceColumn, problemAggregator);
+      super(sourceColumn, problemAggregator, new DoubleHandler());
     }
 
     @Override
@@ -147,10 +209,10 @@ public void increment(double value) {
     }
   }
 
-  private static class RunningMeanStatistic extends RunningStatisticBase {
+  private static class RunningMeanStatistic extends RunningStatisticBase<Double> {
 
     RunningMeanStatistic(Column sourceColumn, ProblemAggregator problemAggregator) {
-      super(sourceColumn, problemAggregator);
+      super(sourceColumn, problemAggregator, new DoubleHandler());
     }
 
     @Override
@@ -181,10 +243,10 @@ public double getCurrent() {
     }
   }
 
-  private static class RunningMinStatistic extends RunningStatisticBase {
+  private static class RunningMinStatistic extends RunningStatisticBase<Double> {
 
     RunningMinStatistic(Column sourceColumn, ProblemAggregator problemAggregator) {
-      super(sourceColumn, problemAggregator);
+      super(sourceColumn, problemAggregator, new DoubleHandler());
     }
 
     @Override
@@ -201,10 +263,31 @@ public void increment(double value) {
     }
   }
 
-  private static class RunningMaxStatistic extends RunningStatisticBase {
+  private static class RunningMinLongStatistic extends RunningStatisticBase<Long> {
+
+    RunningMinLongStatistic(
+        Column sourceColumn, ProblemAggregator problemAggregator, IntegerType type) {
+      super(sourceColumn, problemAggregator, new LongHandler(type));
+    }
+
+    @Override
+    public RunningIterator<Long> getNewIterator() {
+      return new RunningMinLongIterator();
+    }
+
+    private static class RunningMinLongIterator extends RunningIteratorLong {
+
+      @Override
+      public void increment(long value) {
+        current = Math.min(current, value);
+      }
+    }
+  }
+
+  private static class RunningMaxStatistic extends RunningStatisticBase<Double> {
 
     RunningMaxStatistic(Column sourceColumn, ProblemAggregator problemAggregator) {
-      super(sourceColumn, problemAggregator);
+      super(sourceColumn, problemAggregator, new DoubleHandler());
     }
 
     @Override
@@ -220,4 +303,59 @@ public void increment(double value) {
       }
     }
   }
+
+  private static class RunningMaxLongStatistic extends RunningStatisticBase<Long> {
+
+    RunningMaxLongStatistic(
+        Column sourceColumn, ProblemAggregator problemAggregator, IntegerType type) {
+      super(sourceColumn, problemAggregator, new LongHandler(type));
+    }
+
+    @Override
+    public RunningIterator<Long> getNewIterator() {
+      return new RunningMaxLongIterator();
+    }
+
+    private static class RunningMaxLongIterator extends RunningIteratorLong {
+
+      @Override
+      public void increment(long value) {
+        current = Math.max(current, value);
+      }
+    }
+  }
+
+  private abstract static class RunningIteratorLong implements RunningIterator<Long> {
+
+    protected long current;
+    private boolean isInitialized = false;
+
+    @Override
+    public Long next(Long value) {
+      if (value != null) {
+        if (!isInitialized) {
+          isInitialized = true;
+          initialize(value);
+        } else {
+          increment(value);
+        }
+      }
+      return isInitialized ? getCurrent() : null;
+    }
+
+    @Override
+    public Long currentValue() {
+      return isInitialized ? getCurrent() : null;
+    }
+
+    protected void initialize(long value) {
+      current = value;
+    }
+
+    protected abstract void increment(long value);
+
+    protected long getCurrent() {
+      return current;
+    }
+  }
 }