public static <K> KTableHolder<K> build(
      final KTableHolder<K> left,
      final KTableHolder<K> right,
      final TableTableJoin<K> join
  ) {
    final LogicalSchema leftSchema;
    final LogicalSchema rightSchema;

    if (join.getJoinType().equals(RIGHT)) {
      leftSchema = right.getSchema();
      rightSchema = left.getSchema();
    } else {
      leftSchema = left.getSchema();
      rightSchema = right.getSchema();
    }

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

    final KTable<K, GenericRow> result;
    switch (join.getJoinType()) {
      case INNER:
        result = left.getTable().join(right.getTable(), joinParams.getJoiner());
        break;
      case LEFT:
        result = left.getTable().leftJoin(right.getTable(), joinParams.getJoiner());
        break;
      case RIGHT:
        result = right.getTable().leftJoin(left.getTable(), joinParams.getJoiner());
        break;
      case OUTER:
        result = left.getTable().outerJoin(right.getTable(), joinParams.getJoiner());
        break;
      default:
        throw new IllegalStateException("invalid join type: " + join.getJoinType());
    }

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

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

    // Then:
    verify(leftKTable).join(
        same(rightKTable),
        eq(new KsqlValueJoiner(LEFT_SCHEMA.value().size(), RIGHT_SCHEMA.value().size(), 0))
    );
    verifyNoMoreInteractions(leftKTable, rightKTable, resultKTable);
    assertThat(result.getTable(), is(resultKTable));
    assertThat(result.getExecutionKeyFactory(), is(executionKeyFactory));
  }