### Description

Only "search hits" based queries (no aggregations) provide a consistent view on the data when paginating through the result set. With point-in-time, it is possible to support transactional semantics for aggregation queries as well.

This issue is somewhat blocked by https://github.com/elastic/elasticsearch/issues/75528 because a PIT can not immediately be closed if the aggregation result set contains less than `fetch_size` records. This could cause some increased resource utilization due to open PITs. Especially if the client does not close the SQL cursor.
x-pack/plugin/sql/src/test/java/org/elasticsearch/xpack/sql/analysis/CancellationTests.java
listener.onFailure(new TaskCancelledException("cancelled"));
            return null;
        }).when(client).execute(eq(SearchAction.INSTANCE), searchRequestCaptor.capture(), any());
        
        doAnswer(invocation -> {
            ClosePointInTimeRequest request = (ClosePointInTimeRequest) invocation.getArguments()[1];
            assertEquals(pitId, request.getId());
            @SuppressWarnings("unchecked")
            ActionListener<ClosePointInTimeResponse> listener = (ActionListener<ClosePointInTimeResponse>) invocation.getArguments()[2];
            listener.onResponse(new ClosePointInTimeResponse(true, 1));
            return null;
        }).when(client).execute(eq(ClosePointInTimeAction.INSTANCE), any(), any());
        IndexResolver indexResolver = indexResolver(client);
        PlanExecutor planExecutor = new PlanExecutor(client, indexResolver, new NamedWriteableRegistry(Collections.emptyList()));
        SqlQueryRequest request = new SqlQueryRequestBuilder(client, SqlQueryAction.INSTANCE).query("SELECT foo FROM " + indices[0])
            .request();
        CountDownLatch countDownLatch = new CountDownLatch(1);
        TransportSqlQueryAction.operation(planExecutor, task, request, new ActionListener<>() {
            @Override
            public void onResponse(SqlQueryResponse sqlSearchResponse) {
                fail("Shouldn't be here");
                countDownLatch.countDown();
            }
            @Override
            public void onFailure(Exception e) {
                assertThat(e, instanceOf(TaskCancelledException.class));
                countDownLatch.countDown();
            }
        }, "", mock(TransportService.class), mockClusterService);
        assertTrue(countDownLatch.await(5, TimeUnit.SECONDS));