fix: bugs found by unit test

analytic_engine/src/row_iter/chain.rs

@@ -285,6 +285,7 @@ impl ChainIterator {
             self.streams[self.next_prefetch_stream_idx]
                 .start_prefetch()
                 .await;
+            self.next_prefetch_stream_idx += 1;
         }
     }
 }

analytic_engine/src/sst/parquet/async_reader.rs

@@ -585,7 +585,7 @@ impl PrefetchableStream for RecordBatchReceiver {
     }
 
     async fn fetch_next(&mut self) -> Option<Self::Item> {
-        self.fetch_next().await
+        self.next().await
     }
 }
 

common_util/src/prefetchable_stream.rs

@@ -148,3 +148,25 @@ impl<T> PrefetchableStream for NoopPrefetcher<T> {
         self.0.next().await
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use futures::stream;
+
+    use super::*;
+
+    #[tokio::test]
+    async fn test_trait_object_prefetchable_stream() {
+        let numbers = vec![1, 2, 3];
+        let stream = stream::iter(numbers.clone());
+        let stream = NoopPrefetcher(Box::new(stream));
+        let mut stream: Box<dyn PrefetchableStream<Item = i32>> = Box::new(stream);
+
+        let mut fetched_numbers = Vec::with_capacity(numbers.len());
+        while let Some(v) = stream.fetch_next().await {
+            fetched_numbers.push(v);
+        }
+
+        assert_eq!(numbers, fetched_numbers);
+    }
+}