remove prints

src/connector/src/source/filesystem/nd_streaming.rs

@@ -64,14 +64,12 @@ pub async fn split_stream(data_stream: BoxSourceStream) {
             let mut line = String::new();
             match cursor.read_line(&mut line).await {
                 Ok(0) => {
-                    println!("WKXLOG Read line empty, {:?}", msgs);
                     if !msgs.is_empty() {
                         yield msgs;
                     }
                     break;
                 }
                 Ok(_n) => {
-                    println!("WKXLOG Read line: {:?}", line);
                     if line_cnt == 0 && last_message.is_some() {
                         let msg: SourceMessage = std::mem::take(&mut last_message).unwrap();
                         let last_payload = msg.payload.unwrap();
@@ -118,8 +116,6 @@ pub async fn split_stream(data_stream: BoxSourceStream) {
     }
 
     if let Some(msg) = last_message {
-        println!("WKXLOG last_message msg: {:?}", msg);
-
         yield vec![msg];
     }
 }

src/connector/src/source/filesystem/opendal_source/opendal_reader.rs

@@ -196,4 +196,4 @@ impl<Src: OpendalSource> OpendalReader<Src> {
             yield batch;
         }
     }
-}
+}

src/stream/src/executor/source/fetch_executor.rs

@@ -321,8 +321,6 @@ impl<S: StateStore, Src: OpendalSource> FsFetchExecutor<S, Src> {
                         }
                         // StreamChunk from FsSourceReader, and the reader reads only one file.
                         Either::Right(chunk) => {
-                            println!("WKXLOG fetch_executor into_stream chunk: {:?}", chunk);
-                            // println!("WKXLOG fetch_executor mapping: {:?}", mapping);
                             let mapping =
                                 get_split_offset_mapping_from_chunk(&chunk, split_idx, offset_idx)
                                     .unwrap();