feat: move selectKey impl to plan builder

[rodesai]

Description

Moves rekeying from SchemaKStream and into an execution plan builder.

Testing done

Added tests for step builder

Reviewer checklist

• Ensure docs are updated if necessary. (eg. if a user visible feature is being added or changed).
• Ensure relevant issues are linked (description should include text like "Fixes #")

[big-andy-coates]

thanks @rodesai LGTM

[big-andy-coates]

consider moving into method to make tests more readable:

    private ValueMapperWithKey<Object, GenericRow, GenericRow> getMapper() { 
       verify(rekeyedKstream).mapValues(mapperCaptor.capture());
       final ValueMapperWithKey<Object, GenericRow, GenericRow> mapper = mapperCaptor.getValue();
    }

Same for others.

[big-andy-coates]

consider moving into method to make tests more readable:

    private ValueMapperWithKey<Object, GenericRow, GenericRow> getMapper() { 
       verify(rekeyedKstream).mapValues(mapperCaptor.capture());
       final ValueMapperWithKey<Object, GenericRow, GenericRow> mapper = mapperCaptor.getValue();
    }

Same for others.

[big-andy-coates]

double checked this PR after seeing issues with raw types and generics in the other related PRs... looks like this one has some minor generic issues to clean up too.

[big-andy-coates]

use of raw type.

We need to get the generics right here.

Will StreamSelectKey always build a KStream<Struct, GenericRow> regardless of whether it's source is KStream<Windowed<Struct>, GenericRow> or KStream<Struct, GenericRow>??? I think it does, right?

That being the case, StreamSelectKey should be defined as:

public class StreamSelectKey implements ExecutionStep<KStream<Struct, GenericRow>> {

  private final ExecutionStep<KStream<?, GenericRow>> source;
  ...

  public StreamSelectKey(
      ...
      final ExecutionStep<KStream<?, GenericRow>> source,
     ...
  )

  ...
}

And this method should be:

public static StreamSelectKey streamSelectKey(
      final QueryContext.Stacker stacker,
      final ExecutionStep<KStream<?, GenericRow>> source,
      final String fieldName,
      final boolean updateRowKey
  ) {
    final QueryContext queryContext = stacker.getQueryContext();
    return new StreamSelectKey(
        new DefaultExecutionStepProperties(
            source.getProperties().getSchema(),
            queryContext
        ),
        source,
        fieldName,
        updateRowKey
    );
  }

[big-andy-coates]

Use of raw types - can we please use generic types in tests.

[big-andy-coates]

Thanks @rodesai, this looks much better!

[big-andy-coates]

Do we need K here? Can we not just use ExecutionStep<KStream<?, GenericRow>> for the source param / field

[rodesai]

You can't pass an ExecutionStep<KStream<K, GenericRow>>, which is what SchemaKStream has, to ExecutionStep<KStream<?, GenericRow>>. We could make the type ExecutionStep<? extends KStream<?, GenericRow>>, but this felt easier to read.

[big-andy-coates]

consider moving into method to make tests more readable:

    private ValueMapperWithKey<Object, GenericRow, GenericRow> getMapper() { 
       verify(rekeyedKstream).mapValues(mapperCaptor.capture());
       final ValueMapperWithKey<Object, GenericRow, GenericRow> mapper = mapperCaptor.getValue();
    }

Same for others.