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 testSelectUDFs() {
    final String queryString = String.format(
        "SELECT "
            + "ORDERID, "
            + "ITEMID, "
            + "ORDERUNITS*10 AS Col1, "
            + "PRICEARRAY[1]+10 AS Col2, "
            + "KEYVALUEMAP['key1']*KEYVALUEMAP['key2']+10 AS Col3, "
            + "PRICEARRAY[2]>1000 AS Col4 "
            + "FROM %s "
            + "WHERE ORDERUNITS > 20 AND ITEMID LIKE '%%_8';",
        ORDER_DATA_PROVIDER.sourceName()
    );

    final PhysicalSchema resultSchema = PhysicalSchema.from(
        LogicalSchema.builder()
            .keyColumn(ColumnName.of("ORDERID"), SqlTypes.STRING)
            .valueColumn(ColumnName.of("ITEMID"), SqlTypes.STRING)
            .valueColumn(ColumnName.of("COL1"), SqlTypes.DOUBLE)
            .valueColumn(ColumnName.of("COL2"), SqlTypes.DOUBLE)
            .valueColumn(ColumnName.of("COL3"), SqlTypes.DOUBLE)
            .valueColumn(ColumnName.of("COL4"), SqlTypes.BOOLEAN)
            .build(),
        SerdeFeatures.of(),
        SerdeFeatures.of()
    );

    final Map<GenericKey, GenericRow> expectedResults = ImmutableMap.of(
        genericKey("ORDER_6"),
        genericRow("ITEM_8", 800.0, 1110.0, 12.0, true)
    );

    testCreateStreamAsSelect(queryString, resultSchema, expectedResults);
  }