RFC: Paginator Design

[rcoh]

Summary

Smithy models paginated responses. Customers of Smithy generated code & the Rust SDK will have an improved user experience if code is generated to support this.

In this RFC, we propose modeling paginated data as a Stream of output shapes.

• When an output is paginated, a paginate() method will be added to the high level builder
• A <OperationName>Paginator struct will be generated into the paginator module.

The Stream trait enables customers to use a number of abstractions including simple looping, and collect()ing all data in a single call.
A paginator will resend the original input, but with the field marked inputToken to the value of outputToken in the previous output.

Usage example:

let mut tables = vec![];
let mut pages = client
   .list_tables()
   .paginate()
   .send();
while let Some(next_page) = pages.try_next().await? {
    tables.extend(next_page.table_names.unwrap_or_default().into_iter());
}

Paginators are lazy and only retrieve pages when polled by a client.

Details

A draft implementation (not code generated) is here: #344.

Paginators will be generated into the paginator module of service crates. paginators will be an optional but enabled-by-default Cargo feature for AWS services. A paginator struct captures 4 pieces of data:

// dynamodb/src/paginator.rs
struct ListTablesPaginator {
  // holds the low-level client and configuration
  handle: Arc<Handle>

  // generated if the `page_size` member of the paginated trait is set
  page_size: Option<usize>,
  
  input: ListTablesInput
}

In addition to the basic usage example above, when pageSize is modeled, customers can specify the page size during pagination:

let mut tables = vec![];
let mut pages = client
   .list_tables()
   .paginate()
   .page_size(20)
   .send();
while let Some(next_page) = pages.try_next().await? {
    // pages of 20 items requested from DynamoDb
    tables.extend(next_page.table_names.unwrap_or_default().into_iter());
}

Note: Although paginators may specify items, the member where the raw paginated data is stored, that is not currently used by this proposal. A possible enhancement may be to have .paginated().items() return a stream of the items directly.

Paginators define a single public method send(). This method returns impl Stream<Item=Result<OperationOutput, OperationError>. The current plan-of-record is to use AsyncStream internally. AsyncStream essentially implements a rendezvous channel to enable yielding items 1-by-1 in a demand driven way:

// Hand written code, but a model of what would be code generated
impl ListTablesPaginator {
    pub fn send(
        self,
    ) -> impl tokio_stream::Stream<Item = Result<ListTablesOutput, SdkError<ListTablesError>>> + Unpin
    {
        let (mut yield_tx, yield_rx) = async_stream::yielder::pair();
        // Without this, customers will get a 500 line error message about pinning. I think
        // that a single memory allocation is probably worth the trade off given that this will
        // send multiple HTTP requests.
        Box::pin(AsyncStream::new(yield_rx, async move {
            let mut inp = self.inp;
            loop {
                let rsp = self
                    .client
                    .client()
                    .call(inp.make_operation(self.client.conf()).unwrap())
                    .await;
                match rsp {
                    Ok(output) if output.last_evaluated_table_name.is_some() => {
                        // mutate the input to set the next token for the subsequent loop
                        inp.exclusive_start_table_name = output.last_evaluated_table_name.clone();
                        inp.limit = self.page_size.clone();
                        yield_tx.send(Ok(output)).await;
                    }
                    other => {
                        yield_tx.send(other).await;
                        break;
                    }
                }
            }
        }))
    }
}

On Box::pin: The stream returned by AsyncStream does not implement Unpin. Unfortunately, this makes iteration require an invocation of pin_mut! and generates several hundred lines of compiler errors. Box::pin seems a worthwhile trade off to improve the user experience.

On the + Unpin bound: Because auto-traits leak across impl Trait boundaries, + Unpin prevents accidental regressions in the generated code which would break users.

Updates to ergonomic clients

The builders generated by ergonomic clients will gain the following method, iff they represent an operation targeted by the paginated trait:

#[cfg(feature = "paginate")]
pub fn paginate(self) -> Result<ListTablesPaginator, operation::BuildError> {
  let (client, input_builder) = self.into_inner();
  ListTablesPaginator::new(client, input_builder.build()?)
}

Wrinkles

In order to be usable from outside the ergonomic clients, paginators need to be able accept client and conf directly. Currently, Handle is defined within the client module which is feature gated. This would need to be extracted so that it could be used within the paginators. Some considerations also need to be made about ownership. We don't necessarily want to force low-level customers to Arc their clients but this can be an area of ongoing research.

[pkazmier]

Perhaps a dumb question as I don't know Rust, but in the examples above, there is always an outer loop iterating over pages. Will there be a way to simply iterate (lazily) over the results (vs pages) without a double loop? I'm just accustomed to Boto3 resource objects abstracting away pagination entirely for me:

ec2 = session.resource("ec2", "us-east-1")
for vpc in ec2.vpcs.all():
  # Do something with each VPC result

[rcoh]

it's possible in Rust, and we'll consider this option as well when we implement this