Ecto查询仅返回20条而非预期1200条的问题求助

针对你遇到的情况——明明数据库有271K条completed=0的记录，但设置demand=1200、offset=0的Ecto查询只返回20条，结合GenStage的3阶段运行环境，我整理了几个可能的原因和排查步骤：

1. 先确认实际传入的参数和生成的SQL是否正确 #

首先看你的控制台输出：IO.puts "demand/offset: #{demand}/#{offset}"有没有打印出1200/0？如果不是，那问题出在GenStage的demand传递逻辑上——可能下游阶段的实际请求demand被限制了，或者你的producer代码里修改了传入get/2的参数。

如果参数是对的，那把Ecto查询转换成原生SQL直接在数据库里执行，验证数据库层面的返回结果：

# 在你的代码里临时添加这行，生成原生SQL
query = from t in C3poEvent, where: t.completed == 0, order_by: [asc: t.id], limit: 1200, offset: 0
IO.inspect Ecto.Adapters.SQL.to_sql(:all, C3poRepo, query)

把输出的SQL复制到数据库客户端（比如psql、MySQL CLI）执行：

• 如果数据库返回1200条：问题出在Ecto或GenStage的后续处理
• 如果数据库只返回20条：那是数据库层面的过滤，比如**行级安全策略（RLS）**限制了当前用户能看到的数据，或者数据库里的completed字段实际有隐性过滤（比如部分记录的completed是字符串"0"而不是数字0，导致匹配失败）

2. 检查Schema字段类型是否匹配 #

你问到是否和events的实际类型有关，核心要看C3poEvent的completed字段定义是否和数据库一致：
比如数据库里completed是integer类型，但你的schema里定义成了:boolean：

# 错误示例：类型不匹配可能导致过滤逻辑异常
schema "c3po_events" do
  field :completed, :boolean # 数据库里是integer 0/1
end

这种情况下，Ecto会把查询条件t.completed == 0转换成WHERE completed = false，但数据库里的0可能被Ecto解析为true（不同数据库的类型转换规则有差异），导致实际匹配的记录远少于预期。

3. GenStage阶段间的过滤或修改 #

因为代码运行在3阶段的GenStage中，要检查下游阶段有没有对返回的events做过滤：
比如consumer阶段有没有只取前20条，或者中间的processor阶段不小心丢弃了数据？另外，GenStage的demand机制是下游向上游请求，如果你手动设置的demand=1200没有被正确传递到get/2函数，也会导致查询的limit值不对。

4. 数据竞态问题 #

如果数据库在你查询的瞬间，有其他进程正在更新completed字段（比如把0改成1），那Ecto查询到的结果可能会比预期少。可以尝试用快照隔离级别执行查询，避免这种竞态：

events = C3poRepo.transaction(fn ->
  C3poRepo.all(
    from t in C3poEvent,
    where: t.completed == 0,
    order_by: [asc: t.id],
    limit: ^demand,
    offset: ^offset
  )
end, isolation: :snapshot)

排查优先级 #

1. 验证控制台的demand/offset输出是否正确
2. 生成原生SQL在数据库执行，确认数据源的返回量
3. 检查Schema字段类型和数据库的一致性
4. 排查GenStage各阶段的数据处理逻辑

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