fix: clone data instead of moving it - homemade future is dangerous

[waynexia]

I hereby agree to the terms of the GreptimeDB CLA.

Refer to a related PR or issue link (optional)

What's changed and what's your intention?

Fix the change in #3537. Since RecordBatch is all wrapped by Arc, cloning it is not that unacceptable.

Checklist

• I have written the necessary rustdoc comments.
• I have added the necessary unit tests and integration tests.
• This PR does not require documentation updates.

[codecov]

Codecov Report

All modified and coverable lines are covered by tests ✅

Project coverage is 84.99%. Comparing base (a99d6eb) to head (cf13f78).

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #3542      +/-   ##
==========================================
- Coverage   85.37%   84.99%   -0.38%     
==========================================
  Files         909      909              
  Lines      151812   151814       +2     
==========================================
- Hits       129611   129036     -575     
- Misses      22201    22778     +577     

[v0y4g3r]

LGTM, poll master

[tisonkun]

Possibly no clone solution -

diff --git a/src/promql/src/extension_plan/series_divide.rs b/src/promql/src/extension_plan/series_divide.rs
index 73fffbdeeb..4bdd3dda56 100644
--- a/src/promql/src/extension_plan/series_divide.rs
+++ b/src/promql/src/extension_plan/series_divide.rs
@@ -255,17 +255,19 @@ impl Stream for SeriesDivideStream {
 
     fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
         loop {
-            if let Some(batch) = self.buffer.take() {
+            if let Some(batch) = self.buffer.as_ref() {
                 let same_length = self.find_first_diff_row(&batch) + 1;
                 if same_length >= batch.num_rows() {
                     let next_batch = match ready!(self.as_mut().fetch_next_batch(cx)) {
                         Some(Ok(next_batch)) => next_batch,
                         None => {
                             self.num_series += 1;
+                            let batch = self.buffer.take().expect("must be some batch");
                             return Poll::Ready(Some(Ok(batch)));
                         }
                         error => return Poll::Ready(error),
                     };
+                    let batch = self.buffer.take().expect("must be some batch");
                     let new_batch = compute::concat_batches(&batch.schema(), &[batch, next_batch])?;
                     self.buffer = Some(new_batch);
                     continue;

... while a new "unchecked" unwrap while the condition should hold.