### Description

This is to investigate what we could do to support the data streams in SQL. The backing indices can be queried already, but to address them similarly as they're queried over QDSL one would need to construct a pattern learned out of band and not through SQL metadata discovery mechanisms.
x-pack/plugin/sql/qa/server/src/main/java/org/elasticsearch/xpack/sql/qa/rest/RestSqlTestCase.java
}
    }
    static Map<String, Object> runSql(RequestObjectBuilder builder, String mode) throws IOException {
        return toMap(runSql(builder.mode(mode)), mode);
    }
    static Response runSql(RequestObjectBuilder builder) throws IOException {
        return runSqlAsTextWithFormat(builder, null);
    }
    static Response runSqlAsTextWithFormat(RequestObjectBuilder builder, @Nullable String format) throws IOException {
        Request request = new Request("POST", SQL_QUERY_REST_ENDPOINT);
        request.addParameter("error_trace", "true");   
        request.addParameter("pretty", "true");        
        if (format != null) {
            request.addParameter(URL_PARAM_FORMAT, format);        
        }
        request.setEntity(new StringEntity(builder.toString(), ContentType.APPLICATION_JSON));
        return client().performRequest(request);
    }
    static void assertAsyncResponse(Map<String, Object> expected, Map<String, Object> actual) {
        String actualId = (String) actual.get("id");
        assertTrue("async ID missing in response", hasText(actualId));
        expected.put("id", actualId);
        assertResponse(expected, actual);
    }
    private static String expectedTextBody(String format, int fetchSize, int count, Character csvDelimiter) {
        StringBuilder sb = new StringBuilder();
        if (count == 0) { 
            switch (format) {
                case "txt":
                    sb.append("     text      |    number     \n");