fix: list one page in fs source executor (#20292)

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

@@ -18,7 +18,7 @@ use anyhow::anyhow;
 use async_trait::async_trait;
 use chrono::{DateTime, Utc};
 use futures::stream::{self, BoxStream};
-use futures::StreamExt;
+use futures::{StreamExt, TryStreamExt};
 use opendal::{Metakey, Operator};
 use risingwave_common::types::Timestamptz;
 
@@ -54,7 +54,9 @@ impl<Src: OpendalSource> SplitEnumerator for OpendalEnumerator<Src> {
         let empty_split: OpendalFsSplit<Src> = OpendalFsSplit::empty_split();
         let prefix = self.prefix.as_deref().unwrap_or("/");
 
-        match self.op.list(prefix).await {
+        let mut lister = self.op.lister(prefix).await?;
+        // fetch one item as validation, no need to get all
+        match lister.try_next().await {
             Ok(_) => return Ok(vec![empty_split]),
             Err(e) => {
                 return Err(anyhow!(e)

src/connector/src/source/filesystem/s3/enumerator.rs

@@ -99,15 +99,14 @@ impl SplitEnumerator for S3SplitEnumerator {
     }
 
     async fn list_splits(&mut self) -> crate::error::ConnectorResult<Vec<Self::Split>> {
-        let mut objects = Vec::new();
-        loop {
-            let (files, has_finished) = self.get_next_page::<FsSplit>().await?;
-            objects.extend(files);
-            if has_finished {
-                break;
-            }
-        }
-        Ok(objects)
+        // fetch one page as validation, no need to get all pages
+        let (_, _) = self.get_next_page::<FsSplit>().await?;
+
+        Ok(vec![FsSplit {
+            name: "empty_split".to_owned(),
+            offset: 0,
+            size: 0,
+        }])
     }
 }