feat: move setup of the sink to plan builders

[rodesai]

Description

This patch moves stream and table sinks (SchemaKX.into) into execution
plan builders. This change also moves computation of the rowkey/rowtime
indexes to be excluded from the output into the plan builder.

Testing done

Added unit tests for step builders.

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

Few nits, but my main concern is the incorrect use of generics in this new code. I think we need to think about how we want the generics on these new types, and on ExecutionStep to work.

I hope you agree we shouldn't be committing new code with incorrect use of generics and raw types.

Happy to jump on a call and talk it through with you if you want.

[big-andy-coates]

As per comment in other PR the generics on this are wrong.

#3361 (comment)

[big-andy-coates]

wrong generics: kstream would be either KStream<Windowed<Struct>, GenericRow> or KStream<Struct, GenericRow>. Neither of which can be cast to KStream<Object, GenericRow>.

Maybe this method should be something like:

  public static <K> void build(
      final KStream<K, GenericRow> kstream,
      final StreamSink<KStream<K, GenericRow>> streamSink,
      final KsqlQueryBuilder queryBuilder) {

???

[big-andy-coates]

As above, wrong generics.

[big-andy-coates]

Test using raw types - can we please use generic types.

[big-andy-coates]

Test using raw types - can we please use generic types.

[rodesai]

Once we build the streams app from a visitor, we'll actually get a factory from the source node. For now, we can just pass in the serde we have.

[big-andy-coates]

I actually think that the Logical model shouldn't know anything about serializers. The source node should know about schemas, format and serde options only. The physical builder should be able to convert these to a serde, or a serde factory.

[rodesai]

@big-andy-coates that's what I mean. By source node I mean the "source execution step of this step". So the physical builder will do something like:

class KStreamAndKeySerdeFactory<K> {
    KStream<K, GenericRow> kstream;
    KeySerdeFactory<K> factory;
}

KStreamAndKeySerdeFactory<K> visitStreamSink(final StreamSink<K> streamSink) {
    final KStreamAndKeySerdeFactory<K> k = streamSink.getSource().accept(this);
    StreamSinkBuilder.build(streamSink.kstream, streamSink, streamSink.factory, ...);
    ...
}

[big-andy-coates]

Thanks @rodesai looks much better!

[big-andy-coates]

No pertaining to this PR directly, but sometimes rowkeyIndexes is empty. In such a situation we don't actually need the map step. Removing this would obviously change out topology, which is a pain, but maybe we could do:

 mapValues(row -> {
              if (row == null || rowkeyIndexes.isEmpty()) {
                return row;
              }

             ...
}

To avoid the unnecessary copy of the row into another list? Just a small optimisation.

[big-andy-coates]

Can we not use a mock context?

[big-andy-coates]

Use of MockitoRule triggers spotbugs errors for me locally. Can we use @RunWith(MockitoJUnitRunner.class) instead please?

[big-andy-coates]

As above.