chore(iceberg): add extra foreign table options and refactor insert batch processing (#528)

* chore(iceberg): add create_table_if_not_exists foreign table option

* chore: refactor partition batch processing

* add partition rows sorter

* chore: add partition rows sorter

* Update wrappers/src/fdw/iceberg_fdw/iceberg_fdw.rs

Co-authored-by: Copilot <[email protected]>

* Update wrappers/src/fdw/iceberg_fdw/mapper.rs

Co-authored-by: Copilot <[email protected]>

* Update wrappers/src/fdw/iceberg_fdw/README.md

Co-authored-by: Copilot <[email protected]>

* Update wrappers/src/fdw/iceberg_fdw/sorter.rs

Co-authored-by: Copilot <[email protected]>

* modify as per Copilot suggestion

---------

Co-authored-by: Copilot <[email protected]>

Author: burmecia

docs/catalog/iceberg.md

@@ -162,7 +162,7 @@ For any server options need to be stored in Vault, you can add a prefix `vault_`
 
 #### Additional Server Options
 
-- `batch_size` - Controls the batch size of records read from Iceberg (default: 4096)
+- `batch_size` - Controls the batch size of records read from Iceberg (value range: 1 - 65536, default: 8192)
 
 ### Create a schema
 
@@ -178,6 +178,8 @@ The full list of foreign table options are below:
 
 - `table` - Fully qualified source table name with all namespaces in Iceberg, required.
 - `rowid_column` - The column to use as the row identifier for INSERT operations, required for data insertion.
+- `create_table_if_not_exists` - Boolean option (true/false) to automatically create the Iceberg table if it doesn't exist when inserting data, optional (default: false).
+- `partition_buffer_size` - Controls the buffer size for partitioned data during insertion operations, determining how many rows are batched together before being written to Iceberg (value range: 1 - 65536, default: 8192).
 
 ## Entities
 
@@ -332,12 +334,43 @@ values (123, 99.99, '2025-01-15');
 - **Partition Awareness**: When possible, insert data in partition order to optimize file organization
 - **Transaction Size**: Consider breaking very large inserts into smaller transactions
 
-### Limitations for Inserts
+### Automatic Table Creation
+
+When using the `create_table_if_not_exists` option, the Iceberg FDW will automatically create the target table in Iceberg if it doesn't exist when inserting data. This is useful for ad-hoc data insertion scenarios.
+
+```sql
+create foreign table iceberg.new_table (
+  id bigint,
+  name text,
+  created_at timestamp
+)
+  server iceberg_server
+  options (
+    table 'docs_example.new_table',
+    rowid_column 'id',
+    create_table_if_not_exists 'true'
+  );
+
+-- when data is inserted, if the 'docs_example.new_table' doesn't exist in Iceberg,
+-- it will be automatically created with a schema matching the foreign table definition
+insert into iceberg.new_table (id, name, created_at)
+values (1, 'New Record', now());
+```
+
+### Limitations for Insertion
 
 - Schema evolution during insert is not supported
 - Only append operations are supported (no upserts)
 - Complex data types (nested structs, arrays, maps) have limited support
 
+#### Automatic Table Creation Limitations
+
+When using the `create_table_if_not_exists` option, please be aware of the following additional limitations:
+
+- **Type Support**: Only primitive types are supported (such as boolean, integer, text, etc.). Complex types like arrays, structs, and maps are not supported for automatic table creation.
+- **Partitioning**: The automatically created table will use default partitioning settings. You cannot specify custom partition or sort specifications during automatic creation.
+- **Identifier Fields**: The automatically created table will not have any identifier fields specified. If you need identifier fields, you must create the Iceberg table manually beforehand.
+
 ## Limitations
 
 This section describes important limitations and considerations when using this FDW:

supabase-wrappers/src/instance.rs

@@ -3,10 +3,11 @@ use std::ffi::CStr;
 
 use crate::prelude::*;
 use pgrx::pg_sys::panic::ErrorReport;
-use pgrx::prelude::*;
+use pgrx::{pg_sys::Oid, prelude::*};
 
 #[derive(Debug, Clone, Default)]
 pub struct ForeignServer {
+    pub server_oid: Oid,
     pub server_name: String,
     pub server_type: Option<String>,
     pub server_version: Option<String>,
@@ -38,6 +39,7 @@ pub(super) unsafe fn create_fdw_instance_from_server_id<
     };
     let fserver = pg_sys::GetForeignServer(fserver_id);
     let server = ForeignServer {
+        server_oid: fserver_id,
         server_name: to_string((*fserver).servername).unwrap(),
         server_type: to_string((*fserver).servertype),
         server_version: to_string((*fserver).serverversion),

supabase-wrappers/src/modify.rs

@@ -232,9 +232,20 @@ pub(super) extern "C-unwind" fn plan_foreign_modify<
         // use NoLock here.
         let rel = PgRelation::with_lock((*rte).relid, pg_sys::NoLock as _);
 
-        // get rowid column name from table options
         let ftable = pg_sys::GetForeignTable(rel.oid());
-        let opts = options_to_hashmap((*ftable).options).report_unwrap();
+        let mut opts = options_to_hashmap((*ftable).options).report_unwrap();
+
+        // add additional metadata to the options
+        opts.insert(
+            "wrappers.fserver_oid".into(),
+            (*ftable).serverid.to_u32().to_string(),
+        );
+        opts.insert(
+            "wrappers.ftable_oid".into(),
+            (*ftable).relid.to_u32().to_string(),
+        );
+
+        // check if the rowid column name is specified in table options
         let rowid_name = opts.get("rowid_column");
         if rowid_name.is_none() {
             report_error(

supabase-wrappers/src/scan.rs

@@ -187,6 +187,16 @@ pub(super) extern "C-unwind" fn get_foreign_rel_size<
             // get foreign table options
             let ftable = pg_sys::GetForeignTable(foreigntableid);
             state.opts = options_to_hashmap((*ftable).options).report_unwrap();
+
+            // add additional metadata to the options
+            state.opts.insert(
+                "wrappers.fserver_oid".into(),
+                (*ftable).serverid.to_u32().to_string(),
+            );
+            state.opts.insert(
+                "wrappers.ftable_oid".into(),
+                (*ftable).relid.to_u32().to_string(),
+            );
         });
 
         // get estimate row count and mean row width

wrappers/dockerfiles/s3/iceberg_seed.py

@@ -253,6 +253,12 @@ def create_asks_table(catalog, namespace):
         identifier_field_ids=[1],
     )
 
+    partition_spec = PartitionSpec(
+        PartitionField(
+            source_id=1, field_id=1000, transform=DayTransform(), name="datetime_day"
+        ),
+    )
+
     sort_order = SortOrder(SortField(source_id=2, transform=IdentityTransform()))
 
     if catalog.table_exists(tblname):
@@ -261,6 +267,7 @@ def create_asks_table(catalog, namespace):
         identifier=tblname,
         schema=schema,
         #location="s3://iceberg",
+        partition_spec=partition_spec,
         sort_order=sort_order,
     )
     table = catalog.load_table(tblname)

wrappers/src/fdw/iceberg_fdw/README.md

@@ -8,10 +8,11 @@ This is a foreign data wrapper for [Apache Iceberg](https://iceberg.apache.org/)
 
 ## Changelog
 
-| Version | Date       | Notes                                                |
-| ------- | ---------- | ---------------------------------------------------- |
-| 0.1.3   | 2025-09-20 | Add data insertion support                           |
-| 0.1.2   | 2025-07-30 | Large data set query performance improvement         |
-| 0.1.1   | 2025-05-15 | Refactor server options passdown                     |
-| 0.1.0   | 2025-05-07 | Initial version                                      |
+| Version | Date       | Notes                                                                  |
+| ------- | ---------- | ---------------------------------------------------------------------- |
+| 0.1.4   | 2025-11-21 | Add create_table_if_not_exists option and improve insertion performance |
+| 0.1.3   | 2025-09-20 | Add data insertion support                                             |
+| 0.1.2   | 2025-07-30 | Large data set query performance improvement                           |
+| 0.1.1   | 2025-05-15 | Refactor server options passdown                                       |
+| 0.1.0   | 2025-05-07 | Initial version                                                        |