Fix failing unit tests for aggregations

engine/table/src/main/java/io/deephaven/engine/table/impl/by/AggregationContextFactory.java

@@ -8,18 +8,6 @@
  */
 public interface AggregationContextFactory {
 
- /**
- * Should we allow substitution with a {@link KeyOnlyAggregationFactory} (e.g. selectDistinct) when there are only
- * key columns? Instances whose operators could have side effects or are already {@link KeyOnlyAggregationFactory}
- * should return false.
- *
- * @return Whether to allow a {@link KeyOnlyAggregationFactory} to be substituted for this when there are only key
- * columns
- */
- default boolean allowKeyOnlySubstitution() {
- return false;
- }
-
  /**
  * Make an {@link AggregationContext} for this aggregation.
  *

engine/table/src/main/java/io/deephaven/engine/table/impl/by/AggregationFactory.java

@@ -1386,39 +1386,11 @@ public AggregationContextFactory makeAggregationContextFactory() {
  operators.add(formulaChunkedOperator);
  inputNames.add(CollectionUtil.ZERO_LENGTH_STRING_ARRAY);
  } else if (isWeightedAverage || isWeightedSum) {
- final String weightName;
-
- if (isWeightedAverage) {
- weightName = ((WeightedAverageSpecImpl) inputAggregationSpec)
- .getWeightName();
- } else {
- weightName =
- ((WeightedSumSpecImpl) inputAggregationSpec).getWeightName();
- }
-
- final ColumnSource<?> weightSource = table.getColumnSource(weightName);
- final DoubleWeightRecordingInternalOperator weightOperator =
- new DoubleWeightRecordingInternalOperator(weightSource.getChunkType());
- inputColumns.add(weightSource);
- operators.add(weightOperator);
-
- inputNames.add(Stream
- .concat(Stream.of(weightName),
- Arrays.stream(comboMatchPairs).map(MatchPair::rightColumn))
- .toArray(String[]::new));
-
- Arrays.stream(comboMatchPairs).forEach(mp -> {
- final ColumnSource<?> columnSource = table.getColumnSource(mp.rightColumn());
- inputColumns.add(columnSource);
- inputNames.add(new String[] {weightName, mp.rightColumn()});
- if (isWeightedAverage) {
- operators.add(new ChunkedWeightedAverageOperator(columnSource.getChunkType(),
- weightOperator, mp.leftColumn()));
- } else {
- operators.add(new DoubleChunkedWeightedSumOperator(columnSource.getChunkType(),
- weightOperator, mp.leftColumn()));
- }
- });
+ final String weightName = isWeightedAverage
+ ? ((WeightedAverageSpecImpl) inputAggregationSpec).getWeightName()
+ : ((WeightedSumSpecImpl) inputAggregationSpec).getWeightName();
+ WeightedAverageSumAggregationFactory.getOperatorsAndInputs(table,
+ weightName, isWeightedSum, comboMatchPairs, operators, inputNames, inputColumns);
  } else {
  throw new UnsupportedOperationException(
  "Unknown AggregationElementImpl: " + inputAggregationSpec.getClass());

engine/table/src/main/java/io/deephaven/engine/table/impl/by/ChunkedOperatorAggregationHelper.java

@@ -48,11 +48,6 @@
 @SuppressWarnings("rawtypes")
 public class ChunkedOperatorAggregationHelper {
 
- @VisibleForTesting
- public static boolean KEY_ONLY_SUBSTITUTION_ENABLED =
- Configuration.getInstance()
- .getBooleanWithDefault("ChunkedOperatorAggregationHelper.enableKeyOnlySubstitution", true);
-
  static final int CHUNK_SIZE = 1 << 12;
 
  public static QueryTable aggregation(AggregationContextFactory aggregationContextFactory, QueryTable queryTable,
@@ -68,15 +63,6 @@ public static QueryTable aggregation(AggregationControl control,
  && Arrays.stream(groupByColumns).anyMatch(selectColumn -> !selectColumn.isRetain());
  final QueryTable withView = !viewRequired ? queryTable : (QueryTable) queryTable.updateView(groupByColumns);
 
- final AggregationContextFactory aggregationContextFactoryToUse;
- if (KEY_ONLY_SUBSTITUTION_ENABLED
- && withView.getDefinition().getColumns().length == groupByColumns.length
- && aggregationContextFactory.allowKeyOnlySubstitution()) {
- aggregationContextFactoryToUse = new KeyOnlyAggregationFactory();
- } else {
- aggregationContextFactoryToUse = aggregationContextFactory;
- }
-
  if (queryTable.hasAttribute(Table.REVERSE_LOOKUP_ATTRIBUTE)) {
  withView.setAttribute(Table.REVERSE_LOOKUP_ATTRIBUTE,
  queryTable.getAttribute(Table.REVERSE_LOOKUP_ATTRIBUTE));
@@ -86,9 +72,9 @@ public static QueryTable aggregation(AggregationControl control,
  final SwapListener swapListener =
  withView.createSwapListenerIfRefreshing(SwapListener::new);
  withView.initializeWithSnapshot(
- "by(" + aggregationContextFactoryToUse + ", " + Arrays.toString(groupByColumns) + ")", swapListener,
+ "by(" + aggregationContextFactory + ", " + Arrays.toString(groupByColumns) + ")", swapListener,
  (usePrev, beforeClockValue) -> {
- resultHolder.setValue(aggregation(control, swapListener, aggregationContextFactoryToUse, withView,
+ resultHolder.setValue(aggregation(control, swapListener, aggregationContextFactory, withView,
  groupByColumns, usePrev));
  return true;
  });

engine/table/src/main/java/io/deephaven/engine/table/impl/by/FirstOrLastByAggregationFactory.java

@@ -26,11 +26,6 @@ public FirstOrLastByAggregationFactory(boolean isFirst, String exposeRedirection
  this.exposeRedirection = exposeRedirection == null ? null : NameValidator.validateColumnName(exposeRedirection);
  }
 
- @Override
- public boolean allowKeyOnlySubstitution() {
- return true;
- }
-
  @Override
  public AggregationContext makeAggregationContext(@NotNull final Table table,
  @NotNull final String... groupByColumns) {

engine/table/src/main/java/io/deephaven/engine/table/impl/by/FormulaAggregationFactory.java

@@ -27,11 +27,6 @@ public FormulaAggregationFactory(@NotNull final String formula, @NotNull final S
  this.columnParamName = columnParamName;
  }
 
- @Override
- public boolean allowKeyOnlySubstitution() {
- return true;
- }
-
  @Override
  public AggregationContext makeAggregationContext(@NotNull final Table inputTable,
  @NotNull final String... groupByColumnNames) {

engine/table/src/main/java/io/deephaven/engine/table/impl/by/FreezeByAggregationFactory.java

@@ -18,10 +18,6 @@
 import java.util.*;
 
 public class FreezeByAggregationFactory implements AggregationContextFactory {
- @Override
- public boolean allowKeyOnlySubstitution() {
- return true;
- }
 
  @Override
  public AggregationContext makeAggregationContext(@NotNull final Table table,

engine/table/src/main/java/io/deephaven/engine/table/impl/by/GroupByAggregationFactory.java

@@ -26,11 +26,6 @@ public static AggregationContextFactory getInstance() {
 
  private GroupByAggregationFactory() {}
 
- @Override
- public boolean allowKeyOnlySubstitution() {
- return true;
- }
-
  @Override
  public AggregationContext makeAggregationContext(@NotNull final Table inputTable,
  @NotNull final String... groupByColumnNames) {

engine/table/src/main/java/io/deephaven/engine/table/impl/by/NonKeyColumnAggregationFactory.java

@@ -17,11 +17,6 @@ public NonKeyColumnAggregationFactory(IterativeChunkedOperatorFactory iterativeC
  this.iterativeChunkedOperatorFactory = iterativeChunkedOperatorFactory;
  }
 
- @Override
- public boolean allowKeyOnlySubstitution() {
- return true;
- }
-
  @Override
  public AggregationContext makeAggregationContext(@NotNull final Table table,
  @NotNull final String... groupByColumns) {

engine/table/src/main/java/io/deephaven/engine/table/impl/by/SortedFirstOrLastByAggregationFactory.java

@@ -28,11 +28,6 @@ public SortedFirstOrLastByAggregationFactory(final boolean isFirst, final boolea
  this.sortColumns = sortColumns;
  }
 
- @Override
- public boolean allowKeyOnlySubstitution() {
- return true;
- }
-
  @Override
  public AggregationContext makeAggregationContext(@NotNull final Table table,
  @NotNull final String... groupByColumns) {