将SQLAlchemy查询结果序列化跨越进程边界

Question

在使用 Python 3.10.12 和 SQLAlchemy 2.0.23（以及Ubuntu下的PostgreSQL 12，如果这有影响的话）的情况下：

如何序列化现有的SQLAlchemy查询，以便在另一个进程中反序列化并获取其结果？

换句话说，在进程1中，我有一个SqlAlchemy的query对象，但我想在进程2中运行for result in query.yield_per(1000): ...以及query.count()。

请注意，我不打算序列化查询结果，而是想要序列化和反序列化查询本身。

至关重要的一点是，在任何时候都不能缓存/物化结果集，因为它太大以至于无法完全放入RAM中。

ripper234 · Answer

通常人们希望通过传递偏移量/限制窗口或批次的结果ID，将结果集分割到子进程中进行处理。我想我不太理解这种使用场景。您可以尝试使用这个旧版的序列化器，但我认为整个概念将很难维护和调试，即Pickling（序列化）和Unpickling（反序列化）SQLAlchemy表对象及其相关的机制。

废弃扩展

关于该扩展未来讨论的链接

Aaron Maenpaa · Answer

在进程1中：

from sqlalchemy.dialects import postgresql

# 序列化查询语句
query_sql = str(query.statement.compile(dialect=postgresql.dialect(), compile_kwargs={"literal_binds": True}))

在进程2中：

from sqlalchemy import text, select

# 反序列化查询语句并执行
# 使用 `from_statement` 方法以获取完整的ORM对象，而非仅仅通过 `session.execute(text(query_sql))` 得到Row元组
objects = session.execute(
    select(MyTable).execution_options(yield_per=100).from_statement(text(query_sql)),
    execution_options={'yield_per': 100},
).scalars()

# 对大型查询结果使用服务器端游标逐批读取
for obj in objects.yield_per(100):
    ...

上述解决方案并不完美，因为objects是sqlalchemy.engine.result.ScalarResult类型，它不像原始的query那样具有.count()方法，也没有rowcount属性。如何高效地（通过服务器端查询）获取结果计数？

一种选择是在进程1中获取结果计数，并将其作为int类型与query_sql字符串一起传递给进程2。但这会产生竞态条件（计数与查询结果不同步），因此我更倾向于在同一会话中同时获取两者。