DynamoDB Enhanced查询无结果：我的Query语句是否存在问题？

你的Query语句确实存在两处关键问题，导致你看不到预期的查询结果：

1. 异步操作未等待执行完成 #

你使用的是DynamoDbAsyncTable的异步查询API，返回的PagePublisher是Reactive Streams类型的发布者。你的测试方法仅仅订阅了发布者，但没有等待整个异步流处理完毕就直接结束了——测试方法执行结束后，JVM可能直接终止，导致订阅的回调逻辑根本没机会执行，自然看不到输出。

2. 订阅逻辑存在潜在的索引越界风险 #

在订阅回调里，你用atomicInteger.get()去获取page.items()的元素，但如果一个Page包含多个条目，当atomicInteger的值超过当前page的item数量时，会抛出IndexOutOfBoundsException。而且这种方式也没法正确遍历所有返回的条目。

修正后的测试代码 #

下面是修复后的版本，解决了上述两个问题：

@Test
public void testQueryWithPKandSKBegin() throws InterruptedException {
    DynamoDbAsyncTable<DigitalFormDao> digitalformTable = dynamoDbEnhancedAsyncClient
            .table("digital_form", TableSchema.fromBean(DigitalFormDao.class));
    
    // 构建查询条件，这部分你的原代码逻辑是正确的
    PagePublisher<DigitalFormDao> digitalForms = digitalformTable.query(
            r -> r.queryConditional(
                    QueryConditional.sortBeginsWith(
                            Key.builder()
                                    .partitionValue("FORM#ABC123")
                                    .sortValue("INFO")
                                    .build()
                    )
            )
    );

    // 使用CountDownLatch等待异步查询完成
    CountDownLatch latch = new CountDownLatch(1);
    
    digitalForms.subscribe(
            page -> {
                // 遍历当前page的所有条目
                page.items().forEach(form -> System.out.println(form.getSk()));
            },
            // 处理查询过程中的错误
            error -> {
                error.printStackTrace();
                latch.countDown();
            },
            // 流完成时触发，释放等待的线程
            latch::countDown
    );
    
    // 等待异步操作完成，超时时间可根据实际情况调整
    latch.await(5, TimeUnit.SECONDS);
}

或者如果你习惯使用Reactor操作符（AWS SDK的异步API基于Reactor实现），也可以用更简洁的写法：

@Test
public void testQueryWithPKandSKBegin() throws ExecutionException, InterruptedException {
    DynamoDbAsyncTable<DigitalFormDao> digitalformTable = dynamoDbEnhancedAsyncClient
            .table("digital_form", TableSchema.fromBean(DigitalFormDao.class));

    digitalformTable.query(
                    r -> r.queryConditional(
                            QueryConditional.sortBeginsWith(
                                    Key.builder()
                                            .partitionValue("FORM#ABC123")
                                            .sortValue("INFO")
                                            .build()
                            )
                    )
            )
            // 展平所有page的条目为单个流
            .flatMapIterable(Page::items)
            // 打印每个条目的SK
            .doOnNext(form -> System.out.println(form.getSk()))
            // 等待整个流处理完成
            .blockLast();
}

额外说明 #

• 你原代码中sortBeginsWith的条件是正确的：你的SK是INFO#ABC123，前缀INFO可以准确匹配到该条目。
• 异步场景下的测试一定要确保等待所有异步操作完成，否则会出现“假阴性”结果——不是查询没返回数据，而是测试代码没等到数据就提前结束了。

内容的提问来源于stack exchange，提问作者toblerones