public static <K> KTableHolder<K> build(
      final KTableHolder<K> table,
      final TableFilter<K> step,
      final RuntimeBuildContext buildContext) {
    return build(table, step, buildContext, SqlPredicate::new);
  }
@Test
  public void shouldFilterMaterialization() {
    // When:
    step.build(planBuilder, planInfo);

    // Then:
    verify(materializationBuilder).filter(
        predicateFactoryCaptor.capture(),
        eq(queryContext));

    // Given:
    final KsqlTransformer<Object, Optional<GenericRow>> predicate = predicateFactoryCaptor
        .getValue()
        .apply(processingLogger);

    when(preTransformer.transform(any(), any(), any())).thenReturn(Optional.empty());

    // When:
    Optional<GenericRow> result = predicate.transform(key, value, ctx);

    // Then:
    verify(preTransformer).transform(key, value, ctx);
    assertThat(result, is(Optional.empty()));

    // Given:
    when(preTransformer.transform(any(), any(), any()))
        .thenAnswer(inv -> Optional.of(inv.getArgument(1)));

    // When:
    result = predicate.transform(key, value, ctx);

    // Then:
    assertThat(result, is(Optional.of(value)));
  }