@Override
  public LogicalSchema getSchema() {
    return schema;
  }
@Test
  public void shouldHaveFullyQualifiedJoinSchemaWithSyntheticKey() {
    // Given:
    when(joinKey.resolveKeyName(any(), any())).thenReturn(SYNTH_KEY);

    // When:
    final JoinNode joinNode = new JoinNode(nodeId, OUTER, joinKey,       true, left,
        right, empty(),"KAFKA");

    // When:
    assertThat(joinNode.getSchema(), is(LogicalSchema.builder()
        .keyColumn(SYNTH_KEY, SqlTypes.BIGINT)
        .valueColumn(ColumnName.of(LEFT_ALIAS.text() + "_C0"), SqlTypes.BIGINT)
        .valueColumn(ColumnName.of(LEFT_ALIAS.text() + "_L1"), SqlTypes.STRING)
        .valueColumn(ColumnName.of(LEFT_ALIAS.text() + "_ROWTIME"), SqlTypes.BIGINT)
        .valueColumn(ColumnName.of(LEFT_ALIAS.text() + "_ROWPARTITION"), SqlTypes.INTEGER)
        .valueColumn(ColumnName.of(LEFT_ALIAS.text() + "_ROWOFFSET"), SqlTypes.BIGINT)
        .valueColumn(ColumnName.of(LEFT_ALIAS.text() + "_leftKey"), SqlTypes.BIGINT)
        .valueColumn(ColumnName.of(RIGHT_ALIAS.text() + "_C0"), SqlTypes.STRING)
        .valueColumn(ColumnName.of(RIGHT_ALIAS.text() + "_R1"), SqlTypes.BIGINT)
        .valueColumn(ColumnName.of(RIGHT_ALIAS.text() + "_ROWTIME"), SqlTypes.BIGINT)
        .valueColumn(ColumnName.of(RIGHT_ALIAS.text() + "_ROWPARTITION"), SqlTypes.INTEGER)
        .valueColumn(ColumnName.of(RIGHT_ALIAS.text() + "_ROWOFFSET"), SqlTypes.BIGINT)
        .valueColumn(ColumnName.of(RIGHT_ALIAS.text() + "_rightKey"), SqlTypes.BIGINT)
        .valueColumn(SYNTH_KEY, SqlTypes.BIGINT)
        .build()
    ));
  }