Repository: jtablesaw/tablesaw
Pull request #581 (2019-08-09 05:00:20)

Add rolling stream method
Thanks for contributing.

- [X] Tick to sign-off your agreement to the [Developer Certificate of Origin (DCO) 1.1](https://developercertificate.org)

## Description

This adds a rolling `stream(n)` method which allows us to do many different operations such as `forEach`, `map`, `filter`, `anyMatch`, `allMatch`, etc. on rolling groups of rows. `rollWithRows(Consumer<Row[]> rows, int n)` is the same as `stream(n).forEach(consumer)`, so I've deprecated it since it's the same thing, but less powerful

@ryancerf depending on which PR is merged first (i.e. this one or https://github.com/jtablesaw/tablesaw/pull/579) either you or I might need to rebase to put this method on `Relation`

## Testing

Existing unit tests cover this functionality. They're effective too because they caught a bug in my initial implementation.

benmccann: What do you think about the name? Would `rollingStream(n)` be better than `stream(n)`?
lwhite1: > This adds a rolling stream(n) method which allows us to do many different operations such as forEach, map, filter, anyMatch, allMatch, etc. on rolling groups of rows. rollWithRows(Consumer<Row[]> rows, int n) is the same as stream(n).forEach(consumer), so I've deprecated it since it's the same thing, but less powerful

There is so much awesome stuff getting done now. It's really exciting.

> What do you think about the name? Would rollingStream(n) be better than stream(n)?

So long as you're rolling (i.e. moving forward one row at a time as in a rolling average), I think rollingStream is preferable. I often, also, find it useful to step forward by groups of n rows. We may want to support that option. Or is that built in here as well? If you can do either with this implementation I prefer stream().  



ryancerf: This looks good to me, but I would like to discuss the possibility of supporting analytic functions. See #582.  Analytic functions would provide similar rolling stream support.
benmccann: I renamed to `rollingStream` and added a `steppingStream` as well

Patch:
diff --git a/core/src/main/java/tech/tablesaw/aggregate/AggregateFunctions.java b/core/src/main/java/tech/tablesaw/aggregate/AggregateFunctions.java
index 4543677f..8eaaa849 100644
--- a/core/src/main/java/tech/tablesaw/aggregate/AggregateFunctions.java
+++ b/core/src/main/java/tech/tablesaw/aggregate/AggregateFunctions.java
@@ -203,6 +203,17 @@ public class AggregateFunctions {
         }
       };
 
+  /** A function that calculates the coefficient of variation (stdDev/mean) of the values in the column */
+  public static final NumericAggregateFunction cv =
+      new NumericAggregateFunction("CV") {
+
+        @Override
+        public Double summarize(NumericColumn<?> column) {
+          double[] col = removeMissing(column);
+          return Math.sqrt(StatUtils.variance(col)) / StatUtils.mean(col);
+        }
+      };
+
   /** A function that calculates the sum of the values in the column param */
   public static final NumericAggregateFunction sum =
       new NumericAggregateFunction("Sum") {
diff --git a/core/src/main/java/tech/tablesaw/aggregate/Summarizer.java b/core/src/main/java/tech/tablesaw/aggregate/Summarizer.java
index afb88524..cac2bf64 100644
--- a/core/src/main/java/tech/tablesaw/aggregate/Summarizer.java
+++ b/core/src/main/java/tech/tablesaw/aggregate/Summarizer.java
@@ -18,7 +18,6 @@ import com.google.common.base.Preconditions;
 import com.google.common.collect.ArrayListMultimap;
 import java.util.ArrayList;
 import java.util.List;
-import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.Function;
 import java.util.stream.Collectors;
 import tech.tablesaw.api.CategoricalColumn;
@@ -33,6 +32,8 @@ import tech.tablesaw.selection.Selection;
 import tech.tablesaw.table.StandardTableSliceGroup;
 import tech.tablesaw.table.TableSliceGroup;
 
+import static tech.tablesaw.api.QuerySupport.numberColumn;
+
 /**
  * Summarizes the data in a table, by applying functions to a subset of its columns.
  *
@@ -46,7 +47,7 @@ public class Summarizer {
 
   private String[] groupColumnNames = new String[0];
   private final Table original;
-  private final Table temp;
+  private Table temp;
   private final List<String> summarizedColumns = new ArrayList<>();
   private final AggregateFunction<?, ?>[] reductions;
   private static final String GROUP_COL_TEMP_NAME = "_temp_group_col_";
@@ -198,12 +199,7 @@ public class Summarizer {
 
   private Table getSummaryTable(IntColumn groupColumn) {
     TableSliceGroup group = StandardTableSliceGroup.create(temp, groupColumn);
-    Table t = summarize(group);
-
-    // remove the last row if it has no value, that means it didn't have enough records for a full
-    // step
-    t = removeLastRowIfMissingGroup(groupColumn, t);
-    return t;
+    return summarize(group);
   }
 
   /**
@@ -269,15 +265,7 @@ public class Summarizer {
     if (groupColumnNames[0].equals(GROUP_COL_TEMP_NAME)) {
       IntColumn groupColumn = temp.intColumn(GROUP_COL_TEMP_NAME);
       TableSliceGroup group = StandardTableSliceGroup.create(temp, groupColumn);
-      Table t = summarizeForHaving(group, selection);
-
-      // remove the last row if it has no value, that means it didn't have enough records for a full
-      // step
-      t = removeLastRowIfMissingGroup(groupColumn, t);
-
-      t.column(GROUP_COL_TEMP_NAME).setName("Group");
-      return t;
-
+      return summarizeForHaving(group, selection);
     } else {
       TableSliceGroup group = StandardTableSliceGroup.create(temp, groupColumnNames);
       return summarizeForHaving(group, selection);
@@ -308,7 +296,9 @@ public class Summarizer {
 
   public Summarizer groupBy(int step) {
     IntColumn groupColumn = assignToGroupsByStep(step);
-
+    if (tableDoesNotContain(groupColumn.name(), temp)) {
+      temp.addColumns(groupColumn);
+    }
     groupColumnNames = new String[] {GROUP_COL_TEMP_NAME};
 
     return this;
@@ -343,15 +333,28 @@ public class Summarizer {
   private IntColumn assignToGroupsByStep(int step) {
     IntColumn groupColumn = IntColumn.create(GROUP_COL_TEMP_NAME, temp.rowCount());
     temp.addColumns(groupColumn);
-    AtomicInteger groupId = new AtomicInteger(1);
-    temp.stepWithRows(
-        rows -> {
-          int id = groupId.getAndIncrement();
-          for (Row row : rows) {
-            groupColumn.set(row.getRowNumber(), id);
-          }
-        },
-        step);
+
+    int groupId = 1;
+    int withinGroupCount = 0;
+    Row row = new Row(temp);
+
+    while(row.hasNext()) {
+      row.next();
+      if (withinGroupCount < step) {
+        withinGroupCount++;
+        groupColumn.set(row.getRowNumber(), groupId);
+      } else {
+        groupId++;
+        groupColumn.set(row.getRowNumber(), groupId);
+        withinGroupCount = 1;
+      }
+    }
+    int lastGroupSize =
+            temp.where(numberColumn(GROUP_COL_TEMP_NAME).isEqualTo(groupId)).rowCount();
+    if (lastGroupSize < step) {
+      temp = temp.dropWhere(numberColumn(GROUP_COL_TEMP_NAME).isEqualTo(groupId));
+    }
+    temp.addColumns(IntColumn.indexColumn("index", temp.rowCount(), 1));
     return groupColumn;
   }
 
diff --git a/core/src/main/java/tech/tablesaw/api/Row.java b/core/src/main/java/tech/tablesaw/api/Row.java
index 434cda24..140ee138 100644
--- a/core/src/main/java/tech/tablesaw/api/Row.java
+++ b/core/src/main/java/tech/tablesaw/api/Row.java
@@ -39,9 +39,13 @@ public class Row implements Iterator<Row> {
   private int rowNumber;
 
   public Row(Table table) {
+    this(table, -1);
+  }
+
+  public Row(Table table, int rowNumber) {
     this.table = table;
     columnNames = table.columnNames().toArray(new String[0]);
-    rowNumber = -1;
+    this.rowNumber = rowNumber;
     for (Column<?> column : table.columns()) {
       if (column instanceof DoubleColumn) {
         doubleColumnMap.put(column.name(), (DoubleColumn) column);
diff --git a/core/src/main/java/tech/tablesaw/api/Table.java b/core/src/main/java/tech/tablesaw/api/Table.java
index 19f4eef9..45885b8a 100644
--- a/core/src/main/java/tech/tablesaw/api/Table.java
+++ b/core/src/main/java/tech/tablesaw/api/Table.java
@@ -1069,15 +1069,92 @@ public class Table extends Relation implements Iterable<Row> {
     };
   }
 
+  /**
+   * Iterates over rolling sets of rows. I.e. 0 to n-1, 1 to n, 2 to n+1, etc.
+   *
+   * @param n the number of rows to return for each iteration
+   */
+  public Iterator<Row[]> rollingIterator(int n) {
+
+    return new Iterator<Row[]>() {
+
+      private int currRow = 0;
+
+      @Override
+      public Row[] next() {
+        Row[] rows = new Row[n];
+        for (int i = 0; i < n; i++) {
+          rows[i] = new Row(Table.this, currRow + i);
+        }
+        currRow++;
+        return rows;
+      }
+
+      @Override
+      public boolean hasNext() {
+        return currRow + n <= rowCount();
+      }
+    };
+  }
+
+  /**
+   * Streams over stepped sets of rows. I.e. 0 to n-1, n to 2n-1, 2n to 3n-1, etc. Only returns full
+   * sets of rows.
+   *
+   * @param n the number of rows to return for each iteration
+   */
+  public Iterator<Row[]> steppingIterator(int n) {
+
+    return new Iterator<Row[]>() {
+
+      private int currRow = 0;
+
+      @Override
+      public Row[] next() {
+        Row[] rows = new Row[n];
+        for (int i = 0; i < n; i++) {
+          rows[i] = new Row(Table.this, currRow + i);
+        }
+        currRow += n;
+        return rows;
+      }
+
+      @Override
+      public boolean hasNext() {
+        return currRow + n <= rowCount();
+      }
+    };
+  }
+
   public Stream<Row> stream() {
     return Streams.stream(iterator());
   }
 
+  /**
+   * Streams over stepped sets of rows. I.e. 0 to n-1, n to 2n-1, 2n to 3n-1, etc. Only returns full
+   * sets of rows.
+   *
+   * @param n the number of rows to return for each iteration
+   */
+  public Stream<Row[]> steppingStream(int n) {
+    return Streams.stream(steppingIterator(n));
+  }
+
+  /**
+   * Streams over rolling sets of rows. I.e. 0 to n-1, 1 to n, 2 to n+1, etc.
+   *
+   * @param n the number of rows to return for each iteration
+   */
+  public Stream<Row[]> rollingStream(int n) {
+    return Streams.stream(rollingIterator(n));
+  }
+
   /**
    * Applies the operation in {@code doable} to every row in the table
    *
    * @deprecated use {@code stream().forEach}
    */
+  @Deprecated
   public void doWithRows(Consumer<Row> doable) {
     stream().forEach(doable);
   }
@@ -1087,84 +1164,36 @@ public class Table extends Relation implements Iterable<Row> {
    *
    * @deprecated use {@code stream().anyMatch}
    */
+  @Deprecated
   public boolean detect(Predicate<Row> predicate) {
     return stream().anyMatch(predicate);
   }
 
-  /** Applies the operation in {@code rowConsumer} to every series of n rows in the table */
+  /** @deprecated use steppingStream(n).forEach(rowConsumer) */
+  @Deprecated
   public void stepWithRows(Consumer<Row[]> rowConsumer, int n) {
-    if (isEmpty()) {
-      return;
-    }
-    Row[] rows = new Row[n];
-    for (int i = 0; i < n; i++) {
-      rows[i] = new Row(this);
-    }
-
-    int max = rowCount() / n;
-
-    for (int i = 0; i < max; i++) { // 0, 1
-      for (int r = 1; r <= n; r++) {
-        int row = i * n + r - 1;
-        rows[r - 1].at(row);
-      }
-      rowConsumer.accept(rows);
-    }
+    steppingStream(n).forEach(rowConsumer);
   }
 
-  /** Applies the function in {@code pairs} to each consecutive pairs of rows in the table */
+  /** @deprecated use stream(2).forEach(rowConsumer) */
+  @Deprecated
   public void doWithRows(Pairs pairs) {
-    if (isEmpty()) {
-      return;
-    }
-    Row row1 = new Row(this);
-    Row row2 = new Row(this);
-    int max = rowCount();
-    for (int i = 1; i < max; i++) {
-      row1.at(i - 1);
-      row2.at(i);
-      pairs.doWithPair(row1, row2);
-    }
+    rollingStream(2).forEach(rows -> pairs.doWithPair(rows[0], rows[1]));
   }
 
-  /** Applies the function in {@code pairConsumer} to each consecutive pairs of rows in the table */
+  /** @deprecated use stream(2).forEach(rowConsumer) */
+  @Deprecated
   public void doWithRowPairs(Consumer<RowPair> pairConsumer) {
-    if (isEmpty()) {
-      return;
-    }
-    Row row1 = new Row(this);
-    Row row2 = new Row(this);
-    RowPair pair = new RowPair(row1, row2);
-    int max = rowCount();
-    for (int i = 1; i < max; i++) {
-      row1.at(i - 1);
-      row2.at(i);
-      pairConsumer.accept(pair);
-    }
+    rollingStream(2).forEach(rows -> pairConsumer.accept(new RowPair(rows[0], rows[1])));
   }
 
-  /**
-   * Applies the function in {@code rowConsumer} to each group of contiguous rows of size n in the
-   * table This can be used, for example, to calculate a running average of in rows
-   */
+  /** @deprecated use stream(n).forEach(rowConsumer) */
+  @Deprecated
   public void rollWithRows(Consumer<Row[]> rowConsumer, int n) {
-    if (isEmpty()) {
-      return;
-    }
-    Row[] rows = new Row[n];
-    for (int i = 0; i < n; i++) {
-      rows[i] = new Row(this);
-    }
-
-    int max = rowCount() - (n - 2);
-    for (int i = 1; i < max; i++) {
-      for (int r = 0; r < n; r++) {
-        rows[r].at(i + r - 1);
-      }
-      rowConsumer.accept(rows);
-    }
+    rollingStream(n).forEach(rowConsumer);
   }
 
+  @Deprecated
   public static class RowPair {
     private final Row first;
     private final Row second;
@@ -1183,6 +1212,7 @@ public class Table extends Relation implements Iterable<Row> {
     }
   }
 
+  @Deprecated
   interface Pairs {
 
     void doWithPair(Row row1, Row row2);
diff --git a/core/src/main/java/tech/tablesaw/columns/numbers/NumberMapFunctions.java b/core/src/main/java/tech/tablesaw/columns/numbers/NumberMapFunctions.java
index 94144085..f3af7d81 100644
--- a/core/src/main/java/tech/tablesaw/columns/numbers/NumberMapFunctions.java
+++ b/core/src/main/java/tech/tablesaw/columns/numbers/NumberMapFunctions.java
@@ -101,6 +101,7 @@ public interface NumberMapFunctions {
       throw new IllegalArgumentException("The columns must have the same number of elements");
 
     DoubleColumn result = DoubleColumn.create(name() + " * " + column2.name(), col1Size);
+
     for (int r = 0; r < col1Size; r++) {
       result.set(r, multiply(getDouble(r), column2.getDouble(r)));
     }
@@ -197,11 +198,7 @@ public interface NumberMapFunctions {
   default DoubleColumn power(NumericColumn<?> powerColumn) {
     DoubleColumn result = DoubleColumn.create(name() + "[pow]", size());
     for (int i = 0; i < size(); i++) {
-      if (isMissing(i) || (powerColumn.isMissing(i))) {
-        result.setMissing(i);
-      } else {
-        result.set(i, Math.pow(getDouble(i), powerColumn.getDouble(i)));
-      }
+      result.set(i, Math.pow(getDouble(i), powerColumn.getDouble(i)));
     }
     return result;
   }
diff --git a/core/src/main/java/tech/tablesaw/filtering/DeferredNumberColumn.java b/core/src/main/java/tech/tablesaw/filtering/DeferredNumberColumn.java
index e1c73e49..456eb326 100644
--- a/core/src/main/java/tech/tablesaw/filtering/DeferredNumberColumn.java
+++ b/core/src/main/java/tech/tablesaw/filtering/DeferredNumberColumn.java
@@ -16,6 +16,10 @@ public class DeferredNumberColumn extends DeferredColumn {
     return table -> table.numberColumn(getColumnName()).isEqualTo(other);
   }
 
+  public Function<Table, Selection> isEqualTo(double other) {
+    return table -> table.numberColumn(getColumnName()).isEqualTo(other);
+  }
+
   public Function<Table, Selection> isBetweenExclusive(double start, double end) {
     return table -> table.numberColumn(getColumnName()).isBetweenExclusive(start, end);
   }
diff --git a/docs/gettingstarted.md b/docs/gettingstarted.md
index 3642f808..9000fa45 100644
--- a/docs/gettingstarted.md
+++ b/docs/gettingstarted.md
@@ -318,17 +318,14 @@ for (Row row : table) {
 There are better ways, however. Another approach lets you skip the iteration and just provide a Consumer for each row.
 
 ```java
-// Create a consumer as an object or lambda (show below)
-Consumer<Row> doable = row -> {
-    if (row.getRowNumber() < 5) {
-        System.out.println("On "
-                           + row.getDate("date")
-                           + ": "
-                           + row.getDouble("approval"));
-    }
-};
-// apply the lambda
-table.doWithRows(doable);
+table.stream().forEach(row -> {
+  if (row.getRowNumber() < 5) {
+    System.out.println("On "
+                     + row.getDate("date")
+                     + ": "
+                     + row.getDouble("approval"));
+  }
+});
 ```
 
 If you need to process more than one row at a time, there are several methods to help. 
@@ -336,13 +333,11 @@ If you need to process more than one row at a time, there are several methods to
 ```java
 // work with a sliding window of rows
 // for example 0, 1, and 2, then 1, 2, and 3. etc.
-table.rollWithRows(consumer, 3);		
+table.stream(3).forEach(consumer);
 
 // work with a shifting window of rows
 // for example rows 0 through 4, then 5 through 9, etc.
-table.stepWithRows(consumer, 5);		
-
-table.doWithRowPairs(Pairs)	// easy syntax for working with each pair of rows
+table.steppingStream(5).forEach(consumer);
 ```
 
 See [Rows](https://jtablesaw.github.io/tablesaw/userguide/rows) for more information and other options. 


Test patch:
diff --git a/core/src/test/java/tech/tablesaw/aggregate/AggregateFunctionsTest.java b/core/src/test/java/tech/tablesaw/aggregate/AggregateFunctionsTest.java
index e2e01327..ab0e680b 100644
--- a/core/src/test/java/tech/tablesaw/aggregate/AggregateFunctionsTest.java
+++ b/core/src/test/java/tech/tablesaw/aggregate/AggregateFunctionsTest.java
@@ -141,6 +141,13 @@ class AggregateFunctionsTest {
 
   @Test
   void testGroupMean3b() {
+    Summarizer function = table.summarize("approval", mean, stdDev);
+    Table result = function.groupBy(10).apply();
+    assertEquals(32, result.rowCount());
+  }
+
+  @Test
+  void testGroupMean3c() {
     Summarizer function = table.summarize("approval", mean, stdDev);
     Table result = function.groupBy(10).having(num("mean [approval]").isGreaterThan(60));
     assertEquals(21, result.rowCount());