Skip to content

Commit

Permalink
Updated CountWhere and RollingFormula to accept a MatchPair
Browse files Browse the repository at this point in the history
  • Loading branch information
@lbooker42
lbooker42 committed Feb 7, 2025
1 parent 8ca23c4 commit ec41424
Show file tree
Hide file tree
Showing 5 changed files with 114 additions and 69 deletions.
12 changes: 5 additions & 7 deletions .../table/src/main/java/io/deephaven/engine/table/impl/updateby/UpdateByOperatorFactory.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -440,7 +440,7 @@ public Void visit(CumProdSpec cps) {

@Override
public Void visit(CumCountWhereSpec spec) {
ops.add(makeCountWhereOperator(tableDef, spec));
ops.add(makeCountWhereOperator(pairs[0], tableDef, spec));
return null;
}

Expand Down Expand Up @@ -569,7 +569,7 @@ public Void visit(@NotNull final RollingCountSpec spec) {

@Override
public Void visit(@NotNull final RollingCountWhereSpec spec) {
ops.add(makeCountWhereOperator(tableDef, spec));
ops.add(makeCountWhereOperator(pairs[0], tableDef, spec));
return null;
}

Expand All @@ -583,7 +583,7 @@ public Void visit(@NotNull final RollingFormulaSpec spec) {

// noinspection deprecation
if (spec.paramToken().isEmpty()) {
ops.add(makeRollingFormulaMultiColumnOperator(tableDef, spec));
ops.add(makeRollingFormulaMultiColumnOperator(pairs[0], tableDef, spec));
return null;
}

Expand Down Expand Up @@ -1280,6 +1280,7 @@ private UpdateByOperator makeRollingCountOperator(@NotNull final MatchPair pair,
* This is used for Cum/Rolling CountWhere operators
*/
private UpdateByOperator makeCountWhereOperator(
@NotNull final MatchPair pair,
@NotNull final TableDefinition tableDef,
@NotNull final UpdateBySpec spec) {

Expand Down Expand Up @@ -1360,7 +1361,6 @@ private UpdateByOperator makeCountWhereOperator(
final String columnName = isCumulative
? ((CumCountWhereSpec) spec).column().name()
: ((RollingCountWhereSpec) spec).column().name();
final MatchPair pair = new MatchPair(columnName, columnName);

// Create and return the operator.
if (isCumulative) {
Expand Down Expand Up @@ -1595,6 +1595,7 @@ private UpdateByOperator makeRollingFormulaOperator(@NotNull final MatchPair pai
}

private UpdateByOperator makeRollingFormulaMultiColumnOperator(
@NotNull final MatchPair pair,
@NotNull final TableDefinition tableDef,
@NotNull final RollingFormulaSpec rs) {
final long prevWindowScaleUnits = rs.revWindowScale().getTimeScaleUnits();
Expand Down Expand Up @@ -1659,9 +1660,6 @@ private UpdateByOperator makeRollingFormulaMultiColumnOperator(
affectingColumns = ArrayUtils.add(inputNonKeyColumns, rs.revWindowScale().timestampCol());
}

// Create a new column pair with the same name for the left and right columns
final MatchPair pair = new MatchPair(selectColumn.getName(), selectColumn.getName());

return new RollingFormulaMultiColumnOperator(
pair,
affectingColumns,
Expand Down
107 changes: 79 additions & 28 deletions engine/table/src/test/java/io/deephaven/engine/table/impl/updateby/TestUpdateByGeneral.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -8,11 +8,11 @@
import io.deephaven.api.updateby.OperationControl;
import io.deephaven.api.updateby.UpdateByOperation;
import io.deephaven.engine.context.ExecutionContext;
import io.deephaven.engine.context.QueryScope;
import io.deephaven.engine.table.Table;
import io.deephaven.engine.table.impl.QueryTable;
import io.deephaven.engine.table.impl.UpdateErrorReporter;
import io.deephaven.engine.table.impl.util.AsyncClientErrorNotifier;
import io.deephaven.engine.table.impl.util.ColumnHolder;
import io.deephaven.engine.testutil.ControlledUpdateGraph;
import io.deephaven.engine.testutil.EvalNugget;
import io.deephaven.engine.table.impl.TableDefaults;
Expand All @@ -38,8 +38,7 @@
import java.time.Duration;
import java.util.*;

import static io.deephaven.api.updateby.UpdateByOperation.CumMax;
import static io.deephaven.api.updateby.UpdateByOperation.RollingMax;
import static io.deephaven.api.updateby.UpdateByOperation.*;
import static io.deephaven.engine.testutil.GenerateTableUpdates.generateAppends;
import static io.deephaven.engine.testutil.TstUtils.*;
import static io.deephaven.engine.testutil.testcase.RefreshingTableTestCase.simulateShiftAwareStep;
Expand Down Expand Up @@ -157,7 +156,7 @@ protected Table e() {
UpdateByOperation.Ema(skipControl, "ts", 10 * MINUTE,
makeOpColNames(columnNamesArray, "_ema", "Sym", "ts", "boolCol")),
UpdateByOperation.CumSum(makeOpColNames(columnNamesArray, "_sum", "Sym", "ts")),
UpdateByOperation.CumMin(makeOpColNames(columnNamesArray, "_min", "boolCol")),
CumMin(makeOpColNames(columnNamesArray, "_min", "boolCol")),
CumMax(makeOpColNames(columnNamesArray, "_max", "boolCol")),
UpdateByOperation
.CumProd(makeOpColNames(columnNamesArray, "_prod", "Sym", "ts", "boolCol")));
Expand Down Expand Up @@ -275,38 +274,36 @@ public void testInMemoryColumn() {
final Collection<? extends UpdateByOperation> clauses = List.of(
UpdateByOperation.Fill(),

UpdateByOperation.RollingGroup(50, 50,
RollingGroup(50, 50,
makeOpColNames(columnNamesArray, "_rollgroupfwdrev", "Sym", "ts")),
UpdateByOperation.RollingGroup("ts", Duration.ofMinutes(5), Duration.ofMinutes(5),
RollingGroup("ts", Duration.ofMinutes(5), Duration.ofMinutes(5),
makeOpColNames(columnNamesArray, "_rollgrouptimefwdrev", "Sym", "ts")),

UpdateByOperation.RollingSum(100, 0,
RollingSum(100, 0,
makeOpColNames(columnNamesArray, "_rollsumticksrev", "Sym", "ts", "boolCol")),
UpdateByOperation.RollingSum("ts", Duration.ofMinutes(15), Duration.ofMinutes(0),
RollingSum("ts", Duration.ofMinutes(15), Duration.ofMinutes(0),
makeOpColNames(columnNamesArray, "_rollsumtimerev", "Sym", "ts", "boolCol")),

UpdateByOperation.RollingAvg(100, 0,
RollingAvg(100, 0,
makeOpColNames(columnNamesArray, "_rollavgticksrev", "Sym", "ts", "boolCol")),
UpdateByOperation.RollingAvg("ts", Duration.ofMinutes(15), Duration.ofMinutes(0),
RollingAvg("ts", Duration.ofMinutes(15), Duration.ofMinutes(0),
makeOpColNames(columnNamesArray, "_rollavgtimerev", "Sym", "ts", "boolCol")),

UpdateByOperation.RollingMin(100, 0,
RollingMin(100, 0,
makeOpColNames(columnNamesArray, "_rollminticksrev", "Sym", "ts", "boolCol")),
UpdateByOperation.RollingMin("ts", Duration.ofMinutes(5), Duration.ofMinutes(0),
RollingMin("ts", Duration.ofMinutes(5), Duration.ofMinutes(0),
makeOpColNames(columnNamesArray, "_rollmintimerev", "Sym", "ts", "boolCol")),

RollingMax(100, 0,
makeOpColNames(columnNamesArray, "_rollmaxticksrev", "Sym", "ts", "boolCol")),
RollingMax("ts", Duration.ofMinutes(5), Duration.ofMinutes(0),
makeOpColNames(columnNamesArray, "_rollmaxtimerev", "Sym", "ts", "boolCol")),

UpdateByOperation.Ema(skipControl, "ts", 10 * MINUTE,
makeOpColNames(columnNamesArray, "_ema", "Sym", "ts", "boolCol")),
UpdateByOperation.CumSum(makeOpColNames(columnNamesArray, "_sum", "Sym", "ts")),
UpdateByOperation.CumMin(makeOpColNames(columnNamesArray, "_min", "boolCol")),
Ema(skipControl, "ts", 10 * MINUTE, makeOpColNames(columnNamesArray, "_ema", "Sym", "ts", "boolCol")),
CumSum(makeOpColNames(columnNamesArray, "_sum", "Sym", "ts")),
CumMin(makeOpColNames(columnNamesArray, "_min", "boolCol")),
CumMax(makeOpColNames(columnNamesArray, "_max", "boolCol")),
UpdateByOperation
.CumProd(makeOpColNames(columnNamesArray, "_prod", "Sym", "ts", "boolCol")));
CumProd(makeOpColNames(columnNamesArray, "_prod", "Sym", "ts", "boolCol")));
final UpdateByControl control = UpdateByControl.builder().useRedirection(false).build();

final Table table = result.t.updateBy(control, clauses, ColumnName.from("Sym"));
Expand All @@ -328,19 +325,73 @@ public void run() {

@Test
public void testResultColumnOrdering() {
QueryScope.addParam("baseTime", DateTimeUtils.parseInstant("2022-03-09T09:00:00.000 NY"));
final Table source = emptyTable(10).update("Timestamp = baseTime + i * SECOND", "X = ii % 10");
final Table source = emptyTable(5).update("X=ii");

final ColumnHolder<?> x = longCol("X", 0, 1, 2, 3, 4);
final ColumnHolder<?> cumMin = longCol("cumMin", 0, 0, 0, 0, 0);
final ColumnHolder<?> cumMax = longCol("cumMax", 0, 1, 2, 3, 4);
final ColumnHolder<?> rollingMin = longCol("rollingMin", 0, 0, 1, 2, 3);
final ColumnHolder<?> rollingMax = longCol("rollingMax", 0, 1, 2, 3, 4);

final Table result_1 = source.updateBy(List.of(
CumMax("cumMaxTime=Timestamp"),
RollingMax(20, "rollingMaxTime=Timestamp")), "X");
Assert.assertArrayEquals(result_1.getDefinition().getColumnNamesArray(),
new String[] {"Timestamp", "X", "cumMaxTime", "rollingMaxTime"});
CumMin("cumMin=X"),
CumMax("cumMax=X"),
RollingMin(2, "rollingMin=X"),
RollingMax(2, "rollingMax=X")));
final Table expected_1 = TableTools.newTable(x, cumMin, cumMax, rollingMin, rollingMax);
Assert.assertEquals("", diff(result_1, expected_1, 10));

final Table result_2 = source.updateBy(List.of(
RollingMax(20, "rollingMaxTime=Timestamp"),
CumMax("cumMaxTime=Timestamp")), "X");
Assert.assertArrayEquals(result_2.getDefinition().getColumnNamesArray(),
new String[] {"Timestamp", "X", "rollingMaxTime", "cumMaxTime"});
CumMax("cumMax=X"),
CumMin("cumMin=X"),
RollingMax(2, "rollingMax=X"),
RollingMin(2, "rollingMin=X")));
final Table expected_2 = TableTools.newTable(x, cumMax, cumMin, rollingMax, rollingMin);
Assert.assertEquals("", diff(result_2, expected_2, 10));

final Table result_3 = source.updateBy(List.of(
RollingMin(2, "rollingMin=X"),
RollingMax(2, "rollingMax=X"),
CumMin("cumMin=X"),
CumMax("cumMax=X")));
final Table expected_3 = TableTools.newTable(x, rollingMin, rollingMax, cumMin, cumMax);
Assert.assertEquals("", diff(result_3, expected_3, 10));

final Table result_4 = source.updateBy(List.of(
RollingMax(2, "rollingMax=X"),
RollingMin(2, "rollingMin=X"),
CumMax("cumMax=X"),
CumMin("cumMin=X")));
final Table expected_4 = TableTools.newTable(x, rollingMax, rollingMin, cumMax, cumMin);
Assert.assertEquals("", diff(result_4, expected_4, 10));

final Table result_5 = source.updateBy(List.of(
CumMin("cumMin=X"),
RollingMin(2, "rollingMin=X"),
CumMax("cumMax=X"),
RollingMax(2, "rollingMax=X")));
final Table expected_5 = TableTools.newTable(x, cumMin, rollingMin, cumMax, rollingMax);
Assert.assertEquals("", diff(result_5, expected_5, 10));

// Trickiest one, since we internally combine groupBy operations.
final Table source_2 = source.update("Y=ii % 2");
final Table result_6 = source_2.updateBy(List.of(
CumMin("cumMin=X"),
RollingGroup(2, "rollingGroupY=Y"),
RollingMin(2, "rollingMin=X"),
CumMax("cumMax=X"),
RollingGroup(2, "rollingGroupX=X"),
RollingMax(2, "rollingMax=X")));

Assert.assertArrayEquals(result_6.getDefinition().getColumnNamesArray(),
new String[] {
"X",
"Y",
"cumMin",
"rollingGroupY",
"rollingMin",
"cumMax",
"rollingGroupX",
"rollingMax"});
}
}
Loading

0 comments on commit ec41424

Please sign in to comment.