sqlalchemy: with_loader_criteria not applying for select(A, B).join(B), inconsistent for select(A).join(B) vs. select(A).join(A.bs)

Describe the bug

with_loader_criteria does not add criteria to entities referenced in a contains_eager option, even though it does correctly add criteria for other lazy load options like joinedload.

I’ve attached a reproducer that includes a passing test demonstrating similar functionality for joinedload and two failing tests. One for contains_eager, and another more complex example using contains_eager and a subquery with an alias (this is an example used in our application).

The expected behavior is that the test functions test_with_loader_criteria_contains_eager and test_with_loader_criteria_contains_eager_subquery return True.

To Reproduce

from sqlalchemy import Column, ForeignKey, Integer, select, String, create_engine
from sqlalchemy.orm import (
    declarative_base,
    contains_eager,
    joinedload,
    with_loader_criteria,
    relationship,
    aliased,
    Session)

Base = declarative_base()
engine = create_engine('sqlite://')


class A(Base):
    __tablename__ = "a"

    id = Column(Integer, primary_key=True)
    data = Column(String)
    bs = relationship("B", lazy="raise", backref="a")
    a1s = relationship("A1")


class B(Base):
    __tablename__ = "b"
    id = Column(Integer, primary_key=True)
    a_id = Column(ForeignKey("a.id"))
    data = Column(String)


class A1(Base):
    __tablename__ = "a1"
    id = Column(Integer, primary_key=True)
    data = Column(String)
    a_id = Column(ForeignKey("a.id"))


def mock_data(engine):
    Base.metadata.create_all(bind=engine)
    session = Session(bind=engine)

    a1 = A(id=0, data="a")
    a2 = A(id=1, data="b")
    b1 = B(id=1, a=a1)
    b2 = B(id=2, a=a1)

    session.add_all([a1, a2, b1, b2])
    session.commit()


def test_with_loader_criteria_contains_eager_subquery():
    session = Session(bind=engine)

    subquery = select(A).join(B).subquery(name='sub')
    subquery_aliased = aliased(A, alias=subquery, flat=True, adapt_on_names=True)
    query = select(subquery_aliased).options(
        contains_eager(subquery_aliased.bs, alias=subquery),
        with_loader_criteria(B, B.id == 1, include_aliases=True),
    )

    print(str(query))

    results = session.execute(query).unique().all()
    assert len(results) == 1
    assert results[0][0].id == 0
    return len(results[0][0].bs) == 1

def test_with_loader_criteria_contains_eager():
    session = Session(bind=engine)

    query = select(A).join(B).options(
        contains_eager(A.bs),
        with_loader_criteria(B, B.id == 1),
    )

    print(str(query))

    results = session.execute(query).unique().all()
    assert len(results) == 1
    assert results[0][0].id == 0
    # Filter on B not applied as expected
    return len(results[0][0].bs) == 1


def test_with_loader_criteria_joinedload():
    session = Session(bind=engine)

    query = select(A).options(
        joinedload(A.bs),
        with_loader_criteria(B, B.id == 1),
    )

    print(str(query))

    results = session.execute(query).unique().all()
    assert len(results) == 2
    assert results[0][0].id == 0
    return len(results[0][0].bs) == 1



if __name__ == "__main__":
    mock_data(engine)

    print("joinedload")
    # Filter on B applied as expected (len is 1)
    assert test_with_loader_criteria_joinedload()

    print("\ncontains_eager")
    # Filter on B not applied as expected (len is 2)
    assert test_with_loader_criteria_contains_eager() == False
    print("fail\n")

    print("contains_eager subquery")
    # Filter on B not applied as expected (len is 2)
    assert test_with_loader_criteria_contains_eager_subquery() == False
    print("fail\n")

Error

Below is the output from running this script. Note that contains_eager and contains_eager subquery does not include the join condition b_1.id = :id_2

joinedload
SELECT a.id, a.data, b_1.id AS id_1, b_1.a_id, b_1.data AS data_1
FROM a LEFT OUTER JOIN b AS b_1 ON a.id = b_1.a_id AND b_1.id = :id_2

contains_eager
SELECT b.id, b.a_id, b.data, a.id AS id_1, a.data AS data_1
FROM a JOIN b ON a.id = b.a_id
fail

contains_eager subquery
SELECT b.id, b.a_id, b.data, sub.id AS id_1, sub.data AS data_1
FROM b, (SELECT a.id AS id, a.data AS data
FROM a JOIN b ON a.id = b.a_id) AS sub
fail

Versions

• OS: OS X 10.15
• Python: 3.9.6
• SQLAlchemy: 1.4.25
• Database: sqlite
• DBAPI (eg: psycopg, cx_oracle, mysqlclient): sqlite

Additional context

No response