feat: Add prepared/compiled/cached parametrized queries

[RuslanUC]

Description

This pr adds "prepared" (or compiled/cached) parametrized queries that store ready-to-execute sql and list of parameters that need to be replaced when executing such query. Such queries avoid re-calculating filters/joins/etc every time query is being constructed, which improves performance (2x in best-case-scenario).

Motivation and Context

When measuring execution time for one of my projects, I noticed that sometimes _make_query took 2x more time than _execute.
Every time query is created and then executed, Tortoise-ORM and pypika recreate sql query itself and it's parameters list (code 1), though it is probably not the case when query is created first (not await'ed) and then executed after (code 2).

for _ in range(5)
    some_object = await SomeModel.get_or_none(a="b", c=1)

some_query = SomeModel.get_or_none(a="b", c=1)
for _ in range(5)
    some_object = await some_query

But even though second code reuses sql, it has an issue - parameters can't be changed.
So I added ability to both prepare/cache queries and use placeholder parameters in such queries.

How Has This Been Tested?

Ran tests via make test and then separately for all databases.
BUT, changes are not heavily-tested right now (only ~20 tests).

My notes about implementation

Currently, pr is only a draft, because of lack of proper testing, some unresolved todos and me being unsure about some details.
When adding prepared queries, i tried to not touch existing code as much as i can to a) not break anything; b) make sure that I dont negatively affect performance of existing code.

TODOs:

• Add ability to limit cache when collection parameters are present.
• Check for parameters mismatch when replacing placeholders with real values.
• Add more tests
• Add documentation

Details im not sure about:

