feat(clickhouse): [WRA-9] add configurable stream_buffer_size option to ClickHouse FDW

docs/catalog/clickhouse.md

@@ -97,11 +97,14 @@ The following options are available when creating ClickHouse foreign tables:
 
 - `table` - Source table name in ClickHouse, required
 
-This can also be a subquery enclosed in parentheses, for example,
+    This can also be a subquery enclosed in parentheses, for example,
 
-```sql
-table '(select * from my_table)'
-```
+    ```sql
+    table '(select * from my_table)'
+    ```
+
+- `rowid_column` - Primary key column name, optional for data scan, required for data modify
+- `stream_buffer_size` - Size of the internal buffer used for streaming data from ClickHouse, defaults to 1024 rows. Must be between 1 and 100000.
 
 ### Parametrized views
 
@@ -123,8 +126,6 @@ table '(select * from my_table)'
   select * from test_vw where _param1='aaa' and _param2=32;
 ```
 
-- `rowid_column` - Primary key column name, optional for data scan, required for data modify
-
 ## Entities
 
 ### Tables

wrappers/src/fdw/clickhouse_fdw/README.md

@@ -11,6 +11,7 @@ This is a foreign data wrapper for [ClickHouse](https://clickhouse.com/). It is
 
 | Version | Date       | Notes                                                |
 | ------- | ---------- | ---------------------------------------------------- |
+| 0.1.9   | 2025-11-08 | Added stream_buffer_size foreign table option        |
 | 0.1.8   | 2025-10-27 | Refactor to read rows with async streaming           |
 | 0.1.7   | 2025-05-22 | Added more data types support                        |
 | 0.1.6   | 2025-05-06 | Added UUID data type support                         |

wrappers/src/fdw/clickhouse_fdw/clickhouse_fdw.rs

@@ -353,7 +353,7 @@ where
 }
 
 #[wrappers_fdw(
-    version = "0.1.8",
+    version = "0.1.9",
     author = "Supabase",
     website = "https://github.com/supabase/wrappers/tree/main/wrappers/src/fdw/clickhouse_fdw",
     error_type = "ClickHouseFdwError"
@@ -584,8 +584,19 @@ impl ForeignDataWrapper<ClickHouseFdwError> for ClickHouseFdw {
         self.is_scan_complete = false;
         self.current_row_data = None;
 
+        // get stream buffer size from options, with validation
+        let stream_buffer_size = options
+            .get("stream_buffer_size")
+            .map(|s| {
+                s.parse::<usize>()
+                    .map(|size| size.clamp(1, 100_000))
+                    .unwrap_or(1024)
+            })
+            .unwrap_or(1024);
+
         // create bounded channel
-        let (tx, rx) = channel::bounded::<ClickHouseFdwResult<Option<ConvertedRow>>>(1024);
+        let (tx, rx) =
+            channel::bounded::<ClickHouseFdwResult<Option<ConvertedRow>>>(stream_buffer_size);
         self.row_receiver = Some(rx);
 
         // clone data needed by the async task

wrappers/src/fdw/clickhouse_fdw/tests.rs

@@ -85,7 +85,8 @@ mod tests {
                   SERVER my_clickhouse_server
                   OPTIONS (
                     table 'test_table',
-                    rowid_column 'id'
+                    rowid_column 'id',
+                    stream_buffer_size '512'
                   )
              "#,
                 None,