Trabalhar com resultados paginados: verificações e consultas

Os scan batch métodos query e do DynamoDB Enhanced API Client retornam respostas com uma ou mais páginas. Uma página contém um ou mais itens. Seu código pode processar a resposta por página ou pode processar itens individuais.

Uma resposta paginada retornada pelo DynamoDbEnhancedClient cliente síncrono retorna um PageIterableobjeto, enquanto uma resposta retornada pelo DynamoDbEnhancedAsyncClient assíncrono retorna um objeto. PagePublisher

Esta seção analisa o processamento de resultados paginados e fornece exemplos que usam a verificação e a consulta. APIs

Verificar uma tabela

O scanmétodo SDK s corresponde à operação do DynamoDB com o mesmo nome. O DynamoDB Enhanced API Client oferece as mesmas opções, mas usa um modelo de objeto familiar e gerencia a paginação para você.

Primeiro, exploramos a PageIterable interface examinando o scan método da classe de mapeamento síncrono, DynamoDbTable.

Use o síncrono API

O exemplo a seguir mostra o método scan que usa uma expressão para filtrar os itens retornados. ProductCatalogÉ o objeto do modelo que foi mostrado anteriormente.

A expressão de filtragem mostrada após a linha de comentário 2 limita os ProductCatalog itens que são retornados para aqueles com um valor de preço entre 8,00 e 80,00, inclusive.

Este exemplo também exclui os isbn valores usando o attributesToProject método mostrado após a linha de comentário 1.

Depois da linha de comentário 3, o PageIterable objetopagedResults,, é retornado pelo scan método. O método stream de PageIterable retorna um objeto java.util.Stream, que você pode usar para processar as páginas. Neste exemplo, o número de páginas é contado e registrado.

Começando com a linha de comentários 4, o exemplo mostra duas variações de acesso aos itens ProductCatalog. A versão após a linha de comentário 4a percorre cada página e classifica e registra os itens em cada página. A versão após a linha de comentário 4b ignora a iteração da página e acessa os itens diretamente.

A interface PageIterable oferece várias maneiras de processar resultados por causa de suas duas interfaces principais: java.lang.Iterable e SdkIterable. Iterable traz os métodos forEach, iterator e spliterator, e SdkIterable traz o método stream.

    public static void scanSync(DynamoDbTable<ProductCatalog> productCatalog) {

        Map<String, AttributeValue> expressionValues = Map.of(
                ":min_value", numberValue(8.00),
                ":max_value", numberValue(80.00));

        ScanEnhancedRequest request = ScanEnhancedRequest.builder()
                .consistentRead(true)
                // 1. the 'attributesToProject()' method allows you to specify which values you want returned.
                .attributesToProject("id", "title", "authors", "price")
                // 2. Filter expression limits the items returned that match the provided criteria.
                .filterExpression(Expression.builder()
                        .expression("price >= :min_value AND price <= :max_value")
                        .expressionValues(expressionValues)
                        .build())
                .build();

        // 3. A PageIterable object is returned by the scan method.
        PageIterable<ProductCatalog> pagedResults = productCatalog.scan(request);
        logger.info("page count: {}", pagedResults.stream().count());

        // 4. Log the returned ProductCatalog items using two variations.
        // 4a. This version sorts and logs the items of each page.
        pagedResults.stream().forEach(p -> p.items().stream()
                .sorted(Comparator.comparing(ProductCatalog::price))
                .forEach(
                        item -> logger.info(item.toString())
                ));
        // 4b. This version sorts and logs all items for all pages.
        pagedResults.items().stream()
                .sorted(Comparator.comparing(ProductCatalog::price))
                .forEach(
                        item -> logger.info(item.toString())
                );
    }

Use o assíncrono API

O método scan assíncrono retorna os resultados como um objeto PagePublisher. A interface PagePublisher tem dois métodos subscribe que você pode usar para processar páginas de resposta. Um método subscribe vem da interface principal org.reactivestreams.Publisher. Para processar páginas usando essa primeira opção, transmita uma instância Subscriber para o método subscribe. O primeiro exemplo a seguir mostra o uso do método subscribe.

O segundo subscribe método vem da SdkPublisherinterface. Esta versão de subscribe aceita um Consumer em vez de um Subscriber. Essa variação do método subscribe é mostrada no segundo exemplo.

