Struct Pipeline

pub struct Pipeline<'a, B = Owned, const SYNC_MODE: bool = true> { /* private fields */ }

A pipelined sql query type. It lazily batch queries into local buffer and try to send it with the least amount of syscall when pipeline starts.

Examples

use xitca_postgres::{iter::AsyncLendingIterator, pipeline::Pipeline, Client, Execute, ExecuteMut, Statement};

async fn pipeline(client: &Client) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    // prepare a statement that will be called repeatedly.
    // it can be a collection of statements that will be called in iteration.
    let statement = Statement::named("SELECT * FROM public.users", &[]).execute(client).await?;

    // create a new pipeline.
    let mut pipe = Pipeline::new();

    // bind value param to statement and query with the pipeline.
    // pipeline can encode multiple queries locally before send it to database.
    statement.bind([] as [i32; 0]).query_mut(&mut pipe)?;
    statement.bind([] as [i32; 0]).query_mut(&mut pipe)?;
    statement.bind([] as [i32; 0]).query_mut(&mut pipe)?;

    // query the pipeline and on success a streaming response will be returned.
    let mut res = pipe.query(client)?;

    // iterate through the query responses. the response order is the same as the order of
    // queries encoded into pipeline with Pipeline::query_xxx api.
    while let Some(mut item) = res.try_next().await? {
        // every query can contain streaming rows.
        while let Some(row) = item.try_next().await? {
            let _: u32 = row.get("id");
        }
    }

    Ok(())
}

Implementations

impl Pipeline<'_, Owned, true>

pub fn new() -> Self

start a new pipeline.

pipeline is sync by default. which means every query inside is considered separate binding and the pipeline is transparent to database server. the pipeline only happen on socket transport where minimal amount of syscall is needed.

for more relaxed Pipeline Mode see Pipeline::unsync api.

pub fn with_capacity(cap: usize) -> Self

start a new pipeline with given capacity. capacity represent how many queries will be contained by a single pipeline. a determined cap can possibly reduce memory reallocation when constructing the pipeline.

impl Pipeline<'_, Owned, false>

pub fn unsync() -> Self

start a new un-sync pipeline.

in un-sync mode pipeline treat all queries inside as one single binding and database server can see them as no sync point in between which can result in potential performance gain.

it behaves the same on transportation level as Pipeline::new where minimal amount of socket syscall is needed.

pub fn unsync_with_capacity(cap: usize) -> Self

start a new un-sync pipeline with given capacity. capacity represent how many queries will be contained by a single pipeline. a determined cap can possibly reduce memory reallocation when constructing the pipeline.

impl<'a> Pipeline<'_, Borrowed<'a>, true>

pub fn from_buf(buf: &'a mut BytesMut) -> Self

start a new borrowed pipeline. pipeline will use borrowed bytes buffer to store encode messages before sending it to database.

pipeline is sync by default. which means every query inside is considered separate binding and the pipeline is transparent to database server. the pipeline only happen on socket transport where minimal amount of syscall is needed.

for more relaxed Pipeline Mode see Pipeline::unsync_from_buf api.

pub fn with_capacity_from_buf(cap: usize, buf: &'a mut BytesMut) -> Self

start a new borrowed pipeline with given capacity. capacity represent how many queries will be contained by a single pipeline. a determined cap can possibly reduce memory reallocation when constructing the pipeline.

impl<'a> Pipeline<'_, Borrowed<'a>, false>

pub fn unsync_from_buf(buf: &'a mut BytesMut) -> Self

start a new borrowed un-sync pipeline.

in un-sync mode pipeline treat all queries inside as one single binding and database server can see them as no sync point in between which can result in potential performance gain.

it behaves the same on transportation level as Pipeline::from_buf where minimal amount of socket syscall is needed.

pub fn unsync_with_capacity_from_buf(cap: usize, buf: &'a mut BytesMut) -> Self

start a new borrowed un-sync pipeline with given capacity. capacity represent how many queries will be contained by a single pipeline. a determined cap can possibly reduce memory reallocation when constructing the pipeline.

Trait Implementations

impl<'p, C, B, const SYNC_MODE: bool> Execute<'_, C> for Pipeline<'p, B, SYNC_MODE>

where C: Query, B: DerefMut<Target = BytesMut>,

type ExecuteOutput = Pin<Box<dyn Future<Output = Result<u64, Error>> + Send + 'p>>

outcome of execute. used for single time database response: number of rows affected by execution for example.

type QueryOutput = Result<PipelineStream<'p>, Error>

outcome of query. used for repeated database response: database rows for example

fn execute(self, cli: &C) -> Self::ExecuteOutput

define how a statement is executed with single time response

fn query(self, cli: &C) -> Self::QueryOutput

define how a statement is queried with repeated response

impl<'p, C, B, const SYNC_MODE: bool> ExecuteBlocking<'_, C> for Pipeline<'p, B, SYNC_MODE>

where C: Query, B: DerefMut<Target = BytesMut>,

type ExecuteOutput = Result<u64, Error>

type QueryOutput = Result<PipelineStream<'p>, Error>

fn execute_blocking(self, cli: &C) -> Result<u64, Error>

fn query_blocking(self, cli: &C) -> Self::QueryOutput