public static KStreamHolder<GenericKey> build(
      final KStreamHolder<?> stream,
      final StreamSelectKeyV1 selectKey,
      final RuntimeBuildContext buildContext
  ) {
    final LogicalSchema sourceSchema = stream.getSchema();

    final CompiledExpression expression = buildExpressionEvaluator(
        selectKey,
        buildContext,
        sourceSchema
    );

    final ProcessingLogger processingLogger = buildContext
        .getProcessingLogger(selectKey.getProperties().getQueryContext());

    final String errorMsg = "Error extracting new key using expression "
        + selectKey.getKeyExpression();

    final Function<GenericRow, Object> evaluator = val -> expression
        .evaluate(val, null, processingLogger, () -> errorMsg);

    final LogicalSchema resultSchema = new StepSchemaResolver(buildContext.getKsqlConfig(),
        buildContext.getFunctionRegistry()).resolve(selectKey, sourceSchema);

    final KStream<?, GenericRow> kstream = stream.getStream();
    final KStream<GenericKey, GenericRow> rekeyed = kstream
        .filter((key, val) -> val != null && evaluator.apply(val) != null)
        .selectKey((key, val) -> GenericKey.genericKey(evaluator.apply(val)));

    return new KStreamHolder<>(
        rekeyed,
        resultSchema,
        ExecutionKeyFactory.unwindowed(buildContext)
    );
  }
@Test
  public void shouldRekeyCorrectly() {
    // When:
    final KStreamHolder<GenericKey> result = selectKey.build(planBuilder, planInfo);

    // Then:
    final InOrder inOrder = Mockito.inOrder(kstream, filteredKStream, rekeyedKstream);
    inOrder.verify(kstream).filter(any());
    inOrder.verify(filteredKStream).selectKey(any());
    inOrder.verifyNoMoreInteractions();
    assertThat(result.getStream(), is(rekeyedKstream));
  }