public static Cli build(
      final Long streamedQueryRowLimit,
      final Long streamedQueryTimeoutMs,
      final OutputFormat outputFormat,
      final KsqlRestClient restClient
  ) {
    final Console console = Console.build(outputFormat);
    return new Cli(streamedQueryRowLimit, streamedQueryTimeoutMs, restClient, console);
  }
@Test
  public void testSelectProject() {
    final Map<GenericKey, GenericRow> expectedResults = ImmutableMap
        .<GenericKey, GenericRow>builder()
        .put(genericKey("ORDER_1"), genericRow(
            "ITEM_1",
            10.0,
            ImmutableList.of(100.0, 110.99, 90.0)))
        .put(genericKey("ORDER_2"), genericRow(
            "ITEM_2",
            20.0,
            ImmutableList.of(10.0, 10.99, 9.0)))
        .put(genericKey("ORDER_3"), genericRow(
            "ITEM_3",
            30.0,
            ImmutableList.of(10.0, 10.99, 91.0)))
        .put(genericKey("ORDER_4"), genericRow(
            "ITEM_4",
            40.0,
            ImmutableList.of(10.0, 140.99, 94.0)))
        .put(genericKey("ORDER_5"), genericRow(
            "ITEM_5",
            50.0,
            ImmutableList.of(160.0, 160.99, 98.0)))
        .put(genericKey("ORDER_6"), genericRow(
            "ITEM_8",
            80.0,
            ImmutableList.of(1100.0, 1110.99, 970.0)))
        .build();

    final PhysicalSchema resultSchema = PhysicalSchema.from(
        LogicalSchema.builder()
            .keyColumn(ColumnName.of("ORDERID"), SqlTypes.STRING)
            .valueColumn(ColumnName.of("ITEMID"), SqlTypes.STRING)
            .valueColumn(ColumnName.of("ORDERUNITS"), SqlTypes.DOUBLE)
            .valueColumn(ColumnName.of("PRICEARRAY"), SqlTypes.array(SqlTypes.DOUBLE))
            .build(),
        SerdeFeatures.of(),
        SerdeFeatures.of()
    );

    testCreateStreamAsSelect(
        "SELECT ORDERID, ITEMID, ORDERUNITS, PRICEARRAY "
            + "FROM " + ORDER_DATA_PROVIDER.sourceName() + ";",
        resultSchema,
        expectedResults
    );
  }