fix: reset state when begin scan

.github/workflows/test_wrappers.yml

@@ -55,4 +55,4 @@ jobs:
       run: |
         cd wrappers && RUSTFLAGS="-D warnings" cargo clippy --all --tests --no-deps --features all_fdws,helloworld_fdw
 
-    - run: cd wrappers && cargo pgrx test --features all_fdws,pg15
+    - run: cd wrappers && cargo pgrx test --features "all_fdws pg15"

wrappers/.ci/docker-compose.yaml

@@ -28,8 +28,6 @@ services:
   clickhouse:
     image: clickhouse/clickhouse-server
     container_name: clickhouse-wrapped
-    environment:
-      CLICKHOUSE_DB: supa
     ports:
       - "9000:9000" # native interface
       - "8123:8123" # http interface

wrappers/src/fdw/clickhouse_fdw/tests.rs

@@ -9,19 +9,17 @@ mod tests {
     #[pg_test]
     fn clickhouse_smoketest() {
         Spi::connect(|mut c| {
-            let clickhouse_pool = ch::Pool::new("tcp://default:@localhost:9000/supa");
+            let clickhouse_pool = ch::Pool::new("tcp://default:@localhost:9000/default");
 
             let rt = create_async_runtime().expect("failed to create runtime");
             let mut handle = rt
                 .block_on(async { clickhouse_pool.get_handle().await })
                 .expect("handle");
 
             rt.block_on(async {
+                handle.execute("DROP TABLE IF EXISTS test_table").await?;
                 handle
-                    .execute("DROP TABLE IF EXISTS supa.test_table")
-                    .await?;
-                handle
-                    .execute("CREATE TABLE supa.test_table (id INT, name TEXT) engine = Memory")
+                    .execute("CREATE TABLE test_table (id INT, name TEXT) engine = Memory")
                     .await
             })
             .expect("test_table in ClickHouse");
@@ -37,7 +35,7 @@ mod tests {
                 r#"CREATE SERVER my_clickhouse_server
                          FOREIGN DATA WRAPPER clickhouse_wrapper
                          OPTIONS (
-                           conn_string 'tcp://default:@localhost:9000/supa'
+                           conn_string 'tcp://default:@localhost:9000/default'
                          )"#,
                 None,
                 None,
@@ -235,7 +233,7 @@ mod tests {
             let remote_value: String = rt
                 .block_on(async {
                     handle
-                        .query("SELECT name FROM supa.test_table ORDER BY name LIMIT 1")
+                        .query("SELECT name FROM test_table ORDER BY name LIMIT 1")
                         .fetch_all()
                         .await?
                         .rows()

wrappers/src/fdw/mssql_fdw/mssql_fdw.rs

@@ -192,6 +192,8 @@ impl ForeignDataWrapper<MssqlFdwError> for MssqlFdw {
         self.table = require_option("table", options)?.to_string();
         self.tgt_cols = columns.to_vec();
 
+        self.iter_idx = 0;
+
         // create sql server client
         let tcp = self
             .rt

wrappers/src/fdw/redis_fdw/redis_fdw.rs

@@ -74,6 +74,13 @@ impl RedisFdw {
     const FDW_NAME: &'static str = "RedisFdw";
     const BUF_SIZE: isize = 256;
 
+    fn reset(&mut self) {
+        self.iter_idx = 0;
+        self.scan_result.clear();
+        self.iter_idx_stream = "-".to_string();
+        self.scan_result_stream.clear();
+    }
+
     // fetch a target row for list and zset
     fn fetch_row_list(&mut self) -> RedisFdwResult<Option<Row>> {
         if let Some(ref mut conn) = &mut self.conn {
@@ -271,6 +278,8 @@ impl ForeignDataWrapper<RedisFdwError> for RedisFdw {
 
         let mut conn = self.client.get_connection()?;
 
+        self.reset();
+
         match src_type.as_str() {
             "list" | "zset" => {
                 check_target_columns(
@@ -366,10 +375,7 @@ impl ForeignDataWrapper<RedisFdwError> for RedisFdw {
     }
 
     fn re_scan(&mut self) -> RedisFdwResult<()> {
-        self.iter_idx = 0;
-        self.scan_result.clear();
-        self.iter_idx_stream = "-".to_string();
-        self.scan_result_stream.clear();
+        self.reset();
         Ok(())
     }
 

wrappers/src/fdw/stripe_fdw/stripe_fdw.rs

@@ -660,6 +660,8 @@ impl ForeignDataWrapper<StripeFdwError> for StripeFdw {
     ) -> StripeFdwResult<()> {
         let obj = require_option("object", options)?;
 
+        self.iter_idx = 0;
+
         if let Some(client) = &self.client {
             let page_size = 100; // maximum page size limit for Stripe API
             let page_cnt = if let Some(limit) = limit {