SQLAlchemy操作Google Spanner时ASC+nulls_last触发ProgrammingError

需要实现按指定列名动态排序查询结果，且将NULL值始终排在列表末尾。使用desc降序排序时功能正常，但asc升序排序结合nulls_last会触发ProgrammingError，错误由InvalidArgument引发。

依赖版本 #

• sqlalchemy = "~1.4.37"
• sqlalchemy-spanner = "~1.2.0"
  数据库：Google Cloud Spanner

原始查询代码 #

查询生成方法 #

@classmethod
def get_my_models(cls, sort_by: str, order_by: str) -> List["MyModel"]:
    with read_session_scope(get_engine()) as session:
        sort_attr = getattr(MyModel, sort_by)

        query = (
            session.query(MyModel)
            .filter(func.coalesce(MyModel.is_deleted, False).is_(False))
            .order_by(nulls_last(getattr(sort_attr, order_by)()))
        )
    return [MyModel.from_orm(model) for model in query.all()]

转换后的SQL语句 #

• 当传入sort_by="my_column"和order_by="desc"时，生成的SQL：

SELECT ... FROM my_model WHERE coalesce(my_model.is_deleted, false) IS false ORDER BY 
my_model.my_column DESC NULLS LAST

该语句执行正常，但不符合需求（需要NULL始终在末尾，降序时NULLS LAST是正确的，但升序时出错）。

• 当传入sort_by="my_column"和order_by="asc"时，生成的SQL：

SELECT ... FROM my_model WHERE coalesce(my_model.is_deleted, false) IS false ORDER BY 
my_model.my_column ASC NULLS LAST

语句看似正常，但执行query.all()时触发错误：

sqlalchemy.exc.ProgrammingError: (google.cloud.spanner_dbapi.exceptions.ProgrammingError)

错误调用栈 #

