public static <KRightT> ForeignKeyJoinParams<KRightT> create(
      final ExpressionEvaluator expressionEvaluator,
      final LogicalSchema leftSchema,
      final LogicalSchema rightSchema,
      final ProcessingLogger processingLogger
  ) {
    if (rightSchema.key().size() != 1) {
      throw new IllegalStateException("rightSchema must have single column key");
    }
    return new ForeignKeyJoinParams<>(
        new KsqlKeyExtractor<>(expressionEvaluator, processingLogger),
        new KsqlValueJoiner(leftSchema.value().size(), rightSchema.value().size(), 0),
        createSchema(leftSchema, rightSchema)
    );
  }
@Test
  public void shouldBuildCorrectKeyedSchema() {
    // When:
    final ForeignKeyJoinParams<String> joinParams =
        ForeignKeyJoinParamsFactory.create(mock(ExpressionEvaluator.class), LEFT_SCHEMA, RIGHT_SCHEMA, null);

    // 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_FOREIGN_KEY"), 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.INTEGER)
        .build())
    );
  }