public static KTableHolder<GenericKey> build(
      final KGroupedStreamHolder groupedStream,
      final StreamAggregate aggregate,
      final RuntimeBuildContext buildContext,
      final MaterializedFactory materializedFactory) {
    return build(
        groupedStream,
        aggregate,
        buildContext,
        materializedFactory,
        new AggregateParamsFactory()
    );
  }
@Test
  public void shouldReturnCorrectSerdeForWindowedAggregate() {
    for (final Runnable given : given()) {
      // Given:
      clearInvocations(groupedStream, timeWindowedStream, sessionWindowedStream, aggregated, buildContext);
      given.run();

      // When:
      final KTableHolder<Windowed<GenericKey>> tableHolder = windowedAggregate.build(planBuilder, planInfo);

      // Then:
      final ExecutionKeyFactory<Windowed<GenericKey>> serdeFactory = tableHolder.getExecutionKeyFactory();
      final FormatInfo mockFormat = mock(FormatInfo.class);
      final PhysicalSchema mockSchema = mock(PhysicalSchema.class);
      final QueryContext mockCtx = mock(QueryContext.class);
      serdeFactory.buildKeySerde(mockFormat, mockSchema, mockCtx);
      verify(buildContext).buildKeySerde(
          same(mockFormat),
          eq(windowedAggregate.getWindowExpression().getWindowInfo()),
          same(mockSchema),
          same(mockCtx)
      );
    }
  }