File "/opt/pysetup/.venv/lib/python3.8/site-packages/sqlalchemy/orm/query.py", line 2768, in all
    return self._iter().all()
  File "/opt/pysetup/.venv/lib/python3.8/site-packages/sqlalchemy/orm/query.py", line 2903, in _iter
    result = self.session.execute(
  File "/opt/pysetup/.venv/lib/python3.8/site-packages/sqlalchemy/orm/session.py", line 1712, in execute
    result = conn._execute_20(statement, params or {}, execution_options)
  File "/opt/pysetup/.venv/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 1631, in _execute_20
    return meth(self, args_10style, kwargs_10style, execution_options)
  File "/opt/pysetup/.venv/lib/python3.8/site-packages/sqlalchemy/sql/elements.py", line 332, in _execute_on_connection
    return connection._execute_clauseelement(
  File "/opt/pysetup/.venv/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 1498, in _execute_clauseelement
    ret = self._execute_context(
  File "/opt/pysetup/.venv/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 1862, in _execute_context
    self._handle_dbapi_exception(
  File "/opt/pysetup/.venv/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 2043, in _handle_dbapi_exception
    util.raise_(
  File "/opt/pysetup/.venv/lib/python3.8/site-packages/sqlalchemy/util/compat.py", line 208, in raise_
    raise exception
  File "/opt/pysetup/.venv/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 1819, in _execute_context
    self.dialect.do_execute(
  File "/opt/pysetup/.venv/lib/python3.8/site-packages/google/cloud/sqlalchemy_spanner/sqlalchemy_spanner.py", line 1006, in do_execute
    cursor.execute(statement, parameters)
  File "/opt/pysetup/.venv/lib/python3.8/site-packages/google/cloud/spanner_dbapi/cursor.py", line 70, in wrapper
    return function(cursor, *args, **kwargs)
  File "/opt/pysetup/.venv/lib/python3.8/site-packages/google/cloud/spanner_dbapi/cursor.py", line 286, in execute
    raise ProgrammingError(getattr(e, "details", e))
sqlalchemy.exc.ProgrammingError: (google.cloud.spanner_dbapi.exceptions.ProgrammingError) []

错误详情 #

self = <google.cloud.spanner_dbapi.cursor.Cursor object at 0xffffacfca220>
sql = 'SELECT ... coalesce(my_model.is_deleted, @a0) IS false ORDER BY my_model.my_column ASC NULLS LAST'
args = {'a0': False}

    @check_not_closed
    def execute(self, sql, args=None):
        """Prepares and executes a Spanner database operation.
    
        :type sql: str
        :param sql: A SQL query statement.
    
        :type args: list
        :param args: Additional parameters to supplement the SQL query.
        """
        self._result_set = None
    
        try:
            if self.connection.read_only:
                self._handle_DQL(sql, args or None)
                return
    
            class_ = parse_utils.classify_stmt(sql)
            if class_ == parse_utils.STMT_DDL:
                self._batch_DDLs(sql)
                if self.connection.autocommit:
                    self.connection.run_prior_DDL_statements()
                return
    
            # For every other operation, we've got to ensure that
            # any prior DDL statements were run.
            # self._run_prior_DDL_statements()
            self.connection.run_prior_DDL_statements()
    
            if class_ == parse_utils.STMT_UPDATING:
                sql = parse_utils.ensure_where_clause(sql)
    
            if class_ != parse_utils.STMT_INSERT:
                sql, args = sql_pyformat_args_to_spanner(sql, args or None)
    
            if not self.connection.autocommit:
                statement = Statement(
                    sql,
                    args,
                    get_param_types(args or None)
                    if class_ != parse_utils.STMT_INSERT
                    else {},
                    ResultsChecksum(),
                    class_ == parse_utils.STMT_INSERT,
                )
    
                (
                    self._result_set,
                    self._checksum,
                ) = self.connection.run_statement(statement)
                while True:
                    try:
                        self._itr = PeekIterator(self._result_set)
                        break
                    except Aborted:
                        self.connection.retry_transaction()
                return
    
            if class_ == parse_utils.STMT_NON_UPDATING:
                self._handle_DQL(sql, args or None)
            elif class_ == parse_utils.STMT_INSERT:
                _helpers.handle_insert(self.connection, sql, args or None)
            else:
                self.connection.database.run_in_transaction(
                    self._do_execute_update, sql, args or None
                )
        except (AlreadyExists, FailedPrecondition, OutOfRange) as e:
            raise IntegrityError(getattr(e, "details", e))
        except InvalidArgument as e:
>           raise ProgrammingError(getattr(e, "details", e))
E           sqlalchemy.exc.ProgrammingError: (google.cloud.spanner_dbapi.exceptions.ProgrammingError) []

问题原因与解决方法 #

Google Cloud Spanner的SQL语法中，升序（ASC）排序默认将NULL放在末尾，不需要显式指定NULLS LAST；而SQLAlchemy生成的语句强行添加该子句，触发了Spanner的语法校验错误。

方案一：动态控制nulls_last的使用 #

只在降序排序时显式添加NULLS LAST，升序时使用默认行为：

@classmethod
def get_my_models(cls, sort_by: str, order_by: str) -> List["MyModel"]:
    with read_session_scope(get_engine()) as session:
        sort_attr = getattr(MyModel, sort_by)
        sort_expr = getattr(sort_attr, order_by.lower())()
        
        # 仅降序时需要显式指定NULLS LAST
        if order_by.lower() == "desc":
            sort_expr = nulls_last(sort_expr)

        query = (
            session.query(MyModel)
            .filter(func.coalesce(MyModel.is_deleted, False).is_(False))
            .order_by(sort_expr)
        )
    return [MyModel.from_orm(model) for model in query.all()]

方案二：强制NULL始终在末尾（兼容所有排序方向） #

通过先判断字段是否为NULL，再排序的方式，确保无论升序还是降序，NULL都排在最后：

@classmethod
def get_my_models(cls, sort_by: str, order_by: str) -> List["MyModel"]:
    with read_session_scope(get_engine()) as session:
        sort_attr = getattr(MyModel, sort_by)
        # 先按是否为NULL排序（NULL标记为True，升序时排在后面）
        null_priority = func.isnull(sort_attr).asc()
        # 再按指定字段和方向排序
        field_sort = getattr(sort_attr, order_by.lower())()
        
        query = (
            session.query(MyModel)
            .filter(func.coalesce(MyModel.is_deleted, False).is_(False))
            .order_by(null_priority, field_sort)
        )
    return [MyModel.from_orm(model) for model in query.all()]

以上两种方法均能满足需求，同时避免Spanner的语法错误。

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