public static JoinParams create(
      final ColumnName keyColName,
      final LogicalSchema leftSchema,
      final LogicalSchema rightSchema
  ) {
    final boolean appendKey = neitherContain(keyColName, leftSchema, rightSchema);
    return new JoinParams(
        new KsqlValueJoiner(leftSchema.value().size(), rightSchema.value().size(), appendKey ? 1 : 0
        ),
        createSchema(keyColName, leftSchema, rightSchema)
    );
  }
@Test
  public void shouldBuildCorrectLeftKeyedSchema() {
    // Given:
    final ColumnName keyName = Iterables.getOnlyElement(LEFT_SCHEMA.key()).name();

    // When:
    final JoinParams joinParams = JoinParamsFactory.create(keyName, LEFT_SCHEMA, RIGHT_SCHEMA);

    // Then:
    assertThat(joinParams.getSchema(), is(LogicalSchema.builder()
        .keyColumn(ColumnName.of("L_K"), SqlTypes.STRING)
        .valueColumn(ColumnName.of("L_BLUE"), SqlTypes.STRING)
        .valueColumn(ColumnName.of("L_GREEN"), SqlTypes.INTEGER)
        .valueColumn(ColumnName.of("L_K"), SqlTypes.STRING)
        .valueColumn(ColumnName.of("R_RED"), SqlTypes.BIGINT)
        .valueColumn(ColumnName.of("R_ORANGE"), SqlTypes.DOUBLE)
        .valueColumn(ColumnName.of("R_K"), SqlTypes.STRING)
        .build())
    );
  }