### 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
testFetchAllPagesCompositeAggCursorWithFilterOnAggregate("text/csv");
    }
    public void testFetchAllPagesCompositeAggCursorWithFilterOnAggregateTsv() throws IOException {
        testFetchAllPagesCompositeAggCursorWithFilterOnAggregate("text/tab-separated-values");
    }
    public void testFetchAllPagesCompositeAggCursorWithFilterOnAggregate(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, COUNT(*) c FROM " + indexPattern("test") + " GROUP BY field HAVING c = 1 ORDER BY field",
            texts,
            pageSize,
            true
        );
    }
    public void testFetchAllPagesListCursorTxt() throws IOException {
        testFetchAllPagesListCursor("text/plain");
    }
    public void testFetchAllPagesListCursorCsv() throws IOException {
        testFetchAllPagesListCursor("text/csv");
    }
    public void testFetchAllPagesListCursorTsv() throws IOException {
        testFetchAllPagesListCursor("text/tab-separated-values");
    }
    public void testFetchAllPagesListCursor(String format) throws IOException {
        int size = randomIntBetween(4, 20);
        int pageSize = randomIntBetween(1, size + 1);