• Naming of prepared/compiled/cached queryset classes, don't know which one is better, using "prepared" for now.
• PreparingQuerySet can be removed if _cache_key would be in QuerySet class - i think it'd better this way (when changing/adding something - no need to change it in two places), but it also would require changing limit/offset methods, as well as slightly change UpdateQuery/DeleteQuery/CountQuery/ValuesListQuery/ValuesQuery classes.
• Maybe there is some better way to use prepared queries (right now it should be used like query = SomeModel.prepare_sql("query-name").filter(...).prepared()), maybe prepare_sql and prepared methods could be renamed/shortened?
• What is the best way to test everything? Regular QuerySet has 60+ tests (only in test_queryset.py, not counting separate test files for filters, etc.), copying (or writing same tests from scratch) would result in a lot of duplication, i think.
• Is there a better way to create sql for queries with collection parameters than using custom SqlContext (see queryset_prepared.py#L165 and parameter.py#L34)?
• Maybe there is a way to use (or re-use) pypika Parameter class?

Benchmarking

I benchmarked regular and prepared queries, code and results available here.
Prepared queries perform best with .get/.get_or_none/.first queries (especially with complex queies, lots of jois, or both).
Worst case scenario for current implementation is queries with __in filter that every time use collections with different sizes (cache miss and sql recalculation every time), but even then performance is same as with regular queries.

Checklist:

• My code follows the code style of this project.
• My change requires a change to the documentation.
• I have updated the documentation accordingly.
• I have added the changelog accordingly.
• I have read the CONTRIBUTING document.
• I have added tests to cover my changes.
• All new and existing tests passed.

[codspeed-hq]

Merging this PR will not alter performance

✅ 24 untouched benchmarks

---

_{Comparing RuslanUC:feat/add-prepared-parametrized-queries (ebd33cb) with develop (ebc28bf)}

[abondar]

Hi! I looked through changes and have few concerns (I understand that it is only draft)

What would be cache invalidation strategy? Is it manual only?

Did you benchmark it on postgre or any other remote database? Does it provide any significant difference in that case?
Are you aiming at certain usecase, or intend it as general use? In what case users could understand that they should use these prepared queries instead of standard interface?

[RuslanUC]

Hi.

What would be cache invalidation strategy? Is it manual only?

For prepared queries (PreparedQuerySet, etc., not sql queries) I planned only manual invalidation (because cache can't grow "by itself", if someone prepared/cached 10 different queries - cache would contain only that 10 queries). For cached sql, I plan to add lru with max size controlled by ORM user.

Did you benchmark it on postgre or any other remote database? Does it provide any significant difference in that case?

Benchmarked on postgres 14.20 right now (running in local docker container, cant benchmark it using remote server), uploaded results to the repo with benchmarks. Difference is similar to one in sqlite benchmarks.
Upd: ran benchmarks on mysql 8 database (using asyncmy), performance difference is way smaller than in sqlite/postgres, but still noticable. I'll benchmark update, delete, count, exists, etc. queries in the next couple of days and update charts.

Are you aiming at certain usecase, or intend it as general use? In what case users could understand that they should use these prepared queries instead of standard interface?

I intend it for queries that either have complex filters/joins, or executed very frequently (and queries that dont select a lot of rows, because then all time spent either on db side, or creating python objects from query results). Thats why it should be documented and described when it is better to use prepared queries (but i think right now is too early to write documentation).

[abondar]

is it intended behaviour? Seems more logical to throw error on trying to modify prepared query than silently returning same query without changes

[RuslanUC]

Casting to DeleteQuery instead of PreparedDeleteQuery is not an intended behaviour (i'll fix it), but returning same query with different type (e.g. in values_list, values, update, etc.) is an intended behaviour (at least for now). This is because when query is already in cache and is returned from QuerySet.prepare_sql(), all filter and other queryset methods will be executed, e.g.:

query = SomeModel.prepare_sql("some-query").filter(id=123).delete()

First time the code is executed, .prepare_sql returns PreparingQuerySet, then .filter returns PreparingQuerySet and then .delete returns PreparedDeleteQuery (and places it in cache).
Next time the code is executed, .prepare_sql returns PreparedDeleteQuery (not PreparingQuerySet like it was in the first time), then .filter and .delete both retur same PreparedDeleteQuery and in the end, in both cases we have same query. If .delete would throw an error - second execution of the code will fail.

[abondar]

Honestly current implementation seems a bit over the top for what it actually does, as duplicating whole queryset, making implicit cache storages and etc is quite big change with a lot of new APIs.
Also resulting api seems a wacky with all these type casts, queryset mirroring and implicit caching

Have you considered simplifying it other way?

E.g. instead of all that - add new "compile" method to queryset, that will returned prepared query, that user will manage himself, giving responsibility and control for cache management (and possible memory leak) to user

Here is example of what that compiled query could be:

"""                                                                                                                                                                         
  Compiled query support for Tortoise ORM.                                                                                                                                    

  Provides a single CompiledQuery class that caches SQL generation for all
  query types (SELECT, UPDATE, DELETE, COUNT, EXISTS, VALUES, VALUES_LIST).                                                                                                   
                                                                                                                                                                              
  Usage:                                                                                                                                                                      
      compiled = Author.filter(id__gte=Parameter("min_id")).order_by("id").compile("my-query")                                                                                
      results = await compiled.execute(min_id=42)                                                                                                                             
  """                                                                                                                                                                         

  from __future__ import annotations

  from collections import OrderedDict, defaultdict
  from typing import TYPE_CHECKING, Any, Awaitable, Callable, Generic, TypeVar

  from pypika_tortoise.queries import QueryBuilder

  from tortoise.backends.base.client import BaseDBAsyncClient
  from tortoise.exceptions import ParamsError
  from tortoise.parameter import CollectionParameter, Parameter, TortoiseSqlContext
  from tortoise.router import router

  if TYPE_CHECKING:
      from tortoise.models import Model

  MODEL = TypeVar("MODEL", bound="Model")

  # async (db, sql, filled_params) -> result
  ExecuteFn = Callable[[BaseDBAsyncClient, str, list[Any]], Awaitable[Any]]


  # ---------------------------------------------------------------------------
  # Bounded LRU cache
  # ---------------------------------------------------------------------------

  class _BoundedLRU:
      """Minimal bounded LRU cache for collection-size-keyed SQL variants."""

      __slots__ = ("_data", "_maxsize")

      def __init__(self, maxsize: int) -> None:
          self._data: OrderedDict[str, Any] = OrderedDict()
          self._maxsize = maxsize

      def get(self, key: str) -> Any | None:
          try:
              self._data.move_to_end(key)
              return self._data[key]
          except KeyError:
              return None

      def put(self, key: str, value: Any) -> None:
          if key in self._data:
              self._data.move_to_end(key)
              self._data[key] = value
          else:
              if len(self._data) >= self._maxsize:
                  self._data.popitem(last=False)
              self._data[key] = value


  # ---------------------------------------------------------------------------
  # Param template: compiled SQL + name→position mapping
  # ---------------------------------------------------------------------------

  class ParamTemplate:
      """
      Pairs a compiled SQL string with a parameter-position map.

      The ``params`` list mirrors what pypika's ``get_parameterized_sql()``
      returns: concrete values at most positions, but ``Parameter`` /
      ``CollectionParameter`` objects at placeholder positions.  ``fill()``
      copies this list and substitutes runtime values via each Parameter's
      ``encode_value`` chain.
      """

      __slots__ = ("sql", "_params", "_simple", "_collections", "_required")

      def __init__(self, sql: str, params: list[Any]) -> None:
          self.sql = sql
          self._params = params

          self._simple: dict[str, list[tuple[int, Parameter]]] = defaultdict(list)
          self._collections: dict[str, list[tuple[int, CollectionParameter]]] = defaultdict(list)
          names: set[str] = set()

          for idx, p in enumerate(params):
              if isinstance(p, CollectionParameter):
                  self._collections[p.name].append((idx, p))
                  names.add(p.name)
              elif isinstance(p, Parameter):
                  self._simple[p.name].append((idx, p))
                  names.add(p.name)

          self._required: frozenset[str] = frozenset(names)

      def fill(self, values: dict[str, Any]) -> list[Any]:
          """Return a new params list with runtime values substituted."""
          missing = self._required - values.keys()
          if missing:
              raise ParamsError(
                  f"Missing required parameters: {', '.join(sorted(missing))}"
              )

          filled = list(self._params)

          # Scalar parameters — same name may appear at multiple positions
          # (e.g. same Parameter used in both WHERE and a subquery)
          for name, entries in self._simple.items():
              raw = values[name]
              for idx, param in entries:
                  filled[idx] = param.encode_value(raw)

          # Collection parameters — each position corresponds to one element.
          # The SQL was generated with exactly len(entries) placeholders for
          # this collection size, so the caller must supply that many items.
          for name, entries in self._collections.items():
              raw = values[name]
              first_param = entries[0][1]

              # encode_collection calls list_encoder → field.to_db_value per item
              if first_param.collection_encoder:
                  encoded = list(first_param.encode_collection(raw))
              else:
                  if not isinstance(raw, (list, tuple, set)):
                      raise ParamsError(
                          f"Collection parameter {name!r} requires a "
                          f"list/tuple/set, got {type(raw).__name__}"
                      )
                  encoded = list(raw)

              if len(encoded) != len(entries):
                  raise ParamsError(
                      f"Collection parameter {name!r}: SQL was compiled for "
                      f"{len(entries)} items, got {len(encoded)}"
                  )

              for (idx, param), val in zip(entries, encoded):
                  filled[idx] = param.encode_value(val)

          return filled


  # ---------------------------------------------------------------------------
  # CompiledQuery — the single public class
  # ---------------------------------------------------------------------------

  class CompiledQuery(Generic[MODEL]):
      """
      A frozen compiled query that caches SQL generation.

      Created via ``QuerySet.compile(key)`` (or ``.update().compile()``,
      ``.delete().compile()``, etc.).  On the first call ``_make_query()``
      builds the pypika tree and ``get_parameterized_sql()`` extracts the SQL
      template.  On every subsequent ``execute()`` only parameter substitution
      and DB execution happen.

      For queries **without** collection parameters (``__in``), the SQL is
      compiled once and reused forever.  For queries **with** collection
      parameters, a separate SQL template is cached per unique combination of
      collection sizes in a bounded LRU.

      This class deliberately exposes only ``execute(**params)`` and
      ``sql(**params)``.  There are no ``filter`` / ``order_by`` / ``limit``
      methods — the query shape is fixed at compile time.
      """

      __slots__ = (
          "_model",
          "_query_builder",
          "_execute_fn",
          "_db_for_write",
          "_collection_params",
          "_has_collections",
          "_fixed_template",
          "_size_cache",
      )

      DEFAULT_CACHE_SIZE = 128

      def __init__(
          self,
          *,
          model: type[MODEL],
          query_builder: QueryBuilder,
          execute_fn: ExecuteFn,
          db_for_write: bool = False,
          cache_size: int = DEFAULT_CACHE_SIZE,
      ) -> None:
          self._model = model
          self._query_builder = query_builder
          self._execute_fn = execute_fn
          self._db_for_write = db_for_write

          # -- discover collection parameters from an initial parameterization --
          _, initial_params = query_builder.get_parameterized_sql()
          self._collection_params: dict[str, CollectionParameter] = {}
          for p in initial_params:
              if (
                  isinstance(p, CollectionParameter)
                  and p.name not in self._collection_params
              ):
                  self._collection_params[p.name] = p
          self._has_collections = bool(self._collection_params)

          # -- fast path: no collection params → compile once --
          if not self._has_collections:
              sql, params = query_builder.get_parameterized_sql()
              self._fixed_template: ParamTemplate | None = ParamTemplate(sql, params)
              self._size_cache = None
          else:
              self._fixed_template = None

              # -- slow path: bounded cache keyed by collection sizes --
              self._size_cache = _BoundedLRU(maxsize=cache_size)

      # -- public API ---------------------------------------------------------

      def sql(self, **params: Any) -> str:
          """Return the SQL that would be executed (for debugging / logging)."""
          return self._get_template(params).sql

      async def execute(self, **params: Any) -> Any:
          """Execute with the given parameter values and return the result."""
          template = self._get_template(params)
          filled = template.fill(params)
          db = self._choose_db()
          return await self._execute_fn(db, template.sql, filled)

      # -- internals ----------------------------------------------------------

      def _choose_db(self) -> BaseDBAsyncClient:
          if self._db_for_write:
              return router.db_for_write(self._model) or self._model._meta.db
          return router.db_for_read(self._model) or self._model._meta.db

      def _get_template(self, params: dict[str, Any]) -> ParamTemplate:
          # Fast path: no collection params, SQL is always identical
          if self._fixed_template is not None:
              return self._fixed_template

          # Slow path: cache key encodes each collection's length
          size_key = self._make_size_key(params)

          cached = self._size_cache.get(size_key)
          if cached is not None:
              return cached

          template = self._compile_for_sizes(params)
          self._size_cache.put(size_key, template)
          return template

      def _compile_for_sizes(self, params: dict[str, Any]) -> ParamTemplate:
          """
          Re-run ``get_parameterized_sql()`` with ``collection_size`` set on
          each ``CollectionParameter``, producing SQL with the correct number
          of placeholders per IN-clause.

          ``collection_size`` is always reset in a ``finally`` block to avoid
          corrupting the shared query tree.
          """
          try:
              for name, cparam in self._collection_params.items():
                  value = params.get(name)
                  if not isinstance(value, (list, tuple, set)):
                      raise ParamsError(
                          f"Collection parameter {name!r} requires a "
                          f"list/tuple/set, got {type(value).__name__}"
                      )
                  cparam.collection_size = len(value)

              ctx = TortoiseSqlContext.copy(
                  self._query_builder.QUERY_CLS.SQL_CONTEXT,
                  dynamic_params=self._collection_params,
              )
              sql, raw_params = self._query_builder.get_parameterized_sql(ctx)
              return ParamTemplate(sql, raw_params)
          finally:
              for cparam in self._collection_params.values():
                  cparam.collection_size = None

      def _make_size_key(self, params: dict[str, Any]) -> str:
          parts: list[str] = []
          for name in sorted(self._collection_params):
              value = params.get(name, ())
              size = len(value) if isinstance(value, (list, tuple, set)) else 0
              parts.append(f"{name}:{size}")
          return "|".join(parts)

General .compile method can look like this, probably overriden un QuerySet children classes:

  def compile(self) -> CompiledQuery[MODEL]:
      from tortoise.compiled_query import CompiledQuery

      self._db = self._choose_db(self._select_for_update)
      self._make_query()

      model = self.model
      prefetch_map = self._prefetch_map
      prefetch_queries = self._prefetch_queries
      select_related_idx = self._select_related_idx
      custom_fields = list(self._annotations.keys()) or None
      single = self._single
      raise_does_not_exist = self._raise_does_not_exist

      async def execute_fn(db, sql, params):
          instance_list = await db.executor_class(
              model=model,
              db=db,
              prefetch_map=prefetch_map,
              prefetch_queries=prefetch_queries,
              select_related_idx=select_related_idx,
          ).execute_select(sql, params, custom_fields=custom_fields)
          if single:
              if len(instance_list) == 1:
                  return instance_list[0]
              if not instance_list:
                  if raise_does_not_exist:
                      raise DoesNotExist(model)
                  return None
              raise MultipleObjectsReturned(model)
          return instance_list

      return CompiledQuery(
          key=key, model=self.model, query_builder=self.query,
          execute_fn=execute_fn, db_for_write=self._select_for_update,
      )

It's debatable if we should bake-in all execute variables like this, maybe we can pass them to compiled query and make them mutable, but I don't see much of advatages for that

Overall, it would encapsulate majority of new logic to this new module, would not require us to maintain mirrored api of prepared queryset, it looks less prone to memory leaks and gives more control to user on how he wants to manage his queries.

[RuslanUC]

Have you considered simplifying it other way?

I thought about moving _cache_key and cached_sql from Prepared* classes to regular ones. I tried it a some time before and didn't like it for some reason (don't remember why now), but I know that overall implementation was different, but now I don't think that merging regular and prepared classes is that bad of an idea. I'll try it.

I looked a bit into your code, I thought (around the time of first commits) about just adding method like your "compile", that should be called after while query is constructed. Now that I better know how queryset works, I think this is a good idea.

Honestly current implementation seems a bit over the top for what it actually does, as duplicating whole queryset, making implicit cache storages and etc is quite big change with a lot of new APIs.

That's why I created this draft pr instead of continuing to make changes in a separate fork. Thanks for feedback and ideas.