minor

src/source/src/connector_source.rs

@@ -18,7 +18,7 @@ use std::sync::Arc;
 
 use futures::future::try_join_all;
 use futures::stream::pending;
-use futures::{pin_mut, FutureExt, Stream, StreamExt};
+use futures::StreamExt;
 use futures_async_stream::try_stream;
 use itertools::Itertools;
 use risingwave_common::catalog::ColumnId;
@@ -31,7 +31,7 @@ use risingwave_connector::source::filesystem::{FsPage, FsPageItem, S3SplitEnumer
 use risingwave_connector::source::{
     create_split_reader, BoxSourceWithStateStream, BoxTryStream, Column, ConnectorProperties,
     ConnectorState, FsFilterCtrlCtx, FsListInner, SourceColumnDesc, SourceContext,
-    SourceEnumeratorContext, SplitEnumerator, SplitReader, StreamChunkWithState,
+    SourceEnumeratorContext, SplitEnumerator, SplitReader,
 };
 use tokio::time;
 use tokio::time::{Duration, MissedTickBehavior};
@@ -177,9 +177,7 @@ impl ConnectorSource {
                 .await?
             };
 
-            let mut stream = select_all(readers.into_iter().map(|r| r.into_stream()))
-                .boxed()
-                .peekable();
+            let mut stream = select_all(readers.into_iter().map(|r| r.into_stream())).peekable();
             // peek the stream to make sure the underlying connector has been inited
             let _ = Pin::new(&mut stream).peek().await;
             Ok(stream.boxed())

src/stream/src/executor/source/source_executor.rs

@@ -13,12 +13,11 @@
 // limitations under the License.
 
 use std::fmt::Formatter;
-use std::pin::Pin;
 use std::time::Duration;
 
 use anyhow::anyhow;
 use either::Either;
-use futures::{pin_mut, StreamExt};
+use futures::StreamExt;
 use futures_async_stream::try_stream;
 use risingwave_common::system_param::local_manager::SystemParamsReaderRef;
 use risingwave_connector::source::{