feat: move joins to plan builder

[rodesai]

Description

Moves code for joining kstreams/ktables out of SchemaKStream/Table
and into execution plan builders. Also moves KsqlValueJoiner into
ksql-streams, so that it can be used from the plan builders. Also
adds a buildKeySerde API for building both windowed and unwindowed
keys, since most of the plan builders dont have type information
for the key.

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 called out below. Main issue is that the generics on the new code are wrong. We should be striving to have correct generics on new code. See comments inline.

[big-andy-coates]

The generics on this call are wrong. It's a pervasive anti-pattern within the code base, but it's wrong. Returning KeySerde<Object> is saying its a key serde that can handle Objects, but it can't. The returned serde can only handle either Struct or Windowed<Struct>. The fact that you're having to first cast to (KeySerde) is a clear indication of this: KeySerde<Struct> can not be cast to KeySerde<Object> because they are not compatible types.

This is why there are two methods for explicitly handling windowed vs non-windowed serde.

The correct return type for this method is KeySerde<?> i.e. a key serde of an unknown type. This of course gives up type safety, which we should avoid if we can.

The following requires no casts and generates no warnings, though it will of course require other code to change:

   public KeySerde<?> buildKeySerde(
      final KeyFormat keyFormat,
      final PhysicalSchema physicalSchema,
      final QueryContext queryContext
  ) {
    if (keyFormat.getWindowInfo().isPresent()) {
      return buildKeySerde(
          keyFormat.getFormatInfo(),
          keyFormat.getWindowInfo().get(),
          physicalSchema,
          queryContext
      );
    }

    return buildKeySerde(
        keyFormat.getFormatInfo(),
        physicalSchema,
        queryContext
    );
  }

Unfortunately, the generics on our key type are... foobared, and will require some effort to correct and give us type safety. When introducing new functionality or refactoring existing, we should endeavour to make it type safe and then us hacky casts at the point this interacts with old code.

[big-andy-coates]

Generics are wrong again. This should be KStream<?, GenericRow>, which again is unfortunately giving up any type information on the key :(

Really, the signature of this should be:

public static <K>  KStream<K, GenericRow> build(
     final KStream<K, GenericRow> left,
      final KStream<K, GenericRow> right,
      final StreamStreamJoin join,
      final KsqlQueryBuilder queryBuilder,
      final JoinedFactory joinedFactory
) {

[big-andy-coates]

As above

[big-andy-coates]

Use of raw type - this should have generic types set.

[big-andy-coates]

As above

[big-andy-coates]

This test us using raw types. Can we please change the test to use generic types please for all these fields.

Same goes for the other tests.

[big-andy-coates]

Thanks @rodesai, LGTM 'cept a few remaining raw types...

[big-andy-coates]

Feels like this should be somewhere it can be shared with other things that do a join and need the same constant.

[rodesai]

I left them separate because the names don't need to be the same for different join types.

[big-andy-coates]

raw type?

[big-andy-coates]

As per other PR - can we use @RunWith(MockitoJUnitRunner.class) please?

[big-andy-coates]

raw type?