### 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
List<String> texts = IntStream.range(0, size).mapToObj(i -> String.format(Locale.ROOT, "text%02d", i)).toList();
        index(texts.stream().map(t -> "{\"field\": \"" + t + "\"}").toArray(String[]::new));
        testFetchAllPages("text/plain", "SELECT field FROM " + indexPattern("test") + " ORDER BY field", texts, pageSize, false);
    }
    public void testFetchAllPagesCompositeAggCursorTxt() throws IOException {
        testFetchAllPagesCompositeAggCursor("text/plain");
    }
    public void testFetchAllPagesCompositeAggCursorCsv() throws IOException {
        testFetchAllPagesCompositeAggCursor("text/csv");
    }
    public void testFetchAllPagesCompositeAggCursorTsv() throws IOException {
        testFetchAllPagesCompositeAggCursor("text/tab-separated-values");
    }
    public void testFetchAllPagesCompositeAggCursor(String format) throws IOException {
        int size = randomIntBetween(4, 20);
        int pageSize = randomIntBetween(1, size + 1);
        List<String> texts = IntStream.range(0, size).mapToObj(i -> String.format(Locale.ROOT, "text%02d", i)).toList();
        index(texts.stream().map(t -> "{\"field\": \"" + t + "\"}").toArray(String[]::new));
        testFetchAllPages(
            format,
            "SELECT field FROM " + indexPattern("test") + " GROUP BY field ORDER BY field",
            texts,
            pageSize,
            size % pageSize == 0
        );
    }
    public void testFetchAllPagesCompositeAggCursorWithFilterOnAggregateTxt() throws IOException {
        testFetchAllPagesCompositeAggCursorWithFilterOnAggregate("text/plain");
    }
    public void testFetchAllPagesCompositeAggCursorWithFilterOnAggregateCsv() throws IOException {