pub async fn publication_info(
    conn: &str,
    publication: &str,
) -> Result<Vec<TableInfo>, anyhow::Error> {
    let config = conn.parse()?;
    let tls = make_tls(&config)?;
    let (client, connection) = config.connect(tls).await?;
    tokio::spawn(connection);

    let tables = client
        .query(
            "SELECT
                c.oid, p.schemaname, p.tablename
            FROM
                pg_catalog.pg_class AS c
                JOIN pg_namespace AS n ON c.relnamespace = n.oid
                JOIN pg_publication_tables AS p ON
                        c.relname = p.tablename AND n.nspname = p.schemaname
            WHERE
                p.pubname = $1",
            &[&publication],
        )
        .await?;

    let mut table_infos = vec![];
    for row in tables {
        let rel_id = row.get("oid");

        let schema = client
            .query(
                "SELECT
                        a.attname AS name,
                        a.atttypid AS oid,
                        a.atttypmod AS modifier,
                        a.attnotnull AS not_null,
                        b.oid IS NOT NULL AS primary_key
                    FROM pg_catalog.pg_attribute a
                    LEFT JOIN pg_catalog.pg_constraint b
                        ON a.attrelid = b.conrelid
                        AND b.contype = 'p'
                        AND a.attnum = ANY (b.conkey)
                    WHERE a.attnum > 0::pg_catalog.int2
                        AND NOT a.attisdropped
                        AND a.attrelid = $1
                    ORDER BY a.attnum",
                &[&rel_id],
            )
            .await?
            .into_iter()
            .map(|row| {
                let name: String = row.get("name");
                let oid = row.get("oid");
                let pg_type =
                    PgType::from_oid(oid).ok_or_else(|| anyhow!("unknown type OID: {}", oid))?;
                let scalar_type = match pg_type {
                    PgType::NUMERIC | PgType::NUMERIC_ARRAY => {
                        let modifier: i32 = row.get("modifier");
                        // https://github.com/postgres/postgres/blob/REL_13_3/src/backend/utils/adt/numeric.c#L967-L983
                        let typ_mod = if modifier < PG_HEADER_SIZE {
                            None
                        } else {
                            let tmp_mod = modifier - PG_HEADER_SIZE;
                            let scale = (tmp_mod & 0xffff) as u16;
                            let precision = ((tmp_mod >> 16) & 0xffff) as u16;
                            Some(PgNumericMod { scale, precision })
                        };

                        match pg_type {
                            PgType::NUMERIC => PgScalarType::Numeric(typ_mod),
                            PgType::NUMERIC_ARRAY => PgScalarType::NumericArray(typ_mod),
                            _ => unreachable!(),
                        }
                    }
                    PgType::BPCHAR | PgType::BPCHAR_ARRAY => {
                        let modifier: i32 = row.get("modifier");
                        // https://github.com/postgres/postgres/blob/REL_14_0/src/backend/utils/adt/varchar.c#L282-L286
                        let length = if modifier < PG_HEADER_SIZE {
                            adt::char::MAX_LENGTH
                        } else {
                            modifier - PG_HEADER_SIZE
                        };

                        match pg_type {
                            PgType::BPCHAR => PgScalarType::BPChar { length },
                            PgType::BPCHAR_ARRAY => PgScalarType::BPCharArray { length },
                            _ => unreachable!(),
                        }
                    }
                    PgType::VARCHAR | PgType::VARCHAR_ARRAY => {
                        let modifier: i32 = row.get("modifier");
                        // https://github.com/postgres/postgres/blob/REL_14_0/src/backend/utils/adt/varchar.c#L617
                        let length = if modifier < PG_HEADER_SIZE {
                            adt::varchar::MAX_LENGTH
                        } else {
                            modifier - PG_HEADER_SIZE
                        };

                        match pg_type {
                            PgType::VARCHAR => PgScalarType::VarChar { length },
                            PgType::VARCHAR_ARRAY => PgScalarType::VarCharArray { length },
                            _ => unreachable!(),
                        }
                    }
                    other => PgScalarType::Simple(other),
                };
                let not_null: bool = row.get("not_null");
                let primary_key = row.get("primary_key");
                Ok(PgColumn {
                    name: Ident::new(name),
                    scalar_type,
                    nullable: !not_null,
                    primary_key,
                })
            })
            .collect::<Result<Vec<_>, anyhow::Error>>()?;

        table_infos.push(TableInfo {
            rel_id,
            namespace: row.get("schemaname"),
            name: row.get("tablename"),
            schema,
        });
    }

    Ok(table_infos)
}
/// Fetches table schema information from an upstream Postgres source for all tables that are part
/// of a publication, given a connection string and the publication name.
///
/// # Errors
///
/// - Invalid connection string, user information, or user permissions.
/// - Upstream publication does not exist or contains invalid values.
Fetches table schema information from an upstream Postgres source for all tables that are part
of a publication, given a connection string and the publication name.
Fetches table schema information from an upstream Postgres source for all tables that are part of a publication given a connection string and the publication name .
pub async fn publication_info(
    conn: &str,
    publication: &str,
) -> Result<Vec<TableInfo>, anyhow::Error> {
    let config = conn.parse()?;
    let tls = make_tls(&config)?;
    let (client, connection) = config.connect(tls).await?;
    tokio::spawn(connection);
    let tables = client
        .query(
            "SELECT
                c.oid, p.schemaname, p.tablename
            FROM
                pg_catalog.pg_class AS c
                JOIN pg_namespace AS n ON c.relnamespace = n.oid
                JOIN pg_publication_tables AS p ON
                        c.relname = p.tablename AND n.nspname = p.schemaname
            WHERE
                p.pubname = $1",
            &[&publication],
        )
        .await?;
    let mut table_infos = vec![];
    for row in tables {
        let rel_id = row.get("oid");
        let schema = client
            .query(
                "SELECT
                        a.attname AS name,
                        a.atttypid AS oid,
                        a.atttypmod AS modifier,
                        a.attnotnull AS not_null,
                        b.oid IS NOT NULL AS primary_key
                    FROM pg_catalog.pg_attribute a
                    LEFT JOIN pg_catalog.pg_constraint b
                        ON a.attrelid = b.conrelid
                        AND b.contype = 'p'
                        AND a.attnum = ANY (b.conkey)
                    WHERE a.attnum > 0::pg_catalog.int2
                        AND NOT a.attisdropped
                        AND a.attrelid = $1
                    ORDER BY a.attnum",
                &[&rel_id],
            )
            .await?
            .into_iter()
            .map(|row| {
                let name: String = row.get("name");
                let oid = row.get("oid");
                let pg_type =
                    PgType::from_oid(oid).ok_or_else(|| anyhow!("unknown type OID: {}", oid))?;
                let scalar_type = match pg_type {
                    PgType::NUMERIC | PgType::NUMERIC_ARRAY => {
                        let modifier: i32 = row.get("modifier");
                        // https://github.com/postgres/postgres/blob/REL_13_3/src/backend/utils/adt/numeric.c#L967-L983
                        let typ_mod = if modifier < PG_HEADER_SIZE {
                            None
                        } else {
                            let tmp_mod = modifier - PG_HEADER_SIZE;
                            let scale = (tmp_mod & 0xffff) as u16;
                            let precision = ((tmp_mod >> 16) & 0xffff) as u16;
                            Some(PgNumericMod { scale, precision })
                        };
                        match pg_type {
                            PgType::NUMERIC => PgScalarType::Numeric(typ_mod),
                            PgType::NUMERIC_ARRAY => PgScalarType::NumericArray(typ_mod),
                            _ => unreachable!(),
                        }
                    }
                    PgType::BPCHAR | PgType::BPCHAR_ARRAY => {
                        let modifier: i32 = row.get("modifier");
                        // https://github.com/postgres/postgres/blob/REL_14_0/src/backend/utils/adt/varchar.c#L282-L286
                        let length = if modifier < PG_HEADER_SIZE {
                            adt::char::MAX_LENGTH
                        } else {
                            modifier - PG_HEADER_SIZE
                        };
                        match pg_type {
                            PgType::BPCHAR => PgScalarType::BPChar { length },
                            PgType::BPCHAR_ARRAY => PgScalarType::BPCharArray { length },
                            _ => unreachable!(),
                        }
                    }
                    PgType::VARCHAR | PgType::VARCHAR_ARRAY => {
                        let modifier: i32 = row.get("modifier");
                        // https://github.com/postgres/postgres/blob/REL_14_0/src/backend/utils/adt/varchar.c#L617
                        let length = if modifier < PG_HEADER_SIZE {
                            adt::varchar::MAX_LENGTH
                        } else {
                            modifier - PG_HEADER_SIZE
                        };
                        match pg_type {
                            PgType::VARCHAR => PgScalarType::VarChar { length },
                            PgType::VARCHAR_ARRAY => PgScalarType::VarCharArray { length },
                            _ => unreachable!(),
                        }
                    }
                    other => PgScalarType::Simple(other),
                };
                let not_null: bool = row.get("not_null");
                let primary_key = row.get("primary_key");
                Ok(PgColumn {
                    name: Ident::new(name),
                    scalar_type,
                    nullable: !not_null,
                    primary_key,
                })
            })
            .collect::<Result<Vec<_>, anyhow::Error>>()?;
        table_infos.push(TableInfo {
            rel_id,
            namespace: row.get("schemaname"),
            name: row.get("tablename"),
            schema,
        });
    }
    Ok(table_infos)
}
pub async fn publication_info ( conn : & str , publication : & str , ) -> Result < Vec < TableInfo > , anyhow :: Error > { let config = conn . parse ( ) ? ; let tls = make_tls ( & config ) ? ; let ( client , connection ) = config . connect ( tls ) . await ? ; tokio :: spawn ( connection ) ; let tables = client . query ( " " , & [ & publication ] , ) . await ? ; let mut table_infos = vec ! [ ] ; for row in tables { let rel_id = row . get ( " " ) ; let schema = client . query ( " " , & [ & rel_id ] , ) . await ? . into_iter ( ) . map ( | row | { let name : String = row . get ( " " ) ; let oid = row . get ( " " ) ; let pg_type = PgType :: from_oid ( oid ) . ok_or_else ( | | anyhow ! ( " " oid ) ) ? ; let scalar_type = match pg_type { PgType :: NUMERIC | PgType :: NUMERIC_ARRAY => { let modifier : i32 = row . get ( " " ) ; // https://github.com/postgres/postgres/blob/REL_13_3/src/backend/utils/adt/numeric.c#L967-L983 let typ_mod = if modifier < PG_HEADER_SIZE { None } else { let tmp_mod = modifier - PG_HEADER_SIZE ; let scale = ( tmp_mod & 0xffff ) as u16 ; let precision = ( ( tmp_mod >> 16 ) & 0xffff ) as u16 ; Some ( PgNumericMod { scale , precision } ) } ; match pg_type { PgType :: NUMERIC => PgScalarType :: Numeric ( typ_mod ) , PgType :: NUMERIC_ARRAY => PgScalarType :: NumericArray ( typ_mod ) , _ => unreachable ! ( ) , } } PgType :: BPCHAR | PgType :: BPCHAR_ARRAY => { let modifier : i32 = row . get ( " " ) ; // https://github.com/postgres/postgres/blob/REL_14_0/src/backend/utils/adt/varchar.c#L282-L286 let length = if modifier < PG_HEADER_SIZE { adt :: char :: MAX_LENGTH } else { modifier - PG_HEADER_SIZE } ; match pg_type { PgType :: BPCHAR => PgScalarType :: BPChar { length } , PgType :: BPCHAR_ARRAY => PgScalarType :: BPCharArray { length } , _ => unreachable ! ( ) , } } PgType :: VARCHAR | PgType :: VARCHAR_ARRAY => { let modifier : i32 = row . get ( " " ) ; // https://github.com/postgres/postgres/blob/REL_14_0/src/backend/utils/adt/varchar.c#L617 let length = if modifier < PG_HEADER_SIZE { adt :: varchar :: MAX_LENGTH } else { modifier - PG_HEADER_SIZE } ; match pg_type { PgType :: VARCHAR => PgScalarType :: VarChar { length } , PgType :: VARCHAR_ARRAY => PgScalarType :: VarCharArray { length } , _ => unreachable ! ( ) , } } other => PgScalarType :: Simple ( other ) , } ; let not_null : bool = row . get ( " " ) ; let primary_key = row . get ( " " ) ; Ok ( PgColumn { name : Ident :: new ( name ) , scalar_type , nullable : ! not_null , primary_key , } ) } ) . collect :: < Result < Vec < _ > , anyhow :: Error > > ( ) ? ; table_infos . push ( TableInfo { rel_id , namespace : row . get ( " " ) , name : row . get ( " " ) , schema , } ) ; } Ok ( table_infos ) }
Fetches table schema information from an upstream Postgres source for all tables that are part
of a publication, given a connection string and the publication name.
Fetches table schema information from an upstream Postgres source for all tables that are part of a publication given a connection string and the publication name .
conn
&str
publication
&str
conn
&str
publication
&str
errors
Invalid connection string, user information, or user permissions.
Upstream publication does not exist or contains invalid values.
Invalid connection string user information or user permissions . Upstream publication does not exist or contains invalid values .