public static List<FieldInfo> buildSourceSchemaEntity(final LogicalSchema schema) {
    final List<FieldInfo> allFields = schema.columns().stream()
        .map(EntityUtil::toFieldInfo)
        .collect(Collectors.toList());

    if (allFields.isEmpty()) {
      throw new IllegalArgumentException("Root schema should contain columns: " + schema);
    }

    return allFields;
  }
@Test
  public void shouldSupportRowTimeAndKeyInValueSchema() {
    // Given:
    final LogicalSchema schema = LogicalSchema.builder()
        .valueColumn(ColumnName.of("ROWKEY"), SqlTypes.STRING)
        .valueColumn(ColumnName.of("ROWTIME"), SqlTypes.INTEGER)
        .valueColumn(ColumnName.of("field1"), SqlTypes.INTEGER)
        .build();

    // When:
    final List<FieldInfo> fields = EntityUtil.buildSourceSchemaEntity(schema);

    // Then:
    assertThat(fields, hasSize(3));
    assertThat(fields.get(0).getName(), equalTo("ROWKEY"));
    assertThat(fields.get(0).getType(), equalTo(Optional.empty()));
    assertThat(fields.get(1).getName(), equalTo("ROWTIME"));
    assertThat(fields.get(1).getType(), equalTo(Optional.empty()));
  }