public static <K> KStreamHolder<K> build(
      final KStreamHolder<K> left,
      final KTableHolder<K> right,
      final StreamTableJoin<K> join,
      final RuntimeBuildContext buildContext,
      final JoinedFactory joinedFactory
  ) {
    final Formats leftFormats = join.getInternalFormats();
    final QueryContext queryContext = join.getProperties().getQueryContext();
    final QueryContext.Stacker stacker = QueryContext.Stacker.of(queryContext);
    final LogicalSchema leftSchema = left.getSchema();
    final PhysicalSchema leftPhysicalSchema = PhysicalSchema.from(
        leftSchema,
        leftFormats.getKeyFeatures(),
        leftFormats.getValueFeatures()
    );

    final Serde<GenericRow> leftSerde = buildContext.buildValueSerde(
        leftFormats.getValueFormat(),
        leftPhysicalSchema,
        stacker.push(SERDE_CTX).getQueryContext()
    );
    final Serde<K> keySerde = left.getExecutionKeyFactory().buildKeySerde(
        leftFormats.getKeyFormat(),
        leftPhysicalSchema,
        queryContext
    );
    final Joined<K, GenericRow, GenericRow> joined = joinedFactory.create(
        keySerde,
        leftSerde,
        null,
        StreamsUtil.buildOpName(queryContext)
    );
    final LogicalSchema rightSchema = right.getSchema();

    final JoinParams joinParams = JoinParamsFactory
        .create(join.getKeyColName(), leftSchema, rightSchema);

    final KStream<K, GenericRow> result;
    switch (join.getJoinType()) {
      case LEFT:
        result = left.getStream().leftJoin(right.getTable(), joinParams.getJoiner(), joined);
        break;
      case INNER:
        result = left.getStream().join(right.getTable(), joinParams.getJoiner(), joined);
        break;
      default:
        throw new IllegalStateException("invalid join type");
    }
    return left.withStream(result, joinParams.getSchema());
  }
@Test
  public void shouldDoInnerJoinWithSyntheticKey() {
    // Given:
    givenInnerJoin(SYNTH_KEY);

    // When:
    final KStreamHolder<Struct> result = join.build(planBuilder, planInfo);

    // Then:
    verify(leftKStream).join(
        same(rightKTable),
        eq(new KsqlValueJoiner(LEFT_SCHEMA.value().size(), RIGHT_SCHEMA.value().size(), 1)),
        same(joined)
    );
    verifyNoMoreInteractions(leftKStream, rightKTable, resultStream);
    assertThat(result.getStream(), is(resultStream));
    assertThat(result.getExecutionKeyFactory(), is(executionKeyFactory));
  }