fix(s3): parquet file read positioning and text column type conversion (#485)

* fix(s3): parquet file reading positioning and text column type conversion

* fix(s3): parquet file reading positioning and text column type conversion

Author: burmecia

wrappers/dockerfiles/s3/test_data/test_data_titanic.parquet

@@ -10,6 +10,7 @@ This is a foreign data wrapper for [AWS S3](https://aws.amazon.com/s3/). It is d
 
 | Version | Date       | Notes                                                |
 | ------- | ---------- | ---------------------------------------------------- |
+| 0.1.5   | 2025-07-25 | Fixed parquet file reading position issue            |
 | 0.1.4   | 2024-08-20 | Added `path_style_url` server option                 |
 | 0.1.2   | 2023-07-13 | Added fdw stats collection                           |
 | 0.1.1   | 2023-06-05 | Added Parquet file support                           |

wrappers/src/fdw/s3_fdw/README.md

@@ -88,9 +88,10 @@ impl AsyncRead for S3ParquetReader {
         };
 
         // calculate request range
-        let object_remaining = object_size - self.pos;
+        let object_remaining = object_size.saturating_sub(self.pos);
         let remaining = min(buf.remaining() as u64, object_remaining);
-        let range = format!("bytes={}-{}", self.pos, self.pos + remaining - 1);
+        let upper_bound = self.pos.saturating_add(remaining).saturating_sub(1);
+        let range = format!("bytes={}-{}", self.pos, upper_bound);
 
         // wait on current thread to get object contents
         match futures::executor::block_on(
@@ -102,8 +103,20 @@ impl AsyncRead for S3ParquetReader {
                 .send(),
         ) {
             Ok(output) => {
+                let pre_filled = buf.filled().len();
                 let mut rdr = output.body.into_async_read();
-                AsyncRead::poll_read(Pin::new(&mut rdr), cx, buf)
+                match AsyncRead::poll_read(Pin::new(&mut rdr), cx, buf) {
+                    Poll::Ready(result) => match result {
+                        Ok(_) => {
+                            // calculate consumed length of bytes and add it to current position
+                            let consumed = buf.filled().len().saturating_sub(pre_filled);
+                            self.pos = self.pos.saturating_add(consumed as u64);
+                            Poll::Ready(Ok(()))
+                        }
+                        Err(err) => Poll::Ready(Err(err)),
+                    },
+                    Poll::Pending => Poll::Pending,
+                }
             }
             Err(err) => Poll::Ready(Err(to_io_error(err))),
         }
@@ -296,15 +309,23 @@ impl S3Parquet {
                         }
                     }
                     pg_sys::TEXTOID => {
-                        let arr = col
-                            .as_any()
-                            .downcast_ref::<array::BinaryArray>()
-                            .ok_or(S3FdwError::ColumnTypeNotMatch(tgt_col.name.clone()))?;
-                        if arr.is_null(self.batch_idx) {
+                        // 'text' type can be converted from StringArray, StringViewArray and
+                        // generic BinaryArray
+                        if col.is_null(self.batch_idx) {
                             None
-                        } else {
+                        } else if let Some(arr) = col.as_any().downcast_ref::<array::StringArray>()
+                        {
+                            Some(Cell::String(arr.value(self.batch_idx).to_string()))
+                        } else if let Some(arr) =
+                            col.as_any().downcast_ref::<array::StringViewArray>()
+                        {
+                            Some(Cell::String(arr.value(self.batch_idx).to_string()))
+                        } else if let Some(arr) = col.as_any().downcast_ref::<array::BinaryArray>()
+                        {
                             let s = String::from_utf8_lossy(arr.value(self.batch_idx));
                             Some(Cell::String(s.to_string()))
+                        } else {
+                            return Err(S3FdwError::ColumnTypeNotMatch(tgt_col.name.clone()));
                         }
                     }
                     pg_sys::DATEOID => {

wrappers/src/fdw/s3_fdw/parquet.rs

@@ -25,7 +25,7 @@ enum Parser {
 }
 
 #[wrappers_fdw(
-    version = "0.1.4",
+    version = "0.1.5",
     author = "Supabase",
     website = "https://github.com/supabase/wrappers/tree/main/wrappers/src/fdw/s3_fdw",
     error_type = "S3FdwError"

wrappers/src/fdw/s3_fdw/s3_fdw.rs

@@ -155,6 +155,29 @@ mod tests {
             )
             .unwrap();
 
+            c.update(
+                r#"
+                CREATE FOREIGN TABLE s3_test_table_parquet_titanic (
+                    "PassengerId" bigint,
+                    "Survived" bigint,
+                    "Pclass" bigint,
+                    "Name" text,
+                    "Age" double precision,
+                    "SibSp" bigint,
+                    "Parch" bigint,
+                    "Fare" double precision
+                )
+                SERVER s3_server
+                OPTIONS (
+                    uri 's3://warehouse/test_data_titanic.parquet',
+                    format 'parquet'
+                  )
+             "#,
+                None,
+                &[],
+            )
+            .unwrap();
+
             let check_test_table = |table| {
                 let sql = format!("SELECT * FROM {} ORDER BY name LIMIT 1", table);
                 let results = c
@@ -191,6 +214,18 @@ mod tests {
 
             check_parquet_table("s3_test_table_parquet");
             check_parquet_table("s3_test_table_parquet_gz");
+
+            let sql = "SELECT * FROM s3_test_table_parquet_titanic ORDER BY 1 LIMIT 1";
+            let results = c
+                .select(sql, None, &[])
+                .unwrap()
+                .filter_map(|r| {
+                    r.get_by_name::<i64, _>("PassengerId")
+                        .unwrap()
+                        .zip(r.get_by_name::<&str, _>("Name").unwrap())
+                })
+                .collect::<Vec<_>>();
+            assert_eq!(results, vec![(1, "Braund, Mr. Owen Harris")]);
         });
     }
 }