sqlalchemy: sqlalchemy.exc.OperationalError: table xxx has x values for y columns

Describe the bug

(this is a followup of https://groups.google.com/g/sqlalchemy/c/opm2FVNKTTI/m/BCiORAULBQAJ)

There is an issue with column_property marked deferred=True and CTE (but maybe it happens in other cases than CTE …) SQLAlchemy loads column in the WITH RECURSIVE clause (which shouldn’t as they are deferred=True) but not within the SELECT clause. Using explicit orm.undefer() or orm.undefer_group() doesn’t resolve the issue.

import logging
import unittest

from sqlalchemy import Column
from sqlalchemy import create_engine
from sqlalchemy import Integer
from sqlalchemy import select
from sqlalchemy import sql
from sqlalchemy import String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import column_property
from sqlalchemy.orm import declarative_base
from sqlalchemy.orm import Session
from sqlalchemy.orm import undefer
from sqlalchemy.orm import undefer_group

Base = declarative_base()

log = logging.getLogger(__name__)

############
# ENTITIES #
############


class Content(Base):

    __tablename__ = "content"

    id = Column(Integer, primary_key=True)
    col1 = Column(Integer)
    col2 = Column(Integer)
    firstname = Column(String())
    lastname = Column(String())

    fullname = column_property(firstname + " " + lastname, deferred=True)
    colx = column_property(col1 + col2, deferred=True)
    coly = column_property(col1 + col1)
    colz = column_property(col2 + col2, deferred=True, group="gr1")
    colzz = column_property(col1 + col2 + col2, deferred=True, group="gr2")


e = create_engine("sqlite://", echo=False)

Base.metadata.create_all(e)


class FooTest(unittest.TestCase):
    @classmethod
    def setup_class(cls):
        s = Session(e)

        c1 = Content(col1=1, col2=2, firstname="f_a", lastname="l_a")

        c2 = Content(col1=3, col2=4, firstname="f_b", lastname="l_b")

        c3 = Content(col1=5, col2=6, firstname="f_c", lastname="l_c")

        s.add(c1)
        s.add(c2)
        s.add(c3)

        s.commit()
        s.close()

    def test_deferred(self):
        s = Session(e)

        root = (
            select(Content, sql.literal(1, type_=Integer).label("level"))
            .filter(Content.col1 > 1)
            .cte(name="parents", recursive=True)
        )

        root = root.union_all(
            select(Content, root.c.level + 1).join(
                root, root.c.id == Content.col2
            )
        )

        q = (
            select(Content)
            .join(root, root.c.id == Content.id)
            .add_columns(root.c.level.label("level"))
            .order_by(root.c.level)
        )

        s.execute(q)
        s.close()

    def test_explicit_undefer(self):
        s = Session(e)

        root = (
            select(Content, sql.literal(1, type_=Integer).label("level"))
            .filter(Content.col1 > 1)
            .cte(name="parents", recursive=True)
        )

        root = root.union_all(
            select(Content, root.c.level + 1).join(
                root, root.c.id == Content.col2
            )
        )

        q = (
            select(Content)
            .join(root, root.c.id == Content.id)
            .add_columns(root.c.level.label("level"))
            .options(undefer("*"))
            .order_by(root.c.level)
        )

        s.execute(q)
        s.close()

    def test_explicit_undefer_group(self):
        s = Session(e)

        root = (
            select(Content, sql.literal(1, type_=Integer).label("level"))
            .filter(Content.col1 > 1)
            .cte(name="parents", recursive=True)
        )

        root = root.union_all(
            select(Content, root.c.level + 1).join(
                root, root.c.id == Content.col2
            )
        )

        q = (
            select(Content)
            .join(root, root.c.id == Content.id)
            .add_columns(root.c.level.label("level"))
            .options(undefer_group("gr1"))
            .order_by(root.c.level)
        )

        s.execute(q)
        s.close()

    def test_normal(self):
        s = Session(e)

        q = select(Content).options(undefer_group("gr1"))

        s.execute(q)
        s.close()

I’ve attached a minimal MCVE here: poc.txt

Versions.

• OS: FreeBSD home.lan 13.0-RELEASE FreeBSD 13.0-RELEASE #0 releng/13.0-n244733-ea31abc261f: Fri Apr 9 04:24:09 UTC 2021 root@releng1.nyi.freebsd.org:/usr/obj/usr/src/amd64.amd64/sys/GENERIC amd64
• Python: 3.7.9
• SQLAlchemy: 1.4.18
• Database: PostgreSQL
• DBAPI: psycopg2

Have a nice day!