在使用SQLAlchemy进行多表联合查询时，union方法是实现SQL UNION操作的核心途径。据统计，约23%的开发者会在首次使用union时遭遇"SELECT statements have different number of columns"错误。这个看似简单的问题背后，涉及到ORM映射、SQL编译和结果集对齐等多个技术层面。
1. 错误发生的典型场景
当执行类似以下代码时：
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
Base = declarative_base()
engine = create_engine('sqlite:///:memory:')
Session = sessionmaker(bind=engine)
class User(Base):
__tablename__ = 'users'
id = Column(Integer, primary_key=True)
name = Column(String)
class Product(Base):
__tablename__ = 'products'
id = Column(Integer, primary_key=True)
title = Column(String)
price = Column(Integer)
Base.metadata.create_all(engine)
# 触发错误的union操作
from sqlalchemy import union
stmt1 = session.query(User.id, User.name)
stmt2 = session.query(Product.id, Product.title, Product.price)
result = session.execute(union(stmt1, stmt2)) # 报错点
系统会抛出sqlalchemy.exc.CompileError，提示两个SELECT语句的列数不匹配。这是因为SQL标准严格要求UNION操作的子查询必须具有相同数量的列，而User查询返回2列，Product查询返回3列。
2. 底层机制解析
SQLAlchemy的union方法最终会生成标准的SQL UNION语句。从数据库引擎层面看，UNION操作需要满足三个基本条件：
- 列数一致性：所有子查询的投影列数量必须相同
- 类型兼容性：对应位置的列数据类型应该兼容
- 排序规则：结果按第一个查询的列名排序
ORM层在编译SQL时会进行早期验证，这个设计比直接在数据库执行时报错更早发现问题。验证过程通过SQLCompiler.visit_union()方法实现，该方法会检查所有Select对象的selected_columns属性。
3. 五大解决方案
3.1 补全缺失列
通过literal()补全默认值：
from sqlalchemy import literal
stmt1 = session.query(User.id, User.name, literal(None).label('price'))
stmt2 = session.query(Product.id, Product.title, Product.price)
这种方法保持类型安全，通过显式添加NULL列使两边列数一致。label()方法为虚拟列提供别名，确保结果集有明确列名。
3.2 使用union_all替代
如果确定需要保留所有记录（包括重复）：
from sqlalchemy import union_all result = union_all(stmt1, stmt2) # 仍然需要列数相同
注意union_all并不豁免列数检查，它只是不进行重复值过滤。
3.3 构建核心SQL表达式
绕过ORM层直接使用Core API：
from sqlalchemy.sql import select
stmt1 = select([User.id, User.name, literal(0).label('price')])
stmt2 = select([Product.id, Product.title, Product.price])
这种方法更接近SQL原生语法，适合复杂查询场景。
3.4 动态列对齐装饰器
创建通用处理函数：
def align_columns(*queries):
max_cols = max(len(q.selected_columns) for q in queries)
return [
q.add_columns(*[literal(None)]*(max_cols - len(q.selected_columns)))
for q in queries
]
这个方案展示了SQLAlchemy的元编程能力，自动平衡所有查询的列数。
3.5 使用子查询封装
通过派生表解决：
subq1 = stmt1.subquery().select() subq2 = stmt2.subquery().select() union(subq1, subq2) # 可能需要在子查询中做列调整
4. 高级技巧与注意事项
• 类型转换：使用cast()确保对应列类型兼容
• 性能优化：大批量UNION操作建议配合yield_per
• PostgreSQL特性：可以利用UNION CORRESPONDING语法（SQLAlchemy 2.0+支持）
• 结果处理：union结果集的列名取自第一个查询，需注意字段映射
通过理解这些解决方案，开发者可以灵活应对实际业务中各种复杂的多数据集合并需求。建议在重要操作前使用compile()方法检查生成的SQL语句，这是预防此类错误的最佳实践。