O exemplo a seguir mostra a versão assíncrona do método scan que usa a mesma expressão de filtro mostrada no exemplo anterior.

Após a linha de comentário 3, DynamoDbAsyncTable.scan retorna um objeto PagePublisher. Na próxima linha, o código cria uma instância da interface org.reactivestreams.Subscriber, ProductCatalogSubscriber, que assina a PagePublisher após o comentário na linha 4.

O objeto Subscriber coleta os itens ProductCatalog de cada página no método onNext após a linha de comentário 8 no exemplo da classe ProductCatalogSubscriber. Os itens são armazenados na varável List privada e acessados no código de chamada com o método ProductCatalogSubscriber.getSubscribedItems(). A chamado é feita após a linha de comentários 5.

Depois que a lista é recuperada, o código classifica todos os itens ProductCatalog por preço e registra cada item.

O CountDownLatchin the ProductCatalogSubscriber class bloqueia o tópico de chamada até que todos os itens tenham sido adicionados à lista antes de continuar após a linha de comentários 5.

    public static void scanAsync(DynamoDbAsyncTable productCatalog) {
        ScanEnhancedRequest request = ScanEnhancedRequest.builder()
                .consistentRead(true)
                .attributesToProject("id", "title", "authors", "price")
                .filterExpression(Expression.builder()
                        // 1. :min_value and :max_value are placeholders for the values provided by the map
                        .expression("price >= :min_value AND price <= :max_value")
                        // 2. Two values are needed for the expression and each is supplied as a map entry.
                        .expressionValues(
                                Map.of( ":min_value", numberValue(8.00),
                                        ":max_value", numberValue(400_000.00)))
                        .build())
                .build();

        // 3. A PagePublisher object is returned by the scan method.
        PagePublisher<ProductCatalog> pagePublisher = productCatalog.scan(request);
        ProductCatalogSubscriber subscriber = new ProductCatalogSubscriber();
        // 4. Subscribe the ProductCatalogSubscriber to the PagePublisher.
        pagePublisher.subscribe(subscriber);
        // 5. Retrieve all collected ProductCatalog items accumulated by the subscriber.
        subscriber.getSubscribedItems().stream()
                .sorted(Comparator.comparing(ProductCatalog::price))
                .forEach(item ->
                        logger.info(item.toString()));
        // 6. Use a Consumer to work through each page.
        pagePublisher.subscribe(page -> page
                        .items().stream()
                        .sorted(Comparator.comparing(ProductCatalog::price))
                        .forEach(item ->
                                logger.info(item.toString())))
                .join(); // If needed, blocks the subscribe() method thread until it is finished processing.
        // 7. Use a Consumer to work through each ProductCatalog item.
        pagePublisher.items()
                .subscribe(product -> logger.info(product.toString()))
                .exceptionally(failure -> {
                    logger.error("ERROR  - ", failure);
                    return null;
                })
                .join(); // If needed, blocks the subscribe() method thread until it is finished processing.
    }

    private static class ProductCatalogSubscriber implements Subscriber<Page<ProductCatalog>> {
        private CountDownLatch latch = new CountDownLatch(1);
        private Subscription subscription;
        private List<ProductCatalog> itemsFromAllPages = new ArrayList<>();

        @Override
        public void onSubscribe(Subscription sub) {
            subscription = sub;
            subscription.request(1L);
            try {
                latch.await(); // Called by main thread blocking it until latch is released.
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }

        @Override
        public void onNext(Page<ProductCatalog> productCatalogPage) {
            // 8. Collect all the ProductCatalog instances in the page, then ask the publisher for one more page.
            itemsFromAllPages.addAll(productCatalogPage.items());
            subscription.request(1L);
        }

        @Override
        public void onError(Throwable throwable) {
        }

        @Override
        public void onComplete() {
            latch.countDown(); // Call by subscription thread; latch releases.
        }

        List<ProductCatalog> getSubscribedItems() {
            return this.itemsFromAllPages;
        }
    }

O exemplo de trecho a seguir usa a versão do método PagePublisher.subscribe que aceita um Consumer após a linha de comentário 6. O parâmetro lambda em Java consome páginas, que processam ainda mais cada item. Neste exemplo, cada página é processada e os itens em cada página são classificados e, em seguida, registrados.

        // 6. Use a Consumer to work through each page.
        pagePublisher.subscribe(page -> page
                        .items().stream()
                        .sorted(Comparator.comparing(ProductCatalog::price))
                        .forEach(item ->
                                logger.info(item.toString())))
                .join(); // If needed, blocks the subscribe() method thread until it is finished processing.

O método items do PagePublisher desempacota as instâncias do modelo para que seu código possa processar os itens diretamente. Essa abordagem é mostrada no trecho a seguir.

        // 7. Use a Consumer to work through each ProductCatalog item.
        pagePublisher.items()
                .subscribe(product -> logger.info(product.toString()))
                .exceptionally(failure -> {
                    logger.error("ERROR  - ", failure);
                    return null;
                })
                .join(); // If needed, blocks the subscribe() method thread until it is finished processing.

Consultar uma tabela

O método query() da classe DynamoDbTable encontra itens com base nos valores das chaves primárias. A anotação @DynamoDbPartitionKey e a anotação @DynamoDbSortKey opcional são usadas para definir a chave primária em sua classe de dados.

O método query() requer um valor de chave de partição que encontre itens que correspondam ao valor fornecido. Se sua tabela também definir uma chave de classificação, você poderá adicionar um valor para ela à sua consulta como uma condição de comparação adicional para ajustar os resultados.

Exceto pelo processamento dos resultados, as versões síncrona e assíncrona do query() funcionam do mesmo modo. Assim como no scanAPI, o query API retorna um PageIterable para uma chamada síncrona e um PagePublisher para uma chamada assíncrona. Discutimos o uso de PageIterable e PagePublisher anteriormente na seção de verificação.

Exemplos de métodos Query

O exemplo de código do método query() a seguir usa a classe MovieActor. A classe de dados define uma chave primária composta constituída pelo atributo movie para a chave de partição e pelo atributo actor para a chave de classificação.

A classe também sinaliza que usa um índice secundário global chamado acting_award_year. A chave primária composta do índice é constituída pelo atributo actingaward para a chave de partição e pelo actingyear para chave de classificação. Posteriormente neste tópico, quando mostrarmos como criar e usar índices, nos referiremos ao índice acting_award_year.

package org.example.tests.model;

import software.amazon.awssdk.enhanced.dynamodb.mapper.annotations.DynamoDbAttribute;
import software.amazon.awssdk.enhanced.dynamodb.mapper.annotations.DynamoDbBean;
import software.amazon.awssdk.enhanced.dynamodb.mapper.annotations.DynamoDbPartitionKey;
import software.amazon.awssdk.enhanced.dynamodb.mapper.annotations.DynamoDbSecondaryPartitionKey;
import software.amazon.awssdk.enhanced.dynamodb.mapper.annotations.DynamoDbSecondarySortKey;
import software.amazon.awssdk.enhanced.dynamodb.mapper.annotations.DynamoDbSortKey;

import java.util.Objects;

@DynamoDbBean
public class MovieActor implements Comparable<MovieActor> {

    private String movieName;
    private String actorName;
    private String actingAward;
    private Integer actingYear;
    private String actingSchoolName;

    @DynamoDbPartitionKey
    @DynamoDbAttribute("movie")
    public String getMovieName() {
        return movieName;
    }

    public void setMovieName(String movieName) {
        this.movieName = movieName;
    }

    @DynamoDbSortKey
    @DynamoDbAttribute("actor")
    public String getActorName() {
        return actorName;
    }

    public void setActorName(String actorName) {
        this.actorName = actorName;
    }

    @DynamoDbSecondaryPartitionKey(indexNames = "acting_award_year")
    @DynamoDbAttribute("actingaward")
    public String getActingAward() {
        return actingAward;
    }

    public void setActingAward(String actingAward) {
        this.actingAward = actingAward;
    }

    @DynamoDbSecondarySortKey(indexNames = {"acting_award_year", "movie_year"})
    @DynamoDbAttribute("actingyear")
    public Integer getActingYear() {
        return actingYear;
    }

    public void setActingYear(Integer actingYear) {
        this.actingYear = actingYear;
    }

    @DynamoDbAttribute("actingschoolname")
    public String getActingSchoolName() {
        return actingSchoolName;
    }

    public void setActingSchoolName(String actingSchoolName) {
        this.actingSchoolName = actingSchoolName;
    }

    @Override
    public String toString() {
        final StringBuffer sb = new StringBuffer("MovieActor{");
        sb.append("movieName='").append(movieName).append('\'');
        sb.append(", actorName='").append(actorName).append('\'');
        sb.append(", actingAward='").append(actingAward).append('\'');
        sb.append(", actingYear=").append(actingYear);
        sb.append(", actingSchoolName='").append(actingSchoolName).append('\'');
        sb.append('}');
        return sb.toString();
    }

    @Override
    public boolean equals(Object o) {
        if (this == o) return true;
        if (o == null || getClass() != o.getClass()) return false;
        MovieActor that = (MovieActor) o;
        return Objects.equals(movieName, that.movieName) && Objects.equals(actorName, that.actorName) && Objects.equals(actingAward, that.actingAward) && Objects.equals(actingYear, that.actingYear) && Objects.equals(actingSchoolName, that.actingSchoolName);
    }

    @Override
    public int hashCode() {
        return Objects.hash(movieName, actorName, actingAward, actingYear, actingSchoolName);
    }

    @Override
    public int compareTo(MovieActor o) {
        if (this.movieName.compareTo(o.movieName) != 0){
            return this.movieName.compareTo(o.movieName);
        } else {
            return this.actorName.compareTo(o.actorName);
        }
    }
}

Os exemplos de código a seguir são consultados com base nos itens a seguir.

MovieActor{movieName='movie01', actorName='actor0', actingAward='actingaward0', actingYear=2001, actingSchoolName='null'}
MovieActor{movieName='movie01', actorName='actor1', actingAward='actingaward1', actingYear=2001, actingSchoolName='actingschool1'}
MovieActor{movieName='movie01', actorName='actor2', actingAward='actingaward2', actingYear=2001, actingSchoolName='actingschool2'}
MovieActor{movieName='movie01', actorName='actor3', actingAward='actingaward3', actingYear=2001, actingSchoolName='null'}
MovieActor{movieName='movie01', actorName='actor4', actingAward='actingaward4', actingYear=2001, actingSchoolName='actingschool4'}
MovieActor{movieName='movie02', actorName='actor0', actingAward='actingaward0', actingYear=2002, actingSchoolName='null'}
MovieActor{movieName='movie02', actorName='actor1', actingAward='actingaward1', actingYear=2002, actingSchoolName='actingschool1'}
MovieActor{movieName='movie02', actorName='actor2', actingAward='actingaward2', actingYear=2002, actingSchoolName='actingschool2'}
MovieActor{movieName='movie02', actorName='actor3', actingAward='actingaward3', actingYear=2002, actingSchoolName='null'}
MovieActor{movieName='movie02', actorName='actor4', actingAward='actingaward4', actingYear=2002, actingSchoolName='actingschool4'}
MovieActor{movieName='movie03', actorName='actor0', actingAward='actingaward0', actingYear=2003, actingSchoolName='null'}
MovieActor{movieName='movie03', actorName='actor1', actingAward='actingaward1', actingYear=2003, actingSchoolName='actingschool1'}
MovieActor{movieName='movie03', actorName='actor2', actingAward='actingaward2', actingYear=2003, actingSchoolName='actingschool2'}
MovieActor{movieName='movie03', actorName='actor3', actingAward='actingaward3', actingYear=2003, actingSchoolName='null'}
MovieActor{movieName='movie03', actorName='actor4', actingAward='actingaward4', actingYear=2003, actingSchoolName='actingschool4'}

O código a seguir define duas QueryConditionalinstâncias. QueryConditionalstrabalham com valores de chave, seja a chave de partição isolada ou em combinação com a chave de classificação, e correspondem às principais expressões condicionais do serviço DynamoDB. API Depois da linha de comentário 1, o exemplo define a instância keyEqual que corresponde aos itens com um valor de partição de movie01.

Este exemplo também define uma expressão de filtro que filtra qualquer item que não tenha actingschoolname ativado após a linha de comentário 2.

Depois da linha de comentário 3, o exemplo mostra a QueryEnhancedRequestinstância que o código passa para o DynamoDbTable.query() método. Esse objeto combina a condição-chave e o filtro SDK usados para gerar a solicitação para o serviço do DynamoDB.

    public static void query(DynamoDbTable movieActorTable) {

        // 1. Define a QueryConditional instance to return items matching a partition value.
        QueryConditional keyEqual = QueryConditional.keyEqualTo(b -> b.partitionValue("movie01"));
        // 1a. Define a QueryConditional that adds a sort key criteria to the partition value criteria.
        QueryConditional sortGreaterThanOrEqualTo = QueryConditional.sortGreaterThanOrEqualTo(b -> b.partitionValue("movie01").sortValue("actor2"));
        // 2. Define a filter expression that filters out items whose attribute value is null.
        final Expression filterOutNoActingschoolname = Expression.builder().expression("attribute_exists(actingschoolname)").build();

        // 3. Build the query request.
        QueryEnhancedRequest tableQuery = QueryEnhancedRequest.builder()
                .queryConditional(keyEqual)
                .filterExpression(filterOutNoActingschoolname)
                .build();
        // 4. Perform the query.
        PageIterable<MovieActor> pagedResults = movieActorTable.query(tableQuery);
        logger.info("page count: {}", pagedResults.stream().count()); // Log  number of pages.

        pagedResults.items().stream()
                .sorted()
                .forEach(
                        item -> logger.info(item.toString()) // Log the sorted list of items.
                );

Esta é a saída gerada pela execução do método. A saída exibe itens com um valor movieName de movie01 e não exibe nenhum item com actingSchoolName igual a null.

2023-03-05 13:11:05 [main] INFO  org.example.tests.QueryDemo:46 - page count: 1
2023-03-05 13:11:05 [main] INFO  org.example.tests.QueryDemo:51 - MovieActor{movieName='movie01', actorName='actor1', actingAward='actingaward1', actingYear=2001, actingSchoolName='actingschool1'}
2023-03-05 13:11:05 [main] INFO  org.example.tests.QueryDemo:51 - MovieActor{movieName='movie01', actorName='actor2', actingAward='actingaward2', actingYear=2001, actingSchoolName='actingschool2'}
2023-03-05 13:11:05 [main] INFO  org.example.tests.QueryDemo:51 - MovieActor{movieName='movie01', actorName='actor4', actingAward='actingaward4', actingYear=2001, actingSchoolName='actingschool4'}

Na seguinte variação de solicitação de consulta mostrada anteriormente após a linha de comentário 3, o código substitui o keyEqual QueryConditional pelo sortGreaterThanOrEqualTo QueryConditional que foi definido após a linha de comentário 1a. O código a seguir também remove a expressão do filtro.

        QueryEnhancedRequest tableQuery = QueryEnhancedRequest.builder()
                .queryConditional(sortGreaterThanOrEqualTo)

Como essa tabela tem uma chave primária composta, todas as instâncias QueryConditionalexigem um valor de chave de partição. Métodos QueryConditional que começam com sort... indicam que uma chave de classificação é necessária. Os resultados não são classificados.

A saída a seguir exibe os resultados da consulta. A consulta retorna itens que têm um valor movieName igual a movie01 e somente itens que têm um valor actorName maior ou igual a actor2. Como o filtro foi removido, a consulta retorna itens que não têm valor para o atributo actingSchoolName.

2023-03-05 13:15:00 [main] INFO  org.example.tests.QueryDemo:46 - page count: 1
2023-03-05 13:15:00 [main] INFO  org.example.tests.QueryDemo:51 - MovieActor{movieName='movie01', actorName='actor2', actingAward='actingaward2', actingYear=2001, actingSchoolName='actingschool2'}
2023-03-05 13:15:00 [main] INFO  org.example.tests.QueryDemo:51 - MovieActor{movieName='movie01', actorName='actor3', actingAward='actingaward3', actingYear=2001, actingSchoolName='null'}
2023-03-05 13:15:00 [main] INFO  org.example.tests.QueryDemo:51 - MovieActor{movieName='movie01', actorName='actor4', actingAward='actingaward4', actingYear=2001, actingSchoolName='actingschool4'}