求助：基于Scala Scanamo库实现DynamoDB二级索引分页查询示例

问题背景 #

需要基于Scala的Scanamo库实现DynamoDB二级索引的分页查询，手写分页逻辑时遇到问题，且不理解queryPaginatedM()的工作机制，现有尝试代码如下：

protected def querySecondaryIndexPaginated[T, U](
      query: Query[T],
      table: Table[U],
      indexName: String,
      pageSize: Int,
      lastEvaluatedKeyOpt: Option[Map[String, AttributeValue]]
  )(implicit format: DynamoFormat[T]): Future[MyDynamoResult[U]] = {
    val queryRequest: ScanamoOps[List[Either[DynamoReadError, U]]] = lastEvaluatedKeyOpt match {
      case Some(lastEvaluatedKey) => table.index(indexName).limit(pageSize).from(UniqueKey(lastEvaluatedKey)).query(query = query)
      case None                   => table.index(indexName).limit(pageSize).query(query = query)
    }
    scanamo
      .execFuture(queryRequest)
      .map....
}

解决方案 #

Scanamo提供的queryPaginatedM方法已封装好分页逻辑，会自动处理LastEvaluatedKey并返回包含当前页数据和下一页游标的结果，以下是完整实现：

完整分页查询方法 #

import org.scanamo._
import org.scanamo.generic.auto._
import software.amazon.awssdk.services.dynamodb.model.AttributeValue
import scala.concurrent.{Future, ExecutionContext}

// 自定义结果类，封装分页数据和下一页游标
case class MyDynamoResult[U](
  data: List[U],
  lastEvaluatedKey: Option[Map[String, AttributeValue]]
)

class DynamoPaginator(scanamo: ScanamoAsync)(implicit ec: ExecutionContext) {
  def querySecondaryIndexPaginated[T, U](
    tableName: String,
    indexName: String,
    query: Query[T],
    pageSize: Int,
    lastEvaluatedKeyOpt: Option[Map[String, AttributeValue]] = None
  )(implicit formatT: DynamoFormat[T], formatU: DynamoFormat[U]): Future[MyDynamoResult[U]] = {
    val table = Table[U](tableName)
    val baseQuery = table.index(indexName).limit(pageSize)
    
    // 根据是否有游标构建分页查询
    val paginatedQuery = lastEvaluatedKeyOpt match {
      case Some(key) => baseQuery.from(UniqueKey(key)).queryPaginatedM(query)
      case None => baseQuery.queryPaginatedM(query)
    }
    
    scanamo.execFuture(paginatedQuery).map { paginatedResult =>
      MyDynamoResult(
        // 过滤读取成功的数据，若需处理错误可调整逻辑
        data = paginatedResult.items.collect { case Right(item) => item },
        lastEvaluatedKey = paginatedResult.lastEvaluatedKey.map(_.toMap)
      )
    }
  }
}

核心要点说明 #

• queryPaginatedM的工作机制：
  这是Scanamo专为异步场景设计的分页API，返回PaginatedResult对象，包含两个核心字段：
  • items: 当前页的查询结果，以Either[DynamoReadError, U]形式返回，区分读取成功和失败的数据
  • lastEvaluatedKey: 当存在未返回的数据时，该字段为下一页查询的游标；若数据已全部返回，则为None
• 游标传递：使用from(UniqueKey(key))方法指定从上次查询的结束位置继续获取数据
• 错误处理：示例中直接过滤读取失败的数据，实际项目中可根据需求收集错误信息，比如同时返回成功数据和错误列表

调用示例 #

// 定义表对应的Case Class
case class User(id: String, email: String, age: Int)

// 初始化Scanamo异步客户端
val scanamoAsync = ScanamoAsync()
val paginator = new DynamoPaginator(scanamoAsync)

// 构建查询条件：查询age≥18的用户，使用email二级索引
val ageQuery = Query.gte(User.age, 18)

// 第一页查询
paginator.querySecondaryIndexPaginated[String, User](
  tableName = "users",
  indexName = "email-index",
  query = ageQuery,
  pageSize = 20
).map { firstPage =>
  println(s"第一页数据：${firstPage.data}")
  // 存在下一页时，传入游标继续查询
  firstPage.lastEvaluatedKey.foreach { nextCursor =>
    paginator.querySecondaryIndexPaginated[String, User](
      tableName = "users",
      indexName = "email-index",
      query = ageQuery,
      pageSize = 20,
      lastEvaluatedKeyOpt = Some(nextCursor)
    ).map(secondPage => println(s"第二页数据：${secondPage.data}"))
  }
}

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