supabase_wrappers/
interface.rs

1//! Provides interface types and trait to develop Postgres foreign data wrapper
2//!
3
4use crate::instance::ForeignServer;
5use crate::FdwRoutine;
6use pgrx::pg_sys::panic::ErrorReport;
7use pgrx::prelude::{Date, Interval, Time, Timestamp, TimestampWithTimeZone};
8use pgrx::{
9    datum::Uuid,
10    fcinfo,
11    pg_sys::{self, bytea, BuiltinOid, Datum, Oid},
12    AllocatedByRust, AnyNumeric, FromDatum, IntoDatum, JsonB, PgBuiltInOids, PgOid,
13};
14use std::collections::HashMap;
15use std::ffi::CStr;
16use std::fmt;
17use std::iter::Zip;
18use std::mem;
19use std::slice::Iter;
20
21// fdw system catalog oids
22// https://doxygen.postgresql.org/pg__foreign__data__wrapper_8h.html
23// https://doxygen.postgresql.org/pg__foreign__server_8h.html
24// https://doxygen.postgresql.org/pg__foreign__table_8h.html
25
26/// Constant can be used in [validator](ForeignDataWrapper::validator)
27pub const FOREIGN_DATA_WRAPPER_RELATION_ID: Oid = BuiltinOid::ForeignDataWrapperRelationId.value();
28
29/// Constant can be used in [validator](ForeignDataWrapper::validator)
30pub const FOREIGN_SERVER_RELATION_ID: Oid = BuiltinOid::ForeignServerRelationId.value();
31
32/// Constant can be used in [validator](ForeignDataWrapper::validator)
33pub const FOREIGN_TABLE_RELATION_ID: Oid = BuiltinOid::ForeignTableRelationId.value();
34
35/// A data cell in a data row
36#[derive(Debug)]
37pub enum Cell {
38    Bool(bool),
39    I8(i8),
40    I16(i16),
41    F32(f32),
42    I32(i32),
43    F64(f64),
44    I64(i64),
45    Numeric(AnyNumeric),
46    String(String),
47    Date(Date),
48    Time(Time),
49    Timestamp(Timestamp),
50    Timestamptz(TimestampWithTimeZone),
51    Interval(Interval),
52    Json(JsonB),
53    Bytea(*mut bytea),
54    Uuid(Uuid),
55    BoolArray(Vec<Option<bool>>),
56    I16Array(Vec<Option<i16>>),
57    I32Array(Vec<Option<i32>>),
58    I64Array(Vec<Option<i64>>),
59    F32Array(Vec<Option<f32>>),
60    F64Array(Vec<Option<f64>>),
61    StringArray(Vec<Option<String>>),
62}
63
64impl Cell {
65    /// Check if cell is an array type
66    pub fn is_array(&self) -> bool {
67        matches!(
68            self,
69            Cell::BoolArray(_)
70                | Cell::I16Array(_)
71                | Cell::I32Array(_)
72                | Cell::I64Array(_)
73                | Cell::F32Array(_)
74                | Cell::F64Array(_)
75                | Cell::StringArray(_)
76        )
77    }
78}
79
80unsafe impl Send for Cell {}
81
82impl Clone for Cell {
83    fn clone(&self) -> Self {
84        match self {
85            Cell::Bool(v) => Cell::Bool(*v),
86            Cell::I8(v) => Cell::I8(*v),
87            Cell::I16(v) => Cell::I16(*v),
88            Cell::F32(v) => Cell::F32(*v),
89            Cell::I32(v) => Cell::I32(*v),
90            Cell::F64(v) => Cell::F64(*v),
91            Cell::I64(v) => Cell::I64(*v),
92            Cell::Numeric(v) => Cell::Numeric(v.clone()),
93            Cell::String(v) => Cell::String(v.clone()),
94            Cell::Date(v) => Cell::Date(*v),
95            Cell::Time(v) => Cell::Time(*v),
96            Cell::Timestamp(v) => Cell::Timestamp(*v),
97            Cell::Timestamptz(v) => Cell::Timestamptz(*v),
98            Cell::Interval(v) => Cell::Interval(*v),
99            Cell::Json(v) => Cell::Json(JsonB(v.0.clone())),
100            Cell::Bytea(v) => Cell::Bytea(*v),
101            Cell::Uuid(v) => Cell::Uuid(*v),
102            Cell::BoolArray(v) => Cell::BoolArray(v.clone()),
103            Cell::I16Array(v) => Cell::I16Array(v.clone()),
104            Cell::I32Array(v) => Cell::I32Array(v.clone()),
105            Cell::I64Array(v) => Cell::I64Array(v.clone()),
106            Cell::F32Array(v) => Cell::F32Array(v.clone()),
107            Cell::F64Array(v) => Cell::F64Array(v.clone()),
108            Cell::StringArray(v) => Cell::StringArray(v.clone()),
109        }
110    }
111}
112
113fn write_array<T: std::fmt::Display>(
114    array: &[Option<T>],
115    f: &mut fmt::Formatter<'_>,
116) -> fmt::Result {
117    let res = array
118        .iter()
119        .map(|e| match e {
120            Some(val) => format!("{}", val),
121            None => "null".to_owned(),
122        })
123        .collect::<Vec<String>>()
124        .join(",");
125    write!(f, "[{}]", res)
126}
127
128impl fmt::Display for Cell {
129    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
130        match self {
131            Cell::Bool(v) => write!(f, "{}", v),
132            Cell::I8(v) => write!(f, "{}", v),
133            Cell::I16(v) => write!(f, "{}", v),
134            Cell::F32(v) => write!(f, "{}", v),
135            Cell::I32(v) => write!(f, "{}", v),
136            Cell::F64(v) => write!(f, "{}", v),
137            Cell::I64(v) => write!(f, "{}", v),
138            Cell::Numeric(v) => write!(f, "{}", v),
139            Cell::String(v) => write!(f, "'{}'", v),
140            Cell::Date(v) => unsafe {
141                let dt =
142                    fcinfo::direct_function_call_as_datum(pg_sys::date_out, &[(*v).into_datum()])
143                        .expect("cell should be a valid date");
144                let dt_cstr = CStr::from_ptr(dt.cast_mut_ptr());
145                write!(
146                    f,
147                    "'{}'",
148                    dt_cstr.to_str().expect("date should be a valid string")
149                )
150            },
151            Cell::Time(v) => unsafe {
152                let ts =
153                    fcinfo::direct_function_call_as_datum(pg_sys::time_out, &[(*v).into_datum()])
154                        .expect("cell should be a valid time");
155                let ts_cstr = CStr::from_ptr(ts.cast_mut_ptr());
156                write!(
157                    f,
158                    "'{}'",
159                    ts_cstr.to_str().expect("time hould be a valid string")
160                )
161            },
162            Cell::Timestamp(v) => unsafe {
163                let ts = fcinfo::direct_function_call_as_datum(
164                    pg_sys::timestamp_out,
165                    &[(*v).into_datum()],
166                )
167                .expect("cell should be a valid timestamp");
168                let ts_cstr = CStr::from_ptr(ts.cast_mut_ptr());
169                write!(
170                    f,
171                    "'{}'",
172                    ts_cstr
173                        .to_str()
174                        .expect("timestamp should be a valid string")
175                )
176            },
177            Cell::Timestamptz(v) => unsafe {
178                let ts = fcinfo::direct_function_call_as_datum(
179                    pg_sys::timestamptz_out,
180                    &[(*v).into_datum()],
181                )
182                .expect("cell should be a valid timestamptz");
183                let ts_cstr = CStr::from_ptr(ts.cast_mut_ptr());
184                write!(
185                    f,
186                    "'{}'",
187                    ts_cstr
188                        .to_str()
189                        .expect("timestamptz should be a valid string")
190                )
191            },
192            Cell::Interval(v) => write!(f, "{}", v),
193            Cell::Json(v) => write!(f, "{:?}", v),
194            Cell::Bytea(v) => {
195                let byte_u8 = unsafe { pgrx::varlena::varlena_to_byte_slice(*v) };
196                let hex = byte_u8
197                    .iter()
198                    .map(|b| format!("{:02X}", b))
199                    .collect::<Vec<String>>()
200                    .join("");
201                if hex.is_empty() {
202                    write!(f, "''")
203                } else {
204                    write!(f, r#"'\x{}'"#, hex)
205                }
206            }
207            Cell::Uuid(v) => write!(f, "'{}'", v),
208            Cell::BoolArray(v) => write_array(v, f),
209            Cell::I16Array(v) => write_array(v, f),
210            Cell::I32Array(v) => write_array(v, f),
211            Cell::I64Array(v) => write_array(v, f),
212            Cell::F32Array(v) => write_array(v, f),
213            Cell::F64Array(v) => write_array(v, f),
214            Cell::StringArray(v) => write_array(v, f),
215        }
216    }
217}
218
219impl IntoDatum for Cell {
220    fn into_datum(self) -> Option<Datum> {
221        match self {
222            Cell::Bool(v) => v.into_datum(),
223            Cell::I8(v) => v.into_datum(),
224            Cell::I16(v) => v.into_datum(),
225            Cell::F32(v) => v.into_datum(),
226            Cell::I32(v) => v.into_datum(),
227            Cell::F64(v) => v.into_datum(),
228            Cell::I64(v) => v.into_datum(),
229            Cell::Numeric(v) => v.into_datum(),
230            Cell::String(v) => v.into_datum(),
231            Cell::Date(v) => v.into_datum(),
232            Cell::Time(v) => v.into_datum(),
233            Cell::Timestamp(v) => v.into_datum(),
234            Cell::Timestamptz(v) => v.into_datum(),
235            Cell::Interval(v) => v.into_datum(),
236            Cell::Json(v) => v.into_datum(),
237            Cell::Bytea(v) => Some(Datum::from(v)),
238            Cell::Uuid(v) => v.into_datum(),
239            Cell::BoolArray(v) => v.into_datum(),
240            Cell::I16Array(v) => v.into_datum(),
241            Cell::I32Array(v) => v.into_datum(),
242            Cell::I64Array(v) => v.into_datum(),
243            Cell::F32Array(v) => v.into_datum(),
244            Cell::F64Array(v) => v.into_datum(),
245            Cell::StringArray(v) => v.into_datum(),
246        }
247    }
248
249    fn type_oid() -> Oid {
250        Oid::INVALID
251    }
252
253    fn is_compatible_with(other: Oid) -> bool {
254        Self::type_oid() == other
255            || other == pg_sys::BOOLOID
256            || other == pg_sys::CHAROID
257            || other == pg_sys::INT2OID
258            || other == pg_sys::FLOAT4OID
259            || other == pg_sys::INT4OID
260            || other == pg_sys::FLOAT8OID
261            || other == pg_sys::INT8OID
262            || other == pg_sys::NUMERICOID
263            || other == pg_sys::TEXTOID
264            || other == pg_sys::DATEOID
265            || other == pg_sys::TIMEOID
266            || other == pg_sys::TIMESTAMPOID
267            || other == pg_sys::TIMESTAMPTZOID
268            || other == pg_sys::INTERVALOID
269            || other == pg_sys::JSONBOID
270            || other == pg_sys::BYTEAOID
271            || other == pg_sys::UUIDOID
272            || other == pg_sys::BOOLARRAYOID
273            || other == pg_sys::INT2ARRAYOID
274            || other == pg_sys::INT4ARRAYOID
275            || other == pg_sys::INT8ARRAYOID
276            || other == pg_sys::FLOAT4ARRAYOID
277            || other == pg_sys::FLOAT8ARRAYOID
278            || other == pg_sys::TEXTARRAYOID
279    }
280}
281
282impl FromDatum for Cell {
283    unsafe fn from_polymorphic_datum(datum: Datum, is_null: bool, typoid: Oid) -> Option<Self>
284    where
285        Self: Sized,
286    {
287        let oid = PgOid::from(typoid);
288        match oid {
289            PgOid::BuiltIn(PgBuiltInOids::BOOLOID) => {
290                bool::from_datum(datum, is_null).map(Cell::Bool)
291            }
292            PgOid::BuiltIn(PgBuiltInOids::CHAROID) => i8::from_datum(datum, is_null).map(Cell::I8),
293            PgOid::BuiltIn(PgBuiltInOids::INT2OID) => {
294                i16::from_datum(datum, is_null).map(Cell::I16)
295            }
296            PgOid::BuiltIn(PgBuiltInOids::FLOAT4OID) => {
297                f32::from_datum(datum, is_null).map(Cell::F32)
298            }
299            PgOid::BuiltIn(PgBuiltInOids::INT4OID) => {
300                i32::from_datum(datum, is_null).map(Cell::I32)
301            }
302            PgOid::BuiltIn(PgBuiltInOids::FLOAT8OID) => {
303                f64::from_datum(datum, is_null).map(Cell::F64)
304            }
305            PgOid::BuiltIn(PgBuiltInOids::INT8OID) => {
306                i64::from_datum(datum, is_null).map(Cell::I64)
307            }
308            PgOid::BuiltIn(PgBuiltInOids::NUMERICOID) => {
309                AnyNumeric::from_datum(datum, is_null).map(Cell::Numeric)
310            }
311            PgOid::BuiltIn(PgBuiltInOids::TEXTOID) => {
312                String::from_datum(datum, is_null).map(Cell::String)
313            }
314            PgOid::BuiltIn(PgBuiltInOids::DATEOID) => {
315                Date::from_datum(datum, is_null).map(Cell::Date)
316            }
317            PgOid::BuiltIn(PgBuiltInOids::TIMEOID) => {
318                Time::from_datum(datum, is_null).map(Cell::Time)
319            }
320            PgOid::BuiltIn(PgBuiltInOids::TIMESTAMPOID) => {
321                Timestamp::from_datum(datum, is_null).map(Cell::Timestamp)
322            }
323            PgOid::BuiltIn(PgBuiltInOids::TIMESTAMPTZOID) => {
324                TimestampWithTimeZone::from_datum(datum, is_null).map(Cell::Timestamptz)
325            }
326            PgOid::BuiltIn(PgBuiltInOids::INTERVALOID) => {
327                Interval::from_datum(datum, is_null).map(Cell::Interval)
328            }
329            PgOid::BuiltIn(PgBuiltInOids::JSONBOID) => {
330                JsonB::from_datum(datum, is_null).map(Cell::Json)
331            }
332            PgOid::BuiltIn(PgBuiltInOids::BYTEAOID) => {
333                Some(Cell::Bytea(datum.cast_mut_ptr::<bytea>()))
334            }
335            PgOid::BuiltIn(PgBuiltInOids::UUIDOID) => {
336                Uuid::from_datum(datum, is_null).map(Cell::Uuid)
337            }
338            PgOid::BuiltIn(PgBuiltInOids::BOOLARRAYOID) => {
339                Vec::<Option<bool>>::from_datum(datum, false).map(Cell::BoolArray)
340            }
341            PgOid::BuiltIn(PgBuiltInOids::INT2ARRAYOID) => {
342                Vec::<Option<i16>>::from_datum(datum, false).map(Cell::I16Array)
343            }
344            PgOid::BuiltIn(PgBuiltInOids::INT4ARRAYOID) => {
345                Vec::<Option<i32>>::from_datum(datum, false).map(Cell::I32Array)
346            }
347            PgOid::BuiltIn(PgBuiltInOids::INT8ARRAYOID) => {
348                Vec::<Option<i64>>::from_datum(datum, false).map(Cell::I64Array)
349            }
350            PgOid::BuiltIn(PgBuiltInOids::FLOAT4ARRAYOID) => {
351                Vec::<Option<f32>>::from_datum(datum, false).map(Cell::F32Array)
352            }
353            PgOid::BuiltIn(PgBuiltInOids::FLOAT8ARRAYOID) => {
354                Vec::<Option<f64>>::from_datum(datum, false).map(Cell::F64Array)
355            }
356            PgOid::BuiltIn(PgBuiltInOids::TEXTARRAYOID) => {
357                Vec::<Option<String>>::from_datum(datum, false).map(Cell::StringArray)
358            }
359            _ => None,
360        }
361    }
362}
363
364pub trait CellFormatter {
365    fn fmt_cell(&mut self, cell: &Cell) -> String;
366}
367
368struct DefaultFormatter {}
369
370impl DefaultFormatter {
371    fn new() -> Self {
372        Self {}
373    }
374}
375
376impl CellFormatter for DefaultFormatter {
377    fn fmt_cell(&mut self, cell: &Cell) -> String {
378        format!("{}", cell)
379    }
380}
381
382/// A data row in a table
383///
384/// The row contains a column name list and cell list with same number of
385/// elements.
386#[derive(Debug, Clone, Default)]
387pub struct Row {
388    /// column names
389    pub cols: Vec<String>,
390
391    /// column cell list, should match with cols
392    pub cells: Vec<Option<Cell>>,
393}
394
395impl Row {
396    /// Create an empty row
397    pub fn new() -> Self {
398        Self::default()
399    }
400
401    /// Push a cell with column name to this row
402    pub fn push(&mut self, col: &str, cell: Option<Cell>) {
403        self.cols.push(col.to_owned());
404        self.cells.push(cell);
405    }
406
407    /// Return a zipped <column_name, cell> iterator
408    pub fn iter(&self) -> Zip<Iter<'_, String>, Iter<'_, Option<Cell>>> {
409        self.cols.iter().zip(self.cells.iter())
410    }
411
412    /// Remove a cell at the specified index
413    pub fn retain<F>(&mut self, f: F)
414    where
415        F: FnMut((&String, &Option<Cell>)) -> bool,
416    {
417        let keep: Vec<bool> = self.iter().map(f).collect();
418        let mut iter = keep.iter();
419        self.cols.retain(|_| *iter.next().unwrap_or(&true));
420        iter = keep.iter();
421        self.cells.retain(|_| *iter.next().unwrap_or(&true));
422    }
423
424    /// Replace `self` with the source row
425    #[inline]
426    pub fn replace_with(&mut self, src: Row) {
427        let _ = mem::replace(self, src);
428    }
429
430    /// Clear the row, removing all column names and cells
431    pub fn clear(&mut self) {
432        self.cols.clear();
433        self.cells.clear();
434    }
435}
436
437/// A column definition in a table
438///
439/// The column represents a column definition in a table.
440#[derive(Debug, Clone, Default)]
441pub struct Column {
442    /// column name
443    pub name: String,
444
445    /// 1-based column number
446    pub num: usize,
447
448    /// column type OID, can be used to match pg_sys::BuiltinOid
449    pub type_oid: Oid,
450}
451
452/// A restiction value used in [`Qual`], either a [`Cell`] or an array of [`Cell`]
453#[derive(Debug, Clone)]
454pub enum Value {
455    Cell(Cell),
456    Array(Vec<Cell>),
457}
458
459/// Query parameter
460#[derive(Debug, Clone)]
461pub struct Param {
462    /// 1-based parameter id
463    pub id: usize,
464
465    /// parameter type OID
466    pub type_oid: Oid,
467}
468
469/// Query restrictions, a.k.a conditions in `WHERE` clause
470///
471/// A Qual defines a simple condition wich can be used by the FDW to restrict the number
472/// of the results.
473///
474/// <div class="example-wrap" style="display:inline-block"><pre class="compile_fail" style="white-space:normal;font:inherit;">
475/// <strong>Warning</strong>: Currently only simple conditions are supported, see below for examples. Other kinds of conditions, like JSON attribute filter e.g. `where json_col->>'key' = 'value'`, are not supported yet.
476/// </pre></div>
477///
478/// ## Examples
479///
480/// ```sql
481/// where id = 1;
482/// -- [Qual { field: "id", operator: "=", value: Cell(I32(1)), use_or: false }]
483/// ```
484///
485/// ```sql
486/// where id in (1, 2);
487/// -- [Qual { field: "id", operator: "=", value: Array([I64(1), I64(2)]), use_or: true }]
488/// ```
489///
490/// ```sql
491/// where col is null
492/// -- [Qual { field: "col", operator: "is", value: Cell(String("null")), use_or: false }]
493/// ```
494///
495/// ```sql
496/// where bool_col
497/// -- [Qual { field: "bool_col", operator: "=", value: Cell(Bool(true)), use_or: false }]
498/// ```
499///
500/// ```sql
501/// where bool_col is true
502/// -- [Qual { field: "bool_col", operator: "is", value: Cell(Bool(true)), use_or: false }]
503/// ```
504///
505/// ```sql
506/// where id > 1 and col = 'foo';
507/// -- [
508/// --   Qual { field: "id", operator: ">", value: Cell(I32(1)), use_or: false },
509/// --   Qual { field: "col", operator: "=", value: Cell(String("foo")), use_or: false }
510/// -- ]
511/// ```
512#[derive(Debug, Clone)]
513pub struct Qual {
514    pub field: String,
515    pub operator: String,
516    pub value: Value,
517    pub use_or: bool,
518    pub param: Option<Param>,
519}
520
521impl Qual {
522    pub fn deparse(&self) -> String {
523        let mut formatter = DefaultFormatter::new();
524        self.deparse_with_fmt(&mut formatter)
525    }
526
527    pub fn deparse_with_fmt<T: CellFormatter>(&self, t: &mut T) -> String {
528        if self.use_or {
529            match &self.value {
530                Value::Cell(_) => unreachable!(),
531                Value::Array(cells) => {
532                    let conds: Vec<String> = cells
533                        .iter()
534                        .map(|cell| {
535                            format!("{} {} {}", self.field, self.operator, t.fmt_cell(cell))
536                        })
537                        .collect();
538                    conds.join(" or ")
539                }
540            }
541        } else {
542            match &self.value {
543                Value::Cell(cell) => match self.operator.as_str() {
544                    "is" | "is not" => match cell {
545                        Cell::String(cell) if cell == "null" => {
546                            format!("{} {} null", self.field, self.operator)
547                        }
548                        _ => format!("{} {} {}", self.field, self.operator, t.fmt_cell(cell)),
549                    },
550                    "~~" => format!("{} like {}", self.field, t.fmt_cell(cell)),
551                    "!~~" => format!("{} not like {}", self.field, t.fmt_cell(cell)),
552                    _ => format!("{} {} {}", self.field, self.operator, t.fmt_cell(cell)),
553                },
554                Value::Array(_) => unreachable!(),
555            }
556        }
557    }
558}
559
560/// Query sort, a.k.a `ORDER BY` clause
561///
562/// ## Examples
563///
564/// ```sql
565/// order by id;
566/// -- [Sort { field: "id", field_no: 1, reversed: false, nulls_first: false, collate: None]
567/// ```
568///
569/// ```sql
570/// order by id desc;
571/// -- [Sort { field: "id", field_no: 1, reversed: true, nulls_first: true, collate: None]
572/// ```
573///
574/// ```sql
575/// order by id desc, col;
576/// -- [
577/// --   Sort { field: "id", field_no: 1, reversed: true, nulls_first: true, collate: None },
578/// --   Sort { field: "col", field_no: 2, reversed: false, nulls_first: false, collate: None }
579/// -- ]
580/// ```
581///
582/// ```sql
583/// order by id collate "de_DE";
584/// -- [Sort { field: "col", field_no: 2, reversed: false, nulls_first: false, collate: Some("de_DE") }]
585/// ```
586#[derive(Debug, Clone, Default)]
587pub struct Sort {
588    pub field: String,
589    pub field_no: usize,
590    pub reversed: bool,
591    pub nulls_first: bool,
592    pub collate: Option<String>,
593}
594
595impl Sort {
596    pub fn deparse(&self) -> String {
597        let mut sql = self.field.to_string();
598
599        if self.reversed {
600            sql.push_str(" desc");
601        } else {
602            sql.push_str(" asc");
603        }
604
605        if self.nulls_first {
606            sql.push_str(" nulls first")
607        } else {
608            sql.push_str(" nulls last")
609        }
610
611        sql
612    }
613
614    pub fn deparse_with_collate(&self) -> String {
615        let mut sql = self.deparse();
616
617        if let Some(collate) = &self.collate {
618            sql.push_str(&format!(" collate {}", collate));
619        }
620
621        sql
622    }
623}
624
625/// Query limit, a.k.a `LIMIT count OFFSET offset` clause
626///
627/// ## Examples
628///
629/// ```sql
630/// limit 42;
631/// -- Limit { count: 42, offset: 0 }
632/// ```
633///
634/// ```sql
635/// limit 42 offset 7;
636/// -- Limit { count: 42, offset: 7 }
637/// ```
638#[derive(Debug, Clone, Default)]
639pub struct Limit {
640    pub count: i64,
641    pub offset: i64,
642}
643
644impl Limit {
645    pub fn deparse(&self) -> String {
646        format!("limit {} offset {}", self.count, self.offset)
647    }
648}
649
650/// The Foreign Data Wrapper trait
651///
652/// This is the main interface for your foreign data wrapper. Required functions
653/// are listed below, all the others are optional.
654///
655/// 1. new
656/// 2. begin_scan
657/// 3. iter_scan
658/// 4. end_scan
659///
660/// See the module-level document for more details.
661///
662pub trait ForeignDataWrapper<E: Into<ErrorReport>> {
663    /// Create a FDW instance
664    ///
665    /// `options` is the key-value pairs defined in `CREATE SERVER` SQL. For example,
666    ///
667    /// ```sql
668    /// create server my_helloworld_server
669    ///   foreign data wrapper wrappers_helloworld
670    ///   options (
671    ///     foo 'bar'
672    /// );
673    /// ```
674    ///
675    /// `options` passed here will be a hashmap { 'foo' -> 'bar' }.
676    ///
677    /// You can do any initalization in this function, like saving connection
678    /// info or API url in an variable, but don't do heavy works like database
679    /// connection or API call.
680    fn new(server: ForeignServer) -> Result<Self, E>
681    where
682        Self: Sized;
683
684    /// Obtain relation size estimates for a foreign table
685    ///
686    /// Return the expected number of rows and row size (in bytes) by the
687    /// foreign table scan.
688    ///
689    /// [See more details](https://www.postgresql.org/docs/current/fdw-callbacks.html#FDW-CALLBACKS-SCAN).
690    fn get_rel_size(
691        &mut self,
692        _quals: &[Qual],
693        _columns: &[Column],
694        _sorts: &[Sort],
695        _limit: &Option<Limit>,
696        _options: &HashMap<String, String>,
697    ) -> Result<(i64, i32), E> {
698        Ok((0, 0))
699    }
700
701    /// Called when begin executing a foreign scan
702    ///
703    /// - `quals` - `WHERE` clause pushed down
704    /// - `columns` - target columns to be queried
705    /// - `sorts` - `ORDER BY` clause pushed down
706    /// - `limit` - `LIMIT` clause pushed down
707    /// - `options` - the options defined when `CREATE FOREIGN TABLE`
708    ///
709    /// [See more details](https://www.postgresql.org/docs/current/fdw-callbacks.html#FDW-CALLBACKS-SCAN).
710    fn begin_scan(
711        &mut self,
712        quals: &[Qual],
713        columns: &[Column],
714        sorts: &[Sort],
715        limit: &Option<Limit>,
716        options: &HashMap<String, String>,
717    ) -> Result<(), E>;
718
719    /// Called when fetch one row from the foreign source
720    ///
721    /// FDW must save fetched foreign data into the [`Row`], or return `None` if no more rows to read.
722    ///
723    /// [See more details](https://www.postgresql.org/docs/current/fdw-callbacks.html#FDW-CALLBACKS-SCAN).
724    fn iter_scan(&mut self, row: &mut Row) -> Result<Option<()>, E>;
725
726    /// Called when restart the scan from the beginning.
727    ///
728    /// [See more details](https://www.postgresql.org/docs/current/fdw-callbacks.html#FDW-CALLBACKS-SCAN).
729    fn re_scan(&mut self) -> Result<(), E> {
730        Ok(())
731    }
732
733    /// Called when end the scan
734    ///
735    /// [See more details](https://www.postgresql.org/docs/current/fdw-callbacks.html#FDW-CALLBACKS-SCAN).
736    fn end_scan(&mut self) -> Result<(), E>;
737
738    /// Called when begin executing a foreign table modification operation.
739    ///
740    /// - `options` - the options defined when `CREATE FOREIGN TABLE`
741    ///
742    /// The foreign table must include a `rowid_column` option which specify
743    /// the unique identification column of the foreign table to enable data
744    /// modification.
745    ///
746    /// For example,
747    ///
748    /// ```sql
749    /// create foreign table my_foreign_table (
750    ///   id bigint,
751    ///   name text
752    /// )
753    ///   server my_server
754    ///   options (
755    ///     rowid_column 'id'
756    ///   );
757    /// ```
758    ///
759    /// [See more details](https://www.postgresql.org/docs/current/fdw-callbacks.html#FDW-CALLBACKS-UPDATE).
760    fn begin_modify(&mut self, _options: &HashMap<String, String>) -> Result<(), E> {
761        Ok(())
762    }
763
764    /// Called when insert one row into the foreign table
765    ///
766    /// - row - the new row to be inserted
767    ///
768    /// [See more details](https://www.postgresql.org/docs/current/fdw-callbacks.html#FDW-CALLBACKS-UPDATE).
769    fn insert(&mut self, _row: &Row) -> Result<(), E> {
770        Ok(())
771    }
772
773    /// Called when update one row into the foreign table
774    ///
775    /// - rowid - the `rowid_column` cell
776    /// - new_row - the new row with updated cells
777    ///
778    /// [See more details](https://www.postgresql.org/docs/current/fdw-callbacks.html#FDW-CALLBACKS-UPDATE).
779    fn update(&mut self, _rowid: &Cell, _new_row: &Row) -> Result<(), E> {
780        Ok(())
781    }
782
783    /// Called when delete one row into the foreign table
784    ///
785    /// - rowid - the `rowid_column` cell
786    ///
787    /// [See more details](https://www.postgresql.org/docs/current/fdw-callbacks.html#FDW-CALLBACKS-UPDATE).
788    fn delete(&mut self, _rowid: &Cell) -> Result<(), E> {
789        Ok(())
790    }
791
792    /// Called when end the table update
793    ///
794    /// [See more details](https://www.postgresql.org/docs/current/fdw-callbacks.html#FDW-CALLBACKS-UPDATE).
795    fn end_modify(&mut self) -> Result<(), E> {
796        Ok(())
797    }
798
799    /// Obtain a list of foreign table creation commands
800    ///
801    /// Return a list of string, each of which must contain a CREATE FOREIGN TABLE
802    /// which will be executed by the core server.
803    ///
804    /// [See more details](https://www.postgresql.org/docs/current/fdw-callbacks.html#FDW-CALLBACKS-IMPORT).
805    fn import_foreign_schema(
806        &mut self,
807        _stmt: crate::import_foreign_schema::ImportForeignSchemaStmt,
808    ) -> Result<Vec<String>, E> {
809        Ok(Vec::new())
810    }
811
812    /// Returns a FdwRoutine for the FDW
813    ///
814    /// Not to be used directly, use [`wrappers_fdw`](crate::wrappers_fdw) macro instead.
815    fn fdw_routine() -> FdwRoutine
816    where
817        Self: Sized,
818    {
819        unsafe {
820            use crate::{import_foreign_schema, modify, scan};
821            let mut fdw_routine =
822                FdwRoutine::<AllocatedByRust>::alloc_node(pg_sys::NodeTag::T_FdwRoutine);
823
824            // import foreign schema
825            fdw_routine.ImportForeignSchema =
826                Some(import_foreign_schema::import_foreign_schema::<E, Self>);
827
828            // plan phase
829            fdw_routine.GetForeignRelSize = Some(scan::get_foreign_rel_size::<E, Self>);
830            fdw_routine.GetForeignPaths = Some(scan::get_foreign_paths::<E, Self>);
831            fdw_routine.GetForeignPlan = Some(scan::get_foreign_plan::<E, Self>);
832            fdw_routine.ExplainForeignScan = Some(scan::explain_foreign_scan::<E, Self>);
833
834            // scan phase
835            fdw_routine.BeginForeignScan = Some(scan::begin_foreign_scan::<E, Self>);
836            fdw_routine.IterateForeignScan = Some(scan::iterate_foreign_scan::<E, Self>);
837            fdw_routine.ReScanForeignScan = Some(scan::re_scan_foreign_scan::<E, Self>);
838            fdw_routine.EndForeignScan = Some(scan::end_foreign_scan::<E, Self>);
839
840            // modify phase
841            fdw_routine.AddForeignUpdateTargets = Some(modify::add_foreign_update_targets);
842            fdw_routine.PlanForeignModify = Some(modify::plan_foreign_modify::<E, Self>);
843            fdw_routine.BeginForeignModify = Some(modify::begin_foreign_modify::<E, Self>);
844            fdw_routine.ExecForeignInsert = Some(modify::exec_foreign_insert::<E, Self>);
845            fdw_routine.ExecForeignDelete = Some(modify::exec_foreign_delete::<E, Self>);
846            fdw_routine.ExecForeignUpdate = Some(modify::exec_foreign_update::<E, Self>);
847            fdw_routine.EndForeignModify = Some(modify::end_foreign_modify::<E, Self>);
848
849            Self::fdw_routine_hook(&mut fdw_routine);
850            fdw_routine.into_pg_boxed()
851        }
852    }
853
854    /// Additional FwdRoutine setup, called by default `Self::fdw_routine()`
855    /// after completing its initialization.
856    fn fdw_routine_hook(_routine: &mut FdwRoutine<AllocatedByRust>) {}
857
858    /// Validator function for validating options given in `CREATE` and `ALTER`
859    /// commands for its foreign data wrapper, as well as foreign servers, user
860    /// mappings, and foreign tables using the wrapper.
861    ///
862    /// [See more details about validator](https://www.postgresql.org/docs/current/fdw-functions.html)
863    ///
864    /// # Example
865    ///
866    /// ```rust,no_run
867    /// use pgrx::pg_sys::Oid;
868    /// use supabase_wrappers::prelude::check_options_contain;
869    ///
870    /// use pgrx::pg_sys::panic::ErrorReport;
871    /// use pgrx::PgSqlErrorCode;
872    ///
873    /// enum FdwError {
874    ///     InvalidFdwOption,
875    ///     InvalidServerOption,
876    ///     InvalidTableOption,
877    /// }
878    ///
879    /// impl From<FdwError> for ErrorReport {
880    ///     fn from(value: FdwError) -> Self {
881    ///         let error_message = match value {
882    ///             FdwError::InvalidFdwOption => "invalid foreign data wrapper option",
883    ///             FdwError::InvalidServerOption => "invalid foreign server option",
884    ///             FdwError::InvalidTableOption => "invalid foreign table option",
885    ///         };
886    ///         ErrorReport::new(PgSqlErrorCode::ERRCODE_FDW_ERROR, error_message, "")
887    ///     }
888    /// }
889    ///
890    /// fn validator(opt_list: Vec<Option<String>>, catalog: Option<Oid>) -> Result<(), FdwError> {
891    ///     if let Some(oid) = catalog {
892    ///         match oid {
893    ///             FOREIGN_DATA_WRAPPER_RELATION_ID => {
894    ///                 // check a required option when create foreign data wrapper
895    ///                 check_options_contain(&opt_list, "foreign_data_wrapper_required_option");
896    ///             }
897    ///             FOREIGN_SERVER_RELATION_ID => {
898    ///                 // check option here when create server
899    ///                 check_options_contain(&opt_list, "foreign_server_required_option");
900    ///             }
901    ///             FOREIGN_TABLE_RELATION_ID => {
902    ///                 // check option here when create foreign table
903    ///                 check_options_contain(&opt_list, "foreign_table_required_option");
904    ///             }
905    ///             _ => {}
906    ///         }
907    ///     }
908    ///
909    ///     Ok(())
910    /// }
911    /// ```
912    fn validator(_options: Vec<Option<String>>, _catalog: Option<Oid>) -> Result<(), E> {
913        Ok(())
914    }
915}
supabase_wrappers/interface.rs

supabase_wrappers/
interface.rs
