fix(queryset): hydrate select_related by field order

CHANGELOG.rst

@@ -43,6 +43,10 @@ Fixed
 - Field declarations on models now resolve to their concrete type (e.g. ``CharField[str]``) in Pyright/Pylance instead of ``Field[Unknown]``; the ``Field.__new__`` type-check stub now returns ``Self``. (#2216)
 - Type hint for ``TransactionContext`` now returns a ``TransactionalDBClient`` instead of a raw database connection. This change gives the correct inferred type for the transaction context. (#2232)
 - Fix TSVectorField returned value conversion. (#2237)
+- ``select_related`` now maps JOIN columns by recorded field order instead of
+  parsing aliases, so long relation/column names work on Postgres (63-byte
+  identifier limit). (#1902)
+
 
 1.1.7
 -----

tests/test_select_related_long_aliases.py

@@ -0,0 +1,112 @@
+"""Regression tests for Postgres 63-byte identifier truncation on JOIN aliases.
+
+See https://github.com/tortoise/tortoise-orm/issues/1902
+"""
+
+import pytest
+
+from tests.testmodels import (
+    LongJoinAuthor,
+    LongJoinBook,
+    LongJoinChapter,
+    LongJoinParent,
+    ThisIsAnExcessivelyLongOneToOneChildModelName,
+)
+from tortoise.backends.base.executor import BaseExecutor
+
+
+class _IndexRecord:
+    """Row that supports positional access like an asyncpg Record."""
+
+    def __init__(self, keys: list[str], values: list[object]) -> None:
+        self._keys = keys
+        self._values = values
+
+    def __len__(self) -> int:
+        return len(self._values)
+
+    def __getitem__(self, item: int) -> object:
+        if isinstance(item, int):
+            return self._values[item]
+        raise KeyError(item)
+
+    def keys(self) -> list[str]:
+        return self._keys
+
+
+@pytest.mark.asyncio
+async def test_select_related_nested_long_fk_aliases(db):
+    author = await LongJoinAuthor.create()
+    book = await LongJoinBook.create(author_model_relation_with_long_name=author)
+    chapter = await LongJoinChapter.create(book_model_relation_with_long_name=book)
+
+    queryset = LongJoinChapter.all().select_related(
+        "book_model_relation_with_long_name",
+        "book_model_relation_with_long_name__author_model_relation_with_long_name",
+    )
+    sql = queryset.sql()
+    author_alias = (
+        "longjoinchapter__book_model_relation_with_long_name__"
+        "author_model_relation_with_long_name.id"
+    )
+    assert author_alias in sql
+    assert len(author_alias) > 63
+
+    loaded = await queryset.get(id=chapter.id)
+    assert loaded.book_model_relation_with_long_name.id == book.id
+    assert (
+        loaded.book_model_relation_with_long_name.author_model_relation_with_long_name.id
+        == author.id
+    )
+
+
+@pytest.mark.asyncio
+async def test_select_related_long_o2o_model_and_column(db):
+    parent = await LongJoinParent.create(name="parent", this_is_some_long_column_name="long-value")
+    child = await ThisIsAnExcessivelyLongOneToOneChildModelName.create(parent=parent)
+
+    queryset = ThisIsAnExcessivelyLongOneToOneChildModelName.filter(id=child.id).select_related(
+        "parent"
+    )
+    sql = queryset.sql()
+    long_alias = (
+        "thisisanexcessivelylongonetoonechildmodelname__parent.this_is_some_long_column_name"
+    )
+    assert long_alias in sql
+    assert len(long_alias) > 63
+
+    loaded = await queryset.first()
+    assert loaded is not None
+    assert loaded.parent.id == parent.id
+    assert loaded.parent.this_is_some_long_column_name == "long-value"
+    assert loaded.parent.name == "parent"
+
+
+@pytest.mark.asyncio
+async def test_select_related_long_aliases_with_only(db):
+    parent = await LongJoinParent.create(
+        name="only-parent", this_is_some_long_column_name="only-value"
+    )
+    child = await ThisIsAnExcessivelyLongOneToOneChildModelName.create(parent=parent)
+
+    loaded = await (
+        ThisIsAnExcessivelyLongOneToOneChildModelName.filter(id=child.id)
+        .only("id", "parent__this_is_some_long_column_name", "parent__name")
+        .select_related("parent")
+        .first()
+    )
+    assert loaded is not None
+    assert loaded.parent.this_is_some_long_column_name == "only-value"
+    assert loaded.parent.name == "only-parent"
+
+
+def test_row_keys_and_values_prefers_positional_access():
+    keys = ["truncated_alias", "truncated_alias"]
+    values = ["first", "second"]
+    row = _IndexRecord(keys, values)
+
+    got_keys, got_values = BaseExecutor._row_keys_and_values(row)
+    assert got_values == values
+    assert got_keys == keys
+    # dict() would collapse the duplicate truncated keys
+    assert list(dict(zip(got_keys, got_values)).values()) != values

tests/testmodels.py

@@ -1156,3 +1156,34 @@ class Drink(Model):
     toppings = fields.ManyToManyField(
         Flavor, related_name="topping_drinks", through="drink_topping"
     )
+
+
+class LongJoinAuthor(Model):
+    id = fields.UUIDField(primary_key=True)
+
+
+class LongJoinBook(Model):
+    id = fields.UUIDField(primary_key=True)
+    author_model_relation_with_long_name: fields.ForeignKeyRelation[LongJoinAuthor] = (
+        fields.ForeignKeyField("models.LongJoinAuthor", related_name="books")
+    )
+
+
+class LongJoinChapter(Model):
+    id = fields.UUIDField(primary_key=True)
+    book_model_relation_with_long_name: fields.ForeignKeyRelation[LongJoinBook] = (
+        fields.ForeignKeyField("models.LongJoinBook", related_name="chapters")
+    )
+
+
+class LongJoinParent(Model):
+    id = fields.IntField(primary_key=True)
+    name = fields.CharField(max_length=50)
+    this_is_some_long_column_name = fields.CharField(max_length=255)
+
+
+class ThisIsAnExcessivelyLongOneToOneChildModelName(Model):
+    id = fields.IntField(primary_key=True)
+    parent: fields.OneToOneRelation[LongJoinParent] = fields.OneToOneField(
+        "models.LongJoinParent", related_name="child"
+    )

tortoise/backends/base/executor.py

@@ -48,7 +48,8 @@ def __init__(
         prefetch_map: dict[str, set[str | Prefetch]] | None = None,
         prefetch_queries: dict[str, list[tuple[str | None, QuerySet]]] | None = None,
         select_related_idx: (
-            list[tuple[type[Model], int, str, type[Model], Iterable[str | None]]] | None
+            list[tuple[type[Model], int, str, type[Model], Iterable[str | None], Sequence[str]]]
+            | None
         ) = None,
     ) -> None:
         self.model = model
@@ -116,30 +117,27 @@ async def execute_select(
     ) -> list:
         _, raw_results = await self.db.execute_query(sql, values)
         instance_list = []
-        if self.select_related_idx:
-            _split_cache: dict[str, str] = {}
         for row_idx, row in enumerate(raw_results):
             if row_idx != 0 and row_idx % CHUNK_SIZE == 0:
                 # Forcibly yield to the event loop to avoid blocking the event loop
                 # when selecting a large number of rows
                 await asyncio.sleep(0)
 
             if self.select_related_idx:
-                _, current_idx, _, _, path = self.select_related_idx[0]
-                row_items = list(dict(row).items())
-                instance: Model = self.model._init_from_db(**dict(row_items[:current_idx]))
+                _, current_idx, _, _, path, _ = self.select_related_idx[0]
+                row_keys, row_values = self._row_keys_and_values(row)
+                instance: Model = self.model._init_from_db(
+                    **dict(zip(row_keys[:current_idx], row_values[:current_idx]))
+                )
                 instances: dict[Any, Any] = {path: instance}
-                for model, index, *__, full_path in self.select_related_idx[1:]:
+                for model, index, _, _, full_path, field_names in self.select_related_idx[1:]:
                     (*path, attr) = full_path
-                    related_items = row_items[current_idx : current_idx + index]
-                    if any(v for _, v in related_items):
-                        related_kwargs = {}
-                        for k, v in related_items:
-                            fname = _split_cache.get(k)
-                            if fname is None:
-                                fname = _split_cache[k] = k.split(".", 1)[1]
-                            related_kwargs[fname] = v
-                        obj = model._init_from_db(**related_kwargs)
+                    related_values = row_values[current_idx : current_idx + index]
+                    if any(related_values):
+                        # Use field names recorded when the query was built. Postgres
+                        # truncates identifiers to 63 bytes, so related aliases cannot
+                        # be parsed back into field names after a long JOIN.
+                        obj = model._init_from_db(**dict(zip(field_names, related_values)))
                     elif index == 0:
                         # 0 signals that an empty "filler" object should be created in the case
                         # where a field of related model is selected but model itself isn't,
@@ -162,6 +160,26 @@ async def execute_select(
         await self._execute_prefetch_queries(instance_list)
         return instance_list
 
+    @staticmethod
+    def _row_keys_and_values(row: Any) -> tuple[list[Any], list[Any]]:
+        """Return column keys and values in SELECT order.
+
+        Prefer positional access so backends that keep duplicate truncated keys
+        (e.g. asyncpg Records) still yield one value per selected column.
+        """
+        if isinstance(row, dict):
+            return list(row.keys()), list(row.values())
+        try:
+            values = [row[i] for i in range(len(row))]
+        except (KeyError, TypeError, IndexError):
+            items = list(dict(row).items())
+            return [k for k, _ in items], [v for _, v in items]
+        try:
+            keys = list(row.keys())
+        except (AttributeError, TypeError):
+            keys = list(dict(row).keys())
+        return keys, values
+
     async def execute_union(
         self, sql: str, app_field: str, model_field: str, models: set[type[Model]]
     ) -> list:

tortoise/queryset.py

@@ -395,8 +395,15 @@ def __init__(self, model: type[MODEL]) -> None:
         self._select_for_update_no_key: bool = False
         self._select_related: set[str] = set()
         self._select_related_idx: list[
-            tuple[type[Model], int, Table | str, type[Model], Iterable[str | None]]
-        ] = []  # format with: model,idx,model_name,parent_model
+            tuple[
+                type[Model],
+                int,
+                Table | str,
+                type[Model],
+                Iterable[str | None],
+                tuple[str, ...],
+            ]
+        ] = []  # format: model, idx, table/name, parent_model, path, field_names
         self._force_indexes: set[str] = set()
         self._use_indexes: set[str] = set()
 
@@ -1155,23 +1162,28 @@ def _join_select_related(self, lookup_expression: str) -> tuple[type[Model], Tab
             if self._fields_for_select:
                 continue
 
-            related_fields = field.related_model._meta.db_fields
+            related_fields = tuple(field.related_model._meta.db_fields)
             append_item = (
                 field.related_model,
                 len(related_fields),
                 field.model_field_name,
                 model,
                 path,
+                related_fields,
             )
             model = field.related_model
+            # Only select columns the first time this path is recorded. Nested
+            # lookups (e.g. left then left__extra) would otherwise append the
+            # parent columns again while _select_related_idx stays unique, and
+            # positional hydration then reads the wrong slice.
             if append_item not in self._select_related_idx:
                 self._select_related_idx.append(append_item)
-            self.query = self.query.select(
-                *[
-                    table[related_field].as_(f"{table.get_table_name()}.{related_field}")
-                    for related_field in related_fields
-                ]
-            )
+                self.query = self.query.select(
+                    *[
+                        table[related_field].as_(f"{table.get_table_name()}.{related_field}")
+                        for related_field in related_fields
+                    ]
+                )
         return model, table
 
     def _resolve_only(self, only_lookup_expressions: tuple[str, ...]) -> None:
@@ -1197,6 +1209,7 @@ def _resolve_only(self, only_lookup_expressions: tuple[str, ...]) -> None:
                     table,
                     self.model,
                     (None,),
+                    tuple(data_fields),
                 )
             )
             try:
@@ -1222,6 +1235,7 @@ def _resolve_only(self, only_lookup_expressions: tuple[str, ...]) -> None:
                     self.model._meta.basetable,
                     self.model,
                     (None,),
+                    (),
                 )
             )
 
@@ -1252,6 +1266,7 @@ def _resolve_only(self, only_lookup_expressions: tuple[str, ...]) -> None:
                         table,
                         referring_model,
                         path,
+                        tuple(data_fields) if i == len(fetch_fields) - 1 else (),
                     )
                 )
                 added_paths.add(path)
@@ -1277,6 +1292,7 @@ def _make_query(self) -> None:
                 table,
                 self.model,
                 (None,),
+                tuple(self.model._meta.db_fields),
             )
             self._select_related_idx.append(append_item)
         self.resolve_ordering(