public static <KLeftT, KRightT> KTableHolder<KLeftT> build(
      final KTableHolder<KLeftT> left,
      final KTableHolder<KRightT> right,
      final ForeignKeyTableTableJoin<KLeftT, KRightT> join,
      final RuntimeBuildContext buildContext
  ) {
    final LogicalSchema leftSchema = left.getSchema();
    final LogicalSchema rightSchema = right.getSchema();

    final ProcessingLogger logger = buildContext.getProcessingLogger(
        join.getProperties().getQueryContext()
    );

    final ExpressionEvaluator expressionEvaluator;
    final CodeGenRunner codeGenRunner = new CodeGenRunner(
        leftSchema,
        buildContext.getKsqlConfig(),
        buildContext.getFunctionRegistry()
    );

    final Optional<ColumnName> leftColumnName = join.getLeftJoinColumnName();
    final Optional<Expression> leftJoinExpression = join.getLeftJoinExpression();
    if (leftColumnName.isPresent()) {
      expressionEvaluator = codeGenRunner.buildCodeGenFromParseTree(
          new UnqualifiedColumnReferenceExp(leftColumnName.get()),
          "Left Join Expression"
      );
    } else if (leftJoinExpression.isPresent()) {
      expressionEvaluator = codeGenRunner.buildCodeGenFromParseTree(
          leftJoinExpression.get(),
          "Left Join Expression"
      );
    } else {
      throw new IllegalStateException("Both leftColumnName and leftJoinExpression are empty.");
    }

    final ForeignKeyJoinParams<KRightT> joinParams = ForeignKeyJoinParamsFactory
        .create(expressionEvaluator, leftSchema, rightSchema, logger);

    final Formats formats = join.getFormats();

    final PhysicalSchema physicalSchema = PhysicalSchema.from(
        joinParams.getSchema(),
        formats.getKeyFeatures(),
        formats.getValueFeatures()
    );

    final Serde<KLeftT> keySerde = left.getExecutionKeyFactory().buildKeySerde(
        formats.getKeyFormat(),
        physicalSchema,
        join.getProperties().getQueryContext()
    );
    final Serde<GenericRow> valSerde = buildContext.buildValueSerde(
        formats.getValueFormat(),
        physicalSchema,
        join.getProperties().getQueryContext()
    );

    final KTable<KLeftT, GenericRow> result;
    switch (join.getJoinType()) {
      case INNER:
        result = left.getTable().join(
            right.getTable(),
            joinParams.getKeyExtractor(),
            joinParams.getJoiner(),
            buildContext.getMaterializedFactory().create(keySerde, valSerde)
        );
        break;
      case LEFT:
        result = left.getTable().leftJoin(
            right.getTable(),
            joinParams.getKeyExtractor(),
            joinParams.getJoiner(),
            buildContext.getMaterializedFactory().create(keySerde, valSerde)
        );
        break;
      default:
        throw new IllegalStateException("invalid join type: " + join.getJoinType());
    }

    return KTableHolder.unmaterialized(
            result,
            joinParams.getSchema(),
            left.getExecutionKeyFactory()
    );
  }
@Test
  public void shouldReturnCorrectSchema() {
    // Given:
    givenInnerJoin(left, JOIN_COLUMN);

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

    // Then:
    assertThat(
        result.getSchema(),
        is(LogicalSchema.builder()
            .keyColumns(LEFT_SCHEMA.key())
            .valueColumns(LEFT_SCHEMA.value())
            .valueColumns(RIGHT_SCHEMA.value())
            .build()
        )
    );
  }