refactor: avoid double flattening stream

Changes the into_record_batches() method to avoid creating an extra
stream out of the Option that must be flattened (iterating over the
option vs. filtering out all None first).

ingester2/src/query/response.rs

@@ -2,7 +2,7 @@
 //!
 //! [`QueryExec::query_exec()`]: super::QueryExec::query_exec()
 
-use std::pin::Pin;
+use std::{future, pin::Pin};
 
 use arrow::{error::ArrowError, record_batch::RecordBatch};
 use futures::{Stream, StreamExt};
@@ -51,7 +51,7 @@ impl QueryResponse {
     /// Reduce the [`QueryResponse`] to a stream of [`RecordBatch`].
     pub(crate) fn into_record_batches(self) -> impl Stream<Item = Result<RecordBatch, ArrowError>> {
         self.into_partition_stream()
-            .flat_map(|partition| futures::stream::iter(partition.into_record_batch_stream()))
+            .filter_map(|partition| future::ready(partition.into_record_batch_stream()))
             .flatten()
     }